allow bypassing / closing of all pipes and kqueues

if the two ends of the channel are in the same thread, just
run the callback immediately.

diff --git a/src/lib/io/control.c b/src/lib/io/control.c
index 5ed9866a5b893ede17c7db03f2324ab4fd21d831..18423ed6d56a030d418421796652f098d256c2ef 100644 (file)
--- a/src/lib/io/control.c
+++ b/src/lib/io/control.c
@@ -81,6 +81,8 @@ struct fr_control_s {
 
        int                     pipe[2];                //!< our pipes
 
+       bool                    same_thread;            //!< are the two ends in the same thread
+
        fr_control_ctx_t        type[FR_CONTROL_MAX_TYPES];     //!< callbacks
 };
 
@@ -225,7 +227,6 @@ int fr_control_gc(UNUSED fr_control_t *c, fr_ring_buffer_t *rb)
 }
 
 
-
 /** Allocate a control message
  *
  * @param[in] c the control structure
@@ -333,6 +334,13 @@ int fr_control_message_send(fr_control_t *c, fr_ring_buffer_t *rb, uint32_t id,
 {
        (void) talloc_get_type_abort(c, fr_control_t);
 
+       if (c->same_thread) {
+               if (!c->type[id].callback) return -1;
+
+               c->type[id].callback(c->type[id].ctx, data, data_size, fr_time());
+               return 0;
+       }
+
        if (fr_control_message_push(c, rb, id, data, data_size) < 0) return -1;
 
        while (write(c->pipe[1], ".", 1) == 0) {
@@ -449,3 +457,18 @@ int fr_control_callback_delete(fr_control_t *c, uint32_t id)
 
        return 0;
 }
+
+int fr_control_same_thread(fr_control_t *c)
+{
+       c->same_thread = true;
+       (void) fr_event_fd_delete(c->el, c->pipe[0], FR_EVENT_FILTER_IO);
+       close(c->pipe[0]);
+       close(c->pipe[1]);
+
+       /*
+        *      Nothing more to do now that everything is gone.
+        */
+       talloc_set_destructor(c, NULL);
+
+       return 0;
+}

diff --git a/src/lib/io/control.h b/src/lib/io/control.h
index 6236d170295b0ceb096ca9540808687886efe54f..147b9c7ef411856a2495b44b727eda4ca6d82c65 100644 (file)
--- a/src/lib/io/control.h
+++ b/src/lib/io/control.h
@@ -71,6 +71,8 @@ ssize_t fr_control_message_pop(fr_atomic_queue_t *aq, uint32_t *p_id, void *data
 int fr_control_callback_add(fr_control_t *c, uint32_t id, void *ctx, fr_control_callback_t callback) CC_HINT(nonnull(1,4));
 int fr_control_callback_delete(fr_control_t *c, uint32_t id) CC_HINT(nonnull);
 
+int fr_control_same_thread(fr_control_t *c);
+
 #ifdef __cplusplus
 }
 #endif