Fix some nasty bugs related to read/write events and timeout events.

This would have caused users not being able to log in, and in some cases
100% cpu usage.

src/hub.c

@@ -102,7 +102,6 @@ int hub_handle_support(struct hub_info* hub, struct user* u, struct adc_message*
 	int index = 0;
 	int ok = 1;
 	char* arg = adc_msg_get_argument(cmd, index);
-	struct timeval timeout = { TIMEOUT_HANDSHAKE, 0 };
 
 	if (hub->status == hub_status_disabled && u->state == state_protocol)
 	{
@@ -146,8 +145,7 @@ int hub_handle_support(struct hub_info* hub, struct user* u, struct adc_message*
 		if (ok)
 		{
 			hub_send_handshake(hub, u);
-			if (u->net.ev_read)
+			user_set_timeout(u, TIMEOUT_HANDSHAKE);
-				event_add(u->net.ev_read, &timeout);
 		}
 		else
 		{
@@ -929,17 +927,8 @@ void hub_disconnect_user(struct hub_info* hub, struct user* user, int reason)
 	{
 		return;
 	}
-	
-	/* dont read more data from this user */
-	/* FIXME: Remove this from here! */
-	if (user->net.ev_read)
-	{
-		event_del(user->net.ev_read);
-		hub_free(user->net.ev_read);
-		user->net.ev_read = 0;
-	}
 
-	/* this should be enough? */
+	/* stop reading from user */
 	net_shutdown_r(user->net.sd);
 	
 	hub_log(log_trace, "hub_disconnect_user(), user=%p, reason=%d, state=%d", user, reason, user->state);

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 hub_info* hub, struct user* u)
 {
-	struct timeval timeout = { TIMEOUT_IDLE, 0 };
-	
 	/* Send user list of all existing users */
 	if (!uman_send_user_list(hub, u))
 		return;
@@ -71,8 +69,7 @@ void on_login_success(struct hub_info* hub, struct user* u)
 		hub_send_motd(hub, u);
 		
 	/* reset to idle timeout */
-	if (u->net.ev_read)
+	user_set_timeout(u, TIMEOUT_IDLE);
-		event_add(u->net.ev_read, &timeout);
 }
 
 void on_login_failure(struct hub_info* hub, struct user* u, enum status_message msg)

src/netevent.c

@@ -212,7 +212,7 @@ int handle_net_read(struct user* user)
 
 int handle_net_write(struct user* user)
 {
-	for (;;)
+	while (hub_sendq_get_bytes(user->net.send_queue))
 	{
 		int ret = hub_sendq_send(user->net.send_queue, net_user_send, user);
 		if (ret <= 0)
@@ -223,6 +223,10 @@ int handle_net_write(struct user* user)
 	{
 		user_net_io_want_write(user);
 	}
+	else
+	{
+		user_net_io_want_read(user);
+	}
 	return 0;
 }
 
@@ -235,7 +239,7 @@ void net_event(int fd, short ev, void *arg)
 	hub_log(log_trace, "net_on_read() : fd=%d, ev=%d, arg=%p", fd, (int) ev, arg);
 #endif
 
-	if (ev == EV_TIMEOUT)
+	if (ev & EV_TIMEOUT)
 	{
 		if (user_is_connecting(user))
 		{
@@ -247,44 +251,28 @@ void net_event(int fd, short ev, void *arg)
 			// hub_send_ping(hub, user);
 		}
 	}
-	else if (ev == EV_READ)
+
+	if (ev & EV_READ)
 	{
 		flag_close = handle_net_read(user);
 	}
-	else if (ev == EV_WRITE)
+
+	if (ev & EV_WRITE)
 	{
 		flag_close = handle_net_write(user);
 	}
-	
+
 	if (flag_close)
 	{
 		hub_disconnect_user(g_hub, user, flag_close);
 		return;
 	}
-	
-	if (user_is_logged_in(user))
-	{
-		if (user->net.ev_read)
-		{
-			struct timeval timeout = { TIMEOUT_IDLE, 0 };
-			event_add(user->net.ev_read, &timeout);
-		}
-	}
-	else if (user_is_connecting(user))
-	{
-		if (user->net.ev_read)
-		{
-			struct timeval timeout = { TIMEOUT_HANDSHAKE, 0 };
-			event_add(user->net.ev_read, &timeout);
-		}
-	}
 }
 
 
 static void prepare_user_net(struct hub_info* hub, struct user* user)
 {
 		int fd = user->net.sd;
-		struct timeval timeout = { TIMEOUT_CONNECTED, 0 };
 
 #ifdef SET_SENDBUG
 		size_t sendbuf = 0;
@@ -305,10 +293,6 @@ static void prepare_user_net(struct hub_info* hub, struct user* user)
 
 		net_set_nonblocking(fd, 1);
 		net_set_nosigpipe(fd, 1);
-
-		event_set(user->net.ev_read,  fd, EV_READ | EV_PERSIST, net_event, user);
-		event_base_set(hub->evbase, user->net.ev_read);
-		event_add(user->net.ev_read,  &timeout);
 }
 
 void net_on_accept(int server_fd, short ev, void *arg)

src/route.c

@@ -18,7 +18,6 @@
  */
 
 #include "uhub.h"
-#define DEBUG_SENDQ 1
 
 int route_message(struct hub_info* hub, struct user* u, struct adc_message* msg)
 {
@@ -110,8 +109,7 @@ int route_to_user(struct hub_info* hub, struct user* user, struct adc_message* m
 		/* Perform oportunistic write */
 		handle_net_write(user);
 	}
-
+	else
-	if (hub_sendq_get_bytes(user->net.send_queue))
 	{
 		user_net_io_want_write(user);
 	}

src/user.c

@@ -46,20 +46,21 @@ struct user* user_create(struct hub_info* hub, int sd)
 	if (user == NULL)
 		return NULL; /* OOM */
 
-	user->net.ev_read  = hub_malloc_zero(sizeof(struct event));
-
-	if (!user->net.ev_read)
-	{
-	    hub_free(user->net.ev_read);
-	    hub_free(user);
-	    return NULL;
-	}
-	
 	user->net.sd = sd;
 	user->net.tm_connected = time(NULL);
 	user->net.send_queue = hub_sendq_create();
 	user->net.recv_queue = hub_recvq_create();
+
+	event_set(&user->net.event, sd, EV_READ | EV_PERSIST, net_event, user);
+	event_base_set(hub->evbase, &user->net.event);
+	event_add(&user->net.event, 0);
+
+	evtimer_set(&user->net.timeout, net_event, user);
+	event_base_set(hub->evbase, &user->net.timeout);
 	
+
+	user_set_timeout(user, TIMEOUT_CONNECTED);
+
 	user_set_state(user, state_protocol);
 	return user;
 }
@@ -69,12 +70,7 @@ void user_destroy(struct user* user)
 {
 	hub_log(log_trace, "user_destroy(), user=%p", user);
 
-	if (user->net.ev_read)
+	event_del(&user->net.event);
-	{
-		event_del(user->net.ev_read);
-		hub_free(user->net.ev_read);
-		user->net.ev_read = 0;
-	}
 	
 	hub_recvq_destroy(user->net.recv_queue);
 	hub_sendq_destroy(user->net.send_queue);
@@ -335,24 +331,19 @@ void user_net_io_want_write(struct user* user)
 #ifdef DEBUG_SENDQ
 	hub_log(log_trace, "user_net_io_want_write: %s", user_log_str(user));
 #endif
-	if (user && user->net.ev_read)
+	event_del(&user->net.event);
-	{
+	event_set(&user->net.event,  user->net.sd, EV_READ | EV_WRITE | EV_PERSIST, net_event, user);
-		event_set(user->net.ev_read,  user->net.sd, EV_READ | EV_WRITE | EV_PERSIST, net_event, user);
+	event_add(&user->net.event, 0);
-		event_add(user->net.ev_read, 0);
-	}
 }
 
-void user_net_io_want_read(struct user* user, int timeout_s)
+void user_net_io_want_read(struct user* user)
 {
 #ifdef DEBUG_SENDQ
 	hub_log(log_trace, "user_net_io_want_read: %s", user_log_str(user));
 #endif
-	struct timeval timeout = { timeout_s, 0 };
+	event_del(&user->net.event);
-	if (user && user->net.ev_read)
+	event_set(&user->net.event,  user->net.sd, EV_READ | EV_PERSIST, net_event, user);
-	{
+	event_add(&user->net.event, 0);
-		event_set(user->net.ev_read,  user->net.sd, EV_READ | EV_PERSIST, net_event, user);
-		event_add(user->net.ev_read, &timeout);
-	}
 }
 
 void user_reset_last_write(struct user* user)
@@ -365,4 +356,13 @@ void user_reset_last_read(struct user* user)
 	user->net.tm_last_read = time(NULL);
 }
 
+void user_set_timeout(struct user* user, int seconds)
+{
+#ifdef DEBUG_SENDQ
+	hub_log(log_trace, "user_set_timeout to %d seconds: %s", seconds, user_log_str(user));
+#endif
+	struct timeval timeout = { seconds, 0 };
+	evtimer_add(&user->net.timeout, &timeout);
+}
+
 

src/user.h

@@ -95,7 +95,8 @@ struct user_limits
 struct user_net_io
 {
 	int                  sd;                      /** socket descriptor */
-	struct event*        ev_read;                 /** libevent struct for read events */
+	struct event         event;                   /** libevent struct for read/write events */
+	struct event         timeout;                 /** timeout handling */
 
 	struct hub_recvq*    recv_queue;
 	struct hub_sendq*    send_queue;
@@ -286,7 +287,13 @@ extern void user_net_io_want_write(struct user* user);
 /**
  * Mark the user with a want read flag, meaning it should poll for readability.
  */
-extern void user_net_io_want_read(struct user* user, int timeout_s);
+extern void user_net_io_want_read(struct user* user);
+
+/**
+ * Set timeout for connetion.
+ * @param seconds the number of seconds into the future.
+ */
+extern void user_set_timeout(struct user* user, int seconds);
 
 /**
  * Reset the last-write timer.