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     if (resource_id)
 324             *resource_id = endpoint->resource_id;
 325     if (destination)
 326             *destination = endpoint->destination;
 327     if (stream)
 328             *stream = endpoint->stream_id;
 329     if (protocol_info)
 330             *protocol_info = endpoint->endpoint_protocol_info.send;
 331
 332     return 0;
 333 }
 334 //----------------------------//
 335 // frsh_send_endpoint_destroy //
 336 //----------------------------//
 337
 338 int frsh_send_endpoint_destroy
 339      (frsh_send_endpoint_t  endpoint)
 340 {
 341     fna_endpoint_data_t *epoint = endpoint;
 342     int rv;
 343
 344     if (check_valid_send_endpoint(endpoint) < 0) {
 345             return FRSH_ERR_BAD_ARGUMENT;
 346     }
 347
 348     if (fna_operations[epoint->resource_id]->fna_endpoint_destroy) {
 349             rv = fna_operations[epoint->resource_id]->\
 350                         fna_endpoint_destroy(endpoint);
 351             if (rv)
 352                     return rv;
 353     }
 354
 355     endpoint_free(endpoint);
 356     return 0;
 357 }
 358
 359 //-------------------------//
 360 // frsh_send_endpoint_bind //
 361 //-------------------------//
 362
 363 int frsh_send_endpoint_bind
 364         (frsh_vres_id_t      vres,
 365          frsh_send_endpoint_t  endpoint)
 366 {
 367     if (check_valid_send_endpoint(endpoint) < 0) {
 368             return FRSH_ERR_BAD_ARGUMENT;
 369     }
 370
 371     // TODO: check that resource_id and resource_type matches
 372     endpoint->vres = vres;
 373     endpoint->is_bound = true;
 374     /*FRSH_TRACE(FRSH_TRACE_DISTRIBUTED,"%s:vres=%d\n", __func__,
 375                endpoint_data[pos].vres);*/
 376
 377     if (fna_operations[endpoint->resource_id]->fna_send_endpoint_bind) {
 378             return fna_operations[endpoint->resource_id]->\
 379                         fna_send_endpoint_bind(endpoint,vres);
 380     }
 381     else
 382             return 0;
 383 }
 384
 385 //---------------------------//
 386 // frsh_send_endpoint_unbind //
 387 //---------------------------//
 388
 389 int frsh_send_endpoint_unbind
 390   (frsh_send_endpoint_t  endpoint)
 391 {
 392     if (check_valid_send_endpoint(endpoint) < 0) {
 393             return FRSH_ERR_BAD_ARGUMENT;
 394     }
 395
 396     endpoint->is_bound = false;
 397
 398     if (fna_operations[endpoint->resource_id]->fna_send_endpoint_unbind) {
 399             return fna_operations[endpoint->resource_id]->\
 400                         fna_send_endpoint_unbind(endpoint);
 401     }
 402     else
 403             return 0;
 404 }
 405
 406 //--------------------------------//
 407 // frsh_send_endpoint_get_vres_id //
 408 //--------------------------------//
 409
 410 int frsh_send_endpoint_get_vres_id
 411         (const frsh_send_endpoint_t  endpoint,
 412          frsh_vres_id_t            *vres)
 413 {
 414     if (check_valid_send_endpoint(endpoint) < 0) {
 415             return FRSH_ERR_BAD_ARGUMENT;
 416     }
 417
 418     if (vres)
 419             *vres = endpoint->vres;
 420     return 0;
 421 }
 422
 423 //-----------------//
 424 // frsh_send_async //
 425 //-----------------//
 426
 427 int frsh_send_async
 428         (const frsh_send_endpoint_t  endpoint,
 429          const void                  *msg,
 430          const size_t                size)
 431 {
 432     frsh_resource_id_t resource_id;
 433
 434     if (check_valid_send_endpoint(endpoint) < 0) {
 435             return FRSH_ERR_BAD_ARGUMENT;
 436     }
 437
 438     if (endpoint->is_bound == false) {
 439             return FRSH_ERR_NOT_BOUND;
 440     }
 441
 442     resource_id = endpoint->resource_id;
 443
 444     if (fna_operations[resource_id]->fna_send_async == NULL) {
 445             return FRSH_ERR_INTERNAL_ERROR;
 446     }
 447
 448     return fna_operations[resource_id]->fna_send_async
 449                 (endpoint, msg, size);
 450 }
 451
 452 //-----------------//
 453 // frsh_send_sync //
 454 //-----------------//
 455
 456 int frsh_send_sync
 457                 (const frsh_send_endpoint_t  endpoint,
 458                  const void                  *msg,
 459                  const size_t                size)
 460 {
 461     frsh_resource_id_t resource_id;
 462
 463     if (check_valid_send_endpoint(endpoint) < 0) {
 464             return FRSH_ERR_BAD_ARGUMENT;
 465     }
 466
 467     if (endpoint->is_bound == false) {
 468             return FRSH_ERR_NOT_BOUND;
 469     }
 470
 471     resource_id = endpoint->resource_id;
 472
 473     if (fna_operations[resource_id]->fna_send_sync == NULL) {
 474             return FRSH_ERR_INTERNAL_ERROR;
 475     }
 476
 477     return fna_operations[resource_id]->fna_send_sync
 478                         (endpoint, msg, size);
 479 }
 480
 481 // int frsh_send_endpoint_get_status(const frsh_send_endpoint_t endpoint,
 482 //                                   int *number_pending_msg,
 483 //                                   frsh_endpoint_network_status_t *network_status,
 484 //                                   frsh_protocol_status_t *protocol_status);
 485 //
 486
 487 //------------------------------//
 488 // frsh_receive_endpoint_create //
 489 //------------------------------//
 490
 491 int frsh_receive_endpoint_create
 492         (frsh_resource_id_t        resource_id,
 493          frsh_stream_id_t          stream_id,
 494          frsh_endpoint_queueing_info_t queueing_info,
 495          frsh_receive_endpoint_protocol_info_t protocol_info,
 496          frsh_receive_endpoint_t  *endpoint)
 497 {
 498     frsh_receive_endpoint_t epoint;
 499
 500     if (check_valid_resource_id(resource_id)) {
 501         return FRSH_ERR_RESOURCE_ID_INVALID;
 502     }
 503
 504     if (endpoint == NULL) {
 505         return FRSH_ERR_BAD_ARGUMENT;
 506     }
 507
 508     if ((epoint = endpoint_alloc()) == NULL) {
 509         return FRSH_ERR_NO_SPACE;
 510     }
 511
 512     // initialize the receive_endpoint
 513     epoint->endpoint_type = FRSH_RECEIVE_ENDPOINT_TYPE;
 514     epoint->resource_id = resource_id;
 515     epoint->stream_id = stream_id;
 516     epoint->queue_info = queueing_info;
 517     epoint->endpoint_protocol_info.receive = protocol_info;
 518
 519     //FRSH_TRACE(FRSH_TRACE_DISTRIBUTED,"%s:stream_id=%d\n", __func__, stream_id);
 520     // the receive_endpoint identifier for the user is the position in the table
 521     *endpoint = epoint;
 522
 523     if (fna_operations[resource_id]->fna_receive_endpoint_created) {
 524             return fna_operations[resource_id]->\
 525                         fna_receive_endpoint_created(epoint);
 526     }
 527     else
 528             return 0;
 529 }
 530
 531 //----------------------------------//
 532 // frsh_receive_endpoint_get_params //
 533 //----------------------------------//
 534
 535 int frsh_receive_endpoint_get_params
 536         (const frsh_receive_endpoint_t  endpoint,
 537          frsh_resource_id_t        *resource_id,
 538          frsh_stream_id_t          *stream,
 539          frsh_endpoint_queueing_info_t   *queueing_info,
 540          frsh_receive_endpoint_protocol_info_t   *protocol_info)
 541 {
 542     if (check_valid_receive_endpoint(endpoint) < 0) {
 543             return FRSH_ERR_BAD_ARGUMENT;
 544     }
 545
 546     if (resource_id)
 547             *resource_id = endpoint->resource_id;
 548     if (stream)
 549             *stream = endpoint->stream_id;
 550     if (queueing_info)
 551             *queueing_info = endpoint->queue_info;
 552     if (protocol_info)
 553             *protocol_info = endpoint->endpoint_protocol_info.receive;
 554
 555     return 0;
 556 }
 557
 558 //-------------------------------//
 559 // frsh_receive_endpoint_destroy //
 560 //-------------------------------//
 561
 562 int frsh_receive_endpoint_destroy
 563         (frsh_receive_endpoint_t  endpoint)
 564 {
 565     fna_endpoint_data_t *epoint = endpoint;
 566     int rv;
 567
 568     if (check_valid_receive_endpoint(endpoint) < 0) {
 569             return FRSH_ERR_BAD_ARGUMENT;
 570     }
 571
 572     if (fna_operations[epoint->resource_id]->fna_endpoint_destroy) {
 573             rv = fna_operations[epoint->resource_id]->\
 574                         fna_endpoint_destroy(endpoint);
 575             if (rv)
 576                     return rv;
 577     }
 578
 579     endpoint_free(endpoint);
 580     return 0;
 581 }
 582
 583 //-------------------//
 584 // frsh_receive_sync //
 585 //-------------------//
 586
 587 int frsh_receive_sync
 588         (const frsh_receive_endpoint_t  endpoint,
 589          void                           *buffer,
 590          size_t                         buffer_size,
 591          size_t                         *message_size,
 592          frsh_network_address_t         *from)
 593 {
 594     frsh_resource_id_t resource_id;
 595
 596     if (check_valid_receive_endpoint(endpoint) < 0) {
 597             return FRSH_ERR_BAD_ARGUMENT;
 598     }
 599
 600     resource_id = endpoint->resource_id;
 601
 602     if (fna_operations[resource_id]->fna_receive_sync == NULL) {
 603         return FRSH_ERR_INTERNAL_ERROR;
 604     }
 605
 606     return fna_operations[resource_id]->fna_receive_sync
 607             (endpoint, buffer, buffer_size, message_size, from);
 608 }
 609
 610 //
 611 // int frsh_receive_async
 612 //   (const frsh_receive_endpoint_t  endpoint,
 613 //    void                          *buffer,
 614 //    size_t                         buffer_size,
 615 //    size_t                        *message_size,
 616 //    frsh_network_address_t *from);
 617 //
 618 // int frsh_receive_endpoint_get_status(const frsh_receive_endpoint_t endpoint,
 619 //                                   int *number_pending_messages,
 620 //                                   frsh_endpoint_network_status_t *network_status,
 621 //                                   frsh_protocol_status_t *protocol_status);
 622
 623 #endif /* FRSH_DISTRIBUTED_MODULE_SUPPORTED */