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 schedulers; /**< Registered schedulers for this resource (from multiple applications/nodes) */
  87 };
  88
  89 /**
  90  * Resource scheduler registered in different nodes/applications
  91  */
  92 struct res_sched {
  93         gavl_node_t node;
  94         forb_server_id app;
  95         fres_resource_scheduler rs;
  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_sched        /* cust_prefix */,              \
 136                        struct res_mng   /* cust_root_t */,              \
 137                        struct res_sched /* cust_item_t */,              \
 138                        forb_server_id   /* cust_key_t */,               \
 139                        schedulers       /* 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_sched        /* cust_prefix */,              \
 145                        struct res_mng   /* cust_root_t */,              \
 146                        struct res_sched /* cust_item_t */,              \
 147                        forb_server_id   /* cust_key_t */,               \
 148                        schedulers       /* 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_sched *rs;
 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         rs = fcb_sched_find(rm, &app);
 197         if (!rs) {
 198                 char str[60];
 199                 forb_server_id_to_string(str, &app, sizeof(str));
 200                 ul_logerr("No resource scheduler found for %d.%d and %s\n",
 201                           res_key.type, res_key.id, str);
 202                 goto err;
 203         }
 204
 205
 206         /* Reserve contract */
 207         contracts._length = 1;
 208         contract_seq_buf = contract;
 209         contracts._buffer = &contract_seq_buf;
 210         ret = fres_resource_manager_reserve_contracts(rm->rm, &contracts, ev);
 211         if (forb_exception_occured(ev) || ret < 0) {
 212                 goto err;
 213         }
 214         if (ret == 1) {
 215                 ul_logmsg("Contract was not accepted\n");
 216                 goto err;
 217         }
 218
 219         /* Commit contract */
 220         ids._length = ids._maximum = 1;
 221         ids._buffer = id;
 222         fres_resource_manager_commit_contracts(rm->rm, &ids, &schedulable_contracts, ev);
 223         if (forb_exception_occured(ev)) {
 224                 goto err;
 225         }
 226
 227         /* Create VRes */
 228         ret = fres_resource_scheduler_change_vreses(rs->rs, schedulable_contracts, ev);
 229         if (CORBA_sequence_get_release(schedulable_contracts)) {
 230                 int i;
 231                 for (i=0; i<schedulable_contracts->_length; i++) {
 232                         fres_contract_destroy(schedulable_contracts->_buffer[i]);
 233                 }
 234                 forb_free(schedulable_contracts->_buffer);
 235         }
 236         forb_free(schedulable_contracts);
 237         if (forb_exception_occured(ev)) {
 238                 goto err;
 239         }
 240         if (ret != 0) {
 241                 ul_logmsg("VRes was not created\n");
 242                 goto err;
 243         }
 244
 245         return 0;
 246 err:
 247         return ret;
 248 }
 249
 250 CORBA_long register_resource(fres_contract_broker obj,
 251                             const frsh_resource_type_t restype,
 252                             const frsh_resource_id_t resid,
 253                             const fres_resource_desc *desc,
 254                             CORBA_Environment *ev)
 255 {
 256         struct fcb *fcb = o2fcb(obj);
 257         struct res_mng *rm, *rm2;
 258
 259         rm = malloc(sizeof(*rm));
 260         if (!rm) goto err;
 261         memset(rm, 0, sizeof(*rm));
 262         rm->resource.type = restype;
 263         rm->resource.id = resid;
 264         rm2 = fcb_resource_find(fcb, &rm->resource);
 265         if (rm2) {
 266                 if (forb_object_is_stale(rm2->rm)) {
 267                         ul_logmsg("Removing stale manager for resource %d.%d\n",
 268                                   restype, resid);
 269                         forb_object_release(rm2->rm);
 270                         fcb_resource_delete(fcb, rm2);
 271                         /* TODO: Delete also all schedulers associated
 272                          * with this stale resource manager. */
 273                         free(rm);
 274                         rm = rm2;
 275                 } else {
 276                         ul_logerr("Resource manager %d.%d already registered\n",
 277                                   restype, resid);
 278                         goto free_err;
 279                 }
 280         }
 281         rm->rm = forb_object_duplicate(desc->manager);
 282
 283         fcb_sched_init_root_field(rm);
 284         ul_logmsg("Registering manager for resource %d.%d\n",
 285                   restype, resid);
 286         fcb_resource_insert(fcb, rm);
 287         return 0;
 288 free_err:
 289         free(rm);
 290 err:
 291         return -1;
 292 }
 293
 294
 295 CORBA_long register_scheduler(fres_contract_broker obj,
 296                               const frsh_resource_type_t restype,
 297                               const frsh_resource_id_t resid,
 298                               const fres_resource_scheduler rs_obj,
 299                               CORBA_Environment *ev)
 300 {
 301         struct fcb *fcb = o2fcb(obj);
 302         struct res_mng *rm;
 303         struct res_sched *rs;
 304         struct res_key resource;
 305         forb_server_id server_id;
 306         char server_id_str[40];
 307
 308         forb_get_server_id(rs_obj, &server_id);
 309         forb_server_id_to_string(server_id_str, &server_id, sizeof(server_id_str));
 310         ul_logmsg("Registering scheduler for resource %d.%d in app %s\n",
 311                   restype, resid, server_id_str);
 312
 313         resource.type = restype;
 314         resource.id = resid;
 315         rm = fcb_resource_find(fcb, &resource);
 316         if (!rm) {
 317                 ul_logerr("No manager found for %d.%d. Unable to register the scheduler!\n",
 318                           restype, resid);
 319                 goto err;
 320         }
 321         rs = fcb_sched_find(rm, &server_id);
 322         if (rs) {
 323                 char *str = forb_object_to_string(rs_obj);
 324                 ul_logerr("Scheduler from already registered (%s)\n",
 325                           str);
 326                 forb_free(str);
 327                 goto err;
 328         }
 329         rs = malloc(sizeof(*rs));
 330         if (!rs) {
 331                 goto err;
 332         }
 333         rs->app = server_id;
 334         rs->rs = forb_object_duplicate(rs_obj);
 335         fcb_sched_insert(rm, rs);
 336         return 0;
 337 err:
 338         return -1;
 339 }
 340
 341 struct forb_fres_contract_broker_impl impl = {
 342         .negotiate_contract = negotiate_contract,
 343         .register_resource = register_resource,
 344         .register_scheduler = register_scheduler,
 345 };
 346
 347 int main(int argc, char *argv[])
 348 {
 349         forb_orb orb;
 350         struct fcb fcb_data;
 351         fres_contract_broker fcb;
 352         forb_executor_t executor;
 353         int ret;
 354
 355         orb = forb_init(&argc, &argv, "fcb");
 356         if (!orb) error(1, errno, "FORB initialization failed");
 357
 358         fcb_resource_init_root_field(&fcb_data);
 359
 360         fcb = forb_fres_contract_broker_new(orb, &impl, &fcb_data);
 361         if (!fcb) error(1, errno, "forb_fres_contract_broker_new failed");
 362
 363         /* Prepare executor before we register the fcb reference,
 364          * so that no reuqests are lost */
 365         ret = forb_executor_init(&executor);
 366         if (ret) error(1, errno, "forb_executor_init failed");
 367
 368         ret = forb_executor_register_object(&executor, fcb);
 369         if (ret) error(1, errno, "forb_executor_register_object failed");
 370
 371         ret = forb_register_reference(fcb, fres_contract_broker_reg_name);
 372         if (ret) error(1, errno, "forb_register_reference() failed");
 373
 374         ul_logmsg("Waiting for requests\n");
 375         ret = forb_executor_run(&executor);
 376         if (ret) error(1, errno, "forb_executor_run failed");
 377
 378         return 0;
 379 }