blob: 30cf94277d9001caea5993c92b19f12dc216e2e1 [file] [log] [blame]
Emeric Brun2b920a12010-09-23 18:30:22 +02001/*
Emeric Brunb3971ab2015-05-12 18:49:09 +02002 * Peer synchro management.
Emeric Brun2b920a12010-09-23 18:30:22 +02003 *
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 <common/compat.h>
24#include <common/config.h>
25#include <common/time.h>
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +020026#include <common/standard.h>
Christopher Faulet8d8aa0d2017-05-30 15:36:50 +020027#include <common/hathreads.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020028
29#include <types/global.h>
Willy Tarreau3fdb3662012-11-12 00:42:33 +010030#include <types/listener.h>
31#include <types/obj_type.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020032#include <types/peers.h>
33
34#include <proto/acl.h>
Willy Tarreau8a8d83b2015-04-13 13:24:54 +020035#include <proto/applet.h>
Willy Tarreauc7e42382012-08-24 19:22:53 +020036#include <proto/channel.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020037#include <proto/fd.h>
Willy Tarreaud1d48d42015-03-13 16:15:46 +010038#include <proto/frontend.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020039#include <proto/log.h>
40#include <proto/hdr_idx.h>
Willy Tarreau53a47662017-08-28 10:53:00 +020041#include <proto/mux_pt.h>
Frédéric Lécaille1055e682018-04-26 14:35:21 +020042#include <proto/peers.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020043#include <proto/proxy.h>
Willy Tarreaufeb76402015-04-03 14:10:06 +020044#include <proto/session.h>
Willy Tarreau87b09662015-04-03 00:22:06 +020045#include <proto/stream.h>
Willy Tarreau22ec1ea2014-11-27 20:45:39 +010046#include <proto/signal.h>
47#include <proto/stick_table.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020048#include <proto/stream_interface.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020049#include <proto/task.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020050
51
52/*******************************/
53/* Current peer learning state */
54/*******************************/
55
56/******************************/
Emeric Brunb3971ab2015-05-12 18:49:09 +020057/* Current peers section resync state */
Emeric Brun2b920a12010-09-23 18:30:22 +020058/******************************/
Emeric Brunb3971ab2015-05-12 18:49:09 +020059#define PEERS_F_RESYNC_LOCAL 0x00000001 /* Learn from local finished or no more needed */
60#define PEERS_F_RESYNC_REMOTE 0x00000002 /* Learn from remote finished or no more needed */
61#define PEERS_F_RESYNC_ASSIGN 0x00000004 /* A peer was assigned to learn our lesson */
62#define PEERS_F_RESYNC_PROCESS 0x00000008 /* The assigned peer was requested for resync */
63#define PEERS_F_DONOTSTOP 0x00010000 /* Main table sync task block process during soft stop
Emeric Brun2b920a12010-09-23 18:30:22 +020064 to push data to new process */
65
Emeric Brunb3971ab2015-05-12 18:49:09 +020066#define PEERS_RESYNC_STATEMASK (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE)
67#define PEERS_RESYNC_FROMLOCAL 0x00000000
68#define PEERS_RESYNC_FROMREMOTE PEERS_F_RESYNC_LOCAL
69#define PEERS_RESYNC_FINISHED (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE)
70
71/***********************************/
72/* Current shared table sync state */
73/***********************************/
74#define SHTABLE_F_TEACH_STAGE1 0x00000001 /* Teach state 1 complete */
75#define SHTABLE_F_TEACH_STAGE2 0x00000002 /* Teach state 2 complete */
Emeric Brun2b920a12010-09-23 18:30:22 +020076
77/******************************/
78/* Remote peer teaching state */
79/******************************/
80#define PEER_F_TEACH_PROCESS 0x00000001 /* Teach a lesson to current peer */
Emeric Brun2b920a12010-09-23 18:30:22 +020081#define PEER_F_TEACH_FINISHED 0x00000008 /* Teach conclude, (wait for confirm) */
82#define PEER_F_TEACH_COMPLETE 0x00000010 /* All that we know already taught to current peer, used only for a local peer */
83#define PEER_F_LEARN_ASSIGN 0x00000100 /* Current peer was assigned for a lesson */
84#define PEER_F_LEARN_NOTUP2DATE 0x00000200 /* Learn from peer finished but peer is not up to date */
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +020085#define PEER_F_DWNGRD 0x80000000 /* When this flag is enabled, we must downgrade the supported version announced during peer sessions. */
Emeric Brun2b920a12010-09-23 18:30:22 +020086
Emeric Brunb3971ab2015-05-12 18:49:09 +020087#define PEER_TEACH_RESET ~(PEER_F_TEACH_PROCESS|PEER_F_TEACH_FINISHED) /* PEER_F_TEACH_COMPLETE should never be reset */
Emeric Brun2b920a12010-09-23 18:30:22 +020088#define PEER_LEARN_RESET ~(PEER_F_LEARN_ASSIGN|PEER_F_LEARN_NOTUP2DATE)
89
Emeric Brunb3971ab2015-05-12 18:49:09 +020090/*****************************/
91/* Sync message class */
92/*****************************/
93enum {
94 PEER_MSG_CLASS_CONTROL = 0,
95 PEER_MSG_CLASS_ERROR,
96 PEER_MSG_CLASS_STICKTABLE = 10,
97 PEER_MSG_CLASS_RESERVED = 255,
98};
99
100/*****************************/
101/* control message types */
102/*****************************/
103enum {
104 PEER_MSG_CTRL_RESYNCREQ = 0,
105 PEER_MSG_CTRL_RESYNCFINISHED,
106 PEER_MSG_CTRL_RESYNCPARTIAL,
107 PEER_MSG_CTRL_RESYNCCONFIRM,
108};
109
110/*****************************/
111/* error message types */
112/*****************************/
113enum {
114 PEER_MSG_ERR_PROTOCOL = 0,
115 PEER_MSG_ERR_SIZELIMIT,
116};
117
Frédéric Lécailled5fe14b2019-01-24 10:33:40 +0100118/*
119 * Parameters used by functions to build peer protocol messages. */
120struct peer_prep_params {
121 struct {
Frédéric Lécaille7d0ceee2019-01-24 14:24:05 +0100122 struct peer *peer;
123 } hello;
124 struct {
125 unsigned int st1;
126 } error_status;
127 struct {
Frédéric Lécailled5fe14b2019-01-24 10:33:40 +0100128 struct stksess *stksess;
129 struct shared_table *shared_table;
130 unsigned int updateid;
131 int use_identifier;
132 int use_timed;
133 } updt;
134 struct {
135 struct shared_table *shared_table;
136 } swtch;
137 struct {
138 struct shared_table *shared_table;
139 } ack;
Frédéric Lécaille7d0ceee2019-01-24 14:24:05 +0100140 struct {
141 unsigned char head[2];
142 } control;
143 struct {
144 unsigned char head[2];
145 } error;
Frédéric Lécailled5fe14b2019-01-24 10:33:40 +0100146};
Emeric Brunb3971ab2015-05-12 18:49:09 +0200147
148/*******************************/
149/* stick table sync mesg types */
150/* Note: ids >= 128 contains */
151/* id message cotains data */
152/*******************************/
Olivier Houchard33992262018-10-16 18:49:26 +0200153#define PEER_MSG_STKT_UPDATE 0x80
154#define PEER_MSG_STKT_INCUPDATE 0x81
155#define PEER_MSG_STKT_DEFINE 0x82
156#define PEER_MSG_STKT_SWITCH 0x83
157#define PEER_MSG_STKT_ACK 0x84
158#define PEER_MSG_STKT_UPDATE_TIMED 0x85
159#define PEER_MSG_STKT_INCUPDATE_TIMED 0x86
Emeric Brun2b920a12010-09-23 18:30:22 +0200160
161/**********************************/
162/* Peer Session IO handler states */
163/**********************************/
164
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100165enum {
166 PEER_SESS_ST_ACCEPT = 0, /* Initial state for session create by an accept, must be zero! */
167 PEER_SESS_ST_GETVERSION, /* Validate supported protocol version */
168 PEER_SESS_ST_GETHOST, /* Validate host ID correspond to local host id */
169 PEER_SESS_ST_GETPEER, /* Validate peer ID correspond to a known remote peer id */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100170 /* after this point, data were possibly exchanged */
171 PEER_SESS_ST_SENDSUCCESS, /* Send ret code 200 (success) and wait for message */
172 PEER_SESS_ST_CONNECT, /* Initial state for session create on a connect, push presentation into buffer */
173 PEER_SESS_ST_GETSTATUS, /* Wait for the welcome message */
174 PEER_SESS_ST_WAITMSG, /* Wait for data messages */
175 PEER_SESS_ST_EXIT, /* Exit with status code */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200176 PEER_SESS_ST_ERRPROTO, /* Send error proto message before exit */
177 PEER_SESS_ST_ERRSIZE, /* Send error size message before exit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100178 PEER_SESS_ST_END, /* Killed session */
179};
Emeric Brun2b920a12010-09-23 18:30:22 +0200180
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100181/***************************************************/
182/* Peer Session status code - part of the protocol */
183/***************************************************/
Emeric Brun2b920a12010-09-23 18:30:22 +0200184
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100185#define PEER_SESS_SC_CONNECTCODE 100 /* connect in progress */
186#define PEER_SESS_SC_CONNECTEDCODE 110 /* tcp connect success */
Emeric Brun2b920a12010-09-23 18:30:22 +0200187
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100188#define PEER_SESS_SC_SUCCESSCODE 200 /* accept or connect successful */
Emeric Brun2b920a12010-09-23 18:30:22 +0200189
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100190#define PEER_SESS_SC_TRYAGAIN 300 /* try again later */
Emeric Brun2b920a12010-09-23 18:30:22 +0200191
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100192#define PEER_SESS_SC_ERRPROTO 501 /* error protocol */
193#define PEER_SESS_SC_ERRVERSION 502 /* unknown protocol version */
194#define PEER_SESS_SC_ERRHOST 503 /* bad host name */
195#define PEER_SESS_SC_ERRPEER 504 /* unknown peer */
Emeric Brun2b920a12010-09-23 18:30:22 +0200196
197#define PEER_SESSION_PROTO_NAME "HAProxyS"
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200198#define PEER_MAJOR_VER 2
199#define PEER_MINOR_VER 1
200#define PEER_DWNGRD_MINOR_VER 0
Emeric Brun2b920a12010-09-23 18:30:22 +0200201
Willy Tarreau6254a922019-01-29 17:45:23 +0100202static size_t proto_len = sizeof(PEER_SESSION_PROTO_NAME) - 1;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200203struct peers *cfg_peers = NULL;
Willy Tarreau81bc3b02016-10-31 17:37:39 +0100204static void peer_session_forceshutdown(struct appctx *appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +0200205
Emeric Brun18928af2017-03-29 16:32:53 +0200206/* This function encode an uint64 to 'dynamic' length format.
207 The encoded value is written at address *str, and the
208 caller must assure that size after *str is large enought.
209 At return, the *str is set at the next Byte after then
210 encoded integer. The function returns then length of the
211 encoded integer in Bytes */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200212int intencode(uint64_t i, char **str) {
213 int idx = 0;
214 unsigned char *msg;
215
216 if (!*str)
217 return 0;
218
219 msg = (unsigned char *)*str;
220 if (i < 240) {
221 msg[0] = (unsigned char)i;
222 *str = (char *)&msg[idx+1];
223 return (idx+1);
224 }
225
226 msg[idx] =(unsigned char)i | 240;
227 i = (i - 240) >> 4;
228 while (i >= 128) {
229 msg[++idx] = (unsigned char)i | 128;
230 i = (i - 128) >> 7;
231 }
232 msg[++idx] = (unsigned char)i;
233 *str = (char *)&msg[idx+1];
234 return (idx+1);
235}
236
237
238/* This function returns the decoded integer or 0
239 if decode failed
240 *str point on the beginning of the integer to decode
241 at the end of decoding *str point on the end of the
242 encoded integer or to null if end is reached */
Emeric Brun18928af2017-03-29 16:32:53 +0200243uint64_t intdecode(char **str, char *end)
244{
Emeric Brunb3971ab2015-05-12 18:49:09 +0200245 unsigned char *msg;
Emeric Brun18928af2017-03-29 16:32:53 +0200246 uint64_t i;
247 int shift;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200248
249 if (!*str)
250 return 0;
251
252 msg = (unsigned char *)*str;
Emeric Brun18928af2017-03-29 16:32:53 +0200253 if (msg >= (unsigned char *)end)
254 goto fail;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200255
Emeric Brun18928af2017-03-29 16:32:53 +0200256 i = *(msg++);
257 if (i >= 240) {
258 shift = 4;
259 do {
260 if (msg >= (unsigned char *)end)
261 goto fail;
262 i += (uint64_t)*msg << shift;
263 shift += 7;
264 } while (*(msg++) >= 128);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200265 }
Frédéric Lécaillea8725ec2019-01-22 10:31:39 +0100266 *str = (char *)msg;
267 return i;
Emeric Brun18928af2017-03-29 16:32:53 +0200268
Frédéric Lécaillea8725ec2019-01-22 10:31:39 +0100269 fail:
270 *str = NULL;
271 return 0;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200272}
Emeric Brun2b920a12010-09-23 18:30:22 +0200273
Frédéric Lécaille7d0ceee2019-01-24 14:24:05 +0100274/*
275 * Build a "hello" peer protocol message.
276 * Return the number of written bytes written to build this messages if succeeded,
277 * 0 if not.
278 */
279static int peer_prepare_hellomsg(char *msg, size_t size, struct peer_prep_params *p)
280{
281 int min_ver, ret;
282 struct peer *peer;
283
284 peer = p->hello.peer;
285 min_ver = (peer->flags & PEER_F_DWNGRD) ? PEER_DWNGRD_MINOR_VER : PEER_MINOR_VER;
286 /* Prepare headers */
287 ret = snprintf(msg, size, PEER_SESSION_PROTO_NAME " %u.%u\n%s\n%s %d %d\n",
288 PEER_MAJOR_VER, min_ver, peer->id, localpeer, (int)getpid(), relative_pid);
289 if (ret >= size)
290 return 0;
291
292 return ret;
293}
294
295/*
296 * Build a "handshake succeeded" status message.
297 * Return the number of written bytes written to build this messages if succeeded,
298 * 0 if not.
299 */
300static int peer_prepare_status_successmsg(char *msg, size_t size, struct peer_prep_params *p)
301{
302 int ret;
303
304 ret = snprintf(msg, size, "%d\n", PEER_SESS_SC_SUCCESSCODE);
305 if (ret >= size)
306 return 0;
307
308 return ret;
309}
310
311/*
312 * Build an error status message.
313 * Return the number of written bytes written to build this messages if succeeded,
314 * 0 if not.
315 */
316static int peer_prepare_status_errormsg(char *msg, size_t size, struct peer_prep_params *p)
317{
318 int ret;
319 unsigned int st1;
320
321 st1 = p->error_status.st1;
322 ret = snprintf(msg, size, "%d\n", st1);
323 if (ret >= size)
324 return 0;
325
326 return ret;
327}
328
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200329/* Set the stick-table UPDATE message type byte at <msg_type> address,
330 * depending on <use_identifier> and <use_timed> boolean parameters.
331 * Always successful.
332 */
333static inline void peer_set_update_msg_type(char *msg_type, int use_identifier, int use_timed)
334{
335 if (use_timed) {
336 if (use_identifier)
337 *msg_type = PEER_MSG_STKT_UPDATE_TIMED;
338 else
339 *msg_type = PEER_MSG_STKT_INCUPDATE_TIMED;
340 }
341 else {
342 if (use_identifier)
343 *msg_type = PEER_MSG_STKT_UPDATE;
344 else
345 *msg_type = PEER_MSG_STKT_INCUPDATE;
346 }
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200347}
Emeric Brun2b920a12010-09-23 18:30:22 +0200348/*
Emeric Brunb3971ab2015-05-12 18:49:09 +0200349 * This prepare the data update message on the stick session <ts>, <st> is the considered
350 * stick table.
Joseph Herlant82b2f542018-11-15 12:19:14 -0800351 * <msg> is a buffer of <size> to receive data message content
Emeric Brunb3971ab2015-05-12 18:49:09 +0200352 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
353 * check size)
Emeric Brun2b920a12010-09-23 18:30:22 +0200354 */
Frédéric Lécailled5fe14b2019-01-24 10:33:40 +0100355static int peer_prepare_updatemsg(char *msg, size_t size, struct peer_prep_params *p)
Emeric Brun2b920a12010-09-23 18:30:22 +0200356{
357 uint32_t netinteger;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200358 unsigned short datalen;
359 char *cursor, *datamsg;
Emeric Brun94900952015-06-11 18:25:54 +0200360 unsigned int data_type;
361 void *data_ptr;
Frédéric Lécailled5fe14b2019-01-24 10:33:40 +0100362 struct stksess *ts;
363 struct shared_table *st;
364 unsigned int updateid;
365 int use_identifier;
366 int use_timed;
367
368 ts = p->updt.stksess;
369 st = p->updt.shared_table;
370 updateid = p->updt.updateid;
371 use_identifier = p->updt.use_identifier;
372 use_timed = p->updt.use_timed;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200373
374 cursor = datamsg = msg + 1 + 5;
375
Emeric Brun2b920a12010-09-23 18:30:22 +0200376 /* construct message */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200377
378 /* check if we need to send the update identifer */
Emeric Brun819fc6f2017-06-13 19:37:32 +0200379 if (!st->last_pushed || updateid < st->last_pushed || ((updateid - st->last_pushed) != 1)) {
Emeric Bruna6a09982015-09-22 15:34:19 +0200380 use_identifier = 1;
Emeric Brun2b920a12010-09-23 18:30:22 +0200381 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200382
383 /* encode update identifier if needed */
384 if (use_identifier) {
Emeric Brun819fc6f2017-06-13 19:37:32 +0200385 netinteger = htonl(updateid);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200386 memcpy(cursor, &netinteger, sizeof(netinteger));
387 cursor += sizeof(netinteger);
Emeric Brun2b920a12010-09-23 18:30:22 +0200388 }
389
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200390 if (use_timed) {
391 netinteger = htonl(tick_remain(now_ms, ts->expire));
392 memcpy(cursor, &netinteger, sizeof(netinteger));
393 cursor += sizeof(netinteger);
394 }
395
Emeric Brunb3971ab2015-05-12 18:49:09 +0200396 /* encode the key */
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +0200397 if (st->table->type == SMP_T_STR) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200398 int stlen = strlen((char *)ts->key.key);
399
Emeric Brunb3971ab2015-05-12 18:49:09 +0200400 intencode(stlen, &cursor);
401 memcpy(cursor, ts->key.key, stlen);
402 cursor += stlen;
Emeric Brun2b920a12010-09-23 18:30:22 +0200403 }
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +0200404 else if (st->table->type == SMP_T_SINT) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200405 netinteger = htonl(*((uint32_t *)ts->key.key));
Emeric Brunb3971ab2015-05-12 18:49:09 +0200406 memcpy(cursor, &netinteger, sizeof(netinteger));
407 cursor += sizeof(netinteger);
Emeric Brun2b920a12010-09-23 18:30:22 +0200408 }
409 else {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200410 memcpy(cursor, ts->key.key, st->table->key_size);
411 cursor += st->table->key_size;
Emeric Brun2b920a12010-09-23 18:30:22 +0200412 }
413
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100414 HA_RWLOCK_RDLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200415 /* encode values */
Emeric Brun94900952015-06-11 18:25:54 +0200416 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200417
Emeric Brun94900952015-06-11 18:25:54 +0200418 data_ptr = stktable_data_ptr(st->table, ts, data_type);
419 if (data_ptr) {
420 switch (stktable_data_types[data_type].std_type) {
421 case STD_T_SINT: {
422 int data;
423
424 data = stktable_data_cast(data_ptr, std_t_sint);
425 intencode(data, &cursor);
426 break;
427 }
428 case STD_T_UINT: {
429 unsigned int data;
430
431 data = stktable_data_cast(data_ptr, std_t_uint);
432 intencode(data, &cursor);
433 break;
434 }
435 case STD_T_ULL: {
436 unsigned long long data;
437
438 data = stktable_data_cast(data_ptr, std_t_ull);
439 intencode(data, &cursor);
440 break;
441 }
442 case STD_T_FRQP: {
443 struct freq_ctr_period *frqp;
444
445 frqp = &stktable_data_cast(data_ptr, std_t_frqp);
446 intencode((unsigned int)(now_ms - frqp->curr_tick), &cursor);
447 intencode(frqp->curr_ctr, &cursor);
448 intencode(frqp->prev_ctr, &cursor);
449 break;
450 }
451 }
452 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200453 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100454 HA_RWLOCK_RDUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200455
456 /* Compute datalen */
457 datalen = (cursor - datamsg);
458
459 /* prepare message header */
460 msg[0] = PEER_MSG_CLASS_STICKTABLE;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200461 peer_set_update_msg_type(&msg[1], use_identifier, use_timed);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200462 cursor = &msg[2];
463 intencode(datalen, &cursor);
464
465 /* move data after header */
466 memmove(cursor, datamsg, datalen);
467
468 /* return header size + data_len */
469 return (cursor - msg) + datalen;
470}
471
472/*
473 * This prepare the switch table message to targeted share table <st>.
Joseph Herlant82b2f542018-11-15 12:19:14 -0800474 * <msg> is a buffer of <size> to receive data message content
Emeric Brunb3971ab2015-05-12 18:49:09 +0200475 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
476 * check size)
477 */
Frédéric Lécailled5fe14b2019-01-24 10:33:40 +0100478static int peer_prepare_switchmsg(char *msg, size_t size, struct peer_prep_params *params)
Emeric Brunb3971ab2015-05-12 18:49:09 +0200479{
480 int len;
481 unsigned short datalen;
Willy Tarreau83061a82018-07-13 11:56:34 +0200482 struct buffer *chunk;
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200483 char *cursor, *datamsg, *chunkp, *chunkq;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200484 uint64_t data = 0;
Emeric Brun94900952015-06-11 18:25:54 +0200485 unsigned int data_type;
Frédéric Lécailled5fe14b2019-01-24 10:33:40 +0100486 struct shared_table *st;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200487
Frédéric Lécailled5fe14b2019-01-24 10:33:40 +0100488 st = params->swtch.shared_table;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200489 cursor = datamsg = msg + 2 + 5;
490
491 /* Encode data */
492
493 /* encode local id */
494 intencode(st->local_id, &cursor);
495
496 /* encode table name */
497 len = strlen(st->table->id);
498 intencode(len, &cursor);
499 memcpy(cursor, st->table->id, len);
500 cursor += len;
501
502 /* encode table type */
503
504 intencode(st->table->type, &cursor);
505
506 /* encode table key size */
507 intencode(st->table->key_size, &cursor);
508
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200509 chunk = get_trash_chunk();
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200510 chunkp = chunkq = chunk->area;
Emeric Brun94900952015-06-11 18:25:54 +0200511 /* encode available known data types in table */
512 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
513 if (st->table->data_ofs[data_type]) {
514 switch (stktable_data_types[data_type].std_type) {
515 case STD_T_SINT:
516 case STD_T_UINT:
517 case STD_T_ULL:
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200518 data |= 1 << data_type;
519 break;
Emeric Brun94900952015-06-11 18:25:54 +0200520 case STD_T_FRQP:
521 data |= 1 << data_type;
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200522 intencode(data_type, &chunkq);
523 intencode(st->table->data_arg[data_type].u, &chunkq);
Emeric Brun94900952015-06-11 18:25:54 +0200524 break;
525 }
526 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200527 }
528 intencode(data, &cursor);
529
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200530 /* Encode stick-table entries duration. */
531 intencode(st->table->expire, &cursor);
532
533 if (chunkq > chunkp) {
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200534 chunk->data = chunkq - chunkp;
535 memcpy(cursor, chunk->area, chunk->data);
536 cursor += chunk->data;
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200537 }
538
Emeric Brunb3971ab2015-05-12 18:49:09 +0200539 /* Compute datalen */
540 datalen = (cursor - datamsg);
Emeric Brun2b920a12010-09-23 18:30:22 +0200541
Emeric Brunb3971ab2015-05-12 18:49:09 +0200542 /* prepare message header */
543 msg[0] = PEER_MSG_CLASS_STICKTABLE;
544 msg[1] = PEER_MSG_STKT_DEFINE;
545 cursor = &msg[2];
546 intencode(datalen, &cursor);
Emeric Brun2b920a12010-09-23 18:30:22 +0200547
Emeric Brunb3971ab2015-05-12 18:49:09 +0200548 /* move data after header */
549 memmove(cursor, datamsg, datalen);
550
551 /* return header size + data_len */
552 return (cursor - msg) + datalen;
Emeric Brun2b920a12010-09-23 18:30:22 +0200553}
554
Emeric Brunb3971ab2015-05-12 18:49:09 +0200555/*
556 * This prepare the acknowledge message on the stick session <ts>, <st> is the considered
557 * stick table.
Joseph Herlant82b2f542018-11-15 12:19:14 -0800558 * <msg> is a buffer of <size> to receive data message content
Emeric Brunb3971ab2015-05-12 18:49:09 +0200559 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
560 * check size)
561 */
Frédéric Lécailled5fe14b2019-01-24 10:33:40 +0100562static int peer_prepare_ackmsg(char *msg, size_t size, struct peer_prep_params *p)
Emeric Brunb3971ab2015-05-12 18:49:09 +0200563{
564 unsigned short datalen;
565 char *cursor, *datamsg;
566 uint32_t netinteger;
Frédéric Lécailled5fe14b2019-01-24 10:33:40 +0100567 struct shared_table *st;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200568
Emeric Brunb058f1c2015-09-22 15:50:18 +0200569 cursor = datamsg = msg + 2 + 5;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200570
Frédéric Lécailled5fe14b2019-01-24 10:33:40 +0100571 st = p->ack.shared_table;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200572 intencode(st->remote_id, &cursor);
573 netinteger = htonl(st->last_get);
574 memcpy(cursor, &netinteger, sizeof(netinteger));
575 cursor += sizeof(netinteger);
576
577 /* Compute datalen */
578 datalen = (cursor - datamsg);
579
580 /* prepare message header */
581 msg[0] = PEER_MSG_CLASS_STICKTABLE;
Emeric Brune1ab8082015-08-21 11:48:54 +0200582 msg[1] = PEER_MSG_STKT_ACK;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200583 cursor = &msg[2];
584 intencode(datalen, &cursor);
585
586 /* move data after header */
587 memmove(cursor, datamsg, datalen);
588
589 /* return header size + data_len */
590 return (cursor - msg) + datalen;
591}
Emeric Brun2b920a12010-09-23 18:30:22 +0200592
593/*
594 * Callback to release a session with a peer
595 */
Willy Tarreau00a37f02015-04-13 12:05:19 +0200596static void peer_session_release(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +0200597{
Willy Tarreau00a37f02015-04-13 12:05:19 +0200598 struct stream_interface *si = appctx->owner;
Willy Tarreau87b09662015-04-03 00:22:06 +0200599 struct stream *s = si_strm(si);
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200600 struct peer *peer = appctx->ctx.peers.ptr;
601 struct peers *peers = strm_fe(s)->parent;
Emeric Brun2b920a12010-09-23 18:30:22 +0200602
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100603 /* appctx->ctx.peers.ptr is not a peer session */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100604 if (appctx->st0 < PEER_SESS_ST_SENDSUCCESS)
Emeric Brun2b920a12010-09-23 18:30:22 +0200605 return;
606
607 /* peer session identified */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200608 if (peer) {
Willy Tarreau2d372c22018-11-05 17:12:27 +0100609 if (appctx->st0 == PEER_SESS_ST_WAITMSG)
610 HA_ATOMIC_SUB(&connected_peers, 1);
Willy Tarreau199ad242018-11-05 16:31:22 +0100611 HA_ATOMIC_SUB(&active_peers, 1);
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100612 HA_SPIN_LOCK(PEER_LOCK, &peer->lock);
Willy Tarreau9df94c22016-10-31 18:42:52 +0100613 if (peer->appctx == appctx) {
Emeric Brunb157d732015-08-21 12:00:30 +0200614 /* Re-init current table pointers to force announcement on re-connect */
615 peer->remote_table = peer->last_local_table = NULL;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200616 peer->appctx = NULL;
617 if (peer->flags & PEER_F_LEARN_ASSIGN) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200618 /* unassign current peer for learning */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200619 peer->flags &= ~(PEER_F_LEARN_ASSIGN);
620 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brun2b920a12010-09-23 18:30:22 +0200621
622 /* reschedule a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200623 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
Emeric Brun2b920a12010-09-23 18:30:22 +0200624 }
625 /* reset teaching and learning flags to 0 */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200626 peer->flags &= PEER_TEACH_RESET;
627 peer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200628 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100629 HA_SPIN_UNLOCK(PEER_LOCK, &peer->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200630 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200631 }
632}
633
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200634/* Retrieve the major and minor versions of peers protocol
635 * announced by a remote peer. <str> is a null-terminated
636 * string with the following format: "<maj_ver>.<min_ver>".
637 */
638static int peer_get_version(const char *str,
639 unsigned int *maj_ver, unsigned int *min_ver)
640{
641 unsigned int majv, minv;
642 const char *pos, *saved;
643 const char *end;
644
645 saved = pos = str;
646 end = str + strlen(str);
647
648 majv = read_uint(&pos, end);
649 if (saved == pos || *pos++ != '.')
650 return -1;
651
652 saved = pos;
653 minv = read_uint(&pos, end);
654 if (saved == pos || pos != end)
655 return -1;
656
657 *maj_ver = majv;
658 *min_ver = minv;
659
660 return 0;
661}
Emeric Brun2b920a12010-09-23 18:30:22 +0200662
663/*
Frédéric Lécaillece025572019-01-21 13:38:06 +0100664 * Parse a line terminated by an optional '\r' character, followed by a mandatory
665 * '\n' character.
666 * Returns 1 if succeeded or 0 if a '\n' character could not be found, and -1 if
667 * a line could not be read because the communication channel is closed.
668 */
669static inline int peer_getline(struct appctx *appctx)
670{
671 int n;
672 struct stream_interface *si = appctx->owner;
673
674 n = co_getline(si_oc(si), trash.area, trash.size);
675 if (!n)
676 return 0;
677
678 if (n < 0 || trash.area[n - 1] != '\n') {
679 appctx->st0 = PEER_SESS_ST_END;
680 return -1;
681 }
682
683 if (n > 1 && (trash.area[n - 2] == '\r'))
684 trash.area[n - 2] = 0;
685 else
686 trash.area[n - 1] = 0;
687
688 co_skip(si_oc(si), n);
689
690 return n;
691}
692
693/*
Frédéric Lécailleec44ea82019-01-22 15:54:53 +0100694 * Send a message after having called <peer_prepare_msg> to build it.
695 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
696 * Returns -1 if there was not enough room left to send the message,
697 * any other negative returned value must be considered as an error with an appcxt st0
698 * returned value equal to PEER_SESS_ST_END.
699 */
Frédéric Lécailled5fe14b2019-01-24 10:33:40 +0100700static inline int peer_send_msg(struct appctx *appctx,
701 int (*peer_prepare_msg)(char *, size_t, struct peer_prep_params *),
702 struct peer_prep_params *params)
Frédéric Lécailleec44ea82019-01-22 15:54:53 +0100703{
704 int ret, msglen;
705 struct stream_interface *si = appctx->owner;
706
Frédéric Lécailled5fe14b2019-01-24 10:33:40 +0100707 msglen = peer_prepare_msg(trash.area, trash.size, params);
Frédéric Lécailleec44ea82019-01-22 15:54:53 +0100708 if (!msglen) {
709 /* internal error: message does not fit in trash */
710 appctx->st0 = PEER_SESS_ST_END;
711 return 0;
712 }
713
714 /* message to buffer */
715 ret = ci_putblk(si_ic(si), trash.area, msglen);
716 if (ret <= 0) {
717 if (ret == -1) {
718 /* No more write possible */
719 si_rx_room_blk(si);
720 return -1;
721 }
722 appctx->st0 = PEER_SESS_ST_END;
723 }
724
725 return ret;
726}
727
728/*
Frédéric Lécaille7d0ceee2019-01-24 14:24:05 +0100729 * Send a hello message.
730 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
731 * Returns -1 if there was not enough room left to send the message,
732 * any other negative returned value must be considered as an error with an appcxt st0
733 * returned value equal to PEER_SESS_ST_END.
734 */
735static inline int peer_send_hellomsg(struct appctx *appctx, struct peer *peer)
736{
737 struct peer_prep_params p = {
738 .hello.peer = peer,
739 };
740
741 return peer_send_msg(appctx, peer_prepare_hellomsg, &p);
742}
743
744/*
745 * Send a success peer handshake status message.
746 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
747 * Returns -1 if there was not enough room left to send the message,
748 * any other negative returned value must be considered as an error with an appcxt st0
749 * returned value equal to PEER_SESS_ST_END.
750 */
751static inline int peer_send_status_successmsg(struct appctx *appctx)
752{
753 return peer_send_msg(appctx, peer_prepare_status_successmsg, NULL);
754}
755
756/*
757 * Send a peer handshake status error message.
758 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
759 * Returns -1 if there was not enough room left to send the message,
760 * any other negative returned value must be considered as an error with an appcxt st0
761 * returned value equal to PEER_SESS_ST_END.
762 */
763static inline int peer_send_status_errormsg(struct appctx *appctx)
764{
765 struct peer_prep_params p = {
766 .error_status.st1 = appctx->st1,
767 };
768
769 return peer_send_msg(appctx, peer_prepare_status_errormsg, &p);
770}
771
772/*
Frédéric Lécailleec44ea82019-01-22 15:54:53 +0100773 * Send a stick-table switch message.
774 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
775 * Returns -1 if there was not enough room left to send the message,
776 * any other negative returned value must be considered as an error with an appcxt st0
777 * returned value equal to PEER_SESS_ST_END.
778 */
779static inline int peer_send_switchmsg(struct shared_table *st, struct appctx *appctx)
780{
Frédéric Lécailled5fe14b2019-01-24 10:33:40 +0100781 struct peer_prep_params p = {
782 .swtch.shared_table = st,
783 };
784
785 return peer_send_msg(appctx, peer_prepare_switchmsg, &p);
Frédéric Lécailleec44ea82019-01-22 15:54:53 +0100786}
787
788/*
789 * Send a stick-table update acknowledgement message.
790 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
791 * Returns -1 if there was not enough room left to send the message,
792 * any other negative returned value must be considered as an error with an appcxt st0
793 * returned value equal to PEER_SESS_ST_END.
794 */
795static inline int peer_send_ackmsg(struct shared_table *st, struct appctx *appctx)
796{
Frédéric Lécailled5fe14b2019-01-24 10:33:40 +0100797 struct peer_prep_params p = {
798 .ack.shared_table = st,
799 };
800
801 return peer_send_msg(appctx, peer_prepare_ackmsg, &p);
Frédéric Lécailleec44ea82019-01-22 15:54:53 +0100802}
803
804/*
Frédéric Lécaille87f554c2019-01-22 17:26:50 +0100805 * Send a stick-table update message.
806 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
807 * Returns -1 if there was not enough room left to send the message,
808 * any other negative returned value must be considered as an error with an appcxt st0
809 * returned value equal to PEER_SESS_ST_END.
810 */
811static inline int peer_send_updatemsg(struct shared_table *st, struct appctx *appctx, struct stksess *ts,
812 unsigned int updateid, int use_identifier, int use_timed)
813{
Frédéric Lécailled5fe14b2019-01-24 10:33:40 +0100814 struct peer_prep_params p = {
815 .updt.stksess = ts,
816 .updt.shared_table = st,
817 .updt.updateid = updateid,
818 .updt.use_identifier = use_identifier,
819 .updt.use_timed = use_timed,
820 };
821
822 return peer_send_msg(appctx, peer_prepare_updatemsg, &p);
Frédéric Lécaille87f554c2019-01-22 17:26:50 +0100823}
824
Frédéric Lécaille7d0ceee2019-01-24 14:24:05 +0100825/*
826 * Build a peer protocol control class message.
827 * Returns the number of written bytes used to build the message if succeeded,
828 * 0 if not.
829 */
830static int peer_prepare_control_msg(char *msg, size_t size, struct peer_prep_params *p)
831{
832 if (size < sizeof p->control.head)
833 return 0;
834
835 msg[0] = p->control.head[0];
836 msg[1] = p->control.head[1];
837
838 return 2;
839}
Frédéric Lécaille6a8303d2019-01-22 22:25:17 +0100840
841/*
Frédéric Lécaille7d0ceee2019-01-24 14:24:05 +0100842 * Send a stick-table synchronization request message.
843 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
844 * Returns -1 if there was not enough room left to send the message,
845 * any other negative returned value must be considered as an error with an appctx st0
846 * returned value equal to PEER_SESS_ST_END.
847 */
848static inline int peer_send_resync_reqmsg(struct appctx *appctx)
849{
850 struct peer_prep_params p = {
851 .control.head = { PEER_MSG_CLASS_CONTROL, PEER_MSG_CTRL_RESYNCREQ, },
852 };
853
854 return peer_send_msg(appctx, peer_prepare_control_msg, &p);
855}
856
857/*
858 * Send a stick-table synchronization confirmation message.
859 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
860 * Returns -1 if there was not enough room left to send the message,
861 * any other negative returned value must be considered as an error with an appctx st0
862 * returned value equal to PEER_SESS_ST_END.
863 */
864static inline int peer_send_resync_confirmsg(struct appctx *appctx)
865{
866 struct peer_prep_params p = {
867 .control.head = { PEER_MSG_CLASS_CONTROL, PEER_MSG_CTRL_RESYNCCONFIRM, },
868 };
869
870 return peer_send_msg(appctx, peer_prepare_control_msg, &p);
871}
872
873/*
874 * Send a stick-table synchronization finished message.
875 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
876 * Returns -1 if there was not enough room left to send the message,
877 * any other negative returned value must be considered as an error with an appctx st0
878 * returned value equal to PEER_SESS_ST_END.
879 */
880static inline int peer_send_resync_finishedmsg(struct appctx *appctx, struct peer *peer)
881{
882 struct peer_prep_params p = {
883 .control.head = { PEER_MSG_CLASS_CONTROL, },
884 };
885
886 p.control.head[1] = (peer->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FINISHED ?
887 PEER_MSG_CTRL_RESYNCFINISHED : PEER_MSG_CTRL_RESYNCPARTIAL;
888
889 return peer_send_msg(appctx, peer_prepare_control_msg, &p);
890}
891
892/*
893 * Build a peer protocol error class message.
894 * Returns the number of written bytes used to build the message if succeeded,
895 * 0 if not.
896 */
897static int peer_prepare_error_msg(char *msg, size_t size, struct peer_prep_params *p)
898{
899 if (size < sizeof p->error.head)
900 return 0;
901
902 msg[0] = p->error.head[0];
903 msg[1] = p->error.head[1];
904
905 return 2;
906}
907
908/*
909 * Send a "size limit reached" error message.
910 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
911 * Returns -1 if there was not enough room left to send the message,
912 * any other negative returned value must be considered as an error with an appctx st0
913 * returned value equal to PEER_SESS_ST_END.
914 */
915static inline int peer_send_error_size_limitmsg(struct appctx *appctx)
916{
917 struct peer_prep_params p = {
918 .error.head = { PEER_MSG_CLASS_ERROR, PEER_MSG_ERR_SIZELIMIT, },
919 };
920
921 return peer_send_msg(appctx, peer_prepare_error_msg, &p);
922}
923
924/*
925 * Send a "peer protocol" error message.
926 * Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
927 * Returns -1 if there was not enough room left to send the message,
928 * any other negative returned value must be considered as an error with an appctx st0
929 * returned value equal to PEER_SESS_ST_END.
930 */
931static inline int peer_send_error_protomsg(struct appctx *appctx)
932{
933 struct peer_prep_params p = {
934 .error.head = { PEER_MSG_CLASS_ERROR, PEER_MSG_ERR_PROTOCOL, },
935 };
936
937 return peer_send_msg(appctx, peer_prepare_error_msg, &p);
938}
939
940/*
Frédéric Lécaille6a8303d2019-01-22 22:25:17 +0100941 * Function used to lookup for recent stick-table updates associated with
942 * <st> shared stick-table when a lesson must be taught a peer (PEER_F_LEARN_ASSIGN flag set).
943 */
944static inline struct stksess *peer_teach_process_stksess_lookup(struct shared_table *st)
945{
946 struct eb32_node *eb;
947
948 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
949 if (!eb) {
950 eb = eb32_first(&st->table->updates);
951 if (!eb || ((int)(eb->key - st->last_pushed) <= 0)) {
952 st->table->commitupdate = st->last_pushed = st->table->localupdate;
953 return NULL;
954 }
955 }
956
957 if ((int)(eb->key - st->table->localupdate) > 0) {
958 st->table->commitupdate = st->last_pushed = st->table->localupdate;
959 return NULL;
960 }
961
962 return eb32_entry(eb, struct stksess, upd);
963}
964
965/*
966 * Function used to lookup for recent stick-table updates associated with
967 * <st> shared stick-table during teach state 1 step.
968 */
969static inline struct stksess *peer_teach_stage1_stksess_lookup(struct shared_table *st)
970{
971 struct eb32_node *eb;
972
973 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
974 if (!eb) {
975 st->flags |= SHTABLE_F_TEACH_STAGE1;
976 eb = eb32_first(&st->table->updates);
977 if (eb)
978 st->last_pushed = eb->key - 1;
979 return NULL;
980 }
981
982 return eb32_entry(eb, struct stksess, upd);
983}
984
985/*
986 * Function used to lookup for recent stick-table updates associated with
987 * <st> shared stick-table during teach state 2 step.
988 */
989static inline struct stksess *peer_teach_stage2_stksess_lookup(struct shared_table *st)
990{
991 struct eb32_node *eb;
992
993 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
994 if (!eb || eb->key > st->teaching_origin) {
995 st->flags |= SHTABLE_F_TEACH_STAGE2;
996 return NULL;
997 }
998
999 return eb32_entry(eb, struct stksess, upd);
1000}
1001
1002/*
1003 * Generic function to emit update messages for <st> stick-table when a lesson must
1004 * be taught to the peer <p>.
1005 * <locked> must be set to 1 if the shared table <st> is already locked when entering
1006 * this function, 0 if not.
1007 *
1008 * This function temporary unlock/lock <st> when it sends stick-table updates or
1009 * when decrementing its refcount in case of any error when it sends this updates.
1010 *
1011 * Return 0 if any message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1012 * Returns -1 if there was not enough room left to send the message,
1013 * any other negative returned value must be considered as an error with an appcxt st0
1014 * returned value equal to PEER_SESS_ST_END.
1015 * If it returns 0 or -1, this function leave <st> locked if already locked when entering this function
1016 * unlocked if not already locked when entering this function.
1017 */
1018static inline int peer_send_teachmsgs(struct appctx *appctx, struct peer *p,
1019 struct stksess *(*peer_stksess_lookup)(struct shared_table *),
1020 struct shared_table *st, int locked)
1021{
1022 int ret, new_pushed, use_timed;
1023
1024 ret = 1;
1025 use_timed = 0;
1026 if (st != p->last_local_table) {
1027 ret = peer_send_switchmsg(st, appctx);
1028 if (ret <= 0)
1029 return ret;
1030
1031 p->last_local_table = st;
1032 }
1033
1034 if (peer_stksess_lookup != peer_teach_process_stksess_lookup)
1035 use_timed = !(p->flags & PEER_F_DWNGRD);
1036
1037 /* We force new pushed to 1 to force identifier in update message */
1038 new_pushed = 1;
1039
1040 if (!locked)
1041 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
1042
1043 while (1) {
1044 struct stksess *ts;
1045 unsigned updateid;
1046
1047 /* push local updates */
1048 ts = peer_stksess_lookup(st);
1049 if (!ts)
1050 break;
1051
1052 updateid = ts->upd.key;
1053 ts->ref_cnt++;
1054 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
1055
1056 ret = peer_send_updatemsg(st, appctx, ts, updateid, new_pushed, use_timed);
1057 if (ret <= 0) {
1058 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
1059 ts->ref_cnt--;
1060 if (!locked)
1061 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
1062 return ret;
1063 }
1064
1065 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
1066 ts->ref_cnt--;
1067 st->last_pushed = updateid;
1068
1069 if (peer_stksess_lookup == peer_teach_process_stksess_lookup &&
1070 (int)(st->last_pushed - st->table->commitupdate) > 0)
1071 st->table->commitupdate = st->last_pushed;
1072
1073 /* identifier may not needed in next update message */
1074 new_pushed = 0;
1075 }
1076
1077 out:
1078 if (!locked)
1079 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
1080 return 1;
1081}
1082
1083/*
1084 * Function to emit update messages for <st> stick-table when a lesson must
1085 * be taught to the peer <p> (PEER_F_LEARN_ASSIGN flag set).
1086 *
1087 * Note that <st> shared stick-table is locked when calling this function.
1088 *
1089 * Return 0 if any message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1090 * Returns -1 if there was not enough room left to send the message,
1091 * any other negative returned value must be considered as an error with an appcxt st0
1092 * returned value equal to PEER_SESS_ST_END.
1093 */
1094static inline int peer_send_teach_process_msgs(struct appctx *appctx, struct peer *p,
1095 struct shared_table *st)
1096{
1097 return peer_send_teachmsgs(appctx, p, peer_teach_process_stksess_lookup, st, 1);
1098}
1099
1100/*
1101 * Function to emit update messages for <st> stick-table when a lesson must
1102 * be taught to the peer <p> during teach state 1 step.
1103 *
1104 * Return 0 if any message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1105 * Returns -1 if there was not enough room left to send the message,
1106 * any other negative returned value must be considered as an error with an appcxt st0
1107 * returned value equal to PEER_SESS_ST_END.
1108 */
1109static inline int peer_send_teach_stage1_msgs(struct appctx *appctx, struct peer *p,
1110 struct shared_table *st)
1111{
1112 return peer_send_teachmsgs(appctx, p, peer_teach_stage1_stksess_lookup, st, 0);
1113}
1114
1115/*
1116 * Function to emit update messages for <st> stick-table when a lesson must
1117 * be taught to the peer <p> during teach state 1 step.
1118 *
1119 * Return 0 if any message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
1120 * Returns -1 if there was not enough room left to send the message,
1121 * any other negative returned value must be considered as an error with an appcxt st0
1122 * returned value equal to PEER_SESS_ST_END.
1123 */
1124static inline int peer_send_teach_stage2_msgs(struct appctx *appctx, struct peer *p,
1125 struct shared_table *st)
1126{
1127 return peer_send_teachmsgs(appctx, p, peer_teach_stage2_stksess_lookup, st, 0);
1128}
1129
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001130
1131/*
1132 * Function used to parse a stick-table update message after it has been received
1133 * by <p> peer with <msg_cur> as address of the pointer to the position in the
1134 * receipt buffer with <msg_end> being position of the end of the stick-table message.
1135 * Update <msg_curr> accordingly to the peer protocol specs if no peer protocol error
1136 * was encountered.
1137 * <exp> must be set if the stick-table entry expires.
1138 * <updt> must be set for PEER_MSG_STKT_UPDATE or PEER_MSG_STKT_UPDATE_TIMED stick-table
1139 * messages, in this case the stick-table udpate message is received with a stick-table
1140 * update ID.
1141 * <totl> is the length of the stick-table update message computed upon receipt.
1142 */
Frédéric Lécaille444243c2019-01-24 15:40:11 +01001143static int peer_treat_updatemsg(struct appctx *appctx, struct peer *p, int updt, int exp,
1144 char **msg_cur, char *msg_end, int msg_len, int totl)
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001145{
1146 struct stream_interface *si = appctx->owner;
1147 struct shared_table *st = p->remote_table;
1148 struct stksess *ts, *newts;
1149 uint32_t update;
1150 int expire;
1151 unsigned int data_type;
1152 void *data_ptr;
1153
1154 /* Here we have data message */
1155 if (!st)
1156 goto ignore_msg;
1157
1158 expire = MS_TO_TICKS(st->table->expire);
1159
1160 if (updt) {
Willy Tarreau1e82a142019-01-29 11:08:06 +01001161 if (msg_len < sizeof(update))
1162 goto malformed_exit;
1163
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001164 memcpy(&update, *msg_cur, sizeof(update));
1165 *msg_cur += sizeof(update);
1166 st->last_get = htonl(update);
1167 }
1168 else {
1169 st->last_get++;
1170 }
1171
1172 if (exp) {
1173 size_t expire_sz = sizeof expire;
1174
Willy Tarreau1e82a142019-01-29 11:08:06 +01001175 if (*msg_cur + expire_sz > msg_end)
1176 goto malformed_exit;
1177
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001178 memcpy(&expire, *msg_cur, expire_sz);
1179 *msg_cur += expire_sz;
1180 expire = ntohl(expire);
1181 }
1182
1183 newts = stksess_new(st->table, NULL);
1184 if (!newts)
1185 goto ignore_msg;
1186
1187 if (st->table->type == SMP_T_STR) {
1188 unsigned int to_read, to_store;
1189
1190 to_read = intdecode(msg_cur, msg_end);
Willy Tarreau1e82a142019-01-29 11:08:06 +01001191 if (!*msg_cur)
1192 goto malformed_free_newts;
1193
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001194 to_store = MIN(to_read, st->table->key_size - 1);
Willy Tarreau1e82a142019-01-29 11:08:06 +01001195 if (*msg_cur + to_store > msg_end)
1196 goto malformed_free_newts;
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001197
1198 memcpy(newts->key.key, *msg_cur, to_store);
1199 newts->key.key[to_store] = 0;
1200 *msg_cur += to_read;
1201 }
1202 else if (st->table->type == SMP_T_SINT) {
1203 unsigned int netinteger;
1204
Willy Tarreau1e82a142019-01-29 11:08:06 +01001205 if (*msg_cur + sizeof(netinteger) > msg_end)
1206 goto malformed_free_newts;
1207
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001208 memcpy(&netinteger, *msg_cur, sizeof(netinteger));
1209 netinteger = ntohl(netinteger);
1210 memcpy(newts->key.key, &netinteger, sizeof(netinteger));
1211 *msg_cur += sizeof(netinteger);
1212 }
1213 else {
Willy Tarreau1e82a142019-01-29 11:08:06 +01001214 if (*msg_cur + st->table->key_size > msg_end)
1215 goto malformed_free_newts;
1216
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001217 memcpy(newts->key.key, *msg_cur, st->table->key_size);
1218 *msg_cur += st->table->key_size;
1219 }
1220
1221 /* lookup for existing entry */
1222 ts = stktable_set_entry(st->table, newts);
1223 if (ts != newts) {
1224 stksess_free(st->table, newts);
1225 newts = NULL;
1226 }
1227
1228 HA_RWLOCK_WRLOCK(STK_SESS_LOCK, &ts->lock);
1229
1230 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
Willy Tarreau1e82a142019-01-29 11:08:06 +01001231 uint64_t decoded_int;
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001232
1233 if (!((1 << data_type) & st->remote_data))
1234 continue;
1235
Willy Tarreau1e82a142019-01-29 11:08:06 +01001236 decoded_int = intdecode(msg_cur, msg_end);
1237 if (!*msg_cur)
1238 goto malformed_unlock;
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001239
Willy Tarreau1e82a142019-01-29 11:08:06 +01001240 switch (stktable_data_types[data_type].std_type) {
1241 case STD_T_SINT:
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001242 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1243 if (data_ptr)
Willy Tarreau1e82a142019-01-29 11:08:06 +01001244 stktable_data_cast(data_ptr, std_t_sint) = decoded_int;
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001245 break;
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001246
Willy Tarreau1e82a142019-01-29 11:08:06 +01001247 case STD_T_UINT:
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001248 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1249 if (data_ptr)
Willy Tarreau1e82a142019-01-29 11:08:06 +01001250 stktable_data_cast(data_ptr, std_t_uint) = decoded_int;
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001251 break;
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001252
Willy Tarreau1e82a142019-01-29 11:08:06 +01001253 case STD_T_ULL:
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001254 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1255 if (data_ptr)
Willy Tarreau1e82a142019-01-29 11:08:06 +01001256 stktable_data_cast(data_ptr, std_t_ull) = decoded_int;
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001257 break;
Willy Tarreau1e82a142019-01-29 11:08:06 +01001258
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001259 case STD_T_FRQP: {
1260 struct freq_ctr_period data;
1261
1262 /* First bit is reserved for the freq_ctr_period lock
1263 Note: here we're still protected by the stksess lock
1264 so we don't need to update the update the freq_ctr_period
1265 using its internal lock */
1266
Willy Tarreau1e82a142019-01-29 11:08:06 +01001267 data.curr_tick = tick_add(now_ms, -decoded_int) & ~0x1;
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001268 data.curr_ctr = intdecode(msg_cur, msg_end);
Willy Tarreau1e82a142019-01-29 11:08:06 +01001269 if (!*msg_cur)
1270 goto malformed_unlock;
1271
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001272 data.prev_ctr = intdecode(msg_cur, msg_end);
Willy Tarreau1e82a142019-01-29 11:08:06 +01001273 if (!*msg_cur)
1274 goto malformed_unlock;
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001275
1276 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1277 if (data_ptr)
1278 stktable_data_cast(data_ptr, std_t_frqp) = data;
1279 break;
1280 }
1281 }
1282 }
1283 /* Force new expiration */
1284 ts->expire = tick_add(now_ms, expire);
1285
1286 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
1287 stktable_touch_remote(st->table, ts, 1);
1288 return 1;
1289
1290 ignore_msg:
1291 /* skip consumed message */
1292 co_skip(si_oc(si), totl);
1293 return 0;
Willy Tarreau1e82a142019-01-29 11:08:06 +01001294
1295 malformed_unlock:
1296 /* malformed message */
1297 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
1298 stktable_touch_remote(st->table, ts, 1);
1299 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1300 return 0;
1301
1302 malformed_free_newts:
1303 /* malformed message */
1304 stksess_free(st->table, newts);
1305 malformed_exit:
1306 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1307 return 0;
Frédéric Lécaille168a34b2019-01-23 11:16:57 +01001308}
1309
Frédéric Lécaille87f554c2019-01-22 17:26:50 +01001310/*
Frédéric Lécailled27b0942019-01-23 17:31:37 +01001311 * Function used to parse a stick-table update acknowledgement message after it
1312 * has been received by <p> peer with <msg_cur> as address of the pointer to the position in the
1313 * receipt buffer with <msg_end> being the position of the end of the stick-table message.
1314 * Update <msg_curr> accordingly to the peer protocol specs if no peer protocol error
1315 * was encountered.
1316 * Return 1 if succeeded, 0 if not with the appctx state st0 set to PEER_SESS_ST_ERRPROTO.
1317 */
1318static inline int peer_treat_ackmsg(struct appctx *appctx, struct peer *p,
1319 char **msg_cur, char *msg_end)
1320{
1321 /* ack message */
1322 uint32_t table_id ;
1323 uint32_t update;
1324 struct shared_table *st;
1325
1326 table_id = intdecode(msg_cur, msg_end);
1327 if (!*msg_cur || (*msg_cur + sizeof(update) > msg_end)) {
1328 /* malformed message */
1329 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1330 return 0;
1331 }
1332
1333 memcpy(&update, *msg_cur, sizeof(update));
1334 update = ntohl(update);
1335
1336 for (st = p->tables; st; st = st->next) {
1337 if (st->local_id == table_id) {
1338 st->update = update;
1339 break;
1340 }
1341 }
1342
1343 return 1;
1344}
1345
1346/*
1347 * Function used to parse a stick-table switch message after it has been received
1348 * by <p> peer with <msg_cur> as address of the pointer to the position in the
1349 * receipt buffer with <msg_end> being the position of the end of the stick-table message.
1350 * Update <msg_curr> accordingly to the peer protocol specs if no peer protocol error
1351 * was encountered.
1352 * Return 1 if succeeded, 0 if not with the appctx state st0 set to PEER_SESS_ST_ERRPROTO.
1353 */
1354static inline int peer_treat_switchmsg(struct appctx *appctx, struct peer *p,
1355 char **msg_cur, char *msg_end)
1356{
1357 struct shared_table *st;
1358 int table_id;
1359
1360 table_id = intdecode(msg_cur, msg_end);
1361 if (!*msg_cur) {
1362 /* malformed message */
1363 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1364 return 0;
1365 }
1366
1367 p->remote_table = NULL;
1368 for (st = p->tables; st; st = st->next) {
1369 if (st->remote_id == table_id) {
1370 p->remote_table = st;
1371 break;
1372 }
1373 }
1374
1375 return 1;
1376}
1377
1378/*
1379 * Function used to parse a stick-table definition message after it has been received
1380 * by <p> peer with <msg_cur> as address of the pointer to the position in the
1381 * receipt buffer with <msg_end> being the position of the end of the stick-table message.
1382 * Update <msg_curr> accordingly to the peer protocol specs if no peer protocol error
1383 * was encountered.
1384 * <totl> is the length of the stick-table update message computed upon receipt.
1385 * Return 1 if succeeded, 0 if not with the appctx state st0 set to PEER_SESS_ST_ERRPROTO.
1386 */
1387static inline int peer_treat_definemsg(struct appctx *appctx, struct peer *p,
1388 char **msg_cur, char *msg_end, int totl)
1389{
1390 struct stream_interface *si = appctx->owner;
1391 int table_id_len;
1392 struct shared_table *st;
1393 int table_type;
1394 int table_keylen;
1395 int table_id;
1396 uint64_t table_data;
1397
1398 table_id = intdecode(msg_cur, msg_end);
Willy Tarreau6f731f32019-01-29 11:11:23 +01001399 if (!*msg_cur)
1400 goto malformed_exit;
Frédéric Lécailled27b0942019-01-23 17:31:37 +01001401
1402 table_id_len = intdecode(msg_cur, msg_end);
Willy Tarreau6f731f32019-01-29 11:11:23 +01001403 if (!*msg_cur)
1404 goto malformed_exit;
Frédéric Lécailled27b0942019-01-23 17:31:37 +01001405
1406 p->remote_table = NULL;
Willy Tarreau6f731f32019-01-29 11:11:23 +01001407 if (!table_id_len || (*msg_cur + table_id_len) >= msg_end)
1408 goto malformed_exit;
Frédéric Lécailled27b0942019-01-23 17:31:37 +01001409
1410 for (st = p->tables; st; st = st->next) {
1411 /* Reset IDs */
1412 if (st->remote_id == table_id)
1413 st->remote_id = 0;
1414
1415 if (!p->remote_table && (table_id_len == strlen(st->table->id)) &&
1416 (memcmp(st->table->id, *msg_cur, table_id_len) == 0))
1417 p->remote_table = st;
1418 }
1419
1420 if (!p->remote_table)
1421 goto ignore_msg;
1422
1423 *msg_cur += table_id_len;
Willy Tarreau6f731f32019-01-29 11:11:23 +01001424 if (*msg_cur >= msg_end)
1425 goto malformed_exit;
Frédéric Lécailled27b0942019-01-23 17:31:37 +01001426
1427 table_type = intdecode(msg_cur, msg_end);
Willy Tarreau6f731f32019-01-29 11:11:23 +01001428 if (!*msg_cur)
1429 goto malformed_exit;
Frédéric Lécailled27b0942019-01-23 17:31:37 +01001430
1431 table_keylen = intdecode(msg_cur, msg_end);
Willy Tarreau6f731f32019-01-29 11:11:23 +01001432 if (!*msg_cur)
1433 goto malformed_exit;
Frédéric Lécailled27b0942019-01-23 17:31:37 +01001434
1435 table_data = intdecode(msg_cur, msg_end);
Willy Tarreau6f731f32019-01-29 11:11:23 +01001436 if (!*msg_cur)
1437 goto malformed_exit;
Frédéric Lécailled27b0942019-01-23 17:31:37 +01001438
1439 if (p->remote_table->table->type != table_type
1440 || p->remote_table->table->key_size != table_keylen) {
1441 p->remote_table = NULL;
1442 goto ignore_msg;
1443 }
1444
1445 p->remote_table->remote_data = table_data;
1446 p->remote_table->remote_id = table_id;
1447 return 1;
1448
1449 ignore_msg:
1450 co_skip(si_oc(si), totl);
1451 return 0;
Willy Tarreau6f731f32019-01-29 11:11:23 +01001452
1453 malformed_exit:
1454 /* malformed message */
1455 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1456 return 0;
Frédéric Lécailled27b0942019-01-23 17:31:37 +01001457}
1458
1459/*
Frédéric Lécaille95203f22019-01-23 19:38:11 +01001460 * Receive a stick-table message.
1461 * Returns 1 if there was no error, if not, returns 0 if not enough data were available,
1462 * -1 if there was an error updating the appctx state st0 accordingly.
1463 */
1464static inline int peer_recv_msg(struct appctx *appctx, char *msg_head, size_t msg_head_sz,
1465 uint32_t *msg_len, int *totl)
1466{
1467 int reql;
1468 struct stream_interface *si = appctx->owner;
1469
1470 reql = co_getblk(si_oc(si), msg_head, 2 * sizeof(char), *totl);
1471 if (reql <= 0) /* closed or EOL not found */
1472 goto incomplete;
1473
1474 *totl += reql;
1475
1476 if ((unsigned int)msg_head[1] < 128)
1477 return 1;
1478
1479 /* Read and Decode message length */
1480 reql = co_getblk(si_oc(si), &msg_head[2], sizeof(char), *totl);
1481 if (reql <= 0) /* closed */
1482 goto incomplete;
1483
1484 *totl += reql;
1485
1486 if ((unsigned int)msg_head[2] < 240) {
1487 *msg_len = msg_head[2];
1488 }
1489 else {
1490 int i;
1491 char *cur;
1492 char *end;
1493
1494 for (i = 3 ; i < msg_head_sz ; i++) {
1495 reql = co_getblk(si_oc(si), &msg_head[i], sizeof(char), *totl);
1496 if (reql <= 0) /* closed */
1497 goto incomplete;
1498
1499 *totl += reql;
1500
1501 if (!(msg_head[i] & 0x80))
1502 break;
1503 }
1504
1505 if (i == msg_head_sz) {
1506 /* malformed message */
1507 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1508 return -1;
1509 }
1510 end = msg_head + msg_head_sz;
1511 cur = &msg_head[2];
1512 *msg_len = intdecode(&cur, end);
1513 if (!cur) {
1514 /* malformed message */
1515 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1516 return -1;
1517 }
1518 }
1519
1520 /* Read message content */
1521 if (*msg_len) {
1522 if (*msg_len > trash.size) {
1523 /* Status code is not success, abort */
1524 appctx->st0 = PEER_SESS_ST_ERRSIZE;
1525 return -1;
1526 }
1527
1528 reql = co_getblk(si_oc(si), trash.area, *msg_len, *totl);
1529 if (reql <= 0) /* closed */
1530 goto incomplete;
1531 *totl += reql;
1532 }
1533
1534 return 1;
1535
1536 incomplete:
1537 if (reql < 0) {
1538 /* there was an error */
1539 appctx->st0 = PEER_SESS_ST_END;
1540 return -1;
1541 }
1542
1543 return 0;
1544}
Frédéric Lécaille444243c2019-01-24 15:40:11 +01001545
1546/*
1547 * Treat the awaited message with <msg_head> as header.*
1548 * Return 1 if succeeded, 0 if not.
1549 */
1550static inline int peer_treat_awaited_msg(struct appctx *appctx, struct peer *peer, unsigned char *msg_head,
1551 char **msg_cur, char *msg_end, int msg_len, int totl)
1552{
1553 struct stream_interface *si = appctx->owner;
1554 struct stream *s = si_strm(si);
1555 struct peers *peers = strm_fe(s)->parent;
1556
1557 if (msg_head[0] == PEER_MSG_CLASS_CONTROL) {
1558 if (msg_head[1] == PEER_MSG_CTRL_RESYNCREQ) {
1559 struct shared_table *st;
1560 /* Reset message: remote need resync */
1561
1562 /* prepare tables fot a global push */
1563 for (st = peer->tables; st; st = st->next) {
1564 st->teaching_origin = st->last_pushed = st->table->update;
1565 st->flags = 0;
1566 }
1567
1568 /* reset teaching flags to 0 */
1569 peer->flags &= PEER_TEACH_RESET;
1570
1571 /* flag to start to teach lesson */
1572 peer->flags |= PEER_F_TEACH_PROCESS;
1573 }
1574 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCFINISHED) {
1575 if (peer->flags & PEER_F_LEARN_ASSIGN) {
1576 peer->flags &= ~PEER_F_LEARN_ASSIGN;
1577 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
1578 peers->flags |= (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE);
1579 }
1580 peer->confirm++;
1581 }
1582 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCPARTIAL) {
1583 if (peer->flags & PEER_F_LEARN_ASSIGN) {
1584 peer->flags &= ~PEER_F_LEARN_ASSIGN;
1585 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
1586
1587 peer->flags |= PEER_F_LEARN_NOTUP2DATE;
1588 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
1589 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
1590 }
1591 peer->confirm++;
1592 }
1593 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCCONFIRM) {
1594 struct shared_table *st;
1595
1596 /* If stopping state */
1597 if (stopping) {
1598 /* Close session, push resync no more needed */
1599 peer->flags |= PEER_F_TEACH_COMPLETE;
1600 appctx->st0 = PEER_SESS_ST_END;
1601 return 0;
1602 }
1603 for (st = peer->tables; st; st = st->next) {
1604 st->update = st->last_pushed = st->teaching_origin;
1605 st->flags = 0;
1606 }
1607
1608 /* reset teaching flags to 0 */
1609 peer->flags &= PEER_TEACH_RESET;
1610 }
1611 }
1612 else if (msg_head[0] == PEER_MSG_CLASS_STICKTABLE) {
1613 if (msg_head[1] == PEER_MSG_STKT_DEFINE) {
1614 if (!peer_treat_definemsg(appctx, peer, msg_cur, msg_end, totl))
1615 return 0;
1616 }
1617 else if (msg_head[1] == PEER_MSG_STKT_SWITCH) {
1618 if (!peer_treat_switchmsg(appctx, peer, msg_cur, msg_end))
1619 return 0;
1620 }
1621 else if (msg_head[1] == PEER_MSG_STKT_UPDATE ||
1622 msg_head[1] == PEER_MSG_STKT_INCUPDATE ||
1623 msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED ||
1624 msg_head[1] == PEER_MSG_STKT_INCUPDATE_TIMED) {
1625 int update, expire;
1626
1627 update = msg_head[1] == PEER_MSG_STKT_UPDATE || msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED;
1628 expire = msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED || msg_head[1] == PEER_MSG_STKT_INCUPDATE_TIMED;
1629 if (!peer_treat_updatemsg(appctx, peer, update, expire,
1630 msg_cur, msg_end, msg_len, totl))
1631 return 0;
1632
1633 }
1634 else if (msg_head[1] == PEER_MSG_STKT_ACK) {
1635 if (!peer_treat_ackmsg(appctx, peer, msg_cur, msg_end))
1636 return 0;
1637 }
1638 }
1639 else if (msg_head[0] == PEER_MSG_CLASS_RESERVED) {
1640 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1641 return 0;
1642 }
1643
1644 return 1;
1645}
Frédéric Lécaille25e1d5e2019-01-24 17:33:48 +01001646
1647
1648/*
1649 * Send any message to <peer> peer.
1650 * Returns 1 if succeeded, or -1 or 0 if failed.
1651 * -1 means an internal error occured, 0 is for a peer protocol error leading
1652 * to a peer state change (from the peer I/O handler point of view).
1653 */
1654static inline int peer_send_msgs(struct appctx *appctx, struct peer *peer)
1655{
1656 int repl;
1657 struct stream_interface *si = appctx->owner;
1658 struct stream *s = si_strm(si);
1659 struct peers *peers = strm_fe(s)->parent;
1660
1661 /* Need to request a resync */
1662 if ((peer->flags & PEER_F_LEARN_ASSIGN) &&
1663 (peers->flags & PEERS_F_RESYNC_ASSIGN) &&
1664 !(peers->flags & PEERS_F_RESYNC_PROCESS)) {
1665
1666 repl = peer_send_resync_reqmsg(appctx);
1667 if (repl <= 0)
1668 return repl;
1669
1670 peers->flags |= PEERS_F_RESYNC_PROCESS;
1671 }
1672
1673 /* Nothing to read, now we start to write */
1674 if (peer->tables) {
1675 struct shared_table *st;
1676 struct shared_table *last_local_table;
1677
1678 last_local_table = peer->last_local_table;
1679 if (!last_local_table)
1680 last_local_table = peer->tables;
1681 st = last_local_table->next;
1682
1683 while (1) {
1684 if (!st)
1685 st = peer->tables;
1686
1687 /* It remains some updates to ack */
1688 if (st->last_get != st->last_acked) {
1689 repl = peer_send_ackmsg(st, appctx);
1690 if (repl <= 0)
1691 return repl;
1692
1693 st->last_acked = st->last_get;
1694 }
1695
1696 if (!(peer->flags & PEER_F_TEACH_PROCESS)) {
1697 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
1698 if (!(peer->flags & PEER_F_LEARN_ASSIGN) &&
1699 ((int)(st->last_pushed - st->table->localupdate) < 0)) {
1700
1701 repl = peer_send_teach_process_msgs(appctx, peer, st);
1702 if (repl <= 0) {
1703 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
1704 return repl;
1705 }
1706 }
1707 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
1708 }
1709 else {
1710 if (!(st->flags & SHTABLE_F_TEACH_STAGE1)) {
1711 repl = peer_send_teach_stage1_msgs(appctx, peer, st);
1712 if (repl <= 0)
1713 return repl;
1714 }
1715
1716 if (!(st->flags & SHTABLE_F_TEACH_STAGE2)) {
1717 repl = peer_send_teach_stage2_msgs(appctx, peer, st);
1718 if (repl <= 0)
1719 return repl;
1720 }
1721 }
1722
1723 if (st == last_local_table)
1724 break;
1725 st = st->next;
1726 }
1727 }
1728
1729 if ((peer->flags & PEER_F_TEACH_PROCESS) && !(peer->flags & PEER_F_TEACH_FINISHED)) {
1730 repl = peer_send_resync_finishedmsg(appctx, peer);
1731 if (repl <= 0)
1732 return repl;
1733
1734 /* flag finished message sent */
1735 peer->flags |= PEER_F_TEACH_FINISHED;
1736 }
1737
1738 /* Confirm finished or partial messages */
1739 while (peer->confirm) {
1740 repl = peer_send_resync_confirmsg(appctx);
1741 if (repl <= 0)
1742 return repl;
1743
1744 peer->confirm--;
1745 }
1746
1747 return 1;
1748}
1749
Frédéric Lécaille95203f22019-01-23 19:38:11 +01001750/*
Frédéric Lécaille3f0fb9d2019-01-25 08:30:29 +01001751 * Read and parse a first line of a "hello" peer protocol message.
1752 * Returns 0 if could not read a line, -1 if there was a read error or
1753 * the line is malformed, 1 if succeeded.
1754 */
1755static inline int peer_getline_version(struct appctx *appctx,
1756 unsigned int *maj_ver, unsigned int *min_ver)
1757{
1758 int reql;
1759
1760 reql = peer_getline(appctx);
1761 if (!reql)
1762 return 0;
1763
1764 if (reql < 0)
1765 return -1;
1766
1767 /* test protocol */
1768 if (strncmp(PEER_SESSION_PROTO_NAME " ", trash.area, proto_len + 1) != 0) {
1769 appctx->st0 = PEER_SESS_ST_EXIT;
1770 appctx->st1 = PEER_SESS_SC_ERRPROTO;
1771 return -1;
1772 }
1773 if (peer_get_version(trash.area + proto_len + 1, maj_ver, min_ver) == -1 ||
1774 *maj_ver != PEER_MAJOR_VER || *min_ver > PEER_MINOR_VER) {
1775 appctx->st0 = PEER_SESS_ST_EXIT;
1776 appctx->st1 = PEER_SESS_SC_ERRVERSION;
1777 return -1;
1778 }
1779
1780 return 1;
1781}
1782
1783/*
1784 * Read and parse a second line of a "hello" peer protocol message.
1785 * Returns 0 if could not read a line, -1 if there was a read error or
1786 * the line is malformed, 1 if succeeded.
1787 */
1788static inline int peer_getline_host(struct appctx *appctx)
1789{
1790 int reql;
1791
1792 reql = peer_getline(appctx);
1793 if (!reql)
1794 return 0;
1795
1796 if (reql < 0)
1797 return -1;
1798
1799 /* test hostname match */
1800 if (strcmp(localpeer, trash.area) != 0) {
1801 appctx->st0 = PEER_SESS_ST_EXIT;
1802 appctx->st1 = PEER_SESS_SC_ERRHOST;
1803 return -1;
1804 }
1805
1806 return 1;
1807}
1808
1809/*
1810 * Read and parse a last line of a "hello" peer protocol message.
1811 * Returns 0 if could not read a character, -1 if there was a read error or
1812 * the line is malformed, 1 if succeeded.
1813 * Set <curpeer> accordingly (the remote peer sending the "hello" message).
1814 */
1815static inline int peer_getline_last(struct appctx *appctx, struct peer **curpeer)
1816{
1817 char *p;
1818 int reql;
1819 struct peer *peer;
1820 struct stream_interface *si = appctx->owner;
1821 struct stream *s = si_strm(si);
1822 struct peers *peers = strm_fe(s)->parent;
1823
1824 reql = peer_getline(appctx);
1825 if (!reql)
1826 return 0;
1827
1828 if (reql < 0)
1829 return -1;
1830
1831 /* parse line "<peer name> <pid> <relative_pid>" */
1832 p = strchr(trash.area, ' ');
1833 if (!p) {
1834 appctx->st0 = PEER_SESS_ST_EXIT;
1835 appctx->st1 = PEER_SESS_SC_ERRPROTO;
1836 return -1;
1837 }
1838 *p = 0;
1839
1840 /* lookup known peer */
1841 for (peer = peers->remote; peer; peer = peer->next) {
1842 if (strcmp(peer->id, trash.area) == 0)
1843 break;
1844 }
1845
1846 /* if unknown peer */
1847 if (!peer) {
1848 appctx->st0 = PEER_SESS_ST_EXIT;
1849 appctx->st1 = PEER_SESS_SC_ERRPEER;
1850 return -1;
1851 }
1852 *curpeer = peer;
1853
1854 return 1;
1855}
1856
1857/*
Frédéric Lécaille4b2fd9b2019-01-25 08:58:41 +01001858 * Init <peer> peer after having accepted it at peer protocol level.
1859 */
1860static inline void init_accepted_peer(struct peer *peer, struct peers *peers)
1861{
1862 struct shared_table *st;
1863
1864 /* Register status code */
1865 peer->statuscode = PEER_SESS_SC_SUCCESSCODE;
1866
1867 /* Awake main task */
1868 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
1869
1870 /* Init confirm counter */
1871 peer->confirm = 0;
1872
1873 /* Init cursors */
1874 for (st = peer->tables; st ; st = st->next) {
1875 st->last_get = st->last_acked = 0;
1876 st->teaching_origin = st->last_pushed = st->update;
1877 }
1878
1879 /* reset teaching and learning flags to 0 */
1880 peer->flags &= PEER_TEACH_RESET;
1881 peer->flags &= PEER_LEARN_RESET;
1882
1883 /* if current peer is local */
1884 if (peer->local) {
1885 /* if current host need resyncfrom local and no process assined */
1886 if ((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL &&
1887 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
1888 /* assign local peer for a lesson, consider lesson already requested */
1889 peer->flags |= PEER_F_LEARN_ASSIGN;
1890 peers->flags |= (PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
1891 }
1892
1893 }
1894 else if ((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
1895 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
1896 /* assign peer for a lesson */
1897 peer->flags |= PEER_F_LEARN_ASSIGN;
1898 peers->flags |= PEERS_F_RESYNC_ASSIGN;
1899 }
1900}
1901
1902/*
1903 * Init <peer> peer after having connected it at peer protocol level.
1904 */
1905static inline void init_connected_peer(struct peer *peer, struct peers *peers)
1906{
1907 struct shared_table *st;
1908
1909 /* Init cursors */
1910 for (st = peer->tables; st ; st = st->next) {
1911 st->last_get = st->last_acked = 0;
1912 st->teaching_origin = st->last_pushed = st->update;
1913 }
1914
1915 /* Init confirm counter */
1916 peer->confirm = 0;
1917
1918 /* reset teaching and learning flags to 0 */
1919 peer->flags &= PEER_TEACH_RESET;
1920 peer->flags &= PEER_LEARN_RESET;
1921
1922 /* If current peer is local */
1923 if (peer->local) {
1924 /* flag to start to teach lesson */
1925 peer->flags |= PEER_F_TEACH_PROCESS;
1926 }
1927 else if ((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
1928 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
1929 /* If peer is remote and resync from remote is needed,
1930 and no peer currently assigned */
1931
1932 /* assign peer for a lesson */
1933 peer->flags |= PEER_F_LEARN_ASSIGN;
1934 peers->flags |= PEERS_F_RESYNC_ASSIGN;
1935 }
1936}
1937
1938/*
Emeric Brun2b920a12010-09-23 18:30:22 +02001939 * IO Handler to handle message exchance with a peer
1940 */
Willy Tarreau00a37f02015-04-13 12:05:19 +02001941static void peer_io_handler(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +02001942{
Willy Tarreau00a37f02015-04-13 12:05:19 +02001943 struct stream_interface *si = appctx->owner;
Willy Tarreau87b09662015-04-03 00:22:06 +02001944 struct stream *s = si_strm(si);
Vincent Bernat3c2f2f22016-04-03 13:48:42 +02001945 struct peers *curpeers = strm_fe(s)->parent;
Emeric Brun80527f52017-06-19 17:46:37 +02001946 struct peer *curpeer = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001947 int reql = 0;
1948 int repl = 0;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001949 unsigned int maj_ver, min_ver;
Willy Tarreau2d372c22018-11-05 17:12:27 +01001950 int prev_state;
Emeric Brun2b920a12010-09-23 18:30:22 +02001951
Joseph Herlant82b2f542018-11-15 12:19:14 -08001952 /* Check if the input buffer is available. */
Frédéric Lécaille7d0ceee2019-01-24 14:24:05 +01001953 if (si_ic(si)->buf.size == 0) {
1954 si_rx_room_blk(si);
1955 goto out;
1956 }
Christopher Fauleta73e59b2016-12-09 17:30:18 +01001957
Emeric Brun2b920a12010-09-23 18:30:22 +02001958 while (1) {
Willy Tarreau2d372c22018-11-05 17:12:27 +01001959 prev_state = appctx->st0;
Emeric Brun2b920a12010-09-23 18:30:22 +02001960switchstate:
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001961 maj_ver = min_ver = (unsigned int)-1;
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001962 switch(appctx->st0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001963 case PEER_SESS_ST_ACCEPT:
Willy Tarreau2d372c22018-11-05 17:12:27 +01001964 prev_state = appctx->st0;
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001965 appctx->ctx.peers.ptr = NULL;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001966 appctx->st0 = PEER_SESS_ST_GETVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +02001967 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001968 case PEER_SESS_ST_GETVERSION:
Willy Tarreau2d372c22018-11-05 17:12:27 +01001969 prev_state = appctx->st0;
Frédéric Lécaille3f0fb9d2019-01-25 08:30:29 +01001970 reql = peer_getline_version(appctx, &maj_ver, &min_ver);
1971 if (reql <= 0) {
1972 if (!reql)
1973 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001974 goto switchstate;
1975 }
1976
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001977 appctx->st0 = PEER_SESS_ST_GETHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +02001978 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001979 case PEER_SESS_ST_GETHOST:
Willy Tarreau2d372c22018-11-05 17:12:27 +01001980 prev_state = appctx->st0;
Frédéric Lécaille3f0fb9d2019-01-25 08:30:29 +01001981 reql = peer_getline_host(appctx);
1982 if (reql <= 0) {
1983 if (!reql)
1984 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001985 goto switchstate;
1986 }
1987
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001988 appctx->st0 = PEER_SESS_ST_GETPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +02001989 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001990 case PEER_SESS_ST_GETPEER: {
Willy Tarreau2d372c22018-11-05 17:12:27 +01001991 prev_state = appctx->st0;
Frédéric Lécaille3f0fb9d2019-01-25 08:30:29 +01001992 reql = peer_getline_last(appctx, &curpeer);
1993 if (reql <= 0) {
1994 if (!reql)
1995 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001996 goto switchstate;
1997 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001998
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001999 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Willy Tarreau9df94c22016-10-31 18:42:52 +01002000 if (curpeer->appctx && curpeer->appctx != appctx) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02002001 if (curpeer->local) {
2002 /* Local connection, reply a retry */
2003 appctx->st0 = PEER_SESS_ST_EXIT;
2004 appctx->st1 = PEER_SESS_SC_TRYAGAIN;
2005 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02002006 }
Emeric Brun80527f52017-06-19 17:46:37 +02002007
2008 /* we're killing a connection, we must apply a random delay before
2009 * retrying otherwise the other end will do the same and we can loop
2010 * for a while.
2011 */
2012 curpeer->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + random() % 2000));
Willy Tarreau81bc3b02016-10-31 17:37:39 +01002013 peer_session_forceshutdown(curpeer->appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02002014 }
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02002015 if (maj_ver != (unsigned int)-1 && min_ver != (unsigned int)-1) {
2016 if (min_ver == PEER_DWNGRD_MINOR_VER) {
2017 curpeer->flags |= PEER_F_DWNGRD;
2018 }
2019 else {
2020 curpeer->flags &= ~PEER_F_DWNGRD;
2021 }
2022 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02002023 curpeer->appctx = appctx;
2024 appctx->ctx.peers.ptr = curpeer;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002025 appctx->st0 = PEER_SESS_ST_SENDSUCCESS;
Willy Tarreau199ad242018-11-05 16:31:22 +01002026 HA_ATOMIC_ADD(&active_peers, 1);
Emeric Brun2b920a12010-09-23 18:30:22 +02002027 /* fall through */
2028 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002029 case PEER_SESS_ST_SENDSUCCESS: {
Willy Tarreau2d372c22018-11-05 17:12:27 +01002030 prev_state = appctx->st0;
Emeric Brun80527f52017-06-19 17:46:37 +02002031 if (!curpeer) {
2032 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01002033 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02002034 if (curpeer->appctx != appctx) {
2035 appctx->st0 = PEER_SESS_ST_END;
2036 goto switchstate;
2037 }
2038 }
Frédéric Lécaille7d0ceee2019-01-24 14:24:05 +01002039
2040 repl = peer_send_status_successmsg(appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02002041 if (repl <= 0) {
2042 if (repl == -1)
Frédéric Lécaille7d0ceee2019-01-24 14:24:05 +01002043 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02002044 goto switchstate;
2045 }
2046
Frédéric Lécaille4b2fd9b2019-01-25 08:58:41 +01002047 init_accepted_peer(curpeer, curpeers);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002048
Emeric Brun2b920a12010-09-23 18:30:22 +02002049 /* switch to waiting message state */
Willy Tarreau2d372c22018-11-05 17:12:27 +01002050 HA_ATOMIC_ADD(&connected_peers, 1);
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002051 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +02002052 goto switchstate;
2053 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002054 case PEER_SESS_ST_CONNECT: {
Willy Tarreau2d372c22018-11-05 17:12:27 +01002055 prev_state = appctx->st0;
Emeric Brun80527f52017-06-19 17:46:37 +02002056 if (!curpeer) {
2057 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01002058 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02002059 if (curpeer->appctx != appctx) {
2060 appctx->st0 = PEER_SESS_ST_END;
2061 goto switchstate;
2062 }
2063 }
Emeric Brun2b920a12010-09-23 18:30:22 +02002064
Frédéric Lécaille7d0ceee2019-01-24 14:24:05 +01002065 repl = peer_send_hellomsg(appctx, curpeer);
Emeric Brun2b920a12010-09-23 18:30:22 +02002066 if (repl <= 0) {
2067 if (repl == -1)
Frédéric Lécaille7d0ceee2019-01-24 14:24:05 +01002068 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02002069 goto switchstate;
2070 }
2071
2072 /* switch to the waiting statuscode state */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002073 appctx->st0 = PEER_SESS_ST_GETSTATUS;
Emeric Brun2b920a12010-09-23 18:30:22 +02002074 /* fall through */
2075 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002076 case PEER_SESS_ST_GETSTATUS: {
Willy Tarreau2d372c22018-11-05 17:12:27 +01002077 prev_state = appctx->st0;
Emeric Brun80527f52017-06-19 17:46:37 +02002078 if (!curpeer) {
2079 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01002080 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02002081 if (curpeer->appctx != appctx) {
2082 appctx->st0 = PEER_SESS_ST_END;
2083 goto switchstate;
2084 }
2085 }
2086
Willy Tarreau2bb4a962014-11-28 11:11:05 +01002087 if (si_ic(si)->flags & CF_WRITE_PARTIAL)
Emeric Brunb3971ab2015-05-12 18:49:09 +02002088 curpeer->statuscode = PEER_SESS_SC_CONNECTEDCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +02002089
Frédéric Lécaillece025572019-01-21 13:38:06 +01002090 reql = peer_getline(appctx);
2091 if (!reql)
2092 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02002093
Frédéric Lécaillece025572019-01-21 13:38:06 +01002094 if (reql < 0)
2095 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02002096
2097 /* Register status code */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02002098 curpeer->statuscode = atoi(trash.area);
Emeric Brun2b920a12010-09-23 18:30:22 +02002099
2100 /* Awake main task */
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02002101 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +02002102
2103 /* If status code is success */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002104 if (curpeer->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Frédéric Lécaille4b2fd9b2019-01-25 08:58:41 +01002105 init_connected_peer(curpeer, curpeers);
Emeric Brun2b920a12010-09-23 18:30:22 +02002106 }
2107 else {
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02002108 if (curpeer->statuscode == PEER_SESS_SC_ERRVERSION)
2109 curpeer->flags |= PEER_F_DWNGRD;
Emeric Brun2b920a12010-09-23 18:30:22 +02002110 /* Status code is not success, abort */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002111 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02002112 goto switchstate;
2113 }
Willy Tarreau2d372c22018-11-05 17:12:27 +01002114 HA_ATOMIC_ADD(&connected_peers, 1);
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002115 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +02002116 /* fall through */
2117 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002118 case PEER_SESS_ST_WAITMSG: {
Emeric Brunb3971ab2015-05-12 18:49:09 +02002119 uint32_t msg_len = 0;
Willy Tarreau843b7cb2018-07-13 10:54:26 +02002120 char *msg_cur = trash.area;
2121 char *msg_end = trash.area;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002122 unsigned char msg_head[7];
Emeric Brun2b920a12010-09-23 18:30:22 +02002123 int totl = 0;
2124
Willy Tarreau2d372c22018-11-05 17:12:27 +01002125 prev_state = appctx->st0;
Emeric Brun80527f52017-06-19 17:46:37 +02002126 if (!curpeer) {
2127 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01002128 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02002129 if (curpeer->appctx != appctx) {
2130 appctx->st0 = PEER_SESS_ST_END;
2131 goto switchstate;
2132 }
2133 }
2134
Frédéric Lécaille95203f22019-01-23 19:38:11 +01002135 reql = peer_recv_msg(appctx, (char *)msg_head, sizeof msg_head, &msg_len, &totl);
2136 if (reql <= 0) {
2137 if (reql == -1)
2138 goto switchstate;
Frédéric Lécaillebe825e52019-01-24 18:28:44 +01002139 goto send_msgs;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002140 }
Willy Tarreau86a446e2013-11-25 23:02:37 +01002141
Frédéric Lécaille95203f22019-01-23 19:38:11 +01002142 msg_end += msg_len;
Frédéric Lécaille444243c2019-01-24 15:40:11 +01002143 if (!peer_treat_awaited_msg(appctx, curpeer, msg_head, &msg_cur, msg_end, msg_len, totl))
Emeric Brun2b920a12010-09-23 18:30:22 +02002144 goto switchstate;
Frédéric Lécaillebe825e52019-01-24 18:28:44 +01002145
Emeric Brun2b920a12010-09-23 18:30:22 +02002146 /* skip consumed message */
Willy Tarreau06d80a92017-10-19 14:32:15 +02002147 co_skip(si_oc(si), totl);
Emeric Brun2b920a12010-09-23 18:30:22 +02002148 /* loop on that state to peek next message */
Willy Tarreau72d6c162013-04-11 16:14:13 +02002149 goto switchstate;
2150
Frédéric Lécaillebe825e52019-01-24 18:28:44 +01002151send_msgs:
2152 /* we get here when a peer_recv_msg() returns 0 in reql */
Frédéric Lécaille25e1d5e2019-01-24 17:33:48 +01002153 repl = peer_send_msgs(appctx, curpeer);
2154 if (repl <= 0) {
2155 if (repl == -1)
2156 goto out;
2157 goto switchstate;
Emeric Brun597b26e2016-08-12 11:23:31 +02002158 }
2159
Emeric Brun2b920a12010-09-23 18:30:22 +02002160 /* noting more to do */
2161 goto out;
2162 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002163 case PEER_SESS_ST_EXIT:
Willy Tarreau2d372c22018-11-05 17:12:27 +01002164 if (prev_state == PEER_SESS_ST_WAITMSG)
2165 HA_ATOMIC_SUB(&connected_peers, 1);
2166 prev_state = appctx->st0;
Frédéric Lécaille7d0ceee2019-01-24 14:24:05 +01002167 if (peer_send_status_errormsg(appctx) == -1)
2168 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002169 appctx->st0 = PEER_SESS_ST_END;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002170 goto switchstate;
2171 case PEER_SESS_ST_ERRSIZE: {
Willy Tarreau2d372c22018-11-05 17:12:27 +01002172 if (prev_state == PEER_SESS_ST_WAITMSG)
2173 HA_ATOMIC_SUB(&connected_peers, 1);
2174 prev_state = appctx->st0;
Frédéric Lécaille7d0ceee2019-01-24 14:24:05 +01002175 if (peer_send_error_size_limitmsg(appctx) == -1)
2176 goto out;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002177 appctx->st0 = PEER_SESS_ST_END;
2178 goto switchstate;
2179 }
2180 case PEER_SESS_ST_ERRPROTO: {
Willy Tarreau2d372c22018-11-05 17:12:27 +01002181 if (prev_state == PEER_SESS_ST_WAITMSG)
2182 HA_ATOMIC_SUB(&connected_peers, 1);
2183 prev_state = appctx->st0;
Frédéric Lécaille7d0ceee2019-01-24 14:24:05 +01002184 if (peer_send_error_protomsg(appctx) == -1)
2185 goto out;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002186 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau2d372c22018-11-05 17:12:27 +01002187 prev_state = appctx->st0;
Emeric Brun2b920a12010-09-23 18:30:22 +02002188 /* fall through */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002189 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002190 case PEER_SESS_ST_END: {
Willy Tarreau2d372c22018-11-05 17:12:27 +01002191 if (prev_state == PEER_SESS_ST_WAITMSG)
2192 HA_ATOMIC_SUB(&connected_peers, 1);
2193 prev_state = appctx->st0;
Emeric Brun80527f52017-06-19 17:46:37 +02002194 if (curpeer) {
Christopher Faulet2a944ee2017-11-07 10:42:54 +01002195 HA_SPIN_UNLOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02002196 curpeer = NULL;
2197 }
Willy Tarreau73b013b2012-05-21 16:31:45 +02002198 si_shutw(si);
2199 si_shutr(si);
Willy Tarreau2bb4a962014-11-28 11:11:05 +01002200 si_ic(si)->flags |= CF_READ_NULL;
Willy Tarreau828824a2015-04-19 17:20:03 +02002201 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02002202 }
2203 }
2204 }
2205out:
Willy Tarreau2bb4a962014-11-28 11:11:05 +01002206 si_oc(si)->flags |= CF_READ_DONTWAIT;
Emeric Brun80527f52017-06-19 17:46:37 +02002207
2208 if (curpeer)
Christopher Faulet2a944ee2017-11-07 10:42:54 +01002209 HA_SPIN_UNLOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun2b920a12010-09-23 18:30:22 +02002210 return;
2211}
2212
Willy Tarreau30576452015-04-13 13:50:30 +02002213static struct applet peer_applet = {
Willy Tarreau3fdb3662012-11-12 00:42:33 +01002214 .obj_type = OBJ_TYPE_APPLET,
Willy Tarreaub24281b2011-02-13 13:16:36 +01002215 .name = "<PEER>", /* used for logging */
2216 .fct = peer_io_handler,
Aman Gupta9a13e842012-04-02 18:57:53 -07002217 .release = peer_session_release,
Willy Tarreaub24281b2011-02-13 13:16:36 +01002218};
Emeric Brun2b920a12010-09-23 18:30:22 +02002219
2220/*
2221 * Use this function to force a close of a peer session
2222 */
Willy Tarreau81bc3b02016-10-31 17:37:39 +01002223static void peer_session_forceshutdown(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +02002224{
Frédéric Lécaille5df11902017-06-13 16:39:57 +02002225 /* Note that the peer sessions which have just been created
2226 * (->st0 == PEER_SESS_ST_CONNECT) must not
2227 * be shutdown, if not, the TCP session will never be closed
2228 * and stay in CLOSE_WAIT state after having been closed by
2229 * the remote side.
2230 */
2231 if (!appctx || appctx->st0 == PEER_SESS_ST_CONNECT)
Willy Tarreau7b4b4992013-12-01 09:15:12 +01002232 return;
2233
Willy Tarreau81bc3b02016-10-31 17:37:39 +01002234 if (appctx->applet != &peer_applet)
2235 return;
2236
Willy Tarreau2d372c22018-11-05 17:12:27 +01002237 if (appctx->st0 == PEER_SESS_ST_WAITMSG)
2238 HA_ATOMIC_SUB(&connected_peers, 1);
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002239 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau78c0c502016-10-31 17:32:20 +01002240 appctx_wakeup(appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02002241}
2242
Willy Tarreau91d96282015-03-13 15:47:26 +01002243/* Pre-configures a peers frontend to accept incoming connections */
2244void peers_setup_frontend(struct proxy *fe)
2245{
2246 fe->last_change = now.tv_sec;
Frédéric Lécaillec06b5d42018-04-26 10:06:41 +02002247 fe->cap = PR_CAP_FE | PR_CAP_BE;
Willy Tarreau91d96282015-03-13 15:47:26 +01002248 fe->maxconn = 0;
2249 fe->conn_retries = CONN_RETRIES;
2250 fe->timeout.client = MS_TO_TICKS(5000);
Willy Tarreaud1d48d42015-03-13 16:15:46 +01002251 fe->accept = frontend_accept;
Willy Tarreauf87ab942015-03-13 15:55:16 +01002252 fe->default_target = &peer_applet.obj_type;
Willy Tarreau91d96282015-03-13 15:47:26 +01002253 fe->options2 |= PR_O2_INDEPSTR | PR_O2_SMARTCON | PR_O2_SMARTACC;
Willy Tarreau0fca4832015-05-01 19:12:05 +02002254 fe->bind_proc = 0; /* will be filled by users */
Willy Tarreau91d96282015-03-13 15:47:26 +01002255}
2256
Emeric Brun2b920a12010-09-23 18:30:22 +02002257/*
Willy Tarreaubd55e312010-11-11 10:55:09 +01002258 * Create a new peer session in assigned state (connect will start automatically)
Emeric Brun2b920a12010-09-23 18:30:22 +02002259 */
Willy Tarreau9df94c22016-10-31 18:42:52 +01002260static struct appctx *peer_session_create(struct peers *peers, struct peer *peer)
Emeric Brun2b920a12010-09-23 18:30:22 +02002261{
Willy Tarreau04b92862017-09-15 11:01:04 +02002262 struct proxy *p = peers->peers_fe; /* attached frontend */
Willy Tarreau7b4b4992013-12-01 09:15:12 +01002263 struct appctx *appctx;
Willy Tarreau15b5e142015-04-04 14:38:25 +02002264 struct session *sess;
Willy Tarreau87b09662015-04-03 00:22:06 +02002265 struct stream *s;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02002266 struct connection *conn;
Olivier Houchard9aaf7782017-09-13 18:30:23 +02002267 struct conn_stream *cs;
Emeric Brun2b920a12010-09-23 18:30:22 +02002268
Emeric Brunb3971ab2015-05-12 18:49:09 +02002269 peer->reconnect = tick_add(now_ms, MS_TO_TICKS(5000));
2270 peer->statuscode = PEER_SESS_SC_CONNECTCODE;
Willy Tarreaud990baf2015-04-05 00:32:03 +02002271 s = NULL;
2272
Emeric Brun1138fd02017-06-19 12:38:55 +02002273 appctx = appctx_new(&peer_applet, tid_bit);
Willy Tarreaud990baf2015-04-05 00:32:03 +02002274 if (!appctx)
2275 goto out_close;
2276
2277 appctx->st0 = PEER_SESS_ST_CONNECT;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002278 appctx->ctx.peers.ptr = (void *)peer;
Willy Tarreaud990baf2015-04-05 00:32:03 +02002279
Willy Tarreau04b92862017-09-15 11:01:04 +02002280 sess = session_new(p, NULL, &appctx->obj_type);
Willy Tarreau15b5e142015-04-04 14:38:25 +02002281 if (!sess) {
Christopher Faulet767a84b2017-11-24 16:50:31 +01002282 ha_alert("out of memory in peer_session_create().\n");
Willy Tarreaud990baf2015-04-05 00:32:03 +02002283 goto out_free_appctx;
Emeric Brun2b920a12010-09-23 18:30:22 +02002284 }
2285
Willy Tarreau87787ac2017-08-28 16:22:54 +02002286 if ((s = stream_new(sess, &appctx->obj_type)) == NULL) {
Christopher Faulet767a84b2017-11-24 16:50:31 +01002287 ha_alert("Failed to initialize stream in peer_session_create().\n");
Willy Tarreau87787ac2017-08-28 16:22:54 +02002288 goto out_free_sess;
Willy Tarreau8baf9062015-04-05 00:46:36 +02002289 }
2290
Willy Tarreau342bfb12015-04-05 01:35:34 +02002291 /* The tasks below are normally what is supposed to be done by
2292 * fe->accept().
2293 */
Willy Tarreaue7dff022015-04-03 01:14:29 +02002294 s->flags = SF_ASSIGNED|SF_ADDR_SET;
Emeric Brun2b920a12010-09-23 18:30:22 +02002295
Willy Tarreau6e2979c2015-04-27 13:21:15 +02002296 /* applet is waiting for data */
Willy Tarreau0cd3bd62018-11-06 18:46:37 +01002297 si_cant_get(&s->si[0]);
Willy Tarreau6e2979c2015-04-27 13:21:15 +02002298 appctx_wakeup(appctx);
2299
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02002300 /* initiate an outgoing connection */
Willy Tarreaudbd02672017-12-06 17:39:53 +01002301 s->si[1].flags |= SI_FL_NOLINGER;
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02002302 si_set_state(&s->si[1], SI_ST_ASS);
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02002303
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02002304 /* automatically prepare the stream interface to connect to the
Willy Tarreaub363a1f2013-10-01 10:45:07 +02002305 * pre-initialized connection in si->conn.
2306 */
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02002307 if (unlikely((conn = conn_new()) == NULL))
Willy Tarreau8baf9062015-04-05 00:46:36 +02002308 goto out_free_strm;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02002309
Olivier Houchard9aaf7782017-09-13 18:30:23 +02002310 if (unlikely((cs = cs_new(conn)) == NULL))
2311 goto out_free_conn;
2312
Frédéric Lécaille1055e682018-04-26 14:35:21 +02002313 conn->target = s->target = peer_session_target(peer, s);
Willy Tarreaube373152018-09-06 11:45:30 +02002314 memcpy(&conn->addr.to, &peer->addr, sizeof(conn->addr.to));
2315
Frédéric Lécaille1055e682018-04-26 14:35:21 +02002316 conn_prepare(conn, peer->proto, peer_xprt(peer));
Olivier Houchardef60ff32019-01-29 19:00:33 +01002317 if (conn_install_mux(conn, &mux_pt_ops, cs, s->be, NULL) < 0)
2318 goto out_free_cs;
Olivier Houchard9aaf7782017-09-13 18:30:23 +02002319 si_attach_cs(&s->si[1], cs);
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02002320
Emeric Brun2b920a12010-09-23 18:30:22 +02002321 s->do_log = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02002322 s->uniq_id = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +02002323
Willy Tarreau22ec1ea2014-11-27 20:45:39 +01002324 s->res.flags |= CF_READ_DONTWAIT;
Willy Tarreau696a2912014-11-24 11:36:57 +01002325
Emeric Brunb3971ab2015-05-12 18:49:09 +02002326 peer->appctx = appctx;
Willy Tarreau87787ac2017-08-28 16:22:54 +02002327 task_wakeup(s->task, TASK_WOKEN_INIT);
Willy Tarreau199ad242018-11-05 16:31:22 +01002328 HA_ATOMIC_ADD(&active_peers, 1);
Willy Tarreau9df94c22016-10-31 18:42:52 +01002329 return appctx;
Emeric Brun2b920a12010-09-23 18:30:22 +02002330
2331 /* Error unrolling */
Olivier Houchardef60ff32019-01-29 19:00:33 +01002332out_free_cs:
2333 cs_free(cs);
Olivier Houchard9aaf7782017-09-13 18:30:23 +02002334 out_free_conn:
2335 conn_free(conn);
Willy Tarreau15b5e142015-04-04 14:38:25 +02002336 out_free_strm:
Emeric Brun2b920a12010-09-23 18:30:22 +02002337 LIST_DEL(&s->list);
Willy Tarreaubafbe012017-11-24 17:34:44 +01002338 pool_free(pool_head_stream, s);
Willy Tarreau15b5e142015-04-04 14:38:25 +02002339 out_free_sess:
Willy Tarreau11c36242015-04-04 15:54:03 +02002340 session_free(sess);
Willy Tarreaud990baf2015-04-05 00:32:03 +02002341 out_free_appctx:
2342 appctx_free(appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02002343 out_close:
Willy Tarreaub21d08e2016-10-31 17:46:57 +01002344 return NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02002345}
2346
2347/*
2348 * Task processing function to manage re-connect and peer session
2349 * tasks wakeup on local update.
2350 */
Olivier Houchard9f6af332018-05-25 14:04:04 +02002351static struct task *process_peer_sync(struct task * task, void *context, unsigned short state)
Emeric Brun2b920a12010-09-23 18:30:22 +02002352{
Olivier Houchard9f6af332018-05-25 14:04:04 +02002353 struct peers *peers = context;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002354 struct peer *ps;
2355 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +02002356
2357 task->expire = TICK_ETERNITY;
2358
Emeric Brunb3971ab2015-05-12 18:49:09 +02002359 if (!peers->peers_fe) {
Willy Tarreau46dc1ca2015-05-01 18:32:13 +02002360 /* this one was never started, kill it */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002361 signal_unregister_handler(peers->sighandler);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002362 task_delete(peers->sync_task);
2363 task_free(peers->sync_task);
Willy Tarreau37bb7be2015-09-21 15:24:58 +02002364 peers->sync_task = NULL;
Willy Tarreau46dc1ca2015-05-01 18:32:13 +02002365 return NULL;
2366 }
2367
Emeric Brun80527f52017-06-19 17:46:37 +02002368 /* Acquire lock for all peers of the section */
2369 for (ps = peers->remote; ps; ps = ps->next)
Christopher Faulet2a944ee2017-11-07 10:42:54 +01002370 HA_SPIN_LOCK(PEER_LOCK, &ps->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02002371
Emeric Brun2b920a12010-09-23 18:30:22 +02002372 if (!stopping) {
2373 /* Normal case (not soft stop)*/
Emeric Brunb3971ab2015-05-12 18:49:09 +02002374
2375 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL) &&
2376 (!nb_oldpids || tick_is_expired(peers->resync_timeout, now_ms)) &&
2377 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002378 /* Resync from local peer needed
2379 no peer was assigned for the lesson
2380 and no old local peer found
2381 or resync timeout expire */
2382
2383 /* flag no more resync from local, to try resync from remotes */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002384 peers->flags |= PEERS_F_RESYNC_LOCAL;
Emeric Brun2b920a12010-09-23 18:30:22 +02002385
2386 /* reschedule a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002387 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
Emeric Brun2b920a12010-09-23 18:30:22 +02002388 }
2389
2390 /* For each session */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002391 for (ps = peers->remote; ps; ps = ps->next) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002392 /* For each remote peers */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002393 if (!ps->local) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002394 if (!ps->appctx) {
2395 /* no active peer connection */
Emeric Brun2b920a12010-09-23 18:30:22 +02002396 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002397 ((ps->statuscode == PEER_SESS_SC_CONNECTCODE ||
Willy Tarreaub4e34da2015-05-20 10:39:04 +02002398 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002399 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02002400 tick_is_expired(ps->reconnect, now_ms))) {
2401 /* connection never tried
Willy Tarreau9df94c22016-10-31 18:42:52 +01002402 * or previous peer connection established with success
2403 * or previous peer connection failed while connecting
Emeric Brun2b920a12010-09-23 18:30:22 +02002404 * and reconnection timer is expired */
2405
2406 /* retry a connect */
Willy Tarreau9df94c22016-10-31 18:42:52 +01002407 ps->appctx = peer_session_create(peers, ps);
Emeric Brun2b920a12010-09-23 18:30:22 +02002408 }
Willy Tarreaub4e34da2015-05-20 10:39:04 +02002409 else if (!tick_is_expired(ps->reconnect, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002410 /* If previous session failed during connection
2411 * but reconnection timer is not expired */
2412
2413 /* reschedule task for reconnect */
2414 task->expire = tick_first(task->expire, ps->reconnect);
2415 }
2416 /* else do nothing */
Willy Tarreau9df94c22016-10-31 18:42:52 +01002417 } /* !ps->appctx */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002418 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002419 /* current peer connection is active and established */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002420 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
2421 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02002422 !(ps->flags & PEER_F_LEARN_NOTUP2DATE)) {
2423 /* Resync from a remote is needed
2424 * and no peer was assigned for lesson
2425 * and current peer may be up2date */
2426
2427 /* assign peer for the lesson */
2428 ps->flags |= PEER_F_LEARN_ASSIGN;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002429 peers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brun2b920a12010-09-23 18:30:22 +02002430
Willy Tarreau9df94c22016-10-31 18:42:52 +01002431 /* wake up peer handler to handle a request of resync */
Willy Tarreaue5843b32015-04-27 18:40:14 +02002432 appctx_wakeup(ps->appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02002433 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02002434 else {
2435 /* Awake session if there is data to push */
2436 for (st = ps->tables; st ; st = st->next) {
2437 if ((int)(st->last_pushed - st->table->localupdate) < 0) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002438 /* wake up the peer handler to push local updates */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002439 appctx_wakeup(ps->appctx);
2440 break;
2441 }
2442 }
Emeric Brun2b920a12010-09-23 18:30:22 +02002443 }
2444 /* else do nothing */
2445 } /* SUCCESSCODE */
2446 } /* !ps->peer->local */
2447 } /* for */
2448
2449 /* Resync from remotes expired: consider resync is finished */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002450 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
2451 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
2452 tick_is_expired(peers->resync_timeout, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002453 /* Resync from remote peer needed
2454 * no peer was assigned for the lesson
2455 * and resync timeout expire */
2456
2457 /* flag no more resync from remote, consider resync is finished */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002458 peers->flags |= PEERS_F_RESYNC_REMOTE;
Emeric Brun2b920a12010-09-23 18:30:22 +02002459 }
2460
Emeric Brunb3971ab2015-05-12 18:49:09 +02002461 if ((peers->flags & PEERS_RESYNC_STATEMASK) != PEERS_RESYNC_FINISHED) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002462 /* Resync not finished*/
Frédéric Lécaille5d6e5f82017-05-29 13:47:16 +02002463 /* reschedule task to resync timeout if not expired, to ended resync if needed */
2464 if (!tick_is_expired(peers->resync_timeout, now_ms))
2465 task->expire = tick_first(task->expire, peers->resync_timeout);
Emeric Brun2b920a12010-09-23 18:30:22 +02002466 }
2467 } /* !stopping */
2468 else {
2469 /* soft stop case */
Willy Tarreau086735a2018-11-05 15:09:47 +01002470 if (state & TASK_WOKEN_SIGNAL) {
Joseph Herlant82b2f542018-11-15 12:19:14 -08002471 /* We've just received the signal */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002472 if (!(peers->flags & PEERS_F_DONOTSTOP)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002473 /* add DO NOT STOP flag if not present */
Christopher Faulet8d8aa0d2017-05-30 15:36:50 +02002474 HA_ATOMIC_ADD(&jobs, 1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002475 peers->flags |= PEERS_F_DONOTSTOP;
2476 ps = peers->local;
2477 for (st = ps->tables; st ; st = st->next)
2478 st->table->syncing++;
Emeric Brun2b920a12010-09-23 18:30:22 +02002479 }
2480
2481 /* disconnect all connected peers */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002482 for (ps = peers->remote; ps; ps = ps->next) {
Emeric Brun80527f52017-06-19 17:46:37 +02002483 /* we're killing a connection, we must apply a random delay before
2484 * retrying otherwise the other end will do the same and we can loop
2485 * for a while.
2486 */
2487 ps->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + random() % 2000));
Willy Tarreau9df94c22016-10-31 18:42:52 +01002488 if (ps->appctx) {
Willy Tarreau81bc3b02016-10-31 17:37:39 +01002489 peer_session_forceshutdown(ps->appctx);
Willy Tarreaue5843b32015-04-27 18:40:14 +02002490 ps->appctx = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02002491 }
2492 }
2493 }
Emeric Brun2b920a12010-09-23 18:30:22 +02002494
Emeric Brunb3971ab2015-05-12 18:49:09 +02002495 ps = peers->local;
Emeric Brun2b920a12010-09-23 18:30:22 +02002496 if (ps->flags & PEER_F_TEACH_COMPLETE) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02002497 if (peers->flags & PEERS_F_DONOTSTOP) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002498 /* resync of new process was complete, current process can die now */
Willy Tarreaucea85372017-11-29 14:49:30 +01002499 HA_ATOMIC_SUB(&jobs, 1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002500 peers->flags &= ~PEERS_F_DONOTSTOP;
2501 for (st = ps->tables; st ; st = st->next)
2502 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02002503 }
2504 }
Willy Tarreau9df94c22016-10-31 18:42:52 +01002505 else if (!ps->appctx) {
2506 /* If there's no active peer connection */
Emeric Brun2b920a12010-09-23 18:30:22 +02002507 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002508 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
2509 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE ||
2510 ps->statuscode == PEER_SESS_SC_TRYAGAIN) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002511 /* connection never tried
Willy Tarreau9df94c22016-10-31 18:42:52 +01002512 * or previous peer connection was successfully established
2513 * or previous tcp connect succeeded but init state incomplete
Emeric Brun2b920a12010-09-23 18:30:22 +02002514 * or during previous connect, peer replies a try again statuscode */
2515
2516 /* connect to the peer */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002517 peer_session_create(peers, ps);
Emeric Brun2b920a12010-09-23 18:30:22 +02002518 }
2519 else {
2520 /* Other error cases */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002521 if (peers->flags & PEERS_F_DONOTSTOP) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002522 /* unable to resync new process, current process can die now */
Christopher Faulet8d8aa0d2017-05-30 15:36:50 +02002523 HA_ATOMIC_SUB(&jobs, 1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002524 peers->flags &= ~PEERS_F_DONOTSTOP;
2525 for (st = ps->tables; st ; st = st->next)
2526 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02002527 }
2528 }
2529 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02002530 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE ) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002531 /* current peer connection is active and established
2532 * wake up all peer handlers to push remaining local updates */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002533 for (st = ps->tables; st ; st = st->next) {
2534 if ((int)(st->last_pushed - st->table->localupdate) < 0) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02002535 appctx_wakeup(ps->appctx);
2536 break;
2537 }
2538 }
Emeric Brun2b920a12010-09-23 18:30:22 +02002539 }
2540 } /* stopping */
Emeric Brun80527f52017-06-19 17:46:37 +02002541
2542 /* Release lock for all peers of the section */
2543 for (ps = peers->remote; ps; ps = ps->next)
Christopher Faulet2a944ee2017-11-07 10:42:54 +01002544 HA_SPIN_UNLOCK(PEER_LOCK, &ps->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02002545
Emeric Brun2b920a12010-09-23 18:30:22 +02002546 /* Wakeup for re-connect */
2547 return task;
2548}
2549
Emeric Brunb3971ab2015-05-12 18:49:09 +02002550
Emeric Brun2b920a12010-09-23 18:30:22 +02002551/*
Willy Tarreaud9443442018-10-15 11:18:03 +02002552 * returns 0 in case of error.
Emeric Brun2b920a12010-09-23 18:30:22 +02002553 */
Willy Tarreaud9443442018-10-15 11:18:03 +02002554int peers_init_sync(struct peers *peers)
Emeric Brun2b920a12010-09-23 18:30:22 +02002555{
Emeric Brun2b920a12010-09-23 18:30:22 +02002556 struct peer * curpeer;
Willy Tarreau4348fad2012-09-20 16:48:07 +02002557 struct listener *listener;
Emeric Brun2b920a12010-09-23 18:30:22 +02002558
Emeric Brun2b920a12010-09-23 18:30:22 +02002559 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002560 peers->peers_fe->maxconn += 3;
2561 }
2562
Willy Tarreau4348fad2012-09-20 16:48:07 +02002563 list_for_each_entry(listener, &peers->peers_fe->conf.listeners, by_fe)
2564 listener->maxconn = peers->peers_fe->maxconn;
Emeric Brunc60def82017-09-27 14:59:38 +02002565 peers->sync_task = task_new(MAX_THREADS_MASK);
Willy Tarreaud9443442018-10-15 11:18:03 +02002566 if (!peers->sync_task)
2567 return 0;
2568
Emeric Brunb3971ab2015-05-12 18:49:09 +02002569 peers->sync_task->process = process_peer_sync;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002570 peers->sync_task->context = (void *)peers;
2571 peers->sighandler = signal_register_task(0, peers->sync_task, 0);
2572 task_wakeup(peers->sync_task, TASK_WOKEN_INIT);
Willy Tarreaud9443442018-10-15 11:18:03 +02002573 return 1;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002574}
2575
2576
2577
2578/*
2579 * Function used to register a table for sync on a group of peers
2580 *
2581 */
2582void peers_register_table(struct peers *peers, struct stktable *table)
2583{
2584 struct shared_table *st;
2585 struct peer * curpeer;
2586 int id = 0;
2587
2588 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
Vincent Bernat02779b62016-04-03 13:48:43 +02002589 st = calloc(1,sizeof(*st));
Emeric Brunb3971ab2015-05-12 18:49:09 +02002590 st->table = table;
2591 st->next = curpeer->tables;
2592 if (curpeer->tables)
2593 id = curpeer->tables->local_id;
2594 st->local_id = id + 1;
2595
2596 curpeer->tables = st;
2597 }
2598
2599 table->sync_task = peers->sync_task;
Emeric Brun2b920a12010-09-23 18:30:22 +02002600}
2601