Skip to content

Commit afe47ac

Browse files
xiaokai-wangxiaokai-wang
committed
update consistent hash method
1 parent 16cc7b4 commit afe47ac

File tree

1 file changed

+165
-95
lines changed

1 file changed

+165
-95
lines changed

‎src/ngx_stream_upsync_module.h

Lines changed: 165 additions & 95 deletions
Original file line numberDiff line numberDiff line change
@@ -72,8 +72,6 @@ typedef struct {
7272

7373
static int ngx_libc_cdecl ngx_stream_upsync_chash_cmp_points(const void *one,
7474
const void *two);
75-
static void ngx_stream_upsync_chash(ngx_stream_upstream_rr_peer_t *peer,
76-
ngx_stream_upstream_chash_points_t *points);
7775
static ngx_int_t ngx_stream_upsync_chash_init(ngx_stream_upstream_srv_conf_t *uscf,
7876
ngx_stream_upstream_rr_peers_t *tmp_peers);
7977
static ngx_int_t ngx_stream_upsync_del_chash_peer(
@@ -100,114 +98,113 @@ ngx_stream_upsync_chash_cmp_points(const void *one, const void *two)
10098
}
10199

102100

103-
static void
104-
ngx_stream_upsync_chash(ngx_stream_upstream_rr_peer_t *peer,
105-
ngx_stream_upstream_chash_points_t *points)
106-
{
107-
size_t host_len, port_len;
108-
u_char *host, *port, c;
109-
uint32_t hash, base_hash, pre_hash;
110-
ngx_str_t *server;
111-
ngx_uint_t npoints, j;
112-
113-
server = &peer->server;
114-
if (server->len >= 5
115-
&& ngx_strncasecmp(server->data, (u_char *) "unix:", 5) == 0)
116-
{
117-
host = server->data + 5;
118-
host_len = server->len - 5;
119-
port = NULL;
120-
port_len = 0;
121-
goto done;
122-
}
123-
124-
for (j = 0; j < server->len; j++) {
125-
c = server->data[server->len - j - 1];
126-
127-
if (c == ':') {
128-
host = server->data;
129-
host_len = server->len - j - 1;
130-
port = server->data + server->len - j;
131-
port_len = j;
132-
goto done;
133-
}
134-
135-
if (c < '0' || c > '9') {
136-
break;
137-
}
138-
}
139-
140-
host = server->data;
141-
host_len = server->len;
142-
port = NULL;
143-
port_len = 0;
144-
145-
done:
146-
147-
ngx_crc32_init(base_hash);
148-
ngx_crc32_update(&base_hash, host, host_len);
149-
ngx_crc32_update(&base_hash, (u_char *) "", 1);
150-
ngx_crc32_update(&base_hash, port, port_len);
151-
152-
pre_hash = 0;
153-
npoints = peer->weight * 160;
154-
155-
for(j = 0; j < npoints; j++) {
156-
hash = base_hash;
157-
158-
ngx_crc32_update(&hash, (u_char *)&pre_hash, sizeof(uint32_t));
159-
ngx_crc32_final(hash);
160-
161-
points->point[points->number].hash = hash;
162-
points->point[points->number].server = server;
163-
points->number++;
164-
165-
pre_hash = hash;
166-
}
167-
168-
return;
169-
}
170-
171-
172101
static ngx_int_t
173102
ngx_stream_upsync_chash_init(ngx_stream_upstream_srv_conf_t *uscf,
174103
ngx_stream_upstream_rr_peers_t *tmp_peers)
175104
{
176-
size_t old_size, new_size;
177-
ngx_uint_t old_npoints, new_npoints, i, j;
178-
ngx_stream_upstream_rr_peer_t *peer;
179-
ngx_stream_upstream_rr_peers_t *peers;
180-
ngx_stream_upstream_chash_points_t *points;
181-
ngx_stream_upstream_hash_srv_conf_t *hcf;
105+
size_t new_size;
106+
size_t host_len, port_len;
107+
u_char *host, *port, c;
108+
uint32_t hash, base_hash;
109+
ngx_str_t *server;
110+
ngx_uint_t npoints, new_npoints;
111+
ngx_uint_t i, j;
112+
ngx_stream_upstream_rr_peer_t *peer;
113+
ngx_stream_upstream_rr_peers_t *peers;
114+
ngx_stream_upstream_chash_points_t *points;
115+
ngx_stream_upstream_hash_srv_conf_t *hcf;
116+
union {
117+
uint32_t value;
118+
u_char byte[4];
119+
} prev_hash;
182120

183121
hcf = ngx_stream_conf_upstream_srv_conf(uscf, ngx_stream_upstream_hash_module);
184122
if(hcf->points == NULL) {
185123
return 0;
186124
}
187125

188126
peers = uscf->peer.data;
189-
190127
if (tmp_peers != NULL) {
191-
old_npoints = tmp_peers->total_weight * 160;
192128
new_npoints = peers->total_weight * 160;
193129

194-
old_size = sizeof(ngx_stream_upstream_chash_points_t)
195-
+ sizeof(ngx_stream_upstream_chash_point_t) * (old_npoints - 1);
196130
new_size = sizeof(ngx_stream_upstream_chash_points_t)
197131
+ sizeof(ngx_stream_upstream_chash_point_t) * (new_npoints - 1);
198132

199133
points = ngx_calloc(new_size, ngx_cycle->log);
200134
if (points == NULL ) {
201135
return NGX_ERROR;
202136
}
137+
ngx_free(hcf->points); /* free old points */
138+
hcf->points = points;
203139

204-
ngx_memcpy(points, hcf->points, old_size);
205-
ngx_free(hcf->points);
140+
for (peer = peers->peer; peer; peer = peer->next) {
141+
server = &peer->server;
142+
143+
/*
144+
* Hash expression is compatible with Cache::Memcached::Fast:
145+
* crc32(HOST \0 PORT PREV_HASH).
146+
*/
147+
148+
if (server->len >= 5
149+
&& ngx_strncasecmp(server->data, (u_char *) "unix:", 5) == 0)
150+
{
151+
host = server->data + 5;
152+
host_len = server->len - 5;
153+
port = NULL;
154+
port_len = 0;
155+
goto done;
156+
}
157+
158+
for (j = 0; j < server->len; j++) {
159+
c = server->data[server->len - j - 1];
160+
161+
if (c == ':') {
162+
host = server->data;
163+
host_len = server->len - j - 1;
164+
port = server->data + server->len - j;
165+
port_len = j;
166+
goto done;
167+
}
168+
169+
if (c < '0' || c > '9') {
170+
break;
171+
}
172+
}
206173

207-
hcf->points = points;
208-
for (i = 0; i < peers->number - tmp_peers->number; i++) {
209-
peer = &peers->peer[i];
210-
ngx_stream_upsync_chash(peer, points);
174+
host = server->data;
175+
host_len = server->len;
176+
port = NULL;
177+
port_len = 0;
178+
179+
done:
180+
181+
ngx_crc32_init(base_hash);
182+
ngx_crc32_update(&base_hash, host, host_len);
183+
ngx_crc32_update(&base_hash, (u_char *) "", 1);
184+
ngx_crc32_update(&base_hash, port, port_len);
185+
186+
prev_hash.value = 0;
187+
npoints = peer->weight * 160;
188+
189+
for (j = 0; j < npoints; j++) {
190+
hash = base_hash;
191+
192+
ngx_crc32_update(&hash, prev_hash.byte, 4);
193+
ngx_crc32_final(hash);
194+
195+
points->point[points->number].hash = hash;
196+
points->point[points->number].server = server;
197+
points->number++;
198+
199+
#if (NGX_HAVE_LITTLE_ENDIAN)
200+
prev_hash.value = hash;
201+
#else
202+
prev_hash.byte[0] = (u_char) (hash & 0xff);
203+
prev_hash.byte[1] = (u_char) ((hash >> 8) & 0xff);
204+
prev_hash.byte[2] = (u_char) ((hash >> 16) & 0xff);
205+
prev_hash.byte[3] = (u_char) ((hash >> 24) & 0xff);
206+
#endif
207+
}
211208
}
212209

213210
} else {
@@ -249,11 +246,19 @@ ngx_stream_upsync_chash_init(ngx_stream_upstream_srv_conf_t *uscf,
249246
static ngx_int_t
250247
ngx_stream_upsync_del_chash_peer(ngx_stream_upstream_srv_conf_t *uscf)
251248
{
252-
ngx_uint_t i, j;
253-
ngx_stream_upstream_rr_peer_t *peer;
254-
ngx_stream_upstream_rr_peers_t *peers;
255-
ngx_stream_upstream_chash_points_t *points;
256-
ngx_stream_upstream_hash_srv_conf_t *hcf;
249+
size_t host_len, port_len;
250+
u_char *host, *port, c;
251+
uint32_t hash, base_hash;
252+
ngx_str_t *server;
253+
ngx_uint_t npoints, i, j;
254+
ngx_stream_upstream_rr_peer_t *peer;
255+
ngx_stream_upstream_rr_peers_t *peers;
256+
ngx_stream_upstream_chash_points_t *points;
257+
ngx_stream_upstream_hash_srv_conf_t *hcf;
258+
union {
259+
uint32_t value;
260+
u_char byte[4];
261+
} prev_hash;
257262

258263
hcf = ngx_stream_conf_upstream_srv_conf(uscf, ngx_stream_upstream_hash_module);
259264
if(hcf->points == NULL) {
@@ -265,9 +270,74 @@ ngx_stream_upsync_del_chash_peer(ngx_stream_upstream_srv_conf_t *uscf)
265270
points = hcf->points;
266271
points->number = 0;
267272

268-
for (i = 0; i < peers->number; i++) {
269-
peer = &peers->peer[i];
270-
ngx_stream_upsync_chash(peer, points);
273+
for (peer = peers->peer; peer; peer = peer->next) {
274+
server = &peer->server;
275+
276+
/*
277+
* Hash expression is compatible with Cache::Memcached::Fast:
278+
* crc32(HOST \0 PORT PREV_HASH).
279+
*/
280+
281+
if (server->len >= 5
282+
&& ngx_strncasecmp(server->data, (u_char *) "unix:", 5) == 0)
283+
{
284+
host = server->data + 5;
285+
host_len = server->len - 5;
286+
port = NULL;
287+
port_len = 0;
288+
goto done;
289+
}
290+
291+
for (j = 0; j < server->len; j++) {
292+
c = server->data[server->len - j - 1];
293+
294+
if (c == ':') {
295+
host = server->data;
296+
host_len = server->len - j - 1;
297+
port = server->data + server->len - j;
298+
port_len = j;
299+
goto done;
300+
}
301+
302+
if (c < '0' || c > '9') {
303+
break;
304+
}
305+
}
306+
307+
host = server->data;
308+
host_len = server->len;
309+
port = NULL;
310+
port_len = 0;
311+
312+
done:
313+
314+
ngx_crc32_init(base_hash);
315+
ngx_crc32_update(&base_hash, host, host_len);
316+
ngx_crc32_update(&base_hash, (u_char *) "", 1);
317+
ngx_crc32_update(&base_hash, port, port_len);
318+
319+
prev_hash.value = 0;
320+
npoints = peer->weight * 160;
321+
322+
for (j = 0; j < npoints; j++) {
323+
hash = base_hash;
324+
325+
ngx_crc32_update(&hash, prev_hash.byte, 4);
326+
ngx_crc32_final(hash);
327+
328+
points->point[points->number].hash = hash;
329+
points->point[points->number].server = server;
330+
points->number++;
331+
332+
#if (NGX_HAVE_LITTLE_ENDIAN)
333+
prev_hash.value = hash;
334+
#else
335+
prev_hash.byte[0] = (u_char) (hash & 0xff);
336+
prev_hash.byte[1] = (u_char) ((hash >> 8) & 0xff);
337+
prev_hash.byte[2] = (u_char) ((hash >> 16) & 0xff);
338+
prev_hash.byte[3] = (u_char) ((hash >> 24) & 0xff);
339+
#endif
340+
}
271341
}
272342

273343
ngx_qsort(points->point,

0 commit comments

Comments
 (0)