Add coordinator startup / shutdown into scheduler

Author: ndptech

src/lib/io/schedule.c

@@ -110,6 +110,7 @@ struct fr_schedule_s {
 
 	fr_sem_t	*worker_sem;		//!< for inter-thread signaling
 	fr_sem_t	*network_sem;		//!< for inter-thread signaling
+	fr_sem_t	*coord_sem;		//!< for inter-thread signaling
 
 	fr_schedule_thread_instantiate_t	worker_thread_instantiate;	//!< thread instantiation callback
 	fr_schedule_thread_detach_t		worker_thread_detach;
@@ -492,6 +493,14 @@ fr_schedule_t *fr_schedule_create(TALLOC_CTX *ctx, fr_event_list_t *el,
 			return NULL;
 		}
 
+		if (fr_coords_create(sc, el) < 0) {
+			PERROR("Failed creating coordinators");
+			if (unlikely(fr_network_destroy(sc->single_network) < 0)) {
+				PERROR("Failed destroying network");
+			}
+			goto pre_instantiate_st_fail;
+		}
+
 		sc->single_worker = fr_worker_alloc(sc, el, "Worker", sc->log, sc->lvl, &sc->config->worker);
 		if (!sc->single_worker) {
 			PERROR("Failed creating worker");
@@ -545,6 +554,11 @@ fr_schedule_t *fr_schedule_create(TALLOC_CTX *ctx, fr_event_list_t *el,
 			goto st_fail;
 		}
 
+		if (fr_coord_pre_event_insert(el) < 0) {
+			fr_strerror_const("Failed adding coordinator pre-check to event list");
+			goto st_fail;
+		}
+
 		/*
 		 *	Add the event which processes request_t packets.
 		 */
@@ -553,6 +567,11 @@ fr_schedule_t *fr_schedule_create(TALLOC_CTX *ctx, fr_event_list_t *el,
 			goto st_fail;
 		}
 
+		if (fr_coord_post_event_insert(el) < 0) {
+			fr_strerror_const("Failed adding coordinator post-processing to event list");
+			goto st_fail;
+		}
+
 		return sc;
 	}
 
@@ -567,13 +586,17 @@ fr_schedule_t *fr_schedule_create(TALLOC_CTX *ctx, fr_event_list_t *el,
 	sem_fail:
 		ERROR("Failed creating semaphore: %s", fr_syserror(errno));
 		fr_sem_free(sc->network_sem);
+		fr_sem_free(sc->worker_sem);
 		talloc_free(sc);
 		return NULL;
 	}
 
 	sc->worker_sem = fr_sem_alloc();
 	if (!sc->worker_sem) goto sem_fail;
 
+	sc->coord_sem = fr_sem_alloc();
+	if (!sc->coord_sem) goto sem_fail;
+
 	/*
 	 *	Create the network threads first.
 	 */
@@ -635,6 +658,14 @@ fr_schedule_t *fr_schedule_create(TALLOC_CTX *ctx, fr_event_list_t *el,
 		return NULL;
 	}
 
+	/*
+	 *	Create the coordination threads
+	 */
+	if (fr_coord_start(sc->config->max_workers, sc->coord_sem) < 0) {
+		fr_schedule_destroy(&sc);
+		return NULL;
+	};
+
 	/*
 	 *	Create all of the workers.
 	 */
@@ -768,6 +799,7 @@ int fr_schedule_destroy(fr_schedule_t **sc_to_free)
 			ERROR("Failed destroying network");
 		}
 		fr_worker_destroy(sc->single_worker);
+		fr_coords_destroy();
 		goto done;
 	}
 
@@ -841,6 +873,7 @@ int fr_schedule_destroy(fr_schedule_t **sc_to_free)
 		}
 	}
 
+	fr_sem_free(sc->coord_sem);
 	fr_sem_free(sc->network_sem);
 	fr_sem_free(sc->worker_sem);
 done:

src/lib/io/schedule.h

@@ -38,6 +38,7 @@ typedef struct fr_schedule_s fr_schedule_t;
 #include <freeradius-devel/io/channel.h>
 #include <freeradius-devel/io/network.h>
 #include <freeradius-devel/io/worker.h>
+#include <freeradius-devel/io/coord.h>
 #include <freeradius-devel/util/log.h>
 
 #ifdef __cplusplus