text: 
  /*
   * Copyright (C) Roman Arutyunyan
   * Copyright (C) Nginx, Inc.
   */
  
  
  #include <ngx_config.h>
  #include <ngx_core.h>
  #include <ngx_stream.h>
  
  
  typedef struct {
      uint32_t                              hash;
      ngx_str_t                            *server;
  } ngx_stream_upstream_chash_point_t;
  
  
  typedef struct {
      ngx_uint_t                            number;
      ngx_stream_upstream_chash_point_t     point[1];
  } ngx_stream_upstream_chash_points_t;
  
  
  typedef struct {
      ngx_stream_complex_value_t            key;
      ngx_stream_upstream_chash_points_t   *points;
  } ngx_stream_upstream_hash_srv_conf_t;
  
  
  typedef struct {
      /* the round robin data must be first */
      ngx_stream_upstream_rr_peer_data_t    rrp;
      ngx_stream_upstream_hash_srv_conf_t  *conf;
      ngx_str_t                             key;
      ngx_uint_t                            tries;
      ngx_uint_t                            rehash;
      uint32_t                              hash;
      ngx_event_get_peer_pt                 get_rr_peer;
  } ngx_stream_upstream_hash_peer_data_t;
  
  
  static ngx_int_t ngx_stream_upstream_init_hash(ngx_conf_t *cf,
      ngx_stream_upstream_srv_conf_t *us);
  static ngx_int_t ngx_stream_upstream_init_hash_peer(ngx_stream_session_t *s,
      ngx_stream_upstream_srv_conf_t *us);
  static ngx_int_t ngx_stream_upstream_get_hash_peer(ngx_peer_connection_t *pc,
      void *data);
  
  static ngx_int_t ngx_stream_upstream_init_chash(ngx_conf_t *cf,
      ngx_stream_upstream_srv_conf_t *us);
  static int ngx_libc_cdecl
      ngx_stream_upstream_chash_cmp_points(const void *one, const void *two);
  static ngx_uint_t ngx_stream_upstream_find_chash_point(
      ngx_stream_upstream_chash_points_t *points, uint32_t hash);
  static ngx_int_t ngx_stream_upstream_init_chash_peer(ngx_stream_session_t *s,
      ngx_stream_upstream_srv_conf_t *us);
  static ngx_int_t ngx_stream_upstream_get_chash_peer(ngx_peer_connection_t *pc,
      void *data);
  
  static void *ngx_stream_upstream_hash_create_conf(ngx_conf_t *cf);
  static char *ngx_stream_upstream_hash(ngx_conf_t *cf, ngx_command_t *cmd,
      void *conf);
  
  
  static ngx_command_t  ngx_stream_upstream_hash_commands[] = {
  
      { ngx_string("hash"),
        NGX_STREAM_UPS_CONF|NGX_CONF_TAKE12,
        ngx_stream_upstream_hash,
        NGX_STREAM_SRV_CONF_OFFSET,
        0,
        NULL },
  
        ngx_null_command
  };
  
  
  static ngx_stream_module_t  ngx_stream_upstream_hash_module_ctx = {
      NULL,                                  /* preconfiguration */
      NULL,                                  /* postconfiguration */
  
      NULL,                                  /* create main configuration */
      NULL,                                  /* init main configuration */
  
      ngx_stream_upstream_hash_create_conf,  /* create server configuration */
      NULL                                   /* merge server configuration */
  };
  
  
  ngx_module_t  ngx_stream_upstream_hash_module = {
      NGX_MODULE_V1,
      &ngx_stream_upstream_hash_module_ctx,  /* module context */
      ngx_stream_upstream_hash_commands,     /* module directives */
      NGX_STREAM_MODULE,                     /* module type */
      NULL,                                  /* init master */
      NULL,                                  /* init module */
      NULL,                                  /* init process */
      NULL,                                  /* init thread */
      NULL,                                  /* exit thread */
      NULL,                                  /* exit process */
      NULL,                                  /* exit master */
      NGX_MODULE_V1_PADDING
  };
  
  
  static ngx_int_t
  ngx_stream_upstream_init_hash(ngx_conf_t *cf,
      ngx_stream_upstream_srv_conf_t *us)
  {
      if (ngx_stream_upstream_init_round_robin(cf, us) != NGX_OK) {
          return NGX_ERROR;
      }
  
      us->peer.init = ngx_stream_upstream_init_hash_peer;
  
      return NGX_OK;
  }
  
  
  static ngx_int_t
  ngx_stream_upstream_init_hash_peer(ngx_stream_session_t *s,
      ngx_stream_upstream_srv_conf_t *us)
  {
      ngx_stream_upstream_hash_srv_conf_t   *hcf;
      ngx_stream_upstream_hash_peer_data_t  *hp;
  
      hp = ngx_palloc(s->connection->pool,
                      sizeof(ngx_stream_upstream_hash_peer_data_t));
      if (hp == NULL) {
          return NGX_ERROR;
      }
  
      s->upstream->peer.data = &hp->rrp;
  
      if (ngx_stream_upstream_init_round_robin_peer(s, us) != NGX_OK) {
          return NGX_ERROR;
      }
  
      s->upstream->peer.get = ngx_stream_upstream_get_hash_peer;
  
      hcf = ngx_stream_conf_upstream_srv_conf(us,
                                              ngx_stream_upstream_hash_module);
  
      if (ngx_stream_complex_value(s, &hcf->key, &hp->key) != NGX_OK) {
          return NGX_ERROR;
      }
  
      ngx_log_debug1(NGX_LOG_DEBUG_STREAM, s->connection->log, 0,
                     "upstream hash key:\"%V\"", &hp->key);
  
      hp->conf = hcf;
      hp->tries = 0;
      hp->rehash = 0;
      hp->hash = 0;
      hp->get_rr_peer = ngx_stream_upstream_get_round_robin_peer;
  
      return NGX_OK;
  }
  
  
  static ngx_int_t
  ngx_stream_upstream_get_hash_peer(ngx_peer_connection_t *pc, void *data)
  {
      ngx_stream_upstream_hash_peer_data_t *hp = data;
  
      time_t                          now;
      u_char                          buf[NGX_INT_T_LEN];
      size_t                          size;
      uint32_t                        hash;
      ngx_int_t                       w;
      uintptr_t                       m;
      ngx_uint_t                      n, p;
      ngx_stream_upstream_rr_peer_t  *peer;
  
      ngx_log_debug1(NGX_LOG_DEBUG_STREAM, pc->log, 0,
                     "get hash peer, try: %ui", pc->tries);
  
      ngx_stream_upstream_rr_peers_rlock(hp->rrp.peers);
  
      if (hp->tries > 20 || hp->rrp.peers->single || hp->key.len == 0) {
          ngx_stream_upstream_rr_peers_unlock(hp->rrp.peers);
          return hp->get_rr_peer(pc, &hp->rrp);
      }
  
      now = ngx_time();
  
      pc->connection = NULL;
  
      for ( ;; ) {
  
          /*
           * Hash expression is compatible with Cache::Memcached:
           * ((crc32([REHASH] KEY) >> 16) & 0x7fff) + PREV_HASH
           * with REHASH omitted at the first iteration.
           */
  
          ngx_crc32_init(hash);
  
          if (hp->rehash > 0) {
              size = ngx_sprintf(buf, "%ui", hp->rehash) - buf;
              ngx_crc32_update(&hash, buf, size);
          }
  
          ngx_crc32_update(&hash, hp->key.data, hp->key.len);
          ngx_crc32_final(hash);
  
          hash = (hash >> 16) & 0x7fff;
  
          hp->hash += hash;
          hp->rehash++;
  
          w = hp->hash % hp->rrp.peers->total_weight;
          peer = hp->rrp.peers->peer;
          p = 0;
  
          while (w >= peer->weight) {
              w -= peer->weight;
              peer = peer->next;
              p++;
          }
  
          n = p / (8 * sizeof(uintptr_t));
          m = (uintptr_t) 1 << p % (8 * sizeof(uintptr_t));
  
          if (hp->rrp.tried[n] & m) {
              goto next;
          }
  
          ngx_stream_upstream_rr_peer_lock(hp->rrp.peers, peer);
  
          ngx_log_debug2(NGX_LOG_DEBUG_STREAM, pc->log, 0,
                         "get hash peer, value:%uD, peer:%ui", hp->hash, p);
  
          if (peer->down) {
              ngx_stream_upstream_rr_peer_unlock(hp->rrp.peers, peer);
              goto next;
          }
  
          if (peer->max_fails
              && peer->fails >= peer->max_fails
              && now - peer->checked <= peer->fail_timeout)
          {
              ngx_stream_upstream_rr_peer_unlock(hp->rrp.peers, peer);
              goto next;
          }
  
          if (peer->max_conns && peer->conns >= peer->max_conns) {
              ngx_stream_upstream_rr_peer_unlock(hp->rrp.peers, peer);
              goto next;
          }
  
          break;
  
      next:
  
          if (++hp->tries > 20) {
              ngx_stream_upstream_rr_peers_unlock(hp->rrp.peers);
              return hp->get_rr_peer(pc, &hp->rrp);
          }
      }
  
      hp->rrp.current = peer;
  
      pc->sockaddr = peer->sockaddr;
      pc->socklen = peer->socklen;
      pc->name = &peer->name;
  
      peer->conns++;
  
      if (now - peer->checked > peer->fail_timeout) {
          peer->checked = now;
      }
  
      ngx_stream_upstream_rr_peer_unlock(hp->rrp.peers, peer);
      ngx_stream_upstream_rr_peers_unlock(hp->rrp.peers);
  
      hp->rrp.tried[n] |= m;
  
      return NGX_OK;
  }
  
  
  static ngx_int_t
  ngx_stream_upstream_init_chash(ngx_conf_t *cf,
      ngx_stream_upstream_srv_conf_t *us)
  {
      u_char                               *host, *port, c;
      size_t                                host_len, port_len, size;
      uint32_t                              hash, base_hash;
      ngx_str_t                            *server;
      ngx_uint_t                            npoints, i, j;
      ngx_stream_upstream_rr_peer_t        *peer;
      ngx_stream_upstream_rr_peers_t       *peers;
      ngx_stream_upstream_chash_points_t   *points;
      ngx_stream_upstream_hash_srv_conf_t  *hcf;
      union {
          uint32_t                          value;
          u_char                            byte[4];
      } prev_hash;
  
      if (ngx_stream_upstream_init_round_robin(cf, us) != NGX_OK) {
          return NGX_ERROR;
      }
  
      us->peer.init = ngx_stream_upstream_init_chash_peer;
  
      peers = us->peer.data;
      npoints = peers->total_weight * 160;
  
      size = sizeof(ngx_stream_upstream_chash_points_t)
             + sizeof(ngx_stream_upstream_chash_point_t) * (npoints - 1);
  
      points = ngx_palloc(cf->pool, size);
      if (points == NULL) {
          return NGX_ERROR;
      }
  
      points->number = 0;
  
      for (peer = peers->peer; peer; peer = peer->next) {
          server = &peer->server;
  
          /*
           * Hash expression is compatible with Cache::Memcached::Fast:
           * crc32(HOST \0 PORT PREV_HASH).
           */
  
          if (server->len >= 5
              && ngx_strncasecmp(server->data, (u_char *) "unix:", 5) == 0)
          {
              host = server->data + 5;
              host_len = server->len - 5;
              port = NULL;
              port_len = 0;
              goto done;
          }
  
          for (j = 0; j < server->len; j++) {
              c = server->data[server->len - j - 1];
  
              if (c == ':') {
                  host = server->data;
                  host_len = server->len - j - 1;
                  port = server->data + server->len - j;
                  port_len = j;
                  goto done;
              }
  
              if (c < '0' || c > '9') {
                  break;
              }
          }
  
          host = server->data;
          host_len = server->len;
          port = NULL;
          port_len = 0;
  
      done:
  
          ngx_crc32_init(base_hash);
          ngx_crc32_update(&base_hash, host, host_len);
          ngx_crc32_update(&base_hash, (u_char *) "", 1);
          ngx_crc32_update(&base_hash, port, port_len);
  
          prev_hash.value = 0;
          npoints = peer->weight * 160;
  
          for (j = 0; j < npoints; j++) {
              hash = base_hash;
  
              ngx_crc32_update(&hash, prev_hash.byte, 4);
              ngx_crc32_final(hash);
  
              points->point[points->number].hash = hash;
              points->point[points->number].server = server;
              points->number++;
  
  #if (NGX_HAVE_LITTLE_ENDIAN)
              prev_hash.value = hash;
  #else
              prev_hash.byte[0] = (u_char) (hash & 0xff);
              prev_hash.byte[1] = (u_char) ((hash >> 8) & 0xff);
              prev_hash.byte[2] = (u_char) ((hash >> 16) & 0xff);
              prev_hash.byte[3] = (u_char) ((hash >> 24) & 0xff);
  #endif
          }
      }
  
      ngx_qsort(points->point,
                points->number,
                sizeof(ngx_stream_upstream_chash_point_t),
                ngx_stream_upstream_chash_cmp_points);
  
      for (i = 0, j = 1; j < points->number; j++) {
          if (points->point[i].hash != points->point[j].hash) {
              points->point[++i] = points->point[j];
          }
      }
  
      points->number = i + 1;
  
      hcf = ngx_stream_conf_upstream_srv_conf(us,
                                              ngx_stream_upstream_hash_module);
      hcf->points = points;
  
      return NGX_OK;
  }
  
  
  static int ngx_libc_cdecl
  ngx_stream_upstream_chash_cmp_points(const void *one, const void *two)
  {
      ngx_stream_upstream_chash_point_t *first =
                                       (ngx_stream_upstream_chash_point_t *) one;
      ngx_stream_upstream_chash_point_t *second =
                                       (ngx_stream_upstream_chash_point_t *) two;
  
      if (first->hash < second->hash) {
          return -1;
  
      } else if (first->hash > second->hash) {
          return 1;
  
      } else {
          return 0;
      }
  }
  
  
  static ngx_uint_t
  ngx_stream_upstream_find_chash_point(ngx_stream_upstream_chash_points_t *points,
      uint32_t hash)
  {
      ngx_uint_t                          i, j, k;
      ngx_stream_upstream_chash_point_t  *point;
  
      /* find first point >= hash */
  
      point = &points->point[0];
  
      i = 0;
      j = points->number;
  
      while (i < j) {
          k = (i + j) / 2;
  
          if (hash > point[k].hash) {
              i = k + 1;
  
          } else if (hash < point[k].hash) {
              j = k;
  
          } else {
              return k;
          }
      }
  
      return i;
  }
  
  
  static ngx_int_t
  ngx_stream_upstream_init_chash_peer(ngx_stream_session_t *s,
      ngx_stream_upstream_srv_conf_t *us)
  {
      uint32_t                               hash;
      ngx_stream_upstream_hash_srv_conf_t   *hcf;
      ngx_stream_upstream_hash_peer_data_t  *hp;
  
      if (ngx_stream_upstream_init_hash_peer(s, us) != NGX_OK) {
          return NGX_ERROR;
      }
  
      s->upstream->peer.get = ngx_stream_upstream_get_chash_peer;
  
      hp = s->upstream->peer.data;
      hcf = ngx_stream_conf_upstream_srv_conf(us,
                                              ngx_stream_upstream_hash_module);
  
      hash = ngx_crc32_long(hp->key.data, hp->key.len);
  
      ngx_stream_upstream_rr_peers_rlock(hp->rrp.peers);
  
      hp->hash = ngx_stream_upstream_find_chash_point(hcf->points, hash);
  
      ngx_stream_upstream_rr_peers_unlock(hp->rrp.peers);
  
      return NGX_OK;
  }
  
  
  static ngx_int_t
  ngx_stream_upstream_get_chash_peer(ngx_peer_connection_t *pc, void *data)
  {
      ngx_stream_upstream_hash_peer_data_t *hp = data;
  
      time_t                                now;
      intptr_t                              m;
      ngx_str_t                            *server;
      ngx_int_t                             total;
      ngx_uint_t                            i, n, best_i;
      ngx_stream_upstream_rr_peer_t        *peer, *best;
      ngx_stream_upstream_chash_point_t    *point;
      ngx_stream_upstream_chash_points_t   *points;
      ngx_stream_upstream_hash_srv_conf_t  *hcf;
  
      ngx_log_debug1(NGX_LOG_DEBUG_STREAM, pc->log, 0,
                     "get consistent hash peer, try: %ui", pc->tries);
  
      ngx_stream_upstream_rr_peers_wlock(hp->rrp.peers);
  
      if (hp->tries > 20 || hp->rrp.peers->single || hp->key.len == 0) {
          ngx_stream_upstream_rr_peers_unlock(hp->rrp.peers);
          return hp->get_rr_peer(pc, &hp->rrp);
      }
  
      pc->connection = NULL;
  
      now = ngx_time();
      hcf = hp->conf;
  
      points = hcf->points;
      point = &points->point[0];
  
      for ( ;; ) {
          server = point[hp->hash % points->number].server;
  
          ngx_log_debug2(NGX_LOG_DEBUG_STREAM, pc->log, 0,
                         "consistent hash peer:%uD, server:\"%V\"",
                         hp->hash, server);
  
          best = NULL;
          best_i = 0;
          total = 0;
  
          for (peer = hp->rrp.peers->peer, i = 0;
               peer;
               peer = peer->next, i++)
          {
              n = i / (8 * sizeof(uintptr_t));
              m = (uintptr_t) 1 << i % (8 * sizeof(uintptr_t));
  
              if (hp->rrp.tried[n] & m) {
                  continue;
              }
  
              if (peer->down) {
                  continue;
              }
  
              if (peer->max_fails
                  && peer->fails >= peer->max_fails
                  && now - peer->checked <= peer->fail_timeout)
              {
                  continue;
              }
  
              if (peer->max_conns && peer->conns >= peer->max_conns) {
                  continue;
              }
  
              if (peer->server.len != server->len
                  || ngx_strncmp(peer->server.data, server->data, server->len)
                     != 0)
              {
                  continue;
              }
  
              peer->current_weight += peer->effective_weight;
              total += peer->effective_weight;
  
              if (peer->effective_weight < peer->weight) {
                  peer->effective_weight++;
              }
  
              if (best == NULL || peer->current_weight > best->current_weight) {
                  best = peer;
                  best_i = i;
              }
          }
  
          if (best) {
              best->current_weight -= total;
              break;
          }
  
          hp->hash++;
          hp->tries++;
  
          if (hp->tries > 20) {
              ngx_stream_upstream_rr_peers_unlock(hp->rrp.peers);
              return hp->get_rr_peer(pc, &hp->rrp);
          }
      }
  
      hp->rrp.current = best;
  
      pc->sockaddr = best->sockaddr;
      pc->socklen = best->socklen;
      pc->name = &best->name;
  
      best->conns++;
  
      if (now - best->checked > best->fail_timeout) {
          best->checked = now;
      }
  
      ngx_stream_upstream_rr_peers_unlock(hp->rrp.peers);
  
      n = best_i / (8 * sizeof(uintptr_t));
      m = (uintptr_t) 1 << best_i % (8 * sizeof(uintptr_t));
  
      hp->rrp.tried[n] |= m;
  
      return NGX_OK;
  }
  
  
  static void *
  ngx_stream_upstream_hash_create_conf(ngx_conf_t *cf)
  {
      ngx_stream_upstream_hash_srv_conf_t  *conf;
  
      conf = ngx_palloc(cf->pool, sizeof(ngx_stream_upstream_hash_srv_conf_t));
      if (conf == NULL) {
          return NULL;
      }
  
      conf->points = NULL;
  
      return conf;
  }
  
  
  static char *
  ngx_stream_upstream_hash(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
  {
      ngx_stream_upstream_hash_srv_conf_t  *hcf = conf;
  
      ngx_str_t                           *value;
      ngx_stream_upstream_srv_conf_t      *uscf;
      ngx_stream_compile_complex_value_t   ccv;
  
      value = cf->args->elts;
  
      ngx_memzero(&ccv, sizeof(ngx_stream_compile_complex_value_t));
  
      ccv.cf = cf;
      ccv.value = &value[1];
      ccv.complex_value = &hcf->key;
  
      if (ngx_stream_compile_complex_value(&ccv) != NGX_OK) {
          return NGX_CONF_ERROR;
      }
  
      uscf = ngx_stream_conf_get_module_srv_conf(cf, ngx_stream_upstream_module);
  
      if (uscf->peer.init_upstream) {
          ngx_conf_log_error(NGX_LOG_WARN, cf, 0,
                             "load balancing method redefined");
      }
  
      uscf->flags = NGX_STREAM_UPSTREAM_CREATE
                    |NGX_STREAM_UPSTREAM_WEIGHT
                    |NGX_STREAM_UPSTREAM_MAX_CONNS
                    |NGX_STREAM_UPSTREAM_MAX_FAILS
                    |NGX_STREAM_UPSTREAM_FAIL_TIMEOUT
                    |NGX_STREAM_UPSTREAM_DOWN;
  
      if (cf->args->nelts == 2) {
          uscf->peer.init_upstream = ngx_stream_upstream_init_hash;
  
      } else if (ngx_strcmp(value[2].data, "consistent") == 0) {
          uscf->peer.init_upstream = ngx_stream_upstream_init_chash;
  
      } else {
          ngx_conf_log_error(NGX_LOG_EMERG, cf, 0,
                             "invalid parameter \"%V\"", &value[2]);
          return NGX_CONF_ERROR;
      }
  
      return NGX_CONF_OK;
  }
  