udp_connection_t *c; //!< long-term connection
udp_thread_t *thread;
+ uint8_t last_id; //!< Used when replicating to ensure IDs are distributed
+ ///< evenly.
+
uint32_t max_packet_size; //!< Our max packet size. may be different from the parent.
fr_ipaddr_t src_ipaddr; //!< Source IP address. May be altered on bind
MEM(h->buffer = talloc_array(h, uint8_t, h->max_packet_size));
h->buflen = h->max_packet_size;
- MEM(h->tt = radius_track_alloc(h));
+ if (!h->inst->replicate) MEM(h->tt = radius_track_alloc(h));
/*
* Open the outgoing socket.
return c->conn;
}
+/** Read and discard data
+ *
+ */
+static void conn_discard(UNUSED fr_event_list_t *el, int fd, UNUSED int flags, void *uctx)
+{
+ fr_trunk_connection_t *tconn = talloc_get_type_abort(uctx, fr_trunk_connection_t);
+ uint8_t buffer[4096];
+ ssize_t slen;
+
+ while ((slen = read(fd, buffer, sizeof(buffer))) > 0);
+
+ if (slen < 0) {
+ switch (errno) {
+ case EBADF:
+ case ECONNRESET:
+ case ENOTCONN:
+ case ETIMEDOUT:
+ fr_trunk_connection_signal_reconnect(tconn, FR_CONNECTION_FAILED);
+ break;
+
+ default:
+ break;
+ }
+ }
+}
+
/** Standard I/O read function
*
* Underlying FD in now readable, so call the trunk to read any pending requests
}
}
+/** A special version of the trunk/event loop glue function which always discards incoming data
+ *
+ */
+static void thread_conn_notify_replicate(fr_trunk_connection_t *tconn, fr_connection_t *conn,
+ fr_event_list_t *el,
+ fr_trunk_connection_event_t notify_on, UNUSED void *uctx)
+{
+ udp_handle_t *h = talloc_get_type_abort(conn->h, udp_handle_t);
+ fr_event_fd_cb_t read_fn = NULL;
+ fr_event_fd_cb_t write_fn = NULL;
+
+ switch (notify_on) {
+ case FR_TRUNK_CONN_EVENT_NONE:
+ read_fn = conn_discard;
+ write_fn = conn_writable;
+ return;
+
+ case FR_TRUNK_CONN_EVENT_READ:
+ case FR_TRUNK_CONN_EVENT_BOTH:
+ rad_assert(0); /* Should never happen */
+ break;
+
+ case FR_TRUNK_CONN_EVENT_WRITE:
+ read_fn = conn_discard;
+ write_fn = conn_writable;
+ break;
+
+ }
+
+ if (fr_event_fd_insert(h, el, h->fd,
+ read_fn,
+ write_fn,
+ conn_error,
+ tconn) < 0) {
+ ERROR("%s - Failed inserting FD event", h->module_name);
+
+ /*
+ * May free the connection!
+ */
+ fr_trunk_connection_signal_reconnect(tconn, FR_CONNECTION_FAILED);
+ }
+}
+
/*
* Return negative numbers to put 'a' at the top of the heap.
* Return positive numbers to put 'b' at the top of the heap.
* replies which then stopped coming back (and then mark
* it zombie).
*/
- if (h->inst->replicate || h->zombie_ev || !h->last_sent || (h->last_sent <= h->last_idle) ||
+ if (h->zombie_ev || !h->last_sent || (h->last_sent <= h->last_idle) ||
(h->last_reply && (h->last_reply <= h->last_idle))) {
return;
}
return 0;
}
-
static void request_mux(fr_event_list_t *el,
fr_trunk_connection_t *tconn, fr_connection_t *conn, UNUSED void *uctx)
{
udp_handle_t *h = talloc_get_type_abort(conn->h, udp_handle_t);
udp_connection_t *c = h->c;
- rlm_radius_udp_t const *inst = h->thread->inst;
fr_trunk_request_t *treq;
- REQUEST *request;
- udp_request_t *u;
check_for_zombie(el, tconn, 0);
while ((treq = fr_trunk_connection_pop_request(tconn)) != NULL) {
- udp_result_t *r = talloc_get_type_abort(treq->rctx, udp_result_t);
-
- u = treq->preq;
-
- request = treq->request;
+ udp_request_t *u = talloc_get_type_abort(treq->preq, udp_request_t);
+ REQUEST *request = treq->request;
/*
* No existing packet, create it...
if (!u->packet) {
rad_assert(!u->rr);
- RDEBUG("Sending %s ID %d length %ld over connection %s",
- fr_packet_codes[u->code], u->rr->id, u->packet_len, h->name);
-
u->rr = radius_track_entry_alloc(h->tt, request, u->code, treq);
if (!u->rr) {
fail:
continue;
}
+ RDEBUG("Sending %s ID %d length %ld over connection %s",
+ fr_packet_codes[u->code], u->rr->id, u->packet_len, h->name);
+
if (encode(h->inst, request, u, u->rr->id) < 0) {
fail_after_alloc:
udp_request_clear(u, h, 0);
if (write_packet(el, treq, u->packet, u->packet_len) < 0) goto fail_after_alloc;
fr_trunk_request_signal_sent(treq);
- /*
- * We're replicating, so we don't care about the
- * responses. Don't do any retransmission timers, don't
- * look for replies to status checks, etc.
- *
- * Instead, just set the return code to OK, and return.
- *
- * @todo - if replicating, change the mux()
- * routine to allocate a random ID, instead of
- * calling a complex API.
- */
- if (inst->replicate) {
- (void) radius_track_delete(h->tt, u->rr); /* don't set last_idle, we're not checking for zombie */
- u->rr = NULL;
- r->rcode = RLM_MODULE_OK;
- fr_trunk_request_signal_complete(treq);
- continue;
- }
-
/*
* Tell the trunk API that this request is now in
* the "sent" state. And we don't want to see
}
}
+static void request_mux_replicate(fr_event_list_t *el,
+ fr_trunk_connection_t *tconn, fr_connection_t *conn, UNUSED void *uctx)
+{
+ udp_handle_t *h = talloc_get_type_abort(conn->h, udp_handle_t);
+ fr_trunk_request_t *treq;
+
+ while ((treq = fr_trunk_connection_pop_request(tconn)) != NULL) {
+ udp_result_t *r = talloc_get_type_abort(treq->rctx, udp_result_t);
+ udp_request_t *u = talloc_get_type_abort(treq->preq, udp_request_t);
+ REQUEST *request = treq->request;
+ uint8_t id = h->last_id;
+
+ RDEBUG("Sending %s ID %d length %ld over connection %s",
+ fr_packet_codes[u->code], id, u->packet_len, h->name);
+
+ if (!u->packet) {
+ if (encode(h->inst, request, u, id) < 0) {
+ udp_request_clear(u, h, 0);
+ fr_trunk_request_signal_fail(treq);
+ return;
+ }
+ RHEXDUMP3(u->packet, u->packet_len, "Encoded packet");
+ }
+
+ /*
+ * Assume we'll be able to write later, and just break
+ * out of the loop.
+ */
+ if (write_packet(el, treq, u->packet, u->packet_len) < 0) return;
+
+ r->rcode = RLM_MODULE_OK;
+
+ /*
+ * Signal the request as immediately complete
+ */
+ fr_trunk_request_signal_complete(treq);
+
+ h->last_id++; /* Wraps to zero - Defined behaviour*/
+ }
+}
+
/** If we get a reply, the request must come from one of a small
* number of packet types.
*/
}
}
-
-static fr_trunk_io_funcs_t trunk_funcs = {
- .connection_alloc = thread_conn_alloc,
- .connection_notify = thread_conn_notify,
- .request_prioritise = request_prioritise,
- .request_mux = request_mux,
- .request_demux = request_demux,
- .request_complete = request_complete,
- .request_cancel = request_cancel,
-};
-
-
static rlm_rcode_t request_resume(UNUSED void *instance, UNUSED void *thread, UNUSED REQUEST *request, void *ctx)
{
udp_result_t *r = talloc_get_type_abort(ctx, udp_result_t);
*/
static int mod_thread_instantiate(UNUSED CONF_SECTION const *cs, void *instance, fr_event_list_t *el, void *tctx)
{
- rlm_radius_udp_t *inst = talloc_get_type_abort(instance, rlm_radius_udp_t);
- udp_thread_t *thread = talloc_get_type_abort(tctx, udp_thread_t);
+ rlm_radius_udp_t *inst = talloc_get_type_abort(instance, rlm_radius_udp_t);
+ udp_thread_t *thread = talloc_get_type_abort(tctx, udp_thread_t);
+
+ static fr_trunk_io_funcs_t io_funcs = {
+ .connection_alloc = thread_conn_alloc,
+ .connection_notify = thread_conn_notify,
+ .request_prioritise = request_prioritise,
+ .request_mux = request_mux,
+ .request_demux = request_demux,
+ .request_complete = request_complete,
+ .request_cancel = request_cancel,
+ };
+
+ static fr_trunk_io_funcs_t io_funcs_replicate = {
+ .connection_alloc = thread_conn_alloc,
+ .connection_notify = thread_conn_notify_replicate,
+ .request_prioritise = request_prioritise,
+ .request_mux = request_mux_replicate,
+ .request_demux = NULL,
+ .request_complete = request_complete,
+ .request_cancel = NULL,
+ };
inst->trunk_conf = &inst->parent->trunk_conf;
thread->el = el;
thread->inst = inst;
- thread->trunk = fr_trunk_alloc(thread, el, &trunk_funcs, inst->trunk_conf,
- inst->parent->name, thread, false);
+ thread->trunk = fr_trunk_alloc(thread, el, inst->replicate ? &io_funcs_replicate : &io_funcs,
+ inst->trunk_conf, inst->parent->name, thread, false);
if (!thread->trunk) return -1;
return 0;
projects / thirdparty / freeradius-server.git / commitdiff
