1
2/*
3 * Copyright (C) Roman Arutyunyan
4 * Copyright (C) Nginx, Inc.
5 */
6
7
8#include <ngx_config.h>
9#include <ngx_core.h>
10#include <ngx_http.h>
11
12
13typedef struct {
14    uint32_t                            hash;
15    ngx_str_t                          *server;
16} ngx_http_upstream_chash_point_t;
17
18
19typedef struct {
20    ngx_uint_t                          number;
21    ngx_http_upstream_chash_point_t     point[1];
22} ngx_http_upstream_chash_points_t;
23
24
25typedef struct {
26    ngx_http_complex_value_t            key;
27    ngx_http_upstream_chash_points_t   *points;
28} ngx_http_upstream_hash_srv_conf_t;
29
30
31typedef struct {
32    /* the round robin data must be first */
33    ngx_http_upstream_rr_peer_data_t    rrp;
34    ngx_http_upstream_hash_srv_conf_t  *conf;
35    ngx_str_t                           key;
36    ngx_uint_t                          tries;
37    ngx_uint_t                          rehash;
38    uint32_t                            hash;
39    ngx_event_get_peer_pt               get_rr_peer;
40} ngx_http_upstream_hash_peer_data_t;
41
42
43static ngx_int_t ngx_http_upstream_init_hash(ngx_conf_t *cf,
44    ngx_http_upstream_srv_conf_t *us);
45static ngx_int_t ngx_http_upstream_init_hash_peer(ngx_http_request_t *r,
46    ngx_http_upstream_srv_conf_t *us);
47static ngx_int_t ngx_http_upstream_get_hash_peer(ngx_peer_connection_t *pc,
48    void *data);
49
50static ngx_int_t ngx_http_upstream_init_chash(ngx_conf_t *cf,
51    ngx_http_upstream_srv_conf_t *us);
52static int ngx_libc_cdecl
53    ngx_http_upstream_chash_cmp_points(const void *one, const void *two);
54static ngx_uint_t ngx_http_upstream_find_chash_point(
55    ngx_http_upstream_chash_points_t *points, uint32_t hash);
56static ngx_int_t ngx_http_upstream_init_chash_peer(ngx_http_request_t *r,
57    ngx_http_upstream_srv_conf_t *us);
58static ngx_int_t ngx_http_upstream_get_chash_peer(ngx_peer_connection_t *pc,
59    void *data);
60
61static void *ngx_http_upstream_hash_create_conf(ngx_conf_t *cf);
62static char *ngx_http_upstream_hash(ngx_conf_t *cf, ngx_command_t *cmd,
63    void *conf);
64
65
66static ngx_command_t  ngx_http_upstream_hash_commands[] = {
67
68    { ngx_string("hash"),
69      NGX_HTTP_UPS_CONF|NGX_CONF_TAKE12,
70      ngx_http_upstream_hash,
71      NGX_HTTP_SRV_CONF_OFFSET,
72      0,
73      NULL },
74
75      ngx_null_command
76};
77
78
79static ngx_http_module_t  ngx_http_upstream_hash_module_ctx = {
80    NULL,                                  /* preconfiguration */
81    NULL,                                  /* postconfiguration */
82
83    NULL,                                  /* create main configuration */
84    NULL,                                  /* init main configuration */
85
86    ngx_http_upstream_hash_create_conf,    /* create server configuration */
87    NULL,                                  /* merge server configuration */
88
89    NULL,                                  /* create location configuration */
90    NULL                                   /* merge location configuration */
91};
92
93
94ngx_module_t  ngx_http_upstream_hash_module = {
95    NGX_MODULE_V1,
96    &ngx_http_upstream_hash_module_ctx,    /* module context */
97    ngx_http_upstream_hash_commands,       /* module directives */
98    NGX_HTTP_MODULE,                       /* module type */
99    NULL,                                  /* init master */
100    NULL,                                  /* init module */
101    NULL,                                  /* init process */
102    NULL,                                  /* init thread */
103    NULL,                                  /* exit thread */
104    NULL,                                  /* exit process */
105    NULL,                                  /* exit master */
106    NGX_MODULE_V1_PADDING
107};
108
109
110static ngx_int_t
111ngx_http_upstream_init_hash(ngx_conf_t *cf, ngx_http_upstream_srv_conf_t *us)
112{
113    if (ngx_http_upstream_init_round_robin(cf, us) != NGX_OK) {
114        return NGX_ERROR;
115    }
116
117    us->peer.init = ngx_http_upstream_init_hash_peer;
118
119    return NGX_OK;
120}
121
122
123static ngx_int_t
124ngx_http_upstream_init_hash_peer(ngx_http_request_t *r,
125    ngx_http_upstream_srv_conf_t *us)
126{
127    ngx_http_upstream_hash_srv_conf_t   *hcf;
128    ngx_http_upstream_hash_peer_data_t  *hp;
129
130    hp = ngx_palloc(r->pool, sizeof(ngx_http_upstream_hash_peer_data_t));
131    if (hp == NULL) {
132        return NGX_ERROR;
133    }
134
135    r->upstream->peer.data = &hp->rrp;
136
137    if (ngx_http_upstream_init_round_robin_peer(r, us) != NGX_OK) {
138        return NGX_ERROR;
139    }
140
141    r->upstream->peer.get = ngx_http_upstream_get_hash_peer;
142
143    hcf = ngx_http_conf_upstream_srv_conf(us, ngx_http_upstream_hash_module);
144
145    if (ngx_http_complex_value(r, &hcf->key, &hp->key) != NGX_OK) {
146        return NGX_ERROR;
147    }
148
149    ngx_log_debug1(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,
150                   "upstream hash key:\"%V\"", &hp->key);
151
152    hp->conf = hcf;
153    hp->tries = 0;
154    hp->rehash = 0;
155    hp->hash = 0;
156    hp->get_rr_peer = ngx_http_upstream_get_round_robin_peer;
157
158    return NGX_OK;
159}
160
161
162static ngx_int_t
163ngx_http_upstream_get_hash_peer(ngx_peer_connection_t *pc, void *data)
164{
165    ngx_http_upstream_hash_peer_data_t  *hp = data;
166
167    time_t                        now;
168    u_char                        buf[NGX_INT_T_LEN];
169    size_t                        size;
170    uint32_t                      hash;
171    ngx_int_t                     w;
172    uintptr_t                     m;
173    ngx_uint_t                    n, p;
174    ngx_http_upstream_rr_peer_t  *peer;
175
176    ngx_log_debug1(NGX_LOG_DEBUG_HTTP, pc->log, 0,
177                   "get hash peer, try: %ui", pc->tries);
178
179    ngx_http_upstream_rr_peers_wlock(hp->rrp.peers);
180
181    if (hp->tries > 20 || hp->rrp.peers->single) {
182        ngx_http_upstream_rr_peers_unlock(hp->rrp.peers);
183        return hp->get_rr_peer(pc, &hp->rrp);
184    }
185
186    now = ngx_time();
187
188    pc->cached = 0;
189    pc->connection = NULL;
190
191    for ( ;; ) {
192
193        /*
194         * Hash expression is compatible with Cache::Memcached:
195         * ((crc32([REHASH] KEY) >> 16) & 0x7fff) + PREV_HASH
196         * with REHASH omitted at the first iteration.
197         */
198
199        ngx_crc32_init(hash);
200
201        if (hp->rehash > 0) {
202            size = ngx_sprintf(buf, "%ui", hp->rehash) - buf;
203            ngx_crc32_update(&hash, buf, size);
204        }
205
206        ngx_crc32_update(&hash, hp->key.data, hp->key.len);
207        ngx_crc32_final(hash);
208
209        hash = (hash >> 16) & 0x7fff;
210
211        hp->hash += hash;
212        hp->rehash++;
213
214        w = hp->hash % hp->rrp.peers->total_weight;
215        peer = hp->rrp.peers->peer;
216        p = 0;
217
218        while (w >= peer->weight) {
219            w -= peer->weight;
220            peer = peer->next;
221            p++;
222        }
223
224        n = p / (8 * sizeof(uintptr_t));
225        m = (uintptr_t) 1 << p % (8 * sizeof(uintptr_t));
226
227        if (hp->rrp.tried[n] & m) {
228            goto next;
229        }
230
231        ngx_log_debug2(NGX_LOG_DEBUG_HTTP, pc->log, 0,
232                       "get hash peer, value:%uD, peer:%ui", hp->hash, p);
233
234        if (peer->down) {
235            goto next;
236        }
237
238        if (peer->max_fails
239            && peer->fails >= peer->max_fails
240            && now - peer->checked <= peer->fail_timeout)
241        {
242            goto next;
243        }
244
245        if (peer->max_conns && peer->conns >= peer->max_conns) {
246            goto next;
247        }
248
249        break;
250
251    next:
252
253        if (++hp->tries > 20) {
254            ngx_http_upstream_rr_peers_unlock(hp->rrp.peers);
255            return hp->get_rr_peer(pc, &hp->rrp);
256        }
257    }
258
259    hp->rrp.current = peer;
260
261    pc->sockaddr = peer->sockaddr;
262    pc->socklen = peer->socklen;
263    pc->name = &peer->name;
264
265    peer->conns++;
266
267    if (now - peer->checked > peer->fail_timeout) {
268        peer->checked = now;
269    }
270
271    ngx_http_upstream_rr_peers_unlock(hp->rrp.peers);
272
273    hp->rrp.tried[n] |= m;
274
275    return NGX_OK;
276}
277
278
279static ngx_int_t
280ngx_http_upstream_init_chash(ngx_conf_t *cf, ngx_http_upstream_srv_conf_t *us)
281{
282    u_char                             *host, *port, c;
283    size_t                              host_len, port_len, size;
284    uint32_t                            hash, base_hash;
285    ngx_str_t                          *server;
286    ngx_uint_t                          npoints, i, j;
287    ngx_http_upstream_rr_peer_t        *peer;
288    ngx_http_upstream_rr_peers_t       *peers;
289    ngx_http_upstream_chash_points_t   *points;
290    ngx_http_upstream_hash_srv_conf_t  *hcf;
291    union {
292        uint32_t                        value;
293        u_char                          byte[4];
294    } prev_hash;
295
296    if (ngx_http_upstream_init_round_robin(cf, us) != NGX_OK) {
297        return NGX_ERROR;
298    }
299
300    us->peer.init = ngx_http_upstream_init_chash_peer;
301
302    peers = us->peer.data;
303    npoints = peers->total_weight * 160;
304
305    size = sizeof(ngx_http_upstream_chash_points_t)
306           + sizeof(ngx_http_upstream_chash_point_t) * (npoints - 1);
307
308    points = ngx_palloc(cf->pool, size);
309    if (points == NULL) {
310        return NGX_ERROR;
311    }
312
313    points->number = 0;
314
315    for (peer = peers->peer; peer; peer = peer->next) {
316        server = &peer->server;
317
318        /*
319         * Hash expression is compatible with Cache::Memcached::Fast:
320         * crc32(HOST \0 PORT PREV_HASH).
321         */
322
323        if (server->len >= 5
324            && ngx_strncasecmp(server->data, (u_char *) "unix:", 5) == 0)
325        {
326            host = server->data + 5;
327            host_len = server->len - 5;
328            port = NULL;
329            port_len = 0;
330            goto done;
331        }
332
333        for (j = 0; j < server->len; j++) {
334            c = server->data[server->len - j - 1];
335
336            if (c == ':') {
337                host = server->data;
338                host_len = server->len - j - 1;
339                port = server->data + server->len - j;
340                port_len = j;
341                goto done;
342            }
343
344            if (c < '0' || c > '9') {
345                break;
346            }
347        }
348
349        host = server->data;
350        host_len = server->len;
351        port = NULL;
352        port_len = 0;
353
354    done:
355
356        ngx_crc32_init(base_hash);
357        ngx_crc32_update(&base_hash, host, host_len);
358        ngx_crc32_update(&base_hash, (u_char *) "", 1);
359        ngx_crc32_update(&base_hash, port, port_len);
360
361        prev_hash.value = 0;
362        npoints = peer->weight * 160;
363
364        for (j = 0; j < npoints; j++) {
365            hash = base_hash;
366
367            ngx_crc32_update(&hash, prev_hash.byte, 4);
368            ngx_crc32_final(hash);
369
370            points->point[points->number].hash = hash;
371            points->point[points->number].server = server;
372            points->number++;
373
374#if (NGX_HAVE_LITTLE_ENDIAN)
375            prev_hash.value = hash;
376#else
377            prev_hash.byte[0] = (u_char) (hash & 0xff);
378            prev_hash.byte[1] = (u_char) ((hash >> 8) & 0xff);
379            prev_hash.byte[2] = (u_char) ((hash >> 16) & 0xff);
380            prev_hash.byte[3] = (u_char) ((hash >> 24) & 0xff);
381#endif
382        }
383    }
384
385    ngx_qsort(points->point,
386              points->number,
387              sizeof(ngx_http_upstream_chash_point_t),
388              ngx_http_upstream_chash_cmp_points);
389
390    for (i = 0, j = 1; j < points->number; j++) {
391        if (points->point[i].hash != points->point[j].hash) {
392            points->point[++i] = points->point[j];
393        }
394    }
395
396    points->number = i + 1;
397
398    hcf = ngx_http_conf_upstream_srv_conf(us, ngx_http_upstream_hash_module);
399    hcf->points = points;
400
401    return NGX_OK;
402}
403
404
405static int ngx_libc_cdecl
406ngx_http_upstream_chash_cmp_points(const void *one, const void *two)
407{
408    ngx_http_upstream_chash_point_t *first =
409                                       (ngx_http_upstream_chash_point_t *) one;
410    ngx_http_upstream_chash_point_t *second =
411                                       (ngx_http_upstream_chash_point_t *) two;
412
413    if (first->hash < second->hash) {
414        return -1;
415
416    } else if (first->hash > second->hash) {
417        return 1;
418
419    } else {
420        return 0;
421    }
422}
423
424
425static ngx_uint_t
426ngx_http_upstream_find_chash_point(ngx_http_upstream_chash_points_t *points,
427    uint32_t hash)
428{
429    ngx_uint_t                        i, j, k;
430    ngx_http_upstream_chash_point_t  *point;
431
432    /* find first point >= hash */
433
434    point = &points->point[0];
435
436    i = 0;
437    j = points->number;
438
439    while (i < j) {
440        k = (i + j) / 2;
441
442        if (hash > point[k].hash) {
443            i = k + 1;
444
445        } else if (hash < point[k].hash) {
446            j = k;
447
448        } else {
449            return k;
450        }
451    }
452
453    return i;
454}
455
456
457static ngx_int_t
458ngx_http_upstream_init_chash_peer(ngx_http_request_t *r,
459    ngx_http_upstream_srv_conf_t *us)
460{
461    uint32_t                             hash;
462    ngx_http_upstream_hash_srv_conf_t   *hcf;
463    ngx_http_upstream_hash_peer_data_t  *hp;
464
465    if (ngx_http_upstream_init_hash_peer(r, us) != NGX_OK) {
466        return NGX_ERROR;
467    }
468
469    r->upstream->peer.get = ngx_http_upstream_get_chash_peer;
470
471    hp = r->upstream->peer.data;
472    hcf = ngx_http_conf_upstream_srv_conf(us, ngx_http_upstream_hash_module);
473
474    hash = ngx_crc32_long(hp->key.data, hp->key.len);
475
476    ngx_http_upstream_rr_peers_rlock(hp->rrp.peers);
477
478    hp->hash = ngx_http_upstream_find_chash_point(hcf->points, hash);
479
480    ngx_http_upstream_rr_peers_unlock(hp->rrp.peers);
481
482    return NGX_OK;
483}
484
485
486static ngx_int_t
487ngx_http_upstream_get_chash_peer(ngx_peer_connection_t *pc, void *data)
488{
489    ngx_http_upstream_hash_peer_data_t  *hp = data;
490
491    time_t                              now;
492    intptr_t                            m;
493    ngx_str_t                          *server;
494    ngx_int_t                           total;
495    ngx_uint_t                          i, n, best_i;
496    ngx_http_upstream_rr_peer_t        *peer, *best;
497    ngx_http_upstream_chash_point_t    *point;
498    ngx_http_upstream_chash_points_t   *points;
499    ngx_http_upstream_hash_srv_conf_t  *hcf;
500
501    ngx_log_debug1(NGX_LOG_DEBUG_HTTP, pc->log, 0,
502                   "get consistent hash peer, try: %ui", pc->tries);
503
504    ngx_http_upstream_rr_peers_wlock(hp->rrp.peers);
505
506    pc->cached = 0;
507    pc->connection = NULL;
508
509    now = ngx_time();
510    hcf = hp->conf;
511
512    points = hcf->points;
513    point = &points->point[0];
514
515    for ( ;; ) {
516        server = point[hp->hash % points->number].server;
517
518        ngx_log_debug2(NGX_LOG_DEBUG_HTTP, pc->log, 0,
519                       "consistent hash peer:%uD, server:\"%V\"",
520                       hp->hash, server);
521
522        best = NULL;
523        best_i = 0;
524        total = 0;
525
526        for (peer = hp->rrp.peers->peer, i = 0;
527             peer;
528             peer = peer->next, i++)
529        {
530            n = i / (8 * sizeof(uintptr_t));
531            m = (uintptr_t) 1 << i % (8 * sizeof(uintptr_t));
532
533            if (hp->rrp.tried[n] & m) {
534                continue;
535            }
536
537            if (peer->down) {
538                continue;
539            }
540
541            if (peer->server.len != server->len
542                || ngx_strncmp(peer->server.data, server->data, server->len)
543                   != 0)
544            {
545                continue;
546            }
547
548            if (peer->max_fails
549                && peer->fails >= peer->max_fails
550                && now - peer->checked <= peer->fail_timeout)
551            {
552                continue;
553            }
554
555            if (peer->max_conns && peer->conns >= peer->max_conns) {
556                continue;
557            }
558
559            peer->current_weight += peer->effective_weight;
560            total += peer->effective_weight;
561
562            if (peer->effective_weight < peer->weight) {
563                peer->effective_weight++;
564            }
565
566            if (best == NULL || peer->current_weight > best->current_weight) {
567                best = peer;
568                best_i = i;
569            }
570        }
571
572        if (best) {
573            best->current_weight -= total;
574            goto found;
575        }
576
577        hp->hash++;
578        hp->tries++;
579
580        if (hp->tries >= points->number) {
581            pc->name = hp->rrp.peers->name;
582            ngx_http_upstream_rr_peers_unlock(hp->rrp.peers);
583            return NGX_BUSY;
584        }
585    }
586
587found:
588
589    hp->rrp.current = best;
590
591    pc->sockaddr = best->sockaddr;
592    pc->socklen = best->socklen;
593    pc->name = &best->name;
594
595    best->conns++;
596
597    if (now - best->checked > best->fail_timeout) {
598        best->checked = now;
599    }
600
601    ngx_http_upstream_rr_peers_unlock(hp->rrp.peers);
602
603    n = best_i / (8 * sizeof(uintptr_t));
604    m = (uintptr_t) 1 << best_i % (8 * sizeof(uintptr_t));
605
606    hp->rrp.tried[n] |= m;
607
608    return NGX_OK;
609}
610
611
612static void *
613ngx_http_upstream_hash_create_conf(ngx_conf_t *cf)
614{
615    ngx_http_upstream_hash_srv_conf_t  *conf;
616
617    conf = ngx_palloc(cf->pool, sizeof(ngx_http_upstream_hash_srv_conf_t));
618    if (conf == NULL) {
619        return NULL;
620    }
621
622    conf->points = NULL;
623
624    return conf;
625}
626
627
628static char *
629ngx_http_upstream_hash(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
630{
631    ngx_http_upstream_hash_srv_conf_t  *hcf = conf;
632
633    ngx_str_t                         *value;
634    ngx_http_upstream_srv_conf_t      *uscf;
635    ngx_http_compile_complex_value_t   ccv;
636
637    value = cf->args->elts;
638
639    ngx_memzero(&ccv, sizeof(ngx_http_compile_complex_value_t));
640
641    ccv.cf = cf;
642    ccv.value = &value[1];
643    ccv.complex_value = &hcf->key;
644
645    if (ngx_http_compile_complex_value(&ccv) != NGX_OK) {
646        return NGX_CONF_ERROR;
647    }
648
649    uscf = ngx_http_conf_get_module_srv_conf(cf, ngx_http_upstream_module);
650
651    if (uscf->peer.init_upstream) {
652        ngx_conf_log_error(NGX_LOG_WARN, cf, 0,
653                           "load balancing method redefined");
654    }
655
656    uscf->flags = NGX_HTTP_UPSTREAM_CREATE
657                  |NGX_HTTP_UPSTREAM_WEIGHT
658                  |NGX_HTTP_UPSTREAM_MAX_CONNS
659                  |NGX_HTTP_UPSTREAM_MAX_FAILS
660                  |NGX_HTTP_UPSTREAM_FAIL_TIMEOUT
661                  |NGX_HTTP_UPSTREAM_DOWN;
662
663    if (cf->args->nelts == 2) {
664        uscf->peer.init_upstream = ngx_http_upstream_init_hash;
665
666    } else if (ngx_strcmp(value[2].data, "consistent") == 0) {
667        uscf->peer.init_upstream = ngx_http_upstream_init_chash;
668
669    } else {
670        ngx_conf_log_error(NGX_LOG_EMERG, cf, 0,
671                           "invalid parameter \"%V\"", &value[2]);
672        return NGX_CONF_ERROR;
673    }
674
675    return NGX_CONF_OK;
676}
677