Added transports to the create method.

We may add dynamic transports later.  For now, they are statically
allocated when the application starts.

diff --git a/src/util/schedule.c b/src/util/schedule.c
index 22821970809d6bfc2d5e033c62934c94c9c682bf..b7017a663ae412d2a3888549f3a065dc68717fb0 100644 (file)
--- a/src/util/schedule.c
+++ b/src/util/schedule.c
@@ -116,6 +116,9 @@ struct fr_schedule_t {
 
        fr_heap_t       *workers;               //!< heap of workers
        fr_heap_t       *done_workers;          //!< heap of done workers
+
+       uint32_t        num_transports;         //!< how many transport layers we have
+       fr_transport_t  **transports;           //!< array of active transports.
 };
 
 
@@ -193,7 +196,7 @@ static void *fr_schedule_worker_thread(void *arg)
                return NULL;
        }
 
-       sw->worker = fr_worker_create(ctx);
+       sw->worker = fr_worker_create(ctx, sc->num_transports, sc->transports);
        if (!sw->worker) {
                talloc_free(ctx);
                goto fail;
@@ -266,6 +269,7 @@ static void *fr_schedule_worker_thread(void *arg)
  *     - fr_schedule_t new scheduler
  */
 fr_schedule_t *fr_schedule_create(TALLOC_CTX *ctx, int max_inputs, int max_workers,
+                                 uint32_t num_transports, fr_transport_t **transports,
                                  fr_schedule_thread_instantiate_t worker_thread_instantiate,
                                  void *worker_thread_ctx)
 {
@@ -290,6 +294,8 @@ fr_schedule_t *fr_schedule_create(TALLOC_CTX *ctx, int max_inputs, int max_worke
        sc->worker_instantiate_ctx = worker_thread_ctx;
 
        sc->running = true;
+       sc->num_transports = num_transports;
+       sc->transports = transports;
 
        /*
         *      No inputs or workers, we're single threaded mode.

diff --git a/src/util/schedule.h b/src/util/schedule.h
index 6dbf3cdaaee8441db431edaf82c97d4482bd2a48..c2373f016a4c2c5efffe417d8fb29eb1ed785703 100644 (file)
--- a/src/util/schedule.h
+++ b/src/util/schedule.h
@@ -35,6 +35,7 @@ typedef struct fr_schedule_t fr_schedule_t;
 typedef int (*fr_schedule_thread_instantiate_t)(void *ctx);
 
 fr_schedule_t *fr_schedule_create(TALLOC_CTX *ctx, int max_inputs, int max_workers,
+                                 uint32_t num_transports, fr_transport_t **transports,
                                  fr_schedule_thread_instantiate_t worker_thread_instantiate,
                                  void *worker_thread_ctx);
 /* schedulers are async, so there's no fr_schedule_run() */

diff --git a/src/util/worker.c b/src/util/worker.c
index 1c622feffef4bfc78c9494624fbfebed66551078..0bf3f7f5295507e9bee03b4607b94fcb5267122f 100644 (file)
--- a/src/util/worker.c
+++ b/src/util/worker.c
@@ -65,6 +65,7 @@ struct fr_worker_t {
 
        fr_time_tracking_t      tracking;       //!< how much time the worker has spent doing things.
 
+       uint32_t                num_transports; //!< how many transport layers we have
        fr_transport_t          **transports;   //!< array of active transports.
 
        fr_channel_t            *channel[1];    //!< list of channels
@@ -163,6 +164,7 @@ static REQUEST *fr_worker_decode_request(fr_worker_t *worker)
         *      Receive a message to the worker queue, and decode it
         *      to a to a request.
         */
+       rad_assert(cd->transport <= worker->num_transports);
        rad_assert(worker->transports[cd->transport] != NULL);
        request = worker->transports[cd->transport]->recv_request(worker->transports[cd->transport], cd->ctx, ctx, cd->m.data, cd->m.data_size);
 
@@ -583,10 +585,12 @@ void fr_worker_destroy(fr_worker_t *worker)
  *     - NULL on error
  *     - fr_worker_t on success
  */
-fr_worker_t *fr_worker_create(TALLOC_CTX *ctx)
+fr_worker_t *fr_worker_create(TALLOC_CTX *ctx, uint32_t num_transports, fr_transport_t **transports)
 {
        fr_worker_t *worker;
 
+       if (!num_transports || !transports) return NULL;
+
        worker = talloc_zero(ctx, fr_worker_t);
 
        worker->el = fr_event_list_create(worker, fr_worker_idle, worker);
@@ -627,6 +631,9 @@ fr_worker_t *fr_worker_create(TALLOC_CTX *ctx)
                return NULL;
        }
 
+       worker->num_transports = num_transports;
+       worker->transports = transports;
+
        return worker;
 }
 

diff --git a/src/util/worker.h b/src/util/worker.h
index 33eee0ba51c3b58d6a34e9440034e8ebed6c2098..9bf4910acbd23bfd087dba7081c66d136f77ba25 100644 (file)
--- a/src/util/worker.h
+++ b/src/util/worker.h
@@ -42,7 +42,7 @@ extern "C" {
  */
 typedef struct fr_worker_t fr_worker_t;
 
-fr_worker_t *fr_worker_create(TALLOC_CTX *ctx);
+fr_worker_t *fr_worker_create(TALLOC_CTX *ctx, uint32_t num_transports, fr_transport_t **transports);
 void fr_worker_destroy(fr_worker_t *worker) CC_HINT(nonnull);
 int fr_worker_kq(fr_worker_t *worker) CC_HINT(nonnull);
 void fr_worker(fr_worker_t *worker) CC_HINT(nonnull);