/src/freeradius-server/src/lib/io/coord_pair.c

Source
/*
 *   This program is free software; you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation; either version 2 of the License, or
 *   (at your option) any later version.
 *
 *   This program is distributed in the hope that it will be useful,
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   GNU General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with this program; if not, write to the Free Software
 *   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
 */

/**
 * $Id: ec9a7b3c72087b3bf84af340fac59b9f8905dc60 $
 *
 * @brief Sending pair lists to and from coordination threads
 * @file io/coord_pair.c
 *
 * @copyright 2026 Network RADIUS SAS (legal@networkradius.com)
 */
RCSID("$Id: ec9a7b3c72087b3bf84af340fac59b9f8905dc60 $")

#include <freeradius-devel/internal/internal.h>
#include <freeradius-devel/io/listen.h>
#include <freeradius-devel/io/coord_pair.h>
#include <freeradius-devel/io/coord_priv.h>
#include <freeradius-devel/server/main_config.h>
#include <freeradius-devel/unlang/base.h>

static _Atomic(uint64_t) request_number = 0;

FR_SLAB_TYPES(request, request_t)
FR_SLAB_FUNCS(request, request_t)

static fr_dlist_head_t  *coord_pair_regs = NULL;
static module_list_t  *coord_pair_modules;
static fr_dict_attr_t const *attr_worker_id = NULL;

/** Registration of pair list callbacks
 *
 */
struct fr_coord_pair_reg_s {
  char const      *name;      //!< Name for log / request name.
  fr_dlist_t      entry;      //!< Entry in list of pair list registrations
  fr_dict_attr_t const    *attr_packet_type;  //!< Attribute containing packet type
  fr_dict_attr_t const    *root;      //!< Pair list decoding root attribute
  fr_coord_worker_pair_cb_reg_t **callbacks;    //!< Array of pointers to callbacks
  uint32_t      max_packet_type;  //!< Largest valid value for packet type
  uint32_t      cb_id;      //!< The coordinator callback ID used for pair list handling
  fr_time_delta_t     max_request_time; //!< Maximum time for coordinator request processing.
  fr_slab_config_t    reuse;      //!< Request slab allocation config.
  virtual_server_t const    *vs;      //!< Virtual server containing coordinator process sections.
};

struct fr_coord_pair_s {
  fr_coord_t      *coord;     //!< Coordinator which this coord pair is attached to.
  fr_coord_pair_reg_t   *coord_pair_reg;  //!< Registration details for this coord pair
  fr_event_list_t     *el;      //!< Event list for interpreter.
  unlang_interpret_t    *intp;      //!< Interpreter for running requests.
  fr_heap_t     *runnable;    //!< Current runnable requests.

  fr_timer_list_t     *timeout;   //!< Track when requests timeout using a dlist.
  fr_time_delta_t     predicted;    //!< How long we predict a request will take to execute.
  fr_time_tracking_t    tracking;   //!< How much time the coordinator has spent doing things.
  uint64_t      num_active;   //!< Number of active requests.
  request_slab_list_t   *slab;      //!< slab allocator for request_t
};

/** Packet context used when coordinator messages are processed through an interpreter
 *
 * Allows access to the coordinator structure and arbitrary data
 * throughout the state machine.
 */
typedef struct {
  fr_coord_pair_t     *coord_pair;    //!< Coordinator pair this packet is for.
  void        *uctx;      //!< Source specific ctx.
} fr_coord_packet_ctx_t;

/** Conf parser to read slab settings from module config
 */
static const conf_parser_t request_reuse_config[] = {
  FR_SLAB_CONFIG_CONF_PARSER
  CONF_PARSER_TERMINATOR
};

/** Remove a coord pair registration from the list when it is freed
 */
static int _coord_pair_reg_free(fr_coord_pair_reg_t *to_free)
{
  fr_assert(coord_pair_regs);

  fr_dlist_remove(coord_pair_regs, to_free);

  /* If all the registrations are gone, free the list */
  if (fr_dlist_num_elements(coord_pair_regs) == 0) {
    TALLOC_FREE(coord_pair_regs);
    TALLOC_FREE(coord_pair_modules);
  }
  return 0;
}

/** Register a set of callbacks for pair list based coordinator messages
 *
 * Returns a structure to pass as uctx to fr_coord_cb_t using the
 * macro FR_COORD_PAIR_CB_CTX_SET.
 *
 * @param reg_ctx Callback details to register.
 */
fr_coord_pair_reg_t *fr_coord_pair_register(fr_coord_pair_reg_ctx_t *reg_ctx)
{
  fr_coord_pair_reg_t   *coord_pair_reg;
  fr_coord_worker_pair_cb_reg_t *cb_reg = reg_ctx->worker_cb;
  CONF_SECTION      *cs;
  CONF_PAIR     *cp;

  fr_assert(reg_ctx->root);

  /* Resolve the Worker-Id attribute if not already done */
  if (!attr_worker_id) {
    attr_worker_id = fr_dict_attr_by_name(NULL, fr_dict_root(fr_dict_internal()), "Worker-Id");
    if (!attr_worker_id) {
      ERROR("Failed to resolve Worker-Id attribute");
      return NULL;
    }
  }

  if (!coord_pair_regs) {
    MEM(coord_pair_regs = talloc_zero(NULL, fr_dlist_head_t));
    fr_dlist_init(coord_pair_regs, fr_coord_pair_reg_t, entry);
    MEM(coord_pair_modules = module_list_alloc(NULL, &module_list_type_global, "coord", true));
  }

  MEM(coord_pair_reg = talloc(coord_pair_regs, fr_coord_pair_reg_t));
  *coord_pair_reg = (fr_coord_pair_reg_t) {
    .name = reg_ctx->name,
    .root = reg_ctx->root,
    .cb_id = reg_ctx->cb_id,
    .max_request_time = fr_time_delta_eq(reg_ctx->max_request_time, fr_time_delta_from_msec(0)) ?
      main_config->worker.max_request_time : reg_ctx->max_request_time,
  };

  while (cb_reg->callback) {
    if (cb_reg->packet_type > coord_pair_reg->max_packet_type) {
      coord_pair_reg->max_packet_type = cb_reg->packet_type;
    }
    cb_reg++;
  }

  /*
   *  A sane limit on packet type values to avoid a huge array.
   *  If larger values are needed in the future we can use a folded array.
   */
  fr_assert(coord_pair_reg->max_packet_type <= 256);

  MEM(coord_pair_reg->callbacks = talloc_zero_array(coord_pair_reg, fr_coord_worker_pair_cb_reg_t *,
                coord_pair_reg->max_packet_type + 1));

  cb_reg = reg_ctx->worker_cb;
  while (cb_reg->callback) {
    coord_pair_reg->callbacks[cb_reg->packet_type] = cb_reg;
    cb_reg++;
  }

  cs = cf_section_find(reg_ctx->cs, "reuse", NULL);

  /*
   *  Create an empty "reuse" section if one is not found, so defaults are applied
   */
  if (!cs) {
    cs = cf_section_alloc(reg_ctx->cs, reg_ctx->cs, "reuse", NULL);
  }

  if (cf_section_rules_push(cs, request_reuse_config) < 0) {
  fail:
    talloc_free(coord_pair_reg);
    return NULL;
  }
  if (cf_section_parse(coord_pair_reg, &coord_pair_reg->reuse, cs) < 0) goto fail;

  /*
   *  Set defaults for request slab allocation, if not set by conf parsing
   */
  if (!coord_pair_reg->reuse.child_pool_size) coord_pair_reg->reuse.child_pool_size = REQUEST_POOL_SIZE;
  if (!coord_pair_reg->reuse.num_children) coord_pair_reg->reuse.num_children = REQUEST_POOL_NUM_OBJECTS;

  cp = cf_pair_find(reg_ctx->cs, "virtual_server");
  if (!cp) {
    cf_log_err(reg_ctx->cs, "Missing virtual_server option");
    goto fail;
  }

  coord_pair_reg->vs = virtual_server_find(cf_pair_value(cp));
  if (!coord_pair_reg->vs) {
    cf_log_err(cp, "Virtual server not found");
    goto fail;
  }

  /*
   *  Validate that the virtual server uses the correct namespace.
   */
  if (reg_ctx->root->dict != virtual_server_dict_by_cs(virtual_server_cs(coord_pair_reg->vs))) {
    cf_log_err(cp, "Virtual server has namespace %s, should be %s",
         fr_dict_root(virtual_server_dict_by_cs(virtual_server_cs(coord_pair_reg->vs)))->name,
         fr_dict_root(coord_pair_reg->root->dict)->name);
    goto fail;
  }
  coord_pair_reg->attr_packet_type = virtual_server_packet_type_by_cs(virtual_server_cs(coord_pair_reg->vs));

  fr_dlist_insert_tail(coord_pair_regs, coord_pair_reg);
  talloc_set_destructor(coord_pair_reg, _coord_pair_reg_free);

  return coord_pair_reg;
}

/** Return the coordinator callback ID associated with a coord_pair_reg_t
 */
uint32_t fr_coord_pair_reg_cb_id(fr_coord_pair_reg_t *coord_pair_reg)
{
  fr_assert(coord_pair_reg);
  return coord_pair_reg->cb_id;
}

/*
 *  The following set of callbacks for request handling are mirrors of
 *  their equivalent in worker.c
 */

/** Signal the unlang interpreter that it needs to stop running the request
 *
 * @param[in] request request to cancel.  The request may still run to completion.
 */
static void coord_pair_stop_request(request_t *request)
{
  unlang_interpret_signal(request, FR_SIGNAL_CANCEL);
}

/** Enforce max_request_time
 *
 * @param[in] tl  the coordinators's timer list.
 * @param[in] when  the current time
 * @param[in] uctx  the request_t timing out.
 */
static void _coord_pair_request_timeout(UNUSED fr_timer_list_t *tl, UNUSED fr_time_t when, void *uctx)
{
  request_t *request = talloc_get_type_abort(uctx, request_t);

  REDEBUG("Request has reached max_request_time - signalling it to stop");
  coord_pair_stop_request(request);

  request->rcode = RLM_MODULE_TIMEOUT;
}

/** Set, or re-set the request timer
 *
 * @param[in] coord_pair  the coord_pair_t containing the timeout lists.
 * @param[in] request   that we're timing out.
 * @param[in] timeout   the timeout to set.
 * @return
 *  - 0 on success.
 *  - -1 on failure.
 */
static int fr_coord_pair_request_timeout_set(fr_coord_pair_t *coord_pair, request_t *request, fr_time_delta_t timeout)
{
  if (unlikely(fr_timer_in(request, coord_pair->timeout, &request->timeout, timeout,
         true, _coord_pair_request_timeout, request) < 0)) {
    RERROR("Failed to create request timeout timer");
    return -1;
  }

  return 0;
}

/** Start time tracking for a request, and mark it as runnable.
 */
static int coord_pair_request_time_tracking_start(fr_coord_pair_t *coord_pair, request_t *request, fr_time_t now)
{
  fr_assert(!fr_timer_armed(request->timeout));

  if (unlikely(fr_coord_pair_request_timeout_set(coord_pair, request,
                   coord_pair->coord_pair_reg->max_request_time) < 0)) {
    RERROR("Failed to set request timeout");
    return -1;
  }

  RDEBUG3("Time tracking started in yielded state");
  fr_time_tracking_start(&coord_pair->tracking, &request->async->tracking, now);
  fr_time_tracking_yield(&request->async->tracking, now);
  coord_pair->num_active++;

  fr_assert(!fr_heap_entry_inserted(request->runnable));
  (void) fr_heap_insert(&coord_pair->runnable, request);

  return 0;
}

/** End time tracking for a request
 */
static void coord_pair_request_time_tracking_end(fr_coord_pair_t *coord_pair, request_t *request, fr_time_t now)
{
  RDEBUG3("Time tracking ended");
  fr_time_tracking_end(&coord_pair->predicted, &request->async->tracking, now);
  fr_assert(coord_pair->num_active > 0);
  coord_pair->num_active--;

  TALLOC_FREE(request->timeout);  /* Disarm the request timer */
}


static inline CC_HINT(always_inline)
void coord_pair_request_init(fr_event_list_t *el, request_t *request, fr_time_t now, void *packet_ctx)
{
  if (!request->packet) MEM(request->packet = fr_packet_alloc(request, false));
  if (!request->reply) MEM(request->reply = fr_packet_alloc(request, false));

  request->packet->timestamp = now;
  request->async = talloc_zero(request, fr_async_t);
  request->async->recv_time = now;
  request->async->el = el;
  request->async->packet_ctx = packet_ctx;
  fr_dlist_entry_init(&request->async->entry);
}

static inline CC_HINT(always_inline)
void coord_pair_request_name_number(request_t *request, char const *name)
{
  request->number = atomic_fetch_add_explicit(&request_number, 1, memory_order_seq_cst);
  if (request->name) talloc_const_free(request->name);
  request->name = talloc_asprintf(request, "Coord-%s-%"PRIu64, name, request->number);
}

static int _coord_pair_request_deinit( request_t *request, UNUSED void *uctx)
{
  return request_slab_deinit(request);
}

static request_t *coord_pair_request_bootstrap(fr_coord_pair_t *coord_pair, fr_time_t now, void *uctx)
{
  request_t   *request;
  fr_coord_packet_ctx_t *packet_ctx;

  request = request_slab_reserve(coord_pair->slab);
  if (!request) {
    ERROR("Coordinator failed allocating new request");
    return NULL;
  }

  request_slab_element_set_destructor(request, _coord_pair_request_deinit, coord_pair);

  if (request_init(request, REQUEST_TYPE_INTERNAL,
       (&(request_init_args_t){
        .namespace = virtual_server_dict_by_cs(virtual_server_cs(coord_pair->coord_pair_reg->vs))
       }))) {
    ERROR("Coordinator failed initializing new request");
    request_slab_release(request);
    return NULL;
  }

  MEM(packet_ctx = talloc(request, fr_coord_packet_ctx_t));
  *packet_ctx = (fr_coord_packet_ctx_t) {
    .coord_pair = coord_pair,
    .uctx = uctx
  };
  coord_pair_request_init(coord_pair->el, request, now, packet_ctx);
  coord_pair_request_name_number(request, coord_pair->coord_pair_reg->name);

  unlang_interpret_set(request, coord_pair->intp);

  return request;
}

static void coord_pair_request_start(fr_coord_pair_t *coord_pair, request_t *request, fr_time_t now)
{
  fr_pair_t *vp;

  vp = fr_pair_find_by_da(&request->request_pairs, NULL, coord_pair->coord_pair_reg->attr_packet_type);
  if (!vp) {
    RERROR("Missing %s attribute", coord_pair->coord_pair_reg->attr_packet_type->name);
  error:
    request_slab_release(request);
    return;
  }

  request->packet->code = vp->vp_uint32;

  if (virtual_server_push(NULL, request, coord_pair->coord_pair_reg->vs, UNLANG_TOP_FRAME) < 0) {
    RERROR("Protocol failed to set 'process' function");
    goto error;
  }

  if (unlikely(coord_pair_request_time_tracking_start(coord_pair, request, now) < 0)) {
    RERROR("Failed to start request time tracking");
    goto error;
  }
}

static void _coord_pair_request_internal_init(request_t *request, void *uctx)
{
  fr_coord_pair_t *coord_pair = talloc_get_type_abort(uctx, fr_coord_pair_t);
  fr_time_t now = fr_time();

  fr_assert(request->packet);
  fr_assert(request->reply);

  request->packet->timestamp = now;
  request->async = talloc_zero(request, fr_async_t);
  request->async->recv_time = now;
  request->async->el = coord_pair->el;
  fr_dlist_entry_init(&request->async->entry);

  /*
   *  Requests generated by the interpreter
   *  are always marked up as internal.
   */
  fr_assert(request_is_internal(request));
  coord_pair_request_time_tracking_start(coord_pair, request, now);
}

/** External request is now complete - will never happen with coordinators
 *
 */
static void _coord_pair_request_done_external(UNUSED request_t *request, UNUSED rlm_rcode_t rcode, UNUSED void *uctx)
{
  fr_assert(0);
}

/** Internal request (i.e. one generated by the interpreter) is now complete
 *
 * Whatever generated the request is now responsible for freeing it.
 */
static void _coord_pair_request_done_internal(request_t *request, UNUSED rlm_rcode_t rcode, void *uctx)
{
  fr_coord_pair_t *coord_pair = talloc_get_type_abort(uctx, fr_coord_pair_t);

  coord_pair_request_time_tracking_end(coord_pair, request, fr_time());

  fr_assert(!fr_heap_entry_inserted(request->runnable));
  fr_assert(!fr_timer_armed(request->timeout));
  fr_assert(!fr_dlist_entry_in_list(&request->async->entry));
}

/** Detached request (i.e. one generated by the interpreter with no parent) is now complete
 *
 * As the request has no parent, then there's nothing to free it
 * so we have to.
 */
static void _coord_pair_request_done_detached(request_t *request, UNUSED rlm_rcode_t rcode, UNUSED void *uctx)
{
  fr_assert(!fr_heap_entry_inserted(request->runnable));

  TALLOC_FREE(request->timeout);

  fr_assert(!fr_dlist_entry_in_list(&request->async->entry));

  talloc_free(request);
}

/** Make us responsible for running the request
 *
 */
static void _coord_pair_request_detach(request_t *request, void *uctx)
{
  fr_coord_pair_t *coord_pair = talloc_get_type_abort(uctx, fr_coord_pair_t);

  RDEBUG4("%s - Request detaching", __FUNCTION__);

  if (request_is_detachable(request)) {
    /*
    * End the time tracking...  We don't track detached requests,
    * because they don't contribute for the time consumed by an
    * external request.
    */
    if (request->async->tracking.state == FR_TIME_TRACKING_YIELDED) {
      RDEBUG3("Forcing time tracking to running state, from yielded, for request detach");
      fr_time_tracking_resume(&request->async->tracking, fr_time());
    }
    coord_pair_request_time_tracking_end(coord_pair, request, fr_time());

    if (request_detach(request) < 0) RPEDEBUG("Failed detaching request");

    RDEBUG3("Request is detached");
  } else {
    fr_assert_msg(0, "Request is not detachable");
  }
}

/** Request is now runnable
 *
 */
static void _coord_pair_request_runnable(request_t *request, void *uctx)
{
  fr_coord_pair_t *coord_pair = uctx;

  RDEBUG4("%s - Request marked as runnable", __FUNCTION__);
  fr_heap_insert(&coord_pair->runnable, request);
}

/** Interpreter yielded request
 *
 */
static void _coord_pair_request_yield(request_t *request, UNUSED void *uctx)
{
  RDEBUG4("%s - Request yielded", __FUNCTION__);
  if (likely(!request_is_detached(request))) fr_time_tracking_yield(&request->async->tracking, fr_time());
}

/** Interpreter is starting to work on request again
 *
 */
static void _coord_pair_request_resume(request_t *request, UNUSED void *uctx)
{
  RDEBUG4("%s - Request resuming", __FUNCTION__);
  if (likely(!request_is_detached(request))) fr_time_tracking_resume(&request->async->tracking, fr_time());
}

/** Check if a request is scheduled
 *
 */
static bool _coord_pair_request_scheduled(request_t const *request, UNUSED void *uctx)
{
  return fr_heap_entry_inserted(request->runnable);
}

/** Update a request's priority
 *
 */
static void _coord_pair_request_prioritise(request_t *request, void *uctx)
{
  fr_coord_pair_t *coord_pair = talloc_get_type_abort(uctx, fr_coord_pair_t);

  RDEBUG4("%s - Request priority changed", __FUNCTION__);

  /* Extract the request from the runnable queue _if_ it's in the runnable queue */
  if (fr_heap_extract(&coord_pair->runnable, request) < 0) return;

  /* Reinsert it to re-evaluate its new priority */
  fr_heap_insert(&coord_pair->runnable, request);
}

/** Compare two requests by priority and sequence
 */
static int8_t coord_pair_runnable_cmp(void const *one, void const *two)
{
  request_t const *a = one, *b = two;
  int ret;

  ret = CMP(b->priority, a->priority);
  if (ret != 0) return ret;

  return CMP(a->sequence, b->sequence);
}

void fr_coord_pair_inst_destroy(UNUSED fr_coord_t *coord, fr_coord_cb_inst_t *inst, bool single_thread, UNUSED void *uctx) {
  fr_coord_pair_t *coord_pair = talloc_get_type_abort(inst->inst_data, fr_coord_pair_t);
  int   ret, count = 0;

  if (!single_thread) unlang_interpret_set_thread_default(NULL);

  ret = fr_timer_list_force_run(coord_pair->timeout);
  if (unlikely(ret < 0)) {
    fr_assert_msg(0, "Failed to force run the timeout list");
  } else {
    count += ret;
  }

  DEBUG("Coordinator is exiting - stopped %u requests", count);
}

/** Create the coord_pair coord instance data
 */
static fr_coord_pair_t *fr_coord_pair_create(TALLOC_CTX *ctx, fr_coord_t *coord, fr_event_list_t *el,
               bool single_thread, void *uctx)
{
  fr_coord_pair_t   *coord_pair;
  fr_coord_pair_reg_t *coord_pair_reg = talloc_get_type_abort(uctx, fr_coord_pair_reg_t);

  MEM(coord_pair = talloc(ctx, fr_coord_pair_t));
  *coord_pair = (fr_coord_pair_t) {
    .coord = coord,
    .coord_pair_reg = coord_pair_reg,
    .el = el
  };

  coord_pair->runnable = fr_heap_talloc_alloc(coord_pair, coord_pair_runnable_cmp, request_t, runnable, 0);
  if (!coord_pair->runnable) {
    fr_strerror_const("Failed creating runnable heap");
  fail:
    talloc_free(coord_pair);
    return NULL;
  }

  coord_pair->timeout = fr_timer_list_ordered_alloc(coord_pair, el->tl);
  if (!coord_pair->timeout) {
    fr_strerror_const("Failed creating timeouts list");
    goto fail;
  }

  coord_pair->intp = unlang_interpret_init(coord_pair, el,
          &(unlang_request_func_t){
            .init_internal = _coord_pair_request_internal_init,

            .done_external = _coord_pair_request_done_external,
            .done_internal = _coord_pair_request_done_internal,
            .done_detached = _coord_pair_request_done_detached,

            .detach = _coord_pair_request_detach,
            .yield = _coord_pair_request_yield,
            .resume = _coord_pair_request_resume,
            .mark_runnable = _coord_pair_request_runnable,

            .scheduled = _coord_pair_request_scheduled,
            .prioritise = _coord_pair_request_prioritise
          }, coord_pair);

  if (!coord_pair->intp) goto fail;

  if (!(coord_pair->slab = request_slab_list_alloc(coord_pair, el, &coord_pair_reg->reuse, NULL, NULL,
               coord_pair, true, false))) {
    goto fail;
  }

  if (!single_thread) unlang_interpret_set_thread_default(coord_pair->intp);

  return coord_pair;
}

static inline CC_HINT(always_inline) void coord_run_request(fr_coord_pair_t *coord_pair, fr_time_t start)
{
  request_t *request;
  fr_time_t now;

  now = start;

  while (fr_time_delta_lt(fr_time_sub(now, start), fr_time_delta_from_msec(1)) &&
  ((request = fr_heap_pop(&coord_pair->runnable)) != NULL)) {
    REQUEST_VERIFY(request);
    fr_assert(!fr_heap_entry_inserted(request->runnable));

    (void)unlang_interpret(request, UNLANG_REQUEST_RESUME);

    now = fr_time();
  }
}

/*
 *  Pre and post events used in single threaded mode
 */

static int fr_coord_pair_pre_event(UNUSED fr_time_t now, UNUSED fr_time_delta_t wake, void *uctx)
{
  fr_coord_pair_t *coord_pair = talloc_get_type_abort(uctx, fr_coord_pair_t);
  request_t *request;

  request = fr_heap_peek(coord_pair->runnable);
  return request ? 1 : 0;
}

static void fr_coord_pair_post_event(UNUSED fr_event_list_t *el, UNUSED fr_time_t now, void *uctx)
{
  fr_coord_pair_t *coord_pair = talloc_get_type_abort(uctx, fr_coord_pair_t);

  coord_run_request(coord_pair, fr_time());
}

/** Event callback in multi threaded mode
 */
static void fr_coord_pair_event(UNUSED fr_event_list_t *el, void *uctx)
{
  fr_coord_pair_t *coord_pair = talloc_get_type_abort(uctx, fr_coord_pair_t);

  coord_run_request(coord_pair, fr_time());
}

/** Callback run when a coordinator receives pair list data
 *
 * Converts the data into a request.
 */
void fr_coord_pair_data_recv(UNUSED fr_coord_t *coord, uint32_t worker_id, fr_dbuff_t *dbuff, fr_time_t now, void *inst, void *uctx)
{
  fr_coord_pair_reg_t *coord_pair_reg = talloc_get_type_abort(uctx, fr_coord_pair_reg_t);
  fr_coord_pair_t   *coord_pair = talloc_get_type_abort(inst, fr_coord_pair_t);
  request_t   *request;
  fr_pair_t   *vp;

  request = coord_pair_request_bootstrap(coord_pair, now, coord_pair_reg);
  if (!request) return;

  if (fr_pair_append_by_da(request->request_ctx, &vp, &request->request_pairs, attr_worker_id) < 0) {
  error:
    request_slab_release(request);
    return;
  };
  vp->vp_int32 = worker_id;

  if (fr_internal_decode_list_dbuff(request->pair_list.request, &request->request_pairs,
            fr_dict_root(request->proto_dict), dbuff, NULL) < 0) {
    RERROR("Failed decoding packet");
    goto error;
  }

  coord_pair_request_start(coord_pair, request, now);
}

/** Callback run when a worker receives pair list data
 *
 * Finds the packet type attribute in the data and calls the callback
 * registered against the value of that attribute.
 *
 * @param cw  Worker which received the message.
 * @param dbuff Data received.
 * @param now Time the data is received.
 * @param mctx  Module context to pass to callback.
 * @param uctx  The coord_pair registration.
 */
void fr_coord_worker_pair_data_recv(fr_coord_worker_t *cw, fr_dbuff_t *dbuff, fr_time_t now, module_ctx_t *mctx, void *uctx)
{
  fr_coord_pair_reg_t   *coord_pair_reg = talloc_get_type_abort(uctx, fr_coord_pair_reg_t);
  fr_pair_list_t      list;
  fr_pair_t     *vp;

  fr_pair_list_init(&list);
  if (fr_internal_decode_list_dbuff(NULL, &list, coord_pair_reg->root, dbuff, NULL) < 0) {
    PERROR("Failed to decode data as pair list");
    goto free;
  }

  vp = fr_pair_find_by_da_nested(&list, NULL, coord_pair_reg->attr_packet_type);

  if (!vp) {
    ERROR("Message received without %s", coord_pair_reg->attr_packet_type->name);
    goto free;
  }

  if (vp->vp_uint32 > coord_pair_reg->max_packet_type || !coord_pair_reg->callbacks[vp->vp_uint32]) {
    ERROR("Message received with invalid value %pP", vp);
    goto free;
  }

  coord_pair_reg->callbacks[vp->vp_uint32]->callback(cw, coord_pair_reg, &list, now, mctx,
                 coord_pair_reg->callbacks[vp->vp_uint32]->uctx);

free:
  fr_pair_list_free(&list);
}

/** Send a reply list from a coordinator to a worker
 *
 * @param request containing the reply to send.
 * @param worker_id to send the reply to.
 * @return
 *  - 0 on success
 *  - -1 on failure
 */
int fr_coord_to_worker_reply_send(request_t *request, uint32_t worker_id)
{
  fr_dbuff_t    dbuff;
  fr_dbuff_uctx_talloc_t  tctx;
  fr_coord_packet_ctx_t *packet_ctx = talloc_get_type_abort(request->async->packet_ctx, fr_coord_packet_ctx_t);
  fr_coord_pair_reg_t *coord_pair_reg = talloc_get_type_abort(packet_ctx->uctx, fr_coord_pair_reg_t);
  int     ret;

  if (fr_dbuff_init_talloc(NULL, &dbuff, &tctx, 1024, SIZE_MAX) == NULL) return -1;
  if (fr_internal_encode_list(&dbuff, &request->reply_pairs, NULL) < 0) {
    fr_dbuff_free_talloc(&dbuff);
    return -1;
  }

  ret = fr_coord_to_worker_send(packet_ctx->coord_pair->coord, worker_id, coord_pair_reg->cb_id, &dbuff);

  fr_dbuff_free_talloc(&dbuff);

  return ret;
}

/** Send a reply list from a coordinator to all workers
 *
 * @param request containing the reply to send.
 * @return
 *  - 0 on success
 *  - < 0 on failure
 */
int fr_coord_to_worker_reply_broadcast(request_t *request)
{
  fr_dbuff_t    dbuff;
  fr_dbuff_uctx_talloc_t  tctx;
  fr_coord_packet_ctx_t *packet_ctx = talloc_get_type_abort(request->async->packet_ctx, fr_coord_packet_ctx_t);
  fr_coord_pair_reg_t *coord_pair_reg = talloc_get_type_abort(packet_ctx->uctx, fr_coord_pair_reg_t);
  int     ret;

  if (fr_dbuff_init_talloc(NULL, &dbuff, &tctx, 1024, SIZE_MAX) == NULL) return -1;
  if (fr_internal_encode_list(&dbuff, &request->reply_pairs, NULL) < 0) {
    fr_dbuff_free_talloc(&dbuff);
    return -1;
  }

  ret = fr_coord_to_worker_broadcast(packet_ctx->coord_pair->coord, coord_pair_reg->cb_id, &dbuff);

  fr_dbuff_free_talloc(&dbuff);

  return ret;
}

/** Send a pair list from a worker to a coordinator
 *
 * The pair list must include an attribute indicating the packet type
 *
 * @param cw  The coord worker sending the data.
 * @param coord_pair_reg  The coord_pair registration to use.
 * @param list  of pairs to send.
 * @return
 *  - 0 on success
 *  - -1 on failure
 */
int fr_worker_to_coord_pair_send(fr_coord_worker_t *cw, fr_coord_pair_reg_t *coord_pair_reg, fr_pair_list_t *list)
{
  fr_dbuff_t    dbuff;
  fr_dbuff_uctx_talloc_t  tctx;
  int     ret;

  if (fr_dbuff_init_talloc(NULL, &dbuff, &tctx, 1024, SIZE_MAX) == NULL) return -1;
  if (fr_internal_encode_list(&dbuff, list, NULL) < 0) {
    fr_dbuff_free_talloc(&dbuff);
    return -1;
  }

  ret = fr_worker_to_coord_send(cw, coord_pair_reg->cb_id, &dbuff);

  fr_dbuff_free_talloc(&dbuff);
  return ret;
}

/** Instance creation called during coordinator creation.
 *
 * @param ctx   to allocate the instance in.
 * @param coord   Coordinator to create an instance of.
 * @param el    Event list for instance to use.
 * @param single_thread is the server in single thread mode.
 * @param uctx    configured for the callback this instance relates to.
 * @return
 *  - fr_coord_cb_inst_t on success
 *  - NULL on failure
 */
fr_coord_cb_inst_t *fr_coord_pair_inst_create(TALLOC_CTX *ctx, fr_coord_t *coord, fr_event_list_t *el,
                 bool single_thread, void *uctx)
{
  fr_coord_cb_inst_t  *cb_inst;
  fr_coord_pair_t   *coord_pair;

  MEM(cb_inst = talloc(ctx, fr_coord_cb_inst_t));

  *cb_inst = (fr_coord_cb_inst_t) {
    .event_pre_cb = fr_coord_pair_pre_event,
    .event_post_cb = fr_coord_pair_post_event,
    .event_cb = fr_coord_pair_event
  };

  coord_pair = fr_coord_pair_create(ctx, coord, el, single_thread, uctx);
  if (!coord_pair) {
    talloc_free(cb_inst);
    return NULL;
  }

  cb_inst->inst_data = coord_pair;

  return cb_inst;
}

/** Return the coord_pair associated with a coord_pair internal request
 *
 * @param request to fetch associated coordinator for.
 * @return fr_coord_t
 */
fr_coord_pair_t *fr_coord_pair_request_coord_pair(request_t *request)
{
  fr_coord_packet_ctx_t *packet_ctx = talloc_get_type_abort(request->async->packet_ctx, fr_coord_packet_ctx_t);
  return packet_ctx->coord_pair;
}

/** Start a coordinator request to run through a coord_pair process module
 *
 * @param coord_pair  with the process module to run the request.
 * @param list    Pairs to populate the request.
 * @param now   Request time.
 * @return
 *  0 on success
 *  -1 on failure
 */
int fr_coord_pair_coord_request_start(fr_coord_pair_t *coord_pair, fr_pair_list_t *list, fr_time_t now)
{
  request_t *request;

  request = coord_pair_request_bootstrap(coord_pair, now, coord_pair->coord_pair_reg);
  if (!request) return -1;

  fr_pair_list_steal(request->request_ctx, list);
  fr_pair_list_append(&request->request_pairs, list);

  coord_pair_request_start(coord_pair, request, now);
  return 0;
}

Coverage Report

Created: 2026-06-30 07:16