forb: Add forb_executor_synchronize()

This is used to wait for executors to finish processing ongoing requests
during FORB termination. This ensures that all peer sockets are closed
and wvtest does not complain about open file descriptors at the end of
tests.

diff --git a/src/executor.c b/src/executor.c
index 08e00680f448689f782d1b8b68cd167a3484cc96..b24db98b55c00ba1e130a5d70acf2f4b23c285b7 100644 (file)
--- a/src/executor.c
+++ b/src/executor.c
@@ -88,6 +88,9 @@ int forb_executor_init(forb_executor_t *executor)
        forb_exec_req_nolock_init_head(executor);
        
        forb_syncobj_init(&executor->reply_processed, 0);
+
+       executor->finish = false;
+       fosa_cond_init(&executor->finished);
        return 0;
 }
 
@@ -96,6 +99,21 @@ void forb_executor_destroy(forb_executor_t *executor)
        /* TODO: */
 }
 
+/** 
+ * Waits for the executor to finish processing requests.
+ * 
+ * @param executor 
+ */
+void forb_executor_synchronize(forb_executor_t *executor)
+{
+       fosa_mutex_lock(&executor->mutex);
+       executor->finish = true;
+       fosa_cond_signal(&executor->new_request_in_empty_list);
+       fosa_cond_wait(&executor->finished, &executor->mutex);
+       fosa_mutex_unlock(&executor->mutex);
+
+}
+
 /** 
  * Setup the object @a obj so that requests to it are executed within
  * the thread of the @a executor.
@@ -159,6 +177,12 @@ int forb_executor_run(forb_executor_t *executor)
 
                        fosa_mutex_lock(&executor->mutex);
                }
+               if (executor->finish) {
+                       fosa_cond_broadcast(&executor->finished);
+                       executor->finish = false;
+               }
+
+                       
                fosa_cond_wait(&executor->new_request_in_empty_list,
                               &executor->mutex);
        }

diff --git a/src/executor.h b/src/executor.h
index 2c3cdc388cd3bf28befe8abe2bb218d14260f91d..16bd8e10db95af59b78047783af940f6bfdf8b24 100644 (file)
--- a/src/executor.h
+++ b/src/executor.h
@@ -60,7 +60,7 @@
 #include <ul_list.h>
 #include <forb/object_type.h>
 #include <forb/syncobj.h>
-
+#include <stdbool.h>
 
 /**
  * Executor structure.
@@ -76,7 +76,9 @@ typedef struct forb_executor {
        fosa_mutex_t mutex;     /**< Mutex for protecting forb_executor_t::requests. */
        fosa_cond_t new_request_in_empty_list; /**< Signaled when a request was added to the empty list. */
        ul_list_head_t requests; /**< List of pending requests for this executor. */
-       forb_syncobj_t reply_processed;   /**< Synchronization object for signaling the receiver thread to continue processing after the reply is processed by a stub. */       
+       forb_syncobj_t reply_processed;   /**< Synchronization object for signaling the receiver thread to continue processing after the reply is processed by a stub. */
+       bool finish;            /**< Signals the executor to finish */
+       fosa_cond_t finished;
 } forb_executor_t;
 
 
@@ -86,6 +88,7 @@ void forb_executor_destroy(forb_executor_t *executor);
 int forb_executor_register_object(forb_executor_t *executor, forb_object obj);
 void forb_executor_unregister_object(forb_executor_t *executor, forb_object obj);
 int forb_executor_run(forb_executor_t *executor);
+void forb_executor_synchronize(forb_executor_t *executor);
 
 int forb_execute_object(forb_object obj);
 forb_executor_t *forb_get_current_executor(void);