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;
 235         frescan_node_t me;
 236         bool accepted;
 237         frescan_bwres_sa_scenario_t *scenario;
 238
 239         me = frescan_data[req_data->net].local_node;
 240
 241         if (me != FRESCAN_BWRES_MASTER_NODE) {
 242                 DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG,
 243                       "send gn req to master\n");
 244                 ret = frescan_messages_send_request(req_data);
 245                 assert(ret == 0);
 246                 return;
 247         }
 248
 249         scenario = &frescan_data[req_data->net].scenario;
 250
 251         DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG, "prepare new scenario\n");
 252         frescan_manager_gn_prepare_scenario(scenario, req_data);
 253
 254         DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG, "perform sched analysis\n");
 255         ret = frescan_bwres_sa_sched_test(scenario, &accepted);
 256         assert(ret == 0);
 257
 258         if (accepted) {
 259                 DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG, "ACCEPTED!\n");
 260                 req_data->return_value = FRESCAN_BWRES_REQ_ACCEPTED;
 261
 262                 DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG,
 263                       "spare capacity and mode change\n");
 264
 265                 ret = frescan_bwres_sa_spare_capacity(scenario);
 266                 assert(ret == 0);
 267
 268                 ret = frescan_bwres_mode_change_protocol(req_data);
 269                 assert(ret == 0);
 270         } else {
 271                 DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG, "FAILED!\n");
 272                 req_data->return_value = FRESCAN_BWRES_REQ_NOT_ACCEPTED;
 273                 frescan_manager_gn_restore_scenario(scenario, req_data);
 274         }
 275
 276         if (req_data->request_node == me) {
 277                 DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG, "signal robj\n");
 278                 ret = frescan_bwres_robjs_signal(req_data->robj);
 279                 assert(ret == 0);
 280         } else {
 281                 DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG, "send reply\n");
 282                 req_data->type = FRESCAN_BWRES_REP_GN;
 283                 ret = frescan_messages_send_request(req_data);
 284                 assert(ret == 0);
 285         }
 286 }
 287
 288 /**
 289  * frescan_manager_gn_prepare_scenario
 290  */
 291
 292 static void frescan_manager_gn_prepare_scenario
 293                                 (frescan_bwres_sa_scenario_t  *scenario,
 294                                  frescan_bwres_request_data_t *req_data)
 295 {
 296         int ret, i;
 297         frescan_server_params_t server_params;
 298
 299         // NEG-GROUP
 300         server_params.budget         = 0;
 301         server_params.period.tv_sec  = 0;
 302         server_params.period.tv_nsec = 0;
 303         server_params.prio           = 0;
 304
 305         for(i=0; i<req_data->contracts_to_neg->size; i++) {
 306                 ret = frescan_servers_create(req_data->net,
 307                                              &server_params,
 308                                              &req_data->ss_new->ss[i]);
 309                 assert(ret == 0);
 310
 311                 ret = frescan_bwres_sa_add_contract
 312                                 (scenario,
 313                                  req_data->ss_new->ss[i],
 314                                  req_data->request_node,
 315                                  &req_data->contracts_to_neg->contracts[i]);
 316                 assert(ret == 0);
 317         }
 318
 319         // RENEG-GROUP
 320         scenario->backup_contracts_to_reneg.size =
 321                                         req_data->contracts_to_reneg->size;
 322
 323         for(i=0; i<req_data->contracts_to_reneg->size; i++) {
 324                 ret = frescan_bwres_sa_update_contract
 325                         (scenario,
 326                          req_data->ss_to_reneg->ss[i],
 327                          req_data->request_node,
 328                          &req_data->contracts_to_reneg->contracts[i],
 329                          &scenario->backup_contracts_to_reneg.contracts[i]);
 330                 assert(ret == 0);
 331         }
 332
 333         // CANCEL-GROUP
 334         scenario->backup_contracts_to_cancel.size =
 335                                                 req_data->ss_to_cancel->size;
 336
 337         for(i=0; i<req_data->ss_to_cancel->size; i++) {
 338                 ret = frescan_bwres_sa_remove_contract
 339                         (scenario,
 340                          req_data->ss_to_cancel->ss[i],
 341                          req_data->request_node,
 342                          &scenario->backup_contracts_to_cancel.contracts[i]);
 343                 assert(ret == 0);
 344         }
 345 }
 346
 347 /**
 348  * frescan_manager_gn_restore_scenario
 349  */
 350
 351 static void frescan_manager_gn_restore_scenario
 352                 (frescan_bwres_sa_scenario_t  *scenario,
 353                  frescan_bwres_request_data_t *req_data)
 354 {
 355         int ret, i;
 356
 357         // NEG-GROUP
 358         for(i=0; i<req_data->contracts_to_neg->size; i++) {
 359                 ret = frescan_bwres_sa_remove_contract
 360                                 (scenario,
 361                                  req_data->ss_new->ss[i],
 362                                  req_data->request_node,
 363                                  NULL);
 364                 assert(ret == 0);
 365
 366                 ret = frescan_servers_destroy(req_data->net,
 367                                               req_data->ss_new->ss[i]);
 368                 assert(ret == 0);
 369         }
 370
 371         // RENEG-GROUP
 372         for(i=0; i<req_data->contracts_to_reneg->size; i++) {
 373                 ret = frescan_bwres_sa_update_contract
 374                         (scenario,
 375                          req_data->ss_to_reneg->ss[i],
 376                          req_data->request_node,
 377                          &scenario->backup_contracts_to_reneg.contracts[i],
 378                          NULL);
 379                 assert(ret == 0);
 380         }
 381
 382         // CANCEL-GROUP
 383         for(i=0; i<req_data->ss_to_cancel->size; i++) {
 384                 ret = frescan_bwres_sa_add_contract
 385                         (scenario,
 386                          req_data->ss_to_cancel->ss[i],
 387                          req_data->request_node,
 388                          &scenario->backup_contracts_to_cancel.contracts[i]);
 389                 assert(ret == 0);
 390         }
 391 }
 392
 393 /**
 394  * frescan_manager_rep_gn
 395  */
 396
 397 static void frescan_manager_rep_gn(frescan_bwres_request_data_t *req_data)
 398 {
 399         int ret;
 400         frescan_bwres_request_data_t *orig_req_data;
 401
 402         ret = frescan_bwres_requests_get_data(req_data->req, &orig_req_data);
 403         assert(ret == 0);
 404
 405         DEBUG(FRESCAN_BWRES_MANAGER_ENABLE_DEBUG,
 406               "reply to gn, return value:%d\n", req_data->return_value);
 407
 408         orig_req_data->return_value = req_data->return_value;
 409
 410         ret = frescan_bwres_robjs_signal(orig_req_data->robj);
 411         assert(ret == 0);
 412 }
 413
 414 /**
 415  * frescan_manager_req_mc
 416  */
 417
 418 static void frescan_manager_req_mc(frescan_bwres_request_data_t *req_data)
 419 {
 420         int ret;
 421
 422         ret = frescan_bwres_mode_change_local(req_data->net,
 423                                               req_data->mode_change_type);
 424         assert(ret == 0);
 425 }