src_frescan/frescan_bwres_threads.c

   1 /*!
   2  * @file frescan_bwres_threads.c
   3  *
   4  * @brief FRESCAN bandwidth reservation layer: negotiation threads
   5  *
   6  * This module contains the acceptor threads and the master thread for local
   7  * negotiations, with functions to create them.
   8  *
   9  * @version 0.01
  10  *
  11  * @date 2-Apr-2008
  12  *
  13  * @author Daniel Sangorrin <daniel.sangorrin@unican.es>
  14  *
  15  * @license
  16  *
  17  * -----------------------------------------------------------------------
  18  *  Copyright (C) 2006 - 2008 FRESCOR consortium partners:
  19  *
  20  *    Universidad de Cantabria,              SPAIN
  21  *    University of York,                    UK
  22  *    Scuola Superiore Sant'Anna,            ITALY
  23  *    Kaiserslautern University,             GERMANY
  24  *    Univ. Politécnica  Valencia,           SPAIN
  25  *    Czech Technical University in Prague,  CZECH REPUBLIC
  26  *    ENEA                                   SWEDEN
  27  *    Thales Communication S.A.              FRANCE
  28  *    Visual Tools S.A.                      SPAIN
  29  *    Rapita Systems Ltd                     UK
  30  *    Evidence                               ITALY
  31  *
  32  *    See http://www.frescor.org for a link to partners' websites
  33  *
  34  *           FRESCOR project (FP6/2005/IST/5-034026) is funded
  35  *        in part by the European Union Sixth Framework Programme
  36  *        The European Union is not liable of any use that may be
  37  *        made of this code.
  38  *
  39  *  This file is part of FRESCAN
  40  *
  41  *  FRESCAN is free software; you can  redistribute it and/or  modify
  42  *  it under the terms of  the GNU General Public License as published by
  43  *  the Free Software Foundation;  either  version 2, or (at  your option)
  44  *  any later version.
  45  *
  46  *  FRESCAN  is distributed  in  the hope  that  it  will  be useful,  but
  47  *  WITHOUT  ANY  WARRANTY;     without  even the   implied   warranty  of
  48  *  MERCHANTABILITY  or  FITNESS FOR  A  PARTICULAR PURPOSE. See  the  GNU
  49  *  General Public License for more details.
  50  *
  51  *  You should have  received a  copy of  the  GNU  General Public License
  52  *  distributed  with  FRESCAN;  see file COPYING.   If not,  write to the
  53  *  Free Software  Foundation,  59 Temple Place  -  Suite 330,  Boston, MA
  54  *  02111-1307, USA.
  55  *
  56  * As a special exception, including FRESCAN header files in a file,
  57  * instantiating FRESCAN generics or templates, or linking other files
  58  * with FRESCAN objects to produce an executable application, does not
  59  * by itself cause the resulting executable application to be covered
  60  * by the GNU General Public License. This exception does not
  61  * however invalidate any other reasons why the executable file might be
  62  * covered by the GNU Public License.
  63  * -----------------------------------------------------------------------
  64  *
  65  */
  66
  67 #include <assert.h>
  68 #include "fosa_threads_and_signals.h" // fosa_thread_attr_init...
  69 #include "frescan_bwres_threads.h"
  70 #include "frescan_bwres_messages.h"
  71 #include "frescan_bwres_requests.h"
  72 #include "frescan_bwres_robjs.h"
  73 #include "frescan_bwres_analysis.h"
  74 #include "frescan_bwres_mode_change.h"
  75 #include "frescan_config.h"
  76 #include "frescan_debug.h"
  77 #include "frescan_data.h"
  78 #include "frescan_servers.h"
  79
  80 static void *frescan_manager_thread(void *arg);
  81 static void *frescan_acceptor_thread(void *arg);
  82
  83 static void frescan_manager_gn_prepare_scenario
  84                                 (frescan_bwres_sa_scenario_t  *scenario,
  85                                  frescan_bwres_request_data_t *req_data);
  86
  87 static void frescan_manager_gn_restore_scenario
  88                                 (frescan_bwres_sa_scenario_t  *scenario,
  89                                  frescan_bwres_request_data_t *req_data);
  90
  91 static void frescan_manager_req_gn(frescan_bwres_request_data_t *req_data);
  92 static void frescan_manager_rep_gn(frescan_bwres_request_data_t *req_data);
  93 static void frescan_manager_req_mc(frescan_bwres_request_data_t *req_data);
  94
  95 /**
  96  * frescan_manager_thread_create()
  97  *
  98  * This call creates the manager thread at each node which will be waiting
  99  * in a request queue for LOCAL or EXTERNAL requests.
 100  */
 101
 102 int frescan_manager_thread_create(frescan_network_t net)
 103 {
 104         int ret;
 105         fosa_thread_attr_t attr;
 106
 107         ret = fosa_thread_attr_init(&attr);
 108         if (ret != 0) return ret;
 109
 110         ret = fosa_thread_attr_set_prio(&attr, FRESCAN_BWRES_NEG_THREAD_PRIO);
 111         if (ret != 0) return ret;
 112
 113         ret = fosa_thread_create(&frescan_data[net].manager_thread_id,
 114                                  &attr,
 115                                  frescan_manager_thread,
 116                                  (void *)(uint32_t)net);
 117         if (ret != 0) return ret;
 118
 119         ret = fosa_thread_attr_destroy(&attr);
 120         if (ret != 0) return ret;
 121
 122         return 0;
 123 }
 124
 125 /**
 126  * frescan_acceptor_thread_create()
 127  *
 128  * This call creates the acceptor thread which will be waiting negotiation
 129  * messages from the network and converting them into requests.
 130  */
 131
 132 int frescan_acceptor_thread_create(frescan_network_t net)
 133 {
 134         int ret;
 135         fosa_thread_attr_t attr;
 136
 137         ret = fosa_thread_attr_init(&attr);
 138         if (ret != 0) return ret;
 139
 140         ret = fosa_thread_attr_set_prio(&attr, FRESCAN_BWRES_ACCEPTOR_PRIO);
 141         if (ret != 0) return ret;
 142
 143         ret = fosa_thread_create(&frescan_data[net].acceptor_thread_id,
 144                                  &attr,
 145                                  frescan_acceptor_thread,
 146                                  (void *)(uint32_t)net);
 147         if (ret != 0) return ret;
 148
 149         ret = fosa_thread_attr_destroy(&attr);
 150         if (ret != 0) return ret;
 151
 152         return 0;
 153 }
 154
 155 /**
 156  * frescan_manager_thread
 157  */
 158
 159 static void *frescan_manager_thread(void *arg)
 160 {
 161         int ret;
 162         frescan_bwres_request_id_t   req;
 163         frescan_bwres_request_data_t *req_data;
 164         frescan_network_t net = (uint32_t)arg;
 165
 166         DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG, "manager thread starts\n");
 167
 168         while(1) {
 169                 DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG, "wait for a request\n");
 170
 171                 ret = frescan_bwres_requests_dequeue(&req);
 172                 assert(ret == 0);
 173
 174                 ret = frescan_bwres_requests_get_data(req, &req_data);
 175                 assert(ret == 0);
 176
 177                 switch(req_data->type) {
 178                         case FRESCAN_BWRES_REQ_GN:
 179                                 frescan_manager_req_gn(req_data);
 180                                 break;
 181                         case FRESCAN_BWRES_REP_GN:
 182                                 frescan_manager_rep_gn(req_data);
 183                                 break;
 184                         case FRESCAN_BWRES_REQ_MC:
 185                                 frescan_manager_req_mc(req_data);
 186                                 break;
 187                         case FRESCAN_BWRES_REQ_RES:
 188                         case FRESCAN_BWRES_REQ_RES_GET:
 189                         case FRESCAN_BWRES_REP_RES_GET:
 190                         case FRESCAN_BWRES_REQ_RES_SET:
 191                         case FRESCAN_BWRES_REQ_RES_COMMIT:
 192                         case FRESCAN_BWRES_REQ_RES_CANCEL:
 193                         default:
 194                                 FRESCAN_ERROR("request type not supported\n");
 195                                 assert(0);
 196                 }
 197
 198                 if(req_data->request_node != frescan_data[net].local_node) {
 199                         ret = frescan_bwres_requests_free(req);
 200                         assert(ret == 0);
 201                 }
 202         }
 203 }
 204
 205 /**
 206  * frescan_acceptor_thread()
 207  */
 208
 209 static void *frescan_acceptor_thread(void *arg)
 210 {
 211         int ret;
 212         frescan_bwres_request_id_t req;
 213         frescan_network_t net = (uint32_t)arg;
 214
 215         DEBUG(FRESCAN_BWRES_ACCEPTOR_ENABLE_DEBUG, "acceptor thread starts\n");
 216
 217         while(1) {
 218                 ret = frescan_messages_recv_request(net, &req);
 219                 assert(ret == 0);
 220
 221                 ret = frescan_bwres_requests_enqueue(req);
 222                 assert(ret == 0);
 223         }
 224
 225         return NULL;
 226 }
 227
 228 /**
 229  * frescan_manager_req_gn
 230  */
 231
 232 static void frescan_manager_req_gn(frescan_bwres_request_data_t *req_data)
 233 {
 234         int ret, i;
 235         frescan_node_t me;
 236         bool accepted;
 237         frescan_bwres_sa_scenario_t *scenario;
 238         frescan_ss_t ss;
 239         frescan_server_params_t server_params;
 240         frescan_bwres_vres_t *vres;
 241
 242         me = frescan_data[req_data->net].local_node;
 243
 244         if (me != FRESCAN_BWRES_MASTER_NODE) {
 245                 DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG,
 246                       "send gn req to master\n");
 247                 ret = frescan_messages_send_request(req_data);
 248                 assert(ret == 0);
 249                 return;
 250         }
 251
 252         scenario = &frescan_data[req_data->net].scenario;
 253
 254         DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG, "prepare new scenario\n");
 255         frescan_manager_gn_prepare_scenario(scenario, req_data);
 256
 257         DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG, "perform sched analysis\n");
 258         ret = frescan_bwres_sa_sched_test(scenario, &accepted);
 259         assert(ret == 0);
 260
 261         if (accepted) {
 262                 DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG, "ACCEPTED!\n");
 263                 req_data->return_value = FRESCAN_BWRES_REQ_ACCEPTED;
 264
 265                 DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG,
 266                       "spare capacity and mode change\n");
 267
 268                 ret = frescan_bwres_sa_spare_capacity(scenario);
 269                 assert(ret == 0);
 270
 271                 ret = frescan_bwres_mode_change_protocol(req_data);
 272                 assert(ret == 0);
 273         } else {
 274                 DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG, "FAILED!\n");
 275                 req_data->return_value = FRESCAN_BWRES_REQ_NOT_ACCEPTED;
 276                 frescan_manager_gn_restore_scenario(scenario, req_data);
 277         }
 278
 279         if (req_data->request_node == me) {
 280                 // create servers for new contracts
 281                 for(i=0; i<req_data->contracts_to_neg->size; i++) {
 282                         vres = &frescan_data[req_data->net].scenario.
 283                                         vres_pool[me]
 284                                                  [req_data->ss_new->ss[i]];
 285
 286                         server_params.budget = frsh_rel_time_to_usec(
 287                                         frsh_sa_time_to_rel_time(vres->old_c)) /
 288                                         FRESCAN_FRAME_TX_TIME_US;
 289
 290                         server_params.period = frsh_sa_time_to_rel_time
 291                                                                 (vres->old_t);
 292                         server_params.prio   = vres->old_p;
 293
 294                         // Create server
 295                         ret = frescan_servers_create(req_data->net,
 296                                                      &server_params,
 297                                                      &ss);
 298                         assert(ret == 0);
 299                         assert (req_data->ss_new->ss[i] == ss);
 300                 }
 301
 302                 DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG, "signal robj\n");
 303                 ret = frescan_bwres_robjs_signal(req_data->robj);
 304                 assert(ret == 0);
 305         } else {
 306                 DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG, "send reply\n");
 307                 req_data->type = FRESCAN_BWRES_REP_GN;
 308                 ret = frescan_messages_send_request(req_data);
 309                 assert(ret == 0);
 310         }
 311 }
 312
 313 /**
 314  * frescan_manager_gn_prepare_scenario
 315  */
 316
 317 static void frescan_manager_gn_prepare_scenario
 318                                 (frescan_bwres_sa_scenario_t  *scenario,
 319                                  frescan_bwres_request_data_t *req_data)
 320 {
 321         int ret, i;
 322
 323         // NEG-GROUP
 324         for(i=0; i<req_data->contracts_to_neg->size; i++) {
 325                 ret = freelist_alloc(&frescan_data[req_data->net].scenario.
 326                                       ss_id_freelist[req_data->request_node]);
 327                 assert(ret >= 0);
 328
 329                 req_data->ss_new->ss[i] = (frescan_ss_t)ret;
 330
 331                 ret = frescan_bwres_sa_add_contract
 332                                 (scenario,
 333                                  req_data->ss_new->ss[i],
 334                                  req_data->request_node,
 335                                  &req_data->contracts_to_neg->contracts[i]);
 336                 assert(ret == 0);
 337         }
 338
 339         // RENEG-GROUP
 340         scenario->backup_contracts_to_reneg.size =
 341                                         req_data->contracts_to_reneg->size;
 342
 343         for(i=0; i<req_data->contracts_to_reneg->size; i++) {
 344                 ret = frescan_bwres_sa_update_contract
 345                         (scenario,
 346                          req_data->ss_to_reneg->ss[i],
 347                          req_data->request_node,
 348                          &req_data->contracts_to_reneg->contracts[i],
 349                          &scenario->backup_contracts_to_reneg.contracts[i]);
 350                 assert(ret == 0);
 351         }
 352
 353         // CANCEL-GROUP
 354         scenario->backup_contracts_to_cancel.size =
 355                                                 req_data->ss_to_cancel->size;
 356
 357         for(i=0; i<req_data->ss_to_cancel->size; i++) {
 358                 ret = frescan_bwres_sa_remove_contract
 359                         (scenario,
 360                          req_data->ss_to_cancel->ss[i],
 361                          req_data->request_node,
 362                          &scenario->backup_contracts_to_cancel.contracts[i]);
 363                 assert(ret == 0);
 364         }
 365 }
 366
 367 /**
 368  * frescan_manager_gn_restore_scenario
 369  */
 370
 371 static void frescan_manager_gn_restore_scenario
 372                 (frescan_bwres_sa_scenario_t  *scenario,
 373                  frescan_bwres_request_data_t *req_data)
 374 {
 375         int ret, i;
 376
 377         // NEG-GROUP
 378         for(i=0; i<req_data->contracts_to_neg->size; i++) {
 379                 ret = frescan_bwres_sa_remove_contract
 380                                 (scenario,
 381                                  req_data->ss_new->ss[i],
 382                                  req_data->request_node,
 383                                  NULL);
 384                 assert(ret == 0);
 385
 386                 ret = freelist_free(&frescan_data[req_data->net].scenario.
 387                                         ss_id_freelist[req_data->request_node],
 388                                     req_data->ss_new->ss[i]);
 389                 assert(ret == 0);
 390         }
 391
 392         // RENEG-GROUP
 393         for(i=0; i<req_data->contracts_to_reneg->size; i++) {
 394                 ret = frescan_bwres_sa_update_contract
 395                         (scenario,
 396                          req_data->ss_to_reneg->ss[i],
 397                          req_data->request_node,
 398                          &scenario->backup_contracts_to_reneg.contracts[i],
 399                          NULL);
 400                 assert(ret == 0);
 401         }
 402
 403         // CANCEL-GROUP
 404         for(i=0; i<req_data->ss_to_cancel->size; i++) {
 405                 ret = frescan_bwres_sa_add_contract
 406                         (scenario,
 407                          req_data->ss_to_cancel->ss[i],
 408                          req_data->request_node,
 409                          &scenario->backup_contracts_to_cancel.contracts[i]);
 410                 assert(ret == 0);
 411         }
 412 }
 413
 414 /**
 415  * frescan_manager_rep_gn
 416  */
 417
 418 static void frescan_manager_rep_gn(frescan_bwres_request_data_t *req_data)
 419 {
 420         int ret, i;
 421         frescan_bwres_request_data_t *caller_req;
 422
 423         ret = frescan_bwres_requests_get_data(req_data->req, &caller_req);
 424         assert(ret == 0);
 425
 426         DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG,
 427               "reply to gn, return value:%d\n", req_data->return_value);
 428
 429         caller_req->return_value = req_data->return_value;
 430
 431         assert (req_data->ss_new->size == caller_req->contracts_to_neg->size);
 432
 433         for (i=0; i<req_data->ss_new->size; i++) {
 434                 caller_req->ss_new->ss[i] = req_data->ss_new->ss[i];
 435         }
 436
 437         ret = frescan_bwres_robjs_signal(caller_req->robj);
 438         assert(ret == 0);
 439 }
 440
 441 /**
 442  * frescan_manager_req_mc
 443  */
 444
 445 static void frescan_manager_req_mc(frescan_bwres_request_data_t *req_data)
 446 {
 447         int ret;
 448
 449         ret = frescan_bwres_mode_change_local(req_data->net,
 450                                               req_data->mode_change_type,
 451                                               req_data->ss_to_cancel);
 452         assert(ret == 0);
 453 }