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