1 /*
2 * Peer synchro management.
3 *
4 * Copyright 2010 EXCELIANCE, Emeric Brun <ebrun@exceliance.fr>
5 *
6 * This program is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU General Public License
8 * as published by the Free Software Foundation; either version
9 * 2 of the License, or (at your option) any later version.
10 *
11 */
12
13 #include <errno.h>
14 #include <fcntl.h>
15 #include <stdio.h>
16 #include <stdlib.h>
17 #include <string.h>
18
19 #include <sys/socket.h>
20 #include <sys/stat.h>
21 #include <sys/types.h>
22
23 #include <haproxy/api.h>
24 #include <haproxy/applet.h>
25 #include <haproxy/channel.h>
26 #include <haproxy/cli.h>
27 #include <haproxy/dict.h>
28 #include <haproxy/errors.h>
29 #include <haproxy/fd.h>
30 #include <haproxy/frontend.h>
31 #include <haproxy/net_helper.h>
32 #include <haproxy/obj_type-t.h>
33 #include <haproxy/peers.h>
34 #include <haproxy/proxy.h>
35 #include <haproxy/session-t.h>
36 #include <haproxy/signal.h>
37 #include <haproxy/stats-t.h>
38 #include <haproxy/stick_table.h>
39 #include <haproxy/stream.h>
40 #include <haproxy/stream_interface.h>
41 #include <haproxy/task.h>
42 #include <haproxy/thread.h>
43 #include <haproxy/time.h>
44 #include <haproxy/tools.h>
45 #include <haproxy/trace.h>
46
47
48 /*******************************/
49 /* Current peer learning state */
50 /*******************************/
51
52 /******************************/
53 /* Current peers section resync state */
54 /******************************/
55 #define PEERS_F_RESYNC_LOCAL 0x00000001 /* Learn from local finished or no more needed */
56 #define PEERS_F_RESYNC_REMOTE 0x00000002 /* Learn from remote finished or no more needed */
57 #define PEERS_F_RESYNC_ASSIGN 0x00000004 /* A peer was assigned to learn our lesson */
58 #define PEERS_F_RESYNC_PROCESS 0x00000008 /* The assigned peer was requested for resync */
59 #define PEERS_F_RESYNC_LOCALTIMEOUT 0x00000010 /* Timeout waiting for a full resync from a local node */
60 #define PEERS_F_RESYNC_REMOTETIMEOUT 0x00000020 /* Timeout waiting for a full resync from a remote node */
61 #define PEERS_F_RESYNC_LOCALABORT 0x00000040 /* Session aborted learning from a local node */
62 #define PEERS_F_RESYNC_REMOTEABORT 0x00000080 /* Session aborted learning from a remote node */
63 #define PEERS_F_RESYNC_LOCALFINISHED 0x00000100 /* A local node teach us and was fully up to date */
64 #define PEERS_F_RESYNC_REMOTEFINISHED 0x00000200 /* A remote node teach us and was fully up to date */
65 #define PEERS_F_RESYNC_LOCALPARTIAL 0x00000400 /* A local node teach us but was partially up to date */
66 #define PEERS_F_RESYNC_REMOTEPARTIAL 0x00000800 /* A remote node teach us but was partially up to date */
67 #define PEERS_F_RESYNC_LOCALASSIGN 0x00001000 /* A local node was assigned for a full resync */
68 #define PEERS_F_RESYNC_REMOTEASSIGN 0x00002000 /* A remote node was assigned for a full resync */
69 #define PEERS_F_RESYNC_REQUESTED 0x00004000 /* A resync was explicitly requested */
70 #define PEERS_F_DONOTSTOP 0x00010000 /* Main table sync task block process during soft stop
71 to push data to new process */
72
73 #define PEERS_RESYNC_STATEMASK (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE)
74 #define PEERS_RESYNC_FROMLOCAL 0x00000000
75 #define PEERS_RESYNC_FROMREMOTE PEERS_F_RESYNC_LOCAL
76 #define PEERS_RESYNC_FINISHED (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE)
77
78 /***********************************/
79 /* Current shared table sync state */
80 /***********************************/
81 #define SHTABLE_F_TEACH_STAGE1 0x00000001 /* Teach state 1 complete */
82 #define SHTABLE_F_TEACH_STAGE2 0x00000002 /* Teach state 2 complete */
83
84 /******************************/
85 /* Remote peer teaching state */
86 /******************************/
87 #define PEER_F_TEACH_PROCESS 0x00000001 /* Teach a lesson to current peer */
88 #define PEER_F_TEACH_FINISHED 0x00000008 /* Teach conclude, (wait for confirm) */
89 #define PEER_F_TEACH_COMPLETE 0x00000010 /* All that we know already taught to current peer, used only for a local peer */
90 #define PEER_F_LEARN_ASSIGN 0x00000100 /* Current peer was assigned for a lesson */
91 #define PEER_F_LEARN_NOTUP2DATE 0x00000200 /* Learn from peer finished but peer is not up to date */
92 #define PEER_F_ALIVE 0x20000000 /* Used to flag a peer a alive. */
93 #define PEER_F_HEARTBEAT 0x40000000 /* Heartbeat message to send. */
94 #define PEER_F_DWNGRD 0x80000000 /* When this flag is enabled, we must downgrade the supported version announced during peer sessions. */
95
96 #define PEER_TEACH_RESET ~(PEER_F_TEACH_PROCESS|PEER_F_TEACH_FINISHED) /* PEER_F_TEACH_COMPLETE should never be reset */
97 #define PEER_LEARN_RESET ~(PEER_F_LEARN_ASSIGN|PEER_F_LEARN_NOTUP2DATE)
98
99 #define PEER_RESYNC_TIMEOUT 5000 /* 5 seconds */
100 #define PEER_RECONNECT_TIMEOUT 5000 /* 5 seconds */
101 #define PEER_HEARTBEAT_TIMEOUT 3000 /* 3 seconds */
102
103 /*****************************/
104 /* Sync message class */
105 /*****************************/
106 enum {
107 PEER_MSG_CLASS_CONTROL = 0,
108 PEER_MSG_CLASS_ERROR,
109 PEER_MSG_CLASS_STICKTABLE = 10,
110 PEER_MSG_CLASS_RESERVED = 255,
111 };
112
113 /*****************************/
114 /* control message types */
115 /*****************************/
116 enum {
117 PEER_MSG_CTRL_RESYNCREQ = 0,
118 PEER_MSG_CTRL_RESYNCFINISHED,
119 PEER_MSG_CTRL_RESYNCPARTIAL,
120 PEER_MSG_CTRL_RESYNCCONFIRM,
121 PEER_MSG_CTRL_HEARTBEAT,
122 };
123
124 /*****************************/
125 /* error message types */
126 /*****************************/
127 enum {
128 PEER_MSG_ERR_PROTOCOL = 0,
129 PEER_MSG_ERR_SIZELIMIT,
130 };
131
132 /* network key types;
133 * network types were directly and mistakenly
134 * mapped on sample types, to keep backward
135 * compatiblitiy we keep those values but
136 * we now use a internal/network mapping
137 * to avoid further mistakes adding or
138 * modifying internals types
139 */
140 enum {
141 PEER_KT_ANY = 0, /* any type */
142 PEER_KT_RESV1, /* UNUSED */
143 PEER_KT_SINT, /* signed 64bits integer type */
144 PEER_KT_RESV3, /* UNUSED */
145 PEER_KT_IPV4, /* ipv4 type */
146 PEER_KT_IPV6, /* ipv6 type */
147 PEER_KT_STR, /* char string type */
148 PEER_KT_BIN, /* buffer type */
149 PEER_KT_TYPES /* number of types, must always be last */
150 };
151
152 /* Map used to retrieve network type from internal type
153 * Note: Undeclared mapping maps entry to PEER_KT_ANY == 0
154 */
155 static int peer_net_key_type[SMP_TYPES] = {
156 [SMP_T_SINT] = PEER_KT_SINT,
157 [SMP_T_IPV4] = PEER_KT_IPV4,
158 [SMP_T_IPV6] = PEER_KT_IPV6,
159 [SMP_T_STR] = PEER_KT_STR,
160 [SMP_T_BIN] = PEER_KT_BIN,
161 };
162
163 /* Map used to retrieve internal type from external type
164 * Note: Undeclared mapping maps entry to SMP_T_ANY == 0
165 */
166 static int peer_int_key_type[PEER_KT_TYPES] = {
167 [PEER_KT_SINT] = SMP_T_SINT,
168 [PEER_KT_IPV4] = SMP_T_IPV4,
169 [PEER_KT_IPV6] = SMP_T_IPV6,
170 [PEER_KT_STR] = SMP_T_STR,
171 [PEER_KT_BIN] = SMP_T_BIN,
172 };
173
174 /*
175 * Parameters used by functions to build peer protocol messages. */
176 struct peer_prep_params {
177 struct {
178 struct peer *peer;
179 } hello;
180 struct {
181 unsigned int st1;
182 } error_status;
183 struct {
184 struct stksess *stksess;
185 struct shared_table *shared_table;
186 unsigned int updateid;
187 int use_identifier;
188 int use_timed;
189 struct peer *peer;
190 } updt;
191 struct {
192 struct shared_table *shared_table;
193 } swtch;
194 struct {
195 struct shared_table *shared_table;
196 } ack;
197 struct {
198 unsigned char head[2];
199 } control;
200 struct {
201 unsigned char head[2];
202 } error;
203 };
204
205 /*******************************/
206 /* stick table sync mesg types */
207 /* Note: ids >= 128 contains */
208 /* id message contains data */
209 /*******************************/
210 #define PEER_MSG_STKT_UPDATE 0x80
211 #define PEER_MSG_STKT_INCUPDATE 0x81
212 #define PEER_MSG_STKT_DEFINE 0x82
213 #define PEER_MSG_STKT_SWITCH 0x83
214 #define PEER_MSG_STKT_ACK 0x84
215 #define PEER_MSG_STKT_UPDATE_TIMED 0x85
216 #define PEER_MSG_STKT_INCUPDATE_TIMED 0x86
217 /* All the stick-table message identifiers abova have the #7 bit set */
218 #define PEER_MSG_STKT_BIT 7
219 #define PEER_MSG_STKT_BIT_MASK (1 << PEER_MSG_STKT_BIT)
220
221 /* The maximum length of an encoded data length. */
222 #define PEER_MSG_ENC_LENGTH_MAXLEN 5
223
224 /* Minimum 64-bits value encoded with 2 bytes */
225 #define PEER_ENC_2BYTES_MIN 0xf0 /* 0xf0 (or 240) */
226 /* 3 bytes */
227 #define PEER_ENC_3BYTES_MIN ((1ULL << 11) | PEER_ENC_2BYTES_MIN) /* 0x8f0 (or 2288) */
228 /* 4 bytes */
229 #define PEER_ENC_4BYTES_MIN ((1ULL << 18) | PEER_ENC_3BYTES_MIN) /* 0x408f0 (or 264432) */
230 /* 5 bytes */
231 #define PEER_ENC_5BYTES_MIN ((1ULL << 25) | PEER_ENC_4BYTES_MIN) /* 0x20408f0 (or 33818864) */
232 /* 6 bytes */
233 #define PEER_ENC_6BYTES_MIN ((1ULL << 32) | PEER_ENC_5BYTES_MIN) /* 0x1020408f0 (or 4328786160) */
234 /* 7 bytes */
235 #define PEER_ENC_7BYTES_MIN ((1ULL << 39) | PEER_ENC_6BYTES_MIN) /* 0x81020408f0 (or 554084600048) */
236 /* 8 bytes */
237 #define PEER_ENC_8BYTES_MIN ((1ULL << 46) | PEER_ENC_7BYTES_MIN) /* 0x4081020408f0 (or 70922828777712) */
238 /* 9 bytes */
239 #define PEER_ENC_9BYTES_MIN ((1ULL << 53) | PEER_ENC_8BYTES_MIN) /* 0x204081020408f0 (or 9078122083518704) */
240 /* 10 bytes */
241 #define PEER_ENC_10BYTES_MIN ((1ULL << 60) | PEER_ENC_9BYTES_MIN) /* 0x10204081020408f0 (or 1161999626690365680) */
242
243 /* #7 bit used to detect the last byte to be encoded */
244 #define PEER_ENC_STOP_BIT 7
245 /* The byte minimum value with #7 bit set */
246 #define PEER_ENC_STOP_BYTE (1 << PEER_ENC_STOP_BIT)
247 /* The left most number of bits set for PEER_ENC_2BYTES_MIN */
248 #define PEER_ENC_2BYTES_MIN_BITS 4
249
250 #define PEER_MSG_HEADER_LEN 2
251
252 #define PEER_STKT_CACHE_MAX_ENTRIES 128
253
254 /**********************************/
255 /* Peer Session IO handler states */
256 /**********************************/
257
258 enum {
259 PEER_SESS_ST_ACCEPT = 0, /* Initial state for session create by an accept, must be zero! */
260 PEER_SESS_ST_GETVERSION, /* Validate supported protocol version */
261 PEER_SESS_ST_GETHOST, /* Validate host ID correspond to local host id */
262 PEER_SESS_ST_GETPEER, /* Validate peer ID correspond to a known remote peer id */
263 /* after this point, data were possibly exchanged */
264 PEER_SESS_ST_SENDSUCCESS, /* Send ret code 200 (success) and wait for message */
265 PEER_SESS_ST_CONNECT, /* Initial state for session create on a connect, push presentation into buffer */
266 PEER_SESS_ST_GETSTATUS, /* Wait for the welcome message */
267 PEER_SESS_ST_WAITMSG, /* Wait for data messages */
268 PEER_SESS_ST_EXIT, /* Exit with status code */
269 PEER_SESS_ST_ERRPROTO, /* Send error proto message before exit */
270 PEER_SESS_ST_ERRSIZE, /* Send error size message before exit */
271 PEER_SESS_ST_END, /* Killed session */
272 };
273
274 /***************************************************/
275 /* Peer Session status code - part of the protocol */
276 /***************************************************/
277
278 #define PEER_SESS_SC_CONNECTCODE 100 /* connect in progress */
279 #define PEER_SESS_SC_CONNECTEDCODE 110 /* tcp connect success */
280
281 #define PEER_SESS_SC_SUCCESSCODE 200 /* accept or connect successful */
282
283 #define PEER_SESS_SC_TRYAGAIN 300 /* try again later */
284
285 #define PEER_SESS_SC_ERRPROTO 501 /* error protocol */
286 #define PEER_SESS_SC_ERRVERSION 502 /* unknown protocol version */
287 #define PEER_SESS_SC_ERRHOST 503 /* bad host name */
288 #define PEER_SESS_SC_ERRPEER 504 /* unknown peer */
289
290 #define PEER_SESSION_PROTO_NAME "HAProxyS"
291 #define PEER_MAJOR_VER 2
292 #define PEER_MINOR_VER 1
293 #define PEER_DWNGRD_MINOR_VER 0
294
295 static size_t proto_len = sizeof(PEER_SESSION_PROTO_NAME) - 1;
296 struct peers *cfg_peers = NULL;
297 static void peer_session_forceshutdown(struct peer *peer);
298
299 static struct ebpt_node *dcache_tx_insert(struct dcache *dc,
300 struct dcache_tx_entry *i);
301 static inline void flush_dcache(struct peer *peer);
302
303 /* trace source and events */
304 static void peers_trace(enum trace_level level, uint64_t mask,
305 const struct trace_source *src,
306 const struct ist where, const struct ist func,
307 const void *a1, const void *a2, const void *a3, const void *a4);
308
309 static const struct trace_event peers_trace_events[] = {
310 #define PEERS_EV_UPDTMSG (1 << 0)
311 { .mask = PEERS_EV_UPDTMSG, .name = "updtmsg", .desc = "update message received" },
312 #define PEERS_EV_ACKMSG (1 << 1)
313 { .mask = PEERS_EV_ACKMSG, .name = "ackmsg", .desc = "ack message received" },
314 #define PEERS_EV_SWTCMSG (1 << 2)
315 { .mask = PEERS_EV_SWTCMSG, .name = "swtcmsg", .desc = "switch message received" },
316 #define PEERS_EV_DEFMSG (1 << 3)
317 { .mask = PEERS_EV_DEFMSG, .name = "defmsg", .desc = "definition message received" },
318 #define PEERS_EV_CTRLMSG (1 << 4)
319 { .mask = PEERS_EV_CTRLMSG, .name = "ctrlmsg", .desc = "control message sent/received" },
320 #define PEERS_EV_SESSREL (1 << 5)
321 { .mask = PEERS_EV_SESSREL, .name = "sessrl", .desc = "peer session releasing" },
322 #define PEERS_EV_PROTOERR (1 << 6)
323 { .mask = PEERS_EV_PROTOERR, .name = "protoerr", .desc = "protocol error" },
324 };
325
326 static const struct name_desc peers_trace_lockon_args[4] = {
327 /* arg1 */ { /* already used by the connection */ },
328 /* arg2 */ { .name="peers", .desc="Peers protocol" },
329 /* arg3 */ { },
330 /* arg4 */ { }
331 };
332
333 static const struct name_desc peers_trace_decoding[] = {
334 #define PEERS_VERB_CLEAN 1
335 { .name="clean", .desc="only user-friendly stuff, generally suitable for level \"user\"" },
336 { /* end */ }
337 };
338
339
340 struct trace_source trace_peers = {
341 .name = IST("peers"),
342 .desc = "Peers protocol",
343 .arg_def = TRC_ARG1_CONN, /* TRACE()'s first argument is always a connection */
344 .default_cb = peers_trace,
345 .known_events = peers_trace_events,
346 .lockon_args = peers_trace_lockon_args,
347 .decoding = peers_trace_decoding,
348 .report_events = ~0, /* report everything by default */
349 };
350
351 /* Return peer control message types as strings (only for debugging purpose). */
ctrl_msg_type_str(unsigned int type)352 static inline char *ctrl_msg_type_str(unsigned int type)
353 {
354 switch (type) {
355 case PEER_MSG_CTRL_RESYNCREQ:
356 return "RESYNCREQ";
357 case PEER_MSG_CTRL_RESYNCFINISHED:
358 return "RESYNCFINISHED";
359 case PEER_MSG_CTRL_RESYNCPARTIAL:
360 return "RESYNCPARTIAL";
361 case PEER_MSG_CTRL_RESYNCCONFIRM:
362 return "RESYNCCONFIRM";
363 case PEER_MSG_CTRL_HEARTBEAT:
364 return "HEARTBEAT";
365 default:
366 return "???";
367 }
368 }
369
370 #define TRACE_SOURCE &trace_peers
371 INITCALL1(STG_REGISTER, trace_register_source, TRACE_SOURCE);
372
peers_trace(enum trace_level level,uint64_t mask,const struct trace_source * src,const struct ist where,const struct ist func,const void * a1,const void * a2,const void * a3,const void * a4)373 static void peers_trace(enum trace_level level, uint64_t mask,
374 const struct trace_source *src,
375 const struct ist where, const struct ist func,
376 const void *a1, const void *a2, const void *a3, const void *a4)
377 {
378 if (mask & (PEERS_EV_UPDTMSG|PEERS_EV_ACKMSG|PEERS_EV_SWTCMSG)) {
379 if (a2) {
380 const struct peer *peer = a2;
381
382 chunk_appendf(&trace_buf, " peer=%s", peer->id);
383 }
384 if (a3) {
385 const char *p = a3;
386
387 chunk_appendf(&trace_buf, " @%p", p);
388 }
389 if (a4) {
390 const size_t *val = a4;
391
392 chunk_appendf(&trace_buf, " %llu", (unsigned long long)*val);
393 }
394 }
395
396 if (mask & PEERS_EV_DEFMSG) {
397 if (a2) {
398 const struct peer *peer = a2;
399
400 chunk_appendf(&trace_buf, " peer=%s", peer->id);
401 }
402 if (a3) {
403 const char *p = a3;
404
405 chunk_appendf(&trace_buf, " @%p", p);
406 }
407 if (a4) {
408 const int *val = a4;
409
410 chunk_appendf(&trace_buf, " %d", *val);
411 }
412 }
413
414 if (mask & PEERS_EV_CTRLMSG) {
415 if (a2) {
416 const unsigned char *ctrl_msg_type = a2;
417
418 chunk_appendf(&trace_buf, " %s", ctrl_msg_type_str(*ctrl_msg_type));
419
420 }
421 if (a3) {
422 const char *local_peer = a3;
423
424 chunk_appendf(&trace_buf, " %s", local_peer);
425 }
426
427 if (a4) {
428 const char *remote_peer = a4;
429
430 chunk_appendf(&trace_buf, " -> %s", remote_peer);
431 }
432 }
433
434 if (mask & (PEERS_EV_SESSREL|PEERS_EV_PROTOERR)) {
435 if (a2) {
436 const struct peer *peer = a2;
437 struct peers *peers = NULL;
438
439 if (peer && peer->appctx) {
440 struct stream_interface *si;
441
442 si = peer->appctx->owner;
443 if (si) {
444 struct stream *s = si_strm(si);
445
446 peers = strm_fe(s)->parent;
447 }
448 }
449
450 if (peers)
451 chunk_appendf(&trace_buf, " %s", peers->local->id);
452 if (peer)
453 chunk_appendf(&trace_buf, " -> %s", peer->id);
454 }
455
456 if (a3) {
457 const int *prev_state = a3;
458
459 chunk_appendf(&trace_buf, " prev_state=%d\n", *prev_state);
460 }
461 }
462 }
463
statuscode_str(int statuscode)464 static const char *statuscode_str(int statuscode)
465 {
466 switch (statuscode) {
467 case PEER_SESS_SC_CONNECTCODE:
468 return "CONN";
469 case PEER_SESS_SC_CONNECTEDCODE:
470 return "HSHK";
471 case PEER_SESS_SC_SUCCESSCODE:
472 return "ESTA";
473 case PEER_SESS_SC_TRYAGAIN:
474 return "RETR";
475 case PEER_SESS_SC_ERRPROTO:
476 return "PROT";
477 case PEER_SESS_SC_ERRVERSION:
478 return "VERS";
479 case PEER_SESS_SC_ERRHOST:
480 return "NAME";
481 case PEER_SESS_SC_ERRPEER:
482 return "UNKN";
483 default:
484 return "NONE";
485 }
486 }
487
488 /* This function encode an uint64 to 'dynamic' length format.
489 The encoded value is written at address *str, and the
490 caller must assure that size after *str is large enough.
491 At return, the *str is set at the next Byte after then
492 encoded integer. The function returns then length of the
493 encoded integer in Bytes */
intencode(uint64_t i,char ** str)494 int intencode(uint64_t i, char **str) {
495 int idx = 0;
496 unsigned char *msg;
497
498 msg = (unsigned char *)*str;
499 if (i < PEER_ENC_2BYTES_MIN) {
500 msg[0] = (unsigned char)i;
501 *str = (char *)&msg[idx+1];
502 return (idx+1);
503 }
504
505 msg[idx] =(unsigned char)i | PEER_ENC_2BYTES_MIN;
506 i = (i - PEER_ENC_2BYTES_MIN) >> PEER_ENC_2BYTES_MIN_BITS;
507 while (i >= PEER_ENC_STOP_BYTE) {
508 msg[++idx] = (unsigned char)i | PEER_ENC_STOP_BYTE;
509 i = (i - PEER_ENC_STOP_BYTE) >> PEER_ENC_STOP_BIT;
510 }
511 msg[++idx] = (unsigned char)i;
512 *str = (char *)&msg[idx+1];
513 return (idx+1);
514 }
515
516
517 /* This function returns the decoded integer or 0
518 if decode failed
519 *str point on the beginning of the integer to decode
520 at the end of decoding *str point on the end of the
521 encoded integer or to null if end is reached */
intdecode(char ** str,char * end)522 uint64_t intdecode(char **str, char *end)
523 {
524 unsigned char *msg;
525 uint64_t i;
526 int shift;
527
528 if (!*str)
529 return 0;
530
531 msg = (unsigned char *)*str;
532 if (msg >= (unsigned char *)end)
533 goto fail;
534
535 i = *(msg++);
536 if (i >= PEER_ENC_2BYTES_MIN) {
537 shift = PEER_ENC_2BYTES_MIN_BITS;
538 do {
539 if (msg >= (unsigned char *)end)
540 goto fail;
541 i += (uint64_t)*msg << shift;
542 shift += PEER_ENC_STOP_BIT;
543 } while (*(msg++) >= PEER_ENC_STOP_BYTE);
544 }
545 *str = (char *)msg;
546 return i;
547
548 fail:
549 *str = NULL;
550 return 0;
551 }
552
553 /*
554 * Build a "hello" peer protocol message.
555 * Return the number of written bytes written to build this messages if succeeded,
556 * 0 if not.
557 */
peer_prepare_hellomsg(char * msg,size_t size,struct peer_prep_params * p)558 static int peer_prepare_hellomsg(char *msg, size_t size, struct peer_prep_params *p)
559 {
560 int min_ver, ret;
561 struct peer *peer;
562
563 peer = p->hello.peer;
564 min_ver = (peer->flags & PEER_F_DWNGRD) ? PEER_DWNGRD_MINOR_VER : PEER_MINOR_VER;
565 /* Prepare headers */
566 ret = snprintf(msg, size, PEER_SESSION_PROTO_NAME " %u.%u\n%s\n%s %d %d\n",
567 PEER_MAJOR_VER, min_ver, peer->id, localpeer, (int)getpid(), relative_pid);
568 if (ret >= size)
569 return 0;
570
571 return ret;
572 }
573
574 /*
575 * Build a "handshake succeeded" status message.
576 * Return the number of written bytes written to build this messages if succeeded,
577 * 0 if not.
578 */
peer_prepare_status_successmsg(char * msg,size_t size,struct peer_prep_params * p)579 static int peer_prepare_status_successmsg(char *msg, size_t size, struct peer_prep_params *p)
580 {
581 int ret;
582
583 ret = snprintf(msg, size, "%d\n", PEER_SESS_SC_SUCCESSCODE);
584 if (ret >= size)
585 return 0;
586
587 return ret;
588 }
589
590 /*
591 * Build an error status message.
592 * Return the number of written bytes written to build this messages if succeeded,
593 * 0 if not.
594 */
peer_prepare_status_errormsg(char * msg,size_t size,struct peer_prep_params * p)595 static int peer_prepare_status_errormsg(char *msg, size_t size, struct peer_prep_params *p)
596 {
597 int ret;
598 unsigned int st1;
599
600 st1 = p->error_status.st1;
601 ret = snprintf(msg, size, "%d\n", st1);
602 if (ret >= size)
603 return 0;
604
605 return ret;
606 }
607
608 /* Set the stick-table UPDATE message type byte at <msg_type> address,
609 * depending on <use_identifier> and <use_timed> boolean parameters.
610 * Always successful.
611 */
peer_set_update_msg_type(char * msg_type,int use_identifier,int use_timed)612 static inline void peer_set_update_msg_type(char *msg_type, int use_identifier, int use_timed)
613 {
614 if (use_timed) {
615 if (use_identifier)
616 *msg_type = PEER_MSG_STKT_UPDATE_TIMED;
617 else
618 *msg_type = PEER_MSG_STKT_INCUPDATE_TIMED;
619 }
620 else {
621 if (use_identifier)
622 *msg_type = PEER_MSG_STKT_UPDATE;
623 else
624 *msg_type = PEER_MSG_STKT_INCUPDATE;
625 }
626 }
627 /*
628 * This prepare the data update message on the stick session <ts>, <st> is the considered
629 * stick table.
630 * <msg> is a buffer of <size> to receive data message content
631 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
632 * check size)
633 */
peer_prepare_updatemsg(char * msg,size_t size,struct peer_prep_params * p)634 static int peer_prepare_updatemsg(char *msg, size_t size, struct peer_prep_params *p)
635 {
636 uint32_t netinteger;
637 unsigned short datalen;
638 char *cursor, *datamsg;
639 unsigned int data_type;
640 void *data_ptr;
641 struct stksess *ts;
642 struct shared_table *st;
643 unsigned int updateid;
644 int use_identifier;
645 int use_timed;
646 struct peer *peer;
647
648 ts = p->updt.stksess;
649 st = p->updt.shared_table;
650 updateid = p->updt.updateid;
651 use_identifier = p->updt.use_identifier;
652 use_timed = p->updt.use_timed;
653 peer = p->updt.peer;
654
655 cursor = datamsg = msg + PEER_MSG_HEADER_LEN + PEER_MSG_ENC_LENGTH_MAXLEN;
656
657 /* construct message */
658
659 /* check if we need to send the update identifier */
660 if (!st->last_pushed || updateid < st->last_pushed || ((updateid - st->last_pushed) != 1)) {
661 use_identifier = 1;
662 }
663
664 /* encode update identifier if needed */
665 if (use_identifier) {
666 netinteger = htonl(updateid);
667 memcpy(cursor, &netinteger, sizeof(netinteger));
668 cursor += sizeof(netinteger);
669 }
670
671 if (use_timed) {
672 netinteger = htonl(tick_remain(now_ms, ts->expire));
673 memcpy(cursor, &netinteger, sizeof(netinteger));
674 cursor += sizeof(netinteger);
675 }
676
677 /* encode the key */
678 if (st->table->type == SMP_T_STR) {
679 int stlen = strlen((char *)ts->key.key);
680
681 intencode(stlen, &cursor);
682 memcpy(cursor, ts->key.key, stlen);
683 cursor += stlen;
684 }
685 else if (st->table->type == SMP_T_SINT) {
686 netinteger = htonl(read_u32(ts->key.key));
687 memcpy(cursor, &netinteger, sizeof(netinteger));
688 cursor += sizeof(netinteger);
689 }
690 else {
691 memcpy(cursor, ts->key.key, st->table->key_size);
692 cursor += st->table->key_size;
693 }
694
695 HA_RWLOCK_RDLOCK(STK_SESS_LOCK, &ts->lock);
696 /* encode values */
697 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
698
699 data_ptr = stktable_data_ptr(st->table, ts, data_type);
700 if (data_ptr) {
701 switch (stktable_data_types[data_type].std_type) {
702 case STD_T_SINT: {
703 int data;
704
705 data = stktable_data_cast(data_ptr, std_t_sint);
706 intencode(data, &cursor);
707 break;
708 }
709 case STD_T_UINT: {
710 unsigned int data;
711
712 data = stktable_data_cast(data_ptr, std_t_uint);
713 intencode(data, &cursor);
714 break;
715 }
716 case STD_T_ULL: {
717 unsigned long long data;
718
719 data = stktable_data_cast(data_ptr, std_t_ull);
720 intencode(data, &cursor);
721 break;
722 }
723 case STD_T_FRQP: {
724 struct freq_ctr_period *frqp;
725
726 frqp = &stktable_data_cast(data_ptr, std_t_frqp);
727 intencode((unsigned int)(now_ms - frqp->curr_tick), &cursor);
728 intencode(frqp->curr_ctr, &cursor);
729 intencode(frqp->prev_ctr, &cursor);
730 break;
731 }
732 case STD_T_DICT: {
733 struct dict_entry *de;
734 struct ebpt_node *cached_de;
735 struct dcache_tx_entry cde = { };
736 char *beg, *end;
737 size_t value_len, data_len;
738 struct dcache *dc;
739
740 de = stktable_data_cast(data_ptr, std_t_dict);
741 if (!de) {
742 /* No entry */
743 intencode(0, &cursor);
744 break;
745 }
746
747 dc = peer->dcache;
748 cde.entry.key = de;
749 cached_de = dcache_tx_insert(dc, &cde);
750 if (cached_de == &cde.entry) {
751 if (cde.id + 1 >= PEER_ENC_2BYTES_MIN)
752 break;
753 /* Encode the length of the remaining data -> 1 */
754 intencode(1, &cursor);
755 /* Encode the cache entry ID */
756 intencode(cde.id + 1, &cursor);
757 }
758 else {
759 /* Leave enough room to encode the remaining data length. */
760 end = beg = cursor + PEER_MSG_ENC_LENGTH_MAXLEN;
761 /* Encode the dictionary entry key */
762 intencode(cde.id + 1, &end);
763 /* Encode the length of the dictionary entry data */
764 value_len = de->len;
765 intencode(value_len, &end);
766 /* Copy the data */
767 memcpy(end, de->value.key, value_len);
768 end += value_len;
769 /* Encode the length of the data */
770 data_len = end - beg;
771 intencode(data_len, &cursor);
772 memmove(cursor, beg, data_len);
773 cursor += data_len;
774 }
775 break;
776 }
777 }
778 }
779 }
780 HA_RWLOCK_RDUNLOCK(STK_SESS_LOCK, &ts->lock);
781
782 /* Compute datalen */
783 datalen = (cursor - datamsg);
784
785 /* prepare message header */
786 msg[0] = PEER_MSG_CLASS_STICKTABLE;
787 peer_set_update_msg_type(&msg[1], use_identifier, use_timed);
788 cursor = &msg[2];
789 intencode(datalen, &cursor);
790
791 /* move data after header */
792 memmove(cursor, datamsg, datalen);
793
794 /* return header size + data_len */
795 return (cursor - msg) + datalen;
796 }
797
798 /*
799 * This prepare the switch table message to targeted share table <st>.
800 * <msg> is a buffer of <size> to receive data message content
801 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
802 * check size)
803 */
peer_prepare_switchmsg(char * msg,size_t size,struct peer_prep_params * params)804 static int peer_prepare_switchmsg(char *msg, size_t size, struct peer_prep_params *params)
805 {
806 int len;
807 unsigned short datalen;
808 struct buffer *chunk;
809 char *cursor, *datamsg, *chunkp, *chunkq;
810 uint64_t data = 0;
811 unsigned int data_type;
812 struct shared_table *st;
813
814 st = params->swtch.shared_table;
815 cursor = datamsg = msg + PEER_MSG_HEADER_LEN + PEER_MSG_ENC_LENGTH_MAXLEN;
816
817 /* Encode data */
818
819 /* encode local id */
820 intencode(st->local_id, &cursor);
821
822 /* encode table name */
823 len = strlen(st->table->nid);
824 intencode(len, &cursor);
825 memcpy(cursor, st->table->nid, len);
826 cursor += len;
827
828 /* encode table type */
829
830 intencode(peer_net_key_type[st->table->type], &cursor);
831
832 /* encode table key size */
833 intencode(st->table->key_size, &cursor);
834
835 chunk = get_trash_chunk();
836 chunkp = chunkq = chunk->area;
837 /* encode available known data types in table */
838 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
839 if (st->table->data_ofs[data_type]) {
840 switch (stktable_data_types[data_type].std_type) {
841 case STD_T_SINT:
842 case STD_T_UINT:
843 case STD_T_ULL:
844 case STD_T_DICT:
845 data |= 1ULL << data_type;
846 break;
847 case STD_T_FRQP:
848 data |= 1ULL << data_type;
849 intencode(data_type, &chunkq);
850 intencode(st->table->data_arg[data_type].u, &chunkq);
851 break;
852 }
853 }
854 }
855 intencode(data, &cursor);
856
857 /* Encode stick-table entries duration. */
858 intencode(st->table->expire, &cursor);
859
860 if (chunkq > chunkp) {
861 chunk->data = chunkq - chunkp;
862 memcpy(cursor, chunk->area, chunk->data);
863 cursor += chunk->data;
864 }
865
866 /* Compute datalen */
867 datalen = (cursor - datamsg);
868
869 /* prepare message header */
870 msg[0] = PEER_MSG_CLASS_STICKTABLE;
871 msg[1] = PEER_MSG_STKT_DEFINE;
872 cursor = &msg[2];
873 intencode(datalen, &cursor);
874
875 /* move data after header */
876 memmove(cursor, datamsg, datalen);
877
878 /* return header size + data_len */
879 return (cursor - msg) + datalen;
880 }
881
882 /*
883 * This prepare the acknowledge message on the stick session <ts>, <st> is the considered
884 * stick table.
885 * <msg> is a buffer of <size> to receive data message content
886 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
887 * check size)
888 */
peer_prepare_ackmsg(char * msg,size_t size,struct peer_prep_params * p)889 static int peer_prepare_ackmsg(char *msg, size_t size, struct peer_prep_params *p)
890 {
891 unsigned short datalen;
892 char *cursor, *datamsg;
893 uint32_t netinteger;
894 struct shared_table *st;
895
896 cursor = datamsg = msg + PEER_MSG_HEADER_LEN + PEER_MSG_ENC_LENGTH_MAXLEN;
897
898 st = p->ack.shared_table;
899 intencode(st->remote_id, &cursor);
900 netinteger = htonl(st->last_get);
901 memcpy(cursor, &netinteger, sizeof(netinteger));
902 cursor += sizeof(netinteger);
903
904 /* Compute datalen */
905 datalen = (cursor - datamsg);
906
907 /* prepare message header */
908 msg[0] = PEER_MSG_CLASS_STICKTABLE;
909 msg[1] = PEER_MSG_STKT_ACK;
910 cursor = &msg[2];
911 intencode(datalen, &cursor);
912
913 /* move data after header */
914 memmove(cursor, datamsg, datalen);
915
916 /* return header size + data_len */
917 return (cursor - msg) + datalen;
918 }
919
920 /*
921 * Function to deinit connected peer
922 */
__peer_session_deinit(struct peer * peer)923 void __peer_session_deinit(struct peer *peer)
924 {
925 struct stream_interface *si;
926 struct stream *s;
927 struct peers *peers;
928
929 if (!peer->appctx)
930 return;
931
932 si = peer->appctx->owner;
933 if (!si)
934 return;
935
936 s = si_strm(si);
937 if (!s)
938 return;
939
940 peers = strm_fe(s)->parent;
941 if (!peers)
942 return;
943
944 if (peer->appctx->st0 == PEER_SESS_ST_WAITMSG)
945 HA_ATOMIC_SUB(&connected_peers, 1);
946
947 HA_ATOMIC_SUB(&active_peers, 1);
948
949 flush_dcache(peer);
950
951 /* Re-init current table pointers to force announcement on re-connect */
952 peer->remote_table = peer->last_local_table = NULL;
953 peer->appctx = NULL;
954 if (peer->flags & PEER_F_LEARN_ASSIGN) {
955 /* unassign current peer for learning */
956 peer->flags &= ~(PEER_F_LEARN_ASSIGN);
957 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
958
959 if (peer->local)
960 peers->flags |= PEERS_F_RESYNC_LOCALABORT;
961 else
962 peers->flags |= PEERS_F_RESYNC_REMOTEABORT;
963 /* reschedule a resync */
964 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
965 }
966 /* reset teaching and learning flags to 0 */
967 peer->flags &= PEER_TEACH_RESET;
968 peer->flags &= PEER_LEARN_RESET;
969 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
970 }
971
972 /*
973 * Callback to release a session with a peer
974 */
peer_session_release(struct appctx * appctx)975 static void peer_session_release(struct appctx *appctx)
976 {
977 struct peer *peer = appctx->ctx.peers.ptr;
978
979 TRACE_PROTO("releasing peer session", PEERS_EV_SESSREL, NULL, peer);
980 /* appctx->ctx.peers.ptr is not a peer session */
981 if (appctx->st0 < PEER_SESS_ST_SENDSUCCESS)
982 return;
983
984 /* peer session identified */
985 if (peer) {
986 HA_SPIN_LOCK(PEER_LOCK, &peer->lock);
987 if (peer->appctx == appctx)
988 __peer_session_deinit(peer);
989 peer->flags &= ~PEER_F_ALIVE;
990 HA_SPIN_UNLOCK(PEER_LOCK, &peer->lock);
991 }
992 }
993
994 /* Retrieve the major and minor versions of peers protocol
995 * announced by a remote peer. <str> is a null-terminated
996 * string with the following format: "<maj_ver>.<min_ver>".
997 */
peer_get_version(const char * str,unsigned int * maj_ver,unsigned int * min_ver)998 static int peer_get_version(const char *str,
999 unsigned int *maj_ver, unsigned int *min_ver)
1000 {
1001 unsigned int majv, minv;
1002 const char *pos, *saved;
1003 const char *end;
1004
1005 saved = pos = str;
1006 end = str + strlen(str);
1007
1008 majv = read_uint(&pos, end);
1009 if (saved == pos || *pos++ != '.')
1010 return -1;
1011
1012 saved = pos;
1013 minv = read_uint(&pos, end);
1014 if (saved == pos || pos != end)
1015 return -1;
1016
1017 *maj_ver = majv;
1018 *min_ver = minv;
1019
1020 return 0;
1021 }
1022
1023 /*
1024 * Parse a line terminated by an optional '\r' character, followed by a mandatory
1025 * '\n' character.
1026 * Returns 1 if succeeded or 0 if a '\n' character could not be found, and -1 if
1027 * a line could not be read because the communication channel is closed.
1028 */
peer_getline(struct appctx * appctx)1029 static inline int peer_getline(struct appctx *appctx)
1030 {
1031 int n;
1032 struct stream_interface *si = appctx->owner;
1033
1034 n = co_getline(si_oc(si), trash.area, trash.size);
1035 if (!n)
1036 return 0;
1037
1038 if (n < 0 || trash.area[n - 1] != '\n') {
1039 appctx->st0 = PEER_SESS_ST_END;
1040 return -1;
1041 }
1042
1043 if (n > 1 && (trash.area[n - 2] == '\r'))
1044 trash.area[n - 2] = 0;
1045 else
1046 trash.area[n - 1] = 0;
1047
1048 co_skip(si_oc(si), n);
1049
1050 return n;
1051 }
1052
1053 /*
1054 * Send a message after having called <peer_prepare_msg> to build it.
1055 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1056 * Returns -1 if there was not enough room left to send the message,
1057 * any other negative returned value must be considered as an error with an appcxt st0
1058 * returned value equal to PEER_SESS_ST_END.
1059 */
peer_send_msg(struct appctx * appctx,int (* peer_prepare_msg)(char *,size_t,struct peer_prep_params *),struct peer_prep_params * params)1060 static inline int peer_send_msg(struct appctx *appctx,
1061 int (*peer_prepare_msg)(char *, size_t, struct peer_prep_params *),
1062 struct peer_prep_params *params)
1063 {
1064 int ret, msglen;
1065 struct stream_interface *si = appctx->owner;
1066
1067 msglen = peer_prepare_msg(trash.area, trash.size, params);
1068 if (!msglen) {
1069 /* internal error: message does not fit in trash */
1070 appctx->st0 = PEER_SESS_ST_END;
1071 return 0;
1072 }
1073
1074 /* message to buffer */
1075 ret = ci_putblk(si_ic(si), trash.area, msglen);
1076 if (ret <= 0) {
1077 if (ret == -1) {
1078 /* No more write possible */
1079 si_rx_room_blk(si);
1080 return -1;
1081 }
1082 appctx->st0 = PEER_SESS_ST_END;
1083 }
1084
1085 return ret;
1086 }
1087
1088 /*
1089 * Send a hello message.
1090 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1091 * Returns -1 if there was not enough room left to send the message,
1092 * any other negative returned value must be considered as an error with an appcxt st0
1093 * returned value equal to PEER_SESS_ST_END.
1094 */
peer_send_hellomsg(struct appctx * appctx,struct peer * peer)1095 static inline int peer_send_hellomsg(struct appctx *appctx, struct peer *peer)
1096 {
1097 struct peer_prep_params p = {
1098 .hello.peer = peer,
1099 };
1100
1101 return peer_send_msg(appctx, peer_prepare_hellomsg, &p);
1102 }
1103
1104 /*
1105 * Send a success peer handshake status message.
1106 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1107 * Returns -1 if there was not enough room left to send the message,
1108 * any other negative returned value must be considered as an error with an appcxt st0
1109 * returned value equal to PEER_SESS_ST_END.
1110 */
peer_send_status_successmsg(struct appctx * appctx)1111 static inline int peer_send_status_successmsg(struct appctx *appctx)
1112 {
1113 return peer_send_msg(appctx, peer_prepare_status_successmsg, NULL);
1114 }
1115
1116 /*
1117 * Send a peer handshake status error message.
1118 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1119 * Returns -1 if there was not enough room left to send the message,
1120 * any other negative returned value must be considered as an error with an appcxt st0
1121 * returned value equal to PEER_SESS_ST_END.
1122 */
peer_send_status_errormsg(struct appctx * appctx)1123 static inline int peer_send_status_errormsg(struct appctx *appctx)
1124 {
1125 struct peer_prep_params p = {
1126 .error_status.st1 = appctx->st1,
1127 };
1128
1129 return peer_send_msg(appctx, peer_prepare_status_errormsg, &p);
1130 }
1131
1132 /*
1133 * Send a stick-table switch message.
1134 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1135 * Returns -1 if there was not enough room left to send the message,
1136 * any other negative returned value must be considered as an error with an appcxt st0
1137 * returned value equal to PEER_SESS_ST_END.
1138 */
peer_send_switchmsg(struct shared_table * st,struct appctx * appctx)1139 static inline int peer_send_switchmsg(struct shared_table *st, struct appctx *appctx)
1140 {
1141 struct peer_prep_params p = {
1142 .swtch.shared_table = st,
1143 };
1144
1145 return peer_send_msg(appctx, peer_prepare_switchmsg, &p);
1146 }
1147
1148 /*
1149 * Send a stick-table update acknowledgement message.
1150 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1151 * Returns -1 if there was not enough room left to send the message,
1152 * any other negative returned value must be considered as an error with an appcxt st0
1153 * returned value equal to PEER_SESS_ST_END.
1154 */
peer_send_ackmsg(struct shared_table * st,struct appctx * appctx)1155 static inline int peer_send_ackmsg(struct shared_table *st, struct appctx *appctx)
1156 {
1157 struct peer_prep_params p = {
1158 .ack.shared_table = st,
1159 };
1160
1161 return peer_send_msg(appctx, peer_prepare_ackmsg, &p);
1162 }
1163
1164 /*
1165 * Send a stick-table update message.
1166 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1167 * Returns -1 if there was not enough room left to send the message,
1168 * any other negative returned value must be considered as an error with an appcxt st0
1169 * returned value equal to PEER_SESS_ST_END.
1170 */
peer_send_updatemsg(struct shared_table * st,struct appctx * appctx,struct stksess * ts,unsigned int updateid,int use_identifier,int use_timed)1171 static inline int peer_send_updatemsg(struct shared_table *st, struct appctx *appctx, struct stksess *ts,
1172 unsigned int updateid, int use_identifier, int use_timed)
1173 {
1174 struct peer_prep_params p = {
1175 .updt = {
1176 .stksess = ts,
1177 .shared_table = st,
1178 .updateid = updateid,
1179 .use_identifier = use_identifier,
1180 .use_timed = use_timed,
1181 .peer = appctx->ctx.peers.ptr,
1182 },
1183 };
1184
1185 return peer_send_msg(appctx, peer_prepare_updatemsg, &p);
1186 }
1187
1188 /*
1189 * Build a peer protocol control class message.
1190 * Returns the number of written bytes used to build the message if succeeded,
1191 * 0 if not.
1192 */
peer_prepare_control_msg(char * msg,size_t size,struct peer_prep_params * p)1193 static int peer_prepare_control_msg(char *msg, size_t size, struct peer_prep_params *p)
1194 {
1195 if (size < sizeof p->control.head)
1196 return 0;
1197
1198 msg[0] = p->control.head[0];
1199 msg[1] = p->control.head[1];
1200
1201 return 2;
1202 }
1203
1204 /*
1205 * Send a stick-table synchronization request message.
1206 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1207 * Returns -1 if there was not enough room left to send the message,
1208 * any other negative returned value must be considered as an error with an appctx st0
1209 * returned value equal to PEER_SESS_ST_END.
1210 */
peer_send_resync_reqmsg(struct appctx * appctx,struct peer * peer,struct peers * peers)1211 static inline int peer_send_resync_reqmsg(struct appctx *appctx,
1212 struct peer *peer, struct peers *peers)
1213 {
1214 struct peer_prep_params p = {
1215 .control.head = { PEER_MSG_CLASS_CONTROL, PEER_MSG_CTRL_RESYNCREQ, },
1216 };
1217
1218 TRACE_PROTO("send control message", PEERS_EV_CTRLMSG,
1219 NULL, &p.control.head[1], peers->local->id, peer->id);
1220
1221 return peer_send_msg(appctx, peer_prepare_control_msg, &p);
1222 }
1223
1224 /*
1225 * Send a stick-table synchronization confirmation message.
1226 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1227 * Returns -1 if there was not enough room left to send the message,
1228 * any other negative returned value must be considered as an error with an appctx st0
1229 * returned value equal to PEER_SESS_ST_END.
1230 */
peer_send_resync_confirmsg(struct appctx * appctx,struct peer * peer,struct peers * peers)1231 static inline int peer_send_resync_confirmsg(struct appctx *appctx,
1232 struct peer *peer, struct peers *peers)
1233 {
1234 struct peer_prep_params p = {
1235 .control.head = { PEER_MSG_CLASS_CONTROL, PEER_MSG_CTRL_RESYNCCONFIRM, },
1236 };
1237
1238 TRACE_PROTO("send control message", PEERS_EV_CTRLMSG,
1239 NULL, &p.control.head[1], peers->local->id, peer->id);
1240
1241 return peer_send_msg(appctx, peer_prepare_control_msg, &p);
1242 }
1243
1244 /*
1245 * Send a stick-table synchronization finished message.
1246 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1247 * Returns -1 if there was not enough room left to send the message,
1248 * any other negative returned value must be considered as an error with an appctx st0
1249 * returned value equal to PEER_SESS_ST_END.
1250 */
peer_send_resync_finishedmsg(struct appctx * appctx,struct peer * peer,struct peers * peers)1251 static inline int peer_send_resync_finishedmsg(struct appctx *appctx,
1252 struct peer *peer, struct peers *peers)
1253 {
1254 struct peer_prep_params p = {
1255 .control.head = { PEER_MSG_CLASS_CONTROL, },
1256 };
1257
1258 p.control.head[1] = (peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FINISHED ?
1259 PEER_MSG_CTRL_RESYNCFINISHED : PEER_MSG_CTRL_RESYNCPARTIAL;
1260
1261 TRACE_PROTO("send control message", PEERS_EV_CTRLMSG,
1262 NULL, &p.control.head[1], peers->local->id, peer->id);
1263
1264 return peer_send_msg(appctx, peer_prepare_control_msg, &p);
1265 }
1266
1267 /*
1268 * Send a heartbeat message.
1269 * Return 0 if the message could not be built modifying the appctx st0 to PEER_SESS_ST_END value.
1270 * Returns -1 if there was not enough room left to send the message,
1271 * any other negative returned value must be considered as an error with an appctx st0
1272 * returned value equal to PEER_SESS_ST_END.
1273 */
peer_send_heartbeatmsg(struct appctx * appctx,struct peer * peer,struct peers * peers)1274 static inline int peer_send_heartbeatmsg(struct appctx *appctx,
1275 struct peer *peer, struct peers *peers)
1276 {
1277 struct peer_prep_params p = {
1278 .control.head = { PEER_MSG_CLASS_CONTROL, PEER_MSG_CTRL_HEARTBEAT, },
1279 };
1280
1281 TRACE_PROTO("send control message", PEERS_EV_CTRLMSG,
1282 NULL, &p.control.head[1], peers->local->id, peer->id);
1283
1284 return peer_send_msg(appctx, peer_prepare_control_msg, &p);
1285 }
1286
1287 /*
1288 * Build a peer protocol error class message.
1289 * Returns the number of written bytes used to build the message if succeeded,
1290 * 0 if not.
1291 */
peer_prepare_error_msg(char * msg,size_t size,struct peer_prep_params * p)1292 static int peer_prepare_error_msg(char *msg, size_t size, struct peer_prep_params *p)
1293 {
1294 if (size < sizeof p->error.head)
1295 return 0;
1296
1297 msg[0] = p->error.head[0];
1298 msg[1] = p->error.head[1];
1299
1300 return 2;
1301 }
1302
1303 /*
1304 * Send a "size limit reached" error message.
1305 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1306 * Returns -1 if there was not enough room left to send the message,
1307 * any other negative returned value must be considered as an error with an appctx st0
1308 * returned value equal to PEER_SESS_ST_END.
1309 */
peer_send_error_size_limitmsg(struct appctx * appctx)1310 static inline int peer_send_error_size_limitmsg(struct appctx *appctx)
1311 {
1312 struct peer_prep_params p = {
1313 .error.head = { PEER_MSG_CLASS_ERROR, PEER_MSG_ERR_SIZELIMIT, },
1314 };
1315
1316 return peer_send_msg(appctx, peer_prepare_error_msg, &p);
1317 }
1318
1319 /*
1320 * Send a "peer protocol" error message.
1321 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1322 * Returns -1 if there was not enough room left to send the message,
1323 * any other negative returned value must be considered as an error with an appctx st0
1324 * returned value equal to PEER_SESS_ST_END.
1325 */
peer_send_error_protomsg(struct appctx * appctx)1326 static inline int peer_send_error_protomsg(struct appctx *appctx)
1327 {
1328 struct peer_prep_params p = {
1329 .error.head = { PEER_MSG_CLASS_ERROR, PEER_MSG_ERR_PROTOCOL, },
1330 };
1331
1332 return peer_send_msg(appctx, peer_prepare_error_msg, &p);
1333 }
1334
1335 /*
1336 * Function used to lookup for recent stick-table updates associated with
1337 * <st> shared stick-table when a lesson must be taught a peer (PEER_F_LEARN_ASSIGN flag set).
1338 */
peer_teach_process_stksess_lookup(struct shared_table * st)1339 static inline struct stksess *peer_teach_process_stksess_lookup(struct shared_table *st)
1340 {
1341 struct eb32_node *eb;
1342
1343 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
1344 if (!eb) {
1345 eb = eb32_first(&st->table->updates);
1346 if (!eb || (eb->key == st->last_pushed)) {
1347 st->table->commitupdate = st->last_pushed = st->table->localupdate;
1348 return NULL;
1349 }
1350 }
1351
1352 /* if distance between the last pushed and the retrieved key
1353 * is greater than the distance last_pushed and the local_update
1354 * this means we are beyond localupdate.
1355 */
1356 if ((eb->key - st->last_pushed) > (st->table->localupdate - st->last_pushed)) {
1357 st->table->commitupdate = st->last_pushed = st->table->localupdate;
1358 return NULL;
1359 }
1360
1361 return eb32_entry(eb, struct stksess, upd);
1362 }
1363
1364 /*
1365 * Function used to lookup for recent stick-table updates associated with
1366 * <st> shared stick-table during teach state 1 step.
1367 */
peer_teach_stage1_stksess_lookup(struct shared_table * st)1368 static inline struct stksess *peer_teach_stage1_stksess_lookup(struct shared_table *st)
1369 {
1370 struct eb32_node *eb;
1371
1372 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
1373 if (!eb) {
1374 st->flags |= SHTABLE_F_TEACH_STAGE1;
1375 eb = eb32_first(&st->table->updates);
1376 if (eb)
1377 st->last_pushed = eb->key - 1;
1378 return NULL;
1379 }
1380
1381 return eb32_entry(eb, struct stksess, upd);
1382 }
1383
1384 /*
1385 * Function used to lookup for recent stick-table updates associated with
1386 * <st> shared stick-table during teach state 2 step.
1387 */
peer_teach_stage2_stksess_lookup(struct shared_table * st)1388 static inline struct stksess *peer_teach_stage2_stksess_lookup(struct shared_table *st)
1389 {
1390 struct eb32_node *eb;
1391
1392 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
1393 if (!eb || eb->key > st->teaching_origin) {
1394 st->flags |= SHTABLE_F_TEACH_STAGE2;
1395 return NULL;
1396 }
1397
1398 return eb32_entry(eb, struct stksess, upd);
1399 }
1400
1401 /*
1402 * Generic function to emit update messages for <st> stick-table when a lesson must
1403 * be taught to the peer <p>.
1404 * <locked> must be set to 1 if the shared table <st> is already locked when entering
1405 * this function, 0 if not.
1406 *
1407 * This function temporary unlock/lock <st> when it sends stick-table updates or
1408 * when decrementing its refcount in case of any error when it sends this updates.
1409 *
1410 * Return 0 if any message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1411 * Returns -1 if there was not enough room left to send the message,
1412 * any other negative returned value must be considered as an error with an appcxt st0
1413 * returned value equal to PEER_SESS_ST_END.
1414 * If it returns 0 or -1, this function leave <st> locked if already locked when entering this function
1415 * unlocked if not already locked when entering this function.
1416 */
peer_send_teachmsgs(struct appctx * appctx,struct peer * p,struct stksess * (* peer_stksess_lookup)(struct shared_table *),struct shared_table * st,int locked)1417 static inline int peer_send_teachmsgs(struct appctx *appctx, struct peer *p,
1418 struct stksess *(*peer_stksess_lookup)(struct shared_table *),
1419 struct shared_table *st, int locked)
1420 {
1421 int ret, new_pushed, use_timed;
1422
1423 ret = 1;
1424 use_timed = 0;
1425 if (st != p->last_local_table) {
1426 ret = peer_send_switchmsg(st, appctx);
1427 if (ret <= 0)
1428 return ret;
1429
1430 p->last_local_table = st;
1431 }
1432
1433 if (peer_stksess_lookup != peer_teach_process_stksess_lookup)
1434 use_timed = !(p->flags & PEER_F_DWNGRD);
1435
1436 /* We force new pushed to 1 to force identifier in update message */
1437 new_pushed = 1;
1438
1439 if (!locked)
1440 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
1441
1442 while (1) {
1443 struct stksess *ts;
1444 unsigned updateid;
1445
1446 /* push local updates */
1447 ts = peer_stksess_lookup(st);
1448 if (!ts)
1449 break;
1450
1451 updateid = ts->upd.key;
1452 ts->ref_cnt++;
1453 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
1454
1455 ret = peer_send_updatemsg(st, appctx, ts, updateid, new_pushed, use_timed);
1456 if (ret <= 0) {
1457 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
1458 ts->ref_cnt--;
1459 if (!locked)
1460 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
1461 return ret;
1462 }
1463
1464 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
1465 ts->ref_cnt--;
1466 st->last_pushed = updateid;
1467
1468 if (peer_stksess_lookup == peer_teach_process_stksess_lookup &&
1469 (int)(st->last_pushed - st->table->commitupdate) > 0)
1470 st->table->commitupdate = st->last_pushed;
1471
1472 /* identifier may not needed in next update message */
1473 new_pushed = 0;
1474 }
1475
1476 out:
1477 if (!locked)
1478 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
1479 return 1;
1480 }
1481
1482 /*
1483 * Function to emit update messages for <st> stick-table when a lesson must
1484 * be taught to the peer <p> (PEER_F_LEARN_ASSIGN flag set).
1485 *
1486 * Note that <st> shared stick-table is locked when calling this function.
1487 *
1488 * Return 0 if any message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1489 * Returns -1 if there was not enough room left to send the message,
1490 * any other negative returned value must be considered as an error with an appcxt st0
1491 * returned value equal to PEER_SESS_ST_END.
1492 */
peer_send_teach_process_msgs(struct appctx * appctx,struct peer * p,struct shared_table * st)1493 static inline int peer_send_teach_process_msgs(struct appctx *appctx, struct peer *p,
1494 struct shared_table *st)
1495 {
1496 return peer_send_teachmsgs(appctx, p, peer_teach_process_stksess_lookup, st, 1);
1497 }
1498
1499 /*
1500 * Function to emit update messages for <st> stick-table when a lesson must
1501 * be taught to the peer <p> during teach state 1 step.
1502 *
1503 * Return 0 if any message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1504 * Returns -1 if there was not enough room left to send the message,
1505 * any other negative returned value must be considered as an error with an appcxt st0
1506 * returned value equal to PEER_SESS_ST_END.
1507 */
peer_send_teach_stage1_msgs(struct appctx * appctx,struct peer * p,struct shared_table * st)1508 static inline int peer_send_teach_stage1_msgs(struct appctx *appctx, struct peer *p,
1509 struct shared_table *st)
1510 {
1511 return peer_send_teachmsgs(appctx, p, peer_teach_stage1_stksess_lookup, st, 0);
1512 }
1513
1514 /*
1515 * Function to emit update messages for <st> stick-table when a lesson must
1516 * be taught to the peer <p> during teach state 1 step.
1517 *
1518 * Return 0 if any message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1519 * Returns -1 if there was not enough room left to send the message,
1520 * any other negative returned value must be considered as an error with an appcxt st0
1521 * returned value equal to PEER_SESS_ST_END.
1522 */
peer_send_teach_stage2_msgs(struct appctx * appctx,struct peer * p,struct shared_table * st)1523 static inline int peer_send_teach_stage2_msgs(struct appctx *appctx, struct peer *p,
1524 struct shared_table *st)
1525 {
1526 return peer_send_teachmsgs(appctx, p, peer_teach_stage2_stksess_lookup, st, 0);
1527 }
1528
1529
1530 /*
1531 * Function used to parse a stick-table update message after it has been received
1532 * by <p> peer with <msg_cur> as address of the pointer to the position in the
1533 * receipt buffer with <msg_end> being position of the end of the stick-table message.
1534 * Update <msg_curr> accordingly to the peer protocol specs if no peer protocol error
1535 * was encountered.
1536 * <exp> must be set if the stick-table entry expires.
1537 * <updt> must be set for PEER_MSG_STKT_UPDATE or PEER_MSG_STKT_UPDATE_TIMED stick-table
1538 * messages, in this case the stick-table update message is received with a stick-table
1539 * update ID.
1540 * <totl> is the length of the stick-table update message computed upon receipt.
1541 */
peer_treat_updatemsg(struct appctx * appctx,struct peer * p,int updt,int exp,char ** msg_cur,char * msg_end,int msg_len,int totl)1542 static int peer_treat_updatemsg(struct appctx *appctx, struct peer *p, int updt, int exp,
1543 char **msg_cur, char *msg_end, int msg_len, int totl)
1544 {
1545 struct stream_interface *si = appctx->owner;
1546 struct shared_table *st = p->remote_table;
1547 struct stksess *ts, *newts;
1548 uint32_t update;
1549 int expire;
1550 unsigned int data_type;
1551 void *data_ptr;
1552
1553 TRACE_ENTER(PEERS_EV_UPDTMSG, NULL, p);
1554 /* Here we have data message */
1555 if (!st)
1556 goto ignore_msg;
1557
1558 expire = MS_TO_TICKS(st->table->expire);
1559
1560 if (updt) {
1561 if (msg_len < sizeof(update)) {
1562 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG, NULL, p);
1563 goto malformed_exit;
1564 }
1565
1566 memcpy(&update, *msg_cur, sizeof(update));
1567 *msg_cur += sizeof(update);
1568 st->last_get = htonl(update);
1569 }
1570 else {
1571 st->last_get++;
1572 }
1573
1574 if (exp) {
1575 size_t expire_sz = sizeof expire;
1576
1577 if (*msg_cur + expire_sz > msg_end) {
1578 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG,
1579 NULL, p, *msg_cur);
1580 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG,
1581 NULL, p, msg_end, &expire_sz);
1582 goto malformed_exit;
1583 }
1584
1585 memcpy(&expire, *msg_cur, expire_sz);
1586 *msg_cur += expire_sz;
1587 expire = ntohl(expire);
1588 }
1589
1590 newts = stksess_new(st->table, NULL);
1591 if (!newts)
1592 goto ignore_msg;
1593
1594 if (st->table->type == SMP_T_STR) {
1595 unsigned int to_read, to_store;
1596
1597 to_read = intdecode(msg_cur, msg_end);
1598 if (!*msg_cur) {
1599 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG, NULL, p);
1600 goto malformed_free_newts;
1601 }
1602
1603 to_store = MIN(to_read, st->table->key_size - 1);
1604 if (*msg_cur + to_store > msg_end) {
1605 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG,
1606 NULL, p, *msg_cur);
1607 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG,
1608 NULL, p, msg_end, &to_store);
1609 goto malformed_free_newts;
1610 }
1611
1612 memcpy(newts->key.key, *msg_cur, to_store);
1613 newts->key.key[to_store] = 0;
1614 *msg_cur += to_read;
1615 }
1616 else if (st->table->type == SMP_T_SINT) {
1617 unsigned int netinteger;
1618
1619 if (*msg_cur + sizeof(netinteger) > msg_end) {
1620 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG,
1621 NULL, p, *msg_cur);
1622 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG,
1623 NULL, p, msg_end);
1624 goto malformed_free_newts;
1625 }
1626
1627 memcpy(&netinteger, *msg_cur, sizeof(netinteger));
1628 netinteger = ntohl(netinteger);
1629 memcpy(newts->key.key, &netinteger, sizeof(netinteger));
1630 *msg_cur += sizeof(netinteger);
1631 }
1632 else {
1633 if (*msg_cur + st->table->key_size > msg_end) {
1634 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG,
1635 NULL, p, *msg_cur);
1636 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG,
1637 NULL, p, msg_end, &st->table->key_size);
1638 goto malformed_free_newts;
1639 }
1640
1641 memcpy(newts->key.key, *msg_cur, st->table->key_size);
1642 *msg_cur += st->table->key_size;
1643 }
1644
1645 /* lookup for existing entry */
1646 ts = stktable_set_entry(st->table, newts);
1647 if (ts != newts) {
1648 stksess_free(st->table, newts);
1649 newts = NULL;
1650 }
1651
1652 HA_RWLOCK_WRLOCK(STK_SESS_LOCK, &ts->lock);
1653
1654 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
1655 uint64_t decoded_int;
1656
1657 if (!((1ULL << data_type) & st->remote_data))
1658 continue;
1659
1660 decoded_int = intdecode(msg_cur, msg_end);
1661 if (!*msg_cur) {
1662 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG, NULL, p);
1663 goto malformed_unlock;
1664 }
1665
1666 switch (stktable_data_types[data_type].std_type) {
1667 case STD_T_SINT:
1668 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1669 if (data_ptr)
1670 stktable_data_cast(data_ptr, std_t_sint) = decoded_int;
1671 break;
1672
1673 case STD_T_UINT:
1674 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1675 if (data_ptr)
1676 stktable_data_cast(data_ptr, std_t_uint) = decoded_int;
1677 break;
1678
1679 case STD_T_ULL:
1680 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1681 if (data_ptr)
1682 stktable_data_cast(data_ptr, std_t_ull) = decoded_int;
1683 break;
1684
1685 case STD_T_FRQP: {
1686 struct freq_ctr_period data;
1687
1688 /* First bit is reserved for the freq_ctr_period lock
1689 Note: here we're still protected by the stksess lock
1690 so we don't need to update the update the freq_ctr_period
1691 using its internal lock */
1692
1693 data.curr_tick = tick_add(now_ms, -decoded_int) & ~0x1;
1694 data.curr_ctr = intdecode(msg_cur, msg_end);
1695 if (!*msg_cur) {
1696 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG, NULL, p);
1697 goto malformed_unlock;
1698 }
1699
1700 data.prev_ctr = intdecode(msg_cur, msg_end);
1701 if (!*msg_cur) {
1702 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG, NULL, p);
1703 goto malformed_unlock;
1704 }
1705
1706 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1707 if (data_ptr)
1708 stktable_data_cast(data_ptr, std_t_frqp) = data;
1709 break;
1710 }
1711 case STD_T_DICT: {
1712 struct buffer *chunk;
1713 size_t data_len, value_len;
1714 unsigned int id;
1715 struct dict_entry *de;
1716 struct dcache *dc;
1717 char *end;
1718
1719 if (!decoded_int) {
1720 /* No entry. */
1721 break;
1722 }
1723 data_len = decoded_int;
1724 if (*msg_cur + data_len > msg_end) {
1725 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG,
1726 NULL, p, *msg_cur);
1727 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG,
1728 NULL, p, msg_end, &data_len);
1729 goto malformed_unlock;
1730 }
1731
1732 /* Compute the end of the current data, <msg_end> being at the end of
1733 * the entire message.
1734 */
1735 end = *msg_cur + data_len;
1736 id = intdecode(msg_cur, end);
1737 if (!*msg_cur || !id) {
1738 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG,
1739 NULL, p, *msg_cur, &id);
1740 goto malformed_unlock;
1741 }
1742
1743 dc = p->dcache;
1744 if (*msg_cur == end) {
1745 /* Dictionary entry key without value. */
1746 if (id > dc->max_entries) {
1747 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG,
1748 NULL, p, NULL, &id);
1749 goto malformed_unlock;
1750 }
1751 /* IDs sent over the network are numbered from 1. */
1752 de = dc->rx[id - 1].de;
1753 }
1754 else {
1755 chunk = get_trash_chunk();
1756 value_len = intdecode(msg_cur, end);
1757 if (!*msg_cur || *msg_cur + value_len > end ||
1758 unlikely(value_len + 1 >= chunk->size)) {
1759 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG,
1760 NULL, p, *msg_cur, &value_len);
1761 TRACE_PROTO("malformed message", PEERS_EV_UPDTMSG,
1762 NULL, p, end, &chunk->size);
1763 goto malformed_unlock;
1764 }
1765
1766 chunk_memcpy(chunk, *msg_cur, value_len);
1767 chunk->area[chunk->data] = '\0';
1768 *msg_cur += value_len;
1769
1770 de = dict_insert(&server_name_dict, chunk->area);
1771 dc->rx[id - 1].de = de;
1772 }
1773 if (de) {
1774 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1775 if (data_ptr)
1776 stktable_data_cast(data_ptr, std_t_dict) = de;
1777 }
1778 break;
1779 }
1780 }
1781 }
1782 /* Force new expiration */
1783 ts->expire = tick_add(now_ms, expire);
1784
1785 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
1786 stktable_touch_remote(st->table, ts, 1);
1787 TRACE_LEAVE(PEERS_EV_UPDTMSG, NULL, p);
1788 return 1;
1789
1790 ignore_msg:
1791 /* skip consumed message */
1792 co_skip(si_oc(si), totl);
1793 TRACE_DEVEL("leaving in error", PEERS_EV_UPDTMSG);
1794 return 0;
1795
1796 malformed_unlock:
1797 /* malformed message */
1798 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
1799 stktable_touch_remote(st->table, ts, 1);
1800 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1801 TRACE_DEVEL("leaving in error", PEERS_EV_UPDTMSG);
1802 return 0;
1803
1804 malformed_free_newts:
1805 /* malformed message */
1806 stksess_free(st->table, newts);
1807 malformed_exit:
1808 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1809 TRACE_DEVEL("leaving in error", PEERS_EV_UPDTMSG);
1810 return 0;
1811 }
1812
1813 /*
1814 * Function used to parse a stick-table update acknowledgement message after it
1815 * has been received by <p> peer with <msg_cur> as address of the pointer to the position in the
1816 * receipt buffer with <msg_end> being the position of the end of the stick-table message.
1817 * Update <msg_curr> accordingly to the peer protocol specs if no peer protocol error
1818 * was encountered.
1819 * Return 1 if succeeded, 0 if not with the appctx state st0 set to PEER_SESS_ST_ERRPROTO.
1820 */
peer_treat_ackmsg(struct appctx * appctx,struct peer * p,char ** msg_cur,char * msg_end)1821 static inline int peer_treat_ackmsg(struct appctx *appctx, struct peer *p,
1822 char **msg_cur, char *msg_end)
1823 {
1824 /* ack message */
1825 uint32_t table_id ;
1826 uint32_t update;
1827 struct shared_table *st;
1828
1829 /* ignore ack during teaching process */
1830 if (p->flags & PEER_F_TEACH_PROCESS)
1831 return 1;
1832
1833 table_id = intdecode(msg_cur, msg_end);
1834 if (!*msg_cur || (*msg_cur + sizeof(update) > msg_end)) {
1835 /* malformed message */
1836
1837 TRACE_PROTO("malformed message", PEERS_EV_ACKMSG,
1838 NULL, p, *msg_cur);
1839 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1840 return 0;
1841 }
1842
1843 memcpy(&update, *msg_cur, sizeof(update));
1844 update = ntohl(update);
1845
1846 for (st = p->tables; st; st = st->next) {
1847 if (st->local_id == table_id) {
1848 st->update = update;
1849 break;
1850 }
1851 }
1852
1853 return 1;
1854 }
1855
1856 /*
1857 * Function used to parse a stick-table switch message after it has been received
1858 * by <p> peer with <msg_cur> as address of the pointer to the position in the
1859 * receipt buffer with <msg_end> being the position of the end of the stick-table message.
1860 * Update <msg_curr> accordingly to the peer protocol specs if no peer protocol error
1861 * was encountered.
1862 * Return 1 if succeeded, 0 if not with the appctx state st0 set to PEER_SESS_ST_ERRPROTO.
1863 */
peer_treat_switchmsg(struct appctx * appctx,struct peer * p,char ** msg_cur,char * msg_end)1864 static inline int peer_treat_switchmsg(struct appctx *appctx, struct peer *p,
1865 char **msg_cur, char *msg_end)
1866 {
1867 struct shared_table *st;
1868 int table_id;
1869
1870 table_id = intdecode(msg_cur, msg_end);
1871 if (!*msg_cur) {
1872 TRACE_PROTO("malformed message", PEERS_EV_SWTCMSG, NULL, p);
1873 /* malformed message */
1874 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1875 return 0;
1876 }
1877
1878 p->remote_table = NULL;
1879 for (st = p->tables; st; st = st->next) {
1880 if (st->remote_id == table_id) {
1881 p->remote_table = st;
1882 break;
1883 }
1884 }
1885
1886 return 1;
1887 }
1888
1889 /*
1890 * Function used to parse a stick-table definition message after it has been received
1891 * by <p> peer with <msg_cur> as address of the pointer to the position in the
1892 * receipt buffer with <msg_end> being the position of the end of the stick-table message.
1893 * Update <msg_curr> accordingly to the peer protocol specs if no peer protocol error
1894 * was encountered.
1895 * <totl> is the length of the stick-table update message computed upon receipt.
1896 * Return 1 if succeeded, 0 if not with the appctx state st0 set to PEER_SESS_ST_ERRPROTO.
1897 */
peer_treat_definemsg(struct appctx * appctx,struct peer * p,char ** msg_cur,char * msg_end,int totl)1898 static inline int peer_treat_definemsg(struct appctx *appctx, struct peer *p,
1899 char **msg_cur, char *msg_end, int totl)
1900 {
1901 struct stream_interface *si = appctx->owner;
1902 int table_id_len;
1903 struct shared_table *st;
1904 int table_type;
1905 int table_keylen;
1906 int table_id;
1907 uint64_t table_data;
1908
1909 table_id = intdecode(msg_cur, msg_end);
1910 if (!*msg_cur) {
1911 TRACE_PROTO("malformed message", PEERS_EV_DEFMSG, NULL, p);
1912 goto malformed_exit;
1913 }
1914
1915 table_id_len = intdecode(msg_cur, msg_end);
1916 if (!*msg_cur) {
1917 TRACE_PROTO("malformed message", PEERS_EV_DEFMSG, NULL, p, *msg_cur);
1918 goto malformed_exit;
1919 }
1920
1921 p->remote_table = NULL;
1922 if (!table_id_len || (*msg_cur + table_id_len) >= msg_end) {
1923 TRACE_PROTO("malformed message", PEERS_EV_DEFMSG, NULL, p, *msg_cur, &table_id_len);
1924 goto malformed_exit;
1925 }
1926
1927 for (st = p->tables; st; st = st->next) {
1928 /* Reset IDs */
1929 if (st->remote_id == table_id)
1930 st->remote_id = 0;
1931
1932 if (!p->remote_table && (table_id_len == strlen(st->table->nid)) &&
1933 (memcmp(st->table->nid, *msg_cur, table_id_len) == 0))
1934 p->remote_table = st;
1935 }
1936
1937 if (!p->remote_table) {
1938 TRACE_PROTO("ignored message", PEERS_EV_DEFMSG, NULL, p);
1939 goto ignore_msg;
1940 }
1941
1942 *msg_cur += table_id_len;
1943 if (*msg_cur >= msg_end) {
1944 TRACE_PROTO("malformed message", PEERS_EV_DEFMSG, NULL, p);
1945 goto malformed_exit;
1946 }
1947
1948 table_type = intdecode(msg_cur, msg_end);
1949 if (!*msg_cur) {
1950 TRACE_PROTO("malformed message", PEERS_EV_DEFMSG, NULL, p);
1951 goto malformed_exit;
1952 }
1953
1954 table_keylen = intdecode(msg_cur, msg_end);
1955 if (!*msg_cur) {
1956 TRACE_PROTO("malformed message", PEERS_EV_DEFMSG, NULL, p);
1957 goto malformed_exit;
1958 }
1959
1960 table_data = intdecode(msg_cur, msg_end);
1961 if (!*msg_cur) {
1962 TRACE_PROTO("malformed message", PEERS_EV_DEFMSG, NULL, p);
1963 goto malformed_exit;
1964 }
1965
1966 if (p->remote_table->table->type != peer_int_key_type[table_type]
1967 || p->remote_table->table->key_size != table_keylen) {
1968 p->remote_table = NULL;
1969 TRACE_PROTO("ignored message", PEERS_EV_DEFMSG, NULL, p);
1970 goto ignore_msg;
1971 }
1972
1973 p->remote_table->remote_data = table_data;
1974 p->remote_table->remote_id = table_id;
1975 return 1;
1976
1977 ignore_msg:
1978 co_skip(si_oc(si), totl);
1979 return 0;
1980
1981 malformed_exit:
1982 /* malformed message */
1983 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1984 return 0;
1985 }
1986
1987 /*
1988 * Receive a stick-table message or pre-parse any other message.
1989 * The message's header will be sent into <msg_head> which must be at least
1990 * <msg_head_sz> bytes long (at least 7 to store 32-bit variable lengths).
1991 * The first two bytes are always read, and the rest is only read if the
1992 * first bytes indicate a stick-table message. If the message is a stick-table
1993 * message, the varint is decoded and the equivalent number of bytes will be
1994 * copied into the trash at trash.area. <totl> is incremented by the number of
1995 * bytes read EVEN IN CASE OF INCOMPLETE MESSAGES.
1996 * Returns 1 if there was no error, if not, returns 0 if not enough data were available,
1997 * -1 if there was an error updating the appctx state st0 accordingly.
1998 */
peer_recv_msg(struct appctx * appctx,char * msg_head,size_t msg_head_sz,uint32_t * msg_len,int * totl)1999 static inline int peer_recv_msg(struct appctx *appctx, char *msg_head, size_t msg_head_sz,
2000 uint32_t *msg_len, int *totl)
2001 {
2002 int reql;
2003 struct stream_interface *si = appctx->owner;
2004 char *cur;
2005
2006 reql = co_getblk(si_oc(si), msg_head, 2 * sizeof(char), *totl);
2007 if (reql <= 0) /* closed or EOL not found */
2008 goto incomplete;
2009
2010 *totl += reql;
2011
2012 if (!(msg_head[1] & PEER_MSG_STKT_BIT_MASK))
2013 return 1;
2014
2015 /* This is a stick-table message, let's go on */
2016
2017 /* Read and Decode message length */
2018 msg_head += *totl;
2019 msg_head_sz -= *totl;
2020 reql = co_data(si_oc(si)) - *totl;
2021 if (reql > msg_head_sz)
2022 reql = msg_head_sz;
2023
2024 reql = co_getblk(si_oc(si), msg_head, reql, *totl);
2025 if (reql <= 0) /* closed */
2026 goto incomplete;
2027
2028 cur = msg_head;
2029 *msg_len = intdecode(&cur, cur + reql);
2030 if (!cur) {
2031 /* the number is truncated, did we read enough ? */
2032 if (reql < msg_head_sz)
2033 goto incomplete;
2034
2035 /* malformed message */
2036 TRACE_PROTO("malformed message: too large length encoding", PEERS_EV_UPDTMSG);
2037 appctx->st0 = PEER_SESS_ST_ERRPROTO;
2038 return -1;
2039 }
2040 *totl += cur - msg_head;
2041
2042 /* Read message content */
2043 if (*msg_len) {
2044 if (*msg_len > trash.size) {
2045 /* Status code is not success, abort */
2046 appctx->st0 = PEER_SESS_ST_ERRSIZE;
2047 return -1;
2048 }
2049
2050 reql = co_getblk(si_oc(si), trash.area, *msg_len, *totl);
2051 if (reql <= 0) /* closed */
2052 goto incomplete;
2053 *totl += reql;
2054 }
2055
2056 return 1;
2057
2058 incomplete:
2059 if (reql < 0 || (si_oc(si)->flags & (CF_SHUTW|CF_SHUTW_NOW))) {
2060 /* there was an error or the message was truncated */
2061 appctx->st0 = PEER_SESS_ST_END;
2062 return -1;
2063 }
2064
2065 return 0;
2066 }
2067
2068 /*
2069 * Treat the awaited message with <msg_head> as header.*
2070 * Return 1 if succeeded, 0 if not.
2071 */
peer_treat_awaited_msg(struct appctx * appctx,struct peer * peer,unsigned char * msg_head,char ** msg_cur,char * msg_end,int msg_len,int totl)2072 static inline int peer_treat_awaited_msg(struct appctx *appctx, struct peer *peer, unsigned char *msg_head,
2073 char **msg_cur, char *msg_end, int msg_len, int totl)
2074 {
2075 struct stream_interface *si = appctx->owner;
2076 struct stream *s = si_strm(si);
2077 struct peers *peers = strm_fe(s)->parent;
2078
2079 if (msg_head[0] == PEER_MSG_CLASS_CONTROL) {
2080 if (msg_head[1] == PEER_MSG_CTRL_RESYNCREQ) {
2081 struct shared_table *st;
2082 /* Reset message: remote need resync */
2083
2084 TRACE_PROTO("received control message", PEERS_EV_CTRLMSG,
2085 NULL, &msg_head[1], peers->local->id, peer->id);
2086 /* prepare tables for a global push */
2087 for (st = peer->tables; st; st = st->next) {
2088 st->teaching_origin = st->last_pushed = st->update;
2089 st->flags = 0;
2090 }
2091
2092 /* reset teaching flags to 0 */
2093 peer->flags &= PEER_TEACH_RESET;
2094
2095 /* flag to start to teach lesson */
2096 peer->flags |= PEER_F_TEACH_PROCESS;
2097 peers->flags |= PEERS_F_RESYNC_REQUESTED;
2098 }
2099 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCFINISHED) {
2100 TRACE_PROTO("received control message", PEERS_EV_CTRLMSG,
2101 NULL, &msg_head[1], peers->local->id, peer->id);
2102 if (peer->flags & PEER_F_LEARN_ASSIGN) {
2103 peer->flags &= ~PEER_F_LEARN_ASSIGN;
2104 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
2105 peers->flags |= (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE);
2106 if (peer->local)
2107 peers->flags |= PEERS_F_RESYNC_LOCALFINISHED;
2108 else
2109 peers->flags |= PEERS_F_RESYNC_REMOTEFINISHED;
2110 }
2111 peer->confirm++;
2112 }
2113 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCPARTIAL) {
2114 TRACE_PROTO("received control message", PEERS_EV_CTRLMSG,
2115 NULL, &msg_head[1], peers->local->id, peer->id);
2116 if (peer->flags & PEER_F_LEARN_ASSIGN) {
2117 peer->flags &= ~PEER_F_LEARN_ASSIGN;
2118 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
2119
2120 if (peer->local)
2121 peers->flags |= PEERS_F_RESYNC_LOCALPARTIAL;
2122 else
2123 peers->flags |= PEERS_F_RESYNC_REMOTEPARTIAL;
2124 peer->flags |= PEER_F_LEARN_NOTUP2DATE;
2125 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(PEER_RESYNC_TIMEOUT));
2126 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
2127 }
2128 peer->confirm++;
2129 }
2130 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCCONFIRM) {
2131 struct shared_table *st;
2132
2133 TRACE_PROTO("received control message", PEERS_EV_CTRLMSG,
2134 NULL, &msg_head[1], peers->local->id, peer->id);
2135 /* If stopping state */
2136 if (stopping) {
2137 /* Close session, push resync no more needed */
2138 peer->flags |= PEER_F_TEACH_COMPLETE;
2139 appctx->st0 = PEER_SESS_ST_END;
2140 return 0;
2141 }
2142 for (st = peer->tables; st; st = st->next) {
2143 st->update = st->last_pushed = st->teaching_origin;
2144 st->flags = 0;
2145 }
2146
2147 /* reset teaching flags to 0 */
2148 peer->flags &= PEER_TEACH_RESET;
2149 }
2150 else if (msg_head[1] == PEER_MSG_CTRL_HEARTBEAT) {
2151 TRACE_PROTO("received control message", PEERS_EV_CTRLMSG,
2152 NULL, &msg_head[1], peers->local->id, peer->id);
2153 peer->reconnect = tick_add(now_ms, MS_TO_TICKS(PEER_RECONNECT_TIMEOUT));
2154 peer->rx_hbt++;
2155 }
2156 }
2157 else if (msg_head[0] == PEER_MSG_CLASS_STICKTABLE) {
2158 if (msg_head[1] == PEER_MSG_STKT_DEFINE) {
2159 if (!peer_treat_definemsg(appctx, peer, msg_cur, msg_end, totl))
2160 return 0;
2161 }
2162 else if (msg_head[1] == PEER_MSG_STKT_SWITCH) {
2163 if (!peer_treat_switchmsg(appctx, peer, msg_cur, msg_end))
2164 return 0;
2165 }
2166 else if (msg_head[1] == PEER_MSG_STKT_UPDATE ||
2167 msg_head[1] == PEER_MSG_STKT_INCUPDATE ||
2168 msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED ||
2169 msg_head[1] == PEER_MSG_STKT_INCUPDATE_TIMED) {
2170 int update, expire;
2171
2172 update = msg_head[1] == PEER_MSG_STKT_UPDATE || msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED;
2173 expire = msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED || msg_head[1] == PEER_MSG_STKT_INCUPDATE_TIMED;
2174 if (!peer_treat_updatemsg(appctx, peer, update, expire,
2175 msg_cur, msg_end, msg_len, totl))
2176 return 0;
2177
2178 }
2179 else if (msg_head[1] == PEER_MSG_STKT_ACK) {
2180 if (!peer_treat_ackmsg(appctx, peer, msg_cur, msg_end))
2181 return 0;
2182 }
2183 }
2184 else if (msg_head[0] == PEER_MSG_CLASS_RESERVED) {
2185 appctx->st0 = PEER_SESS_ST_ERRPROTO;
2186 return 0;
2187 }
2188
2189 return 1;
2190 }
2191
2192
2193 /*
2194 * Send any message to <peer> peer.
2195 * Returns 1 if succeeded, or -1 or 0 if failed.
2196 * -1 means an internal error occurred, 0 is for a peer protocol error leading
2197 * to a peer state change (from the peer I/O handler point of view).
2198 */
peer_send_msgs(struct appctx * appctx,struct peer * peer,struct peers * peers)2199 static inline int peer_send_msgs(struct appctx *appctx,
2200 struct peer *peer, struct peers *peers)
2201 {
2202 int repl;
2203
2204 /* Need to request a resync */
2205 if ((peer->flags & PEER_F_LEARN_ASSIGN) &&
2206 (peers->flags & PEERS_F_RESYNC_ASSIGN) &&
2207 !(peers->flags & PEERS_F_RESYNC_PROCESS)) {
2208
2209 repl = peer_send_resync_reqmsg(appctx, peer, peers);
2210 if (repl <= 0)
2211 return repl;
2212
2213 peers->flags |= PEERS_F_RESYNC_PROCESS;
2214 }
2215
2216 /* Nothing to read, now we start to write */
2217 if (peer->tables) {
2218 struct shared_table *st;
2219 struct shared_table *last_local_table;
2220
2221 last_local_table = peer->last_local_table;
2222 if (!last_local_table)
2223 last_local_table = peer->tables;
2224 st = last_local_table->next;
2225
2226 while (1) {
2227 if (!st)
2228 st = peer->tables;
2229
2230 /* It remains some updates to ack */
2231 if (st->last_get != st->last_acked) {
2232 repl = peer_send_ackmsg(st, appctx);
2233 if (repl <= 0)
2234 return repl;
2235
2236 st->last_acked = st->last_get;
2237 }
2238
2239 if (!(peer->flags & PEER_F_TEACH_PROCESS)) {
2240 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
2241 if (!(peer->flags & PEER_F_LEARN_ASSIGN) &&
2242 (st->last_pushed != st->table->localupdate)) {
2243
2244 repl = peer_send_teach_process_msgs(appctx, peer, st);
2245 if (repl <= 0) {
2246 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
2247 return repl;
2248 }
2249 }
2250 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
2251 }
2252 else if (!(peer->flags & PEER_F_TEACH_FINISHED)) {
2253 if (!(st->flags & SHTABLE_F_TEACH_STAGE1)) {
2254 repl = peer_send_teach_stage1_msgs(appctx, peer, st);
2255 if (repl <= 0)
2256 return repl;
2257 }
2258
2259 if (!(st->flags & SHTABLE_F_TEACH_STAGE2)) {
2260 repl = peer_send_teach_stage2_msgs(appctx, peer, st);
2261 if (repl <= 0)
2262 return repl;
2263 }
2264 }
2265
2266 if (st == last_local_table)
2267 break;
2268 st = st->next;
2269 }
2270 }
2271
2272 if ((peer->flags & PEER_F_TEACH_PROCESS) && !(peer->flags & PEER_F_TEACH_FINISHED)) {
2273 repl = peer_send_resync_finishedmsg(appctx, peer, peers);
2274 if (repl <= 0)
2275 return repl;
2276
2277 /* flag finished message sent */
2278 peer->flags |= PEER_F_TEACH_FINISHED;
2279 }
2280
2281 /* Confirm finished or partial messages */
2282 while (peer->confirm) {
2283 repl = peer_send_resync_confirmsg(appctx, peer, peers);
2284 if (repl <= 0)
2285 return repl;
2286
2287 peer->confirm--;
2288 }
2289
2290 return 1;
2291 }
2292
2293 /*
2294 * Read and parse a first line of a "hello" peer protocol message.
2295 * Returns 0 if could not read a line, -1 if there was a read error or
2296 * the line is malformed, 1 if succeeded.
2297 */
peer_getline_version(struct appctx * appctx,unsigned int * maj_ver,unsigned int * min_ver)2298 static inline int peer_getline_version(struct appctx *appctx,
2299 unsigned int *maj_ver, unsigned int *min_ver)
2300 {
2301 int reql;
2302
2303 reql = peer_getline(appctx);
2304 if (!reql)
2305 return 0;
2306
2307 if (reql < 0)
2308 return -1;
2309
2310 /* test protocol */
2311 if (strncmp(PEER_SESSION_PROTO_NAME " ", trash.area, proto_len + 1) != 0) {
2312 appctx->st0 = PEER_SESS_ST_EXIT;
2313 appctx->st1 = PEER_SESS_SC_ERRPROTO;
2314 return -1;
2315 }
2316 if (peer_get_version(trash.area + proto_len + 1, maj_ver, min_ver) == -1 ||
2317 *maj_ver != PEER_MAJOR_VER || *min_ver > PEER_MINOR_VER) {
2318 appctx->st0 = PEER_SESS_ST_EXIT;
2319 appctx->st1 = PEER_SESS_SC_ERRVERSION;
2320 return -1;
2321 }
2322
2323 return 1;
2324 }
2325
2326 /*
2327 * Read and parse a second line of a "hello" peer protocol message.
2328 * Returns 0 if could not read a line, -1 if there was a read error or
2329 * the line is malformed, 1 if succeeded.
2330 */
peer_getline_host(struct appctx * appctx)2331 static inline int peer_getline_host(struct appctx *appctx)
2332 {
2333 int reql;
2334
2335 reql = peer_getline(appctx);
2336 if (!reql)
2337 return 0;
2338
2339 if (reql < 0)
2340 return -1;
2341
2342 /* test hostname match */
2343 if (strcmp(localpeer, trash.area) != 0) {
2344 appctx->st0 = PEER_SESS_ST_EXIT;
2345 appctx->st1 = PEER_SESS_SC_ERRHOST;
2346 return -1;
2347 }
2348
2349 return 1;
2350 }
2351
2352 /*
2353 * Read and parse a last line of a "hello" peer protocol message.
2354 * Returns 0 if could not read a character, -1 if there was a read error or
2355 * the line is malformed, 1 if succeeded.
2356 * Set <curpeer> accordingly (the remote peer sending the "hello" message).
2357 */
peer_getline_last(struct appctx * appctx,struct peer ** curpeer)2358 static inline int peer_getline_last(struct appctx *appctx, struct peer **curpeer)
2359 {
2360 char *p;
2361 int reql;
2362 struct peer *peer;
2363 struct stream_interface *si = appctx->owner;
2364 struct stream *s = si_strm(si);
2365 struct peers *peers = strm_fe(s)->parent;
2366
2367 reql = peer_getline(appctx);
2368 if (!reql)
2369 return 0;
2370
2371 if (reql < 0)
2372 return -1;
2373
2374 /* parse line "<peer name> <pid> <relative_pid>" */
2375 p = strchr(trash.area, ' ');
2376 if (!p) {
2377 appctx->st0 = PEER_SESS_ST_EXIT;
2378 appctx->st1 = PEER_SESS_SC_ERRPROTO;
2379 return -1;
2380 }
2381 *p = 0;
2382
2383 /* lookup known peer */
2384 for (peer = peers->remote; peer; peer = peer->next) {
2385 if (strcmp(peer->id, trash.area) == 0)
2386 break;
2387 }
2388
2389 /* if unknown peer */
2390 if (!peer) {
2391 appctx->st0 = PEER_SESS_ST_EXIT;
2392 appctx->st1 = PEER_SESS_SC_ERRPEER;
2393 return -1;
2394 }
2395 *curpeer = peer;
2396
2397 return 1;
2398 }
2399
2400 /*
2401 * Init <peer> peer after having accepted it at peer protocol level.
2402 */
init_accepted_peer(struct peer * peer,struct peers * peers)2403 static inline void init_accepted_peer(struct peer *peer, struct peers *peers)
2404 {
2405 struct shared_table *st;
2406
2407 peer->heartbeat = tick_add(now_ms, MS_TO_TICKS(PEER_HEARTBEAT_TIMEOUT));
2408 /* Register status code */
2409 peer->statuscode = PEER_SESS_SC_SUCCESSCODE;
2410 peer->last_hdshk = now_ms;
2411
2412 /* Awake main task */
2413 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
2414
2415 /* Init confirm counter */
2416 peer->confirm = 0;
2417
2418 /* Init cursors */
2419 for (st = peer->tables; st ; st = st->next) {
2420 st->last_get = st->last_acked = 0;
2421 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
2422 /* if st->update appears to be in future it means
2423 * that the last acked value is very old and we
2424 * remain unconnected a too long time to use this
2425 * acknowlegement as a reset.
2426 * We should update the protocol to be able to
2427 * signal the remote peer that it needs a full resync.
2428 * Here a partial fix consist to set st->update at
2429 * the max past value
2430 */
2431 if ((int)(st->table->localupdate - st->update) < 0)
2432 st->update = st->table->localupdate + (2147483648U);
2433 st->teaching_origin = st->last_pushed = st->update;
2434 st->flags = 0;
2435 if ((int)(st->last_pushed - st->table->commitupdate) > 0)
2436 st->table->commitupdate = st->last_pushed;
2437 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
2438 }
2439
2440 /* reset teaching and learning flags to 0 */
2441 peer->flags &= PEER_TEACH_RESET;
2442 peer->flags &= PEER_LEARN_RESET;
2443
2444 /* if current peer is local */
2445 if (peer->local) {
2446 /* if current host need resyncfrom local and no process assigned */
2447 if ((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL &&
2448 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
2449 /* assign local peer for a lesson, consider lesson already requested */
2450 peer->flags |= PEER_F_LEARN_ASSIGN;
2451 peers->flags |= (PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
2452 peers->flags |= PEERS_F_RESYNC_LOCALASSIGN;
2453 }
2454
2455 }
2456 else if ((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
2457 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
2458 /* assign peer for a lesson */
2459 peer->flags |= PEER_F_LEARN_ASSIGN;
2460 peers->flags |= PEERS_F_RESYNC_ASSIGN;
2461 peers->flags |= PEERS_F_RESYNC_REMOTEASSIGN;
2462 }
2463 }
2464
2465 /*
2466 * Init <peer> peer after having connected it at peer protocol level.
2467 */
init_connected_peer(struct peer * peer,struct peers * peers)2468 static inline void init_connected_peer(struct peer *peer, struct peers *peers)
2469 {
2470 struct shared_table *st;
2471
2472 peer->heartbeat = tick_add(now_ms, MS_TO_TICKS(PEER_HEARTBEAT_TIMEOUT));
2473 /* Init cursors */
2474 for (st = peer->tables; st ; st = st->next) {
2475 st->last_get = st->last_acked = 0;
2476 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
2477 /* if st->update appears to be in future it means
2478 * that the last acked value is very old and we
2479 * remain unconnected a too long time to use this
2480 * acknowlegement as a reset.
2481 * We should update the protocol to be able to
2482 * signal the remote peer that it needs a full resync.
2483 * Here a partial fix consist to set st->update at
2484 * the max past value.
2485 */
2486 if ((int)(st->table->localupdate - st->update) < 0)
2487 st->update = st->table->localupdate + (2147483648U);
2488 st->teaching_origin = st->last_pushed = st->update;
2489 st->flags = 0;
2490 if ((int)(st->last_pushed - st->table->commitupdate) > 0)
2491 st->table->commitupdate = st->last_pushed;
2492 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
2493 }
2494
2495 /* Init confirm counter */
2496 peer->confirm = 0;
2497
2498 /* reset teaching and learning flags to 0 */
2499 peer->flags &= PEER_TEACH_RESET;
2500 peer->flags &= PEER_LEARN_RESET;
2501
2502 /* If current peer is local */
2503 if (peer->local) {
2504 /* flag to start to teach lesson */
2505 peer->flags |= PEER_F_TEACH_PROCESS;
2506 }
2507 else if ((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
2508 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
2509 /* If peer is remote and resync from remote is needed,
2510 and no peer currently assigned */
2511
2512 /* assign peer for a lesson */
2513 peer->flags |= PEER_F_LEARN_ASSIGN;
2514 peers->flags |= PEERS_F_RESYNC_ASSIGN;
2515 peers->flags |= PEERS_F_RESYNC_REMOTEASSIGN;
2516 }
2517 }
2518
2519 /*
2520 * IO Handler to handle message exchange with a peer
2521 */
peer_io_handler(struct appctx * appctx)2522 static void peer_io_handler(struct appctx *appctx)
2523 {
2524 struct stream_interface *si = appctx->owner;
2525 struct stream *s = si_strm(si);
2526 struct peers *curpeers = strm_fe(s)->parent;
2527 struct peer *curpeer = NULL;
2528 int reql = 0;
2529 int repl = 0;
2530 unsigned int maj_ver, min_ver;
2531 int prev_state;
2532
2533 /* Check if the input buffer is available. */
2534 if (si_ic(si)->buf.size == 0) {
2535 si_rx_room_blk(si);
2536 goto out;
2537 }
2538
2539 while (1) {
2540 prev_state = appctx->st0;
2541 switchstate:
2542 maj_ver = min_ver = (unsigned int)-1;
2543 switch(appctx->st0) {
2544 case PEER_SESS_ST_ACCEPT:
2545 prev_state = appctx->st0;
2546 appctx->ctx.peers.ptr = NULL;
2547 appctx->st0 = PEER_SESS_ST_GETVERSION;
2548 /* fall through */
2549 case PEER_SESS_ST_GETVERSION:
2550 prev_state = appctx->st0;
2551 reql = peer_getline_version(appctx, &maj_ver, &min_ver);
2552 if (reql <= 0) {
2553 if (!reql)
2554 goto out;
2555 goto switchstate;
2556 }
2557
2558 appctx->st0 = PEER_SESS_ST_GETHOST;
2559 /* fall through */
2560 case PEER_SESS_ST_GETHOST:
2561 prev_state = appctx->st0;
2562 reql = peer_getline_host(appctx);
2563 if (reql <= 0) {
2564 if (!reql)
2565 goto out;
2566 goto switchstate;
2567 }
2568
2569 appctx->st0 = PEER_SESS_ST_GETPEER;
2570 /* fall through */
2571 case PEER_SESS_ST_GETPEER: {
2572 prev_state = appctx->st0;
2573 reql = peer_getline_last(appctx, &curpeer);
2574 if (reql <= 0) {
2575 if (!reql)
2576 goto out;
2577 goto switchstate;
2578 }
2579
2580 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
2581 if (curpeer->appctx && curpeer->appctx != appctx) {
2582 if (curpeer->local) {
2583 /* Local connection, reply a retry */
2584 appctx->st0 = PEER_SESS_ST_EXIT;
2585 appctx->st1 = PEER_SESS_SC_TRYAGAIN;
2586 goto switchstate;
2587 }
2588
2589 /* we're killing a connection, we must apply a random delay before
2590 * retrying otherwise the other end will do the same and we can loop
2591 * for a while.
2592 */
2593 curpeer->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + ha_random() % 2000));
2594 peer_session_forceshutdown(curpeer);
2595 curpeer->heartbeat = TICK_ETERNITY;
2596 curpeer->coll++;
2597 }
2598 if (maj_ver != (unsigned int)-1 && min_ver != (unsigned int)-1) {
2599 if (min_ver == PEER_DWNGRD_MINOR_VER) {
2600 curpeer->flags |= PEER_F_DWNGRD;
2601 }
2602 else {
2603 curpeer->flags &= ~PEER_F_DWNGRD;
2604 }
2605 }
2606 curpeer->appctx = appctx;
2607 curpeer->flags |= PEER_F_ALIVE;
2608 appctx->ctx.peers.ptr = curpeer;
2609 appctx->st0 = PEER_SESS_ST_SENDSUCCESS;
2610 _HA_ATOMIC_ADD(&active_peers, 1);
2611 }
2612 /* fall through */
2613 case PEER_SESS_ST_SENDSUCCESS: {
2614 prev_state = appctx->st0;
2615 if (!curpeer) {
2616 curpeer = appctx->ctx.peers.ptr;
2617 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
2618 if (curpeer->appctx != appctx) {
2619 appctx->st0 = PEER_SESS_ST_END;
2620 goto switchstate;
2621 }
2622 }
2623
2624 repl = peer_send_status_successmsg(appctx);
2625 if (repl <= 0) {
2626 if (repl == -1)
2627 goto out;
2628 goto switchstate;
2629 }
2630
2631 init_accepted_peer(curpeer, curpeers);
2632
2633 /* switch to waiting message state */
2634 _HA_ATOMIC_ADD(&connected_peers, 1);
2635 appctx->st0 = PEER_SESS_ST_WAITMSG;
2636 goto switchstate;
2637 }
2638 case PEER_SESS_ST_CONNECT: {
2639 prev_state = appctx->st0;
2640 if (!curpeer) {
2641 curpeer = appctx->ctx.peers.ptr;
2642 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
2643 if (curpeer->appctx != appctx) {
2644 appctx->st0 = PEER_SESS_ST_END;
2645 goto switchstate;
2646 }
2647 }
2648
2649 repl = peer_send_hellomsg(appctx, curpeer);
2650 if (repl <= 0) {
2651 if (repl == -1)
2652 goto out;
2653 goto switchstate;
2654 }
2655
2656 /* switch to the waiting statuscode state */
2657 appctx->st0 = PEER_SESS_ST_GETSTATUS;
2658 }
2659 /* fall through */
2660 case PEER_SESS_ST_GETSTATUS: {
2661 prev_state = appctx->st0;
2662 if (!curpeer) {
2663 curpeer = appctx->ctx.peers.ptr;
2664 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
2665 if (curpeer->appctx != appctx) {
2666 appctx->st0 = PEER_SESS_ST_END;
2667 goto switchstate;
2668 }
2669 }
2670
2671 if (si_ic(si)->flags & CF_WRITE_PARTIAL)
2672 curpeer->statuscode = PEER_SESS_SC_CONNECTEDCODE;
2673
2674 reql = peer_getline(appctx);
2675 if (!reql)
2676 goto out;
2677
2678 if (reql < 0)
2679 goto switchstate;
2680
2681 /* Register status code */
2682 curpeer->statuscode = atoi(trash.area);
2683 curpeer->last_hdshk = now_ms;
2684
2685 /* Awake main task */
2686 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
2687
2688 /* If status code is success */
2689 if (curpeer->statuscode == PEER_SESS_SC_SUCCESSCODE) {
2690 init_connected_peer(curpeer, curpeers);
2691 }
2692 else {
2693 if (curpeer->statuscode == PEER_SESS_SC_ERRVERSION)
2694 curpeer->flags |= PEER_F_DWNGRD;
2695 /* Status code is not success, abort */
2696 appctx->st0 = PEER_SESS_ST_END;
2697 goto switchstate;
2698 }
2699 _HA_ATOMIC_ADD(&connected_peers, 1);
2700 appctx->st0 = PEER_SESS_ST_WAITMSG;
2701 }
2702 /* fall through */
2703 case PEER_SESS_ST_WAITMSG: {
2704 uint32_t msg_len = 0;
2705 char *msg_cur = trash.area;
2706 char *msg_end = trash.area;
2707 unsigned char msg_head[7]; // 2 + 5 for varint32
2708 int totl = 0;
2709
2710 prev_state = appctx->st0;
2711 if (!curpeer) {
2712 curpeer = appctx->ctx.peers.ptr;
2713 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
2714 if (curpeer->appctx != appctx) {
2715 appctx->st0 = PEER_SESS_ST_END;
2716 goto switchstate;
2717 }
2718 }
2719
2720 reql = peer_recv_msg(appctx, (char *)msg_head, sizeof msg_head, &msg_len, &totl);
2721 if (reql <= 0) {
2722 if (reql == -1)
2723 goto switchstate;
2724 goto send_msgs;
2725 }
2726
2727 msg_end += msg_len;
2728 if (!peer_treat_awaited_msg(appctx, curpeer, msg_head, &msg_cur, msg_end, msg_len, totl))
2729 goto switchstate;
2730
2731 curpeer->flags |= PEER_F_ALIVE;
2732
2733 /* skip consumed message */
2734 co_skip(si_oc(si), totl);
2735 /* loop on that state to peek next message */
2736 goto switchstate;
2737
2738 send_msgs:
2739 if (curpeer->flags & PEER_F_HEARTBEAT) {
2740 curpeer->flags &= ~PEER_F_HEARTBEAT;
2741 repl = peer_send_heartbeatmsg(appctx, curpeer, curpeers);
2742 if (repl <= 0) {
2743 if (repl == -1)
2744 goto out;
2745 goto switchstate;
2746 }
2747 curpeer->tx_hbt++;
2748 }
2749 /* we get here when a peer_recv_msg() returns 0 in reql */
2750 repl = peer_send_msgs(appctx, curpeer, curpeers);
2751 if (repl <= 0) {
2752 if (repl == -1)
2753 goto out;
2754 goto switchstate;
2755 }
2756
2757 /* noting more to do */
2758 goto out;
2759 }
2760 case PEER_SESS_ST_EXIT:
2761 if (prev_state == PEER_SESS_ST_WAITMSG)
2762 _HA_ATOMIC_SUB(&connected_peers, 1);
2763 prev_state = appctx->st0;
2764 if (peer_send_status_errormsg(appctx) == -1)
2765 goto out;
2766 appctx->st0 = PEER_SESS_ST_END;
2767 goto switchstate;
2768 case PEER_SESS_ST_ERRSIZE: {
2769 if (prev_state == PEER_SESS_ST_WAITMSG)
2770 _HA_ATOMIC_SUB(&connected_peers, 1);
2771 prev_state = appctx->st0;
2772 if (peer_send_error_size_limitmsg(appctx) == -1)
2773 goto out;
2774 appctx->st0 = PEER_SESS_ST_END;
2775 goto switchstate;
2776 }
2777 case PEER_SESS_ST_ERRPROTO: {
2778 TRACE_PROTO("protocol error", PEERS_EV_PROTOERR,
2779 NULL, curpeer, &prev_state);
2780 if (curpeer)
2781 curpeer->proto_err++;
2782 if (prev_state == PEER_SESS_ST_WAITMSG)
2783 _HA_ATOMIC_SUB(&connected_peers, 1);
2784 prev_state = appctx->st0;
2785 if (peer_send_error_protomsg(appctx) == -1) {
2786 TRACE_PROTO("could not send error message", PEERS_EV_PROTOERR);
2787 goto out;
2788 }
2789 appctx->st0 = PEER_SESS_ST_END;
2790 prev_state = appctx->st0;
2791 }
2792 /* fall through */
2793 case PEER_SESS_ST_END: {
2794 if (prev_state == PEER_SESS_ST_WAITMSG)
2795 _HA_ATOMIC_SUB(&connected_peers, 1);
2796 prev_state = appctx->st0;
2797 if (curpeer) {
2798 HA_SPIN_UNLOCK(PEER_LOCK, &curpeer->lock);
2799 curpeer = NULL;
2800 }
2801 si_shutw(si);
2802 si_shutr(si);
2803 si_ic(si)->flags |= CF_READ_NULL;
2804 goto out;
2805 }
2806 }
2807 }
2808 out:
2809 si_oc(si)->flags |= CF_READ_DONTWAIT;
2810
2811 if (curpeer)
2812 HA_SPIN_UNLOCK(PEER_LOCK, &curpeer->lock);
2813 return;
2814 }
2815
2816 static struct applet peer_applet = {
2817 .obj_type = OBJ_TYPE_APPLET,
2818 .name = "<PEER>", /* used for logging */
2819 .fct = peer_io_handler,
2820 .release = peer_session_release,
2821 };
2822
2823
2824 /*
2825 * Use this function to force a close of a peer session
2826 */
peer_session_forceshutdown(struct peer * peer)2827 static void peer_session_forceshutdown(struct peer *peer)
2828 {
2829 struct appctx *appctx = peer->appctx;
2830
2831 /* Note that the peer sessions which have just been created
2832 * (->st0 == PEER_SESS_ST_CONNECT) must not
2833 * be shutdown, if not, the TCP session will never be closed
2834 * and stay in CLOSE_WAIT state after having been closed by
2835 * the remote side.
2836 */
2837 if (!appctx || appctx->st0 == PEER_SESS_ST_CONNECT)
2838 return;
2839
2840 if (appctx->applet != &peer_applet)
2841 return;
2842
2843 __peer_session_deinit(peer);
2844
2845 appctx->st0 = PEER_SESS_ST_END;
2846 appctx_wakeup(appctx);
2847 }
2848
2849 /* Pre-configures a peers frontend to accept incoming connections */
peers_setup_frontend(struct proxy * fe)2850 void peers_setup_frontend(struct proxy *fe)
2851 {
2852 fe->last_change = now.tv_sec;
2853 fe->cap = PR_CAP_FE | PR_CAP_BE;
2854 fe->mode = PR_MODE_PEERS;
2855 fe->maxconn = 0;
2856 fe->conn_retries = CONN_RETRIES;
2857 fe->timeout.client = MS_TO_TICKS(5000);
2858 fe->accept = frontend_accept;
2859 fe->default_target = &peer_applet.obj_type;
2860 fe->options2 |= PR_O2_INDEPSTR | PR_O2_SMARTCON | PR_O2_SMARTACC;
2861 fe->bind_proc = 0; /* will be filled by users */
2862 }
2863
2864 /*
2865 * Create a new peer session in assigned state (connect will start automatically)
2866 */
peer_session_create(struct peers * peers,struct peer * peer)2867 static struct appctx *peer_session_create(struct peers *peers, struct peer *peer)
2868 {
2869 struct proxy *p = peers->peers_fe; /* attached frontend */
2870 struct appctx *appctx;
2871 struct session *sess;
2872 struct stream *s;
2873
2874 peer->new_conn++;
2875 peer->reconnect = tick_add(now_ms, MS_TO_TICKS(PEER_RECONNECT_TIMEOUT));
2876 peer->heartbeat = TICK_ETERNITY;
2877 peer->statuscode = PEER_SESS_SC_CONNECTCODE;
2878 peer->last_hdshk = now_ms;
2879 s = NULL;
2880
2881 appctx = appctx_new(&peer_applet, tid_bit);
2882 if (!appctx)
2883 goto out_close;
2884
2885 appctx->st0 = PEER_SESS_ST_CONNECT;
2886 appctx->ctx.peers.ptr = (void *)peer;
2887
2888 sess = session_new(p, NULL, &appctx->obj_type);
2889 if (!sess) {
2890 ha_alert("out of memory in peer_session_create().\n");
2891 goto out_free_appctx;
2892 }
2893
2894 if ((s = stream_new(sess, &appctx->obj_type)) == NULL) {
2895 ha_alert("Failed to initialize stream in peer_session_create().\n");
2896 goto out_free_sess;
2897 }
2898
2899 /* applet is waiting for data */
2900 si_cant_get(&s->si[0]);
2901 appctx_wakeup(appctx);
2902
2903 /* initiate an outgoing connection */
2904 s->target = peer_session_target(peer, s);
2905 if (!sockaddr_alloc(&s->target_addr, &peer->addr, sizeof(peer->addr)))
2906 goto out_free_strm;
2907 s->flags = SF_ASSIGNED|SF_ADDR_SET;
2908 s->si[1].flags |= SI_FL_NOLINGER;
2909
2910 s->do_log = NULL;
2911 s->uniq_id = 0;
2912
2913 s->res.flags |= CF_READ_DONTWAIT;
2914
2915 peer->appctx = appctx;
2916 task_wakeup(s->task, TASK_WOKEN_INIT);
2917 _HA_ATOMIC_ADD(&active_peers, 1);
2918 return appctx;
2919
2920 /* Error unrolling */
2921 out_free_strm:
2922 LIST_DEL(&s->list);
2923 pool_free(pool_head_stream, s);
2924 out_free_sess:
2925 session_free(sess);
2926 out_free_appctx:
2927 appctx_free(appctx);
2928 out_close:
2929 return NULL;
2930 }
2931
2932 /*
2933 * Task processing function to manage re-connect, peer session
2934 * tasks wakeup on local update and heartbeat.
2935 */
process_peer_sync(struct task * task,void * context,unsigned short state)2936 static struct task *process_peer_sync(struct task * task, void *context, unsigned short state)
2937 {
2938 struct peers *peers = context;
2939 struct peer *ps;
2940 struct shared_table *st;
2941
2942 task->expire = TICK_ETERNITY;
2943
2944 if (!peers->peers_fe) {
2945 /* this one was never started, kill it */
2946 signal_unregister_handler(peers->sighandler);
2947 task_destroy(peers->sync_task);
2948 peers->sync_task = NULL;
2949 return NULL;
2950 }
2951
2952 /* Acquire lock for all peers of the section */
2953 for (ps = peers->remote; ps; ps = ps->next)
2954 HA_SPIN_LOCK(PEER_LOCK, &ps->lock);
2955
2956 if (!stopping) {
2957 /* Normal case (not soft stop)*/
2958
2959 /* resync timeout set to TICK_ETERNITY means we just start
2960 * a new process and timer was not initialized.
2961 * We must arm this timer to switch to a request to a remote
2962 * node if incoming connection from old local process never
2963 * comes.
2964 */
2965 if (peers->resync_timeout == TICK_ETERNITY)
2966 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(PEER_RESYNC_TIMEOUT));
2967
2968 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL) &&
2969 (!nb_oldpids || tick_is_expired(peers->resync_timeout, now_ms)) &&
2970 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
2971 /* Resync from local peer needed
2972 no peer was assigned for the lesson
2973 and no old local peer found
2974 or resync timeout expire */
2975
2976 /* flag no more resync from local, to try resync from remotes */
2977 peers->flags |= PEERS_F_RESYNC_LOCAL;
2978 peers->flags |= PEERS_F_RESYNC_LOCALTIMEOUT;
2979
2980 /* reschedule a resync */
2981 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(PEER_RESYNC_TIMEOUT));
2982 }
2983
2984 /* For each session */
2985 for (ps = peers->remote; ps; ps = ps->next) {
2986 /* For each remote peers */
2987 if (!ps->local) {
2988 if (!ps->appctx) {
2989 /* no active peer connection */
2990 if (ps->statuscode == 0 ||
2991 ((ps->statuscode == PEER_SESS_SC_CONNECTCODE ||
2992 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
2993 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE) &&
2994 tick_is_expired(ps->reconnect, now_ms))) {
2995 /* connection never tried
2996 * or previous peer connection established with success
2997 * or previous peer connection failed while connecting
2998 * and reconnection timer is expired */
2999
3000 /* retry a connect */
3001 ps->appctx = peer_session_create(peers, ps);
3002 }
3003 else if (!tick_is_expired(ps->reconnect, now_ms)) {
3004 /* If previous session failed during connection
3005 * but reconnection timer is not expired */
3006
3007 /* reschedule task for reconnect */
3008 task->expire = tick_first(task->expire, ps->reconnect);
3009 }
3010 /* else do nothing */
3011 } /* !ps->appctx */
3012 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE) {
3013 /* current peer connection is active and established */
3014 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
3015 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
3016 !(ps->flags & PEER_F_LEARN_NOTUP2DATE)) {
3017 /* Resync from a remote is needed
3018 * and no peer was assigned for lesson
3019 * and current peer may be up2date */
3020
3021 /* assign peer for the lesson */
3022 ps->flags |= PEER_F_LEARN_ASSIGN;
3023 peers->flags |= PEERS_F_RESYNC_ASSIGN;
3024 peers->flags |= PEERS_F_RESYNC_REMOTEASSIGN;
3025
3026 /* wake up peer handler to handle a request of resync */
3027 appctx_wakeup(ps->appctx);
3028 }
3029 else {
3030 int update_to_push = 0;
3031
3032 /* Awake session if there is data to push */
3033 for (st = ps->tables; st ; st = st->next) {
3034 if (st->last_pushed != st->table->localupdate) {
3035 /* wake up the peer handler to push local updates */
3036 update_to_push = 1;
3037 /* There is no need to send a heartbeat message
3038 * when some updates must be pushed. The remote
3039 * peer will consider <ps> peer as alive when it will
3040 * receive these updates.
3041 */
3042 ps->flags &= ~PEER_F_HEARTBEAT;
3043 /* Re-schedule another one later. */
3044 ps->heartbeat = tick_add(now_ms, MS_TO_TICKS(PEER_HEARTBEAT_TIMEOUT));
3045 /* We are going to send updates, let's ensure we will
3046 * come back to send heartbeat messages or to reconnect.
3047 */
3048 task->expire = tick_first(ps->reconnect, ps->heartbeat);
3049 appctx_wakeup(ps->appctx);
3050 break;
3051 }
3052 }
3053 /* When there are updates to send we do not reconnect
3054 * and do not send heartbeat message either.
3055 */
3056 if (!update_to_push) {
3057 if (tick_is_expired(ps->reconnect, now_ms)) {
3058 if (ps->flags & PEER_F_ALIVE) {
3059 /* This peer was alive during a 'reconnect' period.
3060 * Flag it as not alive again for the next period.
3061 */
3062 ps->flags &= ~PEER_F_ALIVE;
3063 ps->reconnect = tick_add(now_ms, MS_TO_TICKS(PEER_RECONNECT_TIMEOUT));
3064 }
3065 else {
3066 ps->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + ha_random() % 2000));
3067 ps->heartbeat = TICK_ETERNITY;
3068 peer_session_forceshutdown(ps);
3069 ps->no_hbt++;
3070 }
3071 }
3072 else if (tick_is_expired(ps->heartbeat, now_ms)) {
3073 ps->heartbeat = tick_add(now_ms, MS_TO_TICKS(PEER_HEARTBEAT_TIMEOUT));
3074 ps->flags |= PEER_F_HEARTBEAT;
3075 appctx_wakeup(ps->appctx);
3076 }
3077 task->expire = tick_first(ps->reconnect, ps->heartbeat);
3078 }
3079 }
3080 /* else do nothing */
3081 } /* SUCCESSCODE */
3082 } /* !ps->peer->local */
3083 } /* for */
3084
3085 /* Resync from remotes expired: consider resync is finished */
3086 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
3087 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
3088 tick_is_expired(peers->resync_timeout, now_ms)) {
3089 /* Resync from remote peer needed
3090 * no peer was assigned for the lesson
3091 * and resync timeout expire */
3092
3093 /* flag no more resync from remote, consider resync is finished */
3094 peers->flags |= PEERS_F_RESYNC_REMOTE;
3095 peers->flags |= PEERS_F_RESYNC_REMOTETIMEOUT;
3096 }
3097
3098 if ((peers->flags & PEERS_RESYNC_STATEMASK) != PEERS_RESYNC_FINISHED) {
3099 /* Resync not finished*/
3100 /* reschedule task to resync timeout if not expired, to ended resync if needed */
3101 if (!tick_is_expired(peers->resync_timeout, now_ms))
3102 task->expire = tick_first(task->expire, peers->resync_timeout);
3103 }
3104 } /* !stopping */
3105 else {
3106 /* soft stop case */
3107 if (state & TASK_WOKEN_SIGNAL) {
3108 /* We've just received the signal */
3109 if (!(peers->flags & PEERS_F_DONOTSTOP)) {
3110 /* add DO NOT STOP flag if not present */
3111 _HA_ATOMIC_ADD(&jobs, 1);
3112 peers->flags |= PEERS_F_DONOTSTOP;
3113
3114 /* disconnect all connected peers to process a local sync
3115 * this must be done only the first time we are switching
3116 * in stopping state
3117 */
3118 for (ps = peers->remote; ps; ps = ps->next) {
3119 /* we're killing a connection, we must apply a random delay before
3120 * retrying otherwise the other end will do the same and we can loop
3121 * for a while.
3122 */
3123 ps->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + ha_random() % 2000));
3124 if (ps->appctx) {
3125 peer_session_forceshutdown(ps);
3126 }
3127 }
3128 }
3129 }
3130
3131 ps = peers->local;
3132 if (ps->flags & PEER_F_TEACH_COMPLETE) {
3133 if (peers->flags & PEERS_F_DONOTSTOP) {
3134 /* resync of new process was complete, current process can die now */
3135 _HA_ATOMIC_SUB(&jobs, 1);
3136 peers->flags &= ~PEERS_F_DONOTSTOP;
3137 for (st = ps->tables; st ; st = st->next)
3138 _HA_ATOMIC_SUB(&st->table->refcnt, 1);
3139 }
3140 }
3141 else if (!ps->appctx) {
3142 /* If there's no active peer connection */
3143 if (ps->statuscode == 0 ||
3144 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
3145 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE ||
3146 ps->statuscode == PEER_SESS_SC_TRYAGAIN) {
3147 /* connection never tried
3148 * or previous peer connection was successfully established
3149 * or previous tcp connect succeeded but init state incomplete
3150 * or during previous connect, peer replies a try again statuscode */
3151
3152 /* connect to the local peer if we must push a local sync */
3153 if (peers->flags & PEERS_F_DONOTSTOP) {
3154 peer_session_create(peers, ps);
3155 }
3156 }
3157 else {
3158 /* Other error cases */
3159 if (peers->flags & PEERS_F_DONOTSTOP) {
3160 /* unable to resync new process, current process can die now */
3161 _HA_ATOMIC_SUB(&jobs, 1);
3162 peers->flags &= ~PEERS_F_DONOTSTOP;
3163 for (st = ps->tables; st ; st = st->next)
3164 _HA_ATOMIC_SUB(&st->table->refcnt, 1);
3165 }
3166 }
3167 }
3168 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE ) {
3169 /* current peer connection is active and established
3170 * wake up all peer handlers to push remaining local updates */
3171 for (st = ps->tables; st ; st = st->next) {
3172 if (st->last_pushed != st->table->localupdate) {
3173 appctx_wakeup(ps->appctx);
3174 break;
3175 }
3176 }
3177 }
3178 } /* stopping */
3179
3180 /* Release lock for all peers of the section */
3181 for (ps = peers->remote; ps; ps = ps->next)
3182 HA_SPIN_UNLOCK(PEER_LOCK, &ps->lock);
3183
3184 /* Wakeup for re-connect */
3185 return task;
3186 }
3187
3188
3189 /*
3190 * returns 0 in case of error.
3191 */
peers_init_sync(struct peers * peers)3192 int peers_init_sync(struct peers *peers)
3193 {
3194 struct peer * curpeer;
3195
3196 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
3197 peers->peers_fe->maxconn += 3;
3198 }
3199
3200 peers->sync_task = task_new(MAX_THREADS_MASK);
3201 if (!peers->sync_task)
3202 return 0;
3203
3204 peers->sync_task->process = process_peer_sync;
3205 peers->sync_task->context = (void *)peers;
3206 peers->sighandler = signal_register_task(0, peers->sync_task, 0);
3207 task_wakeup(peers->sync_task, TASK_WOKEN_INIT);
3208 return 1;
3209 }
3210
3211 /*
3212 * Allocate a cache a dictionary entries used upon transmission.
3213 */
new_dcache_tx(size_t max_entries)3214 static struct dcache_tx *new_dcache_tx(size_t max_entries)
3215 {
3216 struct dcache_tx *d;
3217 struct ebpt_node *entries;
3218
3219 d = malloc(sizeof *d);
3220 entries = calloc(max_entries, sizeof *entries);
3221 if (!d || !entries)
3222 goto err;
3223
3224 d->lru_key = 0;
3225 d->prev_lookup = NULL;
3226 d->cached_entries = EB_ROOT_UNIQUE;
3227 d->entries = entries;
3228
3229 return d;
3230
3231 err:
3232 free(d);
3233 free(entries);
3234 return NULL;
3235 }
3236
3237 /*
3238 * Allocate a cache of dictionary entries with <name> as name and <max_entries>
3239 * as maximum of entries.
3240 * Return the dictionary cache if succeeded, NULL if not.
3241 * Must be deallocated calling free_dcache().
3242 */
new_dcache(size_t max_entries)3243 static struct dcache *new_dcache(size_t max_entries)
3244 {
3245 struct dcache_tx *dc_tx;
3246 struct dcache *dc;
3247 struct dcache_rx *dc_rx;
3248
3249 dc = calloc(1, sizeof *dc);
3250 dc_tx = new_dcache_tx(max_entries);
3251 dc_rx = calloc(max_entries, sizeof *dc_rx);
3252 if (!dc || !dc_tx || !dc_rx)
3253 goto err;
3254
3255 dc->tx = dc_tx;
3256 dc->rx = dc_rx;
3257 dc->max_entries = max_entries;
3258
3259 return dc;
3260
3261 err:
3262 free(dc);
3263 free(dc_tx);
3264 free(dc_rx);
3265 return NULL;
3266 }
3267
3268 /*
3269 * Look for the dictionary entry with the value of <i> in <d> cache of dictionary
3270 * entries used upon transmission.
3271 * Return the entry if found, NULL if not.
3272 */
dcache_tx_lookup_value(struct dcache_tx * d,struct dcache_tx_entry * i)3273 static struct ebpt_node *dcache_tx_lookup_value(struct dcache_tx *d,
3274 struct dcache_tx_entry *i)
3275 {
3276 return ebpt_lookup(&d->cached_entries, i->entry.key);
3277 }
3278
3279 /*
3280 * Flush <dc> cache.
3281 * Always succeeds.
3282 */
flush_dcache(struct peer * peer)3283 static inline void flush_dcache(struct peer *peer)
3284 {
3285 int i;
3286 struct dcache *dc = peer->dcache;
3287
3288 for (i = 0; i < dc->max_entries; i++) {
3289 ebpt_delete(&dc->tx->entries[i]);
3290 dc->tx->entries[i].key = NULL;
3291 }
3292 dc->tx->prev_lookup = NULL;
3293 dc->tx->lru_key = 0;
3294
3295 memset(dc->rx, 0, dc->max_entries * sizeof *dc->rx);
3296 }
3297
3298 /*
3299 * Insert a dictionary entry in <dc> cache part used upon transmission (->tx)
3300 * with information provided by <i> dictionary cache entry (especially the value
3301 * to be inserted if not already). Return <i> if already present in the cache
3302 * or something different of <i> if not.
3303 */
dcache_tx_insert(struct dcache * dc,struct dcache_tx_entry * i)3304 static struct ebpt_node *dcache_tx_insert(struct dcache *dc, struct dcache_tx_entry *i)
3305 {
3306 struct dcache_tx *dc_tx;
3307 struct ebpt_node *o;
3308
3309 dc_tx = dc->tx;
3310
3311 if (dc_tx->prev_lookup && dc_tx->prev_lookup->key == i->entry.key) {
3312 o = dc_tx->prev_lookup;
3313 } else {
3314 o = dcache_tx_lookup_value(dc_tx, i);
3315 if (o) {
3316 /* Save it */
3317 dc_tx->prev_lookup = o;
3318 }
3319 }
3320
3321 if (o) {
3322 /* Copy the ID. */
3323 i->id = o - dc->tx->entries;
3324 return &i->entry;
3325 }
3326
3327 /* The new entry to put in cache */
3328 dc_tx->prev_lookup = o = &dc_tx->entries[dc_tx->lru_key];
3329
3330 ebpt_delete(o);
3331 o->key = i->entry.key;
3332 ebpt_insert(&dc_tx->cached_entries, o);
3333 i->id = dc_tx->lru_key;
3334
3335 /* Update the index for the next entry to put in cache */
3336 dc_tx->lru_key = (dc_tx->lru_key + 1) & (dc->max_entries - 1);
3337
3338 return o;
3339 }
3340
3341 /*
3342 * Allocate a dictionary cache for each peer of <peers> section.
3343 * Return 1 if succeeded, 0 if not.
3344 */
peers_alloc_dcache(struct peers * peers)3345 int peers_alloc_dcache(struct peers *peers)
3346 {
3347 struct peer *p;
3348
3349 for (p = peers->remote; p; p = p->next) {
3350 p->dcache = new_dcache(PEER_STKT_CACHE_MAX_ENTRIES);
3351 if (!p->dcache)
3352 return 0;
3353 }
3354
3355 return 1;
3356 }
3357
3358 /*
3359 * Function used to register a table for sync on a group of peers
3360 * Returns 0 in case of success.
3361 */
peers_register_table(struct peers * peers,struct stktable * table)3362 int peers_register_table(struct peers *peers, struct stktable *table)
3363 {
3364 struct shared_table *st;
3365 struct peer * curpeer;
3366 int id = 0;
3367 int retval = 0;
3368
3369 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
3370 st = calloc(1,sizeof(*st));
3371 if (!st) {
3372 retval = 1;
3373 break;
3374 }
3375 st->table = table;
3376 st->next = curpeer->tables;
3377 if (curpeer->tables)
3378 id = curpeer->tables->local_id;
3379 st->local_id = id + 1;
3380
3381 /* If peer is local we inc table
3382 * refcnt to protect against flush
3383 * until this process pushed all
3384 * table content to the new one
3385 */
3386 if (curpeer->local)
3387 _HA_ATOMIC_ADD(&st->table->refcnt, 1);
3388 curpeer->tables = st;
3389 }
3390
3391 table->sync_task = peers->sync_task;
3392
3393 return retval;
3394 }
3395
3396 /*
3397 * Parse the "show peers" command arguments.
3398 * Returns 0 if succeeded, 1 if not with the ->msg of the appctx set as
3399 * error message.
3400 */
cli_parse_show_peers(char ** args,char * payload,struct appctx * appctx,void * private)3401 static int cli_parse_show_peers(char **args, char *payload, struct appctx *appctx, void *private)
3402 {
3403 appctx->ctx.cfgpeers.target = NULL;
3404
3405 if (*args[2]) {
3406 struct peers *p;
3407
3408 for (p = cfg_peers; p; p = p->next) {
3409 if (!strcmp(p->id, args[2])) {
3410 appctx->ctx.cfgpeers.target = p;
3411 break;
3412 }
3413 }
3414
3415 if (!p)
3416 return cli_err(appctx, "No such peers\n");
3417 }
3418
3419 return 0;
3420 }
3421
3422 /*
3423 * This function dumps the peer state information of <peers> "peers" section.
3424 * Returns 0 if the output buffer is full and needs to be called again, non-zero if not.
3425 * Dedicated to be called by cli_io_handler_show_peers() cli I/O handler.
3426 */
peers_dump_head(struct buffer * msg,struct stream_interface * si,struct peers * peers)3427 static int peers_dump_head(struct buffer *msg, struct stream_interface *si, struct peers *peers)
3428 {
3429 struct tm tm;
3430
3431 get_localtime(peers->last_change, &tm);
3432 chunk_appendf(msg, "%p: [%02d/%s/%04d:%02d:%02d:%02d] id=%s disabled=%d flags=0x%x resync_timeout=%s task_calls=%u\n",
3433 peers,
3434 tm.tm_mday, monthname[tm.tm_mon], tm.tm_year+1900,
3435 tm.tm_hour, tm.tm_min, tm.tm_sec,
3436 peers->id, peers->disabled, peers->flags,
3437 peers->resync_timeout ?
3438 tick_is_expired(peers->resync_timeout, now_ms) ? "<PAST>" :
3439 human_time(TICKS_TO_MS(peers->resync_timeout - now_ms),
3440 TICKS_TO_MS(1000)) : "<NEVER>",
3441 peers->sync_task ? peers->sync_task->calls : 0);
3442
3443 if (ci_putchk(si_ic(si), msg) == -1) {
3444 si_rx_room_blk(si);
3445 return 0;
3446 }
3447
3448 return 1;
3449 }
3450
3451 /*
3452 * This function dumps <peer> state information.
3453 * Returns 0 if the output buffer is full and needs to be called again, non-zero
3454 * if not. Dedicated to be called by cli_io_handler_show_peers() cli I/O handler.
3455 */
peers_dump_peer(struct buffer * msg,struct stream_interface * si,struct peer * peer)3456 static int peers_dump_peer(struct buffer *msg, struct stream_interface *si, struct peer *peer)
3457 {
3458 struct connection *conn;
3459 char pn[INET6_ADDRSTRLEN];
3460 struct stream_interface *peer_si;
3461 struct stream *peer_s;
3462 struct appctx *appctx;
3463 struct shared_table *st;
3464
3465 addr_to_str(&peer->addr, pn, sizeof pn);
3466 chunk_appendf(msg, " %p: id=%s(%s,%s) addr=%s:%d last_status=%s",
3467 peer, peer->id,
3468 peer->local ? "local" : "remote",
3469 peer->appctx ? "active" : "inactive",
3470 pn, get_host_port(&peer->addr),
3471 statuscode_str(peer->statuscode));
3472
3473 chunk_appendf(msg, " last_hdshk=%s\n",
3474 peer->last_hdshk ? human_time(TICKS_TO_MS(now_ms - peer->last_hdshk),
3475 TICKS_TO_MS(1000)) : "<NEVER>");
3476
3477 chunk_appendf(msg, " reconnect=%s",
3478 peer->reconnect ?
3479 tick_is_expired(peer->reconnect, now_ms) ? "<PAST>" :
3480 human_time(TICKS_TO_MS(peer->reconnect - now_ms),
3481 TICKS_TO_MS(1000)) : "<NEVER>");
3482
3483 chunk_appendf(msg, " heartbeat=%s",
3484 peer->heartbeat ?
3485 tick_is_expired(peer->heartbeat, now_ms) ? "<PAST>" :
3486 human_time(TICKS_TO_MS(peer->heartbeat - now_ms),
3487 TICKS_TO_MS(1000)) : "<NEVER>");
3488
3489 chunk_appendf(msg, " confirm=%u tx_hbt=%u rx_hbt=%u no_hbt=%u new_conn=%u proto_err=%u coll=%u\n",
3490 peer->confirm, peer->tx_hbt, peer->rx_hbt,
3491 peer->no_hbt, peer->new_conn, peer->proto_err, peer->coll);
3492
3493 chunk_appendf(&trash, " flags=0x%x", peer->flags);
3494
3495 appctx = peer->appctx;
3496 if (!appctx)
3497 goto table_info;
3498
3499 chunk_appendf(&trash, " appctx:%p st0=%d st1=%d task_calls=%u", appctx, appctx->st0, appctx->st1,
3500 appctx->t ? appctx->t->calls : 0);
3501
3502 peer_si = peer->appctx->owner;
3503 if (!peer_si)
3504 goto table_info;
3505
3506 peer_s = si_strm(peer_si);
3507 if (!peer_s)
3508 goto table_info;
3509
3510 chunk_appendf(&trash, " state=%s", si_state_str(si_opposite(peer_si)->state));
3511
3512 conn = objt_conn(strm_orig(peer_s));
3513 if (conn)
3514 chunk_appendf(&trash, "\n xprt=%s", conn_get_xprt_name(conn));
3515
3516 switch (conn && conn_get_src(conn) ? addr_to_str(conn->src, pn, sizeof(pn)) : AF_UNSPEC) {
3517 case AF_INET:
3518 case AF_INET6:
3519 chunk_appendf(&trash, " src=%s:%d", pn, get_host_port(conn->src));
3520 break;
3521 case AF_UNIX:
3522 chunk_appendf(&trash, " src=unix:%d", strm_li(peer_s)->luid);
3523 break;
3524 }
3525
3526 switch (conn && conn_get_dst(conn) ? addr_to_str(conn->dst, pn, sizeof(pn)) : AF_UNSPEC) {
3527 case AF_INET:
3528 case AF_INET6:
3529 chunk_appendf(&trash, " addr=%s:%d", pn, get_host_port(conn->dst));
3530 break;
3531 case AF_UNIX:
3532 chunk_appendf(&trash, " addr=unix:%d", strm_li(peer_s)->luid);
3533 break;
3534 }
3535
3536 table_info:
3537 if (peer->remote_table)
3538 chunk_appendf(&trash, "\n remote_table:%p id=%s local_id=%d remote_id=%d",
3539 peer->remote_table,
3540 peer->remote_table->table->id,
3541 peer->remote_table->local_id,
3542 peer->remote_table->remote_id);
3543
3544 if (peer->last_local_table)
3545 chunk_appendf(&trash, "\n last_local_table:%p id=%s local_id=%d remote_id=%d",
3546 peer->last_local_table,
3547 peer->last_local_table->table->id,
3548 peer->last_local_table->local_id,
3549 peer->last_local_table->remote_id);
3550
3551 if (peer->tables) {
3552 chunk_appendf(&trash, "\n shared tables:");
3553 for (st = peer->tables; st; st = st->next) {
3554 int i, count;
3555 struct stktable *t;
3556 struct dcache *dcache;
3557
3558 t = st->table;
3559 dcache = peer->dcache;
3560
3561 chunk_appendf(&trash, "\n %p local_id=%d remote_id=%d "
3562 "flags=0x%x remote_data=0x%llx",
3563 st, st->local_id, st->remote_id,
3564 st->flags, (unsigned long long)st->remote_data);
3565 chunk_appendf(&trash, "\n last_acked=%u last_pushed=%u last_get=%u"
3566 " teaching_origin=%u update=%u",
3567 st->last_acked, st->last_pushed, st->last_get,
3568 st->teaching_origin, st->update);
3569 chunk_appendf(&trash, "\n table:%p id=%s update=%u localupdate=%u"
3570 " commitupdate=%u refcnt=%u",
3571 t, t->id, t->update, t->localupdate, t->commitupdate, t->refcnt);
3572 chunk_appendf(&trash, "\n TX dictionary cache:");
3573 count = 0;
3574 for (i = 0; i < dcache->max_entries; i++) {
3575 struct ebpt_node *node;
3576 struct dict_entry *de;
3577
3578 node = &dcache->tx->entries[i];
3579 if (!node->key)
3580 break;
3581
3582 if (!count++)
3583 chunk_appendf(&trash, "\n ");
3584 de = node->key;
3585 chunk_appendf(&trash, " %3u -> %s", i, (char *)de->value.key);
3586 count &= 0x3;
3587 }
3588 chunk_appendf(&trash, "\n RX dictionary cache:");
3589 count = 0;
3590 for (i = 0; i < dcache->max_entries; i++) {
3591 if (!count++)
3592 chunk_appendf(&trash, "\n ");
3593 chunk_appendf(&trash, " %3u -> %s", i,
3594 dcache->rx[i].de ?
3595 (char *)dcache->rx[i].de->value.key : "-");
3596 count &= 0x3;
3597 }
3598 }
3599 }
3600
3601 end:
3602 chunk_appendf(&trash, "\n");
3603 if (ci_putchk(si_ic(si), msg) == -1) {
3604 si_rx_room_blk(si);
3605 return 0;
3606 }
3607
3608 return 1;
3609 }
3610
3611 /*
3612 * This function dumps all the peers of "peers" section.
3613 * Returns 0 if the output buffer is full and needs to be called
3614 * again, non-zero if not. It proceeds in an isolated thread, so
3615 * there is no thread safety issue here.
3616 */
cli_io_handler_show_peers(struct appctx * appctx)3617 static int cli_io_handler_show_peers(struct appctx *appctx)
3618 {
3619 int show_all;
3620 int ret = 0, first_peers = 1;
3621 struct stream_interface *si = appctx->owner;
3622
3623 thread_isolate();
3624
3625 show_all = !appctx->ctx.cfgpeers.target;
3626
3627 chunk_reset(&trash);
3628
3629 while (appctx->st2 != STAT_ST_FIN) {
3630 switch (appctx->st2) {
3631 case STAT_ST_INIT:
3632 if (show_all)
3633 appctx->ctx.cfgpeers.peers = cfg_peers;
3634 else
3635 appctx->ctx.cfgpeers.peers = appctx->ctx.cfgpeers.target;
3636
3637 appctx->st2 = STAT_ST_LIST;
3638 /* fall through */
3639
3640 case STAT_ST_LIST:
3641 if (!appctx->ctx.cfgpeers.peers) {
3642 /* No more peers list. */
3643 appctx->st2 = STAT_ST_END;
3644 }
3645 else {
3646 if (!first_peers)
3647 chunk_appendf(&trash, "\n");
3648 else
3649 first_peers = 0;
3650 if (!peers_dump_head(&trash, si, appctx->ctx.cfgpeers.peers))
3651 goto out;
3652
3653 appctx->ctx.cfgpeers.peer = appctx->ctx.cfgpeers.peers->remote;
3654 appctx->ctx.cfgpeers.peers = appctx->ctx.cfgpeers.peers->next;
3655 appctx->st2 = STAT_ST_INFO;
3656 }
3657 break;
3658
3659 case STAT_ST_INFO:
3660 if (!appctx->ctx.cfgpeers.peer) {
3661 /* End of peer list */
3662 if (show_all)
3663 appctx->st2 = STAT_ST_LIST;
3664 else
3665 appctx->st2 = STAT_ST_END;
3666 }
3667 else {
3668 if (!peers_dump_peer(&trash, si, appctx->ctx.cfgpeers.peer))
3669 goto out;
3670
3671 appctx->ctx.cfgpeers.peer = appctx->ctx.cfgpeers.peer->next;
3672 }
3673 break;
3674
3675 case STAT_ST_END:
3676 appctx->st2 = STAT_ST_FIN;
3677 break;
3678 }
3679 }
3680 ret = 1;
3681 out:
3682 thread_release();
3683 return ret;
3684 }
3685
3686 /*
3687 * CLI keywords.
3688 */
3689 static struct cli_kw_list cli_kws = {{ }, {
3690 { { "show", "peers", NULL }, "show peers [peers section]: dump some information about all the peers or this peers section", cli_parse_show_peers, cli_io_handler_show_peers, },
3691 {},
3692 }};
3693
3694 /* Register cli keywords */
3695 INITCALL1(STG_REGISTER, cli_register_kw, &cli_kws);
3696
3697