Making the communication bi-directional with polling

8 files changed, 534 insertions, 1023 deletions

diff --git a/src/http.c b/src/http.c
index a2c897b..bdfe46b 100644
--- a/src/http.c
+++ b/src/http.c
@@ -27,551 +27,28 @@
 
 #define BUFFER_STEP (1 << 12)
 
-struct http_message_t *http_message_new()
+struct http_packet_t *packet_new()
 {
-  struct http_message_t *msg = calloc(1, sizeof(*msg));
-  if (msg == NULL) {
-    ERR("failed to alloc space for http message");
-    return NULL;
-  }
-
-  msg->spare_capacity = 0;
-  msg->spare_filled = 0;
-  msg->spare_buffer = NULL;
-
-  return msg;
-}
-
-void message_free(struct http_message_t *msg)
-{
-  free(msg->spare_buffer);
-  free(msg);
-}
-
-static void packet_check_completion(struct http_packet_t *pkt)
-{
-  struct http_message_t *msg = pkt->parent_message;
-  /* Msg full */
-  if (msg->claimed_size && msg->received_size >= msg->claimed_size) {
-    msg->is_completed = 1;
-    NOTE("http: Message completed: Received size >= claimed size");
-
-    /* Sanity check */
-    if (msg->spare_filled > 0)
-      ERR_AND_EXIT("Msg spare not empty upon completion");
-  }
-
-  /* Pkt full */
-  if (pkt->expected_size && pkt->filled_size >= pkt->expected_size) {
-    pkt->is_completed = 1;
-    NOTE("http: Packet completed: Packet full");
-  }
-
-  /* Pkt over capacity */
-  if (pkt->filled_size > pkt->buffer_capacity) {
-    /* Santiy check */
-    ERR_AND_EXIT("Overflowed packet buffer");
-  }
-}
-
-static int doesMatch(const char *matcher, size_t matcher_len,
-                     const uint8_t *key, size_t key_len)
-{
-  for (size_t i = 0; i < matcher_len; i++)
-    if (i >= key_len || matcher[i] != key[i])
-      return 0;
-  return 1;
-}
-
-static int inspect_header_field(struct http_packet_t *pkt, size_t header_size,
-                                char *key, size_t key_size)
-{
-  /* Find key */
-  uint8_t *pos = memmem(pkt->buffer, header_size, key, key_size);
-  if (pos == NULL)
-    return -1;
-
-  /* Find first digit */
-  size_t number_pos = (size_t) (pos - pkt->buffer) + key_size;
-  while (number_pos < pkt->filled_size && !isdigit(pkt->buffer[number_pos]))
-    ++number_pos;
-
-  /* Find next non-digit */
-  size_t number_end = number_pos;
-  while (number_end < pkt->filled_size && isdigit(pkt->buffer[number_end]))
-    ++number_end;
-
-  /* Failed to find next non-digit
-     header field might be broken */
-  if (number_end >= pkt->filled_size)
-    return -1;
-
-  /* Temporary stringification of buffer for atoi() */
-  uint8_t original_char = pkt->buffer[number_end];
-  pkt->buffer[number_end] = '\0';
-  int val = atoi((const char *)(pkt->buffer + number_pos));
-
-  /* Restore buffer */
-  pkt->buffer[number_end] = original_char;
-  return val;
-}
-
-static void packet_store_excess(struct http_packet_t *pkt)
-{
-  struct http_message_t *msg = pkt->parent_message;
-  if (msg->spare_buffer != NULL)
-    ERR_AND_EXIT("Do not store excess to non-empty packet");
-
-  if (pkt->expected_size >= pkt->filled_size)
-    ERR_AND_EXIT("Do not call packet_store_excess() unless needed");
-
-  size_t spare_size = pkt->filled_size - pkt->expected_size;
-  size_t non_spare = pkt->expected_size;
-  NOTE("HTTP: Storing %d bytes of excess", spare_size);
-
-  /* Align to BUFFER_STEP */
-  size_t needed_size = 0;
-  needed_size += spare_size / BUFFER_STEP;
-  needed_size += (spare_size % BUFFER_STEP) > 0 ? BUFFER_STEP : 0;
-
-  if (msg->spare_buffer == NULL) {
-    uint8_t *buffer = calloc(1, needed_size);
-    if (buffer == NULL)
-      ERR_AND_EXIT("Failed to alloc msg spare buffer");
-    msg->spare_buffer = buffer;
-  }
-
-  memcpy(msg->spare_buffer, pkt->buffer + non_spare, spare_size);
-  pkt->filled_size = non_spare;
-
-  msg->spare_capacity = needed_size;
-  msg->spare_filled = spare_size;
-}
-
-static void packet_take_spare(struct http_packet_t *pkt)
-{
-  struct http_message_t *msg = pkt->parent_message;
-  if (msg->spare_filled == 0)
-    return;
-
-  if (msg->spare_buffer == NULL)
-    return;
-
-  if (pkt->filled_size > 0)
-    ERR_AND_EXIT("pkt should be empty when loading msg spare");
-
-  /* Take message's buffer */
-  size_t msg_size = msg->spare_capacity;
-  size_t msg_filled = msg->spare_filled;
-  uint8_t *msg_buffer = msg->spare_buffer;
-
-  pkt->buffer_capacity = msg_size;
-  pkt->filled_size = msg_filled;
-  pkt->buffer = msg_buffer;
-
-  msg->spare_capacity = 0;
-  msg->spare_filled = 0;
-  msg->spare_buffer = NULL;
-}
-
-static ssize_t packet_find_chunked_size(struct http_packet_t *pkt)
-{
-  /* NOTE:
-     chunks can have trailers which are
-     tacked on http header fields.
-     NOTE:
-     chunks may also have extensions.
-     No one uses or supports them. */
-
-  /* Find end of size string */
-  if (pkt->filled_size >= SSIZE_MAX)
-    ERR_AND_EXIT("Buffer beyond sane size");
-
-  ssize_t max = (ssize_t) pkt->filled_size;
-  ssize_t size_end = -1;
-  ssize_t miniheader_end = -1;
-  ssize_t delimiter_start = -1;
-  for (ssize_t i = 0; i < max; i++) {
-
-    uint8_t *buf = pkt->buffer;
-    if (size_end < 0) {
-      /* No extension */
-      if (i + 1 < max &&
-	  (buf[i] == '\r' &&      /* CR */
-	   buf[i + 1] == '\n')) { /* LF */
-	size_end = i + 1;
-	miniheader_end = size_end;
-	delimiter_start = i;
-	break;
-      }
-
-      /* No extension */
-      if (buf[i] == '\n') { /* LF */
-	size_end = i;
-	miniheader_end = size_end;
-	delimiter_start = i;
-	break;
-      }
-
-      /* Has extensions */
-      if (buf[i] == ';') {
-	size_end = i;
-	continue;
-      }
-    }
-    if (miniheader_end < 0) {
-      if (i + 1 < max &&
-	  (buf[i] == '\r' &&      /* CR */
-	   buf[i + 1] == '\n')) { /* LF */
-	miniheader_end = i + 1;
-	delimiter_start = i;
-	break;
-      }
-
-      if (buf[i] == '\n') { /* LF */
-	miniheader_end = i;
-	delimiter_start = i;
-	break;
-      }
-    }
-  }
-
-  if (miniheader_end < 0) {
-    /* NOTE: knowing just the size field
-       is not enough since the extensions
-       are not included in the size */
-    NOTE("failed to find chunk mini-header so far");
-    return -1;
-  }
-
-  /* Temporary stringification for strtol() */
-  uint8_t original_char = *(pkt->buffer + size_end);
-  *(pkt->buffer + size_end) = '\0';
-  size_t size = strtoul((char *)pkt->buffer, NULL, 16);
-  NOTE("Chunk size raw: %s", pkt->buffer);
-  *(pkt->buffer + size_end) = original_char;
-  if (size > SSIZE_MAX)
-    ERR_AND_EXIT("chunk size is insane");
-
-  if (size > 0) {
-    /* Regular chunk */
-    ssize_t chunk_size = (ssize_t) size; /* Chunk body */
-    chunk_size += miniheader_end + 1; /* Mini-header */
-    chunk_size += 2; /* Trailing CRLF */
-    NOTE("HTTP: Chunk size: %lu", chunk_size);
-    return (ssize_t) chunk_size;
-  }
-
-  /* Terminator chunk
-     May have trailers in body */
-  ssize_t full_size = -1;
-  for (ssize_t i = delimiter_start; i < max; i++) {
-    uint8_t *buf = pkt->buffer;
-    if (i + 3 < max &&
-	(buf[i] == '\r' &&      /* CR */
-	 buf[i + 1] == '\n' &&  /* LF */
-	 buf[i + 2] == '\r' &&  /* CR */
-	 buf[i + 3] == '\n')) { /* LF */
-      full_size = i + 4;
-      break;
-    }
-
-    if (i + 1 < max &&
-	buf[i] == '\n' &&     /* LF */
-	buf[i + 1] == '\n') { /* LF */
-      full_size = i + 2;
-      break;
-    }
-  }
-
-  if (full_size < 0) {
-    NOTE("Chunk miniheader present but body incomplete");
-    return -1;
-  }
-
-  NOTE("Found end chunked packet");
-  pkt->parent_message->is_completed = 1;
-  pkt->is_completed = 1;
-  return full_size;
-}
-
-static ssize_t packet_get_header_size(struct http_packet_t *pkt)
-{
-  if (pkt->header_size != 0)
-    goto found;
-
- /*
-  * RFC2616 recomends we match newline on \n despite full
-  * complience requires the message to use only \r\n
-  * http://www.w3.org/Protocols/rfc2616/rfc2616-sec19.html#sec19.3
-  */
-
-  /* Find header */
-  for (size_t i = 0; i < pkt->filled_size && i < SSIZE_MAX; i++) {
-    /* two \r\n pairs */
-    if ((i + 3) < pkt->filled_size &&
-	'\r' == pkt->buffer[i] &&
-	'\n' == pkt->buffer[i + 1] &&
-	'\r' == pkt->buffer[i + 2] &&
-	'\n' == pkt->buffer[i + 3]) {
-      pkt->header_size = i + 4;
-      goto found;
-    }
-
-    /* two \n pairs */
-    if ((i + 1) < pkt->filled_size &&
-	'\n' == pkt->buffer[i] &&
-	'\n' == pkt->buffer[i + 1]) {
-      pkt->header_size = i + 2;
-      goto found;
-    }
-  }
-
-  return -1;
-
- found:
-  return (ssize_t) pkt->header_size;
-}
-
-enum http_request_t packet_find_type(struct http_packet_t *pkt)
-{
-  enum http_request_t type = HTTP_UNSET;
-  size_t size = 0;
- /*
-  * Valid methods for determining http request
-  * size are defined by W3 in RFC2616 section 4.4
-  * link: http://www.w3.org/Protocols/rfc2616/rfc2616-sec4.html#sec4.4
-  */
-
- /*
-  * This function attempts to find what method this
-  * packet would use. This is only possible in specific case:
-  * 1. if the request uses method 1 we can check the http 
-  *    request type. We must be called on a packet which
-  *    has the full header.
-  * 2. if the request uses method 2 we need the full header
-  *    but a simple network-byte-order-aware string search
-  *    works. This function does not work if called with
-  *    a chunked transport's sub-packet.
-  * 3. if the request uses method 3 we again perform the
-  *    string search.
-  *
-  * All cases require the packat to contain the full header.
-  */
-
-  ssize_t header_size_raw = packet_get_header_size(pkt);
-  if (header_size_raw < 0) {
-    /* We don't have the header yet */
-    goto do_ret;
-  }
-  size_t header_size = (size_t) header_size_raw;
-
-  /* Try Transfer-Encoding Chunked */
-  char xfer_encode_str[] = "Transfer-Encoding: chunked";
-  size_t xfer_encode_str_size = sizeof(xfer_encode_str) - 1;
-  uint8_t *xfer_encode_pos = memmem(pkt->buffer, header_size,
-				    xfer_encode_str,
-				    xfer_encode_str_size);
-  if (xfer_encode_pos != NULL) {
-    size = 0;
-    type = HTTP_CHUNKED;
-    goto do_ret;
-  }
-
-  /* Try Content-Length */
-  char content_length_str[] = "Content-Length: ";
-  ssize_t contlen_size =
-    inspect_header_field(pkt, header_size,
-			 content_length_str, sizeof(content_length_str) - 1);
-  if (contlen_size >= 0) {
-    size = (size_t) contlen_size + header_size;
-    type = HTTP_CONTENT_LENGTH;
-    goto do_ret;
-  }
-
-  /* GET requests (start with GET) or answers from the server (start
-     with HTTP) */
-  if (doesMatch("GET", 3, pkt->buffer, pkt->filled_size) ||
-      doesMatch("HTTP", 4, pkt->buffer, pkt->filled_size)) {
-    size = header_size;
-    type = HTTP_HEADER_ONLY;
-    goto do_ret;
-  }
-
-  /* No size was detectable yet header was found */
-  type = HTTP_UNKNOWN;
-  size = 0;
-
- do_ret:
-  pkt->parent_message->claimed_size = size;
-  pkt->parent_message->type = type;
-  return type;
-}
-
-size_t packet_pending_bytes(struct http_packet_t *pkt)
-{
-  struct http_message_t *msg = pkt->parent_message;
-
-  /* Check Cache */
-  if (pkt->expected_size > 0)
-    goto pending_known;
-
-  if (HTTP_UNSET == msg->type) {
-    msg->type = packet_find_type(pkt);
-
-    if (HTTP_CHUNKED == msg->type) {
-      /* Note: this was the packet with the
-	 header of our chunked message. */
-
-      /* Save any non-header data we got */
-      ssize_t header_size = packet_get_header_size(pkt);
-
-      /* Sanity check */
-      if (header_size < 0 ||
-	  (size_t)header_size > pkt->filled_size)
-	ERR_AND_EXIT("HTTP: Could not find header twice");
-
-      NOTE("HTTP: Chunked header size is %ld bytes",
-	   header_size);
-      pkt->expected_size = (size_t) header_size;
-      msg->claimed_size = 0;
-      goto pending_known;
-    }
-  }
-
-  if (HTTP_CHUNKED == msg->type) {
-    if (pkt->filled_size == 0) {
-      /* Grab chunk's mini-header */
-      goto pending_known;
-    }
-
-    if (pkt->expected_size == 0) {
-      /* Check chunk's mini-header */
-      ssize_t size = packet_find_chunked_size(pkt);
-      if (size <= 0) {
-	ERR("=============================================");
-	ERR("Malformed chunk-transport http header receivd");
-	ERR("Missing chunk's mini-headers in first data");
-	ERR("Have %d bytes", pkt->filled_size);
-	printf("%.*s\n", (int)pkt->filled_size, pkt->buffer);
-	ERR("Malformed chunk-transport http header receivd");
-	ERR("=============================================");
-	goto pending_known;
-      }
-
-      pkt->expected_size = (size_t) size;
-      msg->claimed_size = 0;
-    }
-
-    goto pending_known;
-  }
-  if (HTTP_HEADER_ONLY == msg->type) {
-    /* Note: we can only know it is header only
-       when the buffer already contains the header.
-       So this next call cannot fail. */
-    pkt->expected_size = (size_t) packet_get_header_size(pkt);
-    msg->claimed_size = pkt->expected_size;
-    goto pending_known;
-  }
-  if (HTTP_CONTENT_LENGTH == msg->type) {
-    /* Note: find_header() has
-       filled msg's claimed_size */
-    msg->claimed_size = msg->claimed_size;
-    pkt->expected_size = msg->claimed_size;
-    goto pending_known;
-  }
-
- pending_known:
-
-  /* Save excess data */
-  if (pkt->expected_size && pkt->filled_size > pkt->expected_size)
-    packet_store_excess(pkt);
-
-  size_t expected = pkt->expected_size;
-  if (expected == 0)
-    expected = msg->claimed_size;
-  if (expected == 0)
-    expected = pkt->buffer_capacity;
-
-  /* Sanity check */
-  if (expected < pkt->filled_size)
-    ERR_AND_EXIT("Expected cannot be larger than filled");
-
-  size_t pending = expected - pkt->filled_size;
-
-  /* Expand buffer as needed */
-  while (pending + pkt->filled_size > pkt->buffer_capacity) {
-    ssize_t size_added = packet_expand(pkt);
-    if (size_added < 0) {
-      WARN("packet at max allowed size");
-      return 0;
-    }
-    if (size_added == 0) {
-      ERR("Failed to expand packet");
-      return 0;
-    }
-  }
-
-  packet_check_completion(pkt);
-
-  return pending;
-}
-
-void packet_mark_received(struct http_packet_t *pkt, size_t received)
-{
-  struct http_message_t *msg = pkt->parent_message;
-  msg->received_size += received;
-
-  pkt->filled_size += received;
-  if (received) {
-    NOTE("HTTP: got %lu bytes so: pkt has %lu bytes, "
-	 "msg has %lu bytes",
-	 received, pkt->filled_size, msg->received_size);
-  }
-
-  packet_check_completion(pkt);
-
-  if (pkt->filled_size > pkt->buffer_capacity)
-    ERR_AND_EXIT("Overflowed packet's buffer");
-
-  if (pkt->expected_size && pkt->filled_size > pkt->expected_size) {
-    /* Store excess data */
-    packet_store_excess(pkt);
-  }
-}
-
-struct http_packet_t *packet_new(struct http_message_t *parent_msg)
-{
-  struct http_packet_t *pkt = NULL;
-  uint8_t              *buf = NULL;
   size_t const capacity = BUFFER_STEP;
 
-  assert(parent_msg != NULL);
-  pkt = calloc(1, sizeof(*pkt));
+  struct http_packet_t *pkt = calloc(1, sizeof(*pkt));
   if (pkt == NULL) {
     ERR("failed to alloc packet");
     return NULL;
   }
-  pkt->parent_message = parent_msg;
-  pkt->expected_size = 0;
-
-  /* Claim any spare data from prior packets */
-  packet_take_spare(pkt);
 
-  if (pkt->buffer == NULL) {
-    buf = calloc(capacity, sizeof(*buf));
-    if (buf == NULL) {
-      ERR("failed to alloc space for packet's buffer or space for packet");
-      free(pkt);
-      return NULL;
-    }
-
-    /* Assemble packet */
-    pkt->buffer = buf;
-    pkt->buffer_capacity = capacity;
-    pkt->filled_size = 0;
+  uint8_t *buf = calloc(capacity, sizeof(*buf));
+  if (buf == NULL) {
+    ERR("failed to alloc space for packet's buffer or space for packet");
+    free(pkt);
+    return NULL;
   }
 
+  /* Assemble packet */
+  pkt->buffer = buf;
+  pkt->buffer_capacity = capacity;
+  pkt->filled_size = 0;
+
   return pkt;
 }
 
@@ -580,29 +57,3 @@ void packet_free(struct http_packet_t *pkt)
   free(pkt->buffer);
   free(pkt);
 }
-
-#define MAX_PACKET_SIZE (1 << 26) /* 64MiB */
-ssize_t packet_expand(struct http_packet_t *pkt)
-{
-  size_t cur_size = pkt->buffer_capacity;
-  size_t new_size = cur_size * 2;
-  if (new_size > MAX_PACKET_SIZE) {
-    WARN("HTTP: cannot expand packet beyond limit");
-    return -1;
-  }
-  NOTE("HTTP: doubling packet buffer to %lu", new_size);
-
-  uint8_t *new_buf = realloc(pkt->buffer, new_size);
-  if (new_buf == NULL) {
-    /* If realloc fails the original buffer is still valid */
-    WARN("Failed to expand packet");
-    return 0;
-  }
-  pkt->buffer = new_buf;
-  pkt->buffer_capacity = new_size;
-
-  size_t diff = new_size - cur_size;
-  if (diff > SSIZE_MAX)
-    ERR_AND_EXIT("Buffer expanded beyond sane limit");
-  return (ssize_t) diff;
-}
diff --git a/src/http.h b/src/http.h
index 44744ec..664fae4 100644
--- a/src/http.h
+++ b/src/http.h
@@ -16,51 +16,11 @@
 #include <stdint.h>
 #include <sys/types.h>
 
-enum http_request_t {
-  HTTP_UNSET,
-  HTTP_UNKNOWN,
-  HTTP_CHUNKED,
-  HTTP_CONTENT_LENGTH,
-  HTTP_HEADER_ONLY
-};
-
-struct http_message_t {
-  enum http_request_t type;
-
-  size_t spare_filled;
-  size_t spare_capacity;
-  uint8_t *spare_buffer;
-
-  size_t unreceived_size;
-  uint8_t is_completed;
-
-  /* Detected from child packets */
-  size_t claimed_size;
-  size_t received_size;
-};
-
 struct http_packet_t {
-  /* Cache */
-  size_t header_size;
-
   size_t filled_size;
-  size_t expected_size;
-
   size_t buffer_capacity;
   uint8_t *buffer;
-
-  struct http_message_t *parent_message;
-
-  uint8_t is_completed;
 };
 
-struct http_message_t *http_message_new(void);
-void message_free(struct http_message_t *);
-
-enum http_request_t packet_find_type(struct http_packet_t *pkt);
-size_t packet_pending_bytes(struct http_packet_t *);
-void packet_mark_received(struct http_packet_t *, size_t);
-
-struct http_packet_t *packet_new(struct http_message_t *);
-void packet_free(struct http_packet_t *);
-ssize_t packet_expand(struct http_packet_t *);
+struct http_packet_t *packet_new();
+void packet_free(struct http_packet_t *pkt);
diff --git a/src/ippusbxd.c b/src/ippusbxd.c
index ead0670..d7d00a9 100644
--- a/src/ippusbxd.c
+++ b/src/ippusbxd.c
@@ -22,6 +22,9 @@
 #include <unistd.h>
 #include <getopt.h>
 #include <pthread.h>
+#include <errno.h>
+
+#include <libusb.h>
 
 #include "options.h"
 #include "logging.h"
@@ -33,10 +36,49 @@
 struct service_thread_param {
   struct tcp_conn_t *tcp;
   struct usb_sock_t *usb_sock;
+  struct usb_conn_t *usb_conn;
   pthread_t thread_handle;
-  int thread_num;
+  uint32_t thread_num;
+  pthread_cond_t *cond;
 };
 
+struct libusb_callback_data {
+  int *read_inflight;
+  uint32_t thread_num;
+  struct tcp_conn_t *tcp;
+  struct http_packet_t *pkt;
+  pthread_mutex_t *read_inflight_mutex;
+  pthread_cond_t *read_inflight_cond;
+};
+
+/* Function prototypes */
+static void *service_connection(void *params_void);
+
+static void service_socket_connection(struct service_thread_param *params);
+
+static void *service_printer_connection(void *params_void);
+
+static int allocate_socket_connection(struct service_thread_param *param);
+
+static int setup_socket_connection(struct service_thread_param *param);
+
+static int setup_usb_connection(struct usb_sock_t *usb_sock,
+                                struct service_thread_param *param);
+
+static int setup_communication_thread(void *(*routine)(void *),
+                                      struct service_thread_param *param);
+
+static int get_read_inflight(const int *read_inflight,
+                             pthread_mutex_t *read_inflight_mutex);
+
+static struct libusb_callback_data *setup_libusb_callback_data(
+    struct http_packet_t *pkt, int *read_inflight,
+    struct service_thread_param *thread_param,
+    pthread_mutex_t *read_inflight_mutex);
+
+static int is_socket_open(const struct service_thread_param *param);
+
+/* Global variables */
 static pthread_mutex_t thread_register_mutex;
 static struct service_thread_param **service_threads = NULL;
 static int num_service_threads = 0;
@@ -61,7 +103,7 @@ static void list_service_threads(int num_service_threads,
     snprintf(p, sizeof(buf) - strlen(buf), "None");
   } else {
     for (i = 0; i < num_service_threads; i ++) {
-      snprintf(p, sizeof(buf) - strlen(buf), "#%d, ",
+      snprintf(p, sizeof(buf) - strlen(buf), "#%u, ",
 	       service_threads[i]->thread_num);
       p = buf + strlen(buf);
     }
@@ -76,12 +118,12 @@ static int register_service_thread(int *num_service_threads,
 				   struct service_thread_param ***service_threads,
 				   struct service_thread_param *new_thread)
 {
-  NOTE("Registering thread #%d", new_thread->thread_num);
+  NOTE("Registering thread #%u", new_thread->thread_num);
   (*num_service_threads) ++;
   *service_threads = realloc(*service_threads,
 			     *num_service_threads * sizeof(void*));
   if (*service_threads == NULL) {
-    ERR("Registering thread #%d: Failed to alloc space for thread registration list",
+    ERR("Registering thread #%u: Failed to alloc space for thread registration list",
 	new_thread->thread_num);
     return -1;
   }
@@ -89,18 +131,18 @@ static int register_service_thread(int *num_service_threads,
   return 0;
 }
 
-static int unregister_service_thread(int *num_service_threads,
-				     struct service_thread_param ***service_threads,
-				     int thread_num)
+static int unregister_service_thread(
+    int *num_service_threads, struct service_thread_param ***service_threads,
+    uint32_t thread_num)
 {
   int i;
 
-  NOTE("Unregistering thread #%d", thread_num);
+  NOTE("Unregistering thread #%u", thread_num);
   for (i = 0; i < *num_service_threads; i ++)
     if ((*service_threads)[i]->thread_num == thread_num)
       break;
   if (i >= *num_service_threads) {
-    ERR("Unregistering thread #%d: Cannot unregister, not found", thread_num);
+    ERR("Unregistering thread #%u: Cannot unregister, not found", thread_num);
     return -1;
   }
   (*num_service_threads) --;
@@ -111,7 +153,7 @@ static int unregister_service_thread(int *num_service_threads,
   if (*num_service_threads == 0)
     *service_threads = NULL;
   else if (*service_threads == NULL) {
-    ERR("Unregistering thread #%d: Failed to alloc space for thread registration list",
+    ERR("Unregistering thread #%u: Failed to alloc space for thread registration list",
 	thread_num);
     return -1;
   }
@@ -121,221 +163,444 @@ static int unregister_service_thread(int *num_service_threads,
 static void
 cleanup_handler(void *arg_void)
 {
-  int thread_num = *((int *)(arg_void));
-  NOTE("Thread #%d: Called clean-up handler", thread_num);
+  uint32_t thread_num = *((int *)(arg_void));
+  NOTE("Thread #%u: Called clean-up handler", thread_num);
   pthread_mutex_lock(&thread_register_mutex);
   unregister_service_thread(&num_service_threads, &service_threads, thread_num);
   list_service_threads(num_service_threads, service_threads);
   pthread_mutex_unlock(&thread_register_mutex);
 }
 
-static void *service_connection(void *arg_void)
+static void read_transfer_callback(struct libusb_transfer *transfer)
+{
+  struct libusb_callback_data *user_data =
+      (struct libusb_callback_data *)transfer->user_data;
+
+  uint32_t thread_num = user_data->thread_num;
+  pthread_mutex_t *read_inflight_mutex = user_data->read_inflight_mutex;
+  pthread_cond_t *read_inflight_cond = user_data->read_inflight_cond;
+
+  switch (transfer->status) {
+    case LIBUSB_TRANSFER_COMPLETED:
+      NOTE("Thread #%u: Transfer has completed successfully", thread_num);
+      user_data->pkt->filled_size = transfer->actual_length;
+
+      NOTE("Thread #%u: Pkt from %s (buffer size: %zu)\n===\n%s===", thread_num,
+           "usb", user_data->pkt->filled_size,
+           hexdump(user_data->pkt->buffer, (int)user_data->pkt->filled_size));
+
+      tcp_packet_send(user_data->tcp, user_data->pkt);
+
+      break;
+    case LIBUSB_TRANSFER_ERROR:
+      ERR("Thread #%u: There was an error completing the transfer", thread_num);
+      g_options.terminate = 1;
+      break;
+    case LIBUSB_TRANSFER_TIMED_OUT:
+      ERR("Thread #%u: The transfer timed out before it could be completed: "
+          "Received %u bytes",
+          thread_num, transfer->actual_length);
+      break;
+    case LIBUSB_TRANSFER_CANCELLED:
+      NOTE("Thread #%u: The transfer was cancelled", thread_num);
+      break;
+    case LIBUSB_TRANSFER_STALL:
+      ERR("Thread #%u: The transfer has stalled", thread_num);
+      g_options.terminate = 1;
+      break;
+    case LIBUSB_TRANSFER_NO_DEVICE:
+      ERR("Thread #%u: The printer was disconnected during the transfer",
+          thread_num);
+      g_options.terminate = 1;
+      break;
+    case LIBUSB_TRANSFER_OVERFLOW:
+      ERR("Thread #%u: The printer sent more data than was requested",
+          thread_num);
+      g_options.terminate = 1;
+      break;
+    default:
+      ERR("Thread #%u: Something unexpected happened", thread_num);
+      g_options.terminate = 1;
+  }
+
+  /* Mark the transfer as completed. */
+  pthread_mutex_lock(read_inflight_mutex);
+  *user_data->read_inflight = 0;
+  pthread_cond_broadcast(read_inflight_cond);
+  pthread_mutex_unlock(read_inflight_mutex);
+
+  /* Cleanup the data used for the transfer */
+  packet_free(user_data->pkt);
+  free(user_data);
+  libusb_free_transfer(transfer);
+}
+
+/* This function is responsible for handling connection requests and
+   is run in a separate thread. It detaches itself from the main thread and sets
+   up a USB connection with the printer. This function spawns a partner thread
+   which is responsible for reading from the printer, and then this function
+   calls into service_socket_connection() which is responsible for reading from
+   the socket which made the connection request. Once the socket has closed its
+   end of communiction, this function notifies its partner thread that the
+   connection has been closed and then joins on the partner thread before
+   shutting down. */
+static void *service_connection(void *params_void)
 {
-  struct service_thread_param *arg =
-    (struct service_thread_param *)arg_void;
-  int thread_num = arg->thread_num;
+  struct service_thread_param *params =
+      (struct service_thread_param *)params_void;
+  uint32_t thread_num = params->thread_num;
 
-  NOTE("Thread #%d: Starting", thread_num);
+  NOTE("Thread #%u: Setting up both ends for communication", thread_num);
 
-  /* Detach this thread so that the main thread does not need to join this thread
-     after termination for clean-up */
+  /* Detach this thread so that the main thread does not need to join this
+     thread after termination for clean-up. */
   pthread_detach(pthread_self());
 
-  /* Register clean-up handler */
+  /* Register clean-up handler. */
   pthread_cleanup_push(cleanup_handler, &thread_num);
 
-  /* Allow immediate cancelling of this thread */
+  /* Allow immediate cancelling of this thread. */
   pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
   pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
 
-  /* classify priority */
-  struct usb_conn_t *usb = NULL;
-  int usb_failed = 0;
-  while (!arg->tcp->is_closed && usb_failed == 0 && !g_options.terminate) {
-    struct http_message_t *server_msg = NULL;
-    struct http_message_t *client_msg = NULL;
-
-    /* Client's request */
-    client_msg = http_message_new();
-    if (client_msg == NULL) {
-      ERR("Thread #%d: Failed to create client message", thread_num);
+  /* Attempt to establish a connection with the printer. */
+  if (setup_usb_connection(params->usb_sock, params))
+    goto cleanup;
+
+  /* Condition variable used to broadcast updates to the printer thread. */
+  pthread_cond_t cond;
+  if (pthread_cond_init(&cond, NULL))
+    goto cleanup;
+  params->cond = &cond;
+
+  /* Copy the contents of |params| into |printer_params|. The only
+     differences between the two are the |thread_num| and |thread_handle|. */
+  struct service_thread_param *printer_params =
+      calloc(1, sizeof(*printer_params));
+  memcpy(printer_params, params, sizeof(*printer_params));
+  printer_params->thread_num += 1;
+
+  /* Attempt to start the printer's end of the communication. */
+  NOTE("Thread #%u: Attempting to register thread %u", thread_num,
+       thread_num + 1);
+  if (setup_communication_thread(&service_printer_connection, printer_params))
+    goto cleanup;
+
+  /* This function will run until the socket has been closed. When this function
+     returns it means that the communication has been completed. */
+  service_socket_connection(params);
+
+  /* Notify the printer's end that the socket has closed so that it does not
+     have to wait for any pending asynchronous transfers to complete. */
+  pthread_cond_broadcast(params->cond);
+  
+  /* Wait for the printer thread to exit. */
+  NOTE("Thread #%u: Waiting for thread #%u to complete", thread_num,
+       printer_params->thread_num);
+  if (pthread_join(printer_params->thread_handle, NULL))
+    ERR("Thread #%u: Something went wrong trying to join the printer thread",
+        thread_num);
+
+cleanup:
+  if (params->usb_conn != NULL) {
+    NOTE("Thread #%u: interface #%u: releasing usb conn", thread_num,
+         params->usb_conn->interface_index);
+    usb_conn_release(params->usb_conn);
+    params->usb_conn = NULL;
+  }
+
+  NOTE("Thread #%u: closing, %s", thread_num,
+       g_options.terminate ? "shutdown requested"
+                           : "communication thread terminated");
+  tcp_conn_close(params->tcp);
+  free(params);
+
+  /* Execute clean-up handler. */
+  pthread_cleanup_pop(1);
+  pthread_exit(NULL);
+}
+
+/* Reads from the socket and writes data to the printer. */
+static void service_socket_connection(struct service_thread_param *params)
+{
+  uint32_t thread_num = params->thread_num;
+
+  NOTE("Thread #%u: Starting on socket end", thread_num);
+
+  struct http_packet_t *pkt = NULL;
+
+  while (is_socket_open(params) && !g_options.terminate) {
+    pkt = tcp_packet_get(params->tcp);
+
+    if (pkt == NULL) {
+      if (!is_socket_open(params))
+        NOTE("Thread: #%u: Client closed connection", thread_num);
+      else
+        NOTE("Thread: #%u: There was an error reading from the socket",
+             thread_num);
+      return;
+    }
+
+    NOTE("Thread #%u: Pkt from tcp (buffer size: %zu)\n===\n%s===", thread_num,
+         pkt->filled_size, hexdump(pkt->buffer, (int)pkt->filled_size));
+
+    /* Send pkt to printer. */
+    usb_conn_packet_send(params->usb_conn, pkt);
+    packet_free(pkt);
+  }
+}
+
+/* Returns the value of |read_inflight|. Uses a mutex since another thread which
+   is processing the asynchronous transfer may change the value once the
+   transfer is complete. */
+static int get_read_inflight(const int *read_inflight, pthread_mutex_t *mtx)
+{
+  pthread_mutex_lock(mtx);
+  int val = *read_inflight;
+  pthread_mutex_unlock(mtx);
+
+  return val;
+}
+
+/* Sets the value of |read_inflight| to |val|. Uses a mutex since another thread
+   which is processing the asynchronous transfer may change the value once the
+   transfer is complete. */
+static void set_read_inflight(int val, pthread_mutex_t *mtx, int *read_inflight)
+{
+  pthread_mutex_lock(mtx);
+  *read_inflight = val;
+  pthread_mutex_unlock(mtx);
+}
+
+/* Reads from the printer and writes to the socket. */
+static void *service_printer_connection(void *params_void)
+{
+  struct service_thread_param *params =
+      (struct service_thread_param *)params_void;
+  uint32_t thread_num = params->thread_num;
+
+  NOTE("Thread #%u: Starting on printer end", thread_num);
+
+  /* Register clean-up handler. */
+  pthread_cleanup_push(cleanup_handler, &thread_num);
+
+  int read_inflight = 0;
+  pthread_mutex_t read_inflight_mutex;
+  if (pthread_mutex_init(&read_inflight_mutex, NULL))
+    goto cleanup;
+
+  struct libusb_transfer *read_transfer = NULL;
+
+  while (is_socket_open(params) && !g_options.terminate) {
+    /* If there is already a read from the printer underway, block until it has
+       completed. */
+    pthread_mutex_lock(&read_inflight_mutex);
+    while (is_socket_open(params) && read_inflight)
+      pthread_cond_wait(params->cond, &read_inflight_mutex);
+    pthread_mutex_unlock(&read_inflight_mutex);
+
+    /* After waking up due to a completed transfer, verify that the socket is
+       still open and that the termination flag has not been set before
+       attempting to start another transfer. */
+    if (!is_socket_open(params) || g_options.terminate)
+      break;
+
+    NOTE("Thread #%u: No read in flight, starting a new one", thread_num);
+    struct http_packet_t *pkt = packet_new();
+    if (pkt == NULL) {
+      ERR("Thread #%u: Failed to allocate packet", thread_num);
       break;
     }
-    NOTE("Thread #%d: M %p: Client msg starting",
-	 thread_num, client_msg);
-
-    while (!client_msg->is_completed && !g_options.terminate) {
-      struct http_packet_t *pkt;
-      pkt = tcp_packet_get(arg->tcp, client_msg);
-      if (pkt == NULL) {
-	if (arg->tcp->is_closed) {
-	  NOTE("Thread #%d: M %p: Client closed connection",
-	       thread_num, client_msg);
-	  goto cleanup_subconn;
-	}
-	ERR("Thread #%d: M %p: Got null packet from tcp",
-	    thread_num, client_msg);
-	goto cleanup_subconn;
-      }
-      if (usb == NULL && arg->usb_sock != NULL) {
-	usb = usb_conn_acquire(arg->usb_sock);
-	if (usb == NULL) {
-	  ERR("Thread #%d: M %p: Failed to acquire usb interface",
-	      thread_num, client_msg);
-	  packet_free(pkt);
-	  usb_failed = 1;
-	  goto cleanup_subconn;
-	}
-	usb_failed = 0;
-	NOTE("Thread #%d: M %p: Interface #%d: acquired usb conn",
-	     thread_num, client_msg,
-	     usb->interface_index);
-      }
 
-      if (g_options.terminate)
-	goto cleanup_subconn;
-
-      NOTE("Thread #%d: M %p P %p: Pkt from tcp (buffer size: %d)\n===\n%s===",
-	   thread_num, client_msg, pkt,
-	   pkt->filled_size,
-	   hexdump(pkt->buffer, (int)pkt->filled_size));
-      /* In no-printer mode we simply ignore passing the
-	 client message on to the printer */
-      if (arg->usb_sock != NULL) {
-	if (usb_conn_packet_send(usb, pkt) != 0) {
-	  ERR("Thread #%d: M %p P %p: Interface #%d: Unable to send client package via USB",
-	      thread_num,
-	      client_msg, pkt, usb->interface_index);
-	  packet_free(pkt);
-	  goto cleanup_subconn;
-	}
-	NOTE("Thread #%d: M %p P %p: Interface #%d: Client pkt done",
-	     thread_num,
-	     client_msg, pkt, usb->interface_index);
-      }
-      packet_free(pkt);
+    struct libusb_callback_data *user_data = setup_libusb_callback_data(
+        pkt, &read_inflight, params, &read_inflight_mutex);
+
+    if (user_data == NULL) {
+      ERR("Thread #%u: Failed to allocate memory for libusb_callback_data",
+          thread_num);
+      break;
     }
-    if (usb != NULL)
-      NOTE("Thread #%d: M %p: Interface #%d: Client msg completed",
-	   thread_num, client_msg,
-	   usb->interface_index);
-    else
-      NOTE("Thread #%d: M %p: Client msg completed",
-	   thread_num, client_msg);
-    message_free(client_msg);
-    client_msg = NULL;
-
-    if (g_options.terminate)
-      goto cleanup_subconn;
-
-
-    /* Server's response */
-    server_msg = http_message_new();
-    if (server_msg == NULL) {
-      ERR("Thread #%d: Failed to create server message",
-	  thread_num);
-      goto cleanup_subconn;
+
+    read_transfer = setup_async_read(
+        params->usb_conn, pkt, read_transfer_callback, (void *)user_data, 2000);
+
+    if (read_transfer == NULL) {
+      ERR("Thread #%u: Failed to allocate memory for libusb transfer",
+          thread_num);
+      break;
     }
-    if (usb != NULL)
-      NOTE("Thread #%d: M %p: Interface #%d: Server msg starting",
-	   thread_num, server_msg,
-	   usb->interface_index);
-    else
-      NOTE("Thread #%d: M %p: Server msg starting",
-	   thread_num, server_msg);
-    while (!server_msg->is_completed && !g_options.terminate) {
-      struct http_packet_t *pkt;
-      if (arg->usb_sock != NULL) {
-	pkt = usb_conn_packet_get(usb, server_msg);
-	if (pkt == NULL) {
-	  usb_failed = 1;
-	  goto cleanup_subconn;
-	}
-      } else {
-	/* In no-printer mode we "invent" the answer
-	   of the printer, a simple HTML message as
-	   a pseudo web interface */
-	pkt = packet_new(server_msg);
-	snprintf((char*)(pkt->buffer),
-		 pkt->buffer_capacity - 1,
-		 "HTTP/1.1 200 OK\r\nContent-Type: text/html; name=ippusbxd.html; charset=UTF-8\r\n\r\n<html><h2>ippusbxd</h2><p>Debug/development mode without connection to IPP-over-USB printer</p></html>\r\n");
-	pkt->filled_size = 183;
-	/* End the TCP connection, so that a
-	   web browser does not wait for more data */
-	server_msg->is_completed = 1;
-	arg->tcp->is_closed = 1;
-      }
 
-      if (g_options.terminate)
-	goto cleanup_subconn;
-
-      NOTE("Thread #%d: M %p P %p: Pkt from usb (buffer size: %d)\n===\n%s===",
-	   thread_num, server_msg, pkt, pkt->filled_size,
-	   hexdump(pkt->buffer, (int)pkt->filled_size));
-      if (tcp_packet_send(arg->tcp, pkt) != 0) {
-	ERR("Thread #%d: M %p P %p: Unable to send client package via TCP",
-	    thread_num,
-	    client_msg, pkt);
-	packet_free(pkt);
-	goto cleanup_subconn;
-      }
-      if (usb != NULL)
-	NOTE("Thread #%d: M %p P %p: Interface #%d: Server pkt done",
-	     thread_num, server_msg, pkt,
-	     usb->interface_index);
-      else
-	NOTE("Thread #%d: M %p P %p: Server pkt done",
-	     thread_num, server_msg, pkt);
-      packet_free(pkt);
+    /* Mark that there is a new read in flight. A mutex should not be needed
+       here since the transfer callback won't be fired until after calling
+       libusb_submit_transfer() */
+    read_inflight = 1;
+
+    if (libusb_submit_transfer(read_transfer)) {
+      ERR("Thread #%u: Failed to submit asynchronous USB transfer", thread_num);
+      set_read_inflight(0, &read_inflight_mutex, &read_inflight);
+      break;
     }
-    if (usb != NULL)
-      NOTE("Thread #%d: M %p: Interface #%d: Server msg completed",
-	   thread_num, server_msg,
-	   usb->interface_index);
-    else
-      NOTE("Thread #%d: M %p: Server msg completed",
-	   thread_num, server_msg);
-
-  cleanup_subconn:
-    if (usb != NULL && (arg->tcp->is_closed || usb_failed == 1)) {
-      NOTE("Thread #%d: M %p: Interface #%d: releasing usb conn",
-	   thread_num, server_msg, usb->interface_index);
-      usb_conn_release(usb);
-      usb = NULL;
+  }
+
+  /* If the socket used for communication has closed and there is still a
+     transfer from the printer in flight then we attempt to cancel it. */
+  if (get_read_inflight(&read_inflight, &read_inflight_mutex)) {
+    NOTE(
+        "Thread #%u: There was a read in flight when the connection was "
+        "closed, cancelling transfer", thread_num);
+    int cancel_status = libusb_cancel_transfer(read_transfer);
+    if (!cancel_status) {
+      /* Wait until the cancellation has completed. */
+      NOTE("Thread #%u: Waiting until the transfer has been cancelled",
+           thread_num);
+      pthread_mutex_lock(&read_inflight_mutex);
+      while (read_inflight)
+        pthread_cond_wait(params->cond, &read_inflight_mutex);
+      pthread_mutex_unlock(&read_inflight_mutex);
+    } else if (cancel_status == LIBUSB_ERROR_NOT_FOUND) {
+      NOTE("Thread #%u: The transfer has already completed", thread_num);
+    } else {
+      NOTE("Thread #%u: Failed to cancel transfer");
+      g_options.terminate = 1;
     }
-    if (client_msg != NULL)
-      message_free(client_msg);
-    if (server_msg != NULL)
-      message_free(server_msg);
   }
 
-  NOTE("Thread #%d: Closing, %s", thread_num,
-       g_options.terminate ? "shutdown requested" : "communication thread terminated");
-  tcp_conn_close(arg->tcp);
-  free(arg);
+  pthread_mutex_destroy(&read_inflight_mutex);
 
-  /* Execute clean-up handler */
+cleanup:
+  /* Execute clean-up handler. */
   pthread_cleanup_pop(1);
-
   pthread_exit(NULL);
 }
 
+static uint16_t open_tcp_socket(void)
+{
+  uint16_t desired_port = g_options.desired_port;
+  g_options.tcp_socket = NULL;
+  g_options.tcp6_socket = NULL;
+
+  for (;;) {
+    g_options.tcp_socket = tcp_open(desired_port, g_options.interface);
+    g_options.tcp6_socket = tcp6_open(desired_port, g_options.interface);
+    if (g_options.tcp_socket || g_options.tcp6_socket ||
+        g_options.only_desired_port)
+      break;
+    /* Search for a free port. */
+    desired_port ++;
+    /* We failed with 0 as port number or we reached the max port number. */
+    if (desired_port == 1 || desired_port == 0)
+      /* IANA recommendation of 49152 to 65535 for ephemeral ports. */
+      desired_port = 49152;
+    NOTE("Access to desired port failed, trying alternative port %d",
+         desired_port);
+  }
+
+  return desired_port;
+}
+
+/* Attempts to allocate space for a tcp socket. If the allocation is
+   successful then a value of 0 is returned, otherwise a non-zero value is
+   returned. */
+static int allocate_socket_connection(struct service_thread_param *param)
+{
+  param->tcp = calloc(1, sizeof(*param->tcp));
+
+  if (param->tcp == NULL) {
+    ERR("Preparing thread #%u: Failed to allocate space for cups connection",
+        param->thread_num);
+    return -1;
+  }
+
+  return 0;
+}
+
+/* Attempts to setup a connection for to a tcp socket. Returns a 0 value on
+   success and a non-zero value if something went wrong attempting to establish
+   the connection. */
+static int setup_socket_connection(struct service_thread_param *param)
+{
+  param->tcp = tcp_conn_select(g_options.tcp_socket, g_options.tcp6_socket);
+  if (g_options.terminate || param->tcp == NULL)
+    return -1;
+  return 0;
+}
+
+/* Attempt to create a new usb_conn_t and assign it to |param| by acquiring an
+   available usb interface. Returns 0 if the creating on the connection struct
+   was successful, and non-zero if there was an error attempting to acquire the
+   interface. */
+static int setup_usb_connection(struct usb_sock_t *usb_sock,
+                                struct service_thread_param *param)
+{
+  param->usb_conn = usb_conn_acquire(usb_sock);
+  if (param->usb_conn == NULL) {
+    ERR("Thread #%u: Failed to acquire usb interface", param->thread_num);
+    return -1;
+  }
+
+  return 0;
+}
+
+/* Attempt to register a new communication thread to execute the function
+   |routine| with the given |params|. If successful a 0 value is returned,
+   otherwise a non-zero value is returned. */
+static int setup_communication_thread(void *(*routine)(void *),
+                                      struct service_thread_param *param)
+{
+  pthread_mutex_lock(&thread_register_mutex);
+  register_service_thread(&num_service_threads, &service_threads, param);
+  list_service_threads(num_service_threads, service_threads);
+  pthread_mutex_unlock(&thread_register_mutex);
+
+  int status =
+      pthread_create(&param->thread_handle, NULL, routine, param);
+
+  if (status) {
+    ERR("Creating thread #%u: Failed to spawn thread, error %d",
+        param->thread_num, status);
+    pthread_mutex_lock(&thread_register_mutex);
+    unregister_service_thread(&num_service_threads, &service_threads,
+                              param->thread_num);
+    list_service_threads(num_service_threads, service_threads);
+    pthread_mutex_unlock(&thread_register_mutex);
+    return -1;
+  }
+
+  return 0;
+}
+
+static struct libusb_callback_data *setup_libusb_callback_data(
+    struct http_packet_t *pkt, int *read_inflight,
+    struct service_thread_param *thread_param,
+    pthread_mutex_t *read_inflight_mutex) {
+  struct libusb_callback_data *data = calloc(1, sizeof(*data));
+  if (data == NULL)
+    return NULL;
+
+  data->pkt = pkt;
+  data->read_inflight = read_inflight;
+  data->thread_num = thread_param->thread_num;
+  data->read_inflight_mutex = read_inflight_mutex;
+  data->read_inflight_cond = thread_param->cond;
+  data->tcp = thread_param->tcp;
+
+  return data;
+}
+
+static int is_socket_open(const struct service_thread_param *param) {
+  return !param->tcp->is_closed;
+}
+
 static void start_daemon()
 {
-  /* Capture USB device if not in no-printer mode */
+  /* Capture USB device. */
   struct usb_sock_t *usb_sock;
 
   /* Termination flag */
   g_options.terminate = 0;
 
-  if (g_options.noprinter_mode == 0) {
-    usb_sock = usb_open();
-    if (usb_sock == NULL)
-      goto cleanup_usb;
-  } else {
-    usb_sock = NULL;
-    g_options.device_id = "MFG:Acme;MDL:LaserStar 2000;CMD:AppleRaster,PWGRaster;CLS:PRINTER;DES:Acme LaserStar 2000;SN:001;";
-  }
+  usb_sock = usb_open();
+  if (usb_sock == NULL) goto cleanup_usb;
 
   /* Capture a socket */
   uint16_t desired_port = g_options.desired_port;
@@ -419,45 +684,32 @@ static void start_daemon()
   }
 
   /* Main loop */
-  int i = 0;
+  uint32_t i = 1;
   pthread_mutex_init(&thread_register_mutex, NULL);
   while (!g_options.terminate) {
-    i ++;
     struct service_thread_param *args = calloc(1, sizeof(*args));
     if (args == NULL) {
-      ERR("Preparing thread #%d: Failed to alloc space for thread args",
-	  i);
+      ERR("Preparing thread #%u: Failed to alloc space for thread args", i);
       goto cleanup_thread;
     }
 
     args->thread_num = i;
     args->usb_sock = usb_sock;
 
-    /* For each request/response round we use the socket (IPv4 or
-       IPv6) which receives data first */
-    args->tcp = tcp_conn_select(g_options.tcp_socket, g_options.tcp6_socket);
-    if (g_options.terminate)
+    /* Allocate space for a tcp socket to be used for communication. */
+    if (allocate_socket_connection(args))
       goto cleanup_thread;
-    if (args->tcp == NULL) {
-      ERR("Preparing thread #%d: Failed to open tcp connection", i);
+
+    /* Attempt to establish a connection to the relevant socket. */
+    if (setup_socket_connection(args))
       goto cleanup_thread;
-    }
 
-    pthread_mutex_lock(&thread_register_mutex);
-    register_service_thread(&num_service_threads, &service_threads, args);
-    list_service_threads(num_service_threads, service_threads);
-    pthread_mutex_unlock(&thread_register_mutex);
-    int status = pthread_create(&args->thread_handle, NULL,
-				&service_connection, args);
-    if (status) {
-      ERR("Creating thread #%d: Failed to spawn thread, error %d",
-	  i, status);
-      pthread_mutex_lock(&thread_register_mutex);
-      unregister_service_thread(&num_service_threads, &service_threads, i);
-      list_service_threads(num_service_threads, service_threads);
-      pthread_mutex_unlock(&thread_register_mutex);
+    /* Attempt to start up a new thread to handle the socket's end of
+       communication. */
+    if (setup_communication_thread(&service_connection, args))
       goto cleanup_thread;
-    }
+
+    i += 2;
 
     continue;
 
@@ -479,7 +731,7 @@ static void start_daemon()
      stopping ippusbxd, so that no USB communication with the printer can
      happen after the final reset */
   while (num_service_threads) {
-    NOTE("Thread #%d did not terminate, canceling it now ...",
+    NOTE("Thread #%u did not terminate, canceling it now ...",
 	 service_threads[0]->thread_num);
     i = num_service_threads;
     pthread_cancel(service_threads[0]->thread_handle);
@@ -488,6 +740,7 @@ static void start_daemon()
   }
 
   /* Wait for USB unplug event observer thread to terminate */
+  NOTE("Shutting down usb observer thread");
   pthread_join(g_options.usb_event_thread_handle, NULL);
 
   /* TCP clean-up */
@@ -538,7 +791,6 @@ int main(int argc, char *argv[])
     {"verbose",      no_argument,       0,  'q' },
     {"no-fork",      no_argument,       0,  'n' },
     {"no-broadcast", no_argument,       0,  'B' },
-    {"no-printer",   no_argument,       0,  'N' },
     {"help",         no_argument,       0,  'h' },
     {NULL,           0,                 0,  0   }
   };
@@ -625,9 +877,6 @@ int main(int argc, char *argv[])
     case 's':
       g_options.serial_num = (unsigned char *)optarg;
       break;
-    case 'N':
-      g_options.noprinter_mode = 1;
-      break;
     case 'B':
       g_options.nobroadcast = 1;
       break;
@@ -673,13 +922,11 @@ int main(int argc, char *argv[])
 	   "  -n           No-fork mode\n"
 	   "  --no-broadcast\n"
 	   "  -B           No-broadcast mode, do not DNS-SD-broadcast\n"
-	   "  --no-printer\n"
-	   "  -N           No-printer mode, debug/developer mode which makes ippusbxd\n"
-	   "               run without IPP-over-USB printer\n"
 	   , argv[0], argv[0], argv[0]);
     return 0;
   }
 
   start_daemon();
+  NOTE("ippusbxd completed successfully");
   return 0;
 }
diff --git a/src/options.h b/src/options.h
index 2de31a2..e167514 100644
--- a/src/options.h
+++ b/src/options.h
@@ -35,7 +35,6 @@ struct options {
   int help_mode;
   int verbose_mode;
   int nofork_mode;
-  int noprinter_mode;
   int nobroadcast;
 
   /* Printer identity */
diff --git a/src/tcp.c b/src/tcp.c
index 5e11884..2e08516 100644
--- a/src/tcp.c
+++ b/src/tcp.c
@@ -30,11 +30,11 @@
 #include <unistd.h>
 #include <errno.h>
 
-#include "options.h"
+#include "http.h"
 #include "logging.h"
+#include "options.h"
 #include "tcp.h"
 
-
 struct tcp_sock_t *tcp_open(uint16_t port, char* interface)
 {
   struct tcp_sock_t *this = calloc(1, sizeof *this);
@@ -221,56 +221,37 @@ uint16_t tcp_port_number_get(struct tcp_sock_t *sock)
   return 0;
 }
 
-struct http_packet_t *tcp_packet_get(struct tcp_conn_t *tcp,
-                                     struct http_message_t *msg)
+struct http_packet_t *tcp_packet_get(struct tcp_conn_t *tcp)
 {
   /* Alloc packet ==---------------------------------------------------== */
-  struct http_packet_t *pkt = packet_new(msg);
+  struct http_packet_t *pkt = packet_new();
   if (pkt == NULL) {
     ERR("failed to create packet for incoming tcp message");
     goto error;
   }
 
-  size_t want_size = packet_pending_bytes(pkt);
-  if (want_size == 0) {
-    NOTE("TCP: Got %lu from spare buffer", pkt->filled_size);
-    return pkt;
-  }
-
   struct timeval tv;
   tv.tv_sec = 3;
   tv.tv_usec = 0;
-  setsockopt(tcp->sd, SOL_SOCKET, SO_RCVTIMEO,
-	     (char *)&tv, sizeof(struct timeval));
-
-  while (want_size != 0 && !msg->is_completed && !g_options.terminate) {
-    NOTE("TCP: Getting %d bytes", want_size);
-    uint8_t *subbuffer = pkt->buffer + pkt->filled_size;
-    ssize_t gotten_size = recv(tcp->sd, subbuffer, want_size, 0);
-    if (gotten_size < 0) {
-      int errno_saved = errno;
-      ERR("recv failed with err %d:%s", errno_saved,
-	  strerror(errno_saved));
-      tcp->is_closed = 1;
-      goto error;
-    }
-    NOTE("TCP: Got %d bytes", gotten_size);
-    if (gotten_size == 0) {
-      tcp->is_closed = 1;
-      if (pkt->filled_size == 0) {
-	/* Client closed TCP conn */
-	goto error;
-      } else {
-	break;
-      }
-    }
+  if (setsockopt(tcp->sd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv))) {
+    ERR("TCP: Setting options for tcp connection socket failed");
+    goto error;
+  }
+
+  ssize_t gotten_size = recv(tcp->sd, pkt->buffer, pkt->buffer_capacity, 0);
+
+  if (gotten_size < 0) {
+    int errno_saved = errno;
+    ERR("recv failed with err %d:%s", errno_saved, strerror(errno_saved));
+    tcp->is_closed = 1;
+    goto error;
+  }
 
-    packet_mark_received(pkt, (unsigned) gotten_size);
-    want_size = packet_pending_bytes(pkt);
-    NOTE("TCP: Want more %d bytes; Message %scompleted", want_size, msg->is_completed ? "" : "not ");
+  if (gotten_size == 0) {
+    tcp->is_closed = 1;
   }
 
-  NOTE("TCP: Received %lu bytes", pkt->filled_size);
+  pkt->filled_size = gotten_size;
   return pkt;
 
  error:
@@ -283,9 +264,10 @@ int tcp_packet_send(struct tcp_conn_t *conn, struct http_packet_t *pkt)
 {
   size_t remaining = pkt->filled_size;
   size_t total = 0;
+
   while (remaining > 0 && !g_options.terminate) {
-    ssize_t sent = send(conn->sd, pkt->buffer + total,
-			remaining, MSG_NOSIGNAL);
+    ssize_t sent = send(conn->sd, pkt->buffer + total, remaining, MSG_NOSIGNAL);
+
     if (sent < 0) {
       if (errno == EPIPE) {
 	conn->is_closed = 1;
@@ -295,13 +277,13 @@ int tcp_packet_send(struct tcp_conn_t *conn, struct http_packet_t *pkt)
       return -1;
     }
 
-    size_t sent_ulong = (unsigned) sent;
-    total += sent_ulong;
-    if (sent_ulong >= remaining)
+    total += sent;
+    if (sent >= remaining)
       remaining = 0;
     else
-      remaining -= sent_ulong;
+      remaining -= sent;
   }
+
   NOTE("TCP: sent %lu bytes", total);
   return 0;
 }
diff --git a/src/tcp.h b/src/tcp.h
index e58cb17..5ae6b4f 100644
--- a/src/tcp.h
+++ b/src/tcp.h
@@ -48,6 +48,5 @@ struct tcp_conn_t *tcp_conn_select(struct tcp_sock_t *sock,
 				   struct tcp_sock_t *sock6);
 void tcp_conn_close(struct tcp_conn_t *);
 
-struct http_packet_t *tcp_packet_get(struct tcp_conn_t *,
-                                     struct http_message_t *);
+struct http_packet_t *tcp_packet_get(struct tcp_conn_t *);
 int tcp_packet_send(struct tcp_conn_t *, struct http_packet_t *);
diff --git a/src/usb.c b/src/usb.c
index 39a9a75..cadc006 100644
--- a/src/usb.c
+++ b/src/usb.c
@@ -173,10 +173,8 @@ static void try_detach_kernel_driver(struct usb_sock_t *usb,
   /* Make kernel release interface */
   if (libusb_kernel_driver_active(usb->printer, uf->libusb_interface_index) ==
       1) {
-    /* Only linux supports this
-       other platforms will fail
-       thus we ignore the error code
-       it either works or it does not */
+    /* Only linux supports this other platforms will fail thus we ignore the
+       error code it either works or it does not */
     libusb_detach_kernel_driver(usb->printer, uf->libusb_interface_index);
   }
 }
@@ -186,9 +184,8 @@ static int try_claim_usb_interface(struct usb_sock_t *usb,
   /* Claim the whole interface */
   int status = 0;
   do {
-    /* Spinlock-like
-       Libusb does not offer a blocking call
-       so we're left with a spinlock */
+    /* Spinlock-like Libusb does not offer a blocking call so we're left with a
+       spinlock. */
     status = libusb_claim_interface(usb->printer, uf->libusb_interface_index);
     if (status)
       NOTE("Failed to claim interface %d, retrying",
@@ -412,16 +409,16 @@ struct usb_sock_t *usb_open()
 	goto error;
       }
 
-      /* Try to make the kernel release the usb interface */
+      /* Try to make the kernel release the usb interface. */
       try_detach_kernel_driver(usb, uf);
 
-      /* Try to claim the usb interface */
+      /* Try to claim the usb interface. */
       if (try_claim_usb_interface(usb, uf)) {
         ERR("Failed to claim usb interface #%d", uf->interface_number);
         goto error;
       }
 
-      /* Select the IPP-USB alt setting of the interface */
+      /* Select the IPP-USB alt setting of the interface. */
       if (libusb_set_interface_alt_setting(
               usb->printer, uf->libusb_interface_index, uf->interface_alt)) {
         ERR("Failed to set alt setting for interface #%d",
@@ -492,7 +489,10 @@ void usb_close(struct usb_sock_t *usb)
   NOTE("Resetting printer ...");
   libusb_reset_device(usb->printer);
   NOTE("Reset completed.");
+  NOTE("Closing device handle...");
   libusb_close(usb->printer);
+  NOTE("Closed device handle.");
+
   if (usb != NULL) {
     if (usb->context != NULL)
       libusb_exit(usb->context);
@@ -673,7 +673,6 @@ struct usb_conn_t *usb_conn_acquire(struct usb_sock_t *usb)
       ERR("Timed out waiting for a free USB interface");
       return NULL;
     }
-    usleep(100000);
   }
 
   struct usb_conn_t *conn = calloc(1, sizeof(*conn));
@@ -787,148 +786,18 @@ int usb_conn_packet_send(struct usb_conn_t *conn, struct http_packet_t *pkt)
   return 0;
 }
 
-struct http_packet_t *usb_conn_packet_get(struct usb_conn_t *conn, struct http_message_t *msg)
+struct libusb_transfer *setup_async_read(struct usb_conn_t *conn,
+                                         struct http_packet_t *pkt,
+                                         libusb_transfer_cb_fn callback,
+                                         void *user_data, uint32_t timeout)
 {
-  if (msg->is_completed)
+  struct libusb_transfer *transfer = libusb_alloc_transfer(0);
+  if (transfer == NULL)
     return NULL;
 
-  struct http_packet_t *pkt = packet_new(msg);
-  if (pkt == NULL) {
-    ERR("failed to create packet for incoming usb message");
-    goto cleanup;
-  }
-
-  /* File packet */
-  const int timeout = 1000; /* 1 sec */
-  size_t read_size_ulong = packet_pending_bytes(pkt);
-  if (read_size_ulong == 0)
-    return pkt;
-
-  uint64_t times_staled = 0;
-  while (read_size_ulong > 0 && !msg->is_completed && !g_options.terminate) {
-    if (read_size_ulong >= INT_MAX)
-      goto cleanup;
-    int read_size = (int)read_size_ulong;
-
-    /* Pad read_size to multiple of usb's max packet size */
-    read_size += (512 - (read_size % 512)) % 512;
-
-    /* Expand buffer if needed */
-    if (pkt->buffer_capacity < pkt->filled_size + read_size_ulong)
-      if (packet_expand(pkt) < 0) {
-	ERR("Failed to ensure room for usb pkt");
-	goto cleanup;
-      }
-
-    int gotten_size = 0;
-    int status = libusb_bulk_transfer(conn->parent->printer,
-		                      conn->interface->endpoint_in,
-		                      pkt->buffer + pkt->filled_size,
-		                      read_size,
-		                      &gotten_size, timeout);
-
-    if (status == LIBUSB_ERROR_NO_DEVICE) {
-      ERR("Printer has been disconnected");
-      goto cleanup;
-    }
-
-    if (status != 0 && status != LIBUSB_ERROR_TIMEOUT) {
-      ERR("bulk xfer failed with error code %d", status);
-      ERR("tried reading %d bytes", read_size);
-      goto cleanup;
-    } else if (status == LIBUSB_ERROR_TIMEOUT) {
-      ERR("bulk xfer timed out, retrying ...");
-      ERR("tried reading %d bytes, actually read %d bytes",
-	  read_size, gotten_size);
-    }
-
-    if (gotten_size < 0) {
-      ERR("Negative read size unexpected");
-      goto cleanup;
-    }
-
-    if (gotten_size > 0) {
-      times_staled = 0;
-      usb_conn_mark_moving(conn);
-    } else {
-
-      /* Performance Test ---------------
-	 How long we sleep here has a
-	 dramatic affect on how long it
-	 takes to load a page.
-	 Earlier versions waited a tenth
-	 of a second which resulted in
-	 minute long page loads.
-	 On my HP printer the most obvious
-	 bottleneck is the "Unified.js" file
-	 which weighs 517.87KB. My profiling
-	 looked at how shortening this sleep
-	 could improve this file's load times.
-	 The cycle count is from perf and
-	 covers an entire page load.
-
-	 Below are my results:
-	 1 in    100 == 2447ms, 261M cycles
-	 1 in  1,000 == 483ms,  500M cycles
-	 5 in 10,000 == 433ms,  800M cycles
-	 1 in 10,000 == 320ms, 3000M cycles */
-      #define TIMEOUT_RATIO (10000 / 5)
-      static uint64_t stale_timeout =
-	CONN_STALE_THRESHHOLD * TIMEOUT_RATIO;
-      static uint64_t crash_timeout =
-	PRINTER_CRASH_TIMEOUT_ANSWER * TIMEOUT_RATIO;
-      static uint64_t skip_timeout  =
-	1000000000 / TIMEOUT_RATIO;
-
-      struct timespec sleep_dur;
-      sleep_dur.tv_sec = 0;
-      sleep_dur.tv_nsec = skip_timeout;
-      nanosleep(&sleep_dur, NULL);
-
-      if (status == LIBUSB_ERROR_TIMEOUT)
-	times_staled += TIMEOUT_RATIO * timeout / 1000;
-      else
-	times_staled++;
-      if (times_staled % TIMEOUT_RATIO == 0 ||
-	  status == LIBUSB_ERROR_TIMEOUT) {
-	NOTE("No bytes received for %d sec.",
-	     times_staled / TIMEOUT_RATIO);
-	if (pkt->filled_size > 0)
-	  NOTE("Packet so far \n===\n%s===\n",
-	       hexdump(pkt->buffer,
-		       pkt->filled_size));
-      }
- 
-      if (times_staled > stale_timeout) {
-	usb_conn_mark_staled(conn);
-
-	if (pkt->filled_size > 0 ||
-	    usb_all_conns_staled(conn->parent) ||
-	    times_staled > crash_timeout) {
-	  ERR("USB timed out, giving up waiting for more data");
-	  break;
-	}
-      }
-    }
-
-    if (gotten_size) {
-      NOTE("USB: Getting %d bytes of %d",
-	   read_size, pkt->expected_size);
-      NOTE("USB: Got %d bytes", gotten_size);
-    }
-    packet_mark_received(pkt, (size_t)gotten_size);
-    read_size_ulong = packet_pending_bytes(pkt);
-  }
-  NOTE("USB: Received %d bytes of %d with type %d",
-       pkt->filled_size, pkt->expected_size, msg->type);
-
-  if (pkt->filled_size == 0)
-    goto cleanup;
+  libusb_fill_bulk_transfer(transfer, conn->parent->printer,
+                            conn->interface->endpoint_in, pkt->buffer,
+                            pkt->buffer_capacity, callback, user_data, timeout);
 
-  return pkt;
-
- cleanup:
-  if (pkt != NULL)
-    packet_free(pkt);
-  return NULL;
+  return transfer;
 }
diff --git a/src/usb.h b/src/usb.h
index e23e03c..3aa939c 100644
--- a/src/usb.h
+++ b/src/usb.h
@@ -67,4 +67,8 @@ struct usb_conn_t *usb_conn_acquire(struct usb_sock_t *);
 void usb_conn_release(struct usb_conn_t *);
 
 int usb_conn_packet_send(struct usb_conn_t *, struct http_packet_t *);
-struct http_packet_t *usb_conn_packet_get(struct usb_conn_t *, struct http_message_t *);
+
+struct libusb_transfer *setup_async_read(struct usb_conn_t *conn,
+                                         struct http_packet_t *pkt,
+                                         libusb_transfer_cb_fn callback,
+                                         void *user_data, uint32_t timeout);