Changing the structure of fwp client,server threads

diff --git a/fwp/fwp_manager/fwpmgr.c b/fwp/fwp_manager/fwpmgr.c
index 80280873314a8486c36cf909158c2c70b98a1e13..7963f5b49c7e4334ebe595641745b2d965dc1958 100644 (file)
--- a/fwp/fwp_manager/fwpmgr.c
+++ b/fwp/fwp_manager/fwpmgr.c
@@ -152,16 +152,6 @@ void* fwp_rx_thread(void* arg)
                msgb->addrlen = peer_addrlen;
                memcpy(msgb->head, buffer, len);
                msgb->len = len;
-
-               fwp_msg_header_pull(msgb, &code, &hid);
-
-               if ((code == NEGOTIATE_REQUEST)||(code == NEGOTIATE_RESPONSE))
-                       fwp_msgq_enqueue(&msgq_neg_in, msgb);
-                       sem_post(sem_msq_in);
-               else { 
-                       printf("Error: not valid negotiate frame\n");
-                       fwp_msgb_free(msgb);
-               }
        }
 out:
        close(lsockfd_un);
@@ -169,6 +159,40 @@ out:
        return NULL;
 }
 
+
+void* fwp_server_admctrl(void *arg)
+{
+       struct fwp_msgb *msgb;
+       /* should be invoked periodicaly */
+       
+       fwp_adm_test(&ctable_negotiate);
+       while (fwp_ctable_get(&ctable_negotiate, &ctentry)) {
+               msgb = fwp_msgq_dequeue(&msgq_neg_out);
+               if (ctentry.contract.status == NEGOTIATED )
+                       fwp_ctable_put(&ctable_accepted, &ctentry);
+
+               msgb->data = msgb->head;
+               fwp_msgb_put(msgb, sizeof(struct fwp_msg_header));
+               fwp_msg_contract_put(msgb, &ctentry.contract);  
+                       
+               fwp_vres_send(vres_id, msgb);
+       };
+}
+
+void fwp_server_negotiate_request(unsigned int hid, struct fwp_msgb *msgb)
+{
+       struct fwp_ctable_entry ctentry;
+       struct fwp_contract cnt_req;
+       
+       fwp_msg_contract_pull(msgb, &cnt_req); 
+       memcpy(&ctentry.contract, &cnt_req, sizeof(cnt_req));
+       memcpy(&ctentry.saddr, &msgb->addr, sizeof(msgb->addrlen));
+       ctentry.addrlen = msgb->addrlen;
+       
+       fwp_msgq_enqueue(&msgq_neg_out, msgb);  
+       fwp_ctable_put(&ctable_negotiate, &ctentry);
+}
+
 void* fwp_server_thread(void* arg)
 {
        struct fwp_ctable_entry ctentry;
@@ -176,37 +200,26 @@ void* fwp_server_thread(void* arg)
        struct fwp_msgb *msgb;
        unsigned int code, hid;
 
+       struct fwp_msgb *msgb;
+       unsigned int code, hid;
+
        while (1) {
-               while ((msgb = fwp_msgq_dequeue(&msgq_neg_in))) {
-                       fwp_msg_header_pull(msgb, &code, &hid);
-                       
-                       /*if (h.code != NEGOTIATE_REQUEST)
-                        * printf("Error: Request frame expected\n");
-                        */
-                       fwp_msg_contract_pull(msgb, &cnt); 
-                       memcpy(&ctentry.contract, &cnt, sizeof(cnt));
-                       memcpy(&ctentry.saddr, &msgb->addr, 
-                              sizeof(msgb->addrlen));
-                       ctentry.addrlen = msgb->addrlen;
-                                       
-                       fwp_ctable_put(&ctable_negotiate, &ctentry);
-               
-                       fwp_msgq_enqueue(&msgq_neg_out, msgb);
-               }
+               sem_wait(sem_msg_in);
+               msgb = fwp_msgq_dequeue(&msgq_neg_in);
+               if (!msgb) 
+                       continue;
                        
-               fwp_adm_test(&ctable_negotiate);
-               while (fwp_ctable_get(&ctable_negotiate, &ctentry)) {
-                       msgb = fwp_msgq_dequeue(&msgq_neg_out);
-                       if (ctentry.contract.status == NEGOTIATED )
-                               fwp_ctable_put(&ctable_accepted, &ctentry);
-
-                       msgb->data = msgb->head;
-                       fwp_msgb_put(msgb, sizeof(struct fwp_msg_header));
-                       fwp_msg_contract_put(msgb, &ctentry.contract);  
+               fwp_msg_header_pull(msgb, &code, &hid);
                        
-                       fwp_vres_send(vres_id, msgb);
-               };
+               switch  (code) {
+                       case  FWP_NEGT_REQ: 
+                               fwp_server_negotiate_request(hid, msgb);        
+                               break;
 
+                       case default:
+                               printf("Invalid message\n.");
+                               fwp_msgb_free(msgb);
+               }
        }
 }
 
@@ -231,7 +244,7 @@ void fwp_client negotiate_request(unsigned int hid, struct msgb *msgb)
        fwp_vres_send(vres_id, msgb);
 }
 
-int fwp_client negotiate_response(unsigned int hid, struct msgb *msgb)
+int fwp_client_negotiate_response(unsigned int hid, struct msgb *msgb)
 {
        struct fwp_ctable_entry ctentry;
        struct fwp_contract cnt_res;
@@ -275,8 +288,12 @@ void* fwp_client_thread(void* arg)
                                break;
 
                        case FWP_NEGT_RES: 
-                               fwp_client negotiate_response(hid, msgb);
+                               fwp_client_negotiate_response(hid, msgb);
                                break;
+                       
+                       case default:
+                               printf("Invalid message\n.");
+                               fwp_msgb_free(msgb);
                }
        }
 }
@@ -288,6 +305,15 @@ int main(int argc, char* argv[])
        pthread_attr_t attr;
        
        sem_init(&sem_msg_in, 0, 0);
+
+/*      while ((opt = getopt(argc, argv, "B:b:c:g:j:o:s:T:")) != -1) {
+                             switch (opt) {
+                                                             case 'B':
+                                                                                             opt_def_bandwidth = atoi(optarg);
+                                                                                                                             break;
+
+*/
+
        mode = FWP_SERVER;
        pthread_attr_init(&attr);
        rc = pthread_create(&rx_thread_id, &attr, fwp_rx_thread, (void*) NULL);
@@ -297,7 +323,14 @@ int main(int argc, char* argv[])
        if ( mode == FWP_CLIENT )
                rc = pthread_create(&client_thread_id, &attr, fwp_client_thread, 
                                    NULL);
-
+#if 0
+        fprintf(stderr, "\nWaiting for threads to finish\n");
+               /* Wait for all threads to finish */
+                       for (i=0; i < nr_streams + AC_NUM; i++) {
+                                               sem_wait(&sem_thread_finished);
+                                                       }
+
+#endif
        return 0;
 }