blob: 649af83ffe1a055a55d651269e0606c01e8b1a7e [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>
Emeric Brun2b920a12010-09-23 18:30:22 +020042#include <proto/proxy.h>
Willy Tarreaufeb76402015-04-03 14:10:06 +020043#include <proto/session.h>
Willy Tarreau87b09662015-04-03 00:22:06 +020044#include <proto/stream.h>
Willy Tarreau22ec1ea2014-11-27 20:45:39 +010045#include <proto/signal.h>
46#include <proto/stick_table.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020047#include <proto/stream_interface.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020048#include <proto/task.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020049
50
51/*******************************/
52/* Current peer learning state */
53/*******************************/
54
55/******************************/
Emeric Brunb3971ab2015-05-12 18:49:09 +020056/* Current peers section resync state */
Emeric Brun2b920a12010-09-23 18:30:22 +020057/******************************/
Emeric Brunb3971ab2015-05-12 18:49:09 +020058#define PEERS_F_RESYNC_LOCAL 0x00000001 /* Learn from local finished or no more needed */
59#define PEERS_F_RESYNC_REMOTE 0x00000002 /* Learn from remote finished or no more needed */
60#define PEERS_F_RESYNC_ASSIGN 0x00000004 /* A peer was assigned to learn our lesson */
61#define PEERS_F_RESYNC_PROCESS 0x00000008 /* The assigned peer was requested for resync */
62#define PEERS_F_DONOTSTOP 0x00010000 /* Main table sync task block process during soft stop
Emeric Brun2b920a12010-09-23 18:30:22 +020063 to push data to new process */
64
Emeric Brunb3971ab2015-05-12 18:49:09 +020065#define PEERS_RESYNC_STATEMASK (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE)
66#define PEERS_RESYNC_FROMLOCAL 0x00000000
67#define PEERS_RESYNC_FROMREMOTE PEERS_F_RESYNC_LOCAL
68#define PEERS_RESYNC_FINISHED (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE)
69
70/***********************************/
71/* Current shared table sync state */
72/***********************************/
73#define SHTABLE_F_TEACH_STAGE1 0x00000001 /* Teach state 1 complete */
74#define SHTABLE_F_TEACH_STAGE2 0x00000002 /* Teach state 2 complete */
Emeric Brun2b920a12010-09-23 18:30:22 +020075
76/******************************/
77/* Remote peer teaching state */
78/******************************/
79#define PEER_F_TEACH_PROCESS 0x00000001 /* Teach a lesson to current peer */
Emeric Brun2b920a12010-09-23 18:30:22 +020080#define PEER_F_TEACH_FINISHED 0x00000008 /* Teach conclude, (wait for confirm) */
81#define PEER_F_TEACH_COMPLETE 0x00000010 /* All that we know already taught to current peer, used only for a local peer */
82#define PEER_F_LEARN_ASSIGN 0x00000100 /* Current peer was assigned for a lesson */
83#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 +020084#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 +020085
Emeric Brunb3971ab2015-05-12 18:49:09 +020086#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 +020087#define PEER_LEARN_RESET ~(PEER_F_LEARN_ASSIGN|PEER_F_LEARN_NOTUP2DATE)
88
Emeric Brunb3971ab2015-05-12 18:49:09 +020089/*****************************/
90/* Sync message class */
91/*****************************/
92enum {
93 PEER_MSG_CLASS_CONTROL = 0,
94 PEER_MSG_CLASS_ERROR,
95 PEER_MSG_CLASS_STICKTABLE = 10,
96 PEER_MSG_CLASS_RESERVED = 255,
97};
98
99/*****************************/
100/* control message types */
101/*****************************/
102enum {
103 PEER_MSG_CTRL_RESYNCREQ = 0,
104 PEER_MSG_CTRL_RESYNCFINISHED,
105 PEER_MSG_CTRL_RESYNCPARTIAL,
106 PEER_MSG_CTRL_RESYNCCONFIRM,
107};
108
109/*****************************/
110/* error message types */
111/*****************************/
112enum {
113 PEER_MSG_ERR_PROTOCOL = 0,
114 PEER_MSG_ERR_SIZELIMIT,
115};
116
117
118/*******************************/
119/* stick table sync mesg types */
120/* Note: ids >= 128 contains */
121/* id message cotains data */
122/*******************************/
Olivier Houchard33992262018-10-16 18:49:26 +0200123#define PEER_MSG_STKT_UPDATE 0x80
124#define PEER_MSG_STKT_INCUPDATE 0x81
125#define PEER_MSG_STKT_DEFINE 0x82
126#define PEER_MSG_STKT_SWITCH 0x83
127#define PEER_MSG_STKT_ACK 0x84
128#define PEER_MSG_STKT_UPDATE_TIMED 0x85
129#define PEER_MSG_STKT_INCUPDATE_TIMED 0x86
Emeric Brun2b920a12010-09-23 18:30:22 +0200130
131/**********************************/
132/* Peer Session IO handler states */
133/**********************************/
134
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100135enum {
136 PEER_SESS_ST_ACCEPT = 0, /* Initial state for session create by an accept, must be zero! */
137 PEER_SESS_ST_GETVERSION, /* Validate supported protocol version */
138 PEER_SESS_ST_GETHOST, /* Validate host ID correspond to local host id */
139 PEER_SESS_ST_GETPEER, /* Validate peer ID correspond to a known remote peer id */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100140 /* after this point, data were possibly exchanged */
141 PEER_SESS_ST_SENDSUCCESS, /* Send ret code 200 (success) and wait for message */
142 PEER_SESS_ST_CONNECT, /* Initial state for session create on a connect, push presentation into buffer */
143 PEER_SESS_ST_GETSTATUS, /* Wait for the welcome message */
144 PEER_SESS_ST_WAITMSG, /* Wait for data messages */
145 PEER_SESS_ST_EXIT, /* Exit with status code */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200146 PEER_SESS_ST_ERRPROTO, /* Send error proto message before exit */
147 PEER_SESS_ST_ERRSIZE, /* Send error size message before exit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100148 PEER_SESS_ST_END, /* Killed session */
149};
Emeric Brun2b920a12010-09-23 18:30:22 +0200150
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100151/***************************************************/
152/* Peer Session status code - part of the protocol */
153/***************************************************/
Emeric Brun2b920a12010-09-23 18:30:22 +0200154
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100155#define PEER_SESS_SC_CONNECTCODE 100 /* connect in progress */
156#define PEER_SESS_SC_CONNECTEDCODE 110 /* tcp connect success */
Emeric Brun2b920a12010-09-23 18:30:22 +0200157
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100158#define PEER_SESS_SC_SUCCESSCODE 200 /* accept or connect successful */
Emeric Brun2b920a12010-09-23 18:30:22 +0200159
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100160#define PEER_SESS_SC_TRYAGAIN 300 /* try again later */
Emeric Brun2b920a12010-09-23 18:30:22 +0200161
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100162#define PEER_SESS_SC_ERRPROTO 501 /* error protocol */
163#define PEER_SESS_SC_ERRVERSION 502 /* unknown protocol version */
164#define PEER_SESS_SC_ERRHOST 503 /* bad host name */
165#define PEER_SESS_SC_ERRPEER 504 /* unknown peer */
Emeric Brun2b920a12010-09-23 18:30:22 +0200166
167#define PEER_SESSION_PROTO_NAME "HAProxyS"
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200168#define PEER_MAJOR_VER 2
169#define PEER_MINOR_VER 1
170#define PEER_DWNGRD_MINOR_VER 0
Emeric Brun2b920a12010-09-23 18:30:22 +0200171
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200172struct peers *cfg_peers = NULL;
Willy Tarreau81bc3b02016-10-31 17:37:39 +0100173static void peer_session_forceshutdown(struct appctx *appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +0200174
Emeric Brun18928af2017-03-29 16:32:53 +0200175/* This function encode an uint64 to 'dynamic' length format.
176 The encoded value is written at address *str, and the
177 caller must assure that size after *str is large enought.
178 At return, the *str is set at the next Byte after then
179 encoded integer. The function returns then length of the
180 encoded integer in Bytes */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200181int intencode(uint64_t i, char **str) {
182 int idx = 0;
183 unsigned char *msg;
184
185 if (!*str)
186 return 0;
187
188 msg = (unsigned char *)*str;
189 if (i < 240) {
190 msg[0] = (unsigned char)i;
191 *str = (char *)&msg[idx+1];
192 return (idx+1);
193 }
194
195 msg[idx] =(unsigned char)i | 240;
196 i = (i - 240) >> 4;
197 while (i >= 128) {
198 msg[++idx] = (unsigned char)i | 128;
199 i = (i - 128) >> 7;
200 }
201 msg[++idx] = (unsigned char)i;
202 *str = (char *)&msg[idx+1];
203 return (idx+1);
204}
205
206
207/* This function returns the decoded integer or 0
208 if decode failed
209 *str point on the beginning of the integer to decode
210 at the end of decoding *str point on the end of the
211 encoded integer or to null if end is reached */
Emeric Brun18928af2017-03-29 16:32:53 +0200212uint64_t intdecode(char **str, char *end)
213{
Emeric Brunb3971ab2015-05-12 18:49:09 +0200214 unsigned char *msg;
Emeric Brun18928af2017-03-29 16:32:53 +0200215 uint64_t i;
216 int shift;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200217
218 if (!*str)
219 return 0;
220
221 msg = (unsigned char *)*str;
Emeric Brun18928af2017-03-29 16:32:53 +0200222 if (msg >= (unsigned char *)end)
223 goto fail;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200224
Emeric Brun18928af2017-03-29 16:32:53 +0200225 i = *(msg++);
226 if (i >= 240) {
227 shift = 4;
228 do {
229 if (msg >= (unsigned char *)end)
230 goto fail;
231 i += (uint64_t)*msg << shift;
232 shift += 7;
233 } while (*(msg++) >= 128);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200234 }
Emeric Brun18928af2017-03-29 16:32:53 +0200235 *str = (char *)msg;
236 return i;
237
238 fail:
239 *str = NULL;
240 return 0;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200241}
Emeric Brun2b920a12010-09-23 18:30:22 +0200242
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200243/* Set the stick-table UPDATE message type byte at <msg_type> address,
244 * depending on <use_identifier> and <use_timed> boolean parameters.
245 * Always successful.
246 */
247static inline void peer_set_update_msg_type(char *msg_type, int use_identifier, int use_timed)
248{
249 if (use_timed) {
250 if (use_identifier)
251 *msg_type = PEER_MSG_STKT_UPDATE_TIMED;
252 else
253 *msg_type = PEER_MSG_STKT_INCUPDATE_TIMED;
254 }
255 else {
256 if (use_identifier)
257 *msg_type = PEER_MSG_STKT_UPDATE;
258 else
259 *msg_type = PEER_MSG_STKT_INCUPDATE;
260 }
261
262}
Emeric Brun2b920a12010-09-23 18:30:22 +0200263/*
Emeric Brunb3971ab2015-05-12 18:49:09 +0200264 * This prepare the data update message on the stick session <ts>, <st> is the considered
265 * stick table.
266 * <msg> is a buffer of <size> to recieve data message content
267 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
268 * check size)
Emeric Brun2b920a12010-09-23 18:30:22 +0200269 */
Emeric Brun819fc6f2017-06-13 19:37:32 +0200270static int peer_prepare_updatemsg(struct stksess *ts, struct shared_table *st, unsigned int updateid, char *msg, size_t size, int use_identifier, int use_timed)
Emeric Brun2b920a12010-09-23 18:30:22 +0200271{
272 uint32_t netinteger;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200273 unsigned short datalen;
274 char *cursor, *datamsg;
Emeric Brun94900952015-06-11 18:25:54 +0200275 unsigned int data_type;
276 void *data_ptr;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200277
278 cursor = datamsg = msg + 1 + 5;
279
Emeric Brun2b920a12010-09-23 18:30:22 +0200280 /* construct message */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200281
282 /* check if we need to send the update identifer */
Emeric Brun819fc6f2017-06-13 19:37:32 +0200283 if (!st->last_pushed || updateid < st->last_pushed || ((updateid - st->last_pushed) != 1)) {
Emeric Bruna6a09982015-09-22 15:34:19 +0200284 use_identifier = 1;
Emeric Brun2b920a12010-09-23 18:30:22 +0200285 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200286
287 /* encode update identifier if needed */
288 if (use_identifier) {
Emeric Brun819fc6f2017-06-13 19:37:32 +0200289 netinteger = htonl(updateid);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200290 memcpy(cursor, &netinteger, sizeof(netinteger));
291 cursor += sizeof(netinteger);
Emeric Brun2b920a12010-09-23 18:30:22 +0200292 }
293
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200294 if (use_timed) {
295 netinteger = htonl(tick_remain(now_ms, ts->expire));
296 memcpy(cursor, &netinteger, sizeof(netinteger));
297 cursor += sizeof(netinteger);
298 }
299
Emeric Brunb3971ab2015-05-12 18:49:09 +0200300 /* encode the key */
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +0200301 if (st->table->type == SMP_T_STR) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200302 int stlen = strlen((char *)ts->key.key);
303
Emeric Brunb3971ab2015-05-12 18:49:09 +0200304 intencode(stlen, &cursor);
305 memcpy(cursor, ts->key.key, stlen);
306 cursor += stlen;
Emeric Brun2b920a12010-09-23 18:30:22 +0200307 }
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +0200308 else if (st->table->type == SMP_T_SINT) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200309 netinteger = htonl(*((uint32_t *)ts->key.key));
Emeric Brunb3971ab2015-05-12 18:49:09 +0200310 memcpy(cursor, &netinteger, sizeof(netinteger));
311 cursor += sizeof(netinteger);
Emeric Brun2b920a12010-09-23 18:30:22 +0200312 }
313 else {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200314 memcpy(cursor, ts->key.key, st->table->key_size);
315 cursor += st->table->key_size;
Emeric Brun2b920a12010-09-23 18:30:22 +0200316 }
317
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100318 HA_RWLOCK_RDLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200319 /* encode values */
Emeric Brun94900952015-06-11 18:25:54 +0200320 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200321
Emeric Brun94900952015-06-11 18:25:54 +0200322 data_ptr = stktable_data_ptr(st->table, ts, data_type);
323 if (data_ptr) {
324 switch (stktable_data_types[data_type].std_type) {
325 case STD_T_SINT: {
326 int data;
327
328 data = stktable_data_cast(data_ptr, std_t_sint);
329 intencode(data, &cursor);
330 break;
331 }
332 case STD_T_UINT: {
333 unsigned int data;
334
335 data = stktable_data_cast(data_ptr, std_t_uint);
336 intencode(data, &cursor);
337 break;
338 }
339 case STD_T_ULL: {
340 unsigned long long data;
341
342 data = stktable_data_cast(data_ptr, std_t_ull);
343 intencode(data, &cursor);
344 break;
345 }
346 case STD_T_FRQP: {
347 struct freq_ctr_period *frqp;
348
349 frqp = &stktable_data_cast(data_ptr, std_t_frqp);
350 intencode((unsigned int)(now_ms - frqp->curr_tick), &cursor);
351 intencode(frqp->curr_ctr, &cursor);
352 intencode(frqp->prev_ctr, &cursor);
353 break;
354 }
355 }
356 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200357 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100358 HA_RWLOCK_RDUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200359
360 /* Compute datalen */
361 datalen = (cursor - datamsg);
362
363 /* prepare message header */
364 msg[0] = PEER_MSG_CLASS_STICKTABLE;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200365 peer_set_update_msg_type(&msg[1], use_identifier, use_timed);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200366 cursor = &msg[2];
367 intencode(datalen, &cursor);
368
369 /* move data after header */
370 memmove(cursor, datamsg, datalen);
371
372 /* return header size + data_len */
373 return (cursor - msg) + datalen;
374}
375
376/*
377 * This prepare the switch table message to targeted share table <st>.
378 * <msg> is a buffer of <size> to recieve data message content
379 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
380 * check size)
381 */
382static int peer_prepare_switchmsg(struct shared_table *st, char *msg, size_t size)
383{
384 int len;
385 unsigned short datalen;
Willy Tarreau83061a82018-07-13 11:56:34 +0200386 struct buffer *chunk;
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200387 char *cursor, *datamsg, *chunkp, *chunkq;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200388 uint64_t data = 0;
Emeric Brun94900952015-06-11 18:25:54 +0200389 unsigned int data_type;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200390
391 cursor = datamsg = msg + 2 + 5;
392
393 /* Encode data */
394
395 /* encode local id */
396 intencode(st->local_id, &cursor);
397
398 /* encode table name */
399 len = strlen(st->table->id);
400 intencode(len, &cursor);
401 memcpy(cursor, st->table->id, len);
402 cursor += len;
403
404 /* encode table type */
405
406 intencode(st->table->type, &cursor);
407
408 /* encode table key size */
409 intencode(st->table->key_size, &cursor);
410
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200411 chunk = get_trash_chunk();
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200412 chunkp = chunkq = chunk->area;
Emeric Brun94900952015-06-11 18:25:54 +0200413 /* encode available known data types in table */
414 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
415 if (st->table->data_ofs[data_type]) {
416 switch (stktable_data_types[data_type].std_type) {
417 case STD_T_SINT:
418 case STD_T_UINT:
419 case STD_T_ULL:
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200420 data |= 1 << data_type;
421 break;
Emeric Brun94900952015-06-11 18:25:54 +0200422 case STD_T_FRQP:
423 data |= 1 << data_type;
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200424 intencode(data_type, &chunkq);
425 intencode(st->table->data_arg[data_type].u, &chunkq);
Emeric Brun94900952015-06-11 18:25:54 +0200426 break;
427 }
428 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200429 }
430 intencode(data, &cursor);
431
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200432 /* Encode stick-table entries duration. */
433 intencode(st->table->expire, &cursor);
434
435 if (chunkq > chunkp) {
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200436 chunk->data = chunkq - chunkp;
437 memcpy(cursor, chunk->area, chunk->data);
438 cursor += chunk->data;
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200439 }
440
Emeric Brunb3971ab2015-05-12 18:49:09 +0200441 /* Compute datalen */
442 datalen = (cursor - datamsg);
Emeric Brun2b920a12010-09-23 18:30:22 +0200443
Emeric Brunb3971ab2015-05-12 18:49:09 +0200444 /* prepare message header */
445 msg[0] = PEER_MSG_CLASS_STICKTABLE;
446 msg[1] = PEER_MSG_STKT_DEFINE;
447 cursor = &msg[2];
448 intencode(datalen, &cursor);
Emeric Brun2b920a12010-09-23 18:30:22 +0200449
Emeric Brunb3971ab2015-05-12 18:49:09 +0200450 /* move data after header */
451 memmove(cursor, datamsg, datalen);
452
453 /* return header size + data_len */
454 return (cursor - msg) + datalen;
Emeric Brun2b920a12010-09-23 18:30:22 +0200455}
456
Emeric Brunb3971ab2015-05-12 18:49:09 +0200457/*
458 * This prepare the acknowledge message on the stick session <ts>, <st> is the considered
459 * stick table.
460 * <msg> is a buffer of <size> to recieve data message content
461 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
462 * check size)
463 */
464static int peer_prepare_ackmsg(struct shared_table *st, char *msg, size_t size)
465{
466 unsigned short datalen;
467 char *cursor, *datamsg;
468 uint32_t netinteger;
469
Emeric Brunb058f1c2015-09-22 15:50:18 +0200470 cursor = datamsg = msg + 2 + 5;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200471
472 intencode(st->remote_id, &cursor);
473 netinteger = htonl(st->last_get);
474 memcpy(cursor, &netinteger, sizeof(netinteger));
475 cursor += sizeof(netinteger);
476
477 /* Compute datalen */
478 datalen = (cursor - datamsg);
479
480 /* prepare message header */
481 msg[0] = PEER_MSG_CLASS_STICKTABLE;
Emeric Brune1ab8082015-08-21 11:48:54 +0200482 msg[1] = PEER_MSG_STKT_ACK;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200483 cursor = &msg[2];
484 intencode(datalen, &cursor);
485
486 /* move data after header */
487 memmove(cursor, datamsg, datalen);
488
489 /* return header size + data_len */
490 return (cursor - msg) + datalen;
491}
Emeric Brun2b920a12010-09-23 18:30:22 +0200492
493/*
494 * Callback to release a session with a peer
495 */
Willy Tarreau00a37f02015-04-13 12:05:19 +0200496static void peer_session_release(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +0200497{
Willy Tarreau00a37f02015-04-13 12:05:19 +0200498 struct stream_interface *si = appctx->owner;
Willy Tarreau87b09662015-04-03 00:22:06 +0200499 struct stream *s = si_strm(si);
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200500 struct peer *peer = appctx->ctx.peers.ptr;
501 struct peers *peers = strm_fe(s)->parent;
Emeric Brun2b920a12010-09-23 18:30:22 +0200502
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100503 /* appctx->ctx.peers.ptr is not a peer session */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100504 if (appctx->st0 < PEER_SESS_ST_SENDSUCCESS)
Emeric Brun2b920a12010-09-23 18:30:22 +0200505 return;
506
507 /* peer session identified */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200508 if (peer) {
Willy Tarreau2d372c22018-11-05 17:12:27 +0100509 if (appctx->st0 == PEER_SESS_ST_WAITMSG)
510 HA_ATOMIC_SUB(&connected_peers, 1);
Willy Tarreau199ad242018-11-05 16:31:22 +0100511 HA_ATOMIC_SUB(&active_peers, 1);
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100512 HA_SPIN_LOCK(PEER_LOCK, &peer->lock);
Willy Tarreau9df94c22016-10-31 18:42:52 +0100513 if (peer->appctx == appctx) {
Emeric Brunb157d732015-08-21 12:00:30 +0200514 /* Re-init current table pointers to force announcement on re-connect */
515 peer->remote_table = peer->last_local_table = NULL;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200516 peer->appctx = NULL;
517 if (peer->flags & PEER_F_LEARN_ASSIGN) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200518 /* unassign current peer for learning */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200519 peer->flags &= ~(PEER_F_LEARN_ASSIGN);
520 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brun2b920a12010-09-23 18:30:22 +0200521
522 /* reschedule a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200523 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
Emeric Brun2b920a12010-09-23 18:30:22 +0200524 }
525 /* reset teaching and learning flags to 0 */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200526 peer->flags &= PEER_TEACH_RESET;
527 peer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200528 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100529 HA_SPIN_UNLOCK(PEER_LOCK, &peer->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200530 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200531 }
532}
533
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200534/* Retrieve the major and minor versions of peers protocol
535 * announced by a remote peer. <str> is a null-terminated
536 * string with the following format: "<maj_ver>.<min_ver>".
537 */
538static int peer_get_version(const char *str,
539 unsigned int *maj_ver, unsigned int *min_ver)
540{
541 unsigned int majv, minv;
542 const char *pos, *saved;
543 const char *end;
544
545 saved = pos = str;
546 end = str + strlen(str);
547
548 majv = read_uint(&pos, end);
549 if (saved == pos || *pos++ != '.')
550 return -1;
551
552 saved = pos;
553 minv = read_uint(&pos, end);
554 if (saved == pos || pos != end)
555 return -1;
556
557 *maj_ver = majv;
558 *min_ver = minv;
559
560 return 0;
561}
Emeric Brun2b920a12010-09-23 18:30:22 +0200562
563/*
564 * IO Handler to handle message exchance with a peer
565 */
Willy Tarreau00a37f02015-04-13 12:05:19 +0200566static void peer_io_handler(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +0200567{
Willy Tarreau00a37f02015-04-13 12:05:19 +0200568 struct stream_interface *si = appctx->owner;
Willy Tarreau87b09662015-04-03 00:22:06 +0200569 struct stream *s = si_strm(si);
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200570 struct peers *curpeers = strm_fe(s)->parent;
Emeric Brun80527f52017-06-19 17:46:37 +0200571 struct peer *curpeer = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +0200572 int reql = 0;
573 int repl = 0;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200574 size_t proto_len = strlen(PEER_SESSION_PROTO_NAME);
575 unsigned int maj_ver, min_ver;
Willy Tarreau2d372c22018-11-05 17:12:27 +0100576 int prev_state;
Emeric Brun2b920a12010-09-23 18:30:22 +0200577
Christopher Fauleta73e59b2016-12-09 17:30:18 +0100578 /* Check if the input buffer is avalaible. */
Willy Tarreauc9fa0482018-07-10 17:43:27 +0200579 if (si_ic(si)->buf.size == 0)
Christopher Fauleta73e59b2016-12-09 17:30:18 +0100580 goto full;
581
Emeric Brun2b920a12010-09-23 18:30:22 +0200582 while (1) {
Willy Tarreau2d372c22018-11-05 17:12:27 +0100583 prev_state = appctx->st0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200584switchstate:
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200585 maj_ver = min_ver = (unsigned int)-1;
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100586 switch(appctx->st0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100587 case PEER_SESS_ST_ACCEPT:
Willy Tarreau2d372c22018-11-05 17:12:27 +0100588 prev_state = appctx->st0;
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100589 appctx->ctx.peers.ptr = NULL;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100590 appctx->st0 = PEER_SESS_ST_GETVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200591 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100592 case PEER_SESS_ST_GETVERSION:
Willy Tarreau2d372c22018-11-05 17:12:27 +0100593 prev_state = appctx->st0;
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200594 reql = co_getline(si_oc(si), trash.area,
595 trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200596 if (reql <= 0) { /* closed or EOL not found */
597 if (reql == 0)
598 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100599 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200600 goto switchstate;
601 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200602 if (trash.area[reql-1] != '\n') {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100603 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200604 goto switchstate;
605 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200606 else if (reql > 1 && (trash.area[reql-2] == '\r'))
607 trash.area[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200608 else
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200609 trash.area[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200610
Willy Tarreau06d80a92017-10-19 14:32:15 +0200611 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200612
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200613 /* test protocol */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200614 if (strncmp(PEER_SESSION_PROTO_NAME " ", trash.area, proto_len + 1) != 0) {
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200615 appctx->st0 = PEER_SESS_ST_EXIT;
616 appctx->st1 = PEER_SESS_SC_ERRPROTO;
617 goto switchstate;
618 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200619 if (peer_get_version(trash.area + proto_len + 1, &maj_ver, &min_ver) == -1 ||
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200620 maj_ver != PEER_MAJOR_VER || min_ver > PEER_MINOR_VER) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100621 appctx->st0 = PEER_SESS_ST_EXIT;
622 appctx->st1 = PEER_SESS_SC_ERRVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200623 goto switchstate;
624 }
625
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100626 appctx->st0 = PEER_SESS_ST_GETHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200627 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100628 case PEER_SESS_ST_GETHOST:
Willy Tarreau2d372c22018-11-05 17:12:27 +0100629 prev_state = appctx->st0;
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200630 reql = co_getline(si_oc(si), trash.area,
631 trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200632 if (reql <= 0) { /* closed or EOL not found */
633 if (reql == 0)
634 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100635 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200636 goto switchstate;
637 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200638 if (trash.area[reql-1] != '\n') {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100639 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200640 goto switchstate;
641 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200642 else if (reql > 1 && (trash.area[reql-2] == '\r'))
643 trash.area[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200644 else
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200645 trash.area[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200646
Willy Tarreau06d80a92017-10-19 14:32:15 +0200647 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200648
649 /* test hostname match */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200650 if (strcmp(localpeer, trash.area) != 0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100651 appctx->st0 = PEER_SESS_ST_EXIT;
652 appctx->st1 = PEER_SESS_SC_ERRHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200653 goto switchstate;
654 }
655
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100656 appctx->st0 = PEER_SESS_ST_GETPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200657 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100658 case PEER_SESS_ST_GETPEER: {
Emeric Brun2b920a12010-09-23 18:30:22 +0200659 char *p;
Willy Tarreau2d372c22018-11-05 17:12:27 +0100660
661 prev_state = appctx->st0;
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200662 reql = co_getline(si_oc(si), trash.area,
663 trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200664 if (reql <= 0) { /* closed or EOL not found */
665 if (reql == 0)
666 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100667 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200668 goto switchstate;
669 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200670 if (trash.area[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200671 /* Incomplete line, we quit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100672 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200673 goto switchstate;
674 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200675 else if (reql > 1 && (trash.area[reql-2] == '\r'))
676 trash.area[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200677 else
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200678 trash.area[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200679
Willy Tarreau06d80a92017-10-19 14:32:15 +0200680 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200681
Emeric Brunb3971ab2015-05-12 18:49:09 +0200682 /* parse line "<peer name> <pid> <relative_pid>" */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200683 p = strchr(trash.area, ' ');
Emeric Brun2b920a12010-09-23 18:30:22 +0200684 if (!p) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100685 appctx->st0 = PEER_SESS_ST_EXIT;
686 appctx->st1 = PEER_SESS_SC_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +0200687 goto switchstate;
688 }
689 *p = 0;
690
691 /* lookup known peer */
692 for (curpeer = curpeers->remote; curpeer; curpeer = curpeer->next) {
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200693 if (strcmp(curpeer->id, trash.area) == 0)
Emeric Brun2b920a12010-09-23 18:30:22 +0200694 break;
695 }
696
697 /* if unknown peer */
698 if (!curpeer) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100699 appctx->st0 = PEER_SESS_ST_EXIT;
700 appctx->st1 = PEER_SESS_SC_ERRPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200701 goto switchstate;
702 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200703
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100704 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Willy Tarreau9df94c22016-10-31 18:42:52 +0100705 if (curpeer->appctx && curpeer->appctx != appctx) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200706 if (curpeer->local) {
707 /* Local connection, reply a retry */
708 appctx->st0 = PEER_SESS_ST_EXIT;
709 appctx->st1 = PEER_SESS_SC_TRYAGAIN;
710 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +0200711 }
Emeric Brun80527f52017-06-19 17:46:37 +0200712
713 /* we're killing a connection, we must apply a random delay before
714 * retrying otherwise the other end will do the same and we can loop
715 * for a while.
716 */
717 curpeer->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + random() % 2000));
Willy Tarreau81bc3b02016-10-31 17:37:39 +0100718 peer_session_forceshutdown(curpeer->appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +0200719 }
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200720 if (maj_ver != (unsigned int)-1 && min_ver != (unsigned int)-1) {
721 if (min_ver == PEER_DWNGRD_MINOR_VER) {
722 curpeer->flags |= PEER_F_DWNGRD;
723 }
724 else {
725 curpeer->flags &= ~PEER_F_DWNGRD;
726 }
727 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200728 curpeer->appctx = appctx;
729 appctx->ctx.peers.ptr = curpeer;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100730 appctx->st0 = PEER_SESS_ST_SENDSUCCESS;
Willy Tarreau199ad242018-11-05 16:31:22 +0100731 HA_ATOMIC_ADD(&active_peers, 1);
Emeric Brun2b920a12010-09-23 18:30:22 +0200732 /* fall through */
733 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100734 case PEER_SESS_ST_SENDSUCCESS: {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200735 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +0200736
Willy Tarreau2d372c22018-11-05 17:12:27 +0100737 prev_state = appctx->st0;
Emeric Brun80527f52017-06-19 17:46:37 +0200738 if (!curpeer) {
739 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100740 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +0200741 if (curpeer->appctx != appctx) {
742 appctx->st0 = PEER_SESS_ST_END;
743 goto switchstate;
744 }
745 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200746 repl = snprintf(trash.area, trash.size,
747 "%d\n",
748 PEER_SESS_SC_SUCCESSCODE);
749 repl = ci_putblk(si_ic(si), trash.area, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200750 if (repl <= 0) {
751 if (repl == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +0100752 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100753 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200754 goto switchstate;
755 }
756
757 /* Register status code */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200758 curpeer->statuscode = PEER_SESS_SC_SUCCESSCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200759
760 /* Awake main task */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200761 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200762
763 /* Init confirm counter */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200764 curpeer->confirm = 0;
765
766 /* Init cursors */
767 for (st = curpeer->tables; st ; st = st->next) {
768 st->last_get = st->last_acked = 0;
769 st->teaching_origin = st->last_pushed = st->update;
770 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200771
772 /* reset teaching and learning flags to 0 */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200773 curpeer->flags &= PEER_TEACH_RESET;
774 curpeer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200775
776 /* if current peer is local */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200777 if (curpeer->local) {
778 /* if current host need resyncfrom local and no process assined */
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200779 if ((curpeers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL &&
780 !(curpeers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200781 /* assign local peer for a lesson, consider lesson already requested */
782 curpeer->flags |= PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200783 curpeers->flags |= (PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200784 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200785
Emeric Brunb3971ab2015-05-12 18:49:09 +0200786 }
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200787 else if ((curpeers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
788 !(curpeers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200789 /* assign peer for a lesson */
790 curpeer->flags |= PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200791 curpeers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200792 }
793
794
Emeric Brun2b920a12010-09-23 18:30:22 +0200795 /* switch to waiting message state */
Willy Tarreau2d372c22018-11-05 17:12:27 +0100796 HA_ATOMIC_ADD(&connected_peers, 1);
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100797 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200798 goto switchstate;
799 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100800 case PEER_SESS_ST_CONNECT: {
Willy Tarreau2d372c22018-11-05 17:12:27 +0100801 prev_state = appctx->st0;
Emeric Brun80527f52017-06-19 17:46:37 +0200802 if (!curpeer) {
803 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100804 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +0200805 if (curpeer->appctx != appctx) {
806 appctx->st0 = PEER_SESS_ST_END;
807 goto switchstate;
808 }
809 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200810
811 /* Send headers */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200812 repl = snprintf(trash.area, trash.size,
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200813 PEER_SESSION_PROTO_NAME " %u.%u\n%s\n%s %d %d\n",
814 PEER_MAJOR_VER,
815 (curpeer->flags & PEER_F_DWNGRD) ? PEER_DWNGRD_MINOR_VER : PEER_MINOR_VER,
Emeric Brunb3971ab2015-05-12 18:49:09 +0200816 curpeer->id,
Emeric Brun2b920a12010-09-23 18:30:22 +0200817 localpeer,
Willy Tarreau7b77c9f2012-01-07 22:52:12 +0100818 (int)getpid(),
Emeric Brunb3971ab2015-05-12 18:49:09 +0200819 relative_pid);
Emeric Brun2b920a12010-09-23 18:30:22 +0200820
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100821 if (repl >= trash.size) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100822 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200823 goto switchstate;
824 }
825
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200826 repl = ci_putblk(si_ic(si), trash.area, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200827 if (repl <= 0) {
828 if (repl == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +0100829 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100830 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200831 goto switchstate;
832 }
833
834 /* switch to the waiting statuscode state */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100835 appctx->st0 = PEER_SESS_ST_GETSTATUS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200836 /* fall through */
837 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100838 case PEER_SESS_ST_GETSTATUS: {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200839 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +0200840
Willy Tarreau2d372c22018-11-05 17:12:27 +0100841 prev_state = appctx->st0;
Emeric Brun80527f52017-06-19 17:46:37 +0200842 if (!curpeer) {
843 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100844 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +0200845 if (curpeer->appctx != appctx) {
846 appctx->st0 = PEER_SESS_ST_END;
847 goto switchstate;
848 }
849 }
850
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100851 if (si_ic(si)->flags & CF_WRITE_PARTIAL)
Emeric Brunb3971ab2015-05-12 18:49:09 +0200852 curpeer->statuscode = PEER_SESS_SC_CONNECTEDCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200853
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200854 reql = co_getline(si_oc(si), trash.area,
855 trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200856 if (reql <= 0) { /* closed or EOL not found */
857 if (reql == 0)
858 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100859 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200860 goto switchstate;
861 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200862 if (trash.area[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200863 /* Incomplete line, we quit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100864 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200865 goto switchstate;
866 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200867 else if (reql > 1 && (trash.area[reql-2] == '\r'))
868 trash.area[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200869 else
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200870 trash.area[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200871
Willy Tarreau06d80a92017-10-19 14:32:15 +0200872 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200873
874 /* Register status code */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200875 curpeer->statuscode = atoi(trash.area);
Emeric Brun2b920a12010-09-23 18:30:22 +0200876
877 /* Awake main task */
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200878 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200879
880 /* If status code is success */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200881 if (curpeer->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200882 /* Init cursors */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200883 for (st = curpeer->tables; st ; st = st->next) {
884 st->last_get = st->last_acked = 0;
885 st->teaching_origin = st->last_pushed = st->update;
886 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200887
888 /* Init confirm counter */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200889 curpeer->confirm = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200890
Emeric Brunb3971ab2015-05-12 18:49:09 +0200891 /* reset teaching and learning flags to 0 */
892 curpeer->flags &= PEER_TEACH_RESET;
893 curpeer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200894
Emeric Brunb3971ab2015-05-12 18:49:09 +0200895 /* If current peer is local */
896 if (curpeer->local) {
897 /* flag to start to teach lesson */
898 curpeer->flags |= PEER_F_TEACH_PROCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200899
Emeric Brunb3971ab2015-05-12 18:49:09 +0200900 }
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200901 else if ((curpeers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
902 !(curpeers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200903 /* If peer is remote and resync from remote is needed,
904 and no peer currently assigned */
Emeric Brun2b920a12010-09-23 18:30:22 +0200905
Emeric Brunb3971ab2015-05-12 18:49:09 +0200906 /* assign peer for a lesson */
907 curpeer->flags |= PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200908 curpeers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brun2b920a12010-09-23 18:30:22 +0200909 }
910
911 }
912 else {
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200913 if (curpeer->statuscode == PEER_SESS_SC_ERRVERSION)
914 curpeer->flags |= PEER_F_DWNGRD;
Emeric Brun2b920a12010-09-23 18:30:22 +0200915 /* Status code is not success, abort */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100916 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200917 goto switchstate;
918 }
Willy Tarreau2d372c22018-11-05 17:12:27 +0100919 HA_ATOMIC_ADD(&connected_peers, 1);
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100920 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200921 /* fall through */
922 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100923 case PEER_SESS_ST_WAITMSG: {
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200924 struct stksess *ts, *newts = NULL;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200925 uint32_t msg_len = 0;
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200926 char *msg_cur = trash.area;
927 char *msg_end = trash.area;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200928 unsigned char msg_head[7];
Emeric Brun2b920a12010-09-23 18:30:22 +0200929 int totl = 0;
930
Willy Tarreau2d372c22018-11-05 17:12:27 +0100931 prev_state = appctx->st0;
Emeric Brun80527f52017-06-19 17:46:37 +0200932 if (!curpeer) {
933 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100934 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +0200935 if (curpeer->appctx != appctx) {
936 appctx->st0 = PEER_SESS_ST_END;
937 goto switchstate;
938 }
939 }
940
Willy Tarreau06d80a92017-10-19 14:32:15 +0200941 reql = co_getblk(si_oc(si), (char *)msg_head, 2*sizeof(unsigned char), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200942 if (reql <= 0) /* closed or EOL not found */
943 goto incomplete;
944
Emeric Brun2b920a12010-09-23 18:30:22 +0200945 totl += reql;
946
Emeric Brunb3971ab2015-05-12 18:49:09 +0200947 if (msg_head[1] >= 128) {
948 /* Read and Decode message length */
Willy Tarreau06d80a92017-10-19 14:32:15 +0200949 reql = co_getblk(si_oc(si), (char *)&msg_head[2], sizeof(unsigned char), totl);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200950 if (reql <= 0) /* closed */
951 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +0200952
Emeric Brunb3971ab2015-05-12 18:49:09 +0200953 totl += reql;
954
955 if (msg_head[2] < 240) {
956 msg_len = msg_head[2];
Emeric Brun2b920a12010-09-23 18:30:22 +0200957 }
958 else {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200959 int i;
960 char *cur;
961 char *end;
Willy Tarreau72d6c162013-04-11 16:14:13 +0200962
Emeric Brunb3971ab2015-05-12 18:49:09 +0200963 for (i = 3 ; i < sizeof(msg_head) ; i++) {
Willy Tarreau06d80a92017-10-19 14:32:15 +0200964 reql = co_getblk(si_oc(si), (char *)&msg_head[i], sizeof(char), totl);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200965 if (reql <= 0) /* closed */
966 goto incomplete;
967
968 totl += reql;
969
970 if (!(msg_head[i] & 0x80))
971 break;
972 }
973
974 if (i == sizeof(msg_head)) {
975 /* malformed message */
976 appctx->st0 = PEER_SESS_ST_ERRPROTO;
977 goto switchstate;
978
979 }
980 end = (char *)msg_head + sizeof(msg_head);
981 cur = (char *)&msg_head[2];
982 msg_len = intdecode(&cur, end);
983 if (!cur) {
984 /* malformed message */
985 appctx->st0 = PEER_SESS_ST_ERRPROTO;
986 goto switchstate;
987 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200988 }
989
Willy Tarreau86a446e2013-11-25 23:02:37 +0100990
Emeric Brunb3971ab2015-05-12 18:49:09 +0200991 /* Read message content */
992 if (msg_len) {
993 if (msg_len > trash.size) {
994 /* Status code is not success, abort */
995 appctx->st0 = PEER_SESS_ST_ERRSIZE;
996 goto switchstate;
997 }
Willy Tarreau72d6c162013-04-11 16:14:13 +0200998
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200999 reql = co_getblk(si_oc(si),
1000 trash.area,
1001 msg_len,
1002 totl);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001003 if (reql <= 0) /* closed */
1004 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +02001005 totl += reql;
Willy Tarreau86a446e2013-11-25 23:02:37 +01001006
Emeric Brunb3971ab2015-05-12 18:49:09 +02001007 msg_end += msg_len;
1008 }
1009 }
Willy Tarreau86a446e2013-11-25 23:02:37 +01001010
Emeric Brunb3971ab2015-05-12 18:49:09 +02001011 if (msg_head[0] == PEER_MSG_CLASS_CONTROL) {
1012 if (msg_head[1] == PEER_MSG_CTRL_RESYNCREQ) {
1013 struct shared_table *st;
1014 /* Reset message: remote need resync */
1015
1016 /* prepare tables fot a global push */
1017 for (st = curpeer->tables; st; st = st->next) {
1018 st->teaching_origin = st->last_pushed = st->table->update;
1019 st->flags = 0;
Willy Tarreau86a446e2013-11-25 23:02:37 +01001020 }
Willy Tarreau72d6c162013-04-11 16:14:13 +02001021
Emeric Brunb3971ab2015-05-12 18:49:09 +02001022 /* reset teaching flags to 0 */
1023 curpeer->flags &= PEER_TEACH_RESET;
Willy Tarreau9d9179b2013-04-11 16:56:44 +02001024
Emeric Brunb3971ab2015-05-12 18:49:09 +02001025 /* flag to start to teach lesson */
1026 curpeer->flags |= PEER_F_TEACH_PROCESS;
1027
1028
1029 }
1030 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCFINISHED) {
1031
1032 if (curpeer->flags & PEER_F_LEARN_ASSIGN) {
1033 curpeer->flags &= ~PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001034 curpeers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
1035 curpeers->flags |= (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE);
Willy Tarreau86a446e2013-11-25 23:02:37 +01001036 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001037 curpeer->confirm++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001038 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001039 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCPARTIAL) {
1040
1041 if (curpeer->flags & PEER_F_LEARN_ASSIGN) {
1042 curpeer->flags &= ~PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001043 curpeers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001044
1045 curpeer->flags |= PEER_F_LEARN_NOTUP2DATE;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001046 curpeers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
1047 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Cyril Bonté9a60ff92014-02-16 01:07:07 +01001048 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001049 curpeer->confirm++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001050 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001051 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCCONFIRM) {
Emeric Brun597b26e2016-08-12 11:23:31 +02001052 struct shared_table *st;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001053
1054 /* If stopping state */
1055 if (stopping) {
1056 /* Close session, push resync no more needed */
1057 curpeer->flags |= PEER_F_TEACH_COMPLETE;
1058 appctx->st0 = PEER_SESS_ST_END;
1059 goto switchstate;
1060 }
Emeric Brun597b26e2016-08-12 11:23:31 +02001061 for (st = curpeer->tables; st; st = st->next) {
1062 st->update = st->last_pushed = st->teaching_origin;
1063 st->flags = 0;
1064 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001065
1066 /* reset teaching flags to 0 */
1067 curpeer->flags &= PEER_TEACH_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +02001068 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001069 }
1070 else if (msg_head[0] == PEER_MSG_CLASS_STICKTABLE) {
1071 if (msg_head[1] == PEER_MSG_STKT_DEFINE) {
1072 int table_id_len;
1073 struct shared_table *st;
1074 int table_type;
1075 int table_keylen;
1076 int table_id;
1077 uint64_t table_data;
Emeric Brun2b920a12010-09-23 18:30:22 +02001078
Emeric Brunb3971ab2015-05-12 18:49:09 +02001079 table_id = intdecode(&msg_cur, msg_end);
1080 if (!msg_cur) {
1081 /* malformed message */
1082 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1083 goto switchstate;
1084 }
Willy Tarreau72d6c162013-04-11 16:14:13 +02001085
Emeric Brunb3971ab2015-05-12 18:49:09 +02001086 table_id_len = intdecode(&msg_cur, msg_end);
1087 if (!msg_cur) {
1088 /* malformed message */
1089 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1090 goto switchstate;
1091 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001092
Emeric Brunb3971ab2015-05-12 18:49:09 +02001093 curpeer->remote_table = NULL;
1094 if (!table_id_len || (msg_cur + table_id_len) >= msg_end) {
1095 /* malformed message */
1096 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1097 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001098 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001099
Emeric Brunb3971ab2015-05-12 18:49:09 +02001100 for (st = curpeer->tables; st; st = st->next) {
1101 /* Reset IDs */
1102 if (st->remote_id == table_id)
1103 st->remote_id = 0;
Willy Tarreau9d9179b2013-04-11 16:56:44 +02001104
Emeric Brunb3971ab2015-05-12 18:49:09 +02001105 if (!curpeer->remote_table
1106 && (table_id_len == strlen(st->table->id))
1107 && (memcmp(st->table->id, msg_cur, table_id_len) == 0)) {
1108 curpeer->remote_table = st;
Emeric Brun2b920a12010-09-23 18:30:22 +02001109 }
1110 }
1111
Emeric Brunb3971ab2015-05-12 18:49:09 +02001112 if (!curpeer->remote_table) {
1113 goto ignore_msg;
1114 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001115
Emeric Brunb3971ab2015-05-12 18:49:09 +02001116 msg_cur += table_id_len;
1117 if (msg_cur >= msg_end) {
1118 /* malformed message */
1119 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1120 goto switchstate;
1121 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001122
Emeric Brunb3971ab2015-05-12 18:49:09 +02001123 table_type = intdecode(&msg_cur, msg_end);
1124 if (!msg_cur) {
1125 /* malformed message */
1126 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1127 goto switchstate;
1128 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001129
Emeric Brunb3971ab2015-05-12 18:49:09 +02001130 table_keylen = intdecode(&msg_cur, msg_end);
1131 if (!msg_cur) {
1132 /* malformed message */
1133 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1134 goto switchstate;
1135 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001136
Emeric Brunb3971ab2015-05-12 18:49:09 +02001137 table_data = intdecode(&msg_cur, msg_end);
1138 if (!msg_cur) {
1139 /* malformed message */
1140 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1141 goto switchstate;
1142 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001143
Emeric Brunb3971ab2015-05-12 18:49:09 +02001144 if (curpeer->remote_table->table->type != table_type
1145 || curpeer->remote_table->table->key_size != table_keylen) {
1146 curpeer->remote_table = NULL;
1147 goto ignore_msg;
1148 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001149
Emeric Brunb3971ab2015-05-12 18:49:09 +02001150 curpeer->remote_table->remote_data = table_data;
1151 curpeer->remote_table->remote_id = table_id;
Emeric Brun2b920a12010-09-23 18:30:22 +02001152 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001153 else if (msg_head[1] == PEER_MSG_STKT_SWITCH) {
1154 struct shared_table *st;
1155 int table_id;
Emeric Brun2b920a12010-09-23 18:30:22 +02001156
Emeric Brunb3971ab2015-05-12 18:49:09 +02001157 table_id = intdecode(&msg_cur, msg_end);
1158 if (!msg_cur) {
1159 /* malformed message */
1160 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1161 goto switchstate;
1162 }
1163 curpeer->remote_table = NULL;
1164 for (st = curpeer->tables; st; st = st->next) {
1165 if (st->remote_id == table_id) {
1166 curpeer->remote_table = st;
1167 break;
1168 }
1169 }
1170
Emeric Brun2b920a12010-09-23 18:30:22 +02001171 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001172 else if (msg_head[1] == PEER_MSG_STKT_UPDATE
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001173 || msg_head[1] == PEER_MSG_STKT_INCUPDATE
1174 || msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED
1175 || msg_head[1] == PEER_MSG_STKT_INCUPDATE_TIMED) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001176 struct shared_table *st = curpeer->remote_table;
1177 uint32_t update;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001178 int expire;
Emeric Brun94900952015-06-11 18:25:54 +02001179 unsigned int data_type;
1180 void *data_ptr;
Emeric Brun2b920a12010-09-23 18:30:22 +02001181
Emeric Brunb3971ab2015-05-12 18:49:09 +02001182 /* Here we have data message */
1183 if (!st)
1184 goto ignore_msg;
Emeric Brun2b920a12010-09-23 18:30:22 +02001185
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001186 expire = MS_TO_TICKS(st->table->expire);
1187
1188 if (msg_head[1] == PEER_MSG_STKT_UPDATE ||
1189 msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001190 if (msg_len < sizeof(update)) {
1191 /* malformed message */
1192 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1193 goto switchstate;
1194 }
1195 memcpy(&update, msg_cur, sizeof(update));
1196 msg_cur += sizeof(update);
1197 st->last_get = htonl(update);
1198 }
1199 else {
1200 st->last_get++;
1201 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001202
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001203 if (msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED ||
1204 msg_head[1] == PEER_MSG_STKT_INCUPDATE_TIMED) {
1205 size_t expire_sz = sizeof expire;
1206
1207 if (msg_cur + expire_sz > msg_end) {
1208 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1209 goto switchstate;
1210 }
1211 memcpy(&expire, msg_cur, expire_sz);
1212 msg_cur += expire_sz;
1213 expire = ntohl(expire);
1214 }
1215
Emeric Brunb3971ab2015-05-12 18:49:09 +02001216 newts = stksess_new(st->table, NULL);
1217 if (!newts)
1218 goto ignore_msg;
Emeric Brun55482912018-01-22 15:10:08 +01001219
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +02001220 if (st->table->type == SMP_T_STR) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001221 unsigned int to_read, to_store;
Emeric Brun2b920a12010-09-23 18:30:22 +02001222
Emeric Brunb3971ab2015-05-12 18:49:09 +02001223 to_read = intdecode(&msg_cur, msg_end);
1224 if (!msg_cur) {
1225 /* malformed message */
1226 stksess_free(st->table, newts);
1227 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1228 goto switchstate;
1229 }
1230 to_store = MIN(to_read, st->table->key_size - 1);
1231 if (msg_cur + to_store > msg_end) {
1232 /* malformed message */
1233 stksess_free(st->table, newts);
1234 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1235 goto switchstate;
1236 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001237
Emeric Brunb3971ab2015-05-12 18:49:09 +02001238 memcpy(newts->key.key, msg_cur, to_store);
1239 newts->key.key[to_store] = 0;
1240 msg_cur += to_read;
1241 }
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +02001242 else if (st->table->type == SMP_T_SINT) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001243 unsigned int netinteger;
Willy Tarreau72d6c162013-04-11 16:14:13 +02001244
Emeric Brunb3971ab2015-05-12 18:49:09 +02001245 if (msg_cur + sizeof(netinteger) > msg_end) {
1246 /* malformed message */
1247 stksess_free(st->table, newts);
1248 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1249 goto switchstate;
1250 }
1251 memcpy(&netinteger, msg_cur, sizeof(netinteger));
1252 netinteger = ntohl(netinteger);
1253 memcpy(newts->key.key, &netinteger, sizeof(netinteger));
1254 msg_cur += sizeof(netinteger);
1255 }
1256 else {
1257 if (msg_cur + st->table->key_size > msg_end) {
1258 /* malformed message */
1259 stksess_free(st->table, newts);
1260 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1261 goto switchstate;
1262 }
1263 memcpy(newts->key.key, msg_cur, st->table->key_size);
1264 msg_cur += st->table->key_size;
1265 }
1266
1267 /* lookup for existing entry */
Emeric Brun819fc6f2017-06-13 19:37:32 +02001268 ts = stktable_set_entry(st->table, newts);
1269 if (ts != newts) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001270 stksess_free(st->table, newts);
1271 newts = NULL;
1272 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001273
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001274 HA_RWLOCK_WRLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001275
Emeric Brun94900952015-06-11 18:25:54 +02001276 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001277
Emeric Brun94900952015-06-11 18:25:54 +02001278 if ((1 << data_type) & st->remote_data) {
1279 switch (stktable_data_types[data_type].std_type) {
1280 case STD_T_SINT: {
1281 int data;
1282
1283 data = intdecode(&msg_cur, msg_end);
1284 if (!msg_cur) {
1285 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001286 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001287 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001288 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1289 goto switchstate;
1290 }
1291
1292 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1293 if (data_ptr)
1294 stktable_data_cast(data_ptr, std_t_sint) = data;
1295 break;
1296 }
1297 case STD_T_UINT: {
1298 unsigned int data;
1299
1300 data = intdecode(&msg_cur, msg_end);
1301 if (!msg_cur) {
1302 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001303 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001304 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001305 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1306 goto switchstate;
1307 }
1308
1309 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1310 if (data_ptr)
1311 stktable_data_cast(data_ptr, std_t_uint) = data;
1312 break;
1313 }
1314 case STD_T_ULL: {
1315 unsigned long long data;
1316
1317 data = intdecode(&msg_cur, msg_end);
1318 if (!msg_cur) {
1319 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001320 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001321 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001322 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1323 goto switchstate;
1324 }
1325
1326 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1327 if (data_ptr)
1328 stktable_data_cast(data_ptr, std_t_ull) = data;
1329 break;
1330 }
1331 case STD_T_FRQP: {
1332 struct freq_ctr_period data;
1333
Emeric Brunf2fc1fd2017-11-02 17:32:43 +01001334 /* First bit is reserved for the freq_ctr_period lock
1335 Note: here we're still protected by the stksess lock
1336 so we don't need to update the update the freq_ctr_period
1337 using its internal lock */
1338
1339 data.curr_tick = tick_add(now_ms, -intdecode(&msg_cur, msg_end)) & ~0x1;
Emeric Brun94900952015-06-11 18:25:54 +02001340 if (!msg_cur) {
1341 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001342 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001343 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001344 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1345 goto switchstate;
1346 }
1347 data.curr_ctr = intdecode(&msg_cur, msg_end);
1348 if (!msg_cur) {
1349 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001350 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001351 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001352 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1353 goto switchstate;
1354 }
1355 data.prev_ctr = intdecode(&msg_cur, msg_end);
1356 if (!msg_cur) {
1357 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001358 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001359 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001360 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1361 goto switchstate;
1362 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001363
Emeric Brun94900952015-06-11 18:25:54 +02001364 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1365 if (data_ptr)
1366 stktable_data_cast(data_ptr, std_t_frqp) = data;
1367 break;
1368 }
1369 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001370 }
1371 }
Emeric Brun819fc6f2017-06-13 19:37:32 +02001372
Emeric Brun55482912018-01-22 15:10:08 +01001373 /* Force new expiration */
1374 ts->expire = tick_add(now_ms, expire);
1375
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001376 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001377 stktable_touch_remote(st->table, ts, 1);
1378
Emeric Brunb3971ab2015-05-12 18:49:09 +02001379 }
1380 else if (msg_head[1] == PEER_MSG_STKT_ACK) {
1381 /* ack message */
1382 uint32_t table_id ;
1383 uint32_t update;
1384 struct shared_table *st;
1385
1386 table_id = intdecode(&msg_cur, msg_end);
1387 if (!msg_cur || (msg_cur + sizeof(update) > msg_end)) {
1388 /* malformed message */
1389 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1390 goto switchstate;
1391 }
1392 memcpy(&update, msg_cur, sizeof(update));
1393 update = ntohl(update);
Emeric Brun2b920a12010-09-23 18:30:22 +02001394
Emeric Brunb3971ab2015-05-12 18:49:09 +02001395 for (st = curpeer->tables; st; st = st->next) {
1396 if (st->local_id == table_id) {
1397 st->update = update;
1398 break;
1399 }
1400 }
1401 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001402 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001403 else if (msg_head[0] == PEER_MSG_CLASS_RESERVED) {
1404 appctx->st0 = PEER_SESS_ST_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +02001405 goto switchstate;
1406 }
1407
Emeric Brunb3971ab2015-05-12 18:49:09 +02001408ignore_msg:
Emeric Brun2b920a12010-09-23 18:30:22 +02001409 /* skip consumed message */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001410 co_skip(si_oc(si), totl);
Emeric Brun2b920a12010-09-23 18:30:22 +02001411 /* loop on that state to peek next message */
Willy Tarreau72d6c162013-04-11 16:14:13 +02001412 goto switchstate;
1413
Emeric Brun2b920a12010-09-23 18:30:22 +02001414incomplete:
Willy Tarreau06d80a92017-10-19 14:32:15 +02001415 /* we get here when a co_getblk() returns <= 0 in reql */
Willy Tarreau9d9179b2013-04-11 16:56:44 +02001416
Willy Tarreau72d6c162013-04-11 16:14:13 +02001417 if (reql < 0) {
1418 /* there was an error */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001419 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau72d6c162013-04-11 16:14:13 +02001420 goto switchstate;
1421 }
1422
Emeric Brun2b920a12010-09-23 18:30:22 +02001423
Emeric Brun2b920a12010-09-23 18:30:22 +02001424
Emeric Brunb3971ab2015-05-12 18:49:09 +02001425
Emeric Brun2b920a12010-09-23 18:30:22 +02001426 /* Need to request a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001427 if ((curpeer->flags & PEER_F_LEARN_ASSIGN) &&
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001428 (curpeers->flags & PEERS_F_RESYNC_ASSIGN) &&
1429 !(curpeers->flags & PEERS_F_RESYNC_PROCESS)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001430 unsigned char msg[2];
Emeric Brun2b920a12010-09-23 18:30:22 +02001431
Emeric Brunb3971ab2015-05-12 18:49:09 +02001432 /* Current peer was elected to request a resync */
1433 msg[0] = PEER_MSG_CLASS_CONTROL;
1434 msg[1] = PEER_MSG_CTRL_RESYNCREQ;
Emeric Brun2b920a12010-09-23 18:30:22 +02001435
Emeric Brunb3971ab2015-05-12 18:49:09 +02001436 /* message to buffer */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001437 repl = ci_putblk(si_ic(si), (char *)msg, sizeof(msg));
Emeric Brunb3971ab2015-05-12 18:49:09 +02001438 if (repl <= 0) {
1439 /* no more write possible */
1440 if (repl == -1)
1441 goto full;
1442 appctx->st0 = PEER_SESS_ST_END;
1443 goto switchstate;
1444 }
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001445 curpeers->flags |= PEERS_F_RESYNC_PROCESS;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001446 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001447
Emeric Brunb3971ab2015-05-12 18:49:09 +02001448 /* Nothing to read, now we start to write */
Emeric Brun2b920a12010-09-23 18:30:22 +02001449
Emeric Brunb3971ab2015-05-12 18:49:09 +02001450 if (curpeer->tables) {
1451 struct shared_table *st;
1452 struct shared_table *last_local_table;
Emeric Brun2b920a12010-09-23 18:30:22 +02001453
Emeric Brunb3971ab2015-05-12 18:49:09 +02001454 last_local_table = curpeer->last_local_table;
1455 if (!last_local_table)
1456 last_local_table = curpeer->tables;
1457 st = last_local_table->next;
Emeric Brun2b920a12010-09-23 18:30:22 +02001458
Emeric Brunb3971ab2015-05-12 18:49:09 +02001459 while (1) {
1460 if (!st)
1461 st = curpeer->tables;
Emeric Brun2b920a12010-09-23 18:30:22 +02001462
Emeric Brunb3971ab2015-05-12 18:49:09 +02001463 /* It remains some updates to ack */
1464 if (st->last_get != st->last_acked) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001465 int msglen;
Emeric Brun2b920a12010-09-23 18:30:22 +02001466
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001467 msglen = peer_prepare_ackmsg(st,
1468 trash.area,
1469 trash.size);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001470 if (!msglen) {
1471 /* internal error: message does not fit in trash */
1472 appctx->st0 = PEER_SESS_ST_END;
1473 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001474 }
1475
Emeric Brunb3971ab2015-05-12 18:49:09 +02001476 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001477 repl = ci_putblk(si_ic(si),
1478 trash.area,
1479 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001480 if (repl <= 0) {
1481 /* no more write possible */
1482 if (repl == -1) {
1483 goto full;
Emeric Brun2b920a12010-09-23 18:30:22 +02001484 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001485 appctx->st0 = PEER_SESS_ST_END;
1486 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001487 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001488 st->last_acked = st->last_get;
Emeric Brun2b920a12010-09-23 18:30:22 +02001489 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001490
Emeric Brunb3971ab2015-05-12 18:49:09 +02001491 if (!(curpeer->flags & PEER_F_TEACH_PROCESS)) {
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001492 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001493 if (!(curpeer->flags & PEER_F_LEARN_ASSIGN) &&
1494 ((int)(st->last_pushed - st->table->localupdate) < 0)) {
1495 struct eb32_node *eb;
1496 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001497
Emeric Brunb3971ab2015-05-12 18:49:09 +02001498 if (st != curpeer->last_local_table) {
1499 int msglen;
Emeric Brun2b920a12010-09-23 18:30:22 +02001500
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001501 msglen = peer_prepare_switchmsg(st,
1502 trash.area,
1503 trash.size);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001504 if (!msglen) {
Emeric Brun088c9b72017-12-01 11:37:36 +01001505 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001506 /* internal error: message does not fit in trash */
1507 appctx->st0 = PEER_SESS_ST_END;
1508 goto switchstate;
1509 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001510
Emeric Brunb3971ab2015-05-12 18:49:09 +02001511 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001512 repl = ci_putblk(si_ic(si),
1513 trash.area,
1514 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001515 if (repl <= 0) {
Emeric Brun088c9b72017-12-01 11:37:36 +01001516 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001517 /* no more write possible */
1518 if (repl == -1) {
1519 goto full;
1520 }
1521 appctx->st0 = PEER_SESS_ST_END;
1522 goto switchstate;
1523 }
1524 curpeer->last_local_table = st;
Emeric Brun2b920a12010-09-23 18:30:22 +02001525 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001526
1527 /* We force new pushed to 1 to force identifier in update message */
1528 new_pushed = 1;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001529 while (1) {
1530 uint32_t msglen;
1531 struct stksess *ts;
Emeric Brun819fc6f2017-06-13 19:37:32 +02001532 unsigned updateid;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001533
1534 /* push local updates */
Emeric Brun819fc6f2017-06-13 19:37:32 +02001535 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001536 if (!eb) {
1537 eb = eb32_first(&st->table->updates);
1538 if (!eb || ((int)(eb->key - st->last_pushed) <= 0)) {
Emeric Brunaaf58602015-06-15 17:23:30 +02001539 st->table->commitupdate = st->last_pushed = st->table->localupdate;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001540 break;
1541 }
1542 }
1543
1544 if ((int)(eb->key - st->table->localupdate) > 0) {
Emeric Brunaaf58602015-06-15 17:23:30 +02001545 st->table->commitupdate = st->last_pushed = st->table->localupdate;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001546 break;
1547 }
1548
1549 ts = eb32_entry(eb, struct stksess, upd);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001550 updateid = ts->upd.key;
1551 ts->ref_cnt++;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001552 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001553
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001554 msglen = peer_prepare_updatemsg(ts, st, updateid,
1555 trash.area,
1556 trash.size,
1557 new_pushed,
1558 0);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001559 if (!msglen) {
1560 /* internal error: message does not fit in trash */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001561 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001562 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001563 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001564 appctx->st0 = PEER_SESS_ST_END;
1565 goto switchstate;
1566 }
1567
1568 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001569 repl = ci_putblk(si_ic(si),
1570 trash.area,
1571 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001572 if (repl <= 0) {
1573 /* no more write possible */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001574 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001575 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001576 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001577 if (repl == -1) {
1578 goto full;
1579 }
1580 appctx->st0 = PEER_SESS_ST_END;
1581 goto switchstate;
1582 }
Emeric Brun819fc6f2017-06-13 19:37:32 +02001583
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001584 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001585 ts->ref_cnt--;
1586 st->last_pushed = updateid;
Emeric Brunaaf58602015-06-15 17:23:30 +02001587 if ((int)(st->last_pushed - st->table->commitupdate) > 0)
1588 st->table->commitupdate = st->last_pushed;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001589 /* identifier may not needed in next update message */
1590 new_pushed = 0;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001591 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001592 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001593 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun2b920a12010-09-23 18:30:22 +02001594 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001595 else {
1596 if (!(st->flags & SHTABLE_F_TEACH_STAGE1)) {
1597 struct eb32_node *eb;
1598 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001599
Emeric Brunb3971ab2015-05-12 18:49:09 +02001600 if (st != curpeer->last_local_table) {
1601 int msglen;
1602
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001603 msglen = peer_prepare_switchmsg(st,
1604 trash.area,
1605 trash.size);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001606 if (!msglen) {
1607 /* internal error: message does not fit in trash */
1608 appctx->st0 = PEER_SESS_ST_END;
1609 goto switchstate;
1610 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001611
Emeric Brunb3971ab2015-05-12 18:49:09 +02001612 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001613 repl = ci_putblk(si_ic(si),
1614 trash.area,
1615 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001616 if (repl <= 0) {
1617 /* no more write possible */
1618 if (repl == -1) {
1619 goto full;
1620 }
1621 appctx->st0 = PEER_SESS_ST_END;
1622 goto switchstate;
1623 }
1624 curpeer->last_local_table = st;
1625 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001626
Emeric Brunb3971ab2015-05-12 18:49:09 +02001627 /* We force new pushed to 1 to force identifier in update message */
1628 new_pushed = 1;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001629 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001630 while (1) {
1631 uint32_t msglen;
1632 struct stksess *ts;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001633 int use_timed;
Emeric Brun819fc6f2017-06-13 19:37:32 +02001634 unsigned updateid;
Emeric Brun2b920a12010-09-23 18:30:22 +02001635
Emeric Brunb3971ab2015-05-12 18:49:09 +02001636 /* push local updates */
Emeric Brun819fc6f2017-06-13 19:37:32 +02001637 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001638 if (!eb) {
1639 st->flags |= SHTABLE_F_TEACH_STAGE1;
1640 eb = eb32_first(&st->table->updates);
1641 if (eb)
1642 st->last_pushed = eb->key - 1;
1643 break;
1644 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001645
Emeric Brunb3971ab2015-05-12 18:49:09 +02001646 ts = eb32_entry(eb, struct stksess, upd);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001647 updateid = ts->upd.key;
1648 ts->ref_cnt++;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001649 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001650
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001651 use_timed = !(curpeer->flags & PEER_F_DWNGRD);
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001652 msglen = peer_prepare_updatemsg(ts, st, updateid,
1653 trash.area,
1654 trash.size,
1655 new_pushed,
1656 use_timed);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001657 if (!msglen) {
1658 /* internal error: message does not fit in trash */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001659 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001660 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001661 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001662 appctx->st0 = PEER_SESS_ST_END;
1663 goto switchstate;
1664 }
1665
1666 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001667 repl = ci_putblk(si_ic(si),
1668 trash.area,
1669 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001670 if (repl <= 0) {
1671 /* no more write possible */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001672 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001673 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001674 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001675 if (repl == -1) {
1676 goto full;
1677 }
1678 appctx->st0 = PEER_SESS_ST_END;
1679 goto switchstate;
1680 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001681 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001682 ts->ref_cnt--;
1683 st->last_pushed = updateid;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001684 /* identifier may not needed in next update message */
1685 new_pushed = 0;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001686 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001687 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun2b920a12010-09-23 18:30:22 +02001688 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001689
Emeric Brunb3971ab2015-05-12 18:49:09 +02001690 if (!(st->flags & SHTABLE_F_TEACH_STAGE2)) {
1691 struct eb32_node *eb;
1692 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001693
Emeric Brunb3971ab2015-05-12 18:49:09 +02001694 if (st != curpeer->last_local_table) {
1695 int msglen;
1696
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001697 msglen = peer_prepare_switchmsg(st,
1698 trash.area,
1699 trash.size);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001700 if (!msglen) {
1701 /* internal error: message does not fit in trash */
1702 appctx->st0 = PEER_SESS_ST_END;
1703 goto switchstate;
1704 }
1705
1706 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001707 repl = ci_putblk(si_ic(si),
1708 trash.area,
1709 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001710 if (repl <= 0) {
1711 /* no more write possible */
1712 if (repl == -1) {
1713 goto full;
1714 }
1715 appctx->st0 = PEER_SESS_ST_END;
1716 goto switchstate;
1717 }
1718 curpeer->last_local_table = st;
1719 }
1720
1721 /* We force new pushed to 1 to force identifier in update message */
1722 new_pushed = 1;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001723 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001724 while (1) {
1725 uint32_t msglen;
1726 struct stksess *ts;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001727 int use_timed;
Emeric Brun819fc6f2017-06-13 19:37:32 +02001728 unsigned updateid;
1729
1730 /* push local updates */
1731 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001732
1733 /* push local updates */
1734 if (!eb || eb->key > st->teaching_origin) {
1735 st->flags |= SHTABLE_F_TEACH_STAGE2;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001736 break;
1737 }
1738
1739 ts = eb32_entry(eb, struct stksess, upd);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001740 updateid = ts->upd.key;
1741 ts->ref_cnt++;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001742 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001743
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001744 use_timed = !(curpeer->flags & PEER_F_DWNGRD);
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001745 msglen = peer_prepare_updatemsg(ts, st, updateid,
1746 trash.area,
1747 trash.size,
1748 new_pushed,
1749 use_timed);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001750 if (!msglen) {
1751 /* internal error: message does not fit in trash */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001752 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001753 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001754 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001755 appctx->st0 = PEER_SESS_ST_END;
1756 goto switchstate;
1757 }
1758
1759 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001760 repl = ci_putblk(si_ic(si),
1761 trash.area,
1762 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001763 if (repl <= 0) {
1764 /* no more write possible */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001765 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001766 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001767 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001768 if (repl == -1) {
1769 goto full;
1770 }
1771 appctx->st0 = PEER_SESS_ST_END;
1772 goto switchstate;
1773 }
Emeric Brun819fc6f2017-06-13 19:37:32 +02001774
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001775 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001776 ts->ref_cnt--;
1777 st->last_pushed = updateid;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001778 /* identifier may not needed in next update message */
1779 new_pushed = 0;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001780 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001781 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun2b920a12010-09-23 18:30:22 +02001782 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001783 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001784
1785 if (st == last_local_table)
1786 break;
1787 st = st->next;
Emeric Brun2b920a12010-09-23 18:30:22 +02001788 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001789 }
1790
1791
1792 if ((curpeer->flags & PEER_F_TEACH_PROCESS) && !(curpeer->flags & PEER_F_TEACH_FINISHED)) {
1793 unsigned char msg[2];
1794
1795 /* Current peer was elected to request a resync */
1796 msg[0] = PEER_MSG_CLASS_CONTROL;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001797 msg[1] = ((curpeers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FINISHED) ? PEER_MSG_CTRL_RESYNCFINISHED : PEER_MSG_CTRL_RESYNCPARTIAL;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001798 /* process final lesson message */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001799 repl = ci_putblk(si_ic(si), (char *)msg, sizeof(msg));
Emeric Brunb3971ab2015-05-12 18:49:09 +02001800 if (repl <= 0) {
1801 /* no more write possible */
1802 if (repl == -1)
1803 goto full;
1804 appctx->st0 = PEER_SESS_ST_END;
1805 goto switchstate;
1806 }
1807 /* flag finished message sent */
1808 curpeer->flags |= PEER_F_TEACH_FINISHED;
1809 }
1810
Emeric Brun597b26e2016-08-12 11:23:31 +02001811 /* Confirm finished or partial messages */
1812 while (curpeer->confirm) {
1813 unsigned char msg[2];
1814
1815 /* There is a confirm messages to send */
1816 msg[0] = PEER_MSG_CLASS_CONTROL;
1817 msg[1] = PEER_MSG_CTRL_RESYNCCONFIRM;
1818
1819 /* message to buffer */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001820 repl = ci_putblk(si_ic(si), (char *)msg, sizeof(msg));
Emeric Brun597b26e2016-08-12 11:23:31 +02001821 if (repl <= 0) {
1822 /* no more write possible */
1823 if (repl == -1)
1824 goto full;
1825 appctx->st0 = PEER_SESS_ST_END;
1826 goto switchstate;
1827 }
1828 curpeer->confirm--;
1829 }
1830
Emeric Brun2b920a12010-09-23 18:30:22 +02001831 /* noting more to do */
1832 goto out;
1833 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001834 case PEER_SESS_ST_EXIT:
Willy Tarreau2d372c22018-11-05 17:12:27 +01001835 if (prev_state == PEER_SESS_ST_WAITMSG)
1836 HA_ATOMIC_SUB(&connected_peers, 1);
1837 prev_state = appctx->st0;
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001838 repl = snprintf(trash.area, trash.size,
1839 "%d\n", appctx->st1);
1840 if (ci_putblk(si_ic(si), trash.area, repl) == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +01001841 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001842 appctx->st0 = PEER_SESS_ST_END;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001843 goto switchstate;
1844 case PEER_SESS_ST_ERRSIZE: {
1845 unsigned char msg[2];
1846
Willy Tarreau2d372c22018-11-05 17:12:27 +01001847 if (prev_state == PEER_SESS_ST_WAITMSG)
1848 HA_ATOMIC_SUB(&connected_peers, 1);
1849 prev_state = appctx->st0;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001850 msg[0] = PEER_MSG_CLASS_ERROR;
1851 msg[1] = PEER_MSG_ERR_SIZELIMIT;
1852
Willy Tarreau06d80a92017-10-19 14:32:15 +02001853 if (ci_putblk(si_ic(si), (char *)msg, sizeof(msg)) == -1)
Emeric Brunb3971ab2015-05-12 18:49:09 +02001854 goto full;
1855 appctx->st0 = PEER_SESS_ST_END;
1856 goto switchstate;
1857 }
1858 case PEER_SESS_ST_ERRPROTO: {
1859 unsigned char msg[2];
1860
Willy Tarreau2d372c22018-11-05 17:12:27 +01001861 if (prev_state == PEER_SESS_ST_WAITMSG)
1862 HA_ATOMIC_SUB(&connected_peers, 1);
1863 prev_state = appctx->st0;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001864 msg[0] = PEER_MSG_CLASS_ERROR;
1865 msg[1] = PEER_MSG_ERR_PROTOCOL;
1866
Willy Tarreau06d80a92017-10-19 14:32:15 +02001867 if (ci_putblk(si_ic(si), (char *)msg, sizeof(msg)) == -1)
Emeric Brunb3971ab2015-05-12 18:49:09 +02001868 goto full;
1869 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau2d372c22018-11-05 17:12:27 +01001870 prev_state = appctx->st0;
Emeric Brun2b920a12010-09-23 18:30:22 +02001871 /* fall through */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001872 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001873 case PEER_SESS_ST_END: {
Willy Tarreau2d372c22018-11-05 17:12:27 +01001874 if (prev_state == PEER_SESS_ST_WAITMSG)
1875 HA_ATOMIC_SUB(&connected_peers, 1);
1876 prev_state = appctx->st0;
Emeric Brun80527f52017-06-19 17:46:37 +02001877 if (curpeer) {
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001878 HA_SPIN_UNLOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02001879 curpeer = NULL;
1880 }
Willy Tarreau73b013b2012-05-21 16:31:45 +02001881 si_shutw(si);
1882 si_shutr(si);
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001883 si_ic(si)->flags |= CF_READ_NULL;
Willy Tarreau828824a2015-04-19 17:20:03 +02001884 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001885 }
1886 }
1887 }
1888out:
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001889 si_oc(si)->flags |= CF_READ_DONTWAIT;
Emeric Brun80527f52017-06-19 17:46:37 +02001890
1891 if (curpeer)
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001892 HA_SPIN_UNLOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun2b920a12010-09-23 18:30:22 +02001893 return;
Willy Tarreaubc18da12015-03-13 14:00:47 +01001894full:
Willy Tarreau0cd3bd62018-11-06 18:46:37 +01001895 si_cant_put(si);
Willy Tarreaubc18da12015-03-13 14:00:47 +01001896 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001897}
1898
Willy Tarreau30576452015-04-13 13:50:30 +02001899static struct applet peer_applet = {
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001900 .obj_type = OBJ_TYPE_APPLET,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001901 .name = "<PEER>", /* used for logging */
1902 .fct = peer_io_handler,
Aman Gupta9a13e842012-04-02 18:57:53 -07001903 .release = peer_session_release,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001904};
Emeric Brun2b920a12010-09-23 18:30:22 +02001905
1906/*
1907 * Use this function to force a close of a peer session
1908 */
Willy Tarreau81bc3b02016-10-31 17:37:39 +01001909static void peer_session_forceshutdown(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +02001910{
Frédéric Lécaille5df11902017-06-13 16:39:57 +02001911 /* Note that the peer sessions which have just been created
1912 * (->st0 == PEER_SESS_ST_CONNECT) must not
1913 * be shutdown, if not, the TCP session will never be closed
1914 * and stay in CLOSE_WAIT state after having been closed by
1915 * the remote side.
1916 */
1917 if (!appctx || appctx->st0 == PEER_SESS_ST_CONNECT)
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001918 return;
1919
Willy Tarreau81bc3b02016-10-31 17:37:39 +01001920 if (appctx->applet != &peer_applet)
1921 return;
1922
Willy Tarreau2d372c22018-11-05 17:12:27 +01001923 if (appctx->st0 == PEER_SESS_ST_WAITMSG)
1924 HA_ATOMIC_SUB(&connected_peers, 1);
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001925 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau78c0c502016-10-31 17:32:20 +01001926 appctx_wakeup(appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02001927}
1928
Willy Tarreau91d96282015-03-13 15:47:26 +01001929/* Pre-configures a peers frontend to accept incoming connections */
1930void peers_setup_frontend(struct proxy *fe)
1931{
1932 fe->last_change = now.tv_sec;
1933 fe->cap = PR_CAP_FE;
1934 fe->maxconn = 0;
1935 fe->conn_retries = CONN_RETRIES;
1936 fe->timeout.client = MS_TO_TICKS(5000);
Willy Tarreaud1d48d42015-03-13 16:15:46 +01001937 fe->accept = frontend_accept;
Willy Tarreauf87ab942015-03-13 15:55:16 +01001938 fe->default_target = &peer_applet.obj_type;
Willy Tarreau91d96282015-03-13 15:47:26 +01001939 fe->options2 |= PR_O2_INDEPSTR | PR_O2_SMARTCON | PR_O2_SMARTACC;
Willy Tarreau0fca4832015-05-01 19:12:05 +02001940 fe->bind_proc = 0; /* will be filled by users */
Willy Tarreau91d96282015-03-13 15:47:26 +01001941}
1942
Emeric Brun2b920a12010-09-23 18:30:22 +02001943/*
Willy Tarreaubd55e312010-11-11 10:55:09 +01001944 * Create a new peer session in assigned state (connect will start automatically)
Emeric Brun2b920a12010-09-23 18:30:22 +02001945 */
Willy Tarreau9df94c22016-10-31 18:42:52 +01001946static struct appctx *peer_session_create(struct peers *peers, struct peer *peer)
Emeric Brun2b920a12010-09-23 18:30:22 +02001947{
Willy Tarreau04b92862017-09-15 11:01:04 +02001948 struct proxy *p = peers->peers_fe; /* attached frontend */
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001949 struct appctx *appctx;
Willy Tarreau15b5e142015-04-04 14:38:25 +02001950 struct session *sess;
Willy Tarreau87b09662015-04-03 00:22:06 +02001951 struct stream *s;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001952 struct connection *conn;
Olivier Houchard9aaf7782017-09-13 18:30:23 +02001953 struct conn_stream *cs;
Emeric Brun2b920a12010-09-23 18:30:22 +02001954
Emeric Brunb3971ab2015-05-12 18:49:09 +02001955 peer->reconnect = tick_add(now_ms, MS_TO_TICKS(5000));
1956 peer->statuscode = PEER_SESS_SC_CONNECTCODE;
Willy Tarreaud990baf2015-04-05 00:32:03 +02001957 s = NULL;
1958
Emeric Brun1138fd02017-06-19 12:38:55 +02001959 appctx = appctx_new(&peer_applet, tid_bit);
Willy Tarreaud990baf2015-04-05 00:32:03 +02001960 if (!appctx)
1961 goto out_close;
1962
1963 appctx->st0 = PEER_SESS_ST_CONNECT;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001964 appctx->ctx.peers.ptr = (void *)peer;
Willy Tarreaud990baf2015-04-05 00:32:03 +02001965
Willy Tarreau04b92862017-09-15 11:01:04 +02001966 sess = session_new(p, NULL, &appctx->obj_type);
Willy Tarreau15b5e142015-04-04 14:38:25 +02001967 if (!sess) {
Christopher Faulet767a84b2017-11-24 16:50:31 +01001968 ha_alert("out of memory in peer_session_create().\n");
Willy Tarreaud990baf2015-04-05 00:32:03 +02001969 goto out_free_appctx;
Emeric Brun2b920a12010-09-23 18:30:22 +02001970 }
1971
Willy Tarreau87787ac2017-08-28 16:22:54 +02001972 if ((s = stream_new(sess, &appctx->obj_type)) == NULL) {
Christopher Faulet767a84b2017-11-24 16:50:31 +01001973 ha_alert("Failed to initialize stream in peer_session_create().\n");
Willy Tarreau87787ac2017-08-28 16:22:54 +02001974 goto out_free_sess;
Willy Tarreau8baf9062015-04-05 00:46:36 +02001975 }
1976
Willy Tarreau342bfb12015-04-05 01:35:34 +02001977 /* The tasks below are normally what is supposed to be done by
1978 * fe->accept().
1979 */
Willy Tarreaue7dff022015-04-03 01:14:29 +02001980 s->flags = SF_ASSIGNED|SF_ADDR_SET;
Emeric Brun2b920a12010-09-23 18:30:22 +02001981
Willy Tarreau6e2979c2015-04-27 13:21:15 +02001982 /* applet is waiting for data */
Willy Tarreau0cd3bd62018-11-06 18:46:37 +01001983 si_cant_get(&s->si[0]);
Willy Tarreau6e2979c2015-04-27 13:21:15 +02001984 appctx_wakeup(appctx);
1985
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001986 /* initiate an outgoing connection */
Willy Tarreaudbd02672017-12-06 17:39:53 +01001987 s->si[1].flags |= SI_FL_NOLINGER;
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001988 si_set_state(&s->si[1], SI_ST_ASS);
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001989
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001990 /* automatically prepare the stream interface to connect to the
Willy Tarreaub363a1f2013-10-01 10:45:07 +02001991 * pre-initialized connection in si->conn.
1992 */
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001993 if (unlikely((conn = conn_new()) == NULL))
Willy Tarreau8baf9062015-04-05 00:46:36 +02001994 goto out_free_strm;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001995
Olivier Houchard9aaf7782017-09-13 18:30:23 +02001996 if (unlikely((cs = cs_new(conn)) == NULL))
1997 goto out_free_conn;
1998
Willy Tarreaube373152018-09-06 11:45:30 +02001999 conn->target = s->target = &s->be->obj_type;
2000 memcpy(&conn->addr.to, &peer->addr, sizeof(conn->addr.to));
2001
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02002002 conn_prepare(conn, peer->proto, peer->xprt);
Willy Tarreau175a2bb2018-09-12 12:02:05 +02002003 conn_install_mux(conn, &mux_pt_ops, cs, s->be);
Olivier Houchard9aaf7782017-09-13 18:30:23 +02002004 si_attach_cs(&s->si[1], cs);
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02002005
Emeric Brun2b920a12010-09-23 18:30:22 +02002006 s->do_log = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02002007 s->uniq_id = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +02002008
Willy Tarreau22ec1ea2014-11-27 20:45:39 +01002009 s->res.flags |= CF_READ_DONTWAIT;
Willy Tarreau696a2912014-11-24 11:36:57 +01002010
Emeric Brunb3971ab2015-05-12 18:49:09 +02002011 peer->appctx = appctx;
Willy Tarreau87787ac2017-08-28 16:22:54 +02002012 task_wakeup(s->task, TASK_WOKEN_INIT);
Willy Tarreau199ad242018-11-05 16:31:22 +01002013 HA_ATOMIC_ADD(&active_peers, 1);
Willy Tarreau9df94c22016-10-31 18:42:52 +01002014 return appctx;
Emeric Brun2b920a12010-09-23 18:30:22 +02002015
2016 /* Error unrolling */
Olivier Houchard9aaf7782017-09-13 18:30:23 +02002017 out_free_conn:
2018 conn_free(conn);
Willy Tarreau15b5e142015-04-04 14:38:25 +02002019 out_free_strm:
Emeric Brun2b920a12010-09-23 18:30:22 +02002020 LIST_DEL(&s->list);
Willy Tarreaubafbe012017-11-24 17:34:44 +01002021 pool_free(pool_head_stream, s);
Willy Tarreau15b5e142015-04-04 14:38:25 +02002022 out_free_sess:
Willy Tarreau11c36242015-04-04 15:54:03 +02002023 session_free(sess);
Willy Tarreaud990baf2015-04-05 00:32:03 +02002024 out_free_appctx:
2025 appctx_free(appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02002026 out_close:
Willy Tarreaub21d08e2016-10-31 17:46:57 +01002027 return NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02002028}
2029
2030/*
2031 * Task processing function to manage re-connect and peer session
2032 * tasks wakeup on local update.
2033 */
Olivier Houchard9f6af332018-05-25 14:04:04 +02002034static struct task *process_peer_sync(struct task * task, void *context, unsigned short state)
Emeric Brun2b920a12010-09-23 18:30:22 +02002035{
Olivier Houchard9f6af332018-05-25 14:04:04 +02002036 struct peers *peers = context;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002037 struct peer *ps;
2038 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +02002039
2040 task->expire = TICK_ETERNITY;
2041
Emeric Brunb3971ab2015-05-12 18:49:09 +02002042 if (!peers->peers_fe) {
Willy Tarreau46dc1ca2015-05-01 18:32:13 +02002043 /* this one was never started, kill it */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002044 signal_unregister_handler(peers->sighandler);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002045 task_delete(peers->sync_task);
2046 task_free(peers->sync_task);
Willy Tarreau37bb7be2015-09-21 15:24:58 +02002047 peers->sync_task = NULL;
Willy Tarreau46dc1ca2015-05-01 18:32:13 +02002048 return NULL;
2049 }
2050
Emeric Brun80527f52017-06-19 17:46:37 +02002051 /* Acquire lock for all peers of the section */
2052 for (ps = peers->remote; ps; ps = ps->next)
Christopher Faulet2a944ee2017-11-07 10:42:54 +01002053 HA_SPIN_LOCK(PEER_LOCK, &ps->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02002054
Emeric Brun2b920a12010-09-23 18:30:22 +02002055 if (!stopping) {
2056 /* Normal case (not soft stop)*/
Emeric Brunb3971ab2015-05-12 18:49:09 +02002057
2058 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL) &&
2059 (!nb_oldpids || tick_is_expired(peers->resync_timeout, now_ms)) &&
2060 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002061 /* Resync from local peer needed
2062 no peer was assigned for the lesson
2063 and no old local peer found
2064 or resync timeout expire */
2065
2066 /* flag no more resync from local, to try resync from remotes */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002067 peers->flags |= PEERS_F_RESYNC_LOCAL;
Emeric Brun2b920a12010-09-23 18:30:22 +02002068
2069 /* reschedule a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002070 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
Emeric Brun2b920a12010-09-23 18:30:22 +02002071 }
2072
2073 /* For each session */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002074 for (ps = peers->remote; ps; ps = ps->next) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002075 /* For each remote peers */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002076 if (!ps->local) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002077 if (!ps->appctx) {
2078 /* no active peer connection */
Emeric Brun2b920a12010-09-23 18:30:22 +02002079 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002080 ((ps->statuscode == PEER_SESS_SC_CONNECTCODE ||
Willy Tarreaub4e34da2015-05-20 10:39:04 +02002081 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002082 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02002083 tick_is_expired(ps->reconnect, now_ms))) {
2084 /* connection never tried
Willy Tarreau9df94c22016-10-31 18:42:52 +01002085 * or previous peer connection established with success
2086 * or previous peer connection failed while connecting
Emeric Brun2b920a12010-09-23 18:30:22 +02002087 * and reconnection timer is expired */
2088
2089 /* retry a connect */
Willy Tarreau9df94c22016-10-31 18:42:52 +01002090 ps->appctx = peer_session_create(peers, ps);
Emeric Brun2b920a12010-09-23 18:30:22 +02002091 }
Willy Tarreaub4e34da2015-05-20 10:39:04 +02002092 else if (!tick_is_expired(ps->reconnect, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002093 /* If previous session failed during connection
2094 * but reconnection timer is not expired */
2095
2096 /* reschedule task for reconnect */
2097 task->expire = tick_first(task->expire, ps->reconnect);
2098 }
2099 /* else do nothing */
Willy Tarreau9df94c22016-10-31 18:42:52 +01002100 } /* !ps->appctx */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002101 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002102 /* current peer connection is active and established */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002103 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
2104 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02002105 !(ps->flags & PEER_F_LEARN_NOTUP2DATE)) {
2106 /* Resync from a remote is needed
2107 * and no peer was assigned for lesson
2108 * and current peer may be up2date */
2109
2110 /* assign peer for the lesson */
2111 ps->flags |= PEER_F_LEARN_ASSIGN;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002112 peers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brun2b920a12010-09-23 18:30:22 +02002113
Willy Tarreau9df94c22016-10-31 18:42:52 +01002114 /* wake up peer handler to handle a request of resync */
Willy Tarreaue5843b32015-04-27 18:40:14 +02002115 appctx_wakeup(ps->appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02002116 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02002117 else {
2118 /* Awake session if there is data to push */
2119 for (st = ps->tables; st ; st = st->next) {
2120 if ((int)(st->last_pushed - st->table->localupdate) < 0) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002121 /* wake up the peer handler to push local updates */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002122 appctx_wakeup(ps->appctx);
2123 break;
2124 }
2125 }
Emeric Brun2b920a12010-09-23 18:30:22 +02002126 }
2127 /* else do nothing */
2128 } /* SUCCESSCODE */
2129 } /* !ps->peer->local */
2130 } /* for */
2131
2132 /* Resync from remotes expired: consider resync is finished */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002133 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
2134 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
2135 tick_is_expired(peers->resync_timeout, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002136 /* Resync from remote peer needed
2137 * no peer was assigned for the lesson
2138 * and resync timeout expire */
2139
2140 /* flag no more resync from remote, consider resync is finished */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002141 peers->flags |= PEERS_F_RESYNC_REMOTE;
Emeric Brun2b920a12010-09-23 18:30:22 +02002142 }
2143
Emeric Brunb3971ab2015-05-12 18:49:09 +02002144 if ((peers->flags & PEERS_RESYNC_STATEMASK) != PEERS_RESYNC_FINISHED) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002145 /* Resync not finished*/
Frédéric Lécaille5d6e5f82017-05-29 13:47:16 +02002146 /* reschedule task to resync timeout if not expired, to ended resync if needed */
2147 if (!tick_is_expired(peers->resync_timeout, now_ms))
2148 task->expire = tick_first(task->expire, peers->resync_timeout);
Emeric Brun2b920a12010-09-23 18:30:22 +02002149 }
2150 } /* !stopping */
2151 else {
2152 /* soft stop case */
Willy Tarreau086735a2018-11-05 15:09:47 +01002153 if (state & TASK_WOKEN_SIGNAL) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002154 /* We've just recieved the signal */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002155 if (!(peers->flags & PEERS_F_DONOTSTOP)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002156 /* add DO NOT STOP flag if not present */
Christopher Faulet8d8aa0d2017-05-30 15:36:50 +02002157 HA_ATOMIC_ADD(&jobs, 1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002158 peers->flags |= PEERS_F_DONOTSTOP;
2159 ps = peers->local;
2160 for (st = ps->tables; st ; st = st->next)
2161 st->table->syncing++;
Emeric Brun2b920a12010-09-23 18:30:22 +02002162 }
2163
2164 /* disconnect all connected peers */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002165 for (ps = peers->remote; ps; ps = ps->next) {
Emeric Brun80527f52017-06-19 17:46:37 +02002166 /* we're killing a connection, we must apply a random delay before
2167 * retrying otherwise the other end will do the same and we can loop
2168 * for a while.
2169 */
2170 ps->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + random() % 2000));
Willy Tarreau9df94c22016-10-31 18:42:52 +01002171 if (ps->appctx) {
Willy Tarreau81bc3b02016-10-31 17:37:39 +01002172 peer_session_forceshutdown(ps->appctx);
Willy Tarreaue5843b32015-04-27 18:40:14 +02002173 ps->appctx = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02002174 }
2175 }
2176 }
Emeric Brun2b920a12010-09-23 18:30:22 +02002177
Emeric Brunb3971ab2015-05-12 18:49:09 +02002178 ps = peers->local;
Emeric Brun2b920a12010-09-23 18:30:22 +02002179 if (ps->flags & PEER_F_TEACH_COMPLETE) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02002180 if (peers->flags & PEERS_F_DONOTSTOP) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002181 /* resync of new process was complete, current process can die now */
Willy Tarreaucea85372017-11-29 14:49:30 +01002182 HA_ATOMIC_SUB(&jobs, 1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002183 peers->flags &= ~PEERS_F_DONOTSTOP;
2184 for (st = ps->tables; st ; st = st->next)
2185 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02002186 }
2187 }
Willy Tarreau9df94c22016-10-31 18:42:52 +01002188 else if (!ps->appctx) {
2189 /* If there's no active peer connection */
Emeric Brun2b920a12010-09-23 18:30:22 +02002190 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002191 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
2192 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE ||
2193 ps->statuscode == PEER_SESS_SC_TRYAGAIN) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002194 /* connection never tried
Willy Tarreau9df94c22016-10-31 18:42:52 +01002195 * or previous peer connection was successfully established
2196 * or previous tcp connect succeeded but init state incomplete
Emeric Brun2b920a12010-09-23 18:30:22 +02002197 * or during previous connect, peer replies a try again statuscode */
2198
2199 /* connect to the peer */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002200 peer_session_create(peers, ps);
Emeric Brun2b920a12010-09-23 18:30:22 +02002201 }
2202 else {
2203 /* Other error cases */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002204 if (peers->flags & PEERS_F_DONOTSTOP) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002205 /* unable to resync new process, current process can die now */
Christopher Faulet8d8aa0d2017-05-30 15:36:50 +02002206 HA_ATOMIC_SUB(&jobs, 1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002207 peers->flags &= ~PEERS_F_DONOTSTOP;
2208 for (st = ps->tables; st ; st = st->next)
2209 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02002210 }
2211 }
2212 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02002213 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE ) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002214 /* current peer connection is active and established
2215 * wake up all peer handlers to push remaining local updates */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002216 for (st = ps->tables; st ; st = st->next) {
2217 if ((int)(st->last_pushed - st->table->localupdate) < 0) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02002218 appctx_wakeup(ps->appctx);
2219 break;
2220 }
2221 }
Emeric Brun2b920a12010-09-23 18:30:22 +02002222 }
2223 } /* stopping */
Emeric Brun80527f52017-06-19 17:46:37 +02002224
2225 /* Release lock for all peers of the section */
2226 for (ps = peers->remote; ps; ps = ps->next)
Christopher Faulet2a944ee2017-11-07 10:42:54 +01002227 HA_SPIN_UNLOCK(PEER_LOCK, &ps->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02002228
Emeric Brun2b920a12010-09-23 18:30:22 +02002229 /* Wakeup for re-connect */
2230 return task;
2231}
2232
Emeric Brunb3971ab2015-05-12 18:49:09 +02002233
Emeric Brun2b920a12010-09-23 18:30:22 +02002234/*
Willy Tarreaud9443442018-10-15 11:18:03 +02002235 * returns 0 in case of error.
Emeric Brun2b920a12010-09-23 18:30:22 +02002236 */
Willy Tarreaud9443442018-10-15 11:18:03 +02002237int peers_init_sync(struct peers *peers)
Emeric Brun2b920a12010-09-23 18:30:22 +02002238{
Emeric Brun2b920a12010-09-23 18:30:22 +02002239 struct peer * curpeer;
Willy Tarreau4348fad2012-09-20 16:48:07 +02002240 struct listener *listener;
Emeric Brun2b920a12010-09-23 18:30:22 +02002241
Emeric Brun2b920a12010-09-23 18:30:22 +02002242 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002243 peers->peers_fe->maxconn += 3;
2244 }
2245
Willy Tarreau4348fad2012-09-20 16:48:07 +02002246 list_for_each_entry(listener, &peers->peers_fe->conf.listeners, by_fe)
2247 listener->maxconn = peers->peers_fe->maxconn;
Emeric Brunc60def82017-09-27 14:59:38 +02002248 peers->sync_task = task_new(MAX_THREADS_MASK);
Willy Tarreaud9443442018-10-15 11:18:03 +02002249 if (!peers->sync_task)
2250 return 0;
2251
Emeric Brunb3971ab2015-05-12 18:49:09 +02002252 peers->sync_task->process = process_peer_sync;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002253 peers->sync_task->context = (void *)peers;
2254 peers->sighandler = signal_register_task(0, peers->sync_task, 0);
2255 task_wakeup(peers->sync_task, TASK_WOKEN_INIT);
Willy Tarreaud9443442018-10-15 11:18:03 +02002256 return 1;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002257}
2258
2259
2260
2261/*
2262 * Function used to register a table for sync on a group of peers
2263 *
2264 */
2265void peers_register_table(struct peers *peers, struct stktable *table)
2266{
2267 struct shared_table *st;
2268 struct peer * curpeer;
2269 int id = 0;
2270
2271 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
Vincent Bernat02779b62016-04-03 13:48:43 +02002272 st = calloc(1,sizeof(*st));
Emeric Brunb3971ab2015-05-12 18:49:09 +02002273 st->table = table;
2274 st->next = curpeer->tables;
2275 if (curpeer->tables)
2276 id = curpeer->tables->local_id;
2277 st->local_id = id + 1;
2278
2279 curpeer->tables = st;
2280 }
2281
2282 table->sync_task = peers->sync_task;
Emeric Brun2b920a12010-09-23 18:30:22 +02002283}
2284