Added fwp_vres_tx_thread cancelation

diff --git a/fwp/libfwp/src/fwp_vres.c b/fwp/libfwp/src/fwp_vres.c
index 28af82da350c2d56573c19f208cff0a77451c47c..36557f40d77cf5e2bd211cd9b3b8247c511e3f87 100644 (file)
--- a/fwp/libfwp/src/fwp_vres.c
+++ b/fwp/libfwp/src/fwp_vres.c
@@ -95,6 +95,7 @@ int fwp_vres_open(struct fwp_contract *cnt)
 int fwp_vres_close(unsigned int id)
 {      
        struct fwp_vres *vres = &fwp_vres_table[id];
+       struct fwp_msgb *msgb;
        int rc;
 
        if ((id < 0) || (id >= FWP_VRES_MAX))
@@ -108,17 +109,24 @@ int fwp_vres_close(unsigned int id)
        
        vres->status = FWP_VRES_CLOSED;
        
+       /* locking not needed*/
+       
+       while ((msgb = fwp_msgq_dequeue(&vres->tx_queue))) { 
+               fwp_msgb_free(msgb);
+       }
+
        FWP_DEBUG("Vres id %d closed.\n", id);  
        return  0;
-       //TODO: pthread_cancel(vres->tx_thread,); 
 }
 
 inline int fwp_vres_send(unsigned int id, struct fwp_msgb* msgb)
 {
-       return fwp_msgq_enqueue(&fwp_vres_table[id].tx_queue, msgb);
+       if (fwp_vres_table[id].status == FWP_VRES_OPENED) {
+               return fwp_msgq_enqueue(&fwp_vres_table[id].tx_queue, msgb);
+       }else 
+               return -EPERM;
 }
 
-
 static void* fwp_vres_tx_thread(void *_vres)
 {
        struct fwp_vres *vres = (struct fwp_vres*)_vres;
@@ -140,18 +148,17 @@ static void* fwp_vres_tx_thread(void *_vres)
        FWP_DEBUG("vres tx thread with budget:%d period_usec = %d started.\n", 
                vres->contract.budget, vres->contract.period_usec);
 
-       while (1) {
+       while (vres->status != FWP_VRES_CLOSED) {
                /* TODO-consider: block on semaphore if there is no msg */
                clock_gettime(CLOCK_MONOTONIC, &start_period);
                
-       //      printf("Start in nsec: %lu\n", start_period.tv_nsec);   
                msgb = fwp_msgq_dequeue(msgq);
                
                if (msgb) {
                        /* TODO: rc handling */
                        rc = fwp_ac_send(ac_id, msgb);
                        FWP_DEBUG("Message sent through AC %d\n",ac_id);
-                       //fwp_msgb_free(msgb);
+                       fwp_msgb_free(msgb);
                }
 
                fwp_timespec_add(&end_period, &start_period, &period);
@@ -159,4 +166,6 @@ static void* fwp_vres_tx_thread(void *_vres)
                fwp_timespec_sub(&interval, &end_period, &current_time);
                nanosleep(&interval, NULL);
        }
+       
+       return NULL;
 }