From: Alan T. DeKok <aland@freeradius.org>
Date: Wed, 26 Jul 2017 18:53:52 +0000 (-0400)
Subject: more of the new / better method of managing connections
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=80f281eada8e44b3eb3ee1ca5f594b930277438d;p=thirdparty%2Ffreeradius-server.git

more of the new / better method of managing connections
---

diff --git a/src/modules/rlm_radius/rlm_radius.c b/src/modules/rlm_radius/rlm_radius.c
index f6ea3e0bfcf..318dfdaed7a 100644
--- a/src/modules/rlm_radius/rlm_radius.c
+++ b/src/modules/rlm_radius/rlm_radius.c
@@ -391,7 +391,7 @@ static int mod_instantiate(void *instance, UNUSED CONF_SECTION *conf)
 {
 	rlm_radius_t *inst = talloc_get_type_abort(instance, rlm_radius_t);
 
-	if (inst->io->instantiate(inst->io_instance, inst->io_conf) < 0) {
+	if (inst->io->instantiate(inst, inst->io_instance, inst->io_conf) < 0) {
 		cf_log_err(inst->io_conf, "Instantiate failed for \"%s\"",
 			   inst->io->name);
 		return -1;
diff --git a/src/modules/rlm_radius/rlm_radius.h b/src/modules/rlm_radius/rlm_radius.h
index 5e86b9908da..1e2fbaa121c 100644
--- a/src/modules/rlm_radius/rlm_radius.h
+++ b/src/modules/rlm_radius/rlm_radius.h
@@ -36,6 +36,7 @@ typedef struct rlm_radius_link_t rlm_radius_link_t;
  *
  */
 typedef int (*fr_radius_io_push_t)(void *instance, REQUEST *request, rlm_radius_link_t *link, void *thread);
+typedef int (*fr_radius_io_instantiate_t)(rlm_radius_t *inst, void *io_instance, CONF_SECTION *cs);
 
 
 /** Public structure describing an I/O path for an outgoing socket.
@@ -46,7 +47,7 @@ typedef struct fr_radius_client_io_t {
 	RAD_MODULE_COMMON;				//!< Common fields to all loadable modules.
 
 	fr_app_bootstrap_t		bootstrap;
-	fr_app_instantiate_t		instantiate;
+	fr_radius_io_instantiate_t	instantiate;
 	size_t				io_inst_size;		//!< Size of data for parsing the configuration
 
 	module_thread_t			thread_instantiate;	//!< Callback to configure a module's instance for
diff --git a/src/modules/rlm_radius/rlm_radius_udp.c b/src/modules/rlm_radius/rlm_radius_udp.c
index 77b11acc61e..3bf617918fc 100644
--- a/src/modules/rlm_radius/rlm_radius_udp.c
+++ b/src/modules/rlm_radius/rlm_radius_udp.c
@@ -44,6 +44,8 @@ RCSID("$Id$")
  *
  */
 typedef struct rlm_radius_udp_t {
+	rlm_radius_t		*parent;		//!< rlm_radius instance
+
 	fr_ipaddr_t		dst_ipaddr;		//!< IP of the home server
 	fr_ipaddr_t		src_ipaddr;		//!< IP we open our socket on
 	uint16_t		dst_port;		//!< port of the home server
@@ -73,11 +75,15 @@ typedef struct rlm_radius_udp_thread_t {
 	fr_dlist_t		queued;			//!< queued requests for some new connection
 
 	fr_dlist_t		active;	       	//!< active connections
+	fr_dlist_t		frozen;      	//!< frozen connections
+	fr_dlist_t		opening;      	//!< opening connections
 } rlm_radius_udp_thread_t;
 
 typedef struct rlm_radius_udp_connection_t {
 	rlm_radius_udp_t const	*inst;		//!< our module instance
+	rlm_radius_udp_thread_t *thread;       	//!< our thread-specific data
 	fr_connection_t		*conn;		//!< Connection to our destination.
+	char const		*name;		//!< from IP PORT to IP PORT
 
 	fr_dlist_t		entry;		//!< in the linked list of connections
 
@@ -110,7 +116,9 @@ typedef struct rlm_radius_udp_connection_t {
 typedef struct rlm_radius_udp_request_t {
 	fr_dlist_t		entry;		//!< in the connection list of packets
 
-	rlm_radius_udp_connection_t	*c;		//!< the connection
+	int			code;		//!< packet code
+	rlm_radius_udp_connection_t	*c;     //!< the connection
+	rlm_radius_link_t	*link;		//!< more link stuff
 	rlm_radius_request_t	*rr;		//!< the ID tracking, resend count, etc.
 
 } rlm_radius_udp_request_t;
@@ -140,20 +148,339 @@ static const CONF_PARSER module_config[] = {
 	CONF_PARSER_TERMINATOR
 };
 
+static void conn_error(UNUSED fr_event_list_t *el, UNUSED int fd, UNUSED int flags, int fd_errno, void *uctx);
+static void conn_read(UNUSED fr_event_list_t *el, int fd, UNUSED int flags, void *uctx);
+static void conn_writable(UNUSED fr_event_list_t *el, int fd, UNUSED int flags, void *uctx);
+
+
+/** Set the socket to idle
+ *
+ *  But keep the read event open, just in case the other end sends us
+ *  data  That way we can process it.
+ *
+ * @param[in] c		Connection data structure
+ */
+static void fd_idle(rlm_radius_udp_connection_t *c)
+{
+	rlm_radius_udp_thread_t	*t = c->thread;
+
+	DEBUG3("Marking socket %s as idle", c->name);
+	if (fr_event_fd_insert(c->conn, t->el, c->fd,
+			       conn_read, NULL, conn_error, c) < 0) {
+		PERROR("Failed inserting FD event");
+		talloc_free(c);
+	}
+}
+
+/** Set the socket to active
+ *
+ * We have messages we want to send, so need to know when the socket is writable.
+ *
+ * @param[in] c		Connection data structure
+ */
+static void fd_active(rlm_radius_udp_connection_t *c)
+{
+	rlm_radius_udp_thread_t	*t = c->thread;
+
+	DEBUG3("Marking socket %s as active - Draining requests", c->name);
+
+	if (fr_event_fd_insert(c->conn, t->el, c->fd,
+			       conn_read, conn_writable, conn_error, c) < 0) {
+		PERROR("Failed inserting FD event");
+		talloc_free(c);
+	}
+}
+
+
+/** Connection errored
+ *
+ */
+static void conn_error(UNUSED fr_event_list_t *el, UNUSED int fd, UNUSED int flags, int fd_errno, void *uctx)
+{
+	rlm_radius_udp_connection_t *c = talloc_get_type_abort(uctx, rlm_radius_udp_connection_t);
+
+	ERROR("Connection failed %s: %s", c->name, fr_syserror(fd_errno));
+
+	/*
+	 *	Something bad happened... Fix it...
+	 */
+	fr_connection_reconnect(c->conn);
+}
+
+
+/** Read reply packets.
+ *
+ */
+static void conn_read(fr_event_list_t *el, int fd, UNUSED int flags, void *uctx)
+{
+	rlm_radius_udp_connection_t *c = talloc_get_type_abort(uctx, rlm_radius_udp_connection_t);
+	decode_fail_t reason;
+	size_t packet_len;
+	ssize_t data_len;
+
+	data_len = read(fd, c->buffer, c->buflen);
+	if (data_len == 0) return;
+
+	if (data_len < 0) {
+		conn_error(el, fd, 0, errno, c);
+		return;
+	}
+
+	packet_len = data_len;
+	if (!fr_radius_ok(c->buffer, &packet_len, false, &reason)) {
+		DEBUG("Ignoring malformed packet");
+		return;
+	}
+
+	// look the packet up by ID, and find the matching one
+	// allowing multiple packet codes in the same socket is annoying,
+	// because each request code can have one of multiple reply codes..
+	// and they all share the same ID space. :(
+
+	// verify it
+	// get the REQUEST for it
 
-static rlm_radius_udp_connection_t *mod_connect(rlm_radius_udp_t *inst, rlm_radius_udp_thread_t *t)
+	// remove it from c->sent
+	// update the status of this connection
+	// unlang_resumable(request);
+}
+
+/** There's space available to write data, so do that...
+ *
+ */
+static void conn_writable(fr_event_list_t *el, int fd, UNUSED int flags, void *uctx)
+{
+	rlm_radius_udp_connection_t *c = talloc_get_type_abort(uctx, rlm_radius_udp_connection_t);
+	fr_dlist_t *entry, *next;
+
+	/*
+	 *	Clear our backlog
+	 */
+	for (entry = FR_DLIST_FIRST(c->queued);
+	     entry != NULL;
+	     entry = next) {
+		rlm_radius_udp_request_t *u;
+		ssize_t packet_len;
+		ssize_t rcode;
+
+		next = FR_DLIST_NEXT(c->queued, entry);
+
+		u = fr_ptr_to_type(rlm_radius_udp_request_t, entry, entry);
+
+		packet_len = fr_radius_encode(c->buffer, c->buflen, NULL,
+					      c->inst->secret, u->rr->id, u->code, 0,
+					      u->link->request->packet->vps);
+		if (packet_len <= 0) break;
+
+		rcode = write(fd, c->buffer, packet_len);
+		if (rcode < 0) {
+			if (errno == EWOULDBLOCK) return;
+
+			conn_error(el, fd, 0, errno, c);
+			return;
+		}
+
+		fr_dlist_remove(&u->entry);
+		fr_dlist_insert_tail(&c->sent, &u->entry);
+	}
+
+	/*
+	 *	Check if we have to enable or disable writing on the socket.
+	 */
+	entry = FR_DLIST_FIRST(c->queued);
+	if (!entry) {
+		c->pending = false;
+		fd_idle(c);
+
+	} else if (!c->pending) {
+		c->pending = true;
+		fd_active(c);
+	}
+}
+
+/** Shutdown/close a file descriptor
+ *
+ */
+static void conn_close(int fd, void *uctx)
+{
+	rlm_radius_udp_connection_t *c = talloc_get_type_abort(uctx, rlm_radius_udp_connection_t);
+
+	DEBUG3("Closing socket %s", c->name);
+	if (shutdown(fd, SHUT_RDWR) < 0) DEBUG3("Shutdown on socket %s failed: %s", c->name, fr_syserror(errno));
+	if (close(fd) < 0) DEBUG3("Closing socket %s failed: %s", c->name, fr_syserror(errno));
+
+	c->fd = -1;
+}
+
+/** Process notification that fd is open
+ *
+ */
+static fr_connection_state_t conn_open(UNUSED fr_event_list_t *el, UNUSED int fd, void *uctx)
+{
+	rlm_radius_udp_connection_t *c = talloc_get_type_abort(uctx, rlm_radius_udp_connection_t);
+	rlm_radius_udp_thread_t *t = c->thread;
+	fr_dlist_t *entry, *next;
+
+	talloc_const_free(&c->name);
+	c->name = talloc_strdup(c, "connected");
+
+	/*
+	 *	Remove the connection from the "opening" list, and add
+	 *	it to the "active" list.
+	 */
+	fr_dlist_remove(&c->entry);
+	fr_dlist_insert_tail(&t->active, &c->entry);
+
+	/*
+	 *	Clear the global backlog first.
+	 */
+	for (entry = FR_DLIST_FIRST(t->queued);
+	     entry != NULL;
+	     entry = next) {
+		rlm_radius_udp_request_t *u;
+
+		next = FR_DLIST_NEXT(t->queued, entry);
+
+		u = fr_ptr_to_type(rlm_radius_udp_request_t, entry, entry);
+
+		// @todo - figure out if we can the request to the
+		// connection, by tracking used codes, etc.
+
+		fr_dlist_remove(entry);
+		fr_dlist_insert_tail(&c->queued, &u->entry);
+		c->pending = true;
+	}
+
+
+	/*
+	 *	If we have data pending, add the writable event immediately
+	 */
+	if (c->pending) {
+		fd_active(c);
+	} else {
+		fd_idle(c);
+	}
+
+	return FR_CONNECTION_STATE_CONNECTED;
+}
+
+
+/** Initialise a new outbound connection
+ *
+ * @param[out] fd_out	Where to write the new file descriptor.
+ * @param[in] uctx	A #rlm_radius_thread_t.
+ */
+static fr_connection_state_t conn_init(int *fd_out, void *uctx)
+{
+	int fd;
+	rlm_radius_udp_connection_t *c = talloc_get_type_abort(uctx, rlm_radius_udp_connection_t);
+
+	/*
+	 *	Open the outgoing socket.
+	 *
+	 *	@todo - pass src_ipaddr, and remove later call to fr_socket_bind()
+	 *	which does return the src_port, but doesn't set the "don't fragment" bit.
+	 */
+	fd = fr_socket_client_udp(&c->src_ipaddr, &c->dst_ipaddr, c->dst_port, true);
+	if (fd < 0) {
+		DEBUG("Failed opening RADIUS client UDP socket: %s", fr_strerror());
+		return FR_CONNECTION_STATE_FAILED;
+	}
+
+#if 0
+	if (fr_socket_bind(fd, &io->src_ipaddr, &io->src_port, inst->interface) < 0) {
+		DEBUG("Failed binding RADIUS client UDP socket: %s FD %d %pV port %u interface %s", fr_strerror(), fd, fr_box_ipaddr(io->src_ipaddr),
+			io->src_port, inst->interface);
+		return FR_CONNECTION_STATE_FAILED;
+	}
+#endif
+
+	// @todo - set name properly
+	c->name = talloc_strdup(c, "connecting...");
+
+	// @todo - set recv_buff and send_buff socket options
+
+	c->fd = fd;
+
+	// @todo - initialize the tracking memory, etc.
+
+	*fd_out = fd;
+
+	return FR_CONNECTION_STATE_CONNECTING;
+}
+
+static void mod_connect(rlm_radius_udp_t *inst, rlm_radius_udp_thread_t *t)
 {
 	rlm_radius_udp_connection_t *c;
 
 	c = talloc_zero(t, rlm_radius_udp_connection_t);
+	c->inst = inst;
+	c->thread = t;
 	c->dst_ipaddr = inst->dst_ipaddr;
 	c->dst_port = inst->dst_port;
 	c->src_ipaddr = inst->src_ipaddr;
 	c->src_port = 0;
+	c->max_packet_size = inst->max_packet_size;
+
+	c->buffer = talloc_array(c, uint8_t, c->max_packet_size);
+	if (!c->buffer) {
+		talloc_free(c);
+		return;
+	}
+	c->buflen = c->max_packet_size;
+
+	c->conn = fr_connection_alloc(c, t->el, &inst->parent->connection_timeout, &inst->parent->reconnection_delay,
+				      conn_init, conn_open, conn_close, inst->parent->name, c);
+	if (!c->conn) return;
+
+	fr_connection_start(c->conn);
+
+	fr_dlist_insert_head(&t->opening, &c->entry);
+
+	// @todo - set destructor for connection which removes it from the various lists?
+}
+
+static rlm_radius_udp_connection_t *mod_connection_get(rlm_radius_udp_thread_t *t, UNUSED int code)
+{
+	rlm_radius_udp_connection_t *c;
+	fr_dlist_t *entry;
+
+	entry = FR_DLIST_FIRST(t->active);
+	if (!entry) return NULL;
+
+	c = fr_ptr_to_type(rlm_radius_udp_connection_t, entry, entry);
+	(void) talloc_get_type_abort(c, rlm_radius_udp_connection_t);
 
 	return c;
 }
 
+static void mod_connection_add(UNUSED rlm_radius_udp_connection_t *c, UNUSED rlm_radius_udp_request_t *u)
+{
+	// do stuff
+}
+
+
+static void mod_clear_backlog(rlm_radius_udp_thread_t *t)
+{
+	fr_dlist_t *entry, *next;
+
+	for (entry = FR_DLIST_FIRST(t->queued);
+	     entry != NULL;
+	     entry = next) {
+		rlm_radius_udp_request_t *u;
+		rlm_radius_udp_connection_t *c;
+
+		next = FR_DLIST_NEXT(t->queued, entry);
+
+		u = fr_ptr_to_type(rlm_radius_udp_request_t, entry, entry);
+		c = mod_connection_get(t, u->code);
+		if (!c) return;
+
+		fr_dlist_remove(entry);
+		mod_connection_add(c, u);
+	}
+}
+
 
 static int mod_push(void *instance, REQUEST *request, rlm_radius_link_t *link, void *thread)
 {
@@ -161,38 +488,53 @@ static int mod_push(void *instance, REQUEST *request, rlm_radius_link_t *link, v
 	rlm_radius_udp_thread_t *t = talloc_get_type_abort(thread, rlm_radius_udp_thread_t);
 	rlm_radius_udp_request_t *u = link->request_io_ctx;
 	rlm_radius_udp_connection_t *c;
-	fr_dlist_t *entry;
 
 	rad_assert(request->packet->code > 0);
 	rad_assert(request->packet->code < FR_MAX_PACKET_CODE);
 
-	entry = FR_DLIST_FIRST(t->active);
-	if (!entry) {
-		c = mod_connect(inst, t);
-		if (!c) {
-			RDEBUG("Failed initializing new connection");
-			return -1;
-		}
-	}
-
 	/*
-	 *	Now that we have a connection, use it to send packets.
+	 *	Clear the backlog before sending any new packets.
 	 */
-	c = fr_ptr_to_type(rlm_radius_udp_connection_t, entry, entry);
-	(void) talloc_get_type_abort(c, rlm_radius_udp_connection_t);
+	if (t->pending) mod_clear_backlog(t);
 
-	u->c = c;
+	u->link = link;
+	u->code = request->packet->code;
 
-	if (c->pending) {
-		fr_dlist_insert_head(&c->queued, &u->entry);
+	/*
+	 *	Get a connection.  If they're all full, try to open a
+	 *	new one.
+	 */
+	c = mod_connection_get(t, u->code);
+	if (!c) {
+		fr_dlist_t *entry;
+
+		entry = FR_DLIST_FIRST(t->opening);
+		if (!entry) mod_connect(inst, t);
+
+		/*
+		 *	Add the request to the backlog.  It will be
+		 *	sent either when the new connection is open,
+		 *	or when an existing connection has
+		 *	availability.
+		 */
+		t->pending = true;
+		fr_dlist_insert_head(&t->queued, &u->entry);
 		return 0;
 	}
 
-	// @todo - try to write to the socket.  If we can, return instead of adding it to the queue
-
-	c->pending = true;
+	/*
+	 *	Insert it into the pending queue
+	 */
 	fr_dlist_insert_head(&c->queued, &u->entry);
-//	mod_fd_active(c);
+
+	/*
+	 *	If there are no active packets, try to write one
+	 *	immediately.  This avoids a few context switches in
+	 *	the case where the socket is writable.
+	 */
+	if (!c->pending) {
+		conn_writable(t->el, c->fd, 0, c);
+	}
 
 	return 0;
 }
@@ -210,7 +552,9 @@ static int mod_push(void *instance, REQUEST *request, rlm_radius_link_t *link, v
  */
 static int mod_bootstrap(UNUSED void *instance, UNUSED CONF_SECTION *conf)
 {
-//	rlm_radius_udp_t *inst = talloc_get_type_abort(instance, rlm_radius_udp_t);
+	rlm_radius_udp_t *inst = talloc_get_type_abort(instance, rlm_radius_udp_t);
+
+	(void) talloc_set_type(inst, rlm_radius_udp_t);
 
 	return 0;
 }
@@ -226,10 +570,12 @@ static int mod_bootstrap(UNUSED void *instance, UNUSED CONF_SECTION *conf)
  *	- 0 on success.
  *	- -1 on failure.
  */
-static int mod_instantiate(void *instance, CONF_SECTION *conf)
+static int mod_instantiate(rlm_radius_t *parent, void *instance, CONF_SECTION *conf)
 {
 	rlm_radius_udp_t *inst = talloc_get_type_abort(instance, rlm_radius_udp_t);
 
+	inst->parent = parent;
+
 	/*
 	 *	Ensure that we have a destination address.
 	 */
@@ -295,10 +641,12 @@ static int mod_thread_instantiate(UNUSED CONF_SECTION const *cs, void *instance,
 	t->pending = false;
 	FR_DLIST_INIT(t->queued);
 	FR_DLIST_INIT(t->active);
+	FR_DLIST_INIT(t->frozen);
+	FR_DLIST_INIT(t->opening);
 
 	// @todo - get parent, and initialize the list of IDs by code, from what is permitted by rlm_radius
 
-	// start the connection
+	mod_connect(t->inst, t);
 
 	return 0;
 }