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                 DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG, "send reply\n");
 281                 req_data->type = FRESCAN_BWRES_REP_GN;
 282
 283                 ret = frescan_messages_send_request(req_data);
 284                 assert(ret == 0);
 285                 return;
 286         }
 287
 288         if (req_data->return_value == FRESCAN_BWRES_REQ_ACCEPTED) {
 289                 // create servers for new contracts
 290                 req_data->ss_new->size = req_data->contracts_to_neg->size;
 291                 for(i=0; i<req_data->ss_new->size; i++) {
 292                         vres = &frescan_data[req_data->net].scenario.
 293                                         vres_pool[me]
 294                                                 [req_data->ss_new->ss[i]];
 295
 296                         server_params.budget = frsh_rel_time_to_usec(
 297                                         frsh_sa_time_to_rel_time(vres->old_c)) /
 298                                         FRESCAN_FRAME_TX_TIME_US;
 299
 300                         server_params.period = frsh_sa_time_to_rel_time
 301                                                                 (vres->old_t);
 302                         server_params.prio   = vres->old_p;
 303
 304                         // Create server
 305                         ret = frescan_servers_create(req_data->net,
 306                                                 &server_params,
 307                                                 &ss);
 308                         assert(ret == 0);
 309                         assert (req_data->ss_new->ss[i] == ss);
 310                 }
 311         }
 312
 313         DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG, "signal robj\n");
 314         ret = frescan_bwres_robjs_signal(req_data->robj);
 315         assert(ret == 0);
 316 }
 317
 318 /**
 319  * frescan_manager_gn_prepare_scenario
 320  */
 321
 322 static void frescan_manager_gn_prepare_scenario
 323                                 (frescan_bwres_sa_scenario_t  *scenario,
 324                                  frescan_bwres_request_data_t *req_data)
 325 {
 326         int ret, i;
 327
 328         // NEG-GROUP
 329         for(i=0; i<req_data->contracts_to_neg->size; i++) {
 330                 ret = freelist_alloc(&frescan_data[req_data->net].scenario.
 331                                       ss_id_freelist[req_data->request_node]);
 332                 assert(ret >= 0);
 333
 334                 req_data->ss_new->ss[i] = (frescan_ss_t)ret;
 335
 336                 DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG,
 337                       "prealloc net:%d node:%d ss_new->ss[%d]:%d\n",
 338                       req_data->net, req_data->request_node,
 339                       i, req_data->ss_new->ss[i]);
 340
 341                 ret = frescan_bwres_sa_add_contract
 342                                 (scenario,
 343                                  req_data->ss_new->ss[i],
 344                                  req_data->request_node,
 345                                  &req_data->contracts_to_neg->contracts[i]);
 346                 assert(ret == 0);
 347         }
 348
 349         req_data->ss_new->size = req_data->contracts_to_neg->size;
 350
 351         // RENEG-GROUP
 352         scenario->backup_contracts_to_reneg.size =
 353                                         req_data->contracts_to_reneg->size;
 354
 355         for(i=0; i<req_data->contracts_to_reneg->size; i++) {
 356                 ret = frescan_bwres_sa_update_contract
 357                         (scenario,
 358                          req_data->ss_to_reneg->ss[i],
 359                          req_data->request_node,
 360                          &req_data->contracts_to_reneg->contracts[i],
 361                          &scenario->backup_contracts_to_reneg.contracts[i]);
 362                 assert(ret == 0);
 363         }
 364
 365         // CANCEL-GROUP
 366         scenario->backup_contracts_to_cancel.size =
 367                                                 req_data->ss_to_cancel->size;
 368
 369         for(i=0; i<req_data->ss_to_cancel->size; i++) {
 370                 ret = frescan_bwres_sa_remove_contract
 371                         (scenario,
 372                          req_data->ss_to_cancel->ss[i],
 373                          req_data->request_node,
 374                          &scenario->backup_contracts_to_cancel.contracts[i]);
 375                 assert(ret == 0);
 376         }
 377 }
 378
 379 /**
 380  * frescan_manager_gn_restore_scenario
 381  */
 382
 383 static void frescan_manager_gn_restore_scenario
 384                 (frescan_bwres_sa_scenario_t  *scenario,
 385                  frescan_bwres_request_data_t *req_data)
 386 {
 387         int ret, i;
 388
 389         // NEG-GROUP
 390         for(i=0; i<req_data->contracts_to_neg->size; i++) {
 391                 ret = frescan_bwres_sa_remove_contract
 392                                 (scenario,
 393                                  req_data->ss_new->ss[i],
 394                                  req_data->request_node,
 395                                  NULL);
 396                 assert(ret == 0);
 397
 398                 ret = freelist_free(&frescan_data[req_data->net].scenario.
 399                                         ss_id_freelist[req_data->request_node],
 400                                     req_data->ss_new->ss[i]);
 401                 assert(ret == 0);
 402         }
 403
 404         // RENEG-GROUP
 405         for(i=0; i<req_data->contracts_to_reneg->size; i++) {
 406                 ret = frescan_bwres_sa_update_contract
 407                         (scenario,
 408                          req_data->ss_to_reneg->ss[i],
 409                          req_data->request_node,
 410                          &scenario->backup_contracts_to_reneg.contracts[i],
 411                          NULL);
 412                 assert(ret == 0);
 413         }
 414
 415         // CANCEL-GROUP
 416         for(i=0; i<req_data->ss_to_cancel->size; i++) {
 417                 ret = frescan_bwres_sa_add_contract
 418                         (scenario,
 419                          req_data->ss_to_cancel->ss[i],
 420                          req_data->request_node,
 421                          &scenario->backup_contracts_to_cancel.contracts[i]);
 422                 assert(ret == 0);
 423         }
 424 }
 425
 426 /**
 427  * frescan_manager_rep_gn
 428  */
 429
 430 static void frescan_manager_rep_gn(frescan_bwres_request_data_t *req_data)
 431 {
 432         int ret, i;
 433         frescan_bwres_request_data_t *caller_req;
 434
 435         ret = frescan_bwres_requests_get_data(req_data->req, &caller_req);
 436         assert(ret == 0);
 437
 438         caller_req->return_value = req_data->return_value;
 439
 440         DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG,
 441               "ret:%d -> %s\n", req_data->return_value,
 442               (req_data->return_value ==
 443                FRESCAN_BWRES_REQ_ACCEPTED) ? "OK" : "FAIL");
 444
 445         if (req_data->return_value == FRESCAN_BWRES_REQ_ACCEPTED) {
 446                 assert (req_data->ss_new->size ==
 447                         caller_req->contracts_to_neg->size);
 448                 caller_req->ss_new->size = req_data->ss_new->size;
 449
 450                 DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG,
 451                       "ss_new->size:%u\n", caller_req->ss_new->size);
 452
 453                 for (i=0; i<caller_req->ss_new->size; i++) {
 454                         caller_req->ss_new->ss[i] = req_data->ss_new->ss[i];
 455                 }
 456         }
 457
 458         ret = frescan_bwres_robjs_signal(caller_req->robj);
 459         assert(ret == 0);
 460 }
 461
 462 /**
 463  * frescan_manager_req_mc
 464  */
 465
 466 static void frescan_manager_req_mc(frescan_bwres_request_data_t *req_data)
 467 {
 468         int ret;
 469
 470         ret = frescan_bwres_mode_change_local(req_data->net,
 471                                               req_data->mode_change_type,
 472                                               req_data->ss_to_cancel);
 473         assert(ret == 0);
 474 }