X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=libmemcached%2Fio.cc;h=85e5ed42b82460a69d6c6b364b0af682ed189919;hb=f5838944c3424afa14765d709a127c943f1cf87b;hp=c04deb30502001570333bfc82f24a94ee57d0e6b;hpb=3c4d734f56530d43520f385bff97162c04ac81ac;p=awesomized%2Flibmemcached

diff --git a/libmemcached/io.cc b/libmemcached/io.cc
index c04deb30..85e5ed42 100644
--- a/libmemcached/io.cc
+++ b/libmemcached/io.cc
@@ -44,10 +44,152 @@ enum memc_read_or_write {
   MEM_WRITE
 };
 
-static ssize_t io_flush(memcached_server_write_instance_st ptr,
-                        const bool with_flush,
-                        memcached_return_t *error);
-static void increment_udp_message_id(memcached_server_write_instance_st ptr);
+/*
+ * The udp request id consists of two seperate sections
+ *   1) The thread id
+ *   2) The message number
+ * The thread id should only be set when the memcached_st struct is created
+ * and should not be changed.
+ *
+ * The message num is incremented for each new message we send, this function
+ * extracts the message number from message_id, increments it and then
+ * writes the new value back into the header
+ */
+static void increment_udp_message_id(memcached_server_write_instance_st ptr)
+{
+  struct udp_datagram_header_st *header= (struct udp_datagram_header_st *)ptr->write_buffer;
+  uint16_t cur_req= get_udp_datagram_request_id(header);
+  int msg_num= get_msg_num_from_request_id(cur_req);
+  int thread_id= get_thread_id_from_request_id(cur_req);
+
+  if (((++msg_num) & UDP_REQUEST_ID_THREAD_MASK) != 0)
+    msg_num= 0;
+
+  header->request_id= htons((uint16_t) (thread_id | msg_num));
+}
+
+/**
+ * Try to fill the input buffer for a server with as much
+ * data as possible.
+ *
+ * @param ptr the server to pack
+ */
+static bool repack_input_buffer(memcached_server_write_instance_st ptr)
+{
+  if (ptr->read_ptr != ptr->read_buffer)
+  {
+    /* Move all of the data to the beginning of the buffer so
+     ** that we can fit more data into the buffer...
+   */
+    memmove(ptr->read_buffer, ptr->read_ptr, ptr->read_buffer_length);
+    ptr->read_ptr= ptr->read_buffer;
+    ptr->read_data_length= ptr->read_buffer_length;
+  }
+
+  /* There is room in the buffer, try to fill it! */
+  if (ptr->read_buffer_length != MEMCACHED_MAX_BUFFER)
+  {
+    do {
+      /* Just try a single read to grab what's available */
+      ssize_t nr= recv(ptr->fd,
+                       ptr->read_ptr + ptr->read_data_length,
+                       MEMCACHED_MAX_BUFFER - ptr->read_data_length,
+                       MSG_DONTWAIT);
+
+      switch (nr)
+      {
+      case SOCKET_ERROR:
+        {
+          switch (get_socket_errno())
+          {
+          case EINTR:
+            continue;
+
+          case EWOULDBLOCK:
+#ifdef USE_EAGAIN
+          case EAGAIN:
+#endif
+#ifdef TARGET_OS_LINUX
+          case ERESTART:
+#endif
+            break; // No IO is fine, we can just move on
+
+          default:
+            memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
+          }
+        }
+        break;
+
+      case 0: // Shutdown on the socket has occurred
+        {
+          memcached_set_error(*ptr, MEMCACHED_CONNECTION_FAILURE, MEMCACHED_AT);
+        }
+        break;
+
+      default:
+        {
+          ptr->read_data_length+= size_t(nr);
+          ptr->read_buffer_length+= size_t(nr);
+          return true;
+        }
+        break;
+      }
+    } while (0);
+  }
+  return false;
+}
+
+/**
+ * If the we have callbacks connected to this server structure
+ * we may start process the input queue and fire the callbacks
+ * for the incomming messages. This function is _only_ called
+ * when the input buffer is full, so that we _know_ that we have
+ * at least _one_ message to process.
+ *
+ * @param ptr the server to star processing iput messages for
+ * @return true if we processed anything, false otherwise
+ */
+static bool process_input_buffer(memcached_server_write_instance_st ptr)
+{
+  /*
+   ** We might be able to process some of the response messages if we
+   ** have a callback set up
+ */
+  if (ptr->root->callbacks != NULL && ptr->root->flags.use_udp == false)
+  {
+    /*
+     * We might have responses... try to read them out and fire
+     * callbacks
+   */
+    memcached_callback_st cb= *ptr->root->callbacks;
+
+    memcached_set_processing_input((memcached_st *)ptr->root, true);
+
+    char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE];
+    memcached_return_t error;
+    memcached_st *root= (memcached_st *)ptr->root;
+    error= memcached_response(ptr, buffer, sizeof(buffer),
+                              &root->result);
+
+    memcached_set_processing_input(root, false);
+
+    if (error == MEMCACHED_SUCCESS)
+    {
+      for (unsigned int x= 0; x < cb.number_of_callback; x++)
+      {
+        error= (*cb.callback[x])(ptr->root, &root->result, cb.context);
+        if (error != MEMCACHED_SUCCESS)
+          break;
+      }
+
+      /* @todo what should I do with the error message??? */
+    }
+    /* @todo what should I do with other error messages?? */
+    return true;
+  }
+
+  return false;
+}
 
 static memcached_return_t io_wait(memcached_server_write_instance_st ptr,
                                   memc_read_or_write read_or_write)
@@ -126,16 +268,23 @@ static memcached_return_t io_wait(memcached_server_write_instance_st ptr,
         {
           int err;
           socklen_t len= sizeof (err);
-          (void)getsockopt(ptr->fd, SOL_SOCKET, SO_ERROR, &err, &len);
-          memcached_set_errno(*ptr, (err == 0) ? get_socket_errno() : err, MEMCACHED_AT);
+          if (getsockopt(ptr->fd, SOL_SOCKET, SO_ERROR, &err, &len) == 0)
+          {
+            if (err == 0)
+            {
+              continue;
+            }
+            errno= err;
+          }
         }
         else
         {
           memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
         }
+        int local_errno= get_socket_errno(); // We cache in case memcached_quit_server() modifies errno
         memcached_quit_server(ptr, true);
 
-        return memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
+        return memcached_set_errno(*ptr, local_errno, MEMCACHED_AT);
       }
     }
   }
@@ -145,132 +294,160 @@ static memcached_return_t io_wait(memcached_server_write_instance_st ptr,
   return memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
 }
 
-memcached_return_t memcached_io_wait_for_write(memcached_server_write_instance_st ptr)
+static ssize_t io_flush(memcached_server_write_instance_st ptr,
+                        const bool with_flush,
+                        memcached_return_t *error)
 {
-  return io_wait(ptr, MEM_WRITE);
-}
-
-/**
- * Try to fill the input buffer for a server with as much
- * data as possible.
- *
- * @param ptr the server to pack
+  /*
+   ** We might want to purge the input buffer if we haven't consumed
+   ** any output yet... The test for the limits is the purge is inline
+   ** in the purge function to avoid duplicating the logic..
  */
-static bool repack_input_buffer(memcached_server_write_instance_st ptr)
-{
-  if (ptr->read_ptr != ptr->read_buffer)
   {
-    /* Move all of the data to the beginning of the buffer so
-     ** that we can fit more data into the buffer...
-   */
-    memmove(ptr->read_buffer, ptr->read_ptr, ptr->read_buffer_length);
-    ptr->read_ptr= ptr->read_buffer;
-    ptr->read_data_length= ptr->read_buffer_length;
+    memcached_return_t rc;
+    WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
+    rc= memcached_purge(ptr);
+
+    if (rc != MEMCACHED_SUCCESS && rc != MEMCACHED_STORED)
+    {
+      return -1;
+    }
   }
+  size_t return_length;
+  char *local_write_ptr= ptr->write_buffer;
+  size_t write_length= ptr->write_buffer_offset;
 
-  /* There is room in the buffer, try to fill it! */
-  if (ptr->read_buffer_length != MEMCACHED_MAX_BUFFER)
+  *error= MEMCACHED_SUCCESS;
+
+  WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
+
+  // UDP Sanity check, make sure that we are not sending somthing too big
+  if (memcached_is_udp(ptr->root) and write_length > MAX_UDP_DATAGRAM_LENGTH)
   {
-    do {
-      /* Just try a single read to grab what's available */
-      ssize_t nr= recv(ptr->fd,
-                       ptr->read_ptr + ptr->read_data_length,
-                       MEMCACHED_MAX_BUFFER - ptr->read_data_length,
-                       MSG_DONTWAIT);
+    *error= MEMCACHED_WRITE_FAILURE;
+    return -1;
+  }
 
-      switch (nr)
+  if (ptr->write_buffer_offset == 0 or (memcached_is_udp(ptr->root) and ptr->write_buffer_offset == UDP_DATAGRAM_HEADER_LENGTH))
+  {
+    return 0;
+  }
+
+  /* Looking for memory overflows */
+#if defined(DEBUG)
+  if (write_length == MEMCACHED_MAX_BUFFER)
+    WATCHPOINT_ASSERT(ptr->write_buffer == local_write_ptr);
+  WATCHPOINT_ASSERT((ptr->write_buffer + MEMCACHED_MAX_BUFFER) >= (local_write_ptr + write_length));
+#endif
+
+  return_length= 0;
+  while (write_length)
+  {
+    WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
+    WATCHPOINT_ASSERT(write_length > 0);
+    if (memcached_is_udp(ptr->root))
+    {
+      increment_udp_message_id(ptr);
+    }
+
+    ssize_t sent_length= 0;
+    WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
+    if (with_flush)
+    {
+      sent_length= send(ptr->fd, local_write_ptr, write_length, MSG_NOSIGNAL|MSG_DONTWAIT);
+    }
+    else
+    {
+      sent_length= send(ptr->fd, local_write_ptr, write_length, MSG_NOSIGNAL|MSG_DONTWAIT|MSG_MORE);
+    }
+
+    if (sent_length == SOCKET_ERROR)
+    {
+      memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
+#if 0 // @todo I should look at why we hit this bit of code hard frequently
+      WATCHPOINT_ERRNO(get_socket_errno());
+      WATCHPOINT_NUMBER(get_socket_errno());
+#endif
+      switch (get_socket_errno())
       {
-      case SOCKET_ERROR:
+      case ENOBUFS:
+        continue;
+      case EWOULDBLOCK:
+#ifdef USE_EAGAIN
+      case EAGAIN:
+#endif
         {
-          switch (get_socket_errno())
+          /*
+           * We may be blocked on write because the input buffer
+           * is full. Let's check if we have room in our input
+           * buffer for more data and retry the write before
+           * waiting..
+         */
+          if (repack_input_buffer(ptr) or process_input_buffer(ptr))
           {
-          case EINTR:
             continue;
+          }
 
-          case EWOULDBLOCK:
-#ifdef USE_EAGAIN
-          case EAGAIN:
-#endif
-#ifdef TARGET_OS_LINUX
-          case ERESTART:
-#endif
-            break; // No IO is fine, we can just move on
-
-          default:
-            memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
+          memcached_return_t rc= io_wait(ptr, MEM_WRITE);
+          if (memcached_success(rc))
+          {
+            continue;
+          }
+          else if (rc == MEMCACHED_TIMEOUT)
+          {
+            *error= memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT);
+            return -1;
           }
-        }
-        break;
 
-      case 0: // Shutdown on the socket has occurred
-        {
-          memcached_set_error(*ptr, MEMCACHED_CONNECTION_FAILURE, MEMCACHED_AT);
+          memcached_quit_server(ptr, true);
+          *error= memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
+          return -1;
         }
-        break;
-
+      case ENOTCONN:
+      case EPIPE:
       default:
-        {
-          ptr->read_data_length+= size_t(nr);
-          ptr->read_buffer_length+= size_t(nr);
-          return true;
-        }
-        break;
+        memcached_quit_server(ptr, true);
+        *error= memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
+        WATCHPOINT_ASSERT(ptr->fd == -1);
+        return -1;
       }
-    } while (0);
-  }
-  return false;
-}
-
-/**
- * If the we have callbacks connected to this server structure
- * we may start process the input queue and fire the callbacks
- * for the incomming messages. This function is _only_ called
- * when the input buffer is full, so that we _know_ that we have
- * at least _one_ message to process.
- *
- * @param ptr the server to star processing iput messages for
- * @return true if we processed anything, false otherwise
- */
-static bool process_input_buffer(memcached_server_write_instance_st ptr)
-{
-  /*
-   ** We might be able to process some of the response messages if we
-   ** have a callback set up
- */
-  if (ptr->root->callbacks != NULL && ptr->root->flags.use_udp == false)
-  {
-    /*
-     * We might have responses... try to read them out and fire
-     * callbacks
-   */
-    memcached_callback_st cb= *ptr->root->callbacks;
+    }
 
-    memcached_set_processing_input((memcached_st *)ptr->root, true);
+    if (memcached_is_udp(ptr->root) and size_t(sent_length) != write_length)
+    {
+      memcached_quit_server(ptr, true);
+      *error= memcached_set_error(*ptr, MEMCACHED_WRITE_FAILURE, MEMCACHED_AT);
+      return -1;
+    }
 
-    char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE];
-    memcached_return_t error;
-    memcached_st *root= (memcached_st *)ptr->root;
-    error= memcached_response(ptr, buffer, sizeof(buffer),
-                              &root->result);
+    ptr->io_bytes_sent+= uint32_t(sent_length);
 
-    memcached_set_processing_input(root, false);
+    local_write_ptr+= sent_length;
+    write_length-= uint32_t(sent_length);
+    return_length+= uint32_t(sent_length);
+  }
 
-    if (error == MEMCACHED_SUCCESS)
-    {
-      for (unsigned int x= 0; x < cb.number_of_callback; x++)
-      {
-        error= (*cb.callback[x])(ptr->root, &root->result, cb.context);
-        if (error != MEMCACHED_SUCCESS)
-          break;
-      }
+  WATCHPOINT_ASSERT(write_length == 0);
+  // Need to study this assert() WATCHPOINT_ASSERT(return_length ==
+  // ptr->write_buffer_offset);
 
-      /* @todo what should I do with the error message??? */
-    }
-    /* @todo what should I do with other error messages?? */
-    return true;
+  // if we are a udp server, the begining of the buffer is reserverd for
+  // the upd frame header
+  if (memcached_is_udp(ptr->root))
+  {
+    ptr->write_buffer_offset= UDP_DATAGRAM_HEADER_LENGTH;
+  }
+  else
+  {
+    ptr->write_buffer_offset= 0;
   }
 
-  return false;
+  return (ssize_t) return_length;
+}
+
+memcached_return_t memcached_io_wait_for_write(memcached_server_write_instance_st ptr)
+{
+  return io_wait(ptr, MEM_WRITE);
 }
 
 memcached_return_t memcached_io_read(memcached_server_write_instance_st ptr,
@@ -379,8 +556,8 @@ memcached_return_t memcached_io_read(memcached_server_write_instance_st ptr,
     }
   }
 
-  ptr->server_failure_counter= 0;
   *nread = (ssize_t)(buffer_ptr - (char*)buffer);
+
   return MEMCACHED_SUCCESS;
 }
 
@@ -443,13 +620,10 @@ memcached_return_t memcached_io_slurp(memcached_server_write_instance_st ptr)
 static ssize_t _io_write(memcached_server_write_instance_st ptr,
                          const void *buffer, size_t length, bool with_flush)
 {
-  size_t original_length;
-  const char* buffer_ptr;
-
   WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
 
-  original_length= length;
-  buffer_ptr= static_cast<const char *>(buffer);
+  size_t original_length= length;
+  const char *buffer_ptr= static_cast<const char *>(buffer);
 
   while (length)
   {
@@ -457,7 +631,7 @@ static ssize_t _io_write(memcached_server_write_instance_st ptr,
     size_t should_write;
     size_t buffer_end;
 
-    if (ptr->type == MEMCACHED_CONNECTION_UDP)
+    if (memcached_is_udp(ptr->root))
     {
       //UDP does not support partial writes
       buffer_end= MAX_UDP_DATAGRAM_LENGTH;
@@ -480,7 +654,7 @@ static ssize_t _io_write(memcached_server_write_instance_st ptr,
     buffer_ptr+= should_write;
     length-= should_write;
 
-    if (ptr->write_buffer_offset == buffer_end && ptr->type != MEMCACHED_CONNECTION_UDP)
+    if (ptr->write_buffer_offset == buffer_end and memcached_is_udp(ptr->root) == false)
     {
       WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
 
@@ -639,155 +813,6 @@ memcached_server_write_instance_st memcached_io_get_readable_server(memcached_st
   return NULL;
 }
 
-static ssize_t io_flush(memcached_server_write_instance_st ptr,
-                        const bool with_flush,
-                        memcached_return_t *error)
-{
-  /*
-   ** We might want to purge the input buffer if we haven't consumed
-   ** any output yet... The test for the limits is the purge is inline
-   ** in the purge function to avoid duplicating the logic..
- */
-  {
-    memcached_return_t rc;
-    WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
-    rc= memcached_purge(ptr);
-
-    if (rc != MEMCACHED_SUCCESS && rc != MEMCACHED_STORED)
-    {
-      return -1;
-    }
-  }
-  ssize_t sent_length;
-  size_t return_length;
-  char *local_write_ptr= ptr->write_buffer;
-  size_t write_length= ptr->write_buffer_offset;
-
-  *error= MEMCACHED_SUCCESS;
-
-  WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
-
-  // UDP Sanity check, make sure that we are not sending somthing too big
-  if (ptr->type == MEMCACHED_CONNECTION_UDP && write_length > MAX_UDP_DATAGRAM_LENGTH)
-  {
-    *error= MEMCACHED_WRITE_FAILURE;
-    return -1;
-  }
-
-  if (ptr->write_buffer_offset == 0 || (ptr->type == MEMCACHED_CONNECTION_UDP
-                                        && ptr->write_buffer_offset == UDP_DATAGRAM_HEADER_LENGTH))
-  {
-    return 0;
-  }
-
-  /* Looking for memory overflows */
-#if defined(DEBUG)
-  if (write_length == MEMCACHED_MAX_BUFFER)
-    WATCHPOINT_ASSERT(ptr->write_buffer == local_write_ptr);
-  WATCHPOINT_ASSERT((ptr->write_buffer + MEMCACHED_MAX_BUFFER) >= (local_write_ptr + write_length));
-#endif
-
-  return_length= 0;
-  while (write_length)
-  {
-    WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
-    WATCHPOINT_ASSERT(write_length > 0);
-    sent_length= 0;
-    if (ptr->type == MEMCACHED_CONNECTION_UDP)
-      increment_udp_message_id(ptr);
-
-    WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
-    if (with_flush)
-    {
-      sent_length= send(ptr->fd, local_write_ptr, write_length, MSG_NOSIGNAL|MSG_DONTWAIT);
-    }
-    else
-    {
-      sent_length= send(ptr->fd, local_write_ptr, write_length, MSG_NOSIGNAL|MSG_DONTWAIT|MSG_MORE);
-    }
-
-    if (sent_length == SOCKET_ERROR)
-    {
-      memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
-#if 0 // @todo I should look at why we hit this bit of code hard frequently
-      WATCHPOINT_ERRNO(get_socket_errno());
-      WATCHPOINT_NUMBER(get_socket_errno());
-#endif
-      switch (get_socket_errno())
-      {
-      case ENOBUFS:
-        continue;
-      case EWOULDBLOCK:
-#ifdef USE_EAGAIN
-      case EAGAIN:
-#endif
-        {
-          /*
-           * We may be blocked on write because the input buffer
-           * is full. Let's check if we have room in our input
-           * buffer for more data and retry the write before
-           * waiting..
-         */
-          if (repack_input_buffer(ptr) or
-              process_input_buffer(ptr))
-          {
-            continue;
-          }
-
-          memcached_return_t rc= io_wait(ptr, MEM_WRITE);
-          if (memcached_success(rc))
-          {
-            continue;
-          }
-          else if (rc == MEMCACHED_TIMEOUT)
-          {
-            *error= memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT);
-            return -1;
-          }
-
-          memcached_quit_server(ptr, true);
-          *error= memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
-          return -1;
-        }
-      case ENOTCONN:
-      case EPIPE:
-      default:
-        memcached_quit_server(ptr, true);
-        *error= memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
-        WATCHPOINT_ASSERT(ptr->fd == -1);
-        return -1;
-      }
-    }
-
-    if (ptr->type == MEMCACHED_CONNECTION_UDP and
-        (size_t)sent_length != write_length)
-    {
-      memcached_quit_server(ptr, true);
-      *error= memcached_set_error(*ptr, MEMCACHED_WRITE_FAILURE, MEMCACHED_AT);
-      return -1;
-    }
-
-    ptr->io_bytes_sent += (uint32_t) sent_length;
-
-    local_write_ptr+= sent_length;
-    write_length-= (uint32_t) sent_length;
-    return_length+= (uint32_t) sent_length;
-  }
-
-  WATCHPOINT_ASSERT(write_length == 0);
-  // Need to study this assert() WATCHPOINT_ASSERT(return_length ==
-  // ptr->write_buffer_offset);
-
-  // if we are a udp server, the begining of the buffer is reserverd for
-  // the upd frame header
-  if (ptr->type == MEMCACHED_CONNECTION_UDP)
-    ptr->write_buffer_offset= UDP_DATAGRAM_HEADER_LENGTH;
-  else
-    ptr->write_buffer_offset= 0;
-
-  return (ssize_t) return_length;
-}
-
 /*
   Eventually we will just kill off the server with the problem.
 */
@@ -827,10 +852,11 @@ memcached_return_t memcached_safe_read(memcached_server_write_instance_st ptr,
 
 memcached_return_t memcached_io_readline(memcached_server_write_instance_st ptr,
                                          char *buffer_ptr,
-                                         size_t size)
+                                         size_t size,
+                                         size_t& total_nr)
 {
+  total_nr= 0;
   bool line_complete= false;
-  size_t total_nr= 0;
 
   while (not line_complete)
   {
@@ -879,30 +905,6 @@ memcached_return_t memcached_io_readline(memcached_server_write_instance_st ptr,
   return MEMCACHED_SUCCESS;
 }
 
-/*
- * The udp request id consists of two seperate sections
- *   1) The thread id
- *   2) The message number
- * The thread id should only be set when the memcached_st struct is created
- * and should not be changed.
- *
- * The message num is incremented for each new message we send, this function
- * extracts the message number from message_id, increments it and then
- * writes the new value back into the header
- */
-static void increment_udp_message_id(memcached_server_write_instance_st ptr)
-{
-  struct udp_datagram_header_st *header= (struct udp_datagram_header_st *)ptr->write_buffer;
-  uint16_t cur_req= get_udp_datagram_request_id(header);
-  int msg_num= get_msg_num_from_request_id(cur_req);
-  int thread_id= get_thread_id_from_request_id(cur_req);
-
-  if (((++msg_num) & UDP_REQUEST_ID_THREAD_MASK) != 0)
-    msg_num= 0;
-
-  header->request_id= htons((uint16_t) (thread_id | msg_num));
-}
-
 memcached_return_t memcached_io_init_udp_header(memcached_server_write_instance_st ptr, uint16_t thread_id)
 {
   if (thread_id > UDP_REQUEST_ID_MAX_THREAD_ID)