Add API for coordniator callback instances

Allows callbacks to register an instance create function to create instance data and a set of callbacks to run as part of the event loop.

diff --git a/src/lib/io/coord.c b/src/lib/io/coord.c
index 2e222210dfeeed012c1cb7f43842a84be013fe1d..24d93d4f70a64df6ff657557e66ef3c902a6382f 100644 (file)
--- a/src/lib/io/coord.c
+++ b/src/lib/io/coord.c
@@ -53,6 +53,7 @@ struct fr_coord_s {
        fr_rb_node_t                    node;           //!< Entry in the tree of coordinators.
        fr_coord_cb_reg_t               *callbacks;     //!< Array of callbacks for worker -> coordinator messages.
        uint32_t                        num_callbacks;  //!< Number of callbacks defined.
+       fr_coord_cb_inst_t              **cb_inst;      //!< Array of callback instance specific data.
 
        uint32_t                        max_workers;    //!< Maximum number of workers we expect.
        uint32_t                        num_workers;    //!< How many workers are attached.
@@ -223,7 +224,10 @@ static void coord_data_recv(void *ctx, void const *data, size_t data_size, fr_ti
        }
 
        fr_dbuff_init(&dbuff, (uint8_t const *)cd->m.data, cd->m.data_size);
-       coord->callbacks[cd->coord_cb_id].callback(coord, cm.worker, &dbuff, now, coord->callbacks[cd->coord_cb_id].uctx);
+       coord->callbacks[cd->coord_cb_id].callback(coord, cm.worker, &dbuff, now,
+                                                  coord->cb_inst[cd->coord_cb_id] ?
+                                                  coord->cb_inst[cd->coord_cb_id]->inst_data : NULL,
+                                                  coord->callbacks[cd->coord_cb_id].uctx);
        fr_message_done(&cd->m);
 }
 
@@ -378,6 +382,15 @@ static fr_coord_t *fr_coord_create(TALLOC_CTX *ctx, fr_event_list_t *el, fr_coor
        MEM(coord->coord_send_control = talloc_zero_array(coord, fr_control_t *, coord->max_workers));
        MEM(coord->coord_send_aq = talloc_zero_array(coord, fr_atomic_queue_t *, coord->max_workers));
 
+       MEM(coord->cb_inst = talloc_zero_array(coord, fr_coord_cb_inst_t *, coord->num_callbacks));
+
+       for (i = 0; i < coord->num_callbacks; i++) {
+               if (!coord->callbacks[i].inst_create) continue;
+               coord->cb_inst[i] = coord->callbacks[i].inst_create(coord, coord, coord->el, coord->single_thread,
+                                                                   coord->callbacks[i].uctx);
+               if (!coord->cb_inst[i]) goto fail;
+       }
+
        return coord;
 }
 
@@ -385,6 +398,9 @@ static fr_coord_t *fr_coord_create(TALLOC_CTX *ctx, fr_event_list_t *el, fr_coor
  */
 static void fr_coordinate(fr_coord_t *coord)
 {
+       uint32_t                i;
+       fr_coord_cb_inst_t      *cb_inst;
+
        /*
         *      Run until we're told to exit AND the number of
         *      workers has dropped to zero.
@@ -414,6 +430,14 @@ static void fr_coordinate(fr_coord_t *coord)
                        DEBUG4("Servicing event(s)");
                        fr_event_service(coord->el);
                }
+
+               /*
+                *      Run any registered instance specific event callbacks
+                */
+               for (i = 0; i < coord->num_callbacks; i++) {
+                       cb_inst = coord->cb_inst[i];
+                       if (cb_inst && cb_inst->event_cb) cb_inst->event_cb(coord->el, cb_inst->inst_data);
+               }
        }
 
        return;
@@ -768,3 +792,53 @@ int fr_worker_to_coord_send(fr_coord_worker_t *cw, uint32_t cb_id, fr_dbuff_t *d
        return fr_control_message_send(cw->coord->coord_recv_control, cw->worker_send_rb,
                                       FR_CONTROL_ID_COORD_DATA, &cm, sizeof(fr_coord_msg_t));
 }
+
+/** Insert instance specific pre-event callbacks
+ */
+int fr_coord_pre_event_insert(fr_event_list_t *el)
+{
+       fr_coord_t              *coord;
+       fr_rb_iter_inorder_t    iter;
+       fr_coord_cb_inst_t      *cb_inst;
+       uint32_t                i;
+
+       if (!coord_regs) return 0;
+
+       for (coord = fr_rb_iter_init_inorder(&coords, &iter);
+            coord != NULL;
+            coord = fr_rb_iter_next_inorder(&coords, &iter)) {
+               for (i = 0; i < coord->num_callbacks; i++) {
+                       cb_inst = coord->cb_inst[i];
+                       if (cb_inst && cb_inst->event_pre_cb &&
+                           fr_event_pre_insert(el, cb_inst->event_pre_cb, cb_inst->inst_data) < 0) {
+                               return -1;
+                       }
+               }
+       }
+       return 0;
+}
+
+/** Insert instance specific post-event callbacks
+ */
+int fr_coord_post_event_insert(fr_event_list_t *el)
+{
+       fr_coord_t              *coord;
+       fr_rb_iter_inorder_t    iter;
+       fr_coord_cb_inst_t      *cb_inst;
+       uint32_t                i;
+
+       if (!coord_regs) return 0;
+
+       for (coord = fr_rb_iter_init_inorder(&coords, &iter);
+            coord != NULL;
+            coord = fr_rb_iter_next_inorder(&coords, &iter)) {
+               for (i = 0; i < coord->num_callbacks; i++) {
+                       cb_inst = coord->cb_inst[i];
+                       if (cb_inst && cb_inst->event_post_cb &&
+                           fr_event_post_insert(el, cb_inst->event_post_cb, cb_inst->inst_data) < 0) {
+                               return -1;
+                       }
+               }
+       }
+       return 0;
+}

diff --git a/src/lib/io/coord.h b/src/lib/io/coord.h
index 7cb0854d2ff3d4906a9d28a0336883d84890ac08..f93afd986e308bf1c181410b61234ea58b8be6ec 100644 (file)
--- a/src/lib/io/coord.h
+++ b/src/lib/io/coord.h
@@ -35,13 +35,16 @@ typedef struct fr_coord_reg_s fr_coord_reg_t;
 
 typedef struct fr_coord_s fr_coord_t;
 typedef struct fr_coord_worker_s fr_coord_worker_t;
+typedef struct fr_coord_cb_inst_s fr_coord_cb_inst_t;
 
-typedef        void (*fr_coord_cb_t)(fr_coord_t *coord, uint32_t worker_id, fr_dbuff_t *dbuff, fr_time_t now, void *uctx);
+typedef        void (*fr_coord_cb_t)(fr_coord_t *coord, uint32_t worker_id, fr_dbuff_t *dbuff, fr_time_t now, void *plugin_data, void *uctx);
 typedef void (*fr_coord_worker_cb_t)(fr_coord_worker_t *cw, fr_dbuff_t *dbuff, fr_time_t now, void *uctx);
+typedef fr_coord_cb_inst_t *(*fr_coord_cb_inst_create_t)(TALLOC_CTX *ctx, fr_coord_t *coord, fr_event_list_t *el, bool single_thread, void *uctx);
 
 typedef struct {
        char const                      *name;
        fr_coord_cb_t                   callback;
+       fr_coord_cb_inst_create_t       inst_create;
        void                            *uctx;
 } fr_coord_cb_reg_t;
 
@@ -91,3 +94,7 @@ int           fr_coord_to_worker_send(fr_coord_t *coord, uint32_t worker_id, uint32_t cb_
 int            fr_coord_to_worker_broadcast(fr_coord_t *coord, uint32_t cb_id, fr_dbuff_t *dbuff);
 
 int            fr_worker_to_coord_send(fr_coord_worker_t *cw, uint32_t cb_id, fr_dbuff_t *dbuff);
+
+int            fr_coord_pre_event_insert(fr_event_list_t *el);
+
+int            fr_coord_post_event_insert(fr_event_list_t *el);

diff --git a/src/lib/io/coord_priv.h b/src/lib/io/coord_priv.h
index 105b394c3d8ed91691529747fd4421eb48ac9342..6f4485e967a55922b0eb53b0e974973ec413e3d9 100644 (file)
--- a/src/lib/io/coord_priv.h
+++ b/src/lib/io/coord_priv.h
@@ -39,3 +39,11 @@ typedef struct {
        fr_message_t                    m;                      //!< Message containing data being sent.
        uint32_t                        coord_cb_id;            //!< Callback ID for this message.
 } fr_coord_data_t;
+
+typedef void (*fr_coord_inst_event_cb_t)(fr_event_list_t *el, void *uctx);
+struct fr_coord_cb_inst_s {
+       void                            *inst_data;             //!< Instance data.
+       fr_event_status_cb_t            event_pre_cb;           //!< Pre-event callback in single thread mode.
+       fr_event_post_cb_t              event_post_cb;          //!< Post-event callback in single thread mode.
+       fr_coord_inst_event_cb_t        event_cb;               //!< Event callback in multi thread mode.
+};