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_analysis.h"
  73 #include "frescan_config.h"
  74 #include "frescan_debug.h"
  75 #include "frescan_data.h"
  76 #include "frescan_servers.h"
  77
  78 /**
  79  * frescan_manager_thread_create()
  80  *
  81  * This call creates the manager thread at each node which will be waiting
  82  * in a request queue for LOCAL or EXTERNAL requests.
  83  */
  84
  85 static void *frescan_manager_thread(void *arg);
  86
  87 int frescan_manager_thread_create(frescan_network_t net)
  88 {
  89         int ret;
  90         fosa_thread_attr_t attr;
  91
  92         ret = fosa_thread_attr_init(&attr);
  93         if (ret != 0) return ret;
  94
  95         ret = fosa_thread_attr_set_prio(&attr, FRESCAN_NEG_THREAD_PRIO);
  96         if (ret != 0) return ret;
  97
  98         ret = fosa_thread_create(&the_networks[net].manager_thread_id,
  99                                  &attr,
 100                                  frescan_manager_thread,
 101                                  (void *)(uint32_t)net);
 102         if (ret != 0) return ret;
 103
 104         ret = fosa_thread_attr_destroy(&attr);
 105         if (ret != 0) return ret;
 106
 107         return 0;
 108 }
 109
 110 /**
 111  * frescan_acceptor_thread_create()
 112  *
 113  * This call creates the acceptor thread which will be waiting negotiation
 114  * messages from the network and converting them into requests.
 115  */
 116
 117 static void *frescan_acceptor_thread(void *arg);
 118
 119 int frescan_acceptor_thread_create(frescan_network_t net)
 120 {
 121         int ret;
 122         fosa_thread_attr_t attr;
 123
 124         ret = fosa_thread_attr_init(&attr);
 125         if (ret != 0) return ret;
 126
 127         ret = fosa_thread_attr_set_prio(&attr, FRESCAN_ACCEPTOR_THREAD_PRIO);
 128         if (ret != 0) return ret;
 129
 130         ret = fosa_thread_create(&the_networks[net].acceptor_thread_id,
 131                                  &attr,
 132                                  frescan_acceptor_thread,
 133                                  (void *)(uint32_t)net);
 134         if (ret != 0) return ret;
 135
 136         ret = fosa_thread_attr_destroy(&attr);
 137         if (ret != 0) return ret;
 138
 139         return 0;
 140 }
 141
 142 /**
 143  * frescan_manager_thread
 144  */
 145
 146 static void frescan_manager_neg(frescan_request_data_t *req_data);
 147 static void frescan_manager_reneg(frescan_request_data_t *req_data);
 148 static void frescan_manager_cancel(frescan_request_data_t *req_data);
 149 static void frescan_manager_repneg(frescan_request_data_t *req_data);
 150
 151 static void *frescan_manager_thread(void *arg)
 152 {
 153         int ret;
 154         frescan_request_id_t   req;
 155         frescan_request_data_t *req_data;
 156         frescan_network_t net = (uint32_t)arg;
 157
 158         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "manager thread starts\n");
 159
 160         while(1) {
 161                 DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "wait for a request\n");
 162
 163                 ret = frescan_requests_dequeue(&req);
 164                 assert(ret == 0);
 165
 166                 ret = frescan_requests_get_data(req, &req_data);
 167                 assert(ret == 0);
 168
 169                 switch(req_data->type) {
 170                         case FRESCAN_REQ_NEG:
 171                                 frescan_manager_neg(req_data);
 172                                 break;
 173                         case FRESCAN_REQ_RENEG:
 174                                 frescan_manager_reneg(req_data);
 175                                 break;
 176                         case FRESCAN_REQ_CANCEL:
 177                                 frescan_manager_cancel(req_data);
 178                                 break;
 179                         case FRESCAN_REP_NEG:
 180                                 frescan_manager_repneg(req_data);
 181                                 break;
 182                         default:
 183                                 ERROR("request type not supported\n");
 184                                 assert(0);
 185                 }
 186
 187                 if(req_data->request_node != the_networks[net].local_node) {
 188                         ret = frescan_requests_free(req); // acceptor request
 189                         assert(ret == 0);
 190                 }
 191         }
 192 }
 193
 194 /**
 195  * frescan_acceptor_thread()
 196  */
 197
 198 static void *frescan_acceptor_thread(void *arg)
 199 {
 200         int ret;
 201         frescan_request_id_t req;
 202         frescan_network_t net = (uint32_t)arg;
 203
 204         DEBUG(FRESCAN_ACCEPTOR_ENABLE_DEBUG, "acceptor thread starts\n");
 205
 206         while(1) {
 207                 ret = frescan_messages_recv_request(net, &req);
 208                 assert(ret == 0);
 209
 210                 ret = frescan_requests_enqueue(req);
 211                 assert(ret == 0);
 212         }
 213
 214         return NULL;
 215 }
 216
 217 /**
 218  * frescan_manager_neg
 219  */
 220
 221 static void frescan_manager_neg(frescan_request_data_t *req_data)
 222 {
 223         int ret;
 224         bool accepted;
 225
 226         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "negotiation request\n");
 227
 228         if (the_networks[req_data->net].local_node == FRESCAN_NEG_MASTER_NODE) {
 229                 // scheduling analysis
 230                 ret = frescan_sa_add_contract
 231                                 (&the_networks[req_data->net].scenario,
 232                                  req_data->contract,
 233                                  req_data->ss,
 234                                  req_data->request_node);
 235                 assert(ret == 0);
 236
 237                 ret = frescan_sa_sched_test
 238                                 (&the_networks[req_data->net].scenario,
 239                                  &accepted);
 240                 assert(ret == 0);
 241
 242                 if (accepted) {
 243                         ret = frescan_sa_get_final_values
 244                                         (&the_networks[req_data->net].scenario,
 245                                          req_data->ss,
 246                                          req_data->request_node,
 247                                          &req_data->final_values);
 248                         assert(ret == 0);
 249                         req_data->return_value = FRESCAN_REQ_ACCEPTED;
 250                 } else {
 251                         ret = frescan_sa_remove_contract
 252                                         (&the_networks[req_data->net].scenario,
 253                                          req_data->ss,
 254                                          req_data->request_node);
 255                         assert(ret == 0);
 256                         req_data->return_value = FRESCAN_REQ_NOT_ACCEPTED;
 257                 }
 258
 259                 // signal or reply the results
 260                 if (req_data->request_node == FRESCAN_NEG_MASTER_NODE) {
 261                         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "master local\n");
 262                         ret = frescan_bwres_robjs_signal(req_data->robj);
 263                         assert(ret == 0);
 264                 } else {
 265                         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG,
 266                               "master external, sending reply\n");
 267
 268                         req_data->type = FRESCAN_REP_NEG;
 269                         ret = frescan_messages_send_request(req_data);
 270                         assert(ret == 0);
 271                 }
 272         } else {
 273                 DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG,
 274                       "send negotiation request to master\n");
 275                 ret = frescan_messages_send_request(req_data);
 276                 assert(ret == 0);
 277         }
 278 }
 279
 280 /**
 281  * frescan_manager_reneg
 282  */
 283
 284 static void frescan_manager_reneg(frescan_request_data_t *req_data)
 285 {
 286         int ret;
 287         bool accepted;
 288         frescan_contract_t old_contract;
 289
 290         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "renegotiation request\n");
 291
 292         if (the_networks[req_data->net].local_node == FRESCAN_NEG_MASTER_NODE) {
 293                 // scheduling analysis
 294                 ret = frescan_sa_update_contract
 295                                 (&the_networks[req_data->net].scenario,
 296                                  req_data->ss,
 297                                  req_data->request_node,
 298                                  req_data->contract,
 299                                  &old_contract);
 300                 assert(ret == 0);
 301
 302                 ret = frescan_sa_sched_test
 303                                 (&the_networks[req_data->net].scenario,
 304                                  &accepted);
 305                 assert(ret == 0);
 306
 307                 if (accepted) {
 308                         ret = frescan_sa_get_final_values
 309                                         (&the_networks[req_data->net].scenario,
 310                                           req_data->ss,
 311                                           req_data->request_node,
 312                                           &req_data->final_values);
 313                         assert(ret == 0);
 314                         req_data->return_value = FRESCAN_REQ_ACCEPTED;
 315                 } else {
 316                         ret = frescan_sa_update_contract
 317                                         (&the_networks[req_data->net].scenario,
 318                                          req_data->ss,
 319                                          req_data->request_node,
 320                                          &old_contract,
 321                                          NULL);
 322                         assert(ret == 0);
 323                         req_data->return_value = FRESCAN_REQ_NOT_ACCEPTED;
 324                 }
 325
 326                 // signal or reply the results
 327                 if (req_data->request_node == FRESCAN_NEG_MASTER_NODE) {
 328                         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "master local\n");
 329                         ret = frescan_bwres_robjs_signal(req_data->robj);
 330                         assert(ret == 0);
 331                 } else {
 332                         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG,
 333                               "master external, sending reply\n");
 334
 335                         req_data->type = FRESCAN_REP_NEG;
 336                         ret = frescan_messages_send_request(req_data);
 337                         assert(ret == 0);
 338                 }
 339         } else {
 340                 DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG,
 341                       "send renegotiation request to master\n");
 342                 ret = frescan_messages_send_request(req_data);
 343                 assert(ret == 0);
 344         }
 345 }
 346
 347 /**
 348  * frescan_manager_cancel
 349  */
 350
 351 static void frescan_manager_cancel(frescan_request_data_t *req_data)
 352 {
 353         int ret;
 354
 355         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "cancel request\n");
 356
 357         if (the_networks[req_data->net].local_node == FRESCAN_NEG_MASTER_NODE) {
 358                 ret = frescan_sa_remove_contract
 359                                 (&the_networks[req_data->net].scenario,
 360                                  req_data->ss,
 361                                  req_data->request_node);
 362                 assert(ret == 0);
 363         } else {
 364                 DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG,
 365                       "send cancel request to master\n");
 366                 ret = frescan_messages_send_request(req_data);
 367                 assert(ret == 0);
 368         }
 369
 370         if (req_data->request_node == the_networks[req_data->net].local_node) {
 371                 ret = frescan_bwres_robjs_signal(req_data->robj);
 372                 assert(ret == 0);
 373         }
 374 }
 375
 376 /**
 377  * frescan_manager_repneg
 378  */
 379
 380 static void frescan_manager_repneg(frescan_request_data_t *req_data)
 381 {
 382         int ret;
 383         frescan_request_data_t *neg_req_data;
 384
 385         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "reply to neg request\n");
 386
 387         ret = frescan_requests_get_data(req_data->req, &neg_req_data);
 388         assert(ret == 0);
 389
 390         neg_req_data->return_value = req_data->return_value;
 391         neg_req_data->final_values = req_data->final_values;
 392
 393         ret = frescan_bwres_robjs_signal(neg_req_data->robj);
 394         assert(ret == 0);
 395 }