fres/cbroker/fcb.c

   1 /**************************************************************************/
   2 /* ---------------------------------------------------------------------- */
   3 /* Copyright (C) 2006 - 2008 FRESCOR consortium partners:                 */
   4 /*                                                                        */
   5 /*   Universidad de Cantabria,              SPAIN                         */
   6 /*   University of York,                    UK                            */
   7 /*   Scuola Superiore Sant'Anna,            ITALY                         */
   8 /*   Kaiserslautern University,             GERMANY                       */
   9 /*   Univ. Politécnica  Valencia,           SPAIN                        */
  10 /*   Czech Technical University in Prague,  CZECH REPUBLIC                */
  11 /*   ENEA                                   SWEDEN                        */
  12 /*   Thales Communication S.A.              FRANCE                        */
  13 /*   Visual Tools S.A.                      SPAIN                         */
  14 /*   Rapita Systems Ltd                     UK                            */
  15 /*   Evidence                               ITALY                         */
  16 /*                                                                        */
  17 /*   See http://www.frescor.org for a link to partners' websites          */
  18 /*                                                                        */
  19 /*          FRESCOR project (FP6/2005/IST/5-034026) is funded             */
  20 /*       in part by the European Union Sixth Framework Programme          */
  21 /*       The European Union is not liable of any use that may be          */
  22 /*       made of this code.                                               */
  23 /*                                                                        */
  24 /*                                                                        */
  25 /*  This file is part of FRSH (FRescor ScHeduler)                         */
  26 /*                                                                        */
  27 /* FRSH is free software; you can redistribute it and/or modify it        */
  28 /* under terms of the GNU General Public License as published by the      */
  29 /* Free Software Foundation; either version 2, or (at your option) any    */
  30 /* later version.  FRSH is distributed in the hope that it will be        */
  31 /* useful, but WITHOUT ANY WARRANTY; without even the implied warranty    */
  32 /* of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU    */
  33 /* General Public License for more details. You should have received a    */
  34 /* copy of the GNU General Public License along with FRSH; see file       */
  35 /* COPYING. If not, write to the Free Software Foundation, 675 Mass Ave,  */
  36 /* Cambridge, MA 02139, USA.                                              */
  37 /*                                                                        */
  38 /* As a special exception, including FRSH header files in a file,         */
  39 /* instantiating FRSH generics or templates, or linking other files       */
  40 /* with FRSH objects to produce an executable application, does not       */
  41 /* by itself cause the resulting executable application to be covered     */
  42 /* by the GNU General Public License. This exception does not             */
  43 /* however invalidate any other reasons why the executable file might be  */
  44 /* covered by the GNU Public License.                                     */
  45 /**************************************************************************/
  46
  47 /**
  48  * @file   fcb.c
  49  * @author Michal Sojka <sojkam1@fel.cvut.cz>
  50  * @date   Mon Oct 20 18:00:18 2008
  51  *
  52  * @brief  FRESCOR Contract Broker
  53  *
  54  *
  55  */
  56 #include <forb.h>
  57 #include <fcb.h>
  58 #include <error.h>
  59 #include <errno.h>
  60 #include <stdio.h>
  61 #include <ul_gavlcust.h>
  62 #include <string.h>
  63 #include <ul_log.h>
  64 #include <forb/server_id.h>
  65 #include <fres_contract.h>
  66
  67 UL_LOG_CUST(ulogd_fcb);
  68 ul_log_domain_t ulogd_fcb = {UL_LOGL_MSG, "fcb"};
  69
  70
  71 /**
  72  * Resource identification
  73  */
  74 struct res_key {
  75         frsh_resource_type_t type;
  76         frsh_resource_id_t id;
  77 };
  78
  79 /**
  80  * Registered resource manager
  81  */
  82 struct res_mng {
  83         gavl_node_t node;
  84         struct res_key resource;
  85         fres_resource_manager rm; /**< Object reference of the resource manager */
  86         gavl_cust_root_field_t allocators; /**< Registered allocators for this resource (from multiple applications/nodes) */
  87 };
  88
  89 /**
  90  * Resource allocator registered in different nodes/applications
  91  */
  92 struct res_alloc {
  93         gavl_node_t node;
  94         forb_server_id app;
  95         fres_resource_allocator ra;
  96 };
  97
  98 struct fcb {
  99         gavl_cust_root_field_t resources;
 100 };
 101
 102 static inline int res_key_cmp(struct res_key *a, struct res_key *b)
 103 {
 104         if (a->type < b->type) {
 105                 return -1;
 106         } else if (a->type > b->type) {
 107                 return +1;
 108         } else if (a->id < b->id) {
 109                 return -1;
 110         } else if (a->id > b->id) {
 111                 return +1;
 112         } else {
 113                 return 0;
 114         }
 115 }
 116
 117 GAVL_CUST_NODE_INT_DEC(fcb_resource   /* cust_prefix */,                \
 118                        struct fcb     /* cust_root_t */,                \
 119                        struct res_mng /* cust_item_t */,                \
 120                        struct res_key /* cust_key_t */,                 \
 121                        resources      /* cust_root_node */,             \
 122                        node           /* cust_item_node */,             \
 123                        resource       /* cust_item_key */,              \
 124                        res_key_cmp    /* cust_cmp_fnc */);
 125
 126 GAVL_CUST_NODE_INT_IMP(fcb_resource   /* cust_prefix */,                \
 127                        struct fcb     /* cust_root_t */,                \
 128                        struct res_mng /* cust_item_t */,                \
 129                        struct res_key /* cust_key_t */,                 \
 130                        resources      /* cust_root_node */,             \
 131                        node           /* cust_item_node */,             \
 132                        resource       /* cust_item_key */,              \
 133                        res_key_cmp    /* cust_cmp_fnc */);
 134
 135 GAVL_CUST_NODE_INT_DEC(fcb_alloc        /* cust_prefix */,              \
 136                        struct res_mng   /* cust_root_t */,              \
 137                        struct res_alloc /* cust_item_t */,              \
 138                        forb_server_id   /* cust_key_t */,               \
 139                        allocators       /* cust_root_node */,           \
 140                        node             /* cust_item_node */,           \
 141                        app              /* cust_item_key */,            \
 142                        fres_contract_id_cmp /* cust_cmp_fnc */);
 143
 144 GAVL_CUST_NODE_INT_IMP(fcb_alloc        /* cust_prefix */,              \
 145                        struct res_mng   /* cust_root_t */,              \
 146                        struct res_alloc /* cust_item_t */,              \
 147                        forb_server_id   /* cust_key_t */,               \
 148                        allocators       /* cust_root_node */,           \
 149                        node             /* cust_item_node */,           \
 150                        app              /* cust_item_key */,            \
 151                        forb_server_id_cmp /* cust_cmp_fnc */);
 152
 153
 154 #define o2fcb(o) (struct fcb*)forb_instance_data(o)
 155
 156 CORBA_long
 157 negotiate_contract(fres_contract_broker obj,
 158                    const fres_contract_ptr contract,
 159                    fres_contract_id_t* id,
 160                    CORBA_Environment *ev)
 161 {
 162         struct fcb *fcb = o2fcb(obj);
 163         fres_block_resource *res;
 164         struct res_key res_key;
 165         struct res_mng *rm;
 166         struct res_alloc *ra;
 167         int ret;
 168         forb_server_id app;
 169         fres_contract_ptr contract_seq_buf;
 170         fres_contract_ptr_seq contracts;
 171         fres_contract_ptr_seq *schedulable_contracts;
 172         fres_contract_id_seq ids;
 173
 174
 175         res = fres_contract_get_resource(contract);
 176         if (!res) {
 177                 ul_logerr("No resource specified\n");
 178                 ret = FRSH_ERR_RESOURCE_ID_INVALID;
 179                 goto err;
 180         }
 181         res_key.type = res->resource_type;
 182         res_key.id = res->resource_id;
 183
 184         rm = fcb_resource_find(fcb, &res_key);
 185         if (!rm) {
 186                 ul_logerr("No resource manager for %d.%d registered\n",
 187                           res_key.type, res_key.id);
 188                 ret = -1;
 189                 goto err;
 190         }
 191
 192         forb_uuid_generate((forb_uuid_t *)id);
 193         contract->id = *id;
 194
 195         forb_get_req_source(obj, &app);
 196         ra = fcb_alloc_find(rm, &app);
 197         if (!ra) {
 198                 char str[60];
 199                 forb_server_id_to_string(str, &app, sizeof(str));
 200                 ul_logerr("No resource allocator found for %d.%d and %s\n",
 201                           res_key.type, res_key.id, str);
 202                 ret = -1;
 203                 goto err;
 204         }
 205
 206
 207         /* Reserve contract */
 208         contracts._length = 1;
 209         contract_seq_buf = contract;
 210         contracts._buffer = &contract_seq_buf;
 211         ret = fres_resource_manager_reserve_contracts(rm->rm, &contracts, ev);
 212         if (forb_exception_occurred(ev) || ret < 0) {
 213                 goto err;
 214         }
 215         if (ret == 1) {
 216                 ul_logmsg("Contract was not accepted\n");
 217                 goto err;
 218         }
 219
 220         /* Commit contract */
 221         ids._length = ids._maximum = 1;
 222         ids._buffer = id;
 223         fres_resource_manager_commit_contracts(rm->rm, &ids, &schedulable_contracts, ev);
 224         if (forb_exception_occurred(ev)) {
 225                 ret = FRES_ERR_FORB_EXCEPTION;
 226                 goto err;
 227         }
 228
 229         /* Create VRes */
 230         ret = fres_resource_allocator_change_vreses(ra->ra, schedulable_contracts, ev);
 231         if (CORBA_sequence_get_release(schedulable_contracts)) {
 232                 int i;
 233                 for (i=0; i<schedulable_contracts->_length; i++) {
 234                         fres_contract_destroy(schedulable_contracts->_buffer[i]);
 235                 }
 236                 forb_free(schedulable_contracts->_buffer);
 237         }
 238         forb_free(schedulable_contracts);
 239         if (forb_exception_occurred(ev)) {
 240                 ret = FRES_ERR_FORB_EXCEPTION;
 241                 goto err;
 242         }
 243         if (ret != 0) {
 244                 ul_logmsg("VRes was not created\n");
 245                 goto err;
 246         }
 247
 248         return 0;
 249 err:
 250         return ret;
 251 }
 252
 253 CORBA_long register_resource(fres_contract_broker obj,
 254                             const frsh_resource_type_t restype,
 255                             const frsh_resource_id_t resid,
 256                             const fres_resource_desc *desc,
 257                             CORBA_Environment *ev)
 258 {
 259         struct fcb *fcb = o2fcb(obj);
 260         struct res_mng *rm, *rm2;
 261
 262         rm = malloc(sizeof(*rm));
 263         if (!rm) goto err;
 264         memset(rm, 0, sizeof(*rm));
 265         rm->resource.type = restype;
 266         rm->resource.id = resid;
 267         rm2 = fcb_resource_find(fcb, &rm->resource);
 268         if (rm2) {
 269                 if (forb_object_is_stale(rm2->rm)) {
 270                         ul_logmsg("Removing stale manager for resource %d.%d\n",
 271                                   restype, resid);
 272                         forb_object_release(rm2->rm);
 273                         fcb_resource_delete(fcb, rm2);
 274                         /* TODO: Delete also all allocators associated
 275                          * with this stale resource manager. */
 276                         free(rm);
 277                         rm = rm2;
 278                 } else {
 279                         ul_logerr("Resource manager %d.%d already registered\n",
 280                                   restype, resid);
 281                         goto free_err;
 282                 }
 283         }
 284         rm->rm = forb_object_duplicate(desc->manager);
 285
 286         fcb_alloc_init_root_field(rm);
 287         ul_logmsg("Registering manager for resource %d.%d\n",
 288                   restype, resid);
 289         fcb_resource_insert(fcb, rm);
 290         return 0;
 291 free_err:
 292         free(rm);
 293 err:
 294         return -1;
 295 }
 296
 297
 298 CORBA_long register_allocator(fres_contract_broker obj,
 299                               const frsh_resource_type_t restype,
 300                               const frsh_resource_id_t resid,
 301                               const fres_resource_allocator ra_obj,
 302                               CORBA_Environment *ev)
 303 {
 304         struct fcb *fcb = o2fcb(obj);
 305         struct res_mng *rm;
 306         struct res_alloc *ra;
 307         struct res_key resource;
 308         forb_server_id server_id;
 309         char server_id_str[40];
 310
 311         forb_get_server_id(ra_obj, &server_id);
 312         forb_server_id_to_string(server_id_str, &server_id, sizeof(server_id_str));
 313         ul_logmsg("Registering allocator for resource %d.%d in app %s\n",
 314                   restype, resid, server_id_str);
 315
 316         resource.type = restype;
 317         resource.id = resid;
 318         rm = fcb_resource_find(fcb, &resource);
 319         if (!rm) {
 320                 ul_logerr("No manager found for %d.%d. Unable to register the allocator!\n",
 321                           restype, resid);
 322                 goto err;
 323         }
 324         ra = fcb_alloc_find(rm, &server_id);
 325         if (ra) {
 326                 char *str = forb_object_to_string(ra_obj);
 327                 ul_logerr("Allocator from already registered (%s)\n",
 328                           str);
 329                 forb_free(str);
 330                 goto err;
 331         }
 332         ra = malloc(sizeof(*ra));
 333         if (!ra) {
 334                 goto err;
 335         }
 336         ra->app = server_id;
 337         ra->ra = forb_object_duplicate(ra_obj);
 338         fcb_alloc_insert(rm, ra);
 339         return 0;
 340 err:
 341         return -1;
 342 }
 343
 344 struct forb_fres_contract_broker_impl impl = {
 345         .negotiate_contract = negotiate_contract,
 346         .register_resource = register_resource,
 347         .register_allocator = register_allocator,
 348 };
 349
 350 int main(int argc, char *argv[])
 351 {
 352         forb_orb orb;
 353         struct fcb fcb_data;
 354         fres_contract_broker fcb;
 355         forb_executor_t executor;
 356         int ret;
 357
 358         orb = forb_init(&argc, &argv, "fcb");
 359         if (!orb) error(1, errno, "FORB initialization failed");
 360
 361         fcb_resource_init_root_field(&fcb_data);
 362
 363         fcb = forb_fres_contract_broker_new(orb, &impl, &fcb_data);
 364         if (!fcb) error(1, errno, "forb_fres_contract_broker_new failed");
 365
 366         /* Prepare executor before we register the fcb reference,
 367          * so that no reuqests are lost */
 368         ret = forb_executor_init(&executor);
 369         if (ret) error(1, errno, "forb_executor_init failed");
 370
 371         ret = forb_executor_register_object(&executor, fcb);
 372         if (ret) error(1, errno, "forb_executor_register_object failed");
 373
 374         ret = forb_register_reference(fcb, fres_contract_broker_reg_name);
 375         if (ret) error(1, errno, "forb_register_reference() failed");
 376
 377         ul_logmsg("Waiting for requests\n");
 378         ret = forb_executor_run(&executor);
 379         if (ret) error(1, errno, "forb_executor_run failed");
 380
 381         return 0;
 382 }