fres/resmng/frm_generic.c

   1 /**
   2  * @file   frm_generic.c
   3  * @author Michal Sojka <sojkam1@fel.cvut.cz>
   4  * @date   Tue Nov 11 08:34:34 2008
   5  *
   6  * @brief  Generic resource manager implementation.
   7  *
   8  *
   9  */
  10 #include <frm_generic.h>
  11 #include <forb.h>
  12 #include <ul_log.h>
  13 #include <fres_sa_scenario.h>
  14 #include <fcb.h>
  15
  16 UL_LOG_CUST(ulogd_frm);
  17 ul_log_domain_t ulogd_frm = {UL_LOGL_MSG, "frm"};
  18
  19 #define object_to_frm(o) (struct frm_data*)forb_instance_data(o)
  20 #define save_errno(cmd) do { int _e = errno; cmd; errno = _e; } while(0)
  21
  22 void fres_sa_scenario_reserve_new(struct fres_sa_scenario *scenario)
  23 {
  24         struct fres_sa_contract *c;
  25         fres_sa_scenario_for_each_contract(scenario, c) {
  26                 if (c->new) {
  27                         if (c->reserved) {
  28                                 fres_contract_destroy(c->reserved);
  29                         }
  30                         c->reserved = c->new;
  31                         c->new = NULL;
  32                 }
  33         }
  34 }
  35
  36 void fres_sa_scenario_rollback(struct fres_sa_scenario *scenario)
  37 {
  38         struct fres_sa_contract *c, *c_next;
  39
  40         /* Deleteion safe scenario traverse */
  41         for(c=fres_sa_scenario_contract_first(scenario),
  42                     c_next=c?fres_sa_scenario_contract_next(scenario,c):NULL;
  43             c;
  44             c=c_next,c_next=c?fres_sa_scenario_contract_next(scenario,c):NULL) {
  45                 if (c->new) {
  46                         if (c->committed || c->reserved) {
  47                                         fres_contract_destroy(c->new);
  48                                         c->new = NULL;
  49                                         c->contract = c->reserved ? c->reserved : c->committed;
  50                         } else {
  51                                 fres_sa_scenario_del_contract(scenario, c);
  52                                 fres_sa_contract_destroy(c);
  53                         }
  54                 }
  55         }
  56 }
  57
  58
  59 static CORBA_long reserve_contracts(fres_resource_manager obj,
  60                                     const fres_contract_ptr_seq* contracts,
  61                                     CORBA_Environment *ev)
  62 {
  63         struct frm_data *frm = object_to_frm(obj);
  64         struct fres_sa_scenario *scenario = frm->scenario;
  65         bool schedulable =false;
  66         int i, ret;
  67         struct fres_sa_contract *c;
  68
  69         ul_logmsg("reserve_contracts\n");
  70
  71         for (i=0; i<contracts->_length; i++) {
  72                 struct fres_contract *cin = contracts->_buffer[i];
  73                 c = fres_sa_scenario_find_contract(scenario, &cin->id);
  74                 if (!c) {
  75                         c = fres_sa_contract_new();
  76                         if (c) {
  77                                 c->id = cin->id;
  78                                 fres_sa_scenario_add_contract(scenario, c);
  79                         }
  80                 }
  81                 if (!c) goto err;
  82                 assert(c->new == NULL);
  83                 c->new = fres_contract_duplicate(cin);
  84                 c->contract = c->new;
  85                 if (!c->new) goto err;
  86         }
  87
  88         ret = frm->desc->admission_test(scenario, frm->desc->priv, &schedulable);
  89         if (ret) {
  90                 ul_logerr("admission_test failed: %d\n", ret);
  91                 goto err;
  92         }
  93
  94         if (schedulable) {
  95                 fres_sa_scenario_reserve_new(scenario);
  96         } else {
  97                 fres_sa_scenario_rollback(scenario);
  98         }
  99         return schedulable ? 0 : 1;
 100 err:
 101         fres_sa_scenario_rollback(scenario);
 102         return -1;
 103 }
 104
 105 static void commit_contracts(fres_resource_manager obj,
 106                       const fres_contract_id_seq* ids,
 107                       fres_contract_ptr_seq** contracts_with_scheduling_data,
 108                       CORBA_Environment *ev)
 109 {
 110         struct frm_data *frm = object_to_frm(obj);
 111         int i, num;
 112         struct fres_sa_contract *c;
 113         fres_contract_ptr_seq *contracts;
 114
 115         ul_logmsg("commit_contracts\n");
 116
 117         contracts = forb_malloc(sizeof(*contracts));
 118         if (!contracts) {
 119                 ev->major = FORB_EX_NO_MEMORY;
 120                 goto err;
 121         }
 122         num = ids->_length;
 123         contracts->_buffer = CORBA_sequence_fres_contract_ptr_allocbuf(num);
 124         CORBA_sequence_set_release(contracts, CORBA_TRUE);
 125         contracts->_maximum = contracts->_length = num;
 126
 127         /* TODO: Add also the changed contracts (e.g. because of
 128          * priorities). Question: How to recognize which contracts are
 129          * changed because of this commit? */
 130         for (i=0; i < num; i++) {
 131                 c = fres_sa_scenario_find_contract(frm->scenario, &ids->_buffer[i]);
 132                 if (c && c->reserved) {
 133                         if (c->committed) fres_contract_destroy(c->committed);
 134                         c->committed = c->reserved;
 135                         c->reserved = NULL;
 136                         contracts->_buffer[i] = fres_contract_duplicate(c->committed);
 137                 } else {
 138                         contracts->_buffer[i] = NULL;
 139                         if (!c) ul_logerr("Commit to unknown contract ID\n");
 140                         else if (!c->reserved) ul_logerr("Commit to not reserved contract\n");
 141                 }
 142         }
 143
 144         *contracts_with_scheduling_data = contracts;
 145 err:;
 146 }
 147
 148 static void cancel_contracts(fres_resource_manager obj,
 149                       const fres_contract_id_seq* ids,
 150                       CORBA_Environment *ev)
 151 {
 152         int i;
 153         struct frm_data *frm = object_to_frm(obj);
 154
 155         ul_logmsg("cancel_contracts\n");
 156
 157         for (i=0; i<ids->_length; i++) {
 158                 struct fres_sa_contract *c;
 159                 c = fres_sa_scenario_find_contract(frm->scenario, &ids->_buffer[i]);
 160
 161                 if (c) {
 162                         fres_sa_scenario_del_contract(frm->scenario, c);
 163                         fres_sa_contract_destroy(c);
 164                 }
 165         }
 166 }
 167
 168 static void get_contracts(fres_resource_manager obj,
 169                           fres_contract_ptr_seq** contracts_out,
 170                           CORBA_long* utilization, CORBA_Environment *ev)
 171 {
 172         struct frm_data *frm = object_to_frm(obj);
 173         struct fres_sa_contract *c;
 174         fres_contract_ptr_seq *contracts;
 175
 176         contracts = forb_malloc(sizeof(*contracts));
 177         if (!contracts) {
 178                 ev->major = FORB_EX_NO_MEMORY;
 179                 goto err;
 180         }
 181         contracts->_buffer = CORBA_sequence_fres_contract_ptr_allocbuf(frm->scenario->num_contracts);
 182         CORBA_sequence_set_release(contracts, CORBA_TRUE);
 183         contracts->_maximum = frm->scenario->num_contracts;
 184         contracts->_length = 0;
 185
 186         fres_sa_scenario_for_each_contract(frm->scenario, c) {
 187                 if (c->committed) {
 188                         int i = contracts->_length;
 189                         contracts->_buffer[i] = fres_contract_duplicate(c->committed);
 190                         contracts->_length++;
 191                 }
 192
 193         }
 194         *contracts_out = contracts;
 195         *utilization = frm->scenario->utilization;
 196 err:;
 197 }
 198
 199
 200 static const struct forb_fres_resource_manager_impl frm_impl = {
 201         .reserve_contracts = reserve_contracts,
 202         .commit_contracts  = commit_contracts,
 203         .cancel_contracts  = cancel_contracts,
 204         .get_contracts  = get_contracts,
 205 };
 206
 207 /**
 208  * Initializes generic resource manager. The only thing a
 209  * caller has to supply is admission test function, which is passed in
 210  * @a frm_data->admission_test.
 211  *
 212  * @param orb FORB object used to communicate with other components.
 213  * @param[out] frm_data Pointer to frm_data structure
 214  * @param executor Executor used for this resource manager.
 215  * @param desc Description on the resource manager
 216  *
 217  * @return Resource manager object reference of NULL in case of error.
 218  */
 219 fres_resource_manager frm_register(forb_orb orb, struct frm_data *frm_data,
 220                                    forb_executor_t *executor,
 221                                    const struct fres_res_manager *desc)
 222 {
 223         fres_contract_broker fcb;
 224         fres_resource_desc res_desc;
 225         fres_resource_manager frm;
 226         struct forb_env env;
 227         int ret;
 228
 229         memset(frm_data, 0, sizeof(*frm_data));
 230         frm_data->desc = desc;
 231         frm_data->scenario = fres_sa_scenario_new();
 232         if (!frm_data->scenario) {
 233                 save_errno(ul_logerr("fres_sa_scenario_new failed\n"));
 234                 goto err;
 235         }
 236
 237         fcb = forb_resolve_reference(orb, fres_contract_broker_reg_name);
 238         if (!fcb) {
 239                 save_errno(ul_logerr("Could not find contract broker\n"));
 240                 goto err;
 241         }
 242
 243         frm = forb_fres_resource_manager_new(orb, &frm_impl, frm_data);
 244         if (!frm) {
 245                 save_errno(ul_logerr("forb_fres_resource_manager_new error\n"));
 246                 goto err_release_fcb;
 247         }
 248
 249         ret = forb_executor_register_object(executor, frm);
 250         if (ret) {
 251                 save_errno(ul_logerr("forb_executor_register_object failed\n"));
 252                 goto err_executor;
 253         }
 254
 255         /* Register resource manager */
 256         res_desc.manager = frm;
 257         ret = fres_contract_broker_register_resource(fcb,
 258                                                      desc->res_type,
 259                                                      desc->res_id,
 260                                                      &res_desc, &env);
 261         if (forb_exception_occurred(&env) || ret != 0) {
 262                 save_errno(ul_logerr("fres_contract_broker_register_resource exception\n"));
 263                 goto err_register;
 264         }
 265
 266         forb_object_release(fcb);
 267         return frm;
 268 err_register:
 269         forb_executor_unregister_object(executor, frm);
 270 err_executor:
 271         forb_object_release(frm);
 272 err_release_fcb:
 273         forb_object_release(fcb);
 274 err:
 275         return NULL;
 276 }
 277
 278 /**
 279  * Initializes and runs a generic resource manager. The only thing a
 280  * caller has to supply is admission test function, which is passed in
 281  * @a frm_data->admission_test.
 282  *
 283  * @param orb FORB object used to communicate with other components.
 284  * @param desc Description on the resource manager
 285  *
 286  * @return
 287  */
 288 int frm_register_and_run(forb_orb orb, const struct fres_res_manager *desc)
 289 {
 290         fres_resource_manager frm;
 291         struct frm_data frm_data;
 292         forb_executor_t executor;
 293         int ret = -1;
 294
 295         /* Prepare executor before we register the resource manager
 296          * with contract broker */
 297         ret = forb_executor_init(&executor);
 298         if (ret) {
 299                 save_errno(ul_logerr("forb_executor_init failed\n"));
 300                 goto err;
 301         }
 302
 303         frm = frm_register(orb, &frm_data, &executor, desc);
 304         if (!frm) {
 305                 ret = -1;
 306                 save_errno(ul_logerr("frm_register failed\n"));
 307                 goto err_destroy_executor;
 308         }
 309
 310         /* Start request processing */
 311         ul_logmsg("Waiting for requests\n");
 312         ret = forb_executor_run(&executor);
 313         if (ret) goto err_release_frm;
 314
 315 err_release_frm:
 316         forb_object_release(frm);
 317 err_destroy_executor:
 318         forb_executor_destroy(&executor);
 319 err:
 320         return ret;
 321 }