frsh_api/frsh_distributed.c

   1 // -----------------------------------------------------------------------
   2 //  Copyright (C) 2006 - 2008 FRESCOR consortium partners:
   3 //
   4 //    Universidad de Cantabria,              SPAIN
   5 //    University of York,                    UK
   6 //    Scuola Superiore Sant'Anna,            ITALY
   7 //    Kaiserslautern University,             GERMANY
   8 //    Univ. Politécnica  Valencia,           SPAIN
   9 //    Czech Technical University in Prague,  CZECH REPUBLIC
  10 //    ENEA                                   SWEDEN
  11 //    Thales Communication S.A.              FRANCE
  12 //    Visual Tools S.A.                      SPAIN
  13 //    Rapita Systems Ltd                     UK
  14 //    Evidence                               ITALY
  15 //
  16 //    See http://www.frescor.org for a link to partners' websites
  17 //
  18 //           FRESCOR project (FP6/2005/IST/5-034026) is funded
  19 //        in part by the European Union Sixth Framework Programme
  20 //        The European Union is not liable of any use that may be
  21 //        made of this code.
  22 //
  23 //
  24 //  based on previous work (FSF) done in the FIRST project
  25 //
  26 //   Copyright (C) 2005  Mälardalen University, SWEDEN
  27 //                       Scuola Superiore S.Anna, ITALY
  28 //                       Universidad de Cantabria, SPAIN
  29 //                       University of York, UK
  30 //
  31 //   FSF API web pages: http://marte.unican.es/fsf/docs
  32 //                      http://shark.sssup.it/contrib/first/docs/
  33 //
  34 //   This file is part of FRSH (FRescor ScHeduler)
  35 //
  36 //  FRSH is free software; you can redistribute it and/or modify it
  37 //  under terms of the GNU General Public License as published by the
  38 //  Free Software Foundation; either version 2, or (at your option) any
  39 //  later version.  FRSH is distributed in the hope that it will be
  40 //  useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  41 //  of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  42 //  General Public License for more details. You should have received a
  43 //  copy of the GNU General Public License along with FRSH; see file
  44 //  COPYING. If not, write to the Free Software Foundation, 675 Mass Ave,
  45 //  Cambridge, MA 02139, USA.
  46 //
  47 //  As a special exception, including FRSH header files in a file,
  48 //  instantiating FRSH generics or templates, or linking other files
  49 //  with FRSH objects to produce an executable application, does not
  50 //  by itself cause the resulting executable application to be covered
  51 //  by the GNU General Public License. This exception does not
  52 //  however invalidate any other reasons why the executable file might be
  53 //  covered by the GNU Public License.
  54 // -----------------------------------------------------------------------
  55 //==============================================
  56 //  ******** *******    ********  **      **
  57 //  **///// /**////**  **//////  /**     /**
  58 //  **      /**   /** /**        /**     /**
  59 //  ******* /*******  /********* /**********
  60 //  **////  /**///**  ////////** /**//////**
  61 //  **      /**  //**        /** /**     /**
  62 //  **      /**   //** ********  /**     /**
  63 //  //       //     // ////////   //      //
  64 //
  65 // FRSH(FRescor ScHeduler), pronounced "fresh"
  66 //==============================================
  67 //
  68 //
  69
  70 #include <frsh.h>
  71 /* #include <frsh_internal.h> */
  72
  73 #if FRSH_DISTRIBUTED_MODULE_SUPPORTED
  74
  75 #include "fosa.h"
  76 #include "fna_configuration.h" // for fna_operations, FNA_MAX_NETWORKS
  77 //#include "frsh_debug_and_trace.h" // for FRSH_TRACE
  78 #include <string.h>
  79
  80 #define FRSH_ENDPOINT_SIZE sizeof(fna_endpoint_data_t)
  81
  82 static inline void* endpoint_alloc()
  83 {
  84         return calloc(1, FRSH_ENDPOINT_SIZE);
  85 }
  86
  87 static inline void endpoint_free(void* endpoint)
  88 {
  89         return free(endpoint);
  90 }
  91
  92 //-------------------------//
  93 // check_valid_resource_id //
  94 //-------------------------//
  95 // Internal function to check that a given resource_id is valid
  96
  97 static inline int check_valid_resource_id (const frsh_resource_id_t resource_id)
  98 {
  99     if (resource_id >= FNA_MAX_NETWORKS || resource_id < 0) {
 100         return FRSH_ERR_RESOURCE_ID_INVALID;
 101     }
 102
 103     if (fna_operations[resource_id] == NULL) {
 104         return FRSH_ERR_RESOURCE_ID_INVALID;
 105     }
 106
 107     return 0;
 108 }
 109
 110 //---------------------------//
 111 // check_valid_send_endpoint //
 112 //---------------------------//
 113 // Internal function to check that a given send_endpoint is valid. if it is
 114 // valid it returns the position, if not it returns -1
 115
 116 static inline int check_valid_send_endpoint(frsh_send_endpoint_t endpoint)
 117 {
 118     if ((!endpoint) || (endpoint->endpoint_type != FRSH_SEND_ENDPOINT_TYPE)) {
 119         return -1;
 120     }
 121
 122     return 0;
 123 }
 124
 125 //------------------------------//
 126 // check_valid_receive_endpoint //
 127 //------------------------------//
 128 // Internal function to check that a given receive_endpoint is valid. if it is
 129 // valid it returns the position, if not it returns -1
 130
 131 static inline int check_valid_receive_endpoint(frsh_receive_endpoint_t endpoint)
 132 {
 133     if ((!endpoint) || (endpoint->endpoint_type != FRSH_RECEIVE_ENDPOINT_TYPE)) {
 134         return -1;
 135     }
 136
 137     return 0;
 138 }
 139
 140 //-----------------------//
 141 // frsh_distributed_init //
 142 //-----------------------//
 143 // TODO: use a flag to say if it is initialized or not
 144
 145 int frsh_distributed_init(void)
 146 {
 147     frsh_resource_id_t id = 0;
 148     int err = 0;
 149
 150     // execute the fna_init operation of each installed network
 151     for (id=0; id<FNA_MAX_NETWORKS; id++) {
 152         if (fna_operations[id] != NULL) {
 153             //FRSH_TRACE(FRSH_TRACE_DISTRIBUTED, "init network id=%d\n", id);
 154             if (!fna_operations[id]->fna_init) continue;
 155             err = fna_operations[id]->fna_init(id);
 156             if (err != 0) return -1;
 157         }
 158     }
 159
 160     // initialize the freelist to handle the set of endpoints
 161     //FRSH_TRACE(FRSH_TRACE_DISTRIBUTED, "init endpoints freelist\n");
 162
 163     return 0;
 164 }
 165
 166 //-----------------------------------//
 167 // frsh_network_get_max_message_size //
 168 //-----------------------------------//
 169
 170 int frsh_network_get_max_message_size
 171         (const frsh_resource_id_t resource_id,
 172          const frsh_network_address_t destination,
 173          size_t *max_size)
 174 {
 175     if (check_valid_resource_id(resource_id)) {
 176             return FRSH_ERR_RESOURCE_ID_INVALID;
 177     }
 178
 179     if (fna_operations[resource_id]->fna_network_get_max_message_size == NULL) {
 180             return FRSH_ERR_INTERNAL_ERROR;
 181     }
 182
 183     return fna_operations[resource_id]->fna_network_get_max_message_size
 184                 (resource_id, destination, max_size);
 185 }
 186
 187 //------------------------------//
 188 // frsh_network_bytes_to_budget //
 189 //------------------------------//
 190
 191 int frsh_network_bytes_to_budget
 192         (const frsh_resource_id_t resource_id,
 193          const size_t nbytes,
 194          frsh_rel_time_t *budget)
 195 {
 196     if (check_valid_resource_id(resource_id)) {
 197         return FRSH_ERR_RESOURCE_ID_INVALID;
 198     }
 199
 200     if (fna_operations[resource_id]->fna_network_bytes_to_budget == NULL) {
 201         return FRSH_ERR_INTERNAL_ERROR;
 202     }
 203
 204     return fna_operations[resource_id]->fna_network_bytes_to_budget
 205             (resource_id, nbytes, budget);
 206 }
 207
 208 //------------------------------//
 209 // frsh_network_budget_to_bytes //
 210 //------------------------------//
 211
 212 int frsh_network_budget_to_bytes
 213         (const frsh_resource_id_t resource_id,
 214          const frsh_rel_time_t *budget,
 215          size_t *nbytes)
 216 {
 217     if (check_valid_resource_id(resource_id)) {
 218         return FRSH_ERR_RESOURCE_ID_INVALID;
 219     }
 220
 221     if (fna_operations[resource_id]->fna_network_budget_to_bytes == NULL) {
 222         return FRSH_ERR_INTERNAL_ERROR;
 223     }
 224
 225     return fna_operations[resource_id]->fna_network_budget_to_bytes
 226             (resource_id, budget, nbytes);
 227 }
 228
 229 //---------------------------------------//
 230 // frsh_network_get_min_effective_budget //
 231 //---------------------------------------//
 232
 233 int frsh_network_get_min_effective_budget
 234         (const frsh_resource_id_t resource_id,
 235          frsh_rel_time_t *budget)
 236 {
 237     if (check_valid_resource_id(resource_id)) {
 238         return FRSH_ERR_RESOURCE_ID_INVALID;
 239     }
 240
 241     if (fna_operations[resource_id]->fna_network_get_min_eff_budget == NULL) {
 242         return FRSH_ERR_INTERNAL_ERROR;
 243     }
 244
 245     return fna_operations[resource_id]->fna_network_get_min_eff_budget
 246             (resource_id, budget);
 247 }
 248
 249 //---------------------------//
 250 // frsh_send_endpoint_create //
 251 //---------------------------//
 252
 253 int frsh_send_endpoint_create
 254         (frsh_resource_id_t resource_id,
 255          frsh_network_address_t destination,
 256          frsh_stream_id_t stream_id,
 257          frsh_send_endpoint_protocol_info_t protocol_info,
 258          frsh_send_endpoint_t  *endpoint)
 259 {
 260     frsh_send_endpoint_t epoint;
 261
 262     if (check_valid_resource_id(resource_id)) {
 263         return FRSH_ERR_RESOURCE_ID_INVALID;
 264     }
 265
 266     if (endpoint == NULL) {
 267         return FRSH_ERR_BAD_ARGUMENT;
 268     }
 269
 270     // allocate a free endpoint position in the table
 271     epoint = endpoint_alloc();
 272
 273     if (epoint == NULL) {
 274         return FRSH_ERR_NO_SPACE;
 275     }
 276
 277     // initialize the send endpoint
 278     epoint->endpoint_type = FRSH_SEND_ENDPOINT_TYPE;
 279     epoint->resource_id = resource_id;
 280     epoint->destination = destination;
 281     epoint->stream_id = stream_id;
 282     epoint->endpoint_protocol_info.send = protocol_info;
 283     epoint->is_bound = false;
 284     /*FRSH_TRACE(FRSH_TRACE_DISTRIBUTED,"%s:stream_id=%d\n", __func__, stream_id);*/
 285
 286     *endpoint = epoint;
 287     if (fna_operations[resource_id]->fna_send_endpoint_created) {
 288             return fna_operations[resource_id]->fna_send_endpoint_created(epoint);
 289     }
 290     else
 291             return 0;
 292 }
 293
 294 // int frsh_contract_set_queueing_info(frsh_endpoint_queueing_info_t queueing_info,
 295 //                                     frsh_contract_t *contract);
 296 //
 297 // int frsh_contract_get_queueing_info(const frsh_contract_t *contract,
 298 //                                     frsh_endpoint_queueing_info_t *queueing_info);
 299 //
 300 // int frsh_contract_set_protocol_info(frsh_protocol_info_t protocol_info,
 301 //                                     frsh_contract_t *contract);
 302 //
 303 // int frsh_contract_get_protocol_info(frsh_contract_t *contract,
 304 //                                     frsh_protocol_info_t *protocol_info);
 305 //
 306
 307
 308 //-------------------------------//
 309 // frsh_send_endpoint_get_params //
 310 //-------------------------------//
 311
 312 int frsh_send_endpoint_get_params
 313         (const frsh_send_endpoint_t endpoint,
 314          frsh_resource_id_t *resource_id,
 315          frsh_network_address_t *destination,
 316          frsh_stream_id_t *stream,
 317          frsh_send_endpoint_protocol_info_t *protocol_info)
 318 {
 319     if (check_valid_send_endpoint(endpoint) < 0) {
 320             return FRSH_ERR_BAD_ARGUMENT;
 321     }
 322
 323     *resource_id = endpoint->resource_id;
 324     *destination = endpoint->destination;
 325     *stream = endpoint->stream_id;
 326     *protocol_info = endpoint->endpoint_protocol_info.send;
 327
 328     return 0;
 329 }
 330 //----------------------------//
 331 // frsh_send_endpoint_destroy //
 332 //----------------------------//
 333
 334 int frsh_send_endpoint_destroy
 335      (frsh_send_endpoint_t  endpoint)
 336 {
 337     fna_endpoint_data_t *epoint = endpoint;
 338     int rv;
 339
 340     if (check_valid_send_endpoint(endpoint) < 0) {
 341             return FRSH_ERR_BAD_ARGUMENT;
 342     }
 343
 344     if (fna_operations[epoint->resource_id]->fna_endpoint_destroy) {
 345             rv = fna_operations[epoint->resource_id]->\
 346                         fna_endpoint_destroy(endpoint);
 347             if (rv)
 348                     return rv;
 349     }
 350
 351     endpoint_free(endpoint);
 352     return 0;
 353 }
 354
 355 //-------------------------//
 356 // frsh_send_endpoint_bind //
 357 //-------------------------//
 358
 359 int frsh_send_endpoint_bind
 360         (frsh_vres_id_t      vres,
 361          frsh_send_endpoint_t  endpoint)
 362 {
 363     if (check_valid_send_endpoint(endpoint) < 0) {
 364             return FRSH_ERR_BAD_ARGUMENT;
 365     }
 366
 367     // TODO: check that resource_id and resource_type matches
 368     endpoint->vres = vres;
 369     endpoint->is_bound = true;
 370     /*FRSH_TRACE(FRSH_TRACE_DISTRIBUTED,"%s:vres=%d\n", __func__,
 371                endpoint_data[pos].vres);*/
 372
 373     if (fna_operations[endpoint->resource_id]->fna_send_endpoint_bind) {
 374             return fna_operations[endpoint->resource_id]->\
 375                         fna_send_endpoint_bind(endpoint,vres);
 376     }
 377     else
 378             return 0;
 379 }
 380
 381 //---------------------------//
 382 // frsh_send_endpoint_unbind //
 383 //---------------------------//
 384
 385 int frsh_send_endpoint_unbind
 386   (frsh_send_endpoint_t  endpoint)
 387 {
 388     if (check_valid_send_endpoint(endpoint) < 0) {
 389             return FRSH_ERR_BAD_ARGUMENT;
 390     }
 391
 392     endpoint->is_bound = false;
 393
 394     if (fna_operations[endpoint->resource_id]->fna_send_endpoint_unbind) {
 395             return fna_operations[endpoint->resource_id]->\
 396                         fna_send_endpoint_unbind(endpoint);
 397     }
 398     else
 399             return 0;
 400 }
 401
 402 //--------------------------------//
 403 // frsh_send_endpoint_get_vres_id //
 404 //--------------------------------//
 405
 406 int frsh_send_endpoint_get_vres_id
 407         (const frsh_send_endpoint_t  endpoint,
 408          frsh_vres_id_t            *vres)
 409 {
 410     if (check_valid_send_endpoint(endpoint) < 0) {
 411             return FRSH_ERR_BAD_ARGUMENT;
 412     }
 413
 414     *vres = endpoint->vres;
 415     return 0;
 416 }
 417
 418 //-----------------//
 419 // frsh_send_async //
 420 //-----------------//
 421
 422 int frsh_send_async
 423         (const frsh_send_endpoint_t  endpoint,
 424          const void                  *msg,
 425          const size_t                size)
 426 {
 427     frsh_resource_id_t resource_id;
 428
 429     if (check_valid_send_endpoint(endpoint) < 0) {
 430             return FRSH_ERR_BAD_ARGUMENT;
 431     }
 432
 433     if (endpoint->is_bound == false) {
 434             return FRSH_ERR_NOT_BOUND;
 435     }
 436
 437     resource_id = endpoint->resource_id;
 438
 439     if (fna_operations[resource_id]->fna_send_async == NULL) {
 440             return FRSH_ERR_INTERNAL_ERROR;
 441     }
 442
 443     return fna_operations[resource_id]->fna_send_async
 444                 (endpoint, msg, size);
 445 }
 446
 447 //-----------------//
 448 // frsh_send_sync //
 449 //-----------------//
 450
 451 int frsh_send_sync
 452                 (const frsh_send_endpoint_t  endpoint,
 453                  const void                  *msg,
 454                  const size_t                size)
 455 {
 456     frsh_resource_id_t resource_id;
 457
 458     if (check_valid_send_endpoint(endpoint) < 0) {
 459             return FRSH_ERR_BAD_ARGUMENT;
 460     }
 461
 462     if (endpoint->is_bound == false) {
 463             return FRSH_ERR_NOT_BOUND;
 464     }
 465
 466     resource_id = endpoint->resource_id;
 467
 468     if (fna_operations[resource_id]->fna_send_sync == NULL) {
 469             return FRSH_ERR_INTERNAL_ERROR;
 470     }
 471
 472     return fna_operations[resource_id]->fna_send_sync
 473                         (endpoint, msg, size);
 474 }
 475
 476 // int frsh_send_endpoint_get_status(const frsh_send_endpoint_t endpoint,
 477 //                                   int *number_pending_msg,
 478 //                                   frsh_endpoint_network_status_t *network_status,
 479 //                                   frsh_protocol_status_t *protocol_status);
 480 //
 481
 482 //------------------------------//
 483 // frsh_receive_endpoint_create //
 484 //------------------------------//
 485
 486 int frsh_receive_endpoint_create
 487         (frsh_resource_id_t        resource_id,
 488          frsh_stream_id_t          stream_id,
 489          frsh_endpoint_queueing_info_t queueing_info,
 490          frsh_receive_endpoint_protocol_info_t protocol_info,
 491          frsh_receive_endpoint_t  *endpoint)
 492 {
 493     frsh_receive_endpoint_t epoint;
 494
 495     if (check_valid_resource_id(resource_id)) {
 496         return FRSH_ERR_RESOURCE_ID_INVALID;
 497     }
 498
 499     if (endpoint == NULL) {
 500         return FRSH_ERR_BAD_ARGUMENT;
 501     }
 502
 503     if ((epoint = endpoint_alloc()) == NULL) {
 504         return FRSH_ERR_NO_SPACE;
 505     }
 506
 507     // initialize the receive_endpoint
 508     epoint->endpoint_type = FRSH_RECEIVE_ENDPOINT_TYPE;
 509     epoint->resource_id = resource_id;
 510     epoint->stream_id = stream_id;
 511     epoint->queue_info = queueing_info;
 512     epoint->endpoint_protocol_info.receive = protocol_info;
 513
 514     //FRSH_TRACE(FRSH_TRACE_DISTRIBUTED,"%s:stream_id=%d\n", __func__, stream_id);
 515     // the receive_endpoint identifier for the user is the position in the table
 516     *endpoint = epoint;
 517
 518     if (fna_operations[resource_id]->fna_receive_endpoint_created) {
 519             return fna_operations[resource_id]->\
 520                         fna_receive_endpoint_created(epoint);
 521     }
 522     else
 523             return 0;
 524 }
 525
 526 //----------------------------------//
 527 // frsh_receive_endpoint_get_params //
 528 //----------------------------------//
 529
 530 int frsh_receive_endpoint_get_params
 531         (const frsh_receive_endpoint_t  endpoint,
 532          frsh_resource_id_t        *resource_id,
 533          frsh_stream_id_t          *stream,
 534          frsh_endpoint_queueing_info_t   *queueing_info,
 535          frsh_receive_endpoint_protocol_info_t   *protocol_info)
 536 {
 537     if (check_valid_receive_endpoint(endpoint) < 0) {
 538             return FRSH_ERR_BAD_ARGUMENT;
 539     }
 540
 541     *resource_id = endpoint->resource_id;
 542     *stream = endpoint->stream_id;
 543     *queueing_info = endpoint->queue_info;
 544     *protocol_info = endpoint->endpoint_protocol_info.receive;
 545
 546     return 0;
 547 }
 548
 549 //-------------------------------//
 550 // frsh_receive_endpoint_destroy //
 551 //-------------------------------//
 552
 553 int frsh_receive_endpoint_destroy
 554         (frsh_receive_endpoint_t  endpoint)
 555 {
 556     fna_endpoint_data_t *epoint = endpoint;
 557     int rv;
 558
 559     if (check_valid_receive_endpoint(endpoint) < 0) {
 560             return FRSH_ERR_BAD_ARGUMENT;
 561     }
 562
 563     if (fna_operations[epoint->resource_id]->fna_endpoint_destroy) {
 564             rv = fna_operations[epoint->resource_id]->\
 565                         fna_endpoint_destroy(endpoint);
 566             if (rv)
 567                     return rv;
 568     }
 569
 570     endpoint_free(endpoint);
 571     return 0;
 572 }
 573
 574 //-------------------//
 575 // frsh_receive_sync //
 576 //-------------------//
 577
 578 int frsh_receive_sync
 579         (const frsh_receive_endpoint_t  endpoint,
 580          void                           *buffer,
 581          size_t                         buffer_size,
 582          size_t                         *message_size,
 583          frsh_network_address_t         *from)
 584 {
 585     frsh_resource_id_t resource_id;
 586
 587     if (check_valid_receive_endpoint(endpoint) < 0) {
 588             return FRSH_ERR_BAD_ARGUMENT;
 589     }
 590
 591     resource_id = endpoint->resource_id;
 592
 593     if (fna_operations[resource_id]->fna_receive_sync == NULL) {
 594         return FRSH_ERR_INTERNAL_ERROR;
 595     }
 596
 597     return fna_operations[resource_id]->fna_receive_sync
 598             (endpoint, buffer, buffer_size, message_size, from);
 599 }
 600
 601 //
 602 // int frsh_receive_async
 603 //   (const frsh_receive_endpoint_t  endpoint,
 604 //    void                          *buffer,
 605 //    size_t                         buffer_size,
 606 //    size_t                        *message_size,
 607 //    frsh_network_address_t *from);
 608 //
 609 // int frsh_receive_endpoint_get_status(const frsh_receive_endpoint_t endpoint,
 610 //                                   int *number_pending_messages,
 611 //                                   frsh_endpoint_network_status_t *network_status,
 612 //                                   frsh_protocol_status_t *protocol_status);
 613
 614 #endif /* FRSH_DISTRIBUTED_MODULE_SUPPORTED */