Work in progress: libevent use one event per fd.

Signed-off-by: Jan Vidar Krey <janvidar@extatic.org>

src/hub.c

@@ -146,8 +146,8 @@ int hub_handle_support(struct user* u, struct adc_message* cmd)
 		if (ok)
 		{
 			hub_send_handshake(u);
-			if (u->ev_read)
+			if (u->ev_handle)
-				event_add(u->ev_read, &timeout);
+				event_add(u->ev_handle, &timeout);
 		}
 		else
 		{

src/hubevent.c

@@ -49,8 +49,6 @@ static void log_user_nick_change(struct user* u, const char* nick)
 /* Send MOTD, do logging etc */
 void on_login_success(struct user* u)
 {
-	struct timeval timeout = { TIMEOUT_IDLE, 0 };
-	
 	/* Send user list of all existing users */
 	if (!send_user_list(u))
 		return;
@@ -69,10 +67,6 @@ void on_login_success(struct user* u)
 	/* Send message of the day (if any) */
 	if (user_is_logged_in(u)) /* Previous send() can fail! */
 		hub_send_motd(u);
-		
-	/* reset to idle timeout */
-	if (u->ev_read)
-		event_add(u->ev_read, &timeout);
 }
 
 void on_login_failure(struct user* u, enum status_message msg)

src/netevent.c

@@ -20,31 +20,14 @@
 #include "uhub.h"
 
 
-void net_on_read(int fd, short ev, void *arg)
+static int on_read(struct user* user)
 {
 	static char buf[MAX_RECV_BUF];
-	struct user* user = (struct user*) arg;
-	char* pos;
 	size_t offset;
 	size_t buflen;
 	ssize_t size;
 	int more = 1;
-	int flag_close = 0;
+	char* pos;
-	
-	hub_log(log_trace, "net_on_read() : fd=%d, ev=%d, arg=%p", fd, (int) ev, arg);
-	
-	if (ev == EV_TIMEOUT)
-	{
-		more = 0;
-		if (user_is_connecting(user))
-		{
-			flag_close = quit_timeout;
-		}
-		else
-		{
-			hub_send_ping(user);
-		}
-	}
 
 	while (more)
 	{
@@ -55,17 +38,16 @@ void net_on_read(int fd, short ev, void *arg)
 			offset = user->recv_buf_offset;
 		}
 		
-		size = net_recv(fd, &buf[offset], MAX_RECV_BUF - offset, 0);
+		size = net_recv(user->sd, &buf[offset], MAX_RECV_BUF - offset, 0);
 		if (size == -1)
 		{
 			if (net_error() != EWOULDBLOCK)
-				flag_close = quit_socket_error;
+				return quit_socket_error;
 			break;
 		}
 		else if (size == 0)
 		{
-			flag_close = quit_disconnected;
+			return quit_disconnected;
-			break;
 		}
 		else
 		{
@@ -87,9 +69,7 @@ void net_on_read(int fd, short ev, void *arg)
 					{
 						if (hub_handle_message(user, &buf[handled], msglen) == -1)
 						{
-							flag_close = quit_protocol_error;
+							return quit_protocol_error;
-							more = 0;
-							break;
 						}
 					}
 				}
@@ -124,8 +104,7 @@ void net_on_read(int fd, short ev, void *arg)
 					}
 					else
 					{
-						flag_close = quit_memory_error;
+						return quit_memory_error;
-						break;
 					}
 				}
 			}
@@ -140,39 +119,14 @@ void net_on_read(int fd, short ev, void *arg)
 			}
 		}
 	}
-	
+	return 0;
-	if (flag_close)
-	{
-		user_disconnect(user, flag_close);
-		return;
-	}
-	
-	if (user_is_logged_in(user))
-	{
-		if (user->ev_read)
-		{
-			struct timeval timeout = { TIMEOUT_IDLE, 0 };
-			event_add(user->ev_read, &timeout);
-		}
-	}
-	else if (user_is_connecting(user))
-	{
-		if (user->ev_read)
-		{
-			struct timeval timeout = { TIMEOUT_HANDSHAKE, 0 };
-			event_add(user->ev_read, &timeout);
-		}
-	}
 }
 
-
+static int on_write(struct user* user)
-void net_on_write(int fd, short ev, void *arg)
 {
-	struct user* user = (struct user*) arg;
 	struct adc_message* msg;
 	int ret;
 	int length;
-	int close_flag = 0;
 	
 	msg = list_get_first(user->send_queue);
 	while (msg)
@@ -182,29 +136,16 @@ void net_on_write(int fd, short ev, void *arg)
 	
 		if (ret == 0 || (ret == -1 && net_error() == EWOULDBLOCK))
 		{
-			close_flag = 0;
+			return 0;
-			break;
 		}
 		else if (ret > 0)
 		{
-			
 			user->tm_last_write = time(NULL);
 			
 			if (ret == length)
 			{
-#ifdef DEBUG_SENDQ
-				hub_log(log_error, "SENDQ: sent=%d bytes/%d (all), send_queue_size=%d, offset=%d", ret, (int) msg->length, user->send_queue_size, user->send_queue_offset);
-#endif
 				user->send_queue_size -= ret;
 				user->send_queue_offset = 0;
-			
-#ifdef DEBUG_SENDQ
-				if ((user->send_queue_size < 0) || (user->send_queue_offset < 0))
-				{
-					hub_log(log_error, "INVALID: send_queue_size=%d, send_queue_offset=%d", user->send_queue_size, user->send_queue_offset);
-				}
-#endif
-			
 				list_remove(user->send_queue, msg);
 				
 				if (user_flag_get(user, flag_user_list) && (msg == user->info || user->send_queue_size == 0))
@@ -220,49 +161,81 @@ void net_on_write(int fd, short ev, void *arg)
 			}
 			else
 			{
-#ifdef DEBUG_SENDQ
-                                hub_log(log_error, "SENDQ: sent=%d bytes/%d (part), send_queue_size=%d, offset=%d", ret, (int) msg->length, user->send_queue_size, user->send_queue_offset);
-#endif
 				user->send_queue_size -= ret;
 				user->send_queue_offset += ret;
-				
-#ifdef DEBUG_SENDQ				
-				if ((user->send_queue_size < 0) || (user->send_queue_offset < 0) || (user->send_queue_offset > msg->length))
-				{
-					hub_log(log_error, "INVALID: send_queue_size=%d, send_queue_offset=%d", user->send_queue_size, user->send_queue_offset);
-				}
-#endif
 				break;
 			}
 		}
 		else
 		{
-			close_flag = quit_socket_error;
+			return quit_socket_error;
-			break;
 		}
 		msg = list_get_first(user->send_queue);
 	}
+	return 0;
+}
 
 
-	if (close_flag)
+void on_net_event(int fd, short ev, void *arg)
+{
+	struct user* user = (struct user*) arg;
+	int want_close = 0;
+	int want_write = 0;
+	
+	hub_log(log_debug, "on_net_event() : fd=%d, ev=%d, user=%s", fd, (int) ev, user);
+	
+	if (ev == EV_TIMEOUT)
 	{
-		user_disconnect(user, close_flag);
+		
+		hub_log(log_debug, "EV_TIMEOUT");
+		
+		if (user_is_connecting(user))
+		{
+			want_close = quit_timeout;
 		}
 		else
 		{
-		if (user->send_queue_size > 0 && user->ev_write)
+			hub_send_ping(user);
-			event_add(user->ev_write, NULL);
+		}
+	}
+	else
+	{
+		if (ev & EV_WRITE)
+		{
+			want_close = on_write(user);
+			want_write = (user->send_queue_size != 0);
+		}
+		
+		if (!want_close && ev & EV_READ)
+		{
+			want_close = on_read(user);
+		}
+	}
+		
+	if (want_close)
+	{
+		user_disconnect(user, want_close);
+		return;
+	}
+	
+	if (user_is_logged_in(user))
+	{
+		user_trigger_update(user, want_write, TIMEOUT_IDLE);
+	}
+	else if (user_is_connecting(user))
+	{
+		user_trigger_update(user, want_write, TIMEOUT_HANDSHAKE);
 	}
 }
 
 
+
 void net_on_accept(int server_fd, short ev, void *arg)
 {
 	struct hub_info* hub = (struct hub_info*) arg;
 	struct user* user = 0;
 	struct ip_addr_encap ipaddr;
 	const char* addr;
-	struct timeval timeout = { TIMEOUT_CONNECTED, 0 };
 	
 	for (;;)
 	{
@@ -306,12 +279,7 @@ void net_on_accept(int server_fd, short ev, void *arg)
 
 		net_set_nonblocking(fd, 1);
 		net_set_nosigpipe(fd, 1);
-		
+		user_trigger_init(user);
-		event_set(user->ev_read,  fd, EV_READ | EV_PERSIST, net_on_read,  user);
-		event_set(user->ev_write, fd, EV_WRITE,             net_on_write, user);
-		event_base_set(hub->evbase, user->ev_read);
-		event_base_set(hub->evbase, user->ev_write);
-		event_add(user->ev_read,  &timeout);
 	}
 }
 

src/netevent.h

@@ -20,33 +20,19 @@
 #ifndef HAVE_UHUB_NET_EVENT_H
 #define HAVE_UHUB_NET_EVENT_H
 
-/**
+struct user;
- * Network callback for reading data from a socket.
- */
-extern void net_on_read(int fd, short ev, void *arg);
 
-/**
+extern void on_net_event(int fd, short ev, void *arg);
- * Network callback for writing data to a socket.
- */
-extern void net_on_write(int fd, short ev, void *arg);
-
-/**
- * Network callback for timers.
- */
-extern void net_on_read_timeout(int fd, short ev, void* arg);
 
 /**
  * Network callback to accept incoming connections.
  */
 extern void net_on_accept(int fd, short ev, void *arg);
 
-#ifdef ADC_UDP_OPERATION
 /**
- * Network callback to receive incoming UDP datagram.
+ *
  */
-extern void net_on_packet(int fd, short ev, void *arg);
+extern void net_update_trigger(struct user* user, int write, struct timeval timeout);
-#endif
-
 
 #endif /* HAVE_UHUB_NET_EVENT_H */
 

src/route.c

@@ -135,8 +135,8 @@ int route_to_user(struct user* user, struct adc_message* msg)
 		{
 			queue_command(user, msg, ret);
 			
-			if (user->send_queue_size && user->ev_write)
+			if (user->send_queue_size && user->ev_handle)
-				event_add(user->ev_write, NULL);
+				event_add(user->ev_handle, NULL);
 		}
 		else
 		{
@@ -158,8 +158,8 @@ int route_to_user(struct user* user, struct adc_message* msg)
 		{
 			/* queue command */
 			queue_command(user, msg, 0);
-                        if (user->ev_write)
+                        if (user->ev_handle)
-				event_add(user->ev_write, NULL);
+				event_add(user->ev_handle, NULL);
 
 		}
 		else

src/user.c

@@ -30,13 +30,11 @@ struct user* user_create(struct hub_info* hub, int sd)
 	if (user == NULL)
 		return NULL; /* OOM */
 
-	user->ev_write = hub_malloc_zero(sizeof(struct event));
+	user->ev_handle = hub_malloc_zero(sizeof(struct event));
-	user->ev_read  = hub_malloc_zero(sizeof(struct event));
 
-	if (!user->ev_write || !user->ev_read)
+	if (!user->ev_handle)
 	{
-	    hub_free(user->ev_read);
+	    hub_free(user->ev_handle);
-	    hub_free(user->ev_write);
 	    hub_free(user);
 	    return NULL;
 	}
@@ -65,22 +63,8 @@ void user_destroy(struct user* user)
 {
 	hub_log(log_trace, "user_destroy(), user=%p", user);
 
-	if (user->ev_write)
+	user_trigger_shutdown(user);
-	{
-		event_del(user->ev_write);
-		hub_free(user->ev_write);
-		user->ev_write = 0;
-	}
-	
-	if (user->ev_read)
-	{
-		event_del(user->ev_read);
-		hub_free(user->ev_read);
-		user->ev_read = 0;
-	}
-	
 	net_close(user->sd);
-	
 	adc_msg_free(user->info);
 	user_clear_feature_cast_support(user);
 	
@@ -213,12 +197,7 @@ void user_disconnect(struct user* user, int reason)
 	}
 	
 	/* dont read more data from this user */
-	if (user->ev_read)
+	user_flag_set(user, flag_ignore);
-	{
-		event_del(user->ev_read);
-		hub_free(user->ev_read);
-		user->ev_read = 0;
-	}
 	
 	hub_log(log_trace, "user_disconnect(), user=%p, reason=%d, state=%d", user, reason, user->state);
 	
@@ -303,6 +282,36 @@ int user_is_disconnecting(struct user* user)
 	return 0;
 }
 
+void user_trigger_init(struct user* user)
+{
+	if (user->ev_handle)
+	{
+		struct timeval timeout = { TIMEOUT_CONNECTED, 0 };
+		event_set(user->ev_handle, user->sd, EV_READ | EV_PERSIST, on_net_event, user);
+		event_base_set(user->hub->evbase, user->ev_handle);
+		event_add(user->ev_handle, &timeout);
+	}
+}
 
+void user_trigger_update(struct user* user, int w, int timeout_sec)
+{
+	if (user->ev_handle)
+	{
+		struct timeval timeout = { timeout_sec, 0 };
+		int f = w ? EV_WRITE | EV_READ | EV_PERSIST : EV_READ | EV_PERSIST;
+		event_del(user->ev_handle);
+		event_set(user->ev_handle, user->sd, f, on_net_event, user);
+		event_base_set(user->hub->evbase, user->ev_handle);
+		event_add(user->ev_handle, &timeout);
+	}
+}
 
-
+void user_trigger_shutdown(struct user* user)
+{
+	if (user->ev_handle)
+	{
+		event_del(user->ev_handle);
+		hub_free(user->ev_handle);
+		user->ev_handle = 0;
+	}
+}

src/user.h

@@ -99,8 +99,7 @@ struct user_counts
 struct user
 {
 	int                  sd;                      /** socket descriptor */
-	struct event*        ev_read;                 /** libevent struct for read events */
+	struct event*        ev_handle;
-	struct event*        ev_write;                /** libevent struct for write events */
 	enum user_state      state;                   /** see enum user_state */
 	enum user_credentials credentials;            /** see enum user_credentials */
 	struct user_info     id;                      /** Contains nick name and CID */
@@ -256,7 +255,9 @@ extern int user_set_feature_cast_support(struct user* u, char feature[4]);
  */
 extern void user_clear_feature_cast_support(struct user* u);
 
-
+extern void user_trigger_init(struct user* user);
+extern void user_trigger_update(struct user* user, int w, int timeout_sec);
+extern void user_trigger_shutdown(struct user* user);
 
 #endif /* HAVE_UHUB_USER_H */