Changeset 70e5d7

Timestamp:

01/06/08 21:55:16 (4 years ago)

Author:

Tomash Brechko <tomash.brechko@…>

Branches:

master, ketama-compat

Children:

4c792d

Parents:

3a7104

git-author:

Tomash Brechko <tomash.brechko@…> (01/06/08 18:37:10)

git-committer:

Tomash Brechko <tomash.brechko@…> (01/06/08 21:55:16)

Message:

Add struct array.

Files:

• MANIFEST (modified) (1 diff)
• TODO (modified) (3 diffs)
• src/array.c (added)
• src/array.h (added)
• src/client.c (modified) (36 diffs)
• src/dispatch_key.c (modified) (14 diffs)
• src/dispatch_key.h (modified) (1 diff)

MANIFEST

@@ -24 +24 @@
 src/genparser.pl
 src/reply.kw
+src/array.h
+src/array.c
 src/client.h
 src/client.c

TODO

@@ -21 +21 @@
 - UDP (?).
 
+- Investigate why there are read() errors in no_eagain branch.
+
+- In find_bin() replace binary search with interpolation search.
+
+- Add test for noreply, do not list it in MANIFEST.
+
 
 For 0.07:
@@ -29 +35 @@
   instance.
 
-- Add test for noreply, do not list it in MANIFEST.
-
 - Clean warnings for gcc 4.2.2.
 
 - Fix all ugliness introduced in 0.06 ;).
-
-- Investigate why there are read() errors in no_eagain branch.
 
 - Replace av_push() with av_store(), test result.
@@ -45 +47 @@
 - Rename bins to buckets.
 
-- In find_bin() replace binary search with interpolation search.
-
 - Improve server_versions command.

src/client.c

@@ -23 +23 @@
 
 #include "client.h"
+#include "array.h"
 #include "connect.h"
 #include "parse_keyword.h"
@@ -284 +285 @@
 struct client
 {
-  struct server *servers;
-  int server_count;
-  int server_capacity;
+  struct array servers;
 
   struct dispatch_state dispatch;
@@ -301 +300 @@
   int nowait;
 
-  struct key_node *key_list;
-  int key_list_count;
-  int key_list_capacity;
+  struct array key_list;
 
   generation_type generation;
@@ -357 +354 @@
 get_key_index(struct command_state *state)
 {
-  return state->client->key_list[state->key_head].key;
+  struct key_node *node =
+    array_elem(state->client->key_list, struct key_node, state->key_head);
+  return node->key;
 }
 
@@ -365 +364 @@
 next_key_index(struct command_state *state)
 {
-  state->key_head = state->client->key_list[state->key_head].next;
+  struct key_node *node =
+    array_elem(state->client->key_list, struct key_node, state->key_head);
+  state->key_head = node->next;
 }
 
@@ -376 +377 @@
     return NULL;
 
-  c->servers = NULL;
-  c->server_count = c->server_capacity = 0;
+  array_init(&c->servers);
+  array_init(&c->key_list);
 
   dispatch_init(&c->dispatch);
@@ -392 +393 @@
   c->nowait = 0;
 
-  c->key_list = NULL;
-  c->key_list_count = c->key_list_capacity = 0;
-
   c->generation = 1;            /* Different from initial command state.  */
 
@@ -404 +402 @@
 client_destroy(struct client *c)
 {
+  struct server *s;
   int i;
 
   client_nowait_push(c);
 
-  for (i = 0; i < c->server_count; ++i)
-    server_destroy(&c->servers[i]);
+  for (s = array_beg(c->servers, struct server);
+       s != array_end(c->servers, struct server);
+       ++s)
+    server_destroy(s);
 
   dispatch_destroy(&c->dispatch);
 
-  free(c->servers);
+  array_destroy(&c->servers);
+  array_destroy(&c->key_list);
   free(c->prefix);
-  free(c->key_list);
   free(c);
 }
@@ -424 +425 @@
 {
   /* Should be called before we added any server.  */
-  if (c->server_count > 0 || ketama_points < 0)
+  if (! array_empty(c->servers) || ketama_points < 0)
     return MEMCACHED_FAILURE;
 
@@ -485 +486 @@
     return MEMCACHED_FAILURE;
 
-  if (c->server_count == c->server_capacity)
-    {
-      int capacity = (c->server_capacity > 0 ? c->server_capacity + 1 : 1);
-      struct server *s =
-        (struct server *) realloc(c->servers,
-                                  capacity * sizeof(struct server));
-      if (! s)
-        return MEMCACHED_FAILURE;
-
-      c->servers = s;
-      c->server_capacity = capacity;
-    }
-
-  res = server_init(&c->servers[c->server_count], c,
+  if (array_extend(c->servers, struct server, 1, ARRAY_EXTEND_EXACT) == -1)
+    return MEMCACHED_FAILURE;
+
+  res = server_init(array_end(c->servers, struct server), c,
                     host, host_len, port, port_len, noreply);
   if (res != MEMCACHED_SUCCESS)
@@ -504 +495 @@
 
   res = dispatch_add_server(&c->dispatch, host, host_len, port, port_len,
-                            weight, c->server_count);
+                            weight, array_size(c->servers));
   if (res == -1)
     return MEMCACHED_FAILURE;
 
-  ++c->server_count;
+  array_push(c->servers);
 
   return MEMCACHED_SUCCESS;
@@ -1260 +1251 @@
 static
 void
-client_mark_failed(struct client *c, int server_index)
-{
-  struct server *s;
-
-  s = &c->servers[server_index];
-
+client_mark_failed(struct client *c, struct server *s)
+{
   if (s->cmd_state.fd != -1)
     {
@@ -1318 +1305 @@
   while (1)
     {
-      int max_fd, i, res;
+      struct server *s;
+      int max_fd, res;
 
       max_fd = -1;
-      for (i = 0; i < c->server_count; ++i)
+      for (s = array_beg(c->servers, struct server);
+           s != array_end(c->servers, struct server);
+           ++s)
         {
           int may_write, may_read;
-          struct command_state *state = &c->servers[i].cmd_state;
+          struct command_state *state = &s->cmd_state;
 
           if (! is_active(state))
@@ -1390 +1380 @@
                 case MEMCACHED_CLOSED:
                   deactivate(state);
-                  client_mark_failed(c, i);
+                  client_mark_failed(c, s);
                   break;
                 }
@@ -1412 +1402 @@
       FD_ZERO(&write_set);
       FD_ZERO(&read_set);
-      for (i = 0; i < c->server_count; ++i)
+      for (s = array_beg(c->servers, struct server);
+           s != array_end(c->servers, struct server);
+           ++s)
         {
-          struct command_state *state = &c->servers[i].cmd_state;
+          struct command_state *state = &s->cmd_state;
 
           if (is_active(state))
@@ -1447 +1439 @@
       if (res <= 0)
         {
-          for (i = 0; i < c->server_count; ++i)
+          for (s = array_beg(c->servers, struct server);
+               s != array_end(c->servers, struct server);
+               ++s)
             {
-              struct command_state *state = &c->servers[i].cmd_state;
+              struct command_state *state = &s->cmd_state;
 
               if (is_active(state))
@@ -1460 +1454 @@
                     state->u.value.object->free(state->u.value.opaque);
 
-                  client_mark_failed(c, i);
+                  client_mark_failed(c, s);
                 }
             }
@@ -1484 +1478 @@
 static
 int
-get_server_fd(struct client *c, int index)
-{
-  struct server *s;
+get_server_fd(struct client *c, struct server *s)
+{
   struct command_state *state;
-
-  s = &c->servers[index];
 
   /*
@@ -1518 +1509 @@
 
   if (state->fd == -1)
-    client_mark_failed(c, index);
+    client_mark_failed(c, s);
 
   return state->fd;
@@ -1525 +1516 @@
 
 static
-int
-client_get_server_index(struct client *c, const char *key, size_t key_len)
-{
+struct server *
+client_get_server(struct client *c, const char *key, size_t key_len)
+{
+  struct server *s;
   int index, fd;
 
   index = dispatch_key(&c->dispatch, key, key_len);
   if (index == -1)
-    return -1;
-
-  fd = get_server_fd(c, index);
+    return NULL;
+
+  s = array_elem(c->servers, struct server, index);
+
+  fd = get_server_fd(c, s);
   if (fd == -1)
-    return -1;
-
-  return index;
+    return NULL;
+
+  return s;
 }
 
@@ -1581 +1575 @@
 
   c = state->client;
-  if (c->key_list_count == c->key_list_capacity)
-    {
-      int capacity = (c->key_list_capacity > 0 ? c->key_list_capacity * 2 : 1);
-      struct key_node *list =
-        (struct key_node *) realloc(c->key_list,
-                                    sizeof(struct key_node) * capacity);
-      if (! list)
-        return MEMCACHED_FAILURE;
-
-      c->key_list = list;
-      c->key_list_capacity = capacity;
-    }
+  if (array_extend(c->key_list, struct key_node, 1, ARRAY_EXTEND_TWICE) == -1)
+    return MEMCACHED_FAILURE;
 
   if (state->key_tail != -1)
-    c->key_list[state->key_tail].next = c->key_list_count;
+    array_elem(c->key_list, struct key_node, state->key_tail)->next =
+      array_size(c->key_list);
   else
-    state->key_head = c->key_list_count;
-
-  state->key_tail = c->key_list_count;
-
-  node = &c->key_list[state->key_tail];
+    state->key_head = array_size(c->key_list);
+
+  state->key_tail = array_size(c->key_list);
+
+  node = array_elem(c->key_list, struct key_node, state->key_tail);
   node->key = key_index;
   node->next = -1;
 
-  ++c->key_list_count;
+  array_push(c->key_list);
 
   return MEMCACHED_SUCCESS;
@@ -1615 +1600 @@
 client_reset_for_command(struct client *c)
 {
-  c->key_list_count = 0;
+  array_clear(c->key_list);
   ++c->generation;
 }
@@ -1635 +1620 @@
      + sizeof(" " FLAGS_STUB " " EXPTIME_STUB " " VALUE_SIZE_STUB
               " " NOREPLY "\r\n"));
+  struct server *s;
   struct command_state *state;
   struct iovec *buf_iov;
   char *buf;
-  int server_index, res;
+  int res;
 
   client_reset_for_command(c);
 
-  server_index = client_get_server_index(c, key, key_len);
-  if (server_index == -1)
+  s = client_get_server(c, key, key_len);
+  if (! s)
     return MEMCACHED_CLOSED;
 
-  state = &c->servers[server_index].cmd_state;
+  state = &s->cmd_state;
   use_noreply = (noreply && state->noreply);
   command_state_reset(state, (c->prefix_len ? 2 : 1), 1,
@@ -1712 +1698 @@
      + sizeof(" " FLAGS_STUB " " EXPTIME_STUB " " VALUE_SIZE_STUB
               " " CAS_STUB " " NOREPLY "\r\n"));
+  struct server *s;
   struct command_state *state;
   struct iovec *buf_iov;
   char *buf;
-  int server_index, res;
+  int res;
 
   client_reset_for_command(c);
 
-  server_index = client_get_server_index(c, key, key_len);
-  if (server_index == -1)
+  s = client_get_server(c, key, key_len);
+  if (! s)
     return MEMCACHED_CLOSED;
 
-  state = &c->servers[server_index].cmd_state;
+  state = &s->cmd_state;
   use_noreply = (noreply && state->noreply);
   command_state_reset(state, (c->prefix_len ? 2 : 1), 1,
@@ -1762 +1749 @@
 {
   size_t request_size = (sizeof(struct iovec) * (c->prefix_len ? 4 : 3));
+  struct server *s;
   struct command_state *state;
-  int server_index, res;
+  int res;
 
   client_reset_for_command(c);
 
-  server_index = client_get_server_index(c, key, key_len);
-  if (server_index == -1)
+  s = client_get_server(c, key, key_len);
+  if (! s)
     return MEMCACHED_CLOSED;
 
-  state = &c->servers[server_index].cmd_state;
+  state = &s->cmd_state;
   command_state_reset(state, (c->prefix_len ? 2 : 1), 1, parse_get_reply);
   value_state_reset(&state->u.value, o, (cmd == CMD_GETS));
@@ -1809 +1797 @@
   size_t min_request_size =
     (sizeof(struct iovec) * ((c->prefix_len ? 3 : 2) + 2));
+  struct server *s;
   struct command_state *state;
   int i;
@@ -1819 +1808 @@
       size_t key_len;
       size_t size;
-      int server_index, res;
+      int res;
 
       key = get_key(o->arg, i, &key_len);
 
-      server_index = client_get_server_index(c, key, key_len);
-      if (server_index == -1)
+      s = client_get_server(c, key, key_len);
+      if (! s)
         continue;
 
-      state = &c->servers[server_index].cmd_state;
+      state = &s->cmd_state;
 
       if (is_active(state))
@@ -1873 +1862 @@
     }
 
-  for (i = 0; i < c->server_count; ++i)
-    {
-      state = &c->servers[i].cmd_state;
+  for (s = array_beg(c->servers, struct server);
+       s != array_end(c->servers, struct server);
+       ++s)
+    {
+      state = &s->cmd_state;
 
       if (is_active(state))
@@ -1898 +1889 @@
     (sizeof(struct iovec) * (c->prefix_len ? 4 : 3)
      + sizeof(" " ARITH_STUB " " NOREPLY "\r\n"));
+  struct server *s;
   struct command_state *state;
   struct iovec *buf_iov;
   char *buf;
-  int server_index, res;
+  int res;
 
   client_reset_for_command(c);
 
-  server_index = client_get_server_index(c, key, key_len);
-  if (server_index == -1)
+  s = client_get_server(c, key, key_len);
+  if (! s)
     return MEMCACHED_CLOSED;
 
-  state = &c->servers[server_index].cmd_state;
+  state = &s->cmd_state;
   use_noreply = (noreply && state->noreply);
   command_state_reset(state, (c->prefix_len ? 2 : 1), 1,
@@ -1958 +1950 @@
     (sizeof(struct iovec) * (c->prefix_len ? 4 : 3)
      + sizeof(" " DELAY_STUB " " NOREPLY "\r\n"));
+  struct server *s;
   struct command_state *state;
   struct iovec *buf_iov;
   char *buf;
-  int server_index, res;
+  int res;
 
   client_reset_for_command(c);
 
-  server_index = client_get_server_index(c, key, key_len);
-  if (server_index == -1)
+  s = client_get_server(c, key, key_len);
+  if (! s)
     return MEMCACHED_CLOSED;
 
-  state = &c->servers[server_index].cmd_state;
+  state = &s->cmd_state;
   use_noreply = (noreply && state->noreply);
   command_state_reset(state, (c->prefix_len ? 2 : 1), 1,
@@ -2005 +1998 @@
     (sizeof(struct iovec) * 1
      + sizeof("flush_all " DELAY_STUB " " NOREPLY "\r\n"));
+  struct server *s;
   double ddelay = delay, delay_step = 0.0;
   int i;
@@ -2010 +2004 @@
   client_reset_for_command(c);
 
-  if (c->server_count > 1)
-    delay_step = ddelay / (c->server_count - 1);
+  if (array_size(c->servers) > 1)
+    delay_step = ddelay / (array_size(c->servers) - 1);
   ddelay += delay_step;
 
-  for (i = 0; i < c->server_count; ++i)
+  for (s = array_beg(c->servers, struct server);
+       s != array_end(c->servers, struct server);
+       ++s)
     {
       int use_noreply;
@@ -2024 +2020 @@
       ddelay -= delay_step;
 
-      fd = get_server_fd(c, i);
+      fd = get_server_fd(c, s);
       if (fd == -1)
         continue;
 
-      state = &c->servers[i].cmd_state;
+      state = &s->cmd_state;
       use_noreply = (noreply && state->noreply);
       command_state_reset(state, 0, 0,
@@ -2056 +2052 @@
 client_nowait_push(struct client *c)
 {
+  struct server *s;
   int i;
 
@@ -2063 +2060 @@
   client_reset_for_command(c);
 
-  for (i = 0; i < c->server_count; ++i)
+  for (s = array_beg(c->servers, struct server);
+       s != array_end(c->servers, struct server);
+       ++s)
     {
       struct command_state *state;
       int fd, res;
 
-      state = &c->servers[i].cmd_state;
+      state = &s->cmd_state;
       if (state->nowait_count == 0)
         continue;
 
-      fd = get_server_fd(c, i);
+      fd = get_server_fd(c, s);
       if (fd == -1)
         continue;
@@ -2092 +2091 @@
 {
   static const size_t request_size = (sizeof(struct iovec) * 1);
+  struct server *s;
   int i;
 
   client_reset_for_command(c);
 
-  for (i = 0; i < c->server_count; ++i)
+  for (s = array_beg(c->servers, struct server);
+       s != array_end(c->servers, struct server);
+       ++s)
     {
       struct command_state *state;
       int fd, res;
 
-      fd = get_server_fd(c, i);
+      fd = get_server_fd(c, s);
       if (fd == -1)
         continue;
 
-      state = &c->servers[i].cmd_state;
+      state = &s->cmd_state;
       command_state_reset(state, 0, 0, parse_version_reply);
       embedded_state_reset(&state->u.embedded, o);

src/dispatch_key.c

@@ -39 +39 @@
 
 
-struct dispatch_continuum_point
+struct continuum_point
 {
   unsigned int point;
@@ -46 +46 @@
 
 
-static inline
-int
-extend_bins(struct dispatch_state *state, int add)
-{
-  int capacity =
-    (state->bins_capacity > 0 ? state->bins_capacity + add : add);
-  struct dispatch_continuum_point *b =
-    (struct dispatch_continuum_point *)
-      realloc(state->bins,
-              capacity * sizeof(struct dispatch_continuum_point));
-
-  if (! b)
-    return -1;
-
-  state->bins = b;
-  state->bins_capacity = capacity;
-
-  return 0;
-}
-
-
 static
-int
+struct continuum_point *
 dispatch_find_bin(struct dispatch_state *state, unsigned int point)
 {
-  struct dispatch_continuum_point *left, *right;
-
-  left = state->bins;
-  right = state->bins + state->bins_count;
+  struct continuum_point *beg, *end, *left, *right;
+
+  beg = left = array_beg(state->bins, struct continuum_point);
+  end = right = array_end(state->bins, struct continuum_point);
 
   while (left < right)
     {
-      struct dispatch_continuum_point *middle = left + (right - left) / 2;
+      struct continuum_point *middle = left + (right - left) / 2;
       if (middle->point < point)
         left = middle + 1;
@@ -84 +63 @@
         right = middle;
       else
-        return (middle - state->bins);
+        return middle;
     }
 
   /* Wrap around.  */
-  if (left == state->bins + state->bins_count)
-    left = state->bins;
-
-  return (left - state->bins);
+  if (left == end)
+    left = beg;
+
+  return left;
 }
 
@@ -106 +85 @@
   int i;
   double scale;
-
-  if (state->bins_count == state->bins_capacity)
-    {
-      int res = extend_bins(state, 1);
-      if (res == -1)
-        return -1;
-    }
+  struct continuum_point *p;
+
+  if (array_extend(state->bins, struct continuum_point,
+                   1, ARRAY_EXTEND_EXACT) == -1)
+    return -1;
 
   state->total_weight += weight;
   scale = (1 - weight / state->total_weight);
-  for (i = 0; i < state->bins_count; ++i)
-    state->bins[i].point = ((double) state->bins[i].point * scale + 0.5);
-
-  state->bins[state->bins_count].point = DISPATCH_MAX_POINT;
-  state->bins[state->bins_count].index = index;
-
-  ++state->bins_count;
+  for (p = array_beg(state->bins, struct continuum_point);
+       p != array_end(state->bins, struct continuum_point);
+       ++p)
+    p->point = ((double) p->point * scale + 0.5);
+
+  p->point = DISPATCH_MAX_POINT;
+  p->index = index;
+  array_push(state->bins);
 
   return 0;
@@ -143 +121 @@
     server index.
   */
-  int bin;
+  struct continuum_point *p;
   unsigned int crc32 = compute_crc32(key, key_len);
   unsigned int hash = ((crc32 >> 16) & 0x00007fff);
@@ -155 +133 @@
   point += 1;
 
-  bin = dispatch_find_bin(state, point);
-  return state->bins[bin].index;
+  p = dispatch_find_bin(state, point);
+  return p->index;
 }
 
@@ -173 +151 @@
   count = (state->ketama_points * weight + 0.5);
 
-  if (state->bins_count + count > state->bins_capacity)
-    {
-      int add = state->bins_count + count - state->bins_capacity;
-      int res = extend_bins(state, add);
-      if (res == -1)
-        return -1;
-    }
+  if (array_extend(state->bins, struct continuum_point,
+                   count, ARRAY_EXTEND_EXACT) == -1)
+    return -1;
 
   crc32 = compute_crc32(host, host_len);
@@ -189 +163 @@
       char buf[4];
       unsigned int point;
-      int bin;
+      struct continuum_point *p;
 
       /*
@@ -202 +176 @@
       point = compute_crc32_add(crc32, buf, 4);
 
-      if (state->bins_count > 0)
+      if (! array_empty(state->bins))
         {
-          bin = dispatch_find_bin(state, point);
+          p = dispatch_find_bin(state, point);
 
           /* Check if we wrapped around but actually have new max point.  */
-          if (bin == 0 && point > state->bins[0].point)
+          if (p == array_beg(state->bins, struct continuum_point)
+              && point > p->point)
             {
-              bin = state->bins_count;
+              p = array_end(state->bins, struct continuum_point);
             }
           else
             {
-              if (point == state->bins[bin].point)
+              if (point == p->point)
                 {
                   /*
@@ -222 +197 @@
                     distribution.
                   */
-                  ++bin;
+                  ++p;
                 }
 
               /* Move the tail one position forward.  */
-              memmove(state->bins + bin + 1, state->bins + bin,
-                      (state->bins_count - bin) * sizeof(*state->bins));
+              memmove(p + 1, p,
+                      ((array_end(state->bins, struct continuum_point) - p)
+                       * sizeof(*p)));
             }
         }
       else
         {
-          bin = 0;
+          p = array_beg(state->bins, struct continuum_point);
         }
 
-      state->bins[bin].point = point;
-      state->bins[bin].index = index;
-
-      ++state->bins_count;
+      p->point = point;
+      p->index = index;
+      array_push(state->bins);
     }
 
@@ -251 +226 @@
 {
   unsigned int point = compute_crc32(key, key_len);
-  int bin = dispatch_find_bin(state, point);
-  return state->bins[bin].index;
+  struct continuum_point *p = dispatch_find_bin(state, point);
+  return p->index;
 }
 
@@ -259 +234 @@
 dispatch_init(struct dispatch_state *state)
 {
-  state->bins = NULL;
-  state->bins_count = state->bins_capacity = 0;
+  array_init(&state->bins);
   state->total_weight = 0.0;
   state->ketama_points = 0;
@@ -269 +243 @@
 dispatch_destroy(struct dispatch_state *state)
 {
-  free(state->bins);
+  array_destroy(&state->bins);
 }
 
@@ -297 +271 @@
 dispatch_key(struct dispatch_state *state, const char *key, size_t key_len)
 {
-  if (state->bins_count == 0)
+  if (array_empty(state->bins))
     return -1;
 
-  if (state->bins_count == 1)
-    {
-      return state->bins[0].index;
+  if (array_size(state->bins) == 1)
+    {
+      struct continuum_point *p =
+        array_beg(state->bins, struct continuum_point);
+      return p->index;
     }
   else

src/dispatch_key.h

@@ -25 +25 @@
 #define DISPATCH_KEY_H 1
 
+#include "array.h"
 #include <stddef.h>
-
-
-struct dispatch_continuum_point;
 
 
 struct dispatch_state
 {
-  struct dispatch_continuum_point *bins;
-  int bins_count;
-  int bins_capacity;
+  struct array bins;
   double total_weight;
   int ketama_points;