From:  <brian@gir.local>
Date: Thu, 29 Nov 2007 03:48:02 +0000 (-0800)
Subject: Extended consistent hash logic to handle splice (should distribute keys much
X-Git-Tag: 0.13~65
X-Git-Url: https://git.m6w6.name/?a=commitdiff_plain;h=122f18f5a7bf1c0361ee6edca276a0cabc95ed6d;p=m6w6%2Flibmemcached

Extended consistent hash logic to handle splice (should distribute keys much
more evenly then previous design).
---

diff --git a/include/memcached.h b/include/memcached.h
index 3d1bfcd8..194170ef 100644
--- a/include/memcached.h
+++ b/include/memcached.h
@@ -33,6 +33,8 @@ typedef struct memcached_server_st memcached_server_st;
 #define MEMCACHED_MAX_KEY 251 /* We add one to have it null terminated */
 #define MEMCACHED_MAX_BUFFER HUGE_STRING_LEN
 #define MEMCACHED_MAX_HOST_LENGTH 64
+#define MEMCACHED_WHEEL_SIZE 1024
+#define MEMCACHED_STRIDE 4
 
 typedef enum {
   MEMCACHED_SUCCESS,
@@ -67,6 +69,11 @@ typedef enum {
   MEMCACHED_MAXIMUM_RETURN, /* Always add new error code before */
 } memcached_return;
 
+typedef enum {
+  MEMCACHED_DISTRIBUTION_MODULUS,
+  MEMCACHED_DISTRIBUTION_CONSISTENT,
+} memcached_server_distribution;
+
 typedef enum {
   MEMCACHED_BEHAVIOR_NO_BLOCK,
   MEMCACHED_BEHAVIOR_TCP_NODELAY,
@@ -184,6 +191,8 @@ struct memcached_st {
   int32_t poll_timeout;
   memcached_string_st result_buffer;
   memcached_hash hash;
+  memcached_server_distribution distribution;
+  unsigned int wheel[MEMCACHED_WHEEL_SIZE];
   memcached_return warning; /* Future Use */
 };
 
diff --git a/lib/memcached.c b/lib/memcached.c
index e40c3db4..1ed39a91 100644
--- a/lib/memcached.c
+++ b/lib/memcached.c
@@ -23,6 +23,7 @@ memcached_st *memcached_create(memcached_st *ptr)
   string_ptr= memcached_string_create(ptr, &ptr->result_buffer, 0);
   WATCHPOINT_ASSERT(string_ptr);
   ptr->poll_timeout= -1;
+  ptr->distribution= MEMCACHED_DISTRIBUTION_MODULUS;
 
   return ptr;
 }
@@ -75,6 +76,7 @@ memcached_st *memcached_clone(memcached_st *clone, memcached_st *ptr)
   new_clone->send_size= ptr->send_size;
   new_clone->recv_size= ptr->recv_size;
   new_clone->poll_timeout= ptr->poll_timeout;
+  new_clone->distribution= ptr->distribution;
 
   return new_clone;
 }
diff --git a/lib/memcached_hash.c b/lib/memcached_hash.c
index 9cc846a8..95d7f862 100644
--- a/lib/memcached_hash.c
+++ b/lib/memcached_hash.c
@@ -85,18 +85,18 @@ unsigned int memcached_generate_hash(memcached_st *ptr, char *key, size_t key_le
   }
 
   WATCHPOINT_ASSERT(hash);
-  if (ptr->flags & MEM_USE_KETAMA)
+
+  if (ptr->distribution == MEMCACHED_DISTRIBUTION_MODULUS)
   {
-    WATCHPOINT_ASSERT(0);
-    return 0;
+    return hash % ptr->number_of_hosts;
   }
   else
   {
     unsigned int server_key;
 
-    server_key= hash % ptr->number_of_hosts;
+    server_key= hash % MEMCACHED_WHEEL_SIZE;
 
-    return server_key;
+    return ptr->wheel[server_key];
   }
 }
 
diff --git a/lib/memcached_hosts.c b/lib/memcached_hosts.c
index 0090625c..3517d824 100644
--- a/lib/memcached_hosts.c
+++ b/lib/memcached_hosts.c
@@ -6,6 +6,34 @@ static memcached_return server_add(memcached_st *ptr, char *hostname,
                                    unsigned int port,
                                    memcached_connection type);
 
+#define MEMCACHED_WHEEL_SIZE 1024
+#define MEMCACHED_STRIDE 4
+static void rebalance_wheel(memcached_st *ptr)
+{
+  unsigned int x;
+  unsigned int y;
+  unsigned int latch;
+  unsigned int range;
+
+  range= (MEMCACHED_WHEEL_SIZE / ptr->number_of_hosts);
+
+  /* Seed the Wheel */
+  memset(ptr->wheel, 0, sizeof(unsigned int) * MEMCACHED_WHEEL_SIZE);
+
+  for (latch= y= x= 0; x < MEMCACHED_WHEEL_SIZE; x++, latch++)
+  {
+    if (latch == MEMCACHED_STRIDE)
+    {
+      y++;
+      if (y == ptr->number_of_hosts)
+        y= 0;
+      latch= 0;
+    }
+
+    ptr->wheel[x]= y;
+  }
+}
+
 static void host_reset(memcached_server_st *host, char *hostname, unsigned int port,
                        memcached_connection type)
 {
@@ -48,6 +76,8 @@ memcached_return memcached_server_push(memcached_st *ptr, memcached_server_st *l
   }
   ptr->hosts[0].count= ptr->number_of_hosts;
 
+  rebalance_wheel(ptr);
+
   return MEMCACHED_SUCCESS;
 }
 
@@ -104,6 +134,8 @@ static memcached_return server_add(memcached_st *ptr, char *hostname,
   ptr->number_of_hosts++;
   ptr->hosts[0].count++;
 
+  rebalance_wheel(ptr);
+
   LIBMEMCACHED_MEMCACHED_SERVER_ADD_END();
 
   return MEMCACHED_SUCCESS;