git.m6w6.name Git - m6w6/libmemcached/blob - libhashkit/ketama.c

   1 /* HashKit
   2  * Copyright (C) 2006-2009 Brian Aker
   3  * All rights reserved.
   4  *
   5  * Use and distribution licensed under the BSD license.  See
   6  * the COPYING file in the parent directory for full text.
   7  */
   8
   9 #include "common.h"
  10
  11 static uint32_t ketama_server_hash(const char *key, unsigned int key_length, int alignment)
  12 {
  13   unsigned char results[16];
  14
  15   md5_signature((unsigned char*)key, key_length, results);
  16   return ((uint32_t) (results[3 + alignment * 4] & 0xFF) << 24)
  17     | ((uint32_t) (results[2 + alignment * 4] & 0xFF) << 16)
  18     | ((uint32_t) (results[1 + alignment * 4] & 0xFF) << 8)
  19     | (results[0 + alignment * 4] & 0xFF);
  20 }
  21
  22 static int continuum_points_cmp(const void *t1, const void *t2)
  23 {
  24   hashkit_continuum_point_st *ct1= (hashkit_continuum_point_st *)t1;
  25   hashkit_continuum_point_st *ct2= (hashkit_continuum_point_st *)t2;
  26
  27   if (ct1->value == ct2->value)
  28     return 0;
  29   else if (ct1->value > ct2->value)
  30     return 1;
  31   else
  32     return -1;
  33 }
  34
  35 int update_continuum(hashkit_st *hashkit)
  36 {
  37   uint32_t count;
  38   uint32_t continuum_index= 0;
  39   uint32_t value;
  40   uint32_t points_index;
  41   uint32_t points_count= 0;
  42   uint32_t points_per_server;
  43   uint32_t points_per_hash;
  44   uint64_t total_weight= 0;
  45   uint32_t live_servers;
  46   uint8_t *context;
  47
  48   if (hashkit->active_fn != NULL || hashkit->weight_fn != NULL)
  49   {
  50     live_servers= 0;
  51
  52     for (count= 0, context= hashkit->list; count < hashkit->list_size;
  53          count++, context+= hashkit->context_size)
  54     {
  55       if (hashkit->active_fn != NULL)
  56       {
  57         if (hashkit->active_fn(context))
  58           live_servers++;
  59         else
  60           continue;
  61       }
  62
  63       if (hashkit->weight_fn != NULL)
  64         total_weight+= hashkit->weight_fn(context);
  65     }
  66   }
  67
  68   if (hashkit->active_fn == NULL)
  69     live_servers= (uint32_t)hashkit->list_size;
  70
  71   if (live_servers == 0)
  72     return 0;
  73
  74   if (hashkit->weight_fn == NULL)
  75   {
  76     points_per_server= HASHKIT_POINTS_PER_NODE;
  77     points_per_hash= 1;
  78   }
  79   else
  80   {
  81     points_per_server= HASHKIT_POINTS_PER_NODE_WEIGHTED;
  82     points_per_hash= 4;
  83   }
  84
  85   if (live_servers > hashkit->continuum_count)
  86   {
  87     hashkit_continuum_point_st *new_continuum;
  88
  89     new_continuum= realloc(hashkit->continuum,
  90                            sizeof(hashkit_continuum_point_st) *
  91                            (live_servers + HASHKIT_CONTINUUM_ADDITION) *
  92                            points_per_server);
  93
  94     if (new_continuum == NULL)
  95       return ENOMEM;
  96
  97     hashkit->continuum= new_continuum;
  98     hashkit->continuum_count= live_servers + HASHKIT_CONTINUUM_ADDITION;
  99   }
 100
 101   for (count= 0, context= hashkit->list; count < hashkit->list_size;
 102        count++, context+= hashkit->context_size)
 103   {
 104     if (hashkit->active_fn != NULL && hashkit->active_fn(context) == false)
 105       continue;
 106
 107     if (hashkit->weight_fn != NULL)
 108     {
 109         float pct = (float)hashkit->weight_fn(context) / (float)total_weight;
 110         points_per_server= (uint32_t) ((floorf((float) (pct * HASHKIT_POINTS_PER_NODE_WEIGHTED / 4 * (float)live_servers + 0.0000000001))) * 4);
 111     }
 112
 113     for (points_index= 0;
 114          points_index < points_per_server / points_per_hash;
 115          points_index++)
 116     {
 117       char sort_host[HASHKIT_CONTINUUM_KEY_SIZE]= "";
 118       size_t sort_host_length;
 119
 120       if (hashkit->continuum_key_fn == NULL)
 121       {
 122         sort_host_length= (size_t) snprintf(sort_host, HASHKIT_CONTINUUM_KEY_SIZE, "%u",
 123                                             points_index);
 124       }
 125       else
 126       {
 127         sort_host_length= hashkit->continuum_key_fn(sort_host, HASHKIT_CONTINUUM_KEY_SIZE,
 128                                                  points_index, context);
 129       }
 130
 131       if (hashkit->weight_fn == NULL)
 132       {
 133         if (hashkit->continuum_hash_fn == NULL)
 134           value= hashkit_default(sort_host, sort_host_length);
 135         else
 136           value= hashkit->continuum_hash_fn(sort_host, sort_host_length);
 137
 138         hashkit->continuum[continuum_index].index= count;
 139         hashkit->continuum[continuum_index++].value= value;
 140       }
 141       else
 142       {
 143         unsigned int i;
 144         for (i = 0; i < points_per_hash; i++)
 145         {
 146            value= ketama_server_hash(sort_host, (uint32_t) sort_host_length, (int) i);
 147            hashkit->continuum[continuum_index].index= count;
 148            hashkit->continuum[continuum_index++].value= value;
 149         }
 150       }
 151     }
 152
 153     points_count+= points_per_server;
 154   }
 155
 156   hashkit->continuum_points_count= points_count;
 157   qsort(hashkit->continuum, hashkit->continuum_points_count, sizeof(hashkit_continuum_point_st),
 158         continuum_points_cmp);
 159
 160   return 0;
 161 }