rm queue.c usage

Our use of the queue.c is broken and not needed so remove it.

13 files changed, 304 insertions, 727 deletions

diff --git a/usr/Makefile b/usr/Makefile
index db33ed1..4a8dcae 100644
--- a/usr/Makefile
+++ b/usr/Makefile
@@ -38,7 +38,7 @@ ISCSI_LIB_SRCS = util.o io.o auth.o login.o log.o md5.o sha1.o iscsi_sysfs.o idb
 # sources shared between iscsid and iscsiadm
 COMMON_SRCS =  $(ISCSI_LIB_SRCS)
 # core initiator files
-INITIATOR_SRCS = initiator.o queue.o actor.o mgmt_ipc.o isns.o transport.o
+INITIATOR_SRCS = initiator.o actor.o mgmt_ipc.o isns.o transport.o
 
 all: $(PROGRAMS)
 
diff --git a/usr/actor.c b/usr/actor.c
index 90b12e2..9062f85 100644
--- a/usr/actor.c
+++ b/usr/actor.c
@@ -115,6 +115,7 @@ actor_schedule_private(actor_t *thread, uint32_t ttschedule)
 	switch(thread->state) {
 	case ACTOR_WAITING:
 		log_error("rescheduling a waiting thread!");
+		remque(&thread->item);
 	case ACTOR_NOTSCHEDULED:
 		/* if ttschedule is 0, put in scheduled queue and change
 		 * state to scheduled, else add current time to ttschedule and
@@ -152,6 +153,10 @@ actor_schedule_private(actor_t *thread, uint32_t ttschedule)
 	case ACTOR_SCHEDULED:
 		// don't do anything
 		break;
+	case ACTOR_INVALID:
+		log_error("BUG: Trying to schedule a thread that has not been "
+			  "setup. Ignoring sched.");
+		break;
 	}
 
 }
diff --git a/usr/actor.h b/usr/actor.h
index 88c69d6..b1c966e 100644
--- a/usr/actor.h
+++ b/usr/actor.h
@@ -25,6 +25,7 @@
 #define ACTOR_MAX_LOOPS		1
 
 typedef enum actor_state_e {
+    ACTOR_INVALID,
     ACTOR_WAITING,
     ACTOR_SCHEDULED,
     ACTOR_NOTSCHEDULED,
diff --git a/usr/initiator.c b/usr/initiator.c
index b4f2476..4088d96 100644
--- a/usr/initiator.c
+++ b/usr/initiator.c
@@ -47,8 +47,9 @@
 #include "iscsi_sysfs.h"
 #include "iscsi_settings.h"
 
-static void __session_mainloop(void *data);
 static void __conn_error_handle(iscsi_session_t*, iscsi_conn_t*);
+static void iscsi_login_timedout(void *data);
+static void session_conn_poll(void *data);
 
 #define DEFAULT_TIME2WAIT 2
 
@@ -69,65 +70,75 @@ __padding(unsigned int param)
 	return param + pad;
 }
 
-static int
-__recvpool_alloc(iscsi_conn_t *conn)
+static int iscsi_conn_context_alloc(iscsi_conn_t *conn)
 {
 	int i;
 
-	for (i = 0; i < RECVPOOL_MAX; i++) {
-		conn->recvpool[i] = calloc(1, ipc->ctldev_bufmax);
-		if (!conn->recvpool[i]) {
+	for (i = 0; i < CONTEXT_POOL_MAX; i++) {
+		conn->context_pool[i] = calloc(1,
+					   sizeof(struct iscsi_conn_context) +
+					   ipc->ctldev_bufmax);
+		if (!conn->context_pool[i]) {
 			int j;
 			for (j = 0; j < i; j++)
-				free(conn->recvpool[j]);
+				free(conn->context_pool[j]);
 			return -ENOMEM;
 		}
+		conn->context_pool[i]->conn = conn;
 	}
 
 	return 0;
 }
 
-static void
-__recvpool_free(iscsi_conn_t *conn)
+static void iscsi_conn_context_free(iscsi_conn_t *conn)
 {
 	int i;
 
-	for (i = 0; i < RECVPOOL_MAX; i++) {
-		if (!conn->recvpool[i]) {
-			log_error("recvpool leak: %d bytes",
-				  ipc->ctldev_bufmax);
-		} else
-			free(conn->recvpool[i]);
+	for (i = 0; i < CONTEXT_POOL_MAX; i++) {
+		if (!conn->context_pool[i])
+			continue;
+
+		if (conn->context_pool[i]->allocated)
+			/* missing flush on shutdown */
+			log_error("BUG: context_pool leak");
+		free(conn->context_pool[i]);
 	}
 }
 
-void* recvpool_get(iscsi_conn_t *conn, int ev_size)
+struct iscsi_conn_context *iscsi_conn_context_get(iscsi_conn_t *conn,
+						  int ev_size)
 {
+	struct iscsi_conn_context *conn_context;
 	int i;
 
 	if (ev_size > ipc->ctldev_bufmax)
 		return NULL;
 
-	for (i = 0; i < RECVPOOL_MAX; i++) {
-		if (conn->recvpool[i]) {
-			void *handle = conn->recvpool[i];
-			conn->recvpool[i] = NULL;
-			return handle;
+	for (i = 0; i < CONTEXT_POOL_MAX; i++) {
+		if (!conn->context_pool[i])
+			continue;
+
+		if (!conn->context_pool[i]->allocated) {
+			conn_context = conn->context_pool[i];
+
+			memset(&conn_context->actor, 0,
+				sizeof(struct actor));
+			conn_context->allocated = 1;
+			/* some callers abuse this pointer */
+			conn_context->data = conn_context +
+					sizeof(struct iscsi_conn_context);
+			log_debug(7, "get conn context %p",
+				  &conn_context->actor);
+			return conn_context;
 		}
 	}
 	return NULL;
 }
 
-void recvpool_put(iscsi_conn_t *conn, void *handle)
+void iscsi_conn_context_put(struct iscsi_conn_context *conn_context)
 {
-	int i;
-
-	for (i = 0; i < RECVPOOL_MAX; i++) {
-		if (!conn->recvpool[i]) {
-			conn->recvpool[i] = handle;
-			break;
-		}
-	}
+	log_debug(7, "put conn context %p", &conn_context->actor);
+	conn_context->allocated = 0;
 }
 
 static void session_online_devs(int host_no, int sid)
@@ -343,8 +354,8 @@ __session_conn_create(iscsi_session_t *session, int cid)
 	conn_rec_t *conn_rec = &session->nrec.conn[cid];
 	int err;
 
-	if (__recvpool_alloc(conn)) {
-		log_error("cannot allocate recvpool for conn cid %d", cid);
+	if (iscsi_conn_context_alloc(conn)) {
+		log_error("cannot allocate context_pool for conn cid %d", cid);
 		return -ENOMEM;
 	}
 
@@ -430,81 +441,16 @@ __session_conn_create(iscsi_session_t *session, int cid)
 }
 
 static void
-__send_pdu_timedout(void *data)
-{
-	queue_task_t *qtask = data;
-	iscsi_conn_t *conn = qtask->conn;
-	iscsi_session_t *session = conn->session;
-
-	if (conn->send_pdu_in_progress) {
-		/*
-		 * redirect timeout processing to __session_conn_timer()
-		 */
-		queue_produce(session->queue, EV_CONN_TIMER, qtask, 0, NULL);
-		actor_schedule(&session->mainloop);
-		log_debug(7, "send_pdu timer timedout!");
-	}
-}
-
-static void
-__send_pdu_timer_add(struct iscsi_conn *conn, int timeout)
-{
-	if (conn->state == STATE_IN_LOGIN) {
-		iscsi_login_context_t *c = &conn->login_context;
-		conn->send_pdu_in_progress = 1;
-		actor_timer(&conn->send_pdu_timer, timeout*1000,
-			    __send_pdu_timedout, c->qtask);
-		log_debug(7, "send_pdu timer added %d secs", timeout);
-	}
-}
-
-static void
-__send_pdu_timer_remove(struct iscsi_conn *conn)
-{
-	if (conn->send_pdu_in_progress) {
-		actor_delete(&conn->send_pdu_timer);
-		conn->send_pdu_in_progress = 0;
-		log_debug(7, "send_pdu timer removed");
-	}
-}
-
-
-static void
-session_stop_conn_timers(iscsi_session_t *session, int cid)
-{
-	iscsi_conn_t *conn = &session->conn[cid];
-
-	__send_pdu_timer_remove(conn);
-	actor_delete(&conn->connect_timer);
-}
-
-static void
 session_release(iscsi_session_t *session)
 {
 	log_debug(2, "Releasing session %p", session);
 
 	if (session->target_alias)
 		free(session->target_alias);
-	__recvpool_free(&session->conn[0]);
-	actor_delete(&session->mainloop);
-	queue_destroy(session->queue);
+	iscsi_conn_context_free(&session->conn[0]);
 	free(session);
 }
 
-static void
-session_put(iscsi_session_t *session)
-{
-	session->refcount--;
-	if (session->refcount == 0)
-		session_release(session);
-}
-
-static void
-session_get(iscsi_session_t *session)
-{
-	session->refcount++;
-}
-
 static iscsi_session_t*
 __session_create(node_rec_t *rec, struct iscsi_transport *t)
 {
@@ -518,7 +464,6 @@ __session_create(node_rec_t *rec, struct iscsi_transport *t)
 	log_debug(2, "Allocted session %p", session);
 
 	INIT_LIST_HEAD(&session->list);
-	session_get(session);
 	/* opened at daemon load time (iscsid.c) */
 	session->ctrl_fd = control_fd;
 	session->t = t;
@@ -527,18 +472,6 @@ __session_create(node_rec_t *rec, struct iscsi_transport *t)
 	/* save node record. we might need it for redirection */
 	memcpy(&session->nrec, rec, sizeof(node_rec_t));
 
-	/* initalize per-session queue */
-	session->queue = queue_create(4, 4, NULL, session);
-	if (session->queue == NULL) {
-		log_error("can not create session's queue");
-		free(session);
-		return NULL;
-	}
-
-	/* initalize per-session event processor */
-	actor_new(&session->mainloop, __session_mainloop, session);
-	actor_schedule(&session->mainloop);
-
 	/* session's operational parameters */
 	session->initial_r2t_en = rec->session.iscsi.InitialR2T;
 	session->imm_data_en = rec->session.iscsi.ImmediateData;
@@ -610,20 +543,21 @@ __session_create(node_rec_t *rec, struct iscsi_transport *t)
 	return session;
 }
 
-static void iscsi_queue_flush(queue_t *queue)
+static void iscsi_flush_context_pool(struct iscsi_session *session)
 {
-	unsigned char item_buf[sizeof(queue_item_t) + EVENT_PAYLOAD_MAX];
-	queue_item_t *item = (queue_item_t *)(void *)item_buf;
+	struct iscsi_conn_context *conn_context;
+	struct iscsi_conn *conn = &session->conn[0];
+	int i;
 
-	/* flush queue by consuming all enqueued items */
-	while (queue_consume(queue, EVENT_PAYLOAD_MAX,
-			     item) != QUEUE_IS_EMPTY) {
-		uintptr_t recv_handle = *(uintptr_t *)queue_item_data(item);
+	for (i = 0; i < CONTEXT_POOL_MAX; i++) {
+		conn_context = conn->context_pool[i];
+		if (!conn_context)
+			continue;
 
-		log_debug(7, "item %p(%d) data size %d flushed", item,
-			  item->event_type, item->data_size);
-		if (item->data_size)
-			recvpool_put(item->context, (void*)recv_handle);
+		if (conn_context->allocated) {
+			actor_delete(&(conn->context_pool[i]->actor));
+			iscsi_conn_context_put(conn_context);
+		}
 	}
 }
 
@@ -632,8 +566,8 @@ __session_destroy(iscsi_session_t *session)
 {
 	log_debug(1, "destroying session\n");
 	list_del(&session->list);
-	iscsi_queue_flush(session->queue);
-	session_put(session);
+	iscsi_flush_context_pool(session);
+	session_release(session);
 }
 
 static void
@@ -654,7 +588,6 @@ session_conn_cleanup(queue_task_t *qtask, mgmt_ipc_err_e err)
 	iscsi_session_t *session = conn->session;
 
 	mgmt_ipc_write_rsp(qtask, err);
-	session_stop_conn_timers(session, conn->id);
 	__session_destroy(session);
 }
 
@@ -665,8 +598,7 @@ __session_conn_shutdown(iscsi_conn_t *conn, queue_task_t *qtask,
 	iscsi_session_t *session = conn->session;
 
 	__conn_noop_out_delete(conn);
-	actor_delete(&conn->connect_timer);
-	iscsi_queue_flush(session->queue);
+	iscsi_flush_context_pool(session);
 
 	if (ipc->destroy_conn(session->t->handle, session->id,
 		conn->id)) {
@@ -764,30 +696,26 @@ __conn_noop_out(void *data)
 }
 
 static void
-__connect_timedout(void *data)
-{
-	queue_task_t *qtask = data;
-	iscsi_conn_t *conn = qtask->conn;
-	iscsi_session_t *session = conn->session;
-
-	if (conn->state == STATE_XPT_WAIT) {
-		/* flush any polls or other events queued */
-		iscsi_queue_flush(session->queue);
-		log_debug(3, "__connect_timedout queue EV_CONN_TIMER\n");
-		queue_produce(session->queue, EV_CONN_TIMER, qtask, 0, NULL);
-		actor_schedule(&session->mainloop);
-	}
-}
-
-static void
 queue_delayed_reopen(queue_task_t *qtask, int delay)
 {
 	iscsi_conn_t *conn = qtask->conn;
+	struct iscsi_conn_context *conn_context;
 
 	log_debug(4, "Requeue reopen attempt in %d secs\n", delay);
-	actor_delete(&conn->connect_timer);
-	actor_timer(&conn->connect_timer, delay * 1000,
-		    __connect_timedout, qtask);
+
+	conn_context = iscsi_conn_context_get(conn, 0);
+	if (!conn_context) {
+		/* the queue should be full here */
+		log_error("BUG: queue_delayed_reopen could not get conn "
+			  "context for delayed reopen.");
+	}
+	conn_context->data = qtask;
+	/*
+ 	 * The EV_CONN_LOGIN_TIMER can handle the login resched as
+ 	 * if it were login time out
+ 	 */
+	iscsi_sched_conn_context(conn_context, conn, delay,
+				 EV_CONN_LOGIN_TIMER);
 }
 
 static void
@@ -823,6 +751,7 @@ __session_conn_reopen(iscsi_conn_t *conn, queue_task_t *qtask, int do_stop)
 {
 	int rc, delay;
 	iscsi_session_t *session = conn->session;
+	struct iscsi_conn_context *conn_context;
 
 	log_debug(1, "re-opening session %d (reopen_cnt %d)", session->id,
 			session->reopen_cnt);
@@ -831,11 +760,8 @@ __session_conn_reopen(iscsi_conn_t *conn, queue_task_t *qtask, int do_stop)
 	qtask->conn = conn;
 
 	/* flush stale polls or errors queued */
-	iscsi_queue_flush(session->queue);
-	actor_delete(&conn->connect_timer);
+	iscsi_flush_context_pool(session);
 	__conn_noop_out_delete(conn);
-
-	__send_pdu_timer_remove(conn);
 	conn->state = STATE_XPT_WAIT;
 
 	if (do_stop) {
@@ -868,12 +794,26 @@ __session_conn_reopen(iscsi_conn_t *conn, queue_task_t *qtask, int do_stop)
 		goto queue_reopen;
 	}
 
-	queue_produce(session->queue, EV_CONN_POLL, qtask, 0, NULL);
-	actor_schedule(&session->mainloop);
-
-	actor_timer(&conn->connect_timer, conn->login_timeout*1000,
-		    __connect_timedout, qtask);
+	conn_context = iscsi_conn_context_get(conn, 0);
+	if (!conn_context) {
+		/* while reopening the recv pool should be full */
+		log_error("BUG: __session_conn_reopen could not get conn "
+			  "context for poll.");
+		return ENOMEM;
+	}
+	conn_context->data = qtask;
+	iscsi_sched_conn_context(conn_context, conn, 0, EV_CONN_POLL);
 
+	conn_context = iscsi_conn_context_get(conn, 0);
+	if (!conn_context) {
+		/* while reopening the recv pool should be full */
+		log_error("BUG: __session_conn_reopen could not get conn "
+			  "context for timer.");
+		return ENOMEM;
+	}
+	conn_context->data = qtask;
+	iscsi_sched_conn_context(conn_context, conn, conn->login_timeout,
+				 EV_CONN_LOGIN_TIMER);
 	return 0;
 
 queue_reopen:
@@ -914,8 +854,6 @@ iscsi_login_redirect(iscsi_conn_t *conn)
 
 	log_debug(3, "login redirect ...\n");
 
-	iscsi_queue_flush(session->queue);
-
 	if (session->r_stage == R_STAGE_NO_CHANGE)
 		session->r_stage = R_STAGE_SESSION_REDIRECT;
 
@@ -1193,6 +1131,11 @@ setup_full_feature_phase(iscsi_conn_t *conn)
 		; /* success - fall through */
 	}
 
+	/*
+	 * Remove login timer.
+	 */ 
+	iscsi_flush_context_pool(session);
+
 	/* Entered full-feature phase! */
 	for (i = 0; i < MAX_SESSION_PARAMS; i++) {
 		if (conn->id != 0 && !conntbl[i].conn_only)
@@ -1289,10 +1232,12 @@ failed:
 	return;	
 }
 
-static void iscsi_logout_timeout(void *data)
+static void iscsi_logout_timedout(void *data)
 {
-	iscsi_conn_t *conn = data;
+	struct iscsi_conn_context *conn_context = data;
+	struct iscsi_conn *conn = conn_context->conn;
 
+	iscsi_conn_context_put(conn_context); 
 	/*
 	 * assume we were in STATE_IN_LOGOUT or there
 	 * was some nasty error
@@ -1304,6 +1249,7 @@ static void iscsi_logout_timeout(void *data)
 static int iscsi_send_logout(iscsi_conn_t *conn)
 {
 	struct iscsi_logout hdr;
+	struct iscsi_conn_context *conn_context;
 
 	if (conn->state == STATE_IN_LOGOUT ||
 	    conn->state != STATE_LOGGED_IN)
@@ -1320,9 +1266,18 @@ static int iscsi_send_logout(iscsi_conn_t *conn)
 		return -EIO;
 	conn->state = STATE_IN_LOGOUT;
 
-	actor_timer(&conn->logout_timer, conn->logout_timeout * 1000,
-		    iscsi_logout_timeout, conn);
-	log_debug(3, "logout timeout timer %p start\n", &conn->logout_timer);
+	conn_context = iscsi_conn_context_get(conn, 0);
+	if (!conn_context)
+		/* unbounded logout */
+		log_warning("Could not allocate conn context for logout.");
+	else {
+		iscsi_sched_conn_context(conn_context, conn,
+					 conn->logout_timeout,
+					 EV_CONN_LOGOUT_TIMER);
+		log_debug(3, "logout timeout timer %u\n",
+			  conn->logout_timeout * 1000);
+	}
+
 	return 0;
 }
 
@@ -1396,15 +1351,15 @@ static void iscsi_recv_async_msg(iscsi_conn_t *conn, struct iscsi_hdr *hdr)
 	}
 }
 
-static void
-__session_conn_recv_pdu(queue_item_t *item)
+void session_conn_recv_pdu(void *data)
 {
-	iscsi_conn_t *conn = item->context;
+	struct iscsi_conn_context *conn_context = data;
+	iscsi_conn_t *conn = conn_context->conn;
 	iscsi_session_t *session = conn->session;
 	iscsi_login_context_t *c;
 	struct iscsi_hdr hdr;
 
-	conn->recv_handle = *(uintptr_t*)queue_item_data(item);
+	conn->recv_context = conn_context;
 
 	switch (conn->state) {
 	case STATE_IN_LOGIN:
@@ -1456,17 +1411,17 @@ __session_conn_recv_pdu(queue_item_t *item)
 		}
 		break;
 	case STATE_XPT_WAIT:
-		recvpool_put(conn, (void *)conn->recv_handle);
+		iscsi_conn_context_put(conn_context);
 		log_debug(1, "ignoring incomming PDU in XPT_WAIT. "
 			  "let connection re-establish or fail");
 		break;
 	case STATE_CLEANUP_WAIT:
-		recvpool_put(conn, (void *)conn->recv_handle);
+		iscsi_conn_context_put(conn_context);
 		log_debug(1, "ignoring incomming PDU in XPT_WAIT. "
 			  "let connection cleanup");
 		break;
 	default:
-		recvpool_put(conn, (void *)conn->recv_handle);
+		iscsi_conn_context_put(conn_context);
 		log_error("Invalid state. Dropping PDU.\n");
 	}
 }
@@ -1479,34 +1434,39 @@ setup_kernel_io_callouts(iscsi_conn_t *conn)
 	conn->send_pdu_end = ipc->send_pdu_end;
 	conn->recv_pdu_begin = ipc->recv_pdu_begin;
 	conn->recv_pdu_end = ipc->recv_pdu_end;
-	conn->send_pdu_timer_add = __send_pdu_timer_add;
-	conn->send_pdu_timer_remove = __send_pdu_timer_remove;
 }
 
-static void
-__session_conn_poll(queue_item_t *item)
+static void session_conn_poll(void *data)
 {
+	struct iscsi_conn_context *conn_context = data;
+	iscsi_conn_t *conn = conn_context->conn;
+	struct iscsi_session *session = conn->session;
 	mgmt_ipc_err_e err = MGMT_IPC_OK;
-	queue_task_t *qtask = item->context;
-	iscsi_conn_t *conn = qtask->conn;
+	queue_task_t *qtask = conn_context->data;
 	iscsi_login_context_t *c = &conn->login_context;
-	iscsi_session_t *session = conn->session;
 	int rc;
 
+	iscsi_conn_context_put(conn_context);
+
 	if (conn->state != STATE_XPT_WAIT)
 		return;
 
 	rc = session->t->template->ep_poll(conn, 1);
 	if (rc == 0) {
 		/* timedout: Poll again. */
-		queue_produce(session->queue, EV_CONN_POLL, qtask, 0, NULL);
-		actor_schedule(&session->mainloop);
+		conn_context = iscsi_conn_context_get(conn, 0);
+		if (!conn_context) {
+			/* while polling the recv pool should be full */
+			log_error("BUG: session_conn_poll could not get conn "
+				  "context.");
+			return;
+		}
+		conn_context->data = qtask;
+		iscsi_sched_conn_context(conn_context, conn, 0, EV_CONN_POLL);
 	} else if (rc > 0) {
 		/* connected! */
 		memset(c, 0, sizeof(iscsi_login_context_t));
 
-		actor_delete(&conn->connect_timer);
-
 		/* do not allocate new connection in case of reopen */
 		if (session->r_stage == R_STAGE_NO_CHANGE) {
 			if (conn->id == 0 &&
@@ -1599,12 +1559,20 @@ cleanup:
 	session_conn_cleanup(qtask, err);
 }
 
-static void
-__session_conn_timer(queue_item_t *item)
+static void iscsi_login_timedout(void *data)
 {
-	queue_task_t *qtask = item->context;
-	iscsi_conn_t *conn = qtask->conn;
-	iscsi_session_t *session = conn->session;
+	struct iscsi_conn_context *conn_context = data;
+	queue_task_t *qtask = conn_context->data;
+	iscsi_conn_t *conn = conn_context->conn;
+	struct iscsi_session *session = conn->session;
+
+	iscsi_conn_context_put(conn_context);
+
+	log_debug(3, "session_conn_login_timedout");
+	/*
+	 * Flush polls and other events
+	 */
+	iscsi_flush_context_pool(conn->session);
 
 	switch (conn->state) {
 	case STATE_XPT_WAIT:
@@ -1679,7 +1647,6 @@ __conn_error_handle(iscsi_session_t *session, iscsi_conn_t *conn)
 
 	switch (conn->state) {
 	case STATE_IN_LOGOUT:
-		actor_delete(&conn->logout_timer);
 		/* logout was requested by user */
 		if (conn->logout_qtask) {
 			session_conn_shutdown(conn, conn->logout_qtask,
@@ -1749,60 +1716,56 @@ __conn_error_handle(iscsi_session_t *session, iscsi_conn_t *conn)
 	}
 }
 
-static void
-__session_conn_error(queue_item_t *item)
+static void session_conn_error(void *data)
 {
-	uintptr_t recv_handle = *(uintptr_t *)queue_item_data(item);
-	enum iscsi_err error = *(enum iscsi_err *)recv_handle;
-	iscsi_conn_t *conn = item->context;
+	struct iscsi_conn_context *conn_context = data;
+	enum iscsi_err error = *(enum iscsi_err *)conn_context->data;
+	iscsi_conn_t *conn = conn_context->conn;
 	iscsi_session_t *session = conn->session;
 
 	log_warning("Kernel reported iSCSI connection %d:%d error (%d) "
 		    "state (%d)", session->id, conn->id, error,
 		    conn->state);
+	iscsi_conn_context_put(conn_context);
 	__conn_error_handle(session, conn);
-	recvpool_put(conn, (void*)recv_handle);
 }
 
-static void
-__session_mainloop(void *data)
+void iscsi_sched_conn_context(struct iscsi_conn_context *conn_context,
+			      struct iscsi_conn *conn, unsigned long tmo,
+			      int event)
 {
-	iscsi_session_t *session = data;
-	unsigned char item_buf[sizeof(queue_item_t) + EVENT_PAYLOAD_MAX];
-	queue_item_t *item = (queue_item_t *)(void *)item_buf;
-	int count = session->queue->count, i;
-
-	/*
-	 * grab a reference in case one of these events destroys
-	 * the session
-	 */
-	session_get(session);
-	for (i = 0; i < count; i++) {
-		if (queue_consume(session->queue, EVENT_PAYLOAD_MAX,
-				  item) == QUEUE_IS_EMPTY) {
-			log_debug(4, "%d items flushed while mainloop "
-				  "was processing", count - i);
-			break;
-		}
-
-		switch (item->event_type) {
-		case EV_CONN_RECV_PDU:
-			__session_conn_recv_pdu(item);
-			break;
-		case EV_CONN_POLL:
-			__session_conn_poll(item);
-			break;
-		case EV_CONN_TIMER:
-			__session_conn_timer(item);
-			break;
-		case EV_CONN_ERROR:
-			__session_conn_error(item);
-			break;
-		default:
-			break;
-		}
+	log_debug(7, "sched conn context %p event %d, tmo %lu",
+		  &conn_context->actor, event, tmo);
+
+	conn_context->conn = conn;
+	switch (event) {
+	case EV_CONN_RECV_PDU:
+		actor_new(&conn_context->actor, session_conn_recv_pdu,
+			  conn_context);
+		actor_schedule(&conn_context->actor);
+		break;
+	case EV_CONN_ERROR:
+		actor_new(&conn_context->actor, session_conn_error,
+			  conn_context);
+		actor_schedule(&conn_context->actor);
+		break;
+	case EV_CONN_POLL:
+		actor_new(&conn_context->actor, session_conn_poll,
+			  conn_context);
+		actor_schedule(&conn_context->actor);
+		break;
+	case EV_CONN_LOGIN_TIMER:
+		actor_timer(&conn_context->actor, tmo * 1000,
+			    iscsi_login_timedout, conn_context);
+		break;
+	case EV_CONN_LOGOUT_TIMER:
+		actor_timer(&conn_context->actor, tmo * 1000,
+			    iscsi_logout_timedout, conn_context);
+		break;
+	default:
+		log_error("Invalid event type %d.", event);
+		return;
 	}
-	session_put(session);
 }
 
 iscsi_session_t*
@@ -1862,6 +1825,7 @@ session_login_task(node_rec_t *rec, queue_task_t *qtask)
 	iscsi_session_t *session;
 	iscsi_conn_t *conn;
 	struct iscsi_transport *t;
+	struct iscsi_conn_context *conn_context;
 
 	t = get_transport_by_name(rec->iface.transport_name);
 	if (!t)
@@ -1928,6 +1892,29 @@ session_login_task(node_rec_t *rec, queue_task_t *qtask)
 	conn = &session->conn[0];
 	qtask->conn = conn;
 
+	conn_context = iscsi_conn_context_get(conn, 0);
+	if (!conn_context) {
+		/* while logging the recv pool should be full */
+		log_error("BUG: session_login_task could not get conn "
+			  "context for poll.");
+		__session_destroy(session);
+		return MGMT_IPC_ERR_INTERNAL;
+	}
+	conn_context->data = qtask;
+	iscsi_sched_conn_context(conn_context, conn, 0, EV_CONN_POLL);
+
+	conn_context = iscsi_conn_context_get(conn, 0);
+	if (!conn_context) {
+		/* while logging the recv pool should be full - 1 */
+		log_error("BUG: session_login_task could not get conn "
+			  "context for poll.");
+		__session_destroy(session);
+		return MGMT_IPC_ERR_INTERNAL;
+	}
+	conn_context->data = qtask;
+	iscsi_sched_conn_context(conn_context, conn, conn->login_timeout,
+				 EV_CONN_LOGIN_TIMER);
+
 	rc = session->t->template->ep_connect(conn, 1);
 	if (rc < 0 && errno != EINPROGRESS) {
 		char serv[NI_MAXSERV];
@@ -1939,17 +1926,10 @@ session_login_task(node_rec_t *rec, queue_task_t *qtask)
 
 		log_error("cannot make a connection to %s:%s (%d)",
 			 conn->host, serv, errno);
-		session_stop_conn_timers(session, 0);
 		__session_destroy(session);
 		return MGMT_IPC_ERR_TRANS_FAILURE;
 	}
-
 	conn->state = STATE_XPT_WAIT;
-	queue_produce(session->queue, EV_CONN_POLL, qtask, 0, NULL);
-	actor_schedule(&session->mainloop);
-
-	actor_timer(&conn->connect_timer, conn->login_timeout*1000,
-		    __connect_timedout, qtask);
 
 	qtask->rsp.command = MGMT_IPC_SESSION_LOGIN;
 	qtask->rsp.err = MGMT_IPC_OK;
diff --git a/usr/initiator.h b/usr/initiator.h
index 3b8a224..eacaeb3 100644
--- a/usr/initiator.h
+++ b/usr/initiator.h
@@ -30,7 +30,6 @@
 #include "mgmt_ipc.h"
 #include "config.h"
 #include "actor.h"
-#include "queue.h"
 #include "list.h"
 
 #define ISCSI_CONFIG_ROOT	"/etc/iscsi/"
@@ -66,18 +65,14 @@ enum iscsi_login_status {
 };
 
 typedef enum iscsi_event_e {
-	EV_UNKNOWN			= 0,
-	EV_CONN_RECV_PDU		= 1,
-	EV_CONN_POLL			= 2,
-	EV_CONN_TIMER			= 3,
-	EV_CONN_ERROR			= 4,
+	EV_UNKNOWN,
+	EV_CONN_RECV_PDU,
+	EV_CONN_POLL,
+	EV_CONN_ERROR,
+	EV_CONN_LOGIN_TIMER,
+	EV_CONN_LOGOUT_TIMER,
 } iscsi_event_e;
 
-typedef struct iscsi_event {
-	queue_item_t item;
-	char payload[EVENT_PAYLOAD_MAX];
-} iscsi_event_t;
-
 struct queue_task;
 
 typedef struct iscsi_login_context {
@@ -100,49 +95,37 @@ typedef struct iscsi_login_context {
 
 struct iscsi_session;
 struct iscsi_conn;
+struct iscsi_conn_context;
 
 typedef void (*send_pdu_begin_f) (uint64_t transport_handle, uint32_t sid,
                                   uint32_t cid, int hdr_size, int data_size);
 typedef int (*send_pdu_end_f) (uint64_t transport_handle, uint32_t sid,
 			       uint32_t cid, int *retcode);
-typedef int (*recv_pdu_begin_f) (uint64_t transport_handle,
-				 uintptr_t recv_handle, uintptr_t *pdu_handle,
-				 int *pdu_size);
-typedef int (*recv_pdu_end_f) (uint64_t transport_handle, uintptr_t conn_handle,
-                             uintptr_t pdu_handle);
-typedef void (*send_pdu_timer_add_f)(struct iscsi_conn *conn, int timeout);
-typedef void (*send_pdu_timer_remove_f)(struct iscsi_conn *conn);
+typedef int (*recv_pdu_begin_f) (struct iscsi_conn *conn);
+typedef int (*recv_pdu_end_f) (struct iscsi_conn *conn);
 
 /* daemon's connection structure */
 typedef struct iscsi_conn {
-	struct qelem item; /* must stay at the top */
 	uint32_t id;
-	uintptr_t recv_handle;
 	struct iscsi_session *session;
 	iscsi_login_context_t login_context;
+	struct iscsi_conn_context *recv_context;
 	struct queue_task *logout_qtask;
 	char data[ISCSI_DEF_MAX_RECV_SEG_LEN];
 	char host[NI_MAXHOST];	/* scratch */
 	iscsi_conn_state_e state;
-	actor_t connect_timer;
-	actor_t send_pdu_timer;
-	actor_t logout_timer;
 
 	actor_t noop_out_timer;
 	actor_t noop_out_timeout_timer;
 
-	int send_pdu_in_progress;
-
 	int kernel_io;
 	send_pdu_begin_f send_pdu_begin;
 	send_pdu_end_f send_pdu_end;
 	recv_pdu_begin_f recv_pdu_begin;
 	recv_pdu_end_f recv_pdu_end;
-	send_pdu_timer_add_f send_pdu_timer_add;
-	send_pdu_timer_remove_f send_pdu_timer_remove;
 
-#define RECVPOOL_MAX	32
-	void* recvpool[RECVPOOL_MAX];
+#define CONTEXT_POOL_MAX 32
+	struct iscsi_conn_context *context_pool[CONTEXT_POOL_MAX];
 
 	/* login state machine */
 	int current_stage;
@@ -185,6 +168,13 @@ typedef struct iscsi_conn {
 	uint32_t max_xmit_dlength;	/* the value declared by the target */
 } iscsi_conn_t;
 
+struct iscsi_conn_context {
+	struct actor actor;
+	struct iscsi_conn *conn;
+	int allocated;
+	void *data;
+};
+
 typedef struct queue_task {
 	iscsi_conn_t *conn;
 	iscsiadm_req_t req;
@@ -204,7 +194,6 @@ typedef struct iscsi_session {
 	struct list_head list;
 	uint32_t id;
 	uint32_t hostno;
-	int refcount;
 	char netdev[IFNAMSIZ];
 	struct iscsi_transport *t;
 	node_rec_t nrec; /* copy of original Node record in database */
@@ -261,12 +250,6 @@ typedef struct iscsi_session {
 
 	/* sync up fields */
 	queue_task_t *sync_qtask;
-
-	/* session's processing */
-	actor_t mainloop;
-	queue_t *queue;
-	queue_t *splice_queue;
-
 } iscsi_session_t;
 
 /*
@@ -367,8 +350,12 @@ extern int session_logout_task(iscsi_session_t *session, queue_task_t *qtask);
 extern iscsi_session_t *session_find_by_sid(int sid);
 extern iscsi_session_t *session_find_by_rec(node_rec_t *rec);
 extern int session_is_running(node_rec_t *rec);
-extern void* recvpool_get(iscsi_conn_t *conn, int ev_size);
-extern void recvpool_put(iscsi_conn_t *conn, void *handle);
+extern struct iscsi_conn_context *iscsi_conn_context_get(iscsi_conn_t *conn,
+						   int ev_size);
+extern void iscsi_conn_context_put(struct iscsi_conn_context *conn_context);
+extern void iscsi_sched_conn_context(struct iscsi_conn_context *context,
+				     struct iscsi_conn *conn, unsigned long tmo,
+				     int event);
 extern mgmt_ipc_err_e iscsi_sync_session(node_rec_t *rec, queue_task_t
 					 *tsk, uint32_t sid);
 extern mgmt_ipc_err_e iscsi_host_send_targets(queue_task_t *qtask,
diff --git a/usr/io.c b/usr/io.c
index bb06411..84846cc 100644
--- a/usr/io.c
+++ b/usr/io.c
@@ -632,12 +632,10 @@ iscsi_io_send_pdu(iscsi_conn_t *conn, struct iscsi_hdr *hdr,
 	else
 		pad_bytes = 0;
 
-	if (conn->kernel_io) {
+	if (conn->kernel_io)
 		conn->send_pdu_begin(session->t->handle, session->id,
 			conn->id, end - header,
 			ntoh24(hdr->dlength) + pad_bytes);
-		conn->send_pdu_timer_add(conn, timeout);
-	}
 
 	while (header < end) {
 		vec[0].iov_base = header;
@@ -698,7 +696,6 @@ iscsi_io_send_pdu(iscsi_conn_t *conn, struct iscsi_hdr *hdr,
 	if (conn->kernel_io) {
 		if (conn->send_pdu_end(session->t->handle, session->id,
 				       conn->id, &rc)) {
-			conn->send_pdu_timer_remove(conn);
 			ret = 0;
 			goto done;
 		}
@@ -732,8 +729,6 @@ iscsi_io_recv_pdu(iscsi_conn_t *conn, struct iscsi_hdr *hdr,
 	char *end = data + max_data_length;
 	struct sigaction action;
 	struct sigaction old;
-	uintptr_t pdu_handle;
-	int pdu_size;
 	iscsi_session_t *session = conn->session;
 
 	memset(data, 0, max_data_length);
@@ -751,8 +746,7 @@ iscsi_io_recv_pdu(iscsi_conn_t *conn, struct iscsi_hdr *hdr,
 		timedout = 0;
 		alarm(timeout);
 	} else {
-		if (conn->recv_pdu_begin(session->ctrl_fd,
-				conn->recv_handle, &pdu_handle, &pdu_size)) {
+		if (conn->recv_pdu_begin(conn)) {
 			failed = 1;
 			goto done;
 		}
@@ -905,11 +899,9 @@ done:
 		sigaction(SIGALRM, &old, NULL);
 	} else {
 		/* finalyze receive transaction */
-		if (conn->recv_pdu_end(session->ctrl_fd, (uintptr_t)conn,
-				pdu_handle)) {
+		if (conn->recv_pdu_end(conn)) {
 			failed = 1;
 		}
-		conn->send_pdu_timer_remove(conn);
 	}
 
 	if (timedout || failed) {
diff --git a/usr/iscsi_ipc.h b/usr/iscsi_ipc.h
index 87324db..2172174 100644
--- a/usr/iscsi_ipc.h
+++ b/usr/iscsi_ipc.h
@@ -33,6 +33,8 @@ enum {
 	ISCSI_STRING,
 };
 
+struct iscsi_conn;
+
 /**
  * struct iscsi_ipc - Open-iSCSI Interface for Kernel IPC
  *
@@ -102,11 +104,9 @@ struct iscsi_ipc {
 
 	int (*writev) (enum iscsi_uevent_e type, struct iovec *iovp, int count);
 
-	int (*recv_pdu_begin) (uint64_t transport_handle, uintptr_t recv_handle,
-			       uintptr_t *pdu_handle, int *pdu_size);
+	int (*recv_pdu_begin) (struct iscsi_conn *conn);
 
-	int (*recv_pdu_end) (uint64_t transport_handle, uintptr_t conn_handle,
-			     uintptr_t pdu_handle);
+	int (*recv_pdu_end) (struct iscsi_conn *conn);
 };
 
 #endif /* ISCSI_IPC_H */
diff --git a/usr/iscsi_settings.h b/usr/iscsi_settings.h
index f1d7a1e..4bb7064 100644
--- a/usr/iscsi_settings.h
+++ b/usr/iscsi_settings.h
@@ -3,7 +3,7 @@
  * in the RFC. See iscsi_proto.h for those.
  */
 /* timeouts in seconds */
-#define DEF_LOGIN_TIMEO		15
+#define DEF_LOGIN_TIMEO		30
 #define DEF_LOGOUT_TIMEO	15
 #define DEF_NOOP_OUT_INTERVAL	10
 #define DEF_NOOP_OUT_TIMEO	15
diff --git a/usr/iscsi_sysfs.c b/usr/iscsi_sysfs.c
index c614192..73f9d88 100644
--- a/usr/iscsi_sysfs.c
+++ b/usr/iscsi_sysfs.c
@@ -21,7 +21,6 @@
 #include <ctype.h>
 #include <string.h>
 #include <errno.h>
-#include <search.h>
 #include <dirent.h>
 #include <pwd.h>
 #include <sys/types.h>
diff --git a/usr/isns.c b/usr/isns.c
index b823bf4..6048469 100644
--- a/usr/isns.c
+++ b/usr/isns.c
@@ -50,16 +50,17 @@ struct isns_task {
 	int done;
 	int retry;
 	queue_task_t *qtask;
+	struct actor actor;
 };
 
-static actor_t isns_actor;
-static queue_t *isns_queue = NULL;
 static struct sockaddr_storage ss;
 static uint16_t transaction;
 
 static char isns_address[NI_MAXHOST];
 static int isns_port = 3205, isns_listen_port, max_retry = 10000;
 
+static void isns_poll(void *data);
+
 #define ARRAY_SIZE(x) (sizeof(x) / sizeof((x)[0]))
 
 #define get_hdr_param(hdr, function, length, flags, transaction, sequence)	\
@@ -432,8 +433,8 @@ static int isns_task_done(struct isns_task *task)
 		task->len = length + sizeof(*hdr);
 		task->done = 0;
 
-		queue_produce(isns_queue, EV_CONN_POLL, task, 0, NULL);
-		actor_schedule(&isns_actor);
+		actor_new(&task->actor, isns_poll, task);
+		actor_schedule(&task->actor);
 		finished = 0;
 		break;
 	default:
@@ -502,8 +503,8 @@ int isns_dev_attr_query_task(queue_task_t *qtask)
 
 	qtask->rsp.command = MGMT_IPC_ISNS_DEV_ATTR_QUERY;
 
-	queue_produce(isns_queue, EV_CONN_POLL, task, 0, NULL);
-	actor_schedule(&isns_actor);
+	actor_new(&task->actor, isns_poll, task);
+	actor_schedule(&task->actor);
 
 	return MGMT_IPC_OK;
 }
@@ -532,15 +533,15 @@ void isns_handle(int listen_fd)
 	task->state = ISNS_TASK_RECV_PDU;
 	task->fd = fd;
 
-	queue_produce(isns_queue, EV_CONN_POLL, task, 0, NULL);
-	actor_schedule(&isns_actor);
+	actor_new(&task->actor, isns_poll, task);
+	actor_schedule(&task->actor);
 }
 
-static void isns_poll(queue_item_t *item)
+static void isns_poll(void *data)
 {
 	int err, finished;
 	struct pollfd pfd;
-	struct isns_task *task = item->context;
+	struct isns_task *task = data;
 	struct isns_hdr *hdr = (struct isns_hdr *) task->data;
 	uint16_t function = ntohs(hdr->function);
 
@@ -573,9 +574,8 @@ static void isns_poll(queue_item_t *item)
 						goto free_task;
 				}
 
-				queue_produce(isns_queue, EV_CONN_POLL, task, 0,
-					      NULL);
-				actor_schedule(&isns_actor);
+				actor_new(&task->actor, isns_poll, task);
+				actor_schedule(&task->actor);
 			}
 			break;
 		case ISNS_TASK_RECV_PDU:
@@ -590,9 +590,9 @@ static void isns_poll(queue_item_t *item)
 						goto free_task;
 				} else {
 					/* need to read more */
-					queue_produce(isns_queue, EV_CONN_POLL,
-						      task, 0, NULL);
-					actor_schedule(&isns_actor);
+					actor_new(&task->actor, isns_poll,
+						  task);
+					actor_schedule(&task->actor);
 				}
 			}
 		}
@@ -602,8 +602,8 @@ static void isns_poll(queue_item_t *item)
 			log_error("abort task");
 			goto abort_task;
 		} else {
-			queue_produce(isns_queue, EV_CONN_POLL, task, 0, NULL);
-			actor_schedule(&isns_actor);
+			actor_new(&task->actor, isns_poll, task);
+			actor_schedule(&task->actor);
 		}
 	}
 
@@ -615,32 +615,6 @@ free_task:
 	isns_free_task(task);
 }
 
-static void isns_control(void *data)
-{
-	int count = isns_queue->count, i;
-	int err;
-	unsigned char item_buf[sizeof(queue_item_t) + EVENT_PAYLOAD_MAX];
-	queue_item_t *item = (queue_item_t *)(void *)item_buf;
-
-	for (i = 0; i < count; i++) {
-		err = queue_consume(isns_queue, EVENT_PAYLOAD_MAX, item);
-		if (err == QUEUE_IS_EMPTY) {
-			log_debug(4, "%d items flushed while mainloop "
-				  "was processing", count - i);
-			break;
-		}
-
-		switch (item->event_type) {
-		case EV_CONN_POLL:
-			isns_poll(item);
-			break;
-		default:
-			log_error("%d unknown event type", item->event_type);
-			break;
-		}
-	}
-}
-
 static int isns_dev_register(void)
 {
 	struct isns_task *task;
@@ -658,10 +632,9 @@ static int isns_dev_register(void)
 
 	task->state = ISNS_TASK_WAIT_CONN;
 	build_dev_reg_req(task);
-	queue_produce(isns_queue, EV_CONN_POLL, task, 0, NULL);
 
-	actor_new(&isns_actor, isns_control, NULL);
-	actor_schedule(&isns_actor);
+	actor_new(&task->actor, isns_poll, task);
+	actor_schedule(&task->actor);
 
 	return 0;
 }
@@ -741,62 +714,23 @@ int isns_init(void)
 	if (!strlen(isns_address))
 		return -1;
 
-	isns_queue = queue_create(4, 4, NULL, NULL);
-	if (!isns_queue) {
-		log_error("can't create queue %m");
-		return -ENOMEM;
-	}
-
 	snprintf(port, sizeof(port), "%d", isns_port);
 	err = resolve_address(isns_address, port, &ss);
 	if (err) {
 		log_error("can't resolve address %m, %s", isns_address);
-		goto free_queue;
+		return err;
 	}
 
 	err = isns_listen_init(&fd);
 	if (err)
-		goto free_queue;
+		return err;
 
 	isns_dev_register();
 	return fd;
-
-free_queue:
-	queue_destroy(isns_queue);
-	return err;
 }
 
 void isns_exit(void)
 {
-	int err, count, i;
-	unsigned char item_buf[sizeof(queue_item_t) + EVENT_PAYLOAD_MAX];
-	queue_item_t *item = (queue_item_t *)(void *)item_buf;
-
-	if (!isns_queue)
-		return;
-
-	count = isns_queue->count;
-	/*
-	 * TODO: Add some code to gracefully shutdown.
-	 */
-	for (i = 0; i < count; i++) {
-		err = queue_consume(isns_queue, EVENT_PAYLOAD_MAX, item);
-		if (err == QUEUE_IS_EMPTY) {
-			log_debug(4, "%d items flushed while mainloop "
-				  "was processing", count - i);
-			break;
-		}
-
-		log_debug(4, "Dropping event type %d\n", item->event_type);
-		switch (item->event_type) {
-		case EV_CONN_POLL:
-			isns_free_task(item->context);
-			continue;
-		default:
-			log_error("%d unknown event type", item->event_type);
-			continue;
-		}
-	}
-
-	queue_destroy(isns_queue);
+	/* do nothing for now */
+	;
 }
diff --git a/usr/netlink.c b/usr/netlink.c
index 12cb7f7..65800ab 100644
--- a/usr/netlink.c
+++ b/usr/netlink.c
@@ -668,8 +668,7 @@ kstart_conn(uint64_t transport_handle, uint32_t sid, uint32_t cid,
 }
 
 static int
-krecv_pdu_begin(uint64_t transport_handle, uintptr_t recv_handle,
-		uintptr_t *pdu_handle, int *pdu_size)
+krecv_pdu_begin(struct iscsi_conn *conn)
 {
 	log_debug(7, "in %s", __FUNCTION__);
 
@@ -677,19 +676,17 @@ krecv_pdu_begin(uint64_t transport_handle, uintptr_t recv_handle,
 		log_error("recv's begin state machine bug?");
 		return -EIO;
 	}
-	recvbuf = (void*)recv_handle + sizeof(struct iscsi_uevent);
+	recvbuf = conn->recv_context->data + sizeof(struct iscsi_uevent);
 	recvlen = 0;
-	*pdu_handle = recv_handle;
 
 	log_debug(3, "recv PDU began, pdu handle 0x%p",
-		  (void*)*pdu_handle);
+		  recvbuf);
 
 	return 0;
 }
 
 static int
-krecv_pdu_end(uint64_t transport_handle, uintptr_t conn_handle,
-	      uintptr_t pdu_handle)
+krecv_pdu_end(struct iscsi_conn *conn)
 {
 	log_debug(7, "in %s", __FUNCTION__);
 
@@ -699,9 +696,10 @@ krecv_pdu_end(uint64_t transport_handle, uintptr_t conn_handle,
 	}
 
 	log_debug(3, "recv PDU finished for pdu handle 0x%p",
-		  (void*)pdu_handle);
+		  recvbuf);
 
-	recvpool_put((void*)conn_handle, (void*)pdu_handle);
+	iscsi_conn_context_put(conn->recv_context);
+	conn->recv_context = NULL;
 	recvbuf = NULL;
 	return 0;
 }
@@ -852,9 +850,9 @@ static int ctldev_handle(void)
 	struct iscsi_transport *t;
 	iscsi_session_t *session = NULL;
 	iscsi_conn_t *conn = NULL;
-	uintptr_t recv_handle;
 	char nlm_ev[NLMSG_SPACE(sizeof(struct iscsi_uevent))];
 	struct nlmsghdr *nlh;
+	struct iscsi_conn_context *conn_context;
 	int ev_size;
 
 	log_debug(7, "in %s", __FUNCTION__);
@@ -915,41 +913,41 @@ verify_conn:
 	}
 
 	ev_size = nlh->nlmsg_len - NLMSG_ALIGN(sizeof(struct nlmsghdr));
-	recv_handle = (uintptr_t)recvpool_get(conn, ev_size);
-	if (!recv_handle) {
+	conn_context = iscsi_conn_context_get(conn, ev_size);
+	if (!conn_context) {
 		/* retry later */
-		log_error("Can not allocate memory for receive handle.");
+		log_error("Can not allocate memory for receive context.");
 		return -ENOMEM;
 	}
 
 	log_debug(6, "message real length is %d bytes, recv_handle %p",
-		nlh->nlmsg_len, (void*)recv_handle);
+		nlh->nlmsg_len, conn_context->data);
 
-	if ((rc = nlpayload_read(ctrl_fd, (void*)recv_handle,
+	if ((rc = nlpayload_read(ctrl_fd, conn_context->data,
 				ev_size, 0)) < 0) {
-		recvpool_put(conn, (void*)recv_handle);
+		iscsi_conn_context_put(conn_context);
 		log_error("can not read from NL socket, error %d", rc);
 		/* retry later */
 		return rc;
 	}
 
+	/*
+	 * we sched these events because the handlers could call back
+	 * into ctldev_handle
+	 */
 	switch (ev->type) {
 	case ISCSI_KEVENT_RECV_PDU:
-		/* produce an event, so session manager will handle */
-		queue_produce(session->queue, EV_CONN_RECV_PDU, conn,
-			sizeof(uintptr_t), &recv_handle);
-		actor_schedule(&session->mainloop);
+		iscsi_sched_conn_context(conn_context, conn, 0,
+					 EV_CONN_RECV_PDU);
 		break;
 	case ISCSI_KEVENT_CONN_ERROR:
-		/* produce an event, so session manager will handle */
-		memcpy((void *)recv_handle, &ev->r.connerror.error,
+		memcpy(conn_context->data, &ev->r.connerror.error,
 			sizeof(ev->r.connerror.error));
-		queue_produce(session->queue, EV_CONN_ERROR, conn,
-			sizeof(uintptr_t), &recv_handle);
-		actor_schedule(&session->mainloop);
+		iscsi_sched_conn_context(conn_context, conn, 0,
+					 EV_CONN_ERROR);
 		break;
 	default:
-		recvpool_put(conn, (void*)recv_handle);
+		iscsi_conn_context_put(conn_context);
 		log_error("unknown kernel event %d", ev->type);
 		return -EEXIST;
 	}
diff --git a/usr/queue.c b/usr/queue.c
deleted file mode 100644
index 323bc6f..0000000
--- a/usr/queue.c
+++ /dev/null
@@ -1,251 +0,0 @@
-/*
- * iSCSI event queue
- *
- * Copyright (C) 2004 Dmitry Yusupov, Alex Aizman
- * maintained by open-iscsi@googlegroups.com
- *
- * 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.
- *
- * See the file COPYING included with this distribution for more details.
- */
-
-#include <stdlib.h>
-#include <string.h>
-#include <search.h>
-#include "queue.h"
-#include "log.h"
-#include "actor.h"
-
-queue_t*
-queue_create(int pages_initial, int pages_max, queued_f queued,
-	     void *queued_data)
-{
-	queue_t *queue;
-
-	if ((queue = malloc(sizeof(queue_t))) == NULL) {
-		log_error("out of memory when allocating queue_t");
-		return NULL;
-	}
-
-	queue->queued_func = queued;
-	queue->queued_data = queued_data;
-	queue->pages_current = pages_initial;
-	queue->start_ptr = malloc(queue->pages_current * QUEUE_BUF_SIZE);
-	if (queue->start_ptr == NULL) {
-		log_error("out of memory when allocating queue's pages");
-		free(queue);
-		return NULL;
-	}
-	memset(queue->start_ptr, 0, queue->pages_current * QUEUE_BUF_SIZE);
-	queue->head_ptr = queue->tail_ptr = queue->start_ptr;
-	queue->end_ptr = (char *)queue->start_ptr +
-		queue->pages_current * QUEUE_BUF_SIZE;
-	queue->pages_initial = pages_initial;
-	queue->pages_max = pages_max;
-	queue->list_head.q_forw = &queue->list_head;
-	queue->list_head.q_back = &queue->list_head;
-	queue->count = 0;
-
-	return queue;
-}
-
-void
-queue_destroy(queue_t* queue)
-{
-	if (queue->list_head.q_forw != &queue->list_head) {
-		log_error("destroying non-empty queue 0x%p", queue);
-	}
-	free(queue->start_ptr);
-	free(queue);
-}
-
-static queue_status_e
-__io_queue_grow(queue_t *queue)
-{
-	void *newbuf, *oldbuf;
-	struct qelem *item;
-	queue_item_t *elem;
-
-	log_debug(7, "queue 0x%p:%d is growing", queue, queue->pages_current);
-
-	newbuf = malloc((queue->pages_current + 1) * QUEUE_BUF_SIZE);
-	if (newbuf == NULL) {
-		return QUEUE_OUT_OF_MEMORY;
-	}
-	memcpy(newbuf, queue->start_ptr, queue->pages_current * QUEUE_BUF_SIZE);
-	oldbuf = queue->start_ptr;
-
-	/* adjust queue sizes */
-	queue->start_ptr = newbuf;
-	queue->end_ptr = (char *)newbuf +
-			(queue->pages_current + 1) * QUEUE_BUF_SIZE;
-	queue->tail_ptr = (char *)newbuf + ((char *)queue->tail_ptr -
-					    (char *)oldbuf);
-	queue->head_ptr = (char *)newbuf + ((char *)queue->head_ptr -
-					    (char *)oldbuf);
-	queue->list_head.q_forw = (struct qelem *) (void *)((char *)newbuf +
-			((char *)queue->list_head.q_forw - (char *)oldbuf));
-	queue->list_head.q_back = (struct qelem *) (void *)((char *)newbuf +
-			((char *)queue->list_head.q_back - (char *)oldbuf));
-	/* adjust queue list */
-	for (item = queue->list_head.q_forw;
-	     item != queue->list_head.q_forw; item = item->q_forw) {
-		elem = (queue_item_t *)item;
-		if (elem->item.q_forw != &queue->list_head) {
-			elem->item.q_forw =
-				(struct qelem *)(void *)((char *)newbuf +
-				 ((char *)elem->item.q_forw - (char *)oldbuf));
-		}
-		if (elem->item.q_back != &queue->list_head) {
-			elem->item.q_back =
-				(struct qelem *) (void *)((char *)newbuf +
-				 ((char *)elem->item.q_back - (char *)oldbuf));
-		}
-	}
-	free(oldbuf);
-	queue->pages_current++;
-
-	return QUEUE_OK;
-}
-
-queue_status_e
-queue_consume(queue_t *queue, int data_max_size, queue_item_t *item)
-{
-	int real_size;
-	queue_item_t *elem;
-
-	if (queue->list_head.q_forw == &queue->list_head) {
-		if (queue->count)
-			log_error("queue integrety lost! Bug?");
-		return QUEUE_IS_EMPTY;
-	}
-	elem = (queue_item_t *)queue->list_head.q_forw;
-	if (elem->data_size > data_max_size) {
-		return QUEUE_NOT_ENOUGH_SPACE;
-	}
-	remque(&elem->item);
-	real_size = elem->data_size + sizeof(queue_item_t);
-	if (queue->head_ptr == elem) {
-		queue->head_ptr = (char *)queue->head_ptr + real_size;
-		log_debug(7,
-			"event_type: %d removing from the head: "
-			"0x%p:0x%p:0x%p:0x%p elem 0x%p length %d",
-			elem->event_type,
-			queue->start_ptr,
-			queue->head_ptr,
-			queue->tail_ptr,
-			queue->end_ptr,
-			elem,
-			real_size);
-	} else if ((char *)queue->tail_ptr - real_size == (char*)elem) {
-		queue->tail_ptr = (char *)queue->tail_ptr - real_size;
-		log_debug(7,
-			"event_type: %d removing from the tail: "
-			"0x%p:0x%p:0x%p:0x%p elem 0x%p length %d",
-			elem->event_type,
-			queue->start_ptr,
-			queue->head_ptr,
-			queue->tail_ptr,
-			queue->end_ptr,
-			elem,
-			real_size);
-	} else {
-		log_debug(7,
-			"event_type: %d removing from the list: "
-			"0x%p:0x%p:0x%p:0x%p elem 0x%p length %d",
-			elem->event_type,
-			queue->start_ptr,
-			queue->head_ptr,
-			queue->tail_ptr,
-			queue->end_ptr,
-			elem,
-			real_size);
-	}
-	memcpy(item, elem, sizeof(queue_item_t));
-	memcpy(queue_item_data(item), queue_item_data(elem), elem->data_size);
-
-	if (queue->list_head.q_forw == &queue->list_head) {
-		/* reset buffer pointers just to be clean */
-		queue->head_ptr = queue->tail_ptr = queue->start_ptr;
-	}
-
-	queue->count--;
-
-	return QUEUE_OK;
-}
-
-void*
-queue_item_data (queue_item_t *item)
-{
-	return (char *)item + sizeof(queue_item_t);
-}
-
-queue_status_e
-queue_produce(queue_t *queue, int event_type, void *context,
-	      const int data_size, void *data)
-{
-	int real_size = data_size + sizeof(queue_item_t);
-	queue_item_t *elem;
-
-try_again:
-	if ((char *)queue->tail_ptr + real_size <= (char *)queue->end_ptr) {
-		elem = queue->tail_ptr;
-		queue->tail_ptr = (void *)((char *)queue->tail_ptr + real_size);
-		log_debug(7, "event_type: %d adding to the tail: "
-			"0x%p:0x%p:0x%p:0x%p elem 0x%p length %d",
-			event_type,
-			queue->start_ptr,
-			queue->head_ptr,
-			queue->tail_ptr,
-			queue->end_ptr,
-			elem,
-			real_size);
-	} else if ((char *)queue->head_ptr - real_size >=
-					(char *)queue->start_ptr) {
-		elem = (void *)((char *)queue->head_ptr - real_size);
-		queue->head_ptr = elem;
-		log_debug(7, "event_type: %d adding to the head: "
-			"0x%p:0x%p:0x%p:0x%p length %d",
-			event_type,
-			queue->start_ptr,
-			queue->head_ptr,
-			queue->tail_ptr,
-			queue->end_ptr,
-			real_size);
-	} else {
-		queue_status_e status;
-
-		if (queue->pages_current >= queue->pages_max) {
-			return QUEUE_IS_FULL;
-		}
-
-		/* grow */
-		status = __io_queue_grow(queue);
-		if (status != QUEUE_OK) {
-			return status;
-		}
-
-		goto try_again;
-	}
-	elem->data_size = data_size;
-	elem->event_type = event_type;
-	elem->context = context;
-	memcpy(queue_item_data(elem), data, data_size);
-	insque(&elem->item, queue->list_head.q_back);
-
-	if (queue->queued_func)
-		queue->queued_func(queue->queued_data, event_type);
-
-	queue->count++;
-
-	return QUEUE_OK;
-}
-
diff --git a/usr/queue.h b/usr/queue.h
deleted file mode 100644
index 6abe613..0000000
--- a/usr/queue.h
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * iSCSI event queue
- *
- * Copyright (C) 2004 Dmitry Yusupov, Alex Aizman
- * maintained by open-iscsi@googlegroups.com
- *
- * 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.
- *
- * See the file COPYING included with this distribution for more details.
- */
-
-#ifndef QUEUE_H
-#define QUEUE_H
-
-#include "types.h"
-
-#define QUEUE_BUF_SIZE			4096
-#define EVENT_PAYLOAD_MAX		(DATASEG_MAX+HDRSEG_MAX)
-
-typedef enum queue_status_e {
-	QUEUE_OK		= 0,
-	QUEUE_IS_FULL		= 1,
-	QUEUE_IS_EMPTY		= 2,
-	QUEUE_OUT_OF_MEMORY	= 3,
-	QUEUE_NOT_ENOUGH_SPACE	= 4
-} queue_status_e;
-
-typedef struct queue_item_t {
-	struct qelem	item;
-	int		event_type;
-	int		data_size;
-	void		*context;
-} queue_item_t;
-
-typedef void (*queued_f) (void *data, int event_type);
-
-typedef struct queue_t {
-	void				*start_ptr;
-	void				*end_ptr;
-	void				*head_ptr;
-	void				*tail_ptr;
-	unsigned int			pages_initial;
-	unsigned int			pages_max;
-	unsigned int			pages_current;
-	struct qelem			list_head;
-	queued_f			queued_func;
-	void				*queued_data;
-	int				count;
-} queue_t;
-
-extern queue_t* queue_create(int pages_initial, int pages_max,
-				queued_f queued_func, void *queued_data);
-extern void queue_destroy(queue_t *queue);
-extern void* queue_item_data(queue_item_t *item);
-extern queue_status_e queue_produce(queue_t* queue, int event_type,
-	    void *context, const int data_size, void *data);
-extern queue_status_e queue_consume(queue_t *queue, int	data_max_size,
-				    queue_item_t *item);
-
-#endif /* QUEUE_H */