some reorganization... now the acceptor is simpler.. everything goes through requests...

[frescor/fna.git] / src_frescan / frescan_bwres_threads.c

diff --git a/src_frescan/frescan_bwres_threads.c b/src_frescan/frescan_bwres_threads.c
index 2f78a87ddc257711d9c56bff0abe430c5c6ed069..a083d3eb060c776e461eacbc002b2244a8ae6600 100644 (file)
--- a/src_frescan/frescan_bwres_threads.c
+++ b/src_frescan/frescan_bwres_threads.c
@@ -92,10 +92,10 @@ int frescan_acceptor_thread_create(frescan_network_t net)
  * frescan_manager_thread
  */
 
-static int frescan_manager_neg(frescan_request_data_t *req_data);
-static int frescan_manager_reneg(frescan_request_data_t *req_data);
-static int frescan_manager_cancel(frescan_request_data_t *req_data);
-static int frescan_manager_repneg(frescan_request_data_t *req_data);
+static void frescan_manager_neg(frescan_request_data_t *req_data);
+static void frescan_manager_reneg(frescan_request_data_t *req_data);
+static void frescan_manager_cancel(frescan_request_data_t *req_data);
+static void frescan_manager_repneg(frescan_request_data_t *req_data);
 
 static void *frescan_manager_thread(void *arg)
 {
@@ -117,20 +117,16 @@ static void *frescan_manager_thread(void *arg)
 
                 switch(req_data->type) {
                         case FRESCAN_REQ_NEG:
-                                ret = frescan_manager_neg(req_data);
-                                assert(ret == 0);
+                                frescan_manager_neg(req_data);
                                 break;
                         case FRESCAN_REQ_RENEG:
-                                ret = frescan_manager_reneg(req_data);
-                                assert(ret == 0);
+                                frescan_manager_reneg(req_data);
                                 break;
                         case FRESCAN_REQ_CANCEL:
-                                ret = frescan_manager_cancel(req_data);
-                                assert(ret == 0);
+                                frescan_manager_cancel(req_data);
                                 break;
                         case FRESCAN_REP_NEG:
-                                ret = frescan_manager_repneg(req_data);
-                                assert(ret == 0);
+                                frescan_manager_repneg(req_data);
                                 break;
                         default:
                                 ERROR("request type not supported\n");
@@ -151,48 +147,15 @@ static void *frescan_manager_thread(void *arg)
 static void *frescan_acceptor_thread(void *arg)
 {
         int ret;
-        frescan_recv_params_t recv_params;
-        uint8_t msg[2000]; // TODO: use a constant with the max neg message size
-        size_t recv_bytes;
-        frescan_node_t from;
-        frescan_prio_t prio;
         frescan_request_id_t req;
-        frescan_request_data_t *req_data;
-
-        DEBUG(FRESCAN_ACCEPTOR_THREAD_ENABLE_DEBUG, "acceptor th starts\n");
+        frescan_network_t net = (uint32_t)arg;
 
-        recv_params.net     = (frescan_network_t)(uint32_t)arg;
-        recv_params.channel = FRESCAN_NEG_CHANNEL;
-        recv_params.flags   = FRESCAN_SYNC;
+        DEBUG(FRESCAN_ACCEPTOR_ENABLE_DEBUG, "acceptor thread starts\n");
 
         while(1) {
-                DEBUG(FRESCAN_ACCEPTOR_THREAD_ENABLE_DEBUG,
-                      "waiting for a msg, net:%u chan:%u flags:%u\n",
-                      recv_params.net, recv_params.channel, recv_params.flags);
-
-                ret = frescan_recv(&recv_params,
-                                   msg,
-                                   sizeof(msg),
-                                   &recv_bytes,
-                                   &from,
-                                   &prio);
-                assert(ret == 0);
-
-                DEBUG(FRESCAN_ACCEPTOR_THREAD_ENABLE_DEBUG,
-                      "msg received, from:%u size:%u prio:%u\n",
-                      from, recv_bytes, prio);
-
-                ret = frescan_requests_alloc(&req);
+                ret = frescan_messages_recv_request(net, &req);
                 assert(ret == 0);
 
-                ret = frescan_requests_get_data(req, &req_data);
-                assert(ret == 0);
-
-                ret = frescan_message_to_request(msg, req_data);
-                assert(ret == 0);
-
-                req_data->request_node = from;
-
                 ret = frescan_requests_enqueue(req);
                 assert(ret == 0);
         }
@@ -204,38 +167,66 @@ static void *frescan_acceptor_thread(void *arg)
  * frescan_manager_neg
  */
 
-static int frescan_manager_neg(frescan_request_data_t *req_data)
+static void frescan_manager_neg(frescan_request_data_t *req_data)
 {
+        int ret;
+        bool accepted;
+
         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "negotiation request\n");
-        return 0;
+
+        if (the_networks[req_data->net].local_node == FRESCAN_NEG_MASTER_NODE) {
+
+                accepted = true; // TODO: sched analysis
+                ret = 0;  // TODO: sched analysis
+
+                if (accepted) {
+                        req_data->return_value = FRESCAN_REQ_ACCEPTED;
+                } else {
+                        req_data->return_value = (ret == 0) ?
+                                                  FRESCAN_REQ_NOT_ACCEPTED :
+                                                  FRESCAN_REQ_ERROR;
+                }
+
+                if (req_data->request_node == FRESCAN_NEG_MASTER_NODE) {
+                        DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "local\n");
+                        ret = frescan_bwres_robjs_signal(req_data->robj);
+                        assert(ret == 0);
+                } else {
+                        DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "external\n");
+                        req_data->type = FRESCAN_REP_NEG;
+
+                        ret = frescan_messages_send_request(req_data);
+                        assert(ret == 0);
+                }
+        } else {
+                ret = frescan_messages_send_request(req_data);
+                assert(ret == 0);
+        }
 }
 
 /**
  * frescan_manager_neg
  */
 
-static int frescan_manager_reneg(frescan_request_data_t *req_data)
+static void frescan_manager_reneg(frescan_request_data_t *req_data)
 {
         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "renegotiation request\n");
-        return 0;
 }
 
 /**
  * frescan_manager_neg
  */
 
-static int frescan_manager_cancel(frescan_request_data_t *req_data)
+static void frescan_manager_cancel(frescan_request_data_t *req_data)
 {
         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "cancel request\n");
-        return 0;
 }
 
 /**
  * frescan_manager_neg
  */
 
-static int frescan_manager_repneg(frescan_request_data_t *req_data)
+static void frescan_manager_repneg(frescan_request_data_t *req_data)
 {
         DEBUG(FRESCAN_MANAGER_ENABLE_DEBUG, "reply to neg request\n");
-        return 0;
 }