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 <forb/config.h>
  58 #include <fcb.h>
  59 #include <error.h>
  60 #include <errno.h>
  61 #include <stdio.h>
  62 #include <ul_gavlcust.h>
  63 #include <string.h>
  64 #include <ul_log.h>
  65 #include <forb/server_id.h>
  66 #include <fres_contract.h>
  67 #include "fcb_config.h"
  68 #include <fosa_clocks_and_timers.h>
  69 #if CONFIG_FCB_INET && !CONFIG_FORB_PROTO_INET_DEFAULT
  70 #include <forb/proto_inet.h>
  71 #endif
  72
  73 UL_LOG_CUST(ulogd_fcb);
  74 ul_log_domain_t ulogd_fcb = {UL_LOGL_MSG, "fcb"};
  75
  76 fosa_abs_time_t start_time;
  77
  78
  79 /**
  80  * Resource identification
  81  */
  82 struct res_key {
  83         frsh_resource_type_t type;
  84         frsh_resource_id_t id;
  85 };
  86
  87 /**
  88  * Registered resource
  89  */
  90 struct resource {
  91         gavl_node_t node;
  92         struct res_key key;
  93         char *name;
  94         fres_resource_manager mng; /**< Object reference of the resource manager */
  95         gavl_cust_root_field_t allocators; /**< Registered allocators for this resource (from multiple applications/nodes) */
  96         ul_list_head_t sc_contracts; /**< Negotiated contracts with spare capacity for this resource */
  97 };
  98
  99 /**
 100  * Resource allocator registered in different nodes/applications
 101  */
 102 struct res_alloc {
 103         gavl_node_t node;
 104         forb_server_id app;
 105         fres_resource_allocator ra;
 106 };
 107
 108 struct fcb_contract {
 109         fres_contract_id_t id;
 110         struct res_alloc *ra;   /**< Allocator for this contract TODO: Remove contract if allocator is destroyed */
 111         gavl_node_t node_fcb;
 112         ul_list_node_t node_sc;
 113         ul_list_node_t node_reservation;
 114         struct {
 115                 int initial;
 116                 int try;
 117         } sc_variant;
 118         fosa_abs_time_t end_of_stability_period;
 119         struct fres_contract *user_contract; /* The contract sent by user. Set after sucessfull neotiation. */
 120         struct fres_contract *requested_contract; /* User request for the contract. Set by prepare_reservation_requests() in the beginning of negotiation. */
 121         struct fres_contract *to_be_reserved_contract;
 122         struct fres_contract *schedulable_contract;
 123 };
 124
 125 /**
 126  * Contract broker data
 127  */
 128 struct fcb {
 129         gavl_cust_root_field_t resources; /**< Registered resources */
 130         gavl_cust_root_field_t contracts; /**< Contracts negotiated by this FCB */
 131 };
 132
 133 /** List of contracts to be reserved during spare capacity rebalancing */
 134 struct reservation_list {
 135         ul_list_head_t fcb_contracts;
 136         unsigned length;
 137 };
 138
 139 struct fcb_contract *fcb_contract_new(fres_contract_id_t *id)
 140 {
 141         struct fcb_contract *fcb_contract;
 142
 143         fcb_contract = malloc(sizeof(*fcb_contract));
 144         if (!fcb_contract) {
 145                 return NULL;
 146         }
 147         memset(fcb_contract, 0, sizeof(*fcb_contract));
 148         fcb_contract->id = *id;
 149
 150         return fcb_contract;
 151 }
 152
 153 void fcb_contract_destroy(struct fcb_contract *fcb_contract)
 154 {
 155         if (fcb_contract->user_contract) {
 156                 fres_contract_destroy(fcb_contract->user_contract);
 157         }
 158         if (fcb_contract->to_be_reserved_contract) {
 159                 fres_contract_destroy(fcb_contract->to_be_reserved_contract);
 160         }
 161         if (fcb_contract->requested_contract) {
 162                 fres_contract_destroy(fcb_contract->requested_contract);
 163         }
 164         if (fcb_contract->schedulable_contract) {
 165                 fres_contract_destroy(fcb_contract->requested_contract);
 166         }
 167         free(fcb_contract);
 168 }
 169
 170 static inline int res_key_cmp(const struct res_key *a,
 171                               const struct res_key *b)
 172 {
 173         if (a->type < b->type) {
 174                 return -1;
 175         } else if (a->type > b->type) {
 176                 return +1;
 177         } else if (a->id < b->id) {
 178                 return -1;
 179         } else if (a->id > b->id) {
 180                 return +1;
 181         } else {
 182                 return 0;
 183         }
 184 }
 185
 186 /* Container for registered resources */
 187 GAVL_CUST_NODE_INT_DEC(fcb_resource   /* cust_prefix */,                \
 188                        struct fcb     /* cust_root_t */,                \
 189                        struct resource/* cust_item_t */,                \
 190                        struct res_key /* cust_key_t */,                 \
 191                        resources      /* cust_root_node */,             \
 192                        node           /* cust_item_node */,             \
 193                        key            /* cust_item_key */,              \
 194                        res_key_cmp    /* cust_cmp_fnc */);
 195
 196 GAVL_CUST_NODE_INT_IMP(fcb_resource   /* cust_prefix */,                \
 197                        struct fcb     /* cust_root_t */,                \
 198                        struct resource/* cust_item_t */,                \
 199                        struct res_key /* cust_key_t */,                 \
 200                        resources      /* cust_root_node */,             \
 201                        node           /* cust_item_node */,             \
 202                        key            /* cust_item_key */,              \
 203                        res_key_cmp    /* cust_cmp_fnc */);
 204
 205 /* Container for allocators registered for a given resource */
 206 GAVL_CUST_NODE_INT_DEC(fcb_alloc        /* cust_prefix */,              \
 207                        struct resource  /* cust_root_t */,              \
 208                        struct res_alloc /* cust_item_t */,              \
 209                        forb_server_id   /* cust_key_t */,               \
 210                        allocators       /* cust_root_node */,           \
 211                        node             /* cust_item_node */,           \
 212                        app              /* cust_item_key */,            \
 213                        fres_contract_id_cmp /* cust_cmp_fnc */);
 214
 215 GAVL_CUST_NODE_INT_IMP(fcb_alloc        /* cust_prefix */,              \
 216                        struct resource   /* cust_root_t */,             \
 217                        struct res_alloc /* cust_item_t */,              \
 218                        forb_server_id   /* cust_key_t */,               \
 219                        allocators       /* cust_root_node */,           \
 220                        node             /* cust_item_node */,           \
 221                        app              /* cust_item_key */,            \
 222                        forb_server_id_cmp /* cust_cmp_fnc */);
 223
 224 /* Container for contracts with spare capacity negotiated for a given resource */
 225 UL_LIST_CUST_DEC(sc_contracts        /* cust_prefix */,                 \
 226                  struct resource  /* cust_head_t */,                    \
 227                  struct fcb_contract /* cust_item_t */,                 \
 228                  sc_contracts   /* cust_head_field */,                  \
 229                  node_sc        /* cust_node_field */);
 230
 231 UL_LIST_CUST_DEC(reservation_list        /* cust_prefix */,             \
 232                  struct reservation_list  /* cust_head_t */,            \
 233                  struct fcb_contract /* cust_item_t */,                 \
 234                  fcb_contracts  /* cust_head_field */,                  \
 235                  node_reservation /* cust_node_field */);
 236
 237 /* Container for negotiated contracts */
 238 GAVL_CUST_NODE_INT_DEC(fcb_contract         /* cust_prefix */,          \
 239                        struct fcb           /* cust_root_t */,          \
 240                        struct fcb_contract  /* cust_item_t */,          \
 241                        fres_contract_id_t   /* cust_key_t */,           \
 242                        contracts            /* cust_root_node */,       \
 243                        node_fcb             /* cust_item_node */,       \
 244                        id                   /* cust_item_key */,        \
 245                        fres_contract_id_cmp /* cust_cmp_fnc */);
 246
 247 #if 1
 248 GAVL_CUST_NODE_INT_IMP(fcb_contract         /* cust_prefix */,          \
 249                        struct fcb           /* cust_root_t */,          \
 250                        struct fcb_contract  /* cust_item_t */,          \
 251                        fres_contract_id_t   /* cust_key_t */,           \
 252                        contracts            /* cust_root_node */,       \
 253                        node_fcb             /* cust_item_node */,       \
 254                        id                   /* cust_item_key */,        \
 255                        fres_contract_id_cmp /* cust_cmp_fnc */);
 256 #else
 257 #include "fcb_contract_gavl.inc"
 258 #endif
 259
 260
 261 #define o2fcb(o) (struct fcb*)forb_instance_data(o)
 262
 263 struct res_key*
 264 get_res_key(const struct fres_contract *contract, struct res_key *key)
 265 {
 266         fres_block_resource *block_res;
 267
 268         block_res = fres_contract_get_resource(contract);
 269         if (!block_res) {
 270                 ul_logerr("No resource specified\n");
 271                 return NULL;
 272         } else {
 273                 key->type = block_res->resource_type;
 274                 key->id = block_res->resource_id;
 275         }
 276         return key;
 277 }
 278
 279 /**
 280  * Fills in an array of fcb_contracts according to requests submited
 281  * to negotiate_contracts(). For every contract in @a contracts, there
 282  * is one fcb_contract in @a fcb_contracts array.
 283  *
 284  * @param fcb
 285  * @param fcb_contracts Where to store pointers to fcb_contracts
 286  * @param resource Resource for which negotiation is being done.
 287  * @param app Application which requests negotiation
 288  * @param contracts Contracts submited to negotiate_contracts()
 289  * @param num Number of contracts in contracts (which is the same as the number in fcb_contracts).
 290  *
 291  * @return Zero on success, non-zero error code on error.
 292  */
 293 static int
 294 prepare_fcb_contracts(struct fcb *fcb, struct fcb_contract *fcb_contracts[],
 295                       struct resource **resource,
 296                       forb_server_id *app,
 297                       struct fres_contract *contracts[], int num)
 298 {
 299         unsigned i;
 300         struct fcb_contract *fc;
 301         struct res_alloc *ra;
 302         struct res_key key, key2 = {-1,-1};
 303
 304         for (i=0; i<num; i++) {
 305                 struct fres_contract *c = contracts[i];
 306
 307                 if (fres_contract_id_is_empty(&c->id)) {
 308                         /* Normal negotiation request */
 309                         forb_uuid_generate((forb_uuid_t *)&c->id);
 310                         fc = fcb_contract_new(&c->id);
 311                         if (!fc)
 312                                 return errno;
 313                         fcb_contracts[i] = fc;
 314                         if (!get_res_key(c, &key)) {
 315                                 return FRSH_ERR_RESOURCE_ID_INVALID;
 316                         }
 317                 } else {
 318                         fc = fcb_contract_find(fcb, &c->id);
 319                         if (!fc) {
 320                                 char str[60];
 321                                 fres_contract_id_to_string(str, &c->id, sizeof(str));
 322                                 ul_logerr("Attempt to change unknown contract %s\n", str);
 323                                 return FRES_ERR_NOTHING_TO_RENEGOTIATE;
 324                         } else {
 325                                 fcb_contracts[i] = fc;
 326                                 if (fres_contract_get_num_blocks(c) == 0) {
 327                                         /* Cancelation */
 328                                         get_res_key(fc->user_contract, &key);
 329                                 } else {
 330                                         /* Renegotiation */
 331                                         if (!get_res_key(c, &key)) {
 332                                                 return FRSH_ERR_RESOURCE_ID_INVALID;
 333                                         }
 334                                 }
 335                         }
 336                 }
 337                 fc->requested_contract = c;
 338
 339                 /* Check that all contracts are for the same resource */
 340                 if (i==0) {
 341                         key2 = key;
 342                         *resource = fcb_resource_find(fcb, &key);
 343                         if (!*resource) {
 344                                 ul_logerr("No resource manager for %d.%d registered\n",
 345                                           key.type, key.id);
 346                                 return FRSH_ERR_RESOURCE_ID_INVALID;
 347                         }
 348                 }
 349                 else if (key.type != key2.type ||
 350                          key.id   != key2.id) {
 351                         ul_logerr("Contract #%d differs in resource!\n", i);
 352                         return FRSH_ERR_RESOURCE_ID_INVALID;
 353                 }
 354
 355                 /* Find allocator for this request */
 356                 ra = fcb_alloc_find(*resource, app);
 357                 if (!ra) {
 358                         char str[60];
 359                         forb_server_id_to_string(str, app, sizeof(str));
 360                         ul_logerr("No resource allocator found for %d.%d and %s\n",
 361                                   (*resource)->key.type, (*resource)->key.id, str);
 362                         return FRES_ERR_NO_RESOURCE_ALLOCATOR;
 363                 }
 364                 fc->ra = ra;
 365         }
 366         return 0;
 367 }
 368
 369 /**
 370  * @param resource Resource for which to rebalance capacity and negotiate new contracts
 371  * @param fcb_contract New requests to negotiate
 372  * @param num The number of elements in @a fcb_contract
 373  * @param[out] rl List with changed contracts to be commited
 374  */
 375 static void
 376 prepare_reservation_list(struct resource *resource,
 377                          struct fcb_contract *fcb_contract[], int num,
 378                          struct reservation_list *rl)
 379 {
 380         int i;
 381         fosa_abs_time_t now;
 382         struct fcb_contract *fc;
 383
 384         reservation_list_init_head(rl);
 385         rl->length = 0;
 386         for (i=0; i<num; i++) {
 387                 assert(fcb_contract[i]->requested_contract != NULL);
 388                 reservation_list_insert(rl, fcb_contract[i]);
 389                 rl->length++;
 390         }
 391         fosa_clock_get_time(CLOCK_REALTIME, &now);
 392         ul_list_for_each(sc_contracts, resource, fc) {
 393                 if (fosa_abs_time_smaller(fc->end_of_stability_period, now) &&
 394                     fc->requested_contract == NULL) /* Do not insert contract inserted above */
 395                 {
 396                         reservation_list_insert(rl, fc);
 397                         rl->length++;
 398                 }
 399         }
 400 }
 401
 402 /**
 403  *
 404  *
 405  * @param resource Resource for which to rebalance capacity and negotiate new contracts
 406  * @param rl List with changed contracts to be commited
 407  *
 408  * @return Zero on success, non-zero error code on error
 409  */
 410 static int
 411 rebalance_spare_capacity_and_reserve(struct resource *resource,
 412                                      struct reservation_list *rl)
 413 {
 414         int ret;
 415         unsigned i;
 416         struct fcb_contract *fc;
 417         fres_block_spare_capacity *s;
 418
 419         fres_contract_ptr_seq contracts;
 420         if (!forb_sequence_alloc_buf(contracts, rl->length)) {
 421                 return errno;
 422         }
 423         contracts._length = rl->length;
 424         i=0;
 425         /* Initialize optimization */
 426         ul_list_for_each(reservation_list, rl, fc) {
 427                 fc->to_be_reserved_contract =
 428                         fc->requested_contract ? fres_contract_duplicate(fc->requested_contract) :
 429                         fc->user_contract ? fres_contract_duplicate(fc->user_contract) :
 430                         NULL;
 431                 assert(fc->to_be_reserved_contract != NULL);
 432
 433                 forb_sequence_elem(contracts, i) = fc->to_be_reserved_contract;
 434                 i++;
 435
 436                 s = fres_contract_get_spare_capacity(fc->to_be_reserved_contract);
 437                 if (s && s->granularity == FRSH_GR_DISCRETE) {
 438                         fc->sc_variant.initial = s->variants._length - 1;
 439                         fc->sc_variant.try = fc->sc_variant.initial;
 440                 }
 441         }
 442
 443         bool all_combinations_tried;
 444         int criterion, best_criterion = -1;
 445         struct fcb_contract *fcb_changed;
 446         /* Exhaustive search. Try all combinations of spare capacity
 447          * variants and find the one with best creterion. */
 448         do {
 449                 all_combinations_tried = true;
 450                 fcb_changed = NULL;
 451                 criterion = 0;
 452                 /* Prepare spare capacity variant */
 453                 ul_list_for_each(reservation_list, rl, fc) {
 454                         s = fres_contract_get_spare_capacity(fc->to_be_reserved_contract);
 455                         /* TODO: Simulate continuous granularity by discretization */
 456                         if (s && s->granularity == FRSH_GR_DISCRETE) {
 457                                 fres_container_copy(fc->to_be_reserved_contract->container,
 458                                                     forb_sequence_elem(s->variants, fc->sc_variant.try));
 459                                 criterion += fc->sc_variant.try;
 460
 461                                 if (fcb_changed == NULL) {
 462                                         /* Chnage tried variant for the next round */
 463                                         fc->sc_variant.try = fc->sc_variant.try > 0 ?
 464                                                 fc->sc_variant.try - 1 :
 465                                                 s->variants._length - 1;
 466                                         /* Do not change other
 467                                          * contracts unless we have
 468                                          * tried all variants */
 469                                         if (fc->sc_variant.try != fc->sc_variant.initial) {
 470                                                 fcb_changed = fc;
 471                                         }
 472                                 }
 473                                 if (fc->sc_variant.try != fc->sc_variant.initial)
 474                                         all_combinations_tried = false;
 475                         }
 476                 }
 477
 478                 if (criterion > best_criterion) {
 479                         CORBA_Environment ev;
 480                         /* Reserve contract */
 481                         ret = fres_resource_manager_reserve_contracts(resource->mng, &contracts, &ev);
 482                         if (forb_exception_occurred(&ev)) {
 483                                 ret = fres_forbex2err(&ev);
 484                                 ul_logerr("FORB exception when reserving contracts\n");
 485                                 goto err;
 486                         }
 487                         if (ret < 0) {
 488                                 ul_logerr("Contract reservation error %d\n", ret);
 489                                 ret = FRES_ERR_ADMISSION_TEST;
 490                                 goto err;
 491                         }
 492                         if (ret == 0) { /* negotiation succeeded */
 493                                 best_criterion = criterion;
 494                         }
 495                 }
 496         } while (!all_combinations_tried);
 497
 498         if (best_criterion == -1) {
 499                 ret = FRSH_ERR_CONTRACT_REJECTED;
 500         } else {
 501                 /* At least some variant succeeded */
 502                 ret = 0;
 503                 sc_contracts_init_head(resource);
 504                 ul_list_for_each(reservation_list, rl, fc) {
 505                         s = fres_contract_get_spare_capacity(fc->to_be_reserved_contract);
 506                         if (s && s->granularity == FRSH_GR_DISCRETE) {
 507                                 sc_contracts_insert(resource, fc);
 508                         }
 509                 }
 510         }
 511 err:
 512         ul_list_for_each(reservation_list, rl, fc) {
 513                 fres_contract_destroy(fc->to_be_reserved_contract);
 514                 fc->to_be_reserved_contract = NULL;
 515         }
 516         forb_sequence_free_buf(contracts, forb_no_destructor);
 517         return ret;
 518 }
 519
 520 /**
 521  * Create/change VReses according to @a schedulable_contracts.
 522  *
 523  * There might be more allocators for schedulable contracts, so merge
 524  * adjacent vreses with the same allocator together and create/change
 525  * them in one step. Also the order of schedulable contracts might be
 526  * different from the initial order od ids in commit_contracts()
 527  * therefore we have to search for every contract.
 528  */
 529 static int
 530 change_vreses(struct fcb *fcb, fres_contract_ptr_seq *schedulable_contracts)
 531 {
 532         struct res_alloc *last_ra = NULL;
 533         fres_contract_ptr_seq vreses;
 534         int i, ret;
 535         CORBA_Environment ev;
 536
 537         if (!forb_sequence_alloc_buf(vreses, schedulable_contracts->_length)) {
 538                 return errno;
 539         }
 540         vreses._length = 0;
 541
 542         for (i=0; i<schedulable_contracts->_length; i++) {
 543                 struct fcb_contract *fc;
 544                 struct fres_contract *sc = schedulable_contracts->_buffer[i];
 545
 546                 fc = fcb_contract_find(fcb, &sc->id);
 547                 assert(fc != NULL);
 548
 549                 if (true /* TODO: if the schedulable contract is changed */) {
 550                         if (last_ra != fc->ra) {
 551                                 if (vreses._length) {
 552                                         ret = fres_resource_allocator_change_vreses(last_ra->ra,
 553                                                                                     &vreses, &ev);
 554                                         if (forb_exception_occurred(&ev)) {
 555                                                 ret = fres_forbex2err(&ev);
 556                                                 goto err;
 557                                         }
 558                                         if (ret) goto err;
 559                                 }
 560                                 vreses._length = 0;
 561                         }
 562                         vreses._buffer[vreses._length] = sc;
 563                         vreses._length++;
 564                         fres_contract_destroy(fc->schedulable_contract);
 565                         fc->schedulable_contract = fres_contract_duplicate(sc);
 566                         last_ra = fc->ra;
 567
 568                 }
 569         }
 570         ret = fres_resource_allocator_change_vreses(last_ra->ra,
 571                                                     &vreses, &ev);
 572         if (forb_exception_occurred(&ev))
 573                 ret = fres_forbex2err(&ev);
 574 err:
 575         forb_sequence_free_buf(vreses, forb_no_destructor);
 576         return ret;
 577 }
 578
 579 CORBA_long
 580 negotiate_contracts(fres_contract_broker obj,
 581                     const fres_contract_ptr_seq* contracts,
 582                     fres_contract_id_seq** ids_out,
 583                     CORBA_Environment *ev)
 584 {
 585         struct fcb *fcb = o2fcb(obj);
 586         struct resource *resource = NULL;
 587         int ret = 0;
 588         forb_server_id app;
 589         fres_contract_ptr_seq *schedulable_contracts;
 590         struct fcb_contract **fcb_contracts, *fc;
 591         unsigned i;
 592         fres_contract_id_seq commit_ids;
 593
 594         /* Prepare output sequence for the case we return eariler with
 595          * an error */
 596         forb_sequence_alloc(*ids_out, 0);
 597         if (!*ids_out) {
 598                 ev->major = FORB_EX_NO_MEMORY;
 599                 goto err;
 600         }
 601         CORBA_sequence_set_release(*ids_out, CORBA_TRUE);
 602
 603         forb_get_req_source(obj, &app);
 604
 605         fcb_contracts = malloc(sizeof(fcb_contracts[0])*contracts->_length);
 606         if (!fcb_contracts) {
 607                 ret = errno;
 608                 goto err;
 609         }
 610         memset(fcb_contracts, 0, sizeof(fcb_contracts[0])*contracts->_length);
 611
 612         ret = prepare_fcb_contracts(fcb, fcb_contracts, &resource,
 613                                     &app, contracts->_buffer, contracts->_length);
 614         if (ret)
 615                 goto err_free_fcb_contracts;
 616
 617         struct reservation_list rl;
 618         prepare_reservation_list(resource,
 619                                  fcb_contracts, contracts->_length,
 620                                  &rl);
 621
 622         /* Allocate all the needed memory before doing reservation. If
 623          * there is no enough memory, it has no sense to call resource
 624          * manager. */
 625         if (!forb_sequence_alloc_buf(commit_ids, rl.length)) {
 626                 ret = errno;
 627                 goto err_free_fcb_contracts;
 628         }
 629
 630         /* Reserve contracts */
 631         ret = rebalance_spare_capacity_and_reserve(resource, &rl);
 632         if (ret) {
 633                 if (ret == FRSH_ERR_CONTRACT_REJECTED) {
 634                         ul_logmsg("Contract(s) was/were rejected\n");
 635                 } else {
 636                         char msg[100];
 637                         fres_strerror(ret, msg, sizeof(msg));
 638                         ul_logerr("Reservation error: %s\n", msg);
 639                 }
 640                 goto err_free_fcb_contracts;
 641         }
 642
 643         /* Commit contracts */
 644         commit_ids._length = rl.length;
 645         i=0;
 646         ul_list_for_each(reservation_list, &rl, fc) {
 647                 forb_sequence_elem(commit_ids, i) = fc->id;
 648                 i++;
 649         }
 650
 651         fres_resource_manager_commit_contracts(resource->mng, &commit_ids,
 652                                                &schedulable_contracts, ev);
 653         if (forb_exception_occurred(ev)) {
 654                 ret = fres_forbex2err(ev);
 655                 goto err_cancel_reservation;
 656         }
 657
 658         /* Add new contracts to our fcb database for later
 659          * reference. Canceled contracts are removed below. */
 660         for (i=0; i<contracts->_length; i++) {
 661                 fc =  fcb_contracts[i];
 662
 663                 if (fc->user_contract) {
 664                         if (fres_contract_get_num_blocks(fc->requested_contract) > 0) {
 665                                 /* Renegotiation */
 666                                 fres_contract_destroy(fc->user_contract);
 667                                 fc->user_contract = fres_contract_duplicate(fc->requested_contract);
 668                                 /* Note: requested_contract is also
 669                                  * pointed by contracts parameter and
 670                                  * will be freed by FORB. */
 671                                 fc->requested_contract = NULL;
 672                         }
 673                 } else {
 674                         /* Insert new contracts */
 675                         fcb_contract_insert(fcb, fcb_contracts[i]);
 676                         fc->user_contract = fres_contract_duplicate(fc->requested_contract);
 677                         fc->requested_contract = NULL;
 678                         /* See the note above. */
 679                 }
 680         }
 681
 682         ret = change_vreses(fcb, schedulable_contracts);
 683         if (ret)
 684                 goto err_cancel_reservation;
 685
 686         forb_sequence_free(schedulable_contracts, fres_contract_ptr_destroy);
 687         if (ret != 0) {
 688                 ul_logerr("VRes was not created\n");
 689                 goto err_cancel_contracts;
 690         }
 691
 692
 693         /* Return IDs and delete canceled contracts from out database */
 694         if (!forb_sequence_alloc_buf(**ids_out, contracts->_length)) {
 695                 ev->major = FORB_EX_NO_MEMORY;
 696                 goto err_cancel_contracts;
 697         }
 698         (*ids_out)->_length = contracts->_length;
 699         for (i=0; i<contracts->_length; i++) {
 700                 struct fcb_contract *fc =  fcb_contracts[i];
 701                 forb_sequence_elem(**ids_out, i) = fc->id;
 702
 703                 if (fc->requested_contract &&
 704                     fres_contract_get_num_blocks(fc->requested_contract) == 0) {
 705                         fcb_contract_delete(fcb, fc);
 706                         /* Note: requested_contract is also pointed by
 707                          * contracts parameter and will be freed by FORB. */
 708                         fc->requested_contract = NULL;
 709                         fcb_contract_destroy(fc);
 710                 }
 711         }
 712         return 0;
 713
 714 err_cancel_contracts:
 715         /* TODO */
 716         goto err_free_fcb_contracts;
 717 err_cancel_reservation:
 718         fres_resource_manager_cancel_reservations(resource->mng, &commit_ids, ev);
 719 err_free_fcb_contracts:
 720         for (i=0; i<contracts->_length; i++) {
 721                 fc = fcb_contracts[i];
 722                 if (fc && !fc->user_contract) {
 723                         fcb_contracts[i]->requested_contract = NULL; /* Destroyed by FORB */
 724                         fcb_contract_destroy(fcb_contracts[i]);
 725                 }
 726         }
 727         free(fcb_contracts);
 728 err:
 729         return ret;
 730 }
 731
 732 void redistribute_spare_capacity(fres_contract_broker obj,
 733                                  const frsh_resource_type_t restype,
 734                                  const frsh_resource_id_t resid,
 735                                  CORBA_Environment *ev)
 736 {
 737         struct fcb *fcb = o2fcb(obj);
 738         struct res_key key = {restype, resid };
 739         struct resource *resource;
 740         struct reservation_list rl;
 741
 742         resource = fcb_resource_find(fcb, &key);
 743
 744         prepare_reservation_list(resource, NULL, 0, &rl);
 745
 746 /*      forb_sequence_alloc(ids, rl.length); */
 747 /*      if (!ids || !ids->_buffer) { */
 748 /*              ev->major = FORB_EX_NO_MEMORY; */
 749 /*              goto err_free_fcb_contracts; */
 750 /*      } */
 751 /*      CORBA_sequence_set_release(ids, CORBA_TRUE); */
 752 /*      *ids_out = ids;         /\* ids is freed by FORB *\/ */
 753
 754
 755         rebalance_spare_capacity_and_reserve(resource, &rl);
 756         /* Commit */
 757 }
 758
 759 CORBA_long register_resource(fres_contract_broker obj,
 760                             const frsh_resource_type_t restype,
 761                             const frsh_resource_id_t resid,
 762                             const fres_resource_desc *desc,
 763                             CORBA_Environment *ev)
 764 {
 765         struct fcb *fcb = o2fcb(obj);
 766         struct resource *res, *res2;
 767
 768         res = malloc(sizeof(*res));
 769         if (!res) goto err;
 770         memset(res, 0, sizeof(*res));
 771         res->key.type = restype;
 772         res->key.id = resid;
 773         res2 = fcb_resource_find(fcb, &res->key);
 774         if (res2) {
 775                 if (forb_object_is_stale(res2->mng)) {
 776                         ul_logmsg("Removing stale manager for resource %d.%d\n",
 777                                   restype, resid);
 778                         forb_object_release(res2->mng);
 779                         fcb_resource_delete(fcb, res2);
 780                         /* TODO: Delete also all allocators associated
 781                          * with this stale resource manager. */
 782                         free(res);
 783                         res = res2;
 784                 } else {
 785                         ul_logerr("Resource manager %d.%d already registered\n",
 786                                   restype, resid);
 787                         goto free_err;
 788                 }
 789         }
 790         res->mng = forb_object_duplicate(desc->manager);
 791         res->name = desc->name;
 792
 793         fcb_alloc_init_root_field(res);
 794         sc_contracts_init_head(res);
 795         ul_logmsg("Registering manager for resource \"%s\" (%d.%d)\n",
 796                   res->name, restype, resid);
 797         fcb_resource_insert(fcb, res);
 798         return 0;
 799 free_err:
 800         free(res);
 801 err:
 802         return -1;
 803 }
 804
 805
 806 CORBA_long register_allocator(fres_contract_broker obj,
 807                               const frsh_resource_type_t restype,
 808                               const frsh_resource_id_t resid,
 809                               const fres_resource_allocator ra_obj,
 810                               CORBA_Environment *ev)
 811 {
 812         struct fcb *fcb = o2fcb(obj);
 813         struct resource *res;
 814         struct res_alloc *ra;
 815         struct res_key resource;
 816         forb_server_id server_id;
 817         char server_id_str[40];
 818
 819         forb_get_server_id(ra_obj, &server_id);
 820         forb_server_id_to_string(server_id_str, &server_id, sizeof(server_id_str));
 821         ul_logmsg("Registering allocator for resource %d.%d in app %s\n",
 822                   restype, resid, server_id_str);
 823
 824         resource.type = restype;
 825         resource.id = resid;
 826         res = fcb_resource_find(fcb, &resource);
 827         if (!res) {
 828                 ul_logerr("No manager found for %d.%d. Unable to register the allocator!\n",
 829                           restype, resid);
 830                 goto err;
 831         }
 832         ra = fcb_alloc_find(res, &server_id);
 833         if (ra) {
 834                 char *str = forb_object_to_string(ra_obj);
 835                 ul_logerr("Allocator from already registered (%s)\n",
 836                           str);
 837                 forb_free(str);
 838                 goto err;
 839         }
 840         ra = malloc(sizeof(*ra));
 841         if (!ra) {
 842                 goto err;
 843         }
 844         ra->app = server_id;
 845         ra->ra = forb_object_duplicate(ra_obj);
 846         fcb_alloc_insert(res, ra);
 847         return 0;
 848 err:
 849         return FRSH_ERR_RESOURCE_ID_INVALID;
 850 }
 851
 852 void get_resources(fres_contract_broker obj, fres_resource_seq** resources, CORBA_Environment *ev)
 853 {
 854         struct fcb *fcb = o2fcb(obj);
 855         fres_resource_seq *seq;
 856         struct resource *res;
 857         int n;
 858
 859         seq = malloc(sizeof(*seq));
 860         if (!seq) {
 861                 ev->major = FORB_EX_NO_MEMORY;
 862                 return;
 863         }
 864         memset(seq, 0, sizeof(*seq));
 865
 866         n=0;
 867         gavl_cust_for_each(fcb_resource, fcb, res) {
 868                 n++;
 869         }
 870
 871         seq->_buffer = CORBA_sequence_fres_resource_allocbuf(n);
 872         seq->_maximum = n;
 873         seq->_length = n;
 874
 875         n = 0;
 876         gavl_cust_for_each(fcb_resource, fcb, res) {
 877                 seq->_buffer[n].restype = res->key.type;
 878                 seq->_buffer[n].resid = res->key.id;
 879                 seq->_buffer[n].desc.manager = res->mng;
 880                 seq->_buffer[n].desc.name = res->name;
 881                 n++;
 882         }
 883
 884         *resources = seq;
 885 }
 886
 887 #if CONFIG_FCB_INET && !CONFIG_FORB_PROTO_INET_DEFAULT
 888 static int register_inet_port(forb_orb orb)
 889 {
 890         forb_port_t *port = forb_malloc(sizeof(*port));
 891         int ret;
 892         struct in_addr listen_on;
 893
 894         if (!port)
 895                 return -1;
 896         memset(port, 0, sizeof(*port));
 897         listen_on.s_addr = INADDR_ANY;
 898         ret = forb_inet_port_init(&port->desc, listen_on);
 899         if (ret)
 900                 return ret;
 901         ret = forb_register_port(orb, port);
 902         return ret;
 903 }
 904 #endif
 905
 906 struct forb_fres_contract_broker_impl impl = {
 907         .negotiate_contracts = negotiate_contracts,
 908         .register_resource = register_resource,
 909         .register_allocator = register_allocator,
 910         .redistribute_spare_capacity = redistribute_spare_capacity,
 911         .get_resources = get_resources,
 912 };
 913
 914 void peer_discovery_callback(const forb_orb peer_orb, const char *orb_id)
 915 {
 916         forb_server_id server_id;
 917         char server_id_str[sizeof(forb_server_id)*2+1];
 918
 919         forb_get_server_id(peer_orb, &server_id);
 920         forb_server_id_to_string(server_id_str, &server_id, sizeof(server_id_str));
 921 #if 0
 922         ul_logmsg("peer discovered: %s (orb_id '%s')\n", server_id_str, orb_id);
 923 #endif
 924
 925         if (orb_id == NULL)
 926                 return;
 927
 928         if (strcmp(orb_id, "org.frescor.fcb") == 0) {
 929                 fosa_abs_time_t now;
 930                 fosa_rel_time_t delay;
 931                 fosa_clock_get_time(CLOCK_REALTIME, &now);
 932                 delay = fosa_abs_time_extract_interval(start_time, now);
 933                 ul_logmsg("node joined: %s (time %ld ms)\n", server_id_str,
 934                           fosa_rel_time_to_msec(delay));
 935         }
 936 }
 937
 938 void peer_dead_callback(const forb_orb peer_orb, const char *orb_id)
 939 {
 940 }
 941
 942 int main(int argc, char *argv[])
 943 {
 944         forb_orb orb;
 945         struct fcb fcb_data;
 946         fres_contract_broker fcb;
 947         forb_executor_t executor;
 948         int ret;
 949         forb_init_attr_t attr = {
 950                 .orb_id = "org.frescor.fcb",
 951                 .peer_discovery_callback = peer_discovery_callback,
 952                 .peer_dead_callback = peer_dead_callback,
 953         };
 954
 955
 956         fosa_clock_get_time(CLOCK_REALTIME, &start_time);
 957
 958         orb = forb_init(&argc, &argv, &attr);
 959         if (!orb) error(1, errno, "FORB initialization failed");
 960
 961 #if CONFIG_FCB_INET && !CONFIG_FORB_PROTO_INET_DEFAULT
 962         ret = register_inet_port(orb);
 963         if (ret) error(0, errno, "INET port registration failed");
 964 #endif
 965
 966         fcb_resource_init_root_field(&fcb_data);
 967         fcb_contract_init_root_field(&fcb_data);
 968
 969         fcb = forb_fres_contract_broker_new(orb, &impl, &fcb_data);
 970         if (!fcb) error(1, errno, "forb_fres_contract_broker_new failed");
 971
 972         /* Prepare executor before we register the fcb reference,
 973          * so that no reuqests are lost */
 974         ret = forb_executor_init(&executor);
 975         if (ret) error(1, errno, "forb_executor_init failed");
 976
 977         ret = forb_executor_register_object(&executor, fcb);
 978         if (ret) error(1, errno, "forb_executor_register_object failed");
 979
 980         ret = forb_register_reference(fcb, fres_contract_broker_reg_name);
 981         if (ret) error(1, errno, "forb_register_reference() failed");
 982
 983         ul_logmsg("Waiting for requests\n");
 984         ret = forb_executor_run(&executor);
 985         if (ret) error(1, errno, "forb_executor_run failed");
 986
 987         return 0;
 988 }