blob: 64d5e085d90df4baeac2008f44d82964d824691d [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) {
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100509 HA_SPIN_LOCK(PEER_LOCK, &peer->lock);
Willy Tarreau9df94c22016-10-31 18:42:52 +0100510 if (peer->appctx == appctx) {
Emeric Brunb157d732015-08-21 12:00:30 +0200511 /* Re-init current table pointers to force announcement on re-connect */
512 peer->remote_table = peer->last_local_table = NULL;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200513 peer->appctx = NULL;
514 if (peer->flags & PEER_F_LEARN_ASSIGN) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200515 /* unassign current peer for learning */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200516 peer->flags &= ~(PEER_F_LEARN_ASSIGN);
517 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brun2b920a12010-09-23 18:30:22 +0200518
519 /* reschedule a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200520 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
Emeric Brun2b920a12010-09-23 18:30:22 +0200521 }
522 /* reset teaching and learning flags to 0 */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200523 peer->flags &= PEER_TEACH_RESET;
524 peer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200525 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100526 HA_SPIN_UNLOCK(PEER_LOCK, &peer->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200527 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200528 }
529}
530
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200531/* Retrieve the major and minor versions of peers protocol
532 * announced by a remote peer. <str> is a null-terminated
533 * string with the following format: "<maj_ver>.<min_ver>".
534 */
535static int peer_get_version(const char *str,
536 unsigned int *maj_ver, unsigned int *min_ver)
537{
538 unsigned int majv, minv;
539 const char *pos, *saved;
540 const char *end;
541
542 saved = pos = str;
543 end = str + strlen(str);
544
545 majv = read_uint(&pos, end);
546 if (saved == pos || *pos++ != '.')
547 return -1;
548
549 saved = pos;
550 minv = read_uint(&pos, end);
551 if (saved == pos || pos != end)
552 return -1;
553
554 *maj_ver = majv;
555 *min_ver = minv;
556
557 return 0;
558}
Emeric Brun2b920a12010-09-23 18:30:22 +0200559
560/*
561 * IO Handler to handle message exchance with a peer
562 */
Willy Tarreau00a37f02015-04-13 12:05:19 +0200563static void peer_io_handler(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +0200564{
Willy Tarreau00a37f02015-04-13 12:05:19 +0200565 struct stream_interface *si = appctx->owner;
Willy Tarreau87b09662015-04-03 00:22:06 +0200566 struct stream *s = si_strm(si);
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200567 struct peers *curpeers = strm_fe(s)->parent;
Emeric Brun80527f52017-06-19 17:46:37 +0200568 struct peer *curpeer = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +0200569 int reql = 0;
570 int repl = 0;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200571 size_t proto_len = strlen(PEER_SESSION_PROTO_NAME);
572 unsigned int maj_ver, min_ver;
Emeric Brun2b920a12010-09-23 18:30:22 +0200573
Christopher Fauleta73e59b2016-12-09 17:30:18 +0100574 /* Check if the input buffer is avalaible. */
Willy Tarreauc9fa0482018-07-10 17:43:27 +0200575 if (si_ic(si)->buf.size == 0)
Christopher Fauleta73e59b2016-12-09 17:30:18 +0100576 goto full;
577
Emeric Brun2b920a12010-09-23 18:30:22 +0200578 while (1) {
579switchstate:
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200580 maj_ver = min_ver = (unsigned int)-1;
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100581 switch(appctx->st0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100582 case PEER_SESS_ST_ACCEPT:
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100583 appctx->ctx.peers.ptr = NULL;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100584 appctx->st0 = PEER_SESS_ST_GETVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200585 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100586 case PEER_SESS_ST_GETVERSION:
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200587 reql = co_getline(si_oc(si), trash.area,
588 trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200589 if (reql <= 0) { /* closed or EOL not found */
590 if (reql == 0)
591 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100592 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200593 goto switchstate;
594 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200595 if (trash.area[reql-1] != '\n') {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100596 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200597 goto switchstate;
598 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200599 else if (reql > 1 && (trash.area[reql-2] == '\r'))
600 trash.area[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200601 else
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200602 trash.area[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200603
Willy Tarreau06d80a92017-10-19 14:32:15 +0200604 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200605
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200606 /* test protocol */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200607 if (strncmp(PEER_SESSION_PROTO_NAME " ", trash.area, proto_len + 1) != 0) {
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200608 appctx->st0 = PEER_SESS_ST_EXIT;
609 appctx->st1 = PEER_SESS_SC_ERRPROTO;
610 goto switchstate;
611 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200612 if (peer_get_version(trash.area + proto_len + 1, &maj_ver, &min_ver) == -1 ||
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200613 maj_ver != PEER_MAJOR_VER || min_ver > PEER_MINOR_VER) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100614 appctx->st0 = PEER_SESS_ST_EXIT;
615 appctx->st1 = PEER_SESS_SC_ERRVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200616 goto switchstate;
617 }
618
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100619 appctx->st0 = PEER_SESS_ST_GETHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200620 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100621 case PEER_SESS_ST_GETHOST:
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200622 reql = co_getline(si_oc(si), trash.area,
623 trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200624 if (reql <= 0) { /* closed or EOL not found */
625 if (reql == 0)
626 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100627 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200628 goto switchstate;
629 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200630 if (trash.area[reql-1] != '\n') {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100631 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200632 goto switchstate;
633 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200634 else if (reql > 1 && (trash.area[reql-2] == '\r'))
635 trash.area[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200636 else
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200637 trash.area[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200638
Willy Tarreau06d80a92017-10-19 14:32:15 +0200639 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200640
641 /* test hostname match */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200642 if (strcmp(localpeer, trash.area) != 0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100643 appctx->st0 = PEER_SESS_ST_EXIT;
644 appctx->st1 = PEER_SESS_SC_ERRHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200645 goto switchstate;
646 }
647
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100648 appctx->st0 = PEER_SESS_ST_GETPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200649 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100650 case PEER_SESS_ST_GETPEER: {
Emeric Brun2b920a12010-09-23 18:30:22 +0200651 char *p;
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200652 reql = co_getline(si_oc(si), trash.area,
653 trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200654 if (reql <= 0) { /* closed or EOL not found */
655 if (reql == 0)
656 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100657 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200658 goto switchstate;
659 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200660 if (trash.area[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200661 /* Incomplete line, we quit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100662 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200663 goto switchstate;
664 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200665 else if (reql > 1 && (trash.area[reql-2] == '\r'))
666 trash.area[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200667 else
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200668 trash.area[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200669
Willy Tarreau06d80a92017-10-19 14:32:15 +0200670 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200671
Emeric Brunb3971ab2015-05-12 18:49:09 +0200672 /* parse line "<peer name> <pid> <relative_pid>" */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200673 p = strchr(trash.area, ' ');
Emeric Brun2b920a12010-09-23 18:30:22 +0200674 if (!p) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100675 appctx->st0 = PEER_SESS_ST_EXIT;
676 appctx->st1 = PEER_SESS_SC_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +0200677 goto switchstate;
678 }
679 *p = 0;
680
681 /* lookup known peer */
682 for (curpeer = curpeers->remote; curpeer; curpeer = curpeer->next) {
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200683 if (strcmp(curpeer->id, trash.area) == 0)
Emeric Brun2b920a12010-09-23 18:30:22 +0200684 break;
685 }
686
687 /* if unknown peer */
688 if (!curpeer) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100689 appctx->st0 = PEER_SESS_ST_EXIT;
690 appctx->st1 = PEER_SESS_SC_ERRPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200691 goto switchstate;
692 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200693
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100694 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Willy Tarreau9df94c22016-10-31 18:42:52 +0100695 if (curpeer->appctx && curpeer->appctx != appctx) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200696 if (curpeer->local) {
697 /* Local connection, reply a retry */
698 appctx->st0 = PEER_SESS_ST_EXIT;
699 appctx->st1 = PEER_SESS_SC_TRYAGAIN;
700 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +0200701 }
Emeric Brun80527f52017-06-19 17:46:37 +0200702
703 /* we're killing a connection, we must apply a random delay before
704 * retrying otherwise the other end will do the same and we can loop
705 * for a while.
706 */
707 curpeer->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + random() % 2000));
Willy Tarreau81bc3b02016-10-31 17:37:39 +0100708 peer_session_forceshutdown(curpeer->appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +0200709 }
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200710 if (maj_ver != (unsigned int)-1 && min_ver != (unsigned int)-1) {
711 if (min_ver == PEER_DWNGRD_MINOR_VER) {
712 curpeer->flags |= PEER_F_DWNGRD;
713 }
714 else {
715 curpeer->flags &= ~PEER_F_DWNGRD;
716 }
717 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200718 curpeer->appctx = appctx;
719 appctx->ctx.peers.ptr = curpeer;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100720 appctx->st0 = PEER_SESS_ST_SENDSUCCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200721 /* fall through */
722 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100723 case PEER_SESS_ST_SENDSUCCESS: {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200724 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +0200725
Emeric Brun80527f52017-06-19 17:46:37 +0200726 if (!curpeer) {
727 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100728 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +0200729 if (curpeer->appctx != appctx) {
730 appctx->st0 = PEER_SESS_ST_END;
731 goto switchstate;
732 }
733 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200734 repl = snprintf(trash.area, trash.size,
735 "%d\n",
736 PEER_SESS_SC_SUCCESSCODE);
737 repl = ci_putblk(si_ic(si), trash.area, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200738 if (repl <= 0) {
739 if (repl == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +0100740 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100741 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200742 goto switchstate;
743 }
744
745 /* Register status code */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200746 curpeer->statuscode = PEER_SESS_SC_SUCCESSCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200747
748 /* Awake main task */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200749 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200750
751 /* Init confirm counter */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200752 curpeer->confirm = 0;
753
754 /* Init cursors */
755 for (st = curpeer->tables; st ; st = st->next) {
756 st->last_get = st->last_acked = 0;
757 st->teaching_origin = st->last_pushed = st->update;
758 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200759
760 /* reset teaching and learning flags to 0 */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200761 curpeer->flags &= PEER_TEACH_RESET;
762 curpeer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200763
764 /* if current peer is local */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200765 if (curpeer->local) {
766 /* if current host need resyncfrom local and no process assined */
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200767 if ((curpeers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL &&
768 !(curpeers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200769 /* assign local peer for a lesson, consider lesson already requested */
770 curpeer->flags |= PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200771 curpeers->flags |= (PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200772 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200773
Emeric Brunb3971ab2015-05-12 18:49:09 +0200774 }
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200775 else if ((curpeers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
776 !(curpeers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200777 /* assign peer for a lesson */
778 curpeer->flags |= PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200779 curpeers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200780 }
781
782
Emeric Brun2b920a12010-09-23 18:30:22 +0200783 /* switch to waiting message state */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100784 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200785 goto switchstate;
786 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100787 case PEER_SESS_ST_CONNECT: {
Emeric Brun80527f52017-06-19 17:46:37 +0200788
789 if (!curpeer) {
790 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100791 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +0200792 if (curpeer->appctx != appctx) {
793 appctx->st0 = PEER_SESS_ST_END;
794 goto switchstate;
795 }
796 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200797
798 /* Send headers */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200799 repl = snprintf(trash.area, trash.size,
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200800 PEER_SESSION_PROTO_NAME " %u.%u\n%s\n%s %d %d\n",
801 PEER_MAJOR_VER,
802 (curpeer->flags & PEER_F_DWNGRD) ? PEER_DWNGRD_MINOR_VER : PEER_MINOR_VER,
Emeric Brunb3971ab2015-05-12 18:49:09 +0200803 curpeer->id,
Emeric Brun2b920a12010-09-23 18:30:22 +0200804 localpeer,
Willy Tarreau7b77c9f2012-01-07 22:52:12 +0100805 (int)getpid(),
Emeric Brunb3971ab2015-05-12 18:49:09 +0200806 relative_pid);
Emeric Brun2b920a12010-09-23 18:30:22 +0200807
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100808 if (repl >= trash.size) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100809 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200810 goto switchstate;
811 }
812
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200813 repl = ci_putblk(si_ic(si), trash.area, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200814 if (repl <= 0) {
815 if (repl == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +0100816 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100817 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200818 goto switchstate;
819 }
820
821 /* switch to the waiting statuscode state */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100822 appctx->st0 = PEER_SESS_ST_GETSTATUS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200823 /* fall through */
824 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100825 case PEER_SESS_ST_GETSTATUS: {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200826 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +0200827
Emeric Brun80527f52017-06-19 17:46:37 +0200828 if (!curpeer) {
829 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100830 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +0200831 if (curpeer->appctx != appctx) {
832 appctx->st0 = PEER_SESS_ST_END;
833 goto switchstate;
834 }
835 }
836
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100837 if (si_ic(si)->flags & CF_WRITE_PARTIAL)
Emeric Brunb3971ab2015-05-12 18:49:09 +0200838 curpeer->statuscode = PEER_SESS_SC_CONNECTEDCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200839
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200840 reql = co_getline(si_oc(si), trash.area,
841 trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200842 if (reql <= 0) { /* closed or EOL not found */
843 if (reql == 0)
844 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100845 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200846 goto switchstate;
847 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200848 if (trash.area[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200849 /* Incomplete line, we quit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100850 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200851 goto switchstate;
852 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200853 else if (reql > 1 && (trash.area[reql-2] == '\r'))
854 trash.area[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200855 else
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200856 trash.area[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200857
Willy Tarreau06d80a92017-10-19 14:32:15 +0200858 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200859
860 /* Register status code */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200861 curpeer->statuscode = atoi(trash.area);
Emeric Brun2b920a12010-09-23 18:30:22 +0200862
863 /* Awake main task */
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200864 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200865
866 /* If status code is success */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200867 if (curpeer->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200868 /* Init cursors */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200869 for (st = curpeer->tables; st ; st = st->next) {
870 st->last_get = st->last_acked = 0;
871 st->teaching_origin = st->last_pushed = st->update;
872 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200873
874 /* Init confirm counter */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200875 curpeer->confirm = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200876
Emeric Brunb3971ab2015-05-12 18:49:09 +0200877 /* reset teaching and learning flags to 0 */
878 curpeer->flags &= PEER_TEACH_RESET;
879 curpeer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200880
Emeric Brunb3971ab2015-05-12 18:49:09 +0200881 /* If current peer is local */
882 if (curpeer->local) {
883 /* flag to start to teach lesson */
884 curpeer->flags |= PEER_F_TEACH_PROCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200885
Emeric Brunb3971ab2015-05-12 18:49:09 +0200886 }
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200887 else if ((curpeers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
888 !(curpeers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200889 /* If peer is remote and resync from remote is needed,
890 and no peer currently assigned */
Emeric Brun2b920a12010-09-23 18:30:22 +0200891
Emeric Brunb3971ab2015-05-12 18:49:09 +0200892 /* assign peer for a lesson */
893 curpeer->flags |= PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200894 curpeers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brun2b920a12010-09-23 18:30:22 +0200895 }
896
897 }
898 else {
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200899 if (curpeer->statuscode == PEER_SESS_SC_ERRVERSION)
900 curpeer->flags |= PEER_F_DWNGRD;
Emeric Brun2b920a12010-09-23 18:30:22 +0200901 /* Status code is not success, abort */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100902 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200903 goto switchstate;
904 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100905 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200906 /* fall through */
907 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100908 case PEER_SESS_ST_WAITMSG: {
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200909 struct stksess *ts, *newts = NULL;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200910 uint32_t msg_len = 0;
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200911 char *msg_cur = trash.area;
912 char *msg_end = trash.area;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200913 unsigned char msg_head[7];
Emeric Brun2b920a12010-09-23 18:30:22 +0200914 int totl = 0;
915
Emeric Brun80527f52017-06-19 17:46:37 +0200916 if (!curpeer) {
917 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100918 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +0200919 if (curpeer->appctx != appctx) {
920 appctx->st0 = PEER_SESS_ST_END;
921 goto switchstate;
922 }
923 }
924
Willy Tarreau06d80a92017-10-19 14:32:15 +0200925 reql = co_getblk(si_oc(si), (char *)msg_head, 2*sizeof(unsigned char), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200926 if (reql <= 0) /* closed or EOL not found */
927 goto incomplete;
928
Emeric Brun2b920a12010-09-23 18:30:22 +0200929 totl += reql;
930
Emeric Brunb3971ab2015-05-12 18:49:09 +0200931 if (msg_head[1] >= 128) {
932 /* Read and Decode message length */
Willy Tarreau06d80a92017-10-19 14:32:15 +0200933 reql = co_getblk(si_oc(si), (char *)&msg_head[2], sizeof(unsigned char), totl);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200934 if (reql <= 0) /* closed */
935 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +0200936
Emeric Brunb3971ab2015-05-12 18:49:09 +0200937 totl += reql;
938
939 if (msg_head[2] < 240) {
940 msg_len = msg_head[2];
Emeric Brun2b920a12010-09-23 18:30:22 +0200941 }
942 else {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200943 int i;
944 char *cur;
945 char *end;
Willy Tarreau72d6c162013-04-11 16:14:13 +0200946
Emeric Brunb3971ab2015-05-12 18:49:09 +0200947 for (i = 3 ; i < sizeof(msg_head) ; i++) {
Willy Tarreau06d80a92017-10-19 14:32:15 +0200948 reql = co_getblk(si_oc(si), (char *)&msg_head[i], sizeof(char), totl);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200949 if (reql <= 0) /* closed */
950 goto incomplete;
951
952 totl += reql;
953
954 if (!(msg_head[i] & 0x80))
955 break;
956 }
957
958 if (i == sizeof(msg_head)) {
959 /* malformed message */
960 appctx->st0 = PEER_SESS_ST_ERRPROTO;
961 goto switchstate;
962
963 }
964 end = (char *)msg_head + sizeof(msg_head);
965 cur = (char *)&msg_head[2];
966 msg_len = intdecode(&cur, end);
967 if (!cur) {
968 /* malformed message */
969 appctx->st0 = PEER_SESS_ST_ERRPROTO;
970 goto switchstate;
971 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200972 }
973
Willy Tarreau86a446e2013-11-25 23:02:37 +0100974
Emeric Brunb3971ab2015-05-12 18:49:09 +0200975 /* Read message content */
976 if (msg_len) {
977 if (msg_len > trash.size) {
978 /* Status code is not success, abort */
979 appctx->st0 = PEER_SESS_ST_ERRSIZE;
980 goto switchstate;
981 }
Willy Tarreau72d6c162013-04-11 16:14:13 +0200982
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200983 reql = co_getblk(si_oc(si),
984 trash.area,
985 msg_len,
986 totl);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200987 if (reql <= 0) /* closed */
988 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +0200989 totl += reql;
Willy Tarreau86a446e2013-11-25 23:02:37 +0100990
Emeric Brunb3971ab2015-05-12 18:49:09 +0200991 msg_end += msg_len;
992 }
993 }
Willy Tarreau86a446e2013-11-25 23:02:37 +0100994
Emeric Brunb3971ab2015-05-12 18:49:09 +0200995 if (msg_head[0] == PEER_MSG_CLASS_CONTROL) {
996 if (msg_head[1] == PEER_MSG_CTRL_RESYNCREQ) {
997 struct shared_table *st;
998 /* Reset message: remote need resync */
999
1000 /* prepare tables fot a global push */
1001 for (st = curpeer->tables; st; st = st->next) {
1002 st->teaching_origin = st->last_pushed = st->table->update;
1003 st->flags = 0;
Willy Tarreau86a446e2013-11-25 23:02:37 +01001004 }
Willy Tarreau72d6c162013-04-11 16:14:13 +02001005
Emeric Brunb3971ab2015-05-12 18:49:09 +02001006 /* reset teaching flags to 0 */
1007 curpeer->flags &= PEER_TEACH_RESET;
Willy Tarreau9d9179b2013-04-11 16:56:44 +02001008
Emeric Brunb3971ab2015-05-12 18:49:09 +02001009 /* flag to start to teach lesson */
1010 curpeer->flags |= PEER_F_TEACH_PROCESS;
1011
1012
1013 }
1014 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCFINISHED) {
1015
1016 if (curpeer->flags & PEER_F_LEARN_ASSIGN) {
1017 curpeer->flags &= ~PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001018 curpeers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
1019 curpeers->flags |= (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE);
Willy Tarreau86a446e2013-11-25 23:02:37 +01001020 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001021 curpeer->confirm++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001022 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001023 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCPARTIAL) {
1024
1025 if (curpeer->flags & PEER_F_LEARN_ASSIGN) {
1026 curpeer->flags &= ~PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001027 curpeers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001028
1029 curpeer->flags |= PEER_F_LEARN_NOTUP2DATE;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001030 curpeers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
1031 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Cyril Bonté9a60ff92014-02-16 01:07:07 +01001032 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001033 curpeer->confirm++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001034 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001035 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCCONFIRM) {
Emeric Brun597b26e2016-08-12 11:23:31 +02001036 struct shared_table *st;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001037
1038 /* If stopping state */
1039 if (stopping) {
1040 /* Close session, push resync no more needed */
1041 curpeer->flags |= PEER_F_TEACH_COMPLETE;
1042 appctx->st0 = PEER_SESS_ST_END;
1043 goto switchstate;
1044 }
Emeric Brun597b26e2016-08-12 11:23:31 +02001045 for (st = curpeer->tables; st; st = st->next) {
1046 st->update = st->last_pushed = st->teaching_origin;
1047 st->flags = 0;
1048 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001049
1050 /* reset teaching flags to 0 */
1051 curpeer->flags &= PEER_TEACH_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +02001052 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001053 }
1054 else if (msg_head[0] == PEER_MSG_CLASS_STICKTABLE) {
1055 if (msg_head[1] == PEER_MSG_STKT_DEFINE) {
1056 int table_id_len;
1057 struct shared_table *st;
1058 int table_type;
1059 int table_keylen;
1060 int table_id;
1061 uint64_t table_data;
Emeric Brun2b920a12010-09-23 18:30:22 +02001062
Emeric Brunb3971ab2015-05-12 18:49:09 +02001063 table_id = intdecode(&msg_cur, msg_end);
1064 if (!msg_cur) {
1065 /* malformed message */
1066 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1067 goto switchstate;
1068 }
Willy Tarreau72d6c162013-04-11 16:14:13 +02001069
Emeric Brunb3971ab2015-05-12 18:49:09 +02001070 table_id_len = intdecode(&msg_cur, msg_end);
1071 if (!msg_cur) {
1072 /* malformed message */
1073 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1074 goto switchstate;
1075 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001076
Emeric Brunb3971ab2015-05-12 18:49:09 +02001077 curpeer->remote_table = NULL;
1078 if (!table_id_len || (msg_cur + table_id_len) >= msg_end) {
1079 /* malformed message */
1080 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1081 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001082 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001083
Emeric Brunb3971ab2015-05-12 18:49:09 +02001084 for (st = curpeer->tables; st; st = st->next) {
1085 /* Reset IDs */
1086 if (st->remote_id == table_id)
1087 st->remote_id = 0;
Willy Tarreau9d9179b2013-04-11 16:56:44 +02001088
Emeric Brunb3971ab2015-05-12 18:49:09 +02001089 if (!curpeer->remote_table
1090 && (table_id_len == strlen(st->table->id))
1091 && (memcmp(st->table->id, msg_cur, table_id_len) == 0)) {
1092 curpeer->remote_table = st;
Emeric Brun2b920a12010-09-23 18:30:22 +02001093 }
1094 }
1095
Emeric Brunb3971ab2015-05-12 18:49:09 +02001096 if (!curpeer->remote_table) {
1097 goto ignore_msg;
1098 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001099
Emeric Brunb3971ab2015-05-12 18:49:09 +02001100 msg_cur += table_id_len;
1101 if (msg_cur >= msg_end) {
1102 /* malformed message */
1103 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1104 goto switchstate;
1105 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001106
Emeric Brunb3971ab2015-05-12 18:49:09 +02001107 table_type = intdecode(&msg_cur, msg_end);
1108 if (!msg_cur) {
1109 /* malformed message */
1110 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1111 goto switchstate;
1112 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001113
Emeric Brunb3971ab2015-05-12 18:49:09 +02001114 table_keylen = intdecode(&msg_cur, msg_end);
1115 if (!msg_cur) {
1116 /* malformed message */
1117 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1118 goto switchstate;
1119 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001120
Emeric Brunb3971ab2015-05-12 18:49:09 +02001121 table_data = intdecode(&msg_cur, msg_end);
1122 if (!msg_cur) {
1123 /* malformed message */
1124 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1125 goto switchstate;
1126 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001127
Emeric Brunb3971ab2015-05-12 18:49:09 +02001128 if (curpeer->remote_table->table->type != table_type
1129 || curpeer->remote_table->table->key_size != table_keylen) {
1130 curpeer->remote_table = NULL;
1131 goto ignore_msg;
1132 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001133
Emeric Brunb3971ab2015-05-12 18:49:09 +02001134 curpeer->remote_table->remote_data = table_data;
1135 curpeer->remote_table->remote_id = table_id;
Emeric Brun2b920a12010-09-23 18:30:22 +02001136 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001137 else if (msg_head[1] == PEER_MSG_STKT_SWITCH) {
1138 struct shared_table *st;
1139 int table_id;
Emeric Brun2b920a12010-09-23 18:30:22 +02001140
Emeric Brunb3971ab2015-05-12 18:49:09 +02001141 table_id = intdecode(&msg_cur, msg_end);
1142 if (!msg_cur) {
1143 /* malformed message */
1144 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1145 goto switchstate;
1146 }
1147 curpeer->remote_table = NULL;
1148 for (st = curpeer->tables; st; st = st->next) {
1149 if (st->remote_id == table_id) {
1150 curpeer->remote_table = st;
1151 break;
1152 }
1153 }
1154
Emeric Brun2b920a12010-09-23 18:30:22 +02001155 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001156 else if (msg_head[1] == PEER_MSG_STKT_UPDATE
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001157 || msg_head[1] == PEER_MSG_STKT_INCUPDATE
1158 || msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED
1159 || msg_head[1] == PEER_MSG_STKT_INCUPDATE_TIMED) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001160 struct shared_table *st = curpeer->remote_table;
1161 uint32_t update;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001162 int expire;
Emeric Brun94900952015-06-11 18:25:54 +02001163 unsigned int data_type;
1164 void *data_ptr;
Emeric Brun2b920a12010-09-23 18:30:22 +02001165
Emeric Brunb3971ab2015-05-12 18:49:09 +02001166 /* Here we have data message */
1167 if (!st)
1168 goto ignore_msg;
Emeric Brun2b920a12010-09-23 18:30:22 +02001169
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001170 expire = MS_TO_TICKS(st->table->expire);
1171
1172 if (msg_head[1] == PEER_MSG_STKT_UPDATE ||
1173 msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001174 if (msg_len < sizeof(update)) {
1175 /* malformed message */
1176 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1177 goto switchstate;
1178 }
1179 memcpy(&update, msg_cur, sizeof(update));
1180 msg_cur += sizeof(update);
1181 st->last_get = htonl(update);
1182 }
1183 else {
1184 st->last_get++;
1185 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001186
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001187 if (msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED ||
1188 msg_head[1] == PEER_MSG_STKT_INCUPDATE_TIMED) {
1189 size_t expire_sz = sizeof expire;
1190
1191 if (msg_cur + expire_sz > msg_end) {
1192 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1193 goto switchstate;
1194 }
1195 memcpy(&expire, msg_cur, expire_sz);
1196 msg_cur += expire_sz;
1197 expire = ntohl(expire);
1198 }
1199
Emeric Brunb3971ab2015-05-12 18:49:09 +02001200 newts = stksess_new(st->table, NULL);
1201 if (!newts)
1202 goto ignore_msg;
Emeric Brun55482912018-01-22 15:10:08 +01001203
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +02001204 if (st->table->type == SMP_T_STR) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001205 unsigned int to_read, to_store;
Emeric Brun2b920a12010-09-23 18:30:22 +02001206
Emeric Brunb3971ab2015-05-12 18:49:09 +02001207 to_read = intdecode(&msg_cur, msg_end);
1208 if (!msg_cur) {
1209 /* malformed message */
1210 stksess_free(st->table, newts);
1211 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1212 goto switchstate;
1213 }
1214 to_store = MIN(to_read, st->table->key_size - 1);
1215 if (msg_cur + to_store > msg_end) {
1216 /* malformed message */
1217 stksess_free(st->table, newts);
1218 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1219 goto switchstate;
1220 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001221
Emeric Brunb3971ab2015-05-12 18:49:09 +02001222 memcpy(newts->key.key, msg_cur, to_store);
1223 newts->key.key[to_store] = 0;
1224 msg_cur += to_read;
1225 }
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +02001226 else if (st->table->type == SMP_T_SINT) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001227 unsigned int netinteger;
Willy Tarreau72d6c162013-04-11 16:14:13 +02001228
Emeric Brunb3971ab2015-05-12 18:49:09 +02001229 if (msg_cur + sizeof(netinteger) > msg_end) {
1230 /* malformed message */
1231 stksess_free(st->table, newts);
1232 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1233 goto switchstate;
1234 }
1235 memcpy(&netinteger, msg_cur, sizeof(netinteger));
1236 netinteger = ntohl(netinteger);
1237 memcpy(newts->key.key, &netinteger, sizeof(netinteger));
1238 msg_cur += sizeof(netinteger);
1239 }
1240 else {
1241 if (msg_cur + st->table->key_size > msg_end) {
1242 /* malformed message */
1243 stksess_free(st->table, newts);
1244 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1245 goto switchstate;
1246 }
1247 memcpy(newts->key.key, msg_cur, st->table->key_size);
1248 msg_cur += st->table->key_size;
1249 }
1250
1251 /* lookup for existing entry */
Emeric Brun819fc6f2017-06-13 19:37:32 +02001252 ts = stktable_set_entry(st->table, newts);
1253 if (ts != newts) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001254 stksess_free(st->table, newts);
1255 newts = NULL;
1256 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001257
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001258 HA_RWLOCK_WRLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001259
Emeric Brun94900952015-06-11 18:25:54 +02001260 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001261
Emeric Brun94900952015-06-11 18:25:54 +02001262 if ((1 << data_type) & st->remote_data) {
1263 switch (stktable_data_types[data_type].std_type) {
1264 case STD_T_SINT: {
1265 int data;
1266
1267 data = intdecode(&msg_cur, msg_end);
1268 if (!msg_cur) {
1269 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001270 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001271 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001272 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1273 goto switchstate;
1274 }
1275
1276 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1277 if (data_ptr)
1278 stktable_data_cast(data_ptr, std_t_sint) = data;
1279 break;
1280 }
1281 case STD_T_UINT: {
1282 unsigned int data;
1283
1284 data = intdecode(&msg_cur, msg_end);
1285 if (!msg_cur) {
1286 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001287 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001288 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001289 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1290 goto switchstate;
1291 }
1292
1293 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1294 if (data_ptr)
1295 stktable_data_cast(data_ptr, std_t_uint) = data;
1296 break;
1297 }
1298 case STD_T_ULL: {
1299 unsigned long long data;
1300
1301 data = intdecode(&msg_cur, msg_end);
1302 if (!msg_cur) {
1303 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001304 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001305 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001306 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1307 goto switchstate;
1308 }
1309
1310 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1311 if (data_ptr)
1312 stktable_data_cast(data_ptr, std_t_ull) = data;
1313 break;
1314 }
1315 case STD_T_FRQP: {
1316 struct freq_ctr_period data;
1317
Emeric Brunf2fc1fd2017-11-02 17:32:43 +01001318 /* First bit is reserved for the freq_ctr_period lock
1319 Note: here we're still protected by the stksess lock
1320 so we don't need to update the update the freq_ctr_period
1321 using its internal lock */
1322
1323 data.curr_tick = tick_add(now_ms, -intdecode(&msg_cur, msg_end)) & ~0x1;
Emeric Brun94900952015-06-11 18:25:54 +02001324 if (!msg_cur) {
1325 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001326 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001327 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001328 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1329 goto switchstate;
1330 }
1331 data.curr_ctr = intdecode(&msg_cur, msg_end);
1332 if (!msg_cur) {
1333 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001334 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001335 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001336 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1337 goto switchstate;
1338 }
1339 data.prev_ctr = intdecode(&msg_cur, msg_end);
1340 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 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001347
Emeric Brun94900952015-06-11 18:25:54 +02001348 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1349 if (data_ptr)
1350 stktable_data_cast(data_ptr, std_t_frqp) = data;
1351 break;
1352 }
1353 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001354 }
1355 }
Emeric Brun819fc6f2017-06-13 19:37:32 +02001356
Emeric Brun55482912018-01-22 15:10:08 +01001357 /* Force new expiration */
1358 ts->expire = tick_add(now_ms, expire);
1359
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001360 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001361 stktable_touch_remote(st->table, ts, 1);
1362
Emeric Brunb3971ab2015-05-12 18:49:09 +02001363 }
1364 else if (msg_head[1] == PEER_MSG_STKT_ACK) {
1365 /* ack message */
1366 uint32_t table_id ;
1367 uint32_t update;
1368 struct shared_table *st;
1369
1370 table_id = intdecode(&msg_cur, msg_end);
1371 if (!msg_cur || (msg_cur + sizeof(update) > msg_end)) {
1372 /* malformed message */
1373 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1374 goto switchstate;
1375 }
1376 memcpy(&update, msg_cur, sizeof(update));
1377 update = ntohl(update);
Emeric Brun2b920a12010-09-23 18:30:22 +02001378
Emeric Brunb3971ab2015-05-12 18:49:09 +02001379 for (st = curpeer->tables; st; st = st->next) {
1380 if (st->local_id == table_id) {
1381 st->update = update;
1382 break;
1383 }
1384 }
1385 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001386 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001387 else if (msg_head[0] == PEER_MSG_CLASS_RESERVED) {
1388 appctx->st0 = PEER_SESS_ST_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +02001389 goto switchstate;
1390 }
1391
Emeric Brunb3971ab2015-05-12 18:49:09 +02001392ignore_msg:
Emeric Brun2b920a12010-09-23 18:30:22 +02001393 /* skip consumed message */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001394 co_skip(si_oc(si), totl);
Emeric Brun2b920a12010-09-23 18:30:22 +02001395 /* loop on that state to peek next message */
Willy Tarreau72d6c162013-04-11 16:14:13 +02001396 goto switchstate;
1397
Emeric Brun2b920a12010-09-23 18:30:22 +02001398incomplete:
Willy Tarreau06d80a92017-10-19 14:32:15 +02001399 /* we get here when a co_getblk() returns <= 0 in reql */
Willy Tarreau9d9179b2013-04-11 16:56:44 +02001400
Willy Tarreau72d6c162013-04-11 16:14:13 +02001401 if (reql < 0) {
1402 /* there was an error */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001403 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau72d6c162013-04-11 16:14:13 +02001404 goto switchstate;
1405 }
1406
Emeric Brun2b920a12010-09-23 18:30:22 +02001407
Emeric Brun2b920a12010-09-23 18:30:22 +02001408
Emeric Brunb3971ab2015-05-12 18:49:09 +02001409
Emeric Brun2b920a12010-09-23 18:30:22 +02001410 /* Need to request a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001411 if ((curpeer->flags & PEER_F_LEARN_ASSIGN) &&
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001412 (curpeers->flags & PEERS_F_RESYNC_ASSIGN) &&
1413 !(curpeers->flags & PEERS_F_RESYNC_PROCESS)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001414 unsigned char msg[2];
Emeric Brun2b920a12010-09-23 18:30:22 +02001415
Emeric Brunb3971ab2015-05-12 18:49:09 +02001416 /* Current peer was elected to request a resync */
1417 msg[0] = PEER_MSG_CLASS_CONTROL;
1418 msg[1] = PEER_MSG_CTRL_RESYNCREQ;
Emeric Brun2b920a12010-09-23 18:30:22 +02001419
Emeric Brunb3971ab2015-05-12 18:49:09 +02001420 /* message to buffer */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001421 repl = ci_putblk(si_ic(si), (char *)msg, sizeof(msg));
Emeric Brunb3971ab2015-05-12 18:49:09 +02001422 if (repl <= 0) {
1423 /* no more write possible */
1424 if (repl == -1)
1425 goto full;
1426 appctx->st0 = PEER_SESS_ST_END;
1427 goto switchstate;
1428 }
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001429 curpeers->flags |= PEERS_F_RESYNC_PROCESS;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001430 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001431
Emeric Brunb3971ab2015-05-12 18:49:09 +02001432 /* Nothing to read, now we start to write */
Emeric Brun2b920a12010-09-23 18:30:22 +02001433
Emeric Brunb3971ab2015-05-12 18:49:09 +02001434 if (curpeer->tables) {
1435 struct shared_table *st;
1436 struct shared_table *last_local_table;
Emeric Brun2b920a12010-09-23 18:30:22 +02001437
Emeric Brunb3971ab2015-05-12 18:49:09 +02001438 last_local_table = curpeer->last_local_table;
1439 if (!last_local_table)
1440 last_local_table = curpeer->tables;
1441 st = last_local_table->next;
Emeric Brun2b920a12010-09-23 18:30:22 +02001442
Emeric Brunb3971ab2015-05-12 18:49:09 +02001443 while (1) {
1444 if (!st)
1445 st = curpeer->tables;
Emeric Brun2b920a12010-09-23 18:30:22 +02001446
Emeric Brunb3971ab2015-05-12 18:49:09 +02001447 /* It remains some updates to ack */
1448 if (st->last_get != st->last_acked) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001449 int msglen;
Emeric Brun2b920a12010-09-23 18:30:22 +02001450
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001451 msglen = peer_prepare_ackmsg(st,
1452 trash.area,
1453 trash.size);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001454 if (!msglen) {
1455 /* internal error: message does not fit in trash */
1456 appctx->st0 = PEER_SESS_ST_END;
1457 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001458 }
1459
Emeric Brunb3971ab2015-05-12 18:49:09 +02001460 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001461 repl = ci_putblk(si_ic(si),
1462 trash.area,
1463 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001464 if (repl <= 0) {
1465 /* no more write possible */
1466 if (repl == -1) {
1467 goto full;
Emeric Brun2b920a12010-09-23 18:30:22 +02001468 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001469 appctx->st0 = PEER_SESS_ST_END;
1470 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001471 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001472 st->last_acked = st->last_get;
Emeric Brun2b920a12010-09-23 18:30:22 +02001473 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001474
Emeric Brunb3971ab2015-05-12 18:49:09 +02001475 if (!(curpeer->flags & PEER_F_TEACH_PROCESS)) {
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001476 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001477 if (!(curpeer->flags & PEER_F_LEARN_ASSIGN) &&
1478 ((int)(st->last_pushed - st->table->localupdate) < 0)) {
1479 struct eb32_node *eb;
1480 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001481
Emeric Brunb3971ab2015-05-12 18:49:09 +02001482 if (st != curpeer->last_local_table) {
1483 int msglen;
Emeric Brun2b920a12010-09-23 18:30:22 +02001484
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001485 msglen = peer_prepare_switchmsg(st,
1486 trash.area,
1487 trash.size);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001488 if (!msglen) {
Emeric Brun088c9b72017-12-01 11:37:36 +01001489 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001490 /* internal error: message does not fit in trash */
1491 appctx->st0 = PEER_SESS_ST_END;
1492 goto switchstate;
1493 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001494
Emeric Brunb3971ab2015-05-12 18:49:09 +02001495 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001496 repl = ci_putblk(si_ic(si),
1497 trash.area,
1498 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001499 if (repl <= 0) {
Emeric Brun088c9b72017-12-01 11:37:36 +01001500 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001501 /* no more write possible */
1502 if (repl == -1) {
1503 goto full;
1504 }
1505 appctx->st0 = PEER_SESS_ST_END;
1506 goto switchstate;
1507 }
1508 curpeer->last_local_table = st;
Emeric Brun2b920a12010-09-23 18:30:22 +02001509 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001510
1511 /* We force new pushed to 1 to force identifier in update message */
1512 new_pushed = 1;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001513 while (1) {
1514 uint32_t msglen;
1515 struct stksess *ts;
Emeric Brun819fc6f2017-06-13 19:37:32 +02001516 unsigned updateid;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001517
1518 /* push local updates */
Emeric Brun819fc6f2017-06-13 19:37:32 +02001519 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001520 if (!eb) {
1521 eb = eb32_first(&st->table->updates);
1522 if (!eb || ((int)(eb->key - st->last_pushed) <= 0)) {
Emeric Brunaaf58602015-06-15 17:23:30 +02001523 st->table->commitupdate = st->last_pushed = st->table->localupdate;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001524 break;
1525 }
1526 }
1527
1528 if ((int)(eb->key - st->table->localupdate) > 0) {
Emeric Brunaaf58602015-06-15 17:23:30 +02001529 st->table->commitupdate = st->last_pushed = st->table->localupdate;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001530 break;
1531 }
1532
1533 ts = eb32_entry(eb, struct stksess, upd);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001534 updateid = ts->upd.key;
1535 ts->ref_cnt++;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001536 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001537
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001538 msglen = peer_prepare_updatemsg(ts, st, updateid,
1539 trash.area,
1540 trash.size,
1541 new_pushed,
1542 0);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001543 if (!msglen) {
1544 /* internal error: message does not fit in trash */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001545 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001546 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001547 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001548 appctx->st0 = PEER_SESS_ST_END;
1549 goto switchstate;
1550 }
1551
1552 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001553 repl = ci_putblk(si_ic(si),
1554 trash.area,
1555 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001556 if (repl <= 0) {
1557 /* no more write possible */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001558 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001559 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001560 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001561 if (repl == -1) {
1562 goto full;
1563 }
1564 appctx->st0 = PEER_SESS_ST_END;
1565 goto switchstate;
1566 }
Emeric Brun819fc6f2017-06-13 19:37:32 +02001567
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001568 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001569 ts->ref_cnt--;
1570 st->last_pushed = updateid;
Emeric Brunaaf58602015-06-15 17:23:30 +02001571 if ((int)(st->last_pushed - st->table->commitupdate) > 0)
1572 st->table->commitupdate = st->last_pushed;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001573 /* identifier may not needed in next update message */
1574 new_pushed = 0;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001575 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001576 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001577 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun2b920a12010-09-23 18:30:22 +02001578 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001579 else {
1580 if (!(st->flags & SHTABLE_F_TEACH_STAGE1)) {
1581 struct eb32_node *eb;
1582 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001583
Emeric Brunb3971ab2015-05-12 18:49:09 +02001584 if (st != curpeer->last_local_table) {
1585 int msglen;
1586
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001587 msglen = peer_prepare_switchmsg(st,
1588 trash.area,
1589 trash.size);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001590 if (!msglen) {
1591 /* internal error: message does not fit in trash */
1592 appctx->st0 = PEER_SESS_ST_END;
1593 goto switchstate;
1594 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001595
Emeric Brunb3971ab2015-05-12 18:49:09 +02001596 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001597 repl = ci_putblk(si_ic(si),
1598 trash.area,
1599 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001600 if (repl <= 0) {
1601 /* no more write possible */
1602 if (repl == -1) {
1603 goto full;
1604 }
1605 appctx->st0 = PEER_SESS_ST_END;
1606 goto switchstate;
1607 }
1608 curpeer->last_local_table = st;
1609 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001610
Emeric Brunb3971ab2015-05-12 18:49:09 +02001611 /* We force new pushed to 1 to force identifier in update message */
1612 new_pushed = 1;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001613 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001614 while (1) {
1615 uint32_t msglen;
1616 struct stksess *ts;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001617 int use_timed;
Emeric Brun819fc6f2017-06-13 19:37:32 +02001618 unsigned updateid;
Emeric Brun2b920a12010-09-23 18:30:22 +02001619
Emeric Brunb3971ab2015-05-12 18:49:09 +02001620 /* push local updates */
Emeric Brun819fc6f2017-06-13 19:37:32 +02001621 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001622 if (!eb) {
1623 st->flags |= SHTABLE_F_TEACH_STAGE1;
1624 eb = eb32_first(&st->table->updates);
1625 if (eb)
1626 st->last_pushed = eb->key - 1;
1627 break;
1628 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001629
Emeric Brunb3971ab2015-05-12 18:49:09 +02001630 ts = eb32_entry(eb, struct stksess, upd);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001631 updateid = ts->upd.key;
1632 ts->ref_cnt++;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001633 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001634
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001635 use_timed = !(curpeer->flags & PEER_F_DWNGRD);
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001636 msglen = peer_prepare_updatemsg(ts, st, updateid,
1637 trash.area,
1638 trash.size,
1639 new_pushed,
1640 use_timed);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001641 if (!msglen) {
1642 /* internal error: message does not fit in trash */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001643 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001644 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001645 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001646 appctx->st0 = PEER_SESS_ST_END;
1647 goto switchstate;
1648 }
1649
1650 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001651 repl = ci_putblk(si_ic(si),
1652 trash.area,
1653 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001654 if (repl <= 0) {
1655 /* no more write possible */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001656 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001657 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001658 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001659 if (repl == -1) {
1660 goto full;
1661 }
1662 appctx->st0 = PEER_SESS_ST_END;
1663 goto switchstate;
1664 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001665 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001666 ts->ref_cnt--;
1667 st->last_pushed = updateid;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001668 /* identifier may not needed in next update message */
1669 new_pushed = 0;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001670 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001671 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun2b920a12010-09-23 18:30:22 +02001672 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001673
Emeric Brunb3971ab2015-05-12 18:49:09 +02001674 if (!(st->flags & SHTABLE_F_TEACH_STAGE2)) {
1675 struct eb32_node *eb;
1676 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001677
Emeric Brunb3971ab2015-05-12 18:49:09 +02001678 if (st != curpeer->last_local_table) {
1679 int msglen;
1680
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001681 msglen = peer_prepare_switchmsg(st,
1682 trash.area,
1683 trash.size);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001684 if (!msglen) {
1685 /* internal error: message does not fit in trash */
1686 appctx->st0 = PEER_SESS_ST_END;
1687 goto switchstate;
1688 }
1689
1690 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001691 repl = ci_putblk(si_ic(si),
1692 trash.area,
1693 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001694 if (repl <= 0) {
1695 /* no more write possible */
1696 if (repl == -1) {
1697 goto full;
1698 }
1699 appctx->st0 = PEER_SESS_ST_END;
1700 goto switchstate;
1701 }
1702 curpeer->last_local_table = st;
1703 }
1704
1705 /* We force new pushed to 1 to force identifier in update message */
1706 new_pushed = 1;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001707 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001708 while (1) {
1709 uint32_t msglen;
1710 struct stksess *ts;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001711 int use_timed;
Emeric Brun819fc6f2017-06-13 19:37:32 +02001712 unsigned updateid;
1713
1714 /* push local updates */
1715 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001716
1717 /* push local updates */
1718 if (!eb || eb->key > st->teaching_origin) {
1719 st->flags |= SHTABLE_F_TEACH_STAGE2;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001720 break;
1721 }
1722
1723 ts = eb32_entry(eb, struct stksess, upd);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001724 updateid = ts->upd.key;
1725 ts->ref_cnt++;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001726 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001727
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001728 use_timed = !(curpeer->flags & PEER_F_DWNGRD);
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001729 msglen = peer_prepare_updatemsg(ts, st, updateid,
1730 trash.area,
1731 trash.size,
1732 new_pushed,
1733 use_timed);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001734 if (!msglen) {
1735 /* internal error: message does not fit in trash */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001736 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001737 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001738 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001739 appctx->st0 = PEER_SESS_ST_END;
1740 goto switchstate;
1741 }
1742
1743 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001744 repl = ci_putblk(si_ic(si),
1745 trash.area,
1746 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001747 if (repl <= 0) {
1748 /* no more write possible */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001749 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001750 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001751 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001752 if (repl == -1) {
1753 goto full;
1754 }
1755 appctx->st0 = PEER_SESS_ST_END;
1756 goto switchstate;
1757 }
Emeric Brun819fc6f2017-06-13 19:37:32 +02001758
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001759 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001760 ts->ref_cnt--;
1761 st->last_pushed = updateid;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001762 /* identifier may not needed in next update message */
1763 new_pushed = 0;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001764 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001765 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun2b920a12010-09-23 18:30:22 +02001766 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001767 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001768
1769 if (st == last_local_table)
1770 break;
1771 st = st->next;
Emeric Brun2b920a12010-09-23 18:30:22 +02001772 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001773 }
1774
1775
1776 if ((curpeer->flags & PEER_F_TEACH_PROCESS) && !(curpeer->flags & PEER_F_TEACH_FINISHED)) {
1777 unsigned char msg[2];
1778
1779 /* Current peer was elected to request a resync */
1780 msg[0] = PEER_MSG_CLASS_CONTROL;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001781 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 +02001782 /* process final lesson message */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001783 repl = ci_putblk(si_ic(si), (char *)msg, sizeof(msg));
Emeric Brunb3971ab2015-05-12 18:49:09 +02001784 if (repl <= 0) {
1785 /* no more write possible */
1786 if (repl == -1)
1787 goto full;
1788 appctx->st0 = PEER_SESS_ST_END;
1789 goto switchstate;
1790 }
1791 /* flag finished message sent */
1792 curpeer->flags |= PEER_F_TEACH_FINISHED;
1793 }
1794
Emeric Brun597b26e2016-08-12 11:23:31 +02001795 /* Confirm finished or partial messages */
1796 while (curpeer->confirm) {
1797 unsigned char msg[2];
1798
1799 /* There is a confirm messages to send */
1800 msg[0] = PEER_MSG_CLASS_CONTROL;
1801 msg[1] = PEER_MSG_CTRL_RESYNCCONFIRM;
1802
1803 /* message to buffer */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001804 repl = ci_putblk(si_ic(si), (char *)msg, sizeof(msg));
Emeric Brun597b26e2016-08-12 11:23:31 +02001805 if (repl <= 0) {
1806 /* no more write possible */
1807 if (repl == -1)
1808 goto full;
1809 appctx->st0 = PEER_SESS_ST_END;
1810 goto switchstate;
1811 }
1812 curpeer->confirm--;
1813 }
1814
Emeric Brun2b920a12010-09-23 18:30:22 +02001815 /* noting more to do */
1816 goto out;
1817 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001818 case PEER_SESS_ST_EXIT:
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001819 repl = snprintf(trash.area, trash.size,
1820 "%d\n", appctx->st1);
1821 if (ci_putblk(si_ic(si), trash.area, repl) == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +01001822 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001823 appctx->st0 = PEER_SESS_ST_END;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001824 goto switchstate;
1825 case PEER_SESS_ST_ERRSIZE: {
1826 unsigned char msg[2];
1827
1828 msg[0] = PEER_MSG_CLASS_ERROR;
1829 msg[1] = PEER_MSG_ERR_SIZELIMIT;
1830
Willy Tarreau06d80a92017-10-19 14:32:15 +02001831 if (ci_putblk(si_ic(si), (char *)msg, sizeof(msg)) == -1)
Emeric Brunb3971ab2015-05-12 18:49:09 +02001832 goto full;
1833 appctx->st0 = PEER_SESS_ST_END;
1834 goto switchstate;
1835 }
1836 case PEER_SESS_ST_ERRPROTO: {
1837 unsigned char msg[2];
1838
1839 msg[0] = PEER_MSG_CLASS_ERROR;
1840 msg[1] = PEER_MSG_ERR_PROTOCOL;
1841
Willy Tarreau06d80a92017-10-19 14:32:15 +02001842 if (ci_putblk(si_ic(si), (char *)msg, sizeof(msg)) == -1)
Emeric Brunb3971ab2015-05-12 18:49:09 +02001843 goto full;
1844 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02001845 /* fall through */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001846 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001847 case PEER_SESS_ST_END: {
Emeric Brun80527f52017-06-19 17:46:37 +02001848 if (curpeer) {
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001849 HA_SPIN_UNLOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02001850 curpeer = NULL;
1851 }
Willy Tarreau73b013b2012-05-21 16:31:45 +02001852 si_shutw(si);
1853 si_shutr(si);
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001854 si_ic(si)->flags |= CF_READ_NULL;
Willy Tarreau828824a2015-04-19 17:20:03 +02001855 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001856 }
1857 }
1858 }
1859out:
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001860 si_oc(si)->flags |= CF_READ_DONTWAIT;
Emeric Brun80527f52017-06-19 17:46:37 +02001861
1862 if (curpeer)
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001863 HA_SPIN_UNLOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun2b920a12010-09-23 18:30:22 +02001864 return;
Willy Tarreaubc18da12015-03-13 14:00:47 +01001865full:
Willy Tarreaufe127932015-04-21 19:23:39 +02001866 si_applet_cant_put(si);
Willy Tarreaubc18da12015-03-13 14:00:47 +01001867 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001868}
1869
Willy Tarreau30576452015-04-13 13:50:30 +02001870static struct applet peer_applet = {
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001871 .obj_type = OBJ_TYPE_APPLET,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001872 .name = "<PEER>", /* used for logging */
1873 .fct = peer_io_handler,
Aman Gupta9a13e842012-04-02 18:57:53 -07001874 .release = peer_session_release,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001875};
Emeric Brun2b920a12010-09-23 18:30:22 +02001876
1877/*
1878 * Use this function to force a close of a peer session
1879 */
Willy Tarreau81bc3b02016-10-31 17:37:39 +01001880static void peer_session_forceshutdown(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +02001881{
Frédéric Lécaille5df11902017-06-13 16:39:57 +02001882 /* Note that the peer sessions which have just been created
1883 * (->st0 == PEER_SESS_ST_CONNECT) must not
1884 * be shutdown, if not, the TCP session will never be closed
1885 * and stay in CLOSE_WAIT state after having been closed by
1886 * the remote side.
1887 */
1888 if (!appctx || appctx->st0 == PEER_SESS_ST_CONNECT)
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001889 return;
1890
Willy Tarreau81bc3b02016-10-31 17:37:39 +01001891 if (appctx->applet != &peer_applet)
1892 return;
1893
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001894 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau78c0c502016-10-31 17:32:20 +01001895 appctx_wakeup(appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02001896}
1897
Willy Tarreau91d96282015-03-13 15:47:26 +01001898/* Pre-configures a peers frontend to accept incoming connections */
1899void peers_setup_frontend(struct proxy *fe)
1900{
1901 fe->last_change = now.tv_sec;
1902 fe->cap = PR_CAP_FE;
1903 fe->maxconn = 0;
1904 fe->conn_retries = CONN_RETRIES;
1905 fe->timeout.client = MS_TO_TICKS(5000);
Willy Tarreaud1d48d42015-03-13 16:15:46 +01001906 fe->accept = frontend_accept;
Willy Tarreauf87ab942015-03-13 15:55:16 +01001907 fe->default_target = &peer_applet.obj_type;
Willy Tarreau91d96282015-03-13 15:47:26 +01001908 fe->options2 |= PR_O2_INDEPSTR | PR_O2_SMARTCON | PR_O2_SMARTACC;
Willy Tarreau0fca4832015-05-01 19:12:05 +02001909 fe->bind_proc = 0; /* will be filled by users */
Willy Tarreau91d96282015-03-13 15:47:26 +01001910}
1911
Emeric Brun2b920a12010-09-23 18:30:22 +02001912/*
Willy Tarreaubd55e312010-11-11 10:55:09 +01001913 * Create a new peer session in assigned state (connect will start automatically)
Emeric Brun2b920a12010-09-23 18:30:22 +02001914 */
Willy Tarreau9df94c22016-10-31 18:42:52 +01001915static struct appctx *peer_session_create(struct peers *peers, struct peer *peer)
Emeric Brun2b920a12010-09-23 18:30:22 +02001916{
Willy Tarreau04b92862017-09-15 11:01:04 +02001917 struct proxy *p = peers->peers_fe; /* attached frontend */
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001918 struct appctx *appctx;
Willy Tarreau15b5e142015-04-04 14:38:25 +02001919 struct session *sess;
Willy Tarreau87b09662015-04-03 00:22:06 +02001920 struct stream *s;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001921 struct connection *conn;
Olivier Houchard9aaf7782017-09-13 18:30:23 +02001922 struct conn_stream *cs;
Emeric Brun2b920a12010-09-23 18:30:22 +02001923
Emeric Brunb3971ab2015-05-12 18:49:09 +02001924 peer->reconnect = tick_add(now_ms, MS_TO_TICKS(5000));
1925 peer->statuscode = PEER_SESS_SC_CONNECTCODE;
Willy Tarreaud990baf2015-04-05 00:32:03 +02001926 s = NULL;
1927
Emeric Brun1138fd02017-06-19 12:38:55 +02001928 appctx = appctx_new(&peer_applet, tid_bit);
Willy Tarreaud990baf2015-04-05 00:32:03 +02001929 if (!appctx)
1930 goto out_close;
1931
1932 appctx->st0 = PEER_SESS_ST_CONNECT;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001933 appctx->ctx.peers.ptr = (void *)peer;
Willy Tarreaud990baf2015-04-05 00:32:03 +02001934
Willy Tarreau04b92862017-09-15 11:01:04 +02001935 sess = session_new(p, NULL, &appctx->obj_type);
Willy Tarreau15b5e142015-04-04 14:38:25 +02001936 if (!sess) {
Christopher Faulet767a84b2017-11-24 16:50:31 +01001937 ha_alert("out of memory in peer_session_create().\n");
Willy Tarreaud990baf2015-04-05 00:32:03 +02001938 goto out_free_appctx;
Emeric Brun2b920a12010-09-23 18:30:22 +02001939 }
1940
Willy Tarreau87787ac2017-08-28 16:22:54 +02001941 if ((s = stream_new(sess, &appctx->obj_type)) == NULL) {
Christopher Faulet767a84b2017-11-24 16:50:31 +01001942 ha_alert("Failed to initialize stream in peer_session_create().\n");
Willy Tarreau87787ac2017-08-28 16:22:54 +02001943 goto out_free_sess;
Willy Tarreau8baf9062015-04-05 00:46:36 +02001944 }
1945
Willy Tarreau342bfb12015-04-05 01:35:34 +02001946 /* The tasks below are normally what is supposed to be done by
1947 * fe->accept().
1948 */
Willy Tarreaue7dff022015-04-03 01:14:29 +02001949 s->flags = SF_ASSIGNED|SF_ADDR_SET;
Emeric Brun2b920a12010-09-23 18:30:22 +02001950
Willy Tarreau6e2979c2015-04-27 13:21:15 +02001951 /* applet is waiting for data */
1952 si_applet_cant_get(&s->si[0]);
1953 appctx_wakeup(appctx);
1954
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001955 /* initiate an outgoing connection */
Willy Tarreaudbd02672017-12-06 17:39:53 +01001956 s->si[1].flags |= SI_FL_NOLINGER;
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001957 si_set_state(&s->si[1], SI_ST_ASS);
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001958
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001959 /* automatically prepare the stream interface to connect to the
Willy Tarreaub363a1f2013-10-01 10:45:07 +02001960 * pre-initialized connection in si->conn.
1961 */
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001962 if (unlikely((conn = conn_new()) == NULL))
Willy Tarreau8baf9062015-04-05 00:46:36 +02001963 goto out_free_strm;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001964
Olivier Houchard9aaf7782017-09-13 18:30:23 +02001965 if (unlikely((cs = cs_new(conn)) == NULL))
1966 goto out_free_conn;
1967
Willy Tarreaube373152018-09-06 11:45:30 +02001968 conn->target = s->target = &s->be->obj_type;
1969 memcpy(&conn->addr.to, &peer->addr, sizeof(conn->addr.to));
1970
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001971 conn_prepare(conn, peer->proto, peer->xprt);
Willy Tarreau175a2bb2018-09-12 12:02:05 +02001972 conn_install_mux(conn, &mux_pt_ops, cs, s->be);
Olivier Houchard9aaf7782017-09-13 18:30:23 +02001973 si_attach_cs(&s->si[1], cs);
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001974
Emeric Brun2b920a12010-09-23 18:30:22 +02001975 s->do_log = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001976 s->uniq_id = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +02001977
Willy Tarreau22ec1ea2014-11-27 20:45:39 +01001978 s->res.flags |= CF_READ_DONTWAIT;
Willy Tarreau696a2912014-11-24 11:36:57 +01001979
Emeric Brunb3971ab2015-05-12 18:49:09 +02001980 peer->appctx = appctx;
Willy Tarreau87787ac2017-08-28 16:22:54 +02001981 task_wakeup(s->task, TASK_WOKEN_INIT);
Willy Tarreau9df94c22016-10-31 18:42:52 +01001982 return appctx;
Emeric Brun2b920a12010-09-23 18:30:22 +02001983
1984 /* Error unrolling */
Olivier Houchard9aaf7782017-09-13 18:30:23 +02001985 out_free_conn:
1986 conn_free(conn);
Willy Tarreau15b5e142015-04-04 14:38:25 +02001987 out_free_strm:
Emeric Brun2b920a12010-09-23 18:30:22 +02001988 LIST_DEL(&s->list);
Willy Tarreaubafbe012017-11-24 17:34:44 +01001989 pool_free(pool_head_stream, s);
Willy Tarreau15b5e142015-04-04 14:38:25 +02001990 out_free_sess:
Willy Tarreau11c36242015-04-04 15:54:03 +02001991 session_free(sess);
Willy Tarreaud990baf2015-04-05 00:32:03 +02001992 out_free_appctx:
1993 appctx_free(appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02001994 out_close:
Willy Tarreaub21d08e2016-10-31 17:46:57 +01001995 return NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001996}
1997
1998/*
1999 * Task processing function to manage re-connect and peer session
2000 * tasks wakeup on local update.
2001 */
Olivier Houchard9f6af332018-05-25 14:04:04 +02002002static struct task *process_peer_sync(struct task * task, void *context, unsigned short state)
Emeric Brun2b920a12010-09-23 18:30:22 +02002003{
Olivier Houchard9f6af332018-05-25 14:04:04 +02002004 struct peers *peers = context;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002005 struct peer *ps;
2006 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +02002007
2008 task->expire = TICK_ETERNITY;
2009
Emeric Brunb3971ab2015-05-12 18:49:09 +02002010 if (!peers->peers_fe) {
Willy Tarreau46dc1ca2015-05-01 18:32:13 +02002011 /* this one was never started, kill it */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002012 signal_unregister_handler(peers->sighandler);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002013 task_delete(peers->sync_task);
2014 task_free(peers->sync_task);
Willy Tarreau37bb7be2015-09-21 15:24:58 +02002015 peers->sync_task = NULL;
Willy Tarreau46dc1ca2015-05-01 18:32:13 +02002016 return NULL;
2017 }
2018
Emeric Brun80527f52017-06-19 17:46:37 +02002019 /* Acquire lock for all peers of the section */
2020 for (ps = peers->remote; ps; ps = ps->next)
Christopher Faulet2a944ee2017-11-07 10:42:54 +01002021 HA_SPIN_LOCK(PEER_LOCK, &ps->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02002022
Emeric Brun2b920a12010-09-23 18:30:22 +02002023 if (!stopping) {
2024 /* Normal case (not soft stop)*/
Emeric Brunb3971ab2015-05-12 18:49:09 +02002025
2026 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL) &&
2027 (!nb_oldpids || tick_is_expired(peers->resync_timeout, now_ms)) &&
2028 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002029 /* Resync from local peer needed
2030 no peer was assigned for the lesson
2031 and no old local peer found
2032 or resync timeout expire */
2033
2034 /* flag no more resync from local, to try resync from remotes */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002035 peers->flags |= PEERS_F_RESYNC_LOCAL;
Emeric Brun2b920a12010-09-23 18:30:22 +02002036
2037 /* reschedule a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002038 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
Emeric Brun2b920a12010-09-23 18:30:22 +02002039 }
2040
2041 /* For each session */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002042 for (ps = peers->remote; ps; ps = ps->next) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002043 /* For each remote peers */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002044 if (!ps->local) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002045 if (!ps->appctx) {
2046 /* no active peer connection */
Emeric Brun2b920a12010-09-23 18:30:22 +02002047 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002048 ((ps->statuscode == PEER_SESS_SC_CONNECTCODE ||
Willy Tarreaub4e34da2015-05-20 10:39:04 +02002049 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002050 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02002051 tick_is_expired(ps->reconnect, now_ms))) {
2052 /* connection never tried
Willy Tarreau9df94c22016-10-31 18:42:52 +01002053 * or previous peer connection established with success
2054 * or previous peer connection failed while connecting
Emeric Brun2b920a12010-09-23 18:30:22 +02002055 * and reconnection timer is expired */
2056
2057 /* retry a connect */
Willy Tarreau9df94c22016-10-31 18:42:52 +01002058 ps->appctx = peer_session_create(peers, ps);
Emeric Brun2b920a12010-09-23 18:30:22 +02002059 }
Willy Tarreaub4e34da2015-05-20 10:39:04 +02002060 else if (!tick_is_expired(ps->reconnect, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002061 /* If previous session failed during connection
2062 * but reconnection timer is not expired */
2063
2064 /* reschedule task for reconnect */
2065 task->expire = tick_first(task->expire, ps->reconnect);
2066 }
2067 /* else do nothing */
Willy Tarreau9df94c22016-10-31 18:42:52 +01002068 } /* !ps->appctx */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002069 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002070 /* current peer connection is active and established */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002071 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
2072 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02002073 !(ps->flags & PEER_F_LEARN_NOTUP2DATE)) {
2074 /* Resync from a remote is needed
2075 * and no peer was assigned for lesson
2076 * and current peer may be up2date */
2077
2078 /* assign peer for the lesson */
2079 ps->flags |= PEER_F_LEARN_ASSIGN;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002080 peers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brun2b920a12010-09-23 18:30:22 +02002081
Willy Tarreau9df94c22016-10-31 18:42:52 +01002082 /* wake up peer handler to handle a request of resync */
Willy Tarreaue5843b32015-04-27 18:40:14 +02002083 appctx_wakeup(ps->appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02002084 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02002085 else {
2086 /* Awake session if there is data to push */
2087 for (st = ps->tables; st ; st = st->next) {
2088 if ((int)(st->last_pushed - st->table->localupdate) < 0) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002089 /* wake up the peer handler to push local updates */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002090 appctx_wakeup(ps->appctx);
2091 break;
2092 }
2093 }
Emeric Brun2b920a12010-09-23 18:30:22 +02002094 }
2095 /* else do nothing */
2096 } /* SUCCESSCODE */
2097 } /* !ps->peer->local */
2098 } /* for */
2099
2100 /* Resync from remotes expired: consider resync is finished */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002101 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
2102 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
2103 tick_is_expired(peers->resync_timeout, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002104 /* Resync from remote peer needed
2105 * no peer was assigned for the lesson
2106 * and resync timeout expire */
2107
2108 /* flag no more resync from remote, consider resync is finished */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002109 peers->flags |= PEERS_F_RESYNC_REMOTE;
Emeric Brun2b920a12010-09-23 18:30:22 +02002110 }
2111
Emeric Brunb3971ab2015-05-12 18:49:09 +02002112 if ((peers->flags & PEERS_RESYNC_STATEMASK) != PEERS_RESYNC_FINISHED) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002113 /* Resync not finished*/
Frédéric Lécaille5d6e5f82017-05-29 13:47:16 +02002114 /* reschedule task to resync timeout if not expired, to ended resync if needed */
2115 if (!tick_is_expired(peers->resync_timeout, now_ms))
2116 task->expire = tick_first(task->expire, peers->resync_timeout);
Emeric Brun2b920a12010-09-23 18:30:22 +02002117 }
2118 } /* !stopping */
2119 else {
2120 /* soft stop case */
2121 if (task->state & TASK_WOKEN_SIGNAL) {
2122 /* We've just recieved the signal */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002123 if (!(peers->flags & PEERS_F_DONOTSTOP)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002124 /* add DO NOT STOP flag if not present */
Christopher Faulet8d8aa0d2017-05-30 15:36:50 +02002125 HA_ATOMIC_ADD(&jobs, 1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002126 peers->flags |= PEERS_F_DONOTSTOP;
2127 ps = peers->local;
2128 for (st = ps->tables; st ; st = st->next)
2129 st->table->syncing++;
Emeric Brun2b920a12010-09-23 18:30:22 +02002130 }
2131
2132 /* disconnect all connected peers */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002133 for (ps = peers->remote; ps; ps = ps->next) {
Emeric Brun80527f52017-06-19 17:46:37 +02002134 /* we're killing a connection, we must apply a random delay before
2135 * retrying otherwise the other end will do the same and we can loop
2136 * for a while.
2137 */
2138 ps->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + random() % 2000));
Willy Tarreau9df94c22016-10-31 18:42:52 +01002139 if (ps->appctx) {
Willy Tarreau81bc3b02016-10-31 17:37:39 +01002140 peer_session_forceshutdown(ps->appctx);
Willy Tarreaue5843b32015-04-27 18:40:14 +02002141 ps->appctx = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02002142 }
2143 }
2144 }
Emeric Brun2b920a12010-09-23 18:30:22 +02002145
Emeric Brunb3971ab2015-05-12 18:49:09 +02002146 ps = peers->local;
Emeric Brun2b920a12010-09-23 18:30:22 +02002147 if (ps->flags & PEER_F_TEACH_COMPLETE) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02002148 if (peers->flags & PEERS_F_DONOTSTOP) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002149 /* resync of new process was complete, current process can die now */
Willy Tarreaucea85372017-11-29 14:49:30 +01002150 HA_ATOMIC_SUB(&jobs, 1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002151 peers->flags &= ~PEERS_F_DONOTSTOP;
2152 for (st = ps->tables; st ; st = st->next)
2153 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02002154 }
2155 }
Willy Tarreau9df94c22016-10-31 18:42:52 +01002156 else if (!ps->appctx) {
2157 /* If there's no active peer connection */
Emeric Brun2b920a12010-09-23 18:30:22 +02002158 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002159 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
2160 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE ||
2161 ps->statuscode == PEER_SESS_SC_TRYAGAIN) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002162 /* connection never tried
Willy Tarreau9df94c22016-10-31 18:42:52 +01002163 * or previous peer connection was successfully established
2164 * or previous tcp connect succeeded but init state incomplete
Emeric Brun2b920a12010-09-23 18:30:22 +02002165 * or during previous connect, peer replies a try again statuscode */
2166
2167 /* connect to the peer */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002168 peer_session_create(peers, ps);
Emeric Brun2b920a12010-09-23 18:30:22 +02002169 }
2170 else {
2171 /* Other error cases */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002172 if (peers->flags & PEERS_F_DONOTSTOP) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002173 /* unable to resync new process, current process can die now */
Christopher Faulet8d8aa0d2017-05-30 15:36:50 +02002174 HA_ATOMIC_SUB(&jobs, 1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002175 peers->flags &= ~PEERS_F_DONOTSTOP;
2176 for (st = ps->tables; st ; st = st->next)
2177 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02002178 }
2179 }
2180 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02002181 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE ) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002182 /* current peer connection is active and established
2183 * wake up all peer handlers to push remaining local updates */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002184 for (st = ps->tables; st ; st = st->next) {
2185 if ((int)(st->last_pushed - st->table->localupdate) < 0) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02002186 appctx_wakeup(ps->appctx);
2187 break;
2188 }
2189 }
Emeric Brun2b920a12010-09-23 18:30:22 +02002190 }
2191 } /* stopping */
Emeric Brun80527f52017-06-19 17:46:37 +02002192
2193 /* Release lock for all peers of the section */
2194 for (ps = peers->remote; ps; ps = ps->next)
Christopher Faulet2a944ee2017-11-07 10:42:54 +01002195 HA_SPIN_UNLOCK(PEER_LOCK, &ps->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02002196
Emeric Brun2b920a12010-09-23 18:30:22 +02002197 /* Wakeup for re-connect */
2198 return task;
2199}
2200
Emeric Brunb3971ab2015-05-12 18:49:09 +02002201
Emeric Brun2b920a12010-09-23 18:30:22 +02002202/*
Willy Tarreaud9443442018-10-15 11:18:03 +02002203 * returns 0 in case of error.
Emeric Brun2b920a12010-09-23 18:30:22 +02002204 */
Willy Tarreaud9443442018-10-15 11:18:03 +02002205int peers_init_sync(struct peers *peers)
Emeric Brun2b920a12010-09-23 18:30:22 +02002206{
Emeric Brun2b920a12010-09-23 18:30:22 +02002207 struct peer * curpeer;
Willy Tarreau4348fad2012-09-20 16:48:07 +02002208 struct listener *listener;
Emeric Brun2b920a12010-09-23 18:30:22 +02002209
Emeric Brun2b920a12010-09-23 18:30:22 +02002210 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002211 peers->peers_fe->maxconn += 3;
2212 }
2213
Willy Tarreau4348fad2012-09-20 16:48:07 +02002214 list_for_each_entry(listener, &peers->peers_fe->conf.listeners, by_fe)
2215 listener->maxconn = peers->peers_fe->maxconn;
Emeric Brunc60def82017-09-27 14:59:38 +02002216 peers->sync_task = task_new(MAX_THREADS_MASK);
Willy Tarreaud9443442018-10-15 11:18:03 +02002217 if (!peers->sync_task)
2218 return 0;
2219
Emeric Brunb3971ab2015-05-12 18:49:09 +02002220 peers->sync_task->process = process_peer_sync;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002221 peers->sync_task->context = (void *)peers;
2222 peers->sighandler = signal_register_task(0, peers->sync_task, 0);
2223 task_wakeup(peers->sync_task, TASK_WOKEN_INIT);
Willy Tarreaud9443442018-10-15 11:18:03 +02002224 return 1;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002225}
2226
2227
2228
2229/*
2230 * Function used to register a table for sync on a group of peers
2231 *
2232 */
2233void peers_register_table(struct peers *peers, struct stktable *table)
2234{
2235 struct shared_table *st;
2236 struct peer * curpeer;
2237 int id = 0;
2238
2239 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
Vincent Bernat02779b62016-04-03 13:48:43 +02002240 st = calloc(1,sizeof(*st));
Emeric Brunb3971ab2015-05-12 18:49:09 +02002241 st->table = table;
2242 st->next = curpeer->tables;
2243 if (curpeer->tables)
2244 id = curpeer->tables->local_id;
2245 st->local_id = id + 1;
2246
2247 curpeer->tables = st;
2248 }
2249
2250 table->sync_task = peers->sync_task;
Emeric Brun2b920a12010-09-23 18:30:22 +02002251}
2252