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  */
  16
  17 #include <assert.h>
  18 #include "fosa_threads_and_signals.h" // fosa_thread_attr_init...
  19 #include "frescan_bwres_threads.h"
  20 #include "frescan_bwres_messages.h"
  21 #include "frescan_bwres_requests.h"
  22 #include "frescan_config.h"
  23 #include "frescan_debug.h"
  24 #include "frescan_data.h"
  25 #include "frescan_servers.h"
  26
  27 /**
  28  * frescan_manager_thread_create()
  29  *
  30  * This call creates the manager thread at each node which will be waiting
  31  * in a request queue for LOCAL or EXTERNAL requests.
  32  */
  33
  34 static void *frescan_manager_thread(void *arg);
  35
  36 int frescan_manager_thread_create(frescan_network_t net)
  37 {
  38         int ret;
  39         fosa_thread_attr_t attr;
  40
  41         ret = fosa_thread_attr_init(&attr);
  42         if (ret != 0) return ret;
  43
  44         ret = fosa_thread_attr_set_prio(&attr, FRESCAN_NEG_THREAD_PRIO);
  45         if (ret != 0) return ret;
  46
  47         ret = fosa_thread_create(&the_networks[net].manager_thread_id,
  48                                  &attr,
  49                                  frescan_manager_thread,
  50                                  (void *)(uint32_t)net);
  51         if (ret != 0) return ret;
  52
  53         ret = fosa_thread_attr_destroy(&attr);
  54         if (ret != 0) return ret;
  55
  56         return 0;
  57 }
  58
  59 /**
  60  * frescan_acceptor_thread_create()
  61  *
  62  * This call creates the acceptor thread which will be waiting negotiation
  63  * messages from the network and converting them into requests.
  64  */
  65
  66 static void *frescan_acceptor_thread(void *arg);
  67
  68 int frescan_acceptor_thread_create(frescan_network_t net)
  69 {
  70         int ret;
  71         fosa_thread_attr_t attr;
  72
  73         ret = fosa_thread_attr_init(&attr);
  74         if (ret != 0) return ret;
  75
  76         ret = fosa_thread_attr_set_prio(&attr, FRESCAN_ACCEPTOR_THREAD_PRIO);
  77         if (ret != 0) return ret;
  78
  79         ret = fosa_thread_create(&the_networks[net].acceptor_thread_id,
  80                                  &attr,
  81                                  frescan_acceptor_thread,
  82                                  (void *)(uint32_t)net);
  83         if (ret != 0) return ret;
  84
  85         ret = fosa_thread_attr_destroy(&attr);
  86         if (ret != 0) return ret;
  87
  88         return 0;
  89 }
  90
  91 /**
  92  * frescan_manager_thread
  93  */
  94
  95 static void frescan_manager_neg(frescan_request_data_t *req_data);
  96 static void frescan_manager_reneg(frescan_request_data_t *req_data);
  97 static void frescan_manager_cancel(frescan_request_data_t *req_data);
  98 static void frescan_manager_repneg(frescan_request_data_t *req_data);
  99
 100 static void *frescan_manager_thread(void *arg)
 101 {
 102         int ret;
 103         frescan_request_id_t   req;
 104         frescan_request_data_t *req_data;
 105         frescan_network_t net = (uint32_t)arg;
 106
 107         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "manager thread starts\n");
 108
 109         while(1) {
 110                 DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "wait for a request\n");
 111
 112                 ret = frescan_requests_dequeue(&req);
 113                 assert(ret == 0);
 114
 115                 ret = frescan_requests_get_data(req, &req_data);
 116                 assert(ret == 0);
 117
 118                 switch(req_data->type) {
 119                         case FRESCAN_REQ_NEG:
 120                                 frescan_manager_neg(req_data);
 121                                 break;
 122                         case FRESCAN_REQ_RENEG:
 123                                 frescan_manager_reneg(req_data);
 124                                 break;
 125                         case FRESCAN_REQ_CANCEL:
 126                                 frescan_manager_cancel(req_data);
 127                                 break;
 128                         case FRESCAN_REP_NEG:
 129                                 frescan_manager_repneg(req_data);
 130                                 break;
 131                         default:
 132                                 ERROR("request type not supported\n");
 133                                 assert(0);
 134                 }
 135
 136                 if(req_data->request_node != the_networks[net].local_node) {
 137                         ret = frescan_requests_free(req);
 138                         assert(ret == 0);
 139                 }
 140         }
 141 }
 142
 143 /**
 144  * frescan_acceptor_thread()
 145  */
 146
 147 static void *frescan_acceptor_thread(void *arg)
 148 {
 149         int ret;
 150         frescan_request_id_t req;
 151         frescan_network_t net = (uint32_t)arg;
 152
 153         DEBUG(FRESCAN_ACCEPTOR_ENABLE_DEBUG, "acceptor thread starts\n");
 154
 155         while(1) {
 156                 ret = frescan_messages_recv_request(net, &req);
 157                 assert(ret == 0);
 158
 159                 ret = frescan_requests_enqueue(req);
 160                 assert(ret == 0);
 161         }
 162
 163         return NULL;
 164 }
 165
 166 /**
 167  * frescan_manager_neg
 168  */
 169
 170 static void frescan_manager_neg(frescan_request_data_t *req_data)
 171 {
 172         int ret;
 173         bool accepted;
 174
 175         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "negotiation request\n");
 176
 177         if (the_networks[req_data->net].local_node == FRESCAN_NEG_MASTER_NODE) {
 178
 179                 // NOTE: req_data->ss is an INPUT parameter
 180                 accepted = true; // TODO: sched analysis
 181                 ret = 0;  // TODO: sched analysis
 182
 183                 if (accepted) {
 184                         req_data->return_value = FRESCAN_REQ_ACCEPTED;
 185                 } else {
 186                         req_data->return_value = (ret == 0) ?
 187                                                   FRESCAN_REQ_NOT_ACCEPTED :
 188                                                   FRESCAN_REQ_ERROR;
 189                 }
 190
 191                 if (req_data->request_node == FRESCAN_NEG_MASTER_NODE) {
 192                         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "master local\n");
 193                         ret = frescan_bwres_robjs_signal(req_data->robj);
 194                         assert(ret == 0);
 195                 } else {
 196                         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG,
 197                               "master external, sending reply\n");
 198
 199                         req_data->type = FRESCAN_REP_NEG;
 200                         ret = frescan_messages_send_request(req_data);
 201                         assert(ret == 0);
 202                 }
 203         } else {
 204                 DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG,
 205                       "send negotiation request to master\n");
 206                 ret = frescan_messages_send_request(req_data);
 207                 assert(ret == 0);
 208         }
 209 }
 210
 211 /**
 212  * frescan_manager_neg
 213  */
 214
 215 static void frescan_manager_reneg(frescan_request_data_t *req_data)
 216 {
 217         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "renegotiation request\n");
 218 }
 219
 220 /**
 221  * frescan_manager_neg
 222  */
 223
 224 static void frescan_manager_cancel(frescan_request_data_t *req_data)
 225 {
 226         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "cancel request\n");
 227 }
 228
 229 /**
 230  * frescan_manager_neg
 231  */
 232
 233 static void frescan_manager_repneg(frescan_request_data_t *req_data)
 234 {
 235         int ret;
 236         frescan_request_data_t *neg_req_data;
 237
 238         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "reply to neg request\n");
 239
 240         ret = frescan_requests_get_data(req_data->req, &neg_req_data);
 241         assert(ret == 0);
 242
 243         neg_req_data->return_value = req_data->return_value;
 244
 245         ret = frescan_bwres_robjs_signal(neg_req_data->robj);
 246         assert(ret == 0);
 247 }