Fix descriptor mutex errno reporting and synchronize probe workers (#2481)

src/OVAL/probes/SEAP/seap-descriptor.h

@@ -82,37 +82,36 @@ void SEAP_desc_free(SEAP_desc_t *dsc);
 
 static inline int SEAP_desc_trylock (pthread_mutex_t *m)
 {
-        switch (pthread_mutex_trylock (m)) {
+        int ret = pthread_mutex_trylock (m);
+        switch (ret) {
         case 0:
                 return (1);
         case EBUSY:
                 return (0);
-        case EINVAL:
-                errno = EINVAL;
-                /* FALLTHROUGH */
         default:
+                errno = ret;
                 return (-1);
         }
 }
 
 static inline int SEAP_desc_lock (pthread_mutex_t *m)
 {
-        switch (pthread_mutex_lock (m)) {
-        case 0:
-                return (1);
-        default:
+        int ret = pthread_mutex_lock (m);
+        if (ret != 0) {
+                errno = ret;
                 return (-1);
         }
+        return (1);
 }
 
 static inline int SEAP_desc_unlock (pthread_mutex_t *m)
 {
-        switch (pthread_mutex_unlock (m)) {
-        case 0:
-                return (1);
-        default:
+        int ret = pthread_mutex_unlock (m);
+        if (ret != 0) {
+                errno = ret;
                 return (-1);
         }
+        return (1);
 }
 
 #define DESC_TRYRLOCK(d) SEAP_desc_trylock (&((d)->r_lock))

src/OVAL/probes/probe/input_handler.c

@@ -180,9 +180,11 @@ void *probe_input_handler(void *arg)
 					} else {
 						/* OK */
 
+						probe_worker_start(probe);
 						if (pthread_create(&pair->pth->tid, &pth_attr, &probe_worker_runfn, pair))
 						{
 							dE("Cannot start a new worker thread: %d, %s.", errno, strerror(errno));
+							probe_worker_end(probe);
 
 							if (rbt_i32_del(probe->workers, pair->pth->sid, NULL) != 0)
 								dE("rbt_i32_del: failed to remove worker thread (ID=%u)", pair->pth->sid);

src/OVAL/probes/probe/probe.h

@@ -69,6 +69,9 @@ typedef struct {
 	pthread_barrier_t *th_barrier;
 
         rbt_t    *workers;
+	pthread_mutex_t workers_mutex; /**< guards workers_running */
+	pthread_cond_t  workers_cond;  /**< signaled when a worker thread ends */
+	uint32_t  workers_running;     /**< worker threads that have yet to end */
         uint32_t  max_threads;
         uint32_t  max_chdepth;
 

src/OVAL/probes/probe/probe_main.c

@@ -154,6 +154,9 @@ static void probe_common_main_cleanup(void *arg)
 	}
 	dD("probe_input_handler thread has joined with status %ld", (long) status);
 
+	/* The worker threads use the SEAP descriptor and the caches until they end. */
+	probe_worker_wait(probe);
+
 	probe_fini_function_t fini_function = probe_table_get_fini_function(probe->subtype);
 	if (fini_function != NULL) {
 		fini_function(probe->probe_arg);
@@ -162,6 +165,8 @@ static void probe_common_main_cleanup(void *arg)
 	probe_rcache_free(probe->rcache);
 	probe_icache_free(probe->icache);
 	rbt_i32_free(probe->workers);
+	pthread_cond_destroy(&probe->workers_cond);
+	pthread_mutex_destroy(&probe->workers_mutex);
 	SEAP_CTX_free(probe->SEAP_ctx);
 	free(probe->option);
 
@@ -244,6 +249,9 @@ void *probe_common_main(void *arg)
 	 * Create input handler (detached)
 	 */
         probe.workers   = rbt_i32_new();
+	pthread_mutex_init(&probe.workers_mutex, NULL);
+	pthread_cond_init(&probe.workers_cond, NULL);
+	probe.workers_running = 0;
 
 	probe_init_function_t init_function = probe_table_get_init_function(probe.subtype);
 	if (init_function != NULL) {

src/OVAL/probes/probe/worker.c

@@ -79,10 +79,42 @@ static void pthread_pair_cleanup_handler(void *arg)
 	SEAP_replyerr(pair->probe->SEAP_ctx, pair->probe->sd, pair->pth->msg, -100);
 }
 
+/*
+ * The library may close the probe as soon as a worker thread has sent its
+ * reply, while the worker still unlocks the SEAP descriptor and frees its
+ * message. probe_worker_start counts a worker thread before it starts,
+ * probe_worker_end is the last use of the probe by the thread, and
+ * probe_worker_wait, called by probe_common_main_cleanup before it frees
+ * the SEAP context and the caches, returns when every worker has ended.
+ */
+void probe_worker_start(probe_t *probe)
+{
+	pthread_mutex_lock(&probe->workers_mutex);
+	probe->workers_running++;
+	pthread_mutex_unlock(&probe->workers_mutex);
+}
+
+void probe_worker_end(probe_t *probe)
+{
+	pthread_mutex_lock(&probe->workers_mutex);
+	probe->workers_running--;
+	pthread_cond_broadcast(&probe->workers_cond);
+	pthread_mutex_unlock(&probe->workers_mutex);
+}
+
+void probe_worker_wait(probe_t *probe)
+{
+	pthread_mutex_lock(&probe->workers_mutex);
+	while (probe->workers_running > 0)
+		pthread_cond_wait(&probe->workers_cond, &probe->workers_mutex);
+	pthread_mutex_unlock(&probe->workers_mutex);
+}
+
 void *probe_worker_runfn(void *arg)
 {
 	dD("probe_worker_runfn has started");
 	probe_pwpair_t *pair = (probe_pwpair_t *)arg;
+	probe_t *probe = pair->probe;
 
 	SEXP_t *probe_res, *obj, *oid;
 	int     probe_ret;
@@ -120,6 +152,7 @@ void *probe_worker_runfn(void *arg)
                 SEAP_msg_free(pair->pth->msg);
                 SEXP_free(probe_res);
                 free(pair);
+		probe_worker_end(probe);
 
 		dD("probe_worker_runfn has finished");
                 return (NULL);
@@ -187,6 +220,7 @@ void *probe_worker_runfn(void *arg)
 	pthread_detach(pthread_self());
 
 	pthread_cleanup_pop(0);
+	probe_worker_end(probe);
 
 	dD("probe_worker_runfn has finished");
 	return (NULL);

src/OVAL/probes/probe/worker.h

@@ -48,6 +48,9 @@ typedef struct {
 } probe_pwpair_t;
 
 probe_worker_t *probe_worker_new(void);
+void probe_worker_start(probe_t *probe);
+void probe_worker_end(probe_t *probe);
+void probe_worker_wait(probe_t *probe);
 void *probe_worker_runfn(void *arg);
 SEXP_t *probe_worker(probe_t *probe, SEAP_msg_t *msg_in, int *ret);