blob: f26c3dfb9edda103b0055e4cf7733fd69c2f4613 [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/proto_tcp.h>
43#include <proto/proto_http.h>
44#include <proto/proxy.h>
Willy Tarreaufeb76402015-04-03 14:10:06 +020045#include <proto/session.h>
Willy Tarreau87b09662015-04-03 00:22:06 +020046#include <proto/stream.h>
Willy Tarreau22ec1ea2014-11-27 20:45:39 +010047#include <proto/signal.h>
48#include <proto/stick_table.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020049#include <proto/stream_interface.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020050#include <proto/task.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020051
52
53/*******************************/
54/* Current peer learning state */
55/*******************************/
56
57/******************************/
Emeric Brunb3971ab2015-05-12 18:49:09 +020058/* Current peers section resync state */
Emeric Brun2b920a12010-09-23 18:30:22 +020059/******************************/
Emeric Brunb3971ab2015-05-12 18:49:09 +020060#define PEERS_F_RESYNC_LOCAL 0x00000001 /* Learn from local finished or no more needed */
61#define PEERS_F_RESYNC_REMOTE 0x00000002 /* Learn from remote finished or no more needed */
62#define PEERS_F_RESYNC_ASSIGN 0x00000004 /* A peer was assigned to learn our lesson */
63#define PEERS_F_RESYNC_PROCESS 0x00000008 /* The assigned peer was requested for resync */
64#define PEERS_F_DONOTSTOP 0x00010000 /* Main table sync task block process during soft stop
Emeric Brun2b920a12010-09-23 18:30:22 +020065 to push data to new process */
66
Emeric Brunb3971ab2015-05-12 18:49:09 +020067#define PEERS_RESYNC_STATEMASK (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE)
68#define PEERS_RESYNC_FROMLOCAL 0x00000000
69#define PEERS_RESYNC_FROMREMOTE PEERS_F_RESYNC_LOCAL
70#define PEERS_RESYNC_FINISHED (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE)
71
72/***********************************/
73/* Current shared table sync state */
74/***********************************/
75#define SHTABLE_F_TEACH_STAGE1 0x00000001 /* Teach state 1 complete */
76#define SHTABLE_F_TEACH_STAGE2 0x00000002 /* Teach state 2 complete */
Emeric Brun2b920a12010-09-23 18:30:22 +020077
78/******************************/
79/* Remote peer teaching state */
80/******************************/
81#define PEER_F_TEACH_PROCESS 0x00000001 /* Teach a lesson to current peer */
Emeric Brun2b920a12010-09-23 18:30:22 +020082#define PEER_F_TEACH_FINISHED 0x00000008 /* Teach conclude, (wait for confirm) */
83#define PEER_F_TEACH_COMPLETE 0x00000010 /* All that we know already taught to current peer, used only for a local peer */
84#define PEER_F_LEARN_ASSIGN 0x00000100 /* Current peer was assigned for a lesson */
85#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 +020086#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 +020087
Emeric Brunb3971ab2015-05-12 18:49:09 +020088#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 +020089#define PEER_LEARN_RESET ~(PEER_F_LEARN_ASSIGN|PEER_F_LEARN_NOTUP2DATE)
90
Emeric Brunb3971ab2015-05-12 18:49:09 +020091/*****************************/
92/* Sync message class */
93/*****************************/
94enum {
95 PEER_MSG_CLASS_CONTROL = 0,
96 PEER_MSG_CLASS_ERROR,
97 PEER_MSG_CLASS_STICKTABLE = 10,
98 PEER_MSG_CLASS_RESERVED = 255,
99};
100
101/*****************************/
102/* control message types */
103/*****************************/
104enum {
105 PEER_MSG_CTRL_RESYNCREQ = 0,
106 PEER_MSG_CTRL_RESYNCFINISHED,
107 PEER_MSG_CTRL_RESYNCPARTIAL,
108 PEER_MSG_CTRL_RESYNCCONFIRM,
109};
110
111/*****************************/
112/* error message types */
113/*****************************/
114enum {
115 PEER_MSG_ERR_PROTOCOL = 0,
116 PEER_MSG_ERR_SIZELIMIT,
117};
118
119
120/*******************************/
121/* stick table sync mesg types */
122/* Note: ids >= 128 contains */
123/* id message cotains data */
124/*******************************/
125enum {
126 PEER_MSG_STKT_UPDATE = 128,
127 PEER_MSG_STKT_INCUPDATE,
128 PEER_MSG_STKT_DEFINE,
129 PEER_MSG_STKT_SWITCH,
130 PEER_MSG_STKT_ACK,
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200131 PEER_MSG_STKT_UPDATE_TIMED,
132 PEER_MSG_STKT_INCUPDATE_TIMED,
Emeric Brunb3971ab2015-05-12 18:49:09 +0200133};
Emeric Brun2b920a12010-09-23 18:30:22 +0200134
135/**********************************/
136/* Peer Session IO handler states */
137/**********************************/
138
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100139enum {
140 PEER_SESS_ST_ACCEPT = 0, /* Initial state for session create by an accept, must be zero! */
141 PEER_SESS_ST_GETVERSION, /* Validate supported protocol version */
142 PEER_SESS_ST_GETHOST, /* Validate host ID correspond to local host id */
143 PEER_SESS_ST_GETPEER, /* Validate peer ID correspond to a known remote peer id */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100144 /* after this point, data were possibly exchanged */
145 PEER_SESS_ST_SENDSUCCESS, /* Send ret code 200 (success) and wait for message */
146 PEER_SESS_ST_CONNECT, /* Initial state for session create on a connect, push presentation into buffer */
147 PEER_SESS_ST_GETSTATUS, /* Wait for the welcome message */
148 PEER_SESS_ST_WAITMSG, /* Wait for data messages */
149 PEER_SESS_ST_EXIT, /* Exit with status code */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200150 PEER_SESS_ST_ERRPROTO, /* Send error proto message before exit */
151 PEER_SESS_ST_ERRSIZE, /* Send error size message before exit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100152 PEER_SESS_ST_END, /* Killed session */
153};
Emeric Brun2b920a12010-09-23 18:30:22 +0200154
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100155/***************************************************/
156/* Peer Session status code - part of the protocol */
157/***************************************************/
Emeric Brun2b920a12010-09-23 18:30:22 +0200158
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100159#define PEER_SESS_SC_CONNECTCODE 100 /* connect in progress */
160#define PEER_SESS_SC_CONNECTEDCODE 110 /* tcp connect success */
Emeric Brun2b920a12010-09-23 18:30:22 +0200161
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100162#define PEER_SESS_SC_SUCCESSCODE 200 /* accept or connect successful */
Emeric Brun2b920a12010-09-23 18:30:22 +0200163
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100164#define PEER_SESS_SC_TRYAGAIN 300 /* try again later */
Emeric Brun2b920a12010-09-23 18:30:22 +0200165
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100166#define PEER_SESS_SC_ERRPROTO 501 /* error protocol */
167#define PEER_SESS_SC_ERRVERSION 502 /* unknown protocol version */
168#define PEER_SESS_SC_ERRHOST 503 /* bad host name */
169#define PEER_SESS_SC_ERRPEER 504 /* unknown peer */
Emeric Brun2b920a12010-09-23 18:30:22 +0200170
171#define PEER_SESSION_PROTO_NAME "HAProxyS"
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200172#define PEER_MAJOR_VER 2
173#define PEER_MINOR_VER 1
174#define PEER_DWNGRD_MINOR_VER 0
Emeric Brun2b920a12010-09-23 18:30:22 +0200175
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200176struct peers *cfg_peers = NULL;
Willy Tarreau81bc3b02016-10-31 17:37:39 +0100177static void peer_session_forceshutdown(struct appctx *appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +0200178
Emeric Brun18928af2017-03-29 16:32:53 +0200179/* This function encode an uint64 to 'dynamic' length format.
180 The encoded value is written at address *str, and the
181 caller must assure that size after *str is large enought.
182 At return, the *str is set at the next Byte after then
183 encoded integer. The function returns then length of the
184 encoded integer in Bytes */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200185int intencode(uint64_t i, char **str) {
186 int idx = 0;
187 unsigned char *msg;
188
189 if (!*str)
190 return 0;
191
192 msg = (unsigned char *)*str;
193 if (i < 240) {
194 msg[0] = (unsigned char)i;
195 *str = (char *)&msg[idx+1];
196 return (idx+1);
197 }
198
199 msg[idx] =(unsigned char)i | 240;
200 i = (i - 240) >> 4;
201 while (i >= 128) {
202 msg[++idx] = (unsigned char)i | 128;
203 i = (i - 128) >> 7;
204 }
205 msg[++idx] = (unsigned char)i;
206 *str = (char *)&msg[idx+1];
207 return (idx+1);
208}
209
210
211/* This function returns the decoded integer or 0
212 if decode failed
213 *str point on the beginning of the integer to decode
214 at the end of decoding *str point on the end of the
215 encoded integer or to null if end is reached */
Emeric Brun18928af2017-03-29 16:32:53 +0200216uint64_t intdecode(char **str, char *end)
217{
Emeric Brunb3971ab2015-05-12 18:49:09 +0200218 unsigned char *msg;
Emeric Brun18928af2017-03-29 16:32:53 +0200219 uint64_t i;
220 int shift;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200221
222 if (!*str)
223 return 0;
224
225 msg = (unsigned char *)*str;
Emeric Brun18928af2017-03-29 16:32:53 +0200226 if (msg >= (unsigned char *)end)
227 goto fail;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200228
Emeric Brun18928af2017-03-29 16:32:53 +0200229 i = *(msg++);
230 if (i >= 240) {
231 shift = 4;
232 do {
233 if (msg >= (unsigned char *)end)
234 goto fail;
235 i += (uint64_t)*msg << shift;
236 shift += 7;
237 } while (*(msg++) >= 128);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200238 }
Emeric Brun18928af2017-03-29 16:32:53 +0200239 *str = (char *)msg;
240 return i;
241
242 fail:
243 *str = NULL;
244 return 0;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200245}
Emeric Brun2b920a12010-09-23 18:30:22 +0200246
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200247/* Set the stick-table UPDATE message type byte at <msg_type> address,
248 * depending on <use_identifier> and <use_timed> boolean parameters.
249 * Always successful.
250 */
251static inline void peer_set_update_msg_type(char *msg_type, int use_identifier, int use_timed)
252{
253 if (use_timed) {
254 if (use_identifier)
255 *msg_type = PEER_MSG_STKT_UPDATE_TIMED;
256 else
257 *msg_type = PEER_MSG_STKT_INCUPDATE_TIMED;
258 }
259 else {
260 if (use_identifier)
261 *msg_type = PEER_MSG_STKT_UPDATE;
262 else
263 *msg_type = PEER_MSG_STKT_INCUPDATE;
264 }
265
266}
Emeric Brun2b920a12010-09-23 18:30:22 +0200267/*
Emeric Brunb3971ab2015-05-12 18:49:09 +0200268 * This prepare the data update message on the stick session <ts>, <st> is the considered
269 * stick table.
270 * <msg> is a buffer of <size> to recieve data message content
271 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
272 * check size)
Emeric Brun2b920a12010-09-23 18:30:22 +0200273 */
Emeric Brun819fc6f2017-06-13 19:37:32 +0200274static 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 +0200275{
276 uint32_t netinteger;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200277 unsigned short datalen;
278 char *cursor, *datamsg;
Emeric Brun94900952015-06-11 18:25:54 +0200279 unsigned int data_type;
280 void *data_ptr;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200281
282 cursor = datamsg = msg + 1 + 5;
283
Emeric Brun2b920a12010-09-23 18:30:22 +0200284 /* construct message */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200285
286 /* check if we need to send the update identifer */
Emeric Brun819fc6f2017-06-13 19:37:32 +0200287 if (!st->last_pushed || updateid < st->last_pushed || ((updateid - st->last_pushed) != 1)) {
Emeric Bruna6a09982015-09-22 15:34:19 +0200288 use_identifier = 1;
Emeric Brun2b920a12010-09-23 18:30:22 +0200289 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200290
291 /* encode update identifier if needed */
292 if (use_identifier) {
Emeric Brun819fc6f2017-06-13 19:37:32 +0200293 netinteger = htonl(updateid);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200294 memcpy(cursor, &netinteger, sizeof(netinteger));
295 cursor += sizeof(netinteger);
Emeric Brun2b920a12010-09-23 18:30:22 +0200296 }
297
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200298 if (use_timed) {
299 netinteger = htonl(tick_remain(now_ms, ts->expire));
300 memcpy(cursor, &netinteger, sizeof(netinteger));
301 cursor += sizeof(netinteger);
302 }
303
Emeric Brunb3971ab2015-05-12 18:49:09 +0200304 /* encode the key */
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +0200305 if (st->table->type == SMP_T_STR) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200306 int stlen = strlen((char *)ts->key.key);
307
Emeric Brunb3971ab2015-05-12 18:49:09 +0200308 intencode(stlen, &cursor);
309 memcpy(cursor, ts->key.key, stlen);
310 cursor += stlen;
Emeric Brun2b920a12010-09-23 18:30:22 +0200311 }
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +0200312 else if (st->table->type == SMP_T_SINT) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200313 netinteger = htonl(*((uint32_t *)ts->key.key));
Emeric Brunb3971ab2015-05-12 18:49:09 +0200314 memcpy(cursor, &netinteger, sizeof(netinteger));
315 cursor += sizeof(netinteger);
Emeric Brun2b920a12010-09-23 18:30:22 +0200316 }
317 else {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200318 memcpy(cursor, ts->key.key, st->table->key_size);
319 cursor += st->table->key_size;
Emeric Brun2b920a12010-09-23 18:30:22 +0200320 }
321
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100322 HA_RWLOCK_RDLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200323 /* encode values */
Emeric Brun94900952015-06-11 18:25:54 +0200324 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200325
Emeric Brun94900952015-06-11 18:25:54 +0200326 data_ptr = stktable_data_ptr(st->table, ts, data_type);
327 if (data_ptr) {
328 switch (stktable_data_types[data_type].std_type) {
329 case STD_T_SINT: {
330 int data;
331
332 data = stktable_data_cast(data_ptr, std_t_sint);
333 intencode(data, &cursor);
334 break;
335 }
336 case STD_T_UINT: {
337 unsigned int data;
338
339 data = stktable_data_cast(data_ptr, std_t_uint);
340 intencode(data, &cursor);
341 break;
342 }
343 case STD_T_ULL: {
344 unsigned long long data;
345
346 data = stktable_data_cast(data_ptr, std_t_ull);
347 intencode(data, &cursor);
348 break;
349 }
350 case STD_T_FRQP: {
351 struct freq_ctr_period *frqp;
352
353 frqp = &stktable_data_cast(data_ptr, std_t_frqp);
354 intencode((unsigned int)(now_ms - frqp->curr_tick), &cursor);
355 intencode(frqp->curr_ctr, &cursor);
356 intencode(frqp->prev_ctr, &cursor);
357 break;
358 }
359 }
360 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200361 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100362 HA_RWLOCK_RDUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200363
364 /* Compute datalen */
365 datalen = (cursor - datamsg);
366
367 /* prepare message header */
368 msg[0] = PEER_MSG_CLASS_STICKTABLE;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200369 peer_set_update_msg_type(&msg[1], use_identifier, use_timed);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200370 cursor = &msg[2];
371 intencode(datalen, &cursor);
372
373 /* move data after header */
374 memmove(cursor, datamsg, datalen);
375
376 /* return header size + data_len */
377 return (cursor - msg) + datalen;
378}
379
380/*
381 * This prepare the switch table message to targeted share table <st>.
382 * <msg> is a buffer of <size> to recieve data message content
383 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
384 * check size)
385 */
386static int peer_prepare_switchmsg(struct shared_table *st, char *msg, size_t size)
387{
388 int len;
389 unsigned short datalen;
Willy Tarreau83061a82018-07-13 11:56:34 +0200390 struct buffer *chunk;
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200391 char *cursor, *datamsg, *chunkp, *chunkq;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200392 uint64_t data = 0;
Emeric Brun94900952015-06-11 18:25:54 +0200393 unsigned int data_type;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200394
395 cursor = datamsg = msg + 2 + 5;
396
397 /* Encode data */
398
399 /* encode local id */
400 intencode(st->local_id, &cursor);
401
402 /* encode table name */
403 len = strlen(st->table->id);
404 intencode(len, &cursor);
405 memcpy(cursor, st->table->id, len);
406 cursor += len;
407
408 /* encode table type */
409
410 intencode(st->table->type, &cursor);
411
412 /* encode table key size */
413 intencode(st->table->key_size, &cursor);
414
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200415 chunk = get_trash_chunk();
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200416 chunkp = chunkq = chunk->area;
Emeric Brun94900952015-06-11 18:25:54 +0200417 /* encode available known data types in table */
418 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
419 if (st->table->data_ofs[data_type]) {
420 switch (stktable_data_types[data_type].std_type) {
421 case STD_T_SINT:
422 case STD_T_UINT:
423 case STD_T_ULL:
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200424 data |= 1 << data_type;
425 break;
Emeric Brun94900952015-06-11 18:25:54 +0200426 case STD_T_FRQP:
427 data |= 1 << data_type;
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200428 intencode(data_type, &chunkq);
429 intencode(st->table->data_arg[data_type].u, &chunkq);
Emeric Brun94900952015-06-11 18:25:54 +0200430 break;
431 }
432 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200433 }
434 intencode(data, &cursor);
435
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200436 /* Encode stick-table entries duration. */
437 intencode(st->table->expire, &cursor);
438
439 if (chunkq > chunkp) {
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200440 chunk->data = chunkq - chunkp;
441 memcpy(cursor, chunk->area, chunk->data);
442 cursor += chunk->data;
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200443 }
444
Emeric Brunb3971ab2015-05-12 18:49:09 +0200445 /* Compute datalen */
446 datalen = (cursor - datamsg);
Emeric Brun2b920a12010-09-23 18:30:22 +0200447
Emeric Brunb3971ab2015-05-12 18:49:09 +0200448 /* prepare message header */
449 msg[0] = PEER_MSG_CLASS_STICKTABLE;
450 msg[1] = PEER_MSG_STKT_DEFINE;
451 cursor = &msg[2];
452 intencode(datalen, &cursor);
Emeric Brun2b920a12010-09-23 18:30:22 +0200453
Emeric Brunb3971ab2015-05-12 18:49:09 +0200454 /* move data after header */
455 memmove(cursor, datamsg, datalen);
456
457 /* return header size + data_len */
458 return (cursor - msg) + datalen;
Emeric Brun2b920a12010-09-23 18:30:22 +0200459}
460
Emeric Brunb3971ab2015-05-12 18:49:09 +0200461/*
462 * This prepare the acknowledge message on the stick session <ts>, <st> is the considered
463 * stick table.
464 * <msg> is a buffer of <size> to recieve data message content
465 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
466 * check size)
467 */
468static int peer_prepare_ackmsg(struct shared_table *st, char *msg, size_t size)
469{
470 unsigned short datalen;
471 char *cursor, *datamsg;
472 uint32_t netinteger;
473
Emeric Brunb058f1c2015-09-22 15:50:18 +0200474 cursor = datamsg = msg + 2 + 5;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200475
476 intencode(st->remote_id, &cursor);
477 netinteger = htonl(st->last_get);
478 memcpy(cursor, &netinteger, sizeof(netinteger));
479 cursor += sizeof(netinteger);
480
481 /* Compute datalen */
482 datalen = (cursor - datamsg);
483
484 /* prepare message header */
485 msg[0] = PEER_MSG_CLASS_STICKTABLE;
Emeric Brune1ab8082015-08-21 11:48:54 +0200486 msg[1] = PEER_MSG_STKT_ACK;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200487 cursor = &msg[2];
488 intencode(datalen, &cursor);
489
490 /* move data after header */
491 memmove(cursor, datamsg, datalen);
492
493 /* return header size + data_len */
494 return (cursor - msg) + datalen;
495}
Emeric Brun2b920a12010-09-23 18:30:22 +0200496
497/*
498 * Callback to release a session with a peer
499 */
Willy Tarreau00a37f02015-04-13 12:05:19 +0200500static void peer_session_release(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +0200501{
Willy Tarreau00a37f02015-04-13 12:05:19 +0200502 struct stream_interface *si = appctx->owner;
Willy Tarreau87b09662015-04-03 00:22:06 +0200503 struct stream *s = si_strm(si);
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200504 struct peer *peer = appctx->ctx.peers.ptr;
505 struct peers *peers = strm_fe(s)->parent;
Emeric Brun2b920a12010-09-23 18:30:22 +0200506
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100507 /* appctx->ctx.peers.ptr is not a peer session */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100508 if (appctx->st0 < PEER_SESS_ST_SENDSUCCESS)
Emeric Brun2b920a12010-09-23 18:30:22 +0200509 return;
510
511 /* peer session identified */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200512 if (peer) {
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100513 HA_SPIN_LOCK(PEER_LOCK, &peer->lock);
Willy Tarreau9df94c22016-10-31 18:42:52 +0100514 if (peer->appctx == appctx) {
Emeric Brunb157d732015-08-21 12:00:30 +0200515 /* Re-init current table pointers to force announcement on re-connect */
516 peer->remote_table = peer->last_local_table = NULL;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200517 peer->appctx = NULL;
518 if (peer->flags & PEER_F_LEARN_ASSIGN) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200519 /* unassign current peer for learning */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200520 peer->flags &= ~(PEER_F_LEARN_ASSIGN);
521 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brun2b920a12010-09-23 18:30:22 +0200522
523 /* reschedule a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200524 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
Emeric Brun2b920a12010-09-23 18:30:22 +0200525 }
526 /* reset teaching and learning flags to 0 */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200527 peer->flags &= PEER_TEACH_RESET;
528 peer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200529 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100530 HA_SPIN_UNLOCK(PEER_LOCK, &peer->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200531 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200532 }
533}
534
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200535/* Retrieve the major and minor versions of peers protocol
536 * announced by a remote peer. <str> is a null-terminated
537 * string with the following format: "<maj_ver>.<min_ver>".
538 */
539static int peer_get_version(const char *str,
540 unsigned int *maj_ver, unsigned int *min_ver)
541{
542 unsigned int majv, minv;
543 const char *pos, *saved;
544 const char *end;
545
546 saved = pos = str;
547 end = str + strlen(str);
548
549 majv = read_uint(&pos, end);
550 if (saved == pos || *pos++ != '.')
551 return -1;
552
553 saved = pos;
554 minv = read_uint(&pos, end);
555 if (saved == pos || pos != end)
556 return -1;
557
558 *maj_ver = majv;
559 *min_ver = minv;
560
561 return 0;
562}
Emeric Brun2b920a12010-09-23 18:30:22 +0200563
564/*
565 * IO Handler to handle message exchance with a peer
566 */
Willy Tarreau00a37f02015-04-13 12:05:19 +0200567static void peer_io_handler(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +0200568{
Willy Tarreau00a37f02015-04-13 12:05:19 +0200569 struct stream_interface *si = appctx->owner;
Willy Tarreau87b09662015-04-03 00:22:06 +0200570 struct stream *s = si_strm(si);
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200571 struct peers *curpeers = strm_fe(s)->parent;
Emeric Brun80527f52017-06-19 17:46:37 +0200572 struct peer *curpeer = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +0200573 int reql = 0;
574 int repl = 0;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200575 size_t proto_len = strlen(PEER_SESSION_PROTO_NAME);
576 unsigned int maj_ver, min_ver;
Emeric Brun2b920a12010-09-23 18:30:22 +0200577
Christopher Fauleta73e59b2016-12-09 17:30:18 +0100578 /* Check if the input buffer is avalaible. */
Willy Tarreauc9fa0482018-07-10 17:43:27 +0200579 if (si_ic(si)->buf.size == 0)
Christopher Fauleta73e59b2016-12-09 17:30:18 +0100580 goto full;
581
Emeric Brun2b920a12010-09-23 18:30:22 +0200582 while (1) {
583switchstate:
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200584 maj_ver = min_ver = (unsigned int)-1;
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100585 switch(appctx->st0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100586 case PEER_SESS_ST_ACCEPT:
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100587 appctx->ctx.peers.ptr = NULL;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100588 appctx->st0 = PEER_SESS_ST_GETVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200589 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100590 case PEER_SESS_ST_GETVERSION:
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200591 reql = co_getline(si_oc(si), trash.area,
592 trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200593 if (reql <= 0) { /* closed or EOL not found */
594 if (reql == 0)
595 goto out;
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 if (trash.area[reql-1] != '\n') {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100600 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200601 goto switchstate;
602 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200603 else if (reql > 1 && (trash.area[reql-2] == '\r'))
604 trash.area[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200605 else
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200606 trash.area[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200607
Willy Tarreau06d80a92017-10-19 14:32:15 +0200608 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200609
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200610 /* test protocol */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200611 if (strncmp(PEER_SESSION_PROTO_NAME " ", trash.area, proto_len + 1) != 0) {
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200612 appctx->st0 = PEER_SESS_ST_EXIT;
613 appctx->st1 = PEER_SESS_SC_ERRPROTO;
614 goto switchstate;
615 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200616 if (peer_get_version(trash.area + proto_len + 1, &maj_ver, &min_ver) == -1 ||
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200617 maj_ver != PEER_MAJOR_VER || min_ver > PEER_MINOR_VER) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100618 appctx->st0 = PEER_SESS_ST_EXIT;
619 appctx->st1 = PEER_SESS_SC_ERRVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200620 goto switchstate;
621 }
622
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100623 appctx->st0 = PEER_SESS_ST_GETHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200624 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100625 case PEER_SESS_ST_GETHOST:
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200626 reql = co_getline(si_oc(si), trash.area,
627 trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200628 if (reql <= 0) { /* closed or EOL not found */
629 if (reql == 0)
630 goto out;
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 if (trash.area[reql-1] != '\n') {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100635 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200636 goto switchstate;
637 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200638 else if (reql > 1 && (trash.area[reql-2] == '\r'))
639 trash.area[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200640 else
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200641 trash.area[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200642
Willy Tarreau06d80a92017-10-19 14:32:15 +0200643 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200644
645 /* test hostname match */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200646 if (strcmp(localpeer, trash.area) != 0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100647 appctx->st0 = PEER_SESS_ST_EXIT;
648 appctx->st1 = PEER_SESS_SC_ERRHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200649 goto switchstate;
650 }
651
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100652 appctx->st0 = PEER_SESS_ST_GETPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200653 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100654 case PEER_SESS_ST_GETPEER: {
Emeric Brun2b920a12010-09-23 18:30:22 +0200655 char *p;
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200656 reql = co_getline(si_oc(si), trash.area,
657 trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200658 if (reql <= 0) { /* closed or EOL not found */
659 if (reql == 0)
660 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100661 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200662 goto switchstate;
663 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200664 if (trash.area[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200665 /* Incomplete line, we quit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100666 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200667 goto switchstate;
668 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200669 else if (reql > 1 && (trash.area[reql-2] == '\r'))
670 trash.area[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200671 else
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200672 trash.area[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200673
Willy Tarreau06d80a92017-10-19 14:32:15 +0200674 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200675
Emeric Brunb3971ab2015-05-12 18:49:09 +0200676 /* parse line "<peer name> <pid> <relative_pid>" */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200677 p = strchr(trash.area, ' ');
Emeric Brun2b920a12010-09-23 18:30:22 +0200678 if (!p) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100679 appctx->st0 = PEER_SESS_ST_EXIT;
680 appctx->st1 = PEER_SESS_SC_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +0200681 goto switchstate;
682 }
683 *p = 0;
684
685 /* lookup known peer */
686 for (curpeer = curpeers->remote; curpeer; curpeer = curpeer->next) {
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200687 if (strcmp(curpeer->id, trash.area) == 0)
Emeric Brun2b920a12010-09-23 18:30:22 +0200688 break;
689 }
690
691 /* if unknown peer */
692 if (!curpeer) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100693 appctx->st0 = PEER_SESS_ST_EXIT;
694 appctx->st1 = PEER_SESS_SC_ERRPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200695 goto switchstate;
696 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200697
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100698 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Willy Tarreau9df94c22016-10-31 18:42:52 +0100699 if (curpeer->appctx && curpeer->appctx != appctx) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200700 if (curpeer->local) {
701 /* Local connection, reply a retry */
702 appctx->st0 = PEER_SESS_ST_EXIT;
703 appctx->st1 = PEER_SESS_SC_TRYAGAIN;
704 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +0200705 }
Emeric Brun80527f52017-06-19 17:46:37 +0200706
707 /* we're killing a connection, we must apply a random delay before
708 * retrying otherwise the other end will do the same and we can loop
709 * for a while.
710 */
711 curpeer->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + random() % 2000));
Willy Tarreau81bc3b02016-10-31 17:37:39 +0100712 peer_session_forceshutdown(curpeer->appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +0200713 }
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200714 if (maj_ver != (unsigned int)-1 && min_ver != (unsigned int)-1) {
715 if (min_ver == PEER_DWNGRD_MINOR_VER) {
716 curpeer->flags |= PEER_F_DWNGRD;
717 }
718 else {
719 curpeer->flags &= ~PEER_F_DWNGRD;
720 }
721 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200722 curpeer->appctx = appctx;
723 appctx->ctx.peers.ptr = curpeer;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100724 appctx->st0 = PEER_SESS_ST_SENDSUCCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200725 /* fall through */
726 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100727 case PEER_SESS_ST_SENDSUCCESS: {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200728 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +0200729
Emeric Brun80527f52017-06-19 17:46:37 +0200730 if (!curpeer) {
731 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100732 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +0200733 if (curpeer->appctx != appctx) {
734 appctx->st0 = PEER_SESS_ST_END;
735 goto switchstate;
736 }
737 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200738 repl = snprintf(trash.area, trash.size,
739 "%d\n",
740 PEER_SESS_SC_SUCCESSCODE);
741 repl = ci_putblk(si_ic(si), trash.area, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200742 if (repl <= 0) {
743 if (repl == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +0100744 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100745 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200746 goto switchstate;
747 }
748
749 /* Register status code */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200750 curpeer->statuscode = PEER_SESS_SC_SUCCESSCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200751
752 /* Awake main task */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200753 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200754
755 /* Init confirm counter */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200756 curpeer->confirm = 0;
757
758 /* Init cursors */
759 for (st = curpeer->tables; st ; st = st->next) {
760 st->last_get = st->last_acked = 0;
761 st->teaching_origin = st->last_pushed = st->update;
762 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200763
764 /* reset teaching and learning flags to 0 */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200765 curpeer->flags &= PEER_TEACH_RESET;
766 curpeer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200767
768 /* if current peer is local */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200769 if (curpeer->local) {
770 /* if current host need resyncfrom local and no process assined */
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200771 if ((curpeers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL &&
772 !(curpeers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200773 /* assign local peer for a lesson, consider lesson already requested */
774 curpeer->flags |= PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200775 curpeers->flags |= (PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200776 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200777
Emeric Brunb3971ab2015-05-12 18:49:09 +0200778 }
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200779 else if ((curpeers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
780 !(curpeers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200781 /* assign peer for a lesson */
782 curpeer->flags |= PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200783 curpeers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200784 }
785
786
Emeric Brun2b920a12010-09-23 18:30:22 +0200787 /* switch to waiting message state */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100788 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200789 goto switchstate;
790 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100791 case PEER_SESS_ST_CONNECT: {
Emeric Brun80527f52017-06-19 17:46:37 +0200792
793 if (!curpeer) {
794 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100795 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +0200796 if (curpeer->appctx != appctx) {
797 appctx->st0 = PEER_SESS_ST_END;
798 goto switchstate;
799 }
800 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200801
802 /* Send headers */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200803 repl = snprintf(trash.area, trash.size,
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200804 PEER_SESSION_PROTO_NAME " %u.%u\n%s\n%s %d %d\n",
805 PEER_MAJOR_VER,
806 (curpeer->flags & PEER_F_DWNGRD) ? PEER_DWNGRD_MINOR_VER : PEER_MINOR_VER,
Emeric Brunb3971ab2015-05-12 18:49:09 +0200807 curpeer->id,
Emeric Brun2b920a12010-09-23 18:30:22 +0200808 localpeer,
Willy Tarreau7b77c9f2012-01-07 22:52:12 +0100809 (int)getpid(),
Emeric Brunb3971ab2015-05-12 18:49:09 +0200810 relative_pid);
Emeric Brun2b920a12010-09-23 18:30:22 +0200811
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100812 if (repl >= trash.size) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100813 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200814 goto switchstate;
815 }
816
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200817 repl = ci_putblk(si_ic(si), trash.area, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200818 if (repl <= 0) {
819 if (repl == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +0100820 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100821 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200822 goto switchstate;
823 }
824
825 /* switch to the waiting statuscode state */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100826 appctx->st0 = PEER_SESS_ST_GETSTATUS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200827 /* fall through */
828 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100829 case PEER_SESS_ST_GETSTATUS: {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200830 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +0200831
Emeric Brun80527f52017-06-19 17:46:37 +0200832 if (!curpeer) {
833 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100834 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +0200835 if (curpeer->appctx != appctx) {
836 appctx->st0 = PEER_SESS_ST_END;
837 goto switchstate;
838 }
839 }
840
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100841 if (si_ic(si)->flags & CF_WRITE_PARTIAL)
Emeric Brunb3971ab2015-05-12 18:49:09 +0200842 curpeer->statuscode = PEER_SESS_SC_CONNECTEDCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200843
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200844 reql = co_getline(si_oc(si), trash.area,
845 trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200846 if (reql <= 0) { /* closed or EOL not found */
847 if (reql == 0)
848 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100849 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200850 goto switchstate;
851 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200852 if (trash.area[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200853 /* Incomplete line, we quit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100854 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200855 goto switchstate;
856 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200857 else if (reql > 1 && (trash.area[reql-2] == '\r'))
858 trash.area[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200859 else
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200860 trash.area[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200861
Willy Tarreau06d80a92017-10-19 14:32:15 +0200862 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200863
864 /* Register status code */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200865 curpeer->statuscode = atoi(trash.area);
Emeric Brun2b920a12010-09-23 18:30:22 +0200866
867 /* Awake main task */
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200868 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200869
870 /* If status code is success */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200871 if (curpeer->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200872 /* Init cursors */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200873 for (st = curpeer->tables; st ; st = st->next) {
874 st->last_get = st->last_acked = 0;
875 st->teaching_origin = st->last_pushed = st->update;
876 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200877
878 /* Init confirm counter */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200879 curpeer->confirm = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200880
Emeric Brunb3971ab2015-05-12 18:49:09 +0200881 /* reset teaching and learning flags to 0 */
882 curpeer->flags &= PEER_TEACH_RESET;
883 curpeer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200884
Emeric Brunb3971ab2015-05-12 18:49:09 +0200885 /* If current peer is local */
886 if (curpeer->local) {
887 /* flag to start to teach lesson */
888 curpeer->flags |= PEER_F_TEACH_PROCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200889
Emeric Brunb3971ab2015-05-12 18:49:09 +0200890 }
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200891 else if ((curpeers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
892 !(curpeers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200893 /* If peer is remote and resync from remote is needed,
894 and no peer currently assigned */
Emeric Brun2b920a12010-09-23 18:30:22 +0200895
Emeric Brunb3971ab2015-05-12 18:49:09 +0200896 /* assign peer for a lesson */
897 curpeer->flags |= PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200898 curpeers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brun2b920a12010-09-23 18:30:22 +0200899 }
900
901 }
902 else {
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200903 if (curpeer->statuscode == PEER_SESS_SC_ERRVERSION)
904 curpeer->flags |= PEER_F_DWNGRD;
Emeric Brun2b920a12010-09-23 18:30:22 +0200905 /* Status code is not success, abort */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100906 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200907 goto switchstate;
908 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100909 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200910 /* fall through */
911 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100912 case PEER_SESS_ST_WAITMSG: {
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200913 struct stksess *ts, *newts = NULL;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200914 uint32_t msg_len = 0;
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200915 char *msg_cur = trash.area;
916 char *msg_end = trash.area;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200917 unsigned char msg_head[7];
Emeric Brun2b920a12010-09-23 18:30:22 +0200918 int totl = 0;
919
Emeric Brun80527f52017-06-19 17:46:37 +0200920 if (!curpeer) {
921 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100922 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +0200923 if (curpeer->appctx != appctx) {
924 appctx->st0 = PEER_SESS_ST_END;
925 goto switchstate;
926 }
927 }
928
Willy Tarreau06d80a92017-10-19 14:32:15 +0200929 reql = co_getblk(si_oc(si), (char *)msg_head, 2*sizeof(unsigned char), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200930 if (reql <= 0) /* closed or EOL not found */
931 goto incomplete;
932
Emeric Brun2b920a12010-09-23 18:30:22 +0200933 totl += reql;
934
Emeric Brunb3971ab2015-05-12 18:49:09 +0200935 if (msg_head[1] >= 128) {
936 /* Read and Decode message length */
Willy Tarreau06d80a92017-10-19 14:32:15 +0200937 reql = co_getblk(si_oc(si), (char *)&msg_head[2], sizeof(unsigned char), totl);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200938 if (reql <= 0) /* closed */
939 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +0200940
Emeric Brunb3971ab2015-05-12 18:49:09 +0200941 totl += reql;
942
943 if (msg_head[2] < 240) {
944 msg_len = msg_head[2];
Emeric Brun2b920a12010-09-23 18:30:22 +0200945 }
946 else {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200947 int i;
948 char *cur;
949 char *end;
Willy Tarreau72d6c162013-04-11 16:14:13 +0200950
Emeric Brunb3971ab2015-05-12 18:49:09 +0200951 for (i = 3 ; i < sizeof(msg_head) ; i++) {
Willy Tarreau06d80a92017-10-19 14:32:15 +0200952 reql = co_getblk(si_oc(si), (char *)&msg_head[i], sizeof(char), totl);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200953 if (reql <= 0) /* closed */
954 goto incomplete;
955
956 totl += reql;
957
958 if (!(msg_head[i] & 0x80))
959 break;
960 }
961
962 if (i == sizeof(msg_head)) {
963 /* malformed message */
964 appctx->st0 = PEER_SESS_ST_ERRPROTO;
965 goto switchstate;
966
967 }
968 end = (char *)msg_head + sizeof(msg_head);
969 cur = (char *)&msg_head[2];
970 msg_len = intdecode(&cur, end);
971 if (!cur) {
972 /* malformed message */
973 appctx->st0 = PEER_SESS_ST_ERRPROTO;
974 goto switchstate;
975 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200976 }
977
Willy Tarreau86a446e2013-11-25 23:02:37 +0100978
Emeric Brunb3971ab2015-05-12 18:49:09 +0200979 /* Read message content */
980 if (msg_len) {
981 if (msg_len > trash.size) {
982 /* Status code is not success, abort */
983 appctx->st0 = PEER_SESS_ST_ERRSIZE;
984 goto switchstate;
985 }
Willy Tarreau72d6c162013-04-11 16:14:13 +0200986
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200987 reql = co_getblk(si_oc(si),
988 trash.area,
989 msg_len,
990 totl);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200991 if (reql <= 0) /* closed */
992 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +0200993 totl += reql;
Willy Tarreau86a446e2013-11-25 23:02:37 +0100994
Emeric Brunb3971ab2015-05-12 18:49:09 +0200995 msg_end += msg_len;
996 }
997 }
Willy Tarreau86a446e2013-11-25 23:02:37 +0100998
Emeric Brunb3971ab2015-05-12 18:49:09 +0200999 if (msg_head[0] == PEER_MSG_CLASS_CONTROL) {
1000 if (msg_head[1] == PEER_MSG_CTRL_RESYNCREQ) {
1001 struct shared_table *st;
1002 /* Reset message: remote need resync */
1003
1004 /* prepare tables fot a global push */
1005 for (st = curpeer->tables; st; st = st->next) {
1006 st->teaching_origin = st->last_pushed = st->table->update;
1007 st->flags = 0;
Willy Tarreau86a446e2013-11-25 23:02:37 +01001008 }
Willy Tarreau72d6c162013-04-11 16:14:13 +02001009
Emeric Brunb3971ab2015-05-12 18:49:09 +02001010 /* reset teaching flags to 0 */
1011 curpeer->flags &= PEER_TEACH_RESET;
Willy Tarreau9d9179b2013-04-11 16:56:44 +02001012
Emeric Brunb3971ab2015-05-12 18:49:09 +02001013 /* flag to start to teach lesson */
1014 curpeer->flags |= PEER_F_TEACH_PROCESS;
1015
1016
1017 }
1018 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCFINISHED) {
1019
1020 if (curpeer->flags & PEER_F_LEARN_ASSIGN) {
1021 curpeer->flags &= ~PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001022 curpeers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
1023 curpeers->flags |= (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE);
Willy Tarreau86a446e2013-11-25 23:02:37 +01001024 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001025 curpeer->confirm++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001026 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001027 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCPARTIAL) {
1028
1029 if (curpeer->flags & PEER_F_LEARN_ASSIGN) {
1030 curpeer->flags &= ~PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001031 curpeers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001032
1033 curpeer->flags |= PEER_F_LEARN_NOTUP2DATE;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001034 curpeers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
1035 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Cyril Bonté9a60ff92014-02-16 01:07:07 +01001036 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001037 curpeer->confirm++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001038 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001039 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCCONFIRM) {
Emeric Brun597b26e2016-08-12 11:23:31 +02001040 struct shared_table *st;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001041
1042 /* If stopping state */
1043 if (stopping) {
1044 /* Close session, push resync no more needed */
1045 curpeer->flags |= PEER_F_TEACH_COMPLETE;
1046 appctx->st0 = PEER_SESS_ST_END;
1047 goto switchstate;
1048 }
Emeric Brun597b26e2016-08-12 11:23:31 +02001049 for (st = curpeer->tables; st; st = st->next) {
1050 st->update = st->last_pushed = st->teaching_origin;
1051 st->flags = 0;
1052 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001053
1054 /* reset teaching flags to 0 */
1055 curpeer->flags &= PEER_TEACH_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +02001056 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001057 }
1058 else if (msg_head[0] == PEER_MSG_CLASS_STICKTABLE) {
1059 if (msg_head[1] == PEER_MSG_STKT_DEFINE) {
1060 int table_id_len;
1061 struct shared_table *st;
1062 int table_type;
1063 int table_keylen;
1064 int table_id;
1065 uint64_t table_data;
Emeric Brun2b920a12010-09-23 18:30:22 +02001066
Emeric Brunb3971ab2015-05-12 18:49:09 +02001067 table_id = intdecode(&msg_cur, msg_end);
1068 if (!msg_cur) {
1069 /* malformed message */
1070 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1071 goto switchstate;
1072 }
Willy Tarreau72d6c162013-04-11 16:14:13 +02001073
Emeric Brunb3971ab2015-05-12 18:49:09 +02001074 table_id_len = intdecode(&msg_cur, msg_end);
1075 if (!msg_cur) {
1076 /* malformed message */
1077 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1078 goto switchstate;
1079 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001080
Emeric Brunb3971ab2015-05-12 18:49:09 +02001081 curpeer->remote_table = NULL;
1082 if (!table_id_len || (msg_cur + table_id_len) >= msg_end) {
1083 /* malformed message */
1084 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1085 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001086 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001087
Emeric Brunb3971ab2015-05-12 18:49:09 +02001088 for (st = curpeer->tables; st; st = st->next) {
1089 /* Reset IDs */
1090 if (st->remote_id == table_id)
1091 st->remote_id = 0;
Willy Tarreau9d9179b2013-04-11 16:56:44 +02001092
Emeric Brunb3971ab2015-05-12 18:49:09 +02001093 if (!curpeer->remote_table
1094 && (table_id_len == strlen(st->table->id))
1095 && (memcmp(st->table->id, msg_cur, table_id_len) == 0)) {
1096 curpeer->remote_table = st;
Emeric Brun2b920a12010-09-23 18:30:22 +02001097 }
1098 }
1099
Emeric Brunb3971ab2015-05-12 18:49:09 +02001100 if (!curpeer->remote_table) {
1101 goto ignore_msg;
1102 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001103
Emeric Brunb3971ab2015-05-12 18:49:09 +02001104 msg_cur += table_id_len;
1105 if (msg_cur >= msg_end) {
1106 /* malformed message */
1107 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1108 goto switchstate;
1109 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001110
Emeric Brunb3971ab2015-05-12 18:49:09 +02001111 table_type = intdecode(&msg_cur, msg_end);
1112 if (!msg_cur) {
1113 /* malformed message */
1114 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1115 goto switchstate;
1116 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001117
Emeric Brunb3971ab2015-05-12 18:49:09 +02001118 table_keylen = intdecode(&msg_cur, msg_end);
1119 if (!msg_cur) {
1120 /* malformed message */
1121 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1122 goto switchstate;
1123 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001124
Emeric Brunb3971ab2015-05-12 18:49:09 +02001125 table_data = intdecode(&msg_cur, msg_end);
1126 if (!msg_cur) {
1127 /* malformed message */
1128 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1129 goto switchstate;
1130 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001131
Emeric Brunb3971ab2015-05-12 18:49:09 +02001132 if (curpeer->remote_table->table->type != table_type
1133 || curpeer->remote_table->table->key_size != table_keylen) {
1134 curpeer->remote_table = NULL;
1135 goto ignore_msg;
1136 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001137
Emeric Brunb3971ab2015-05-12 18:49:09 +02001138 curpeer->remote_table->remote_data = table_data;
1139 curpeer->remote_table->remote_id = table_id;
Emeric Brun2b920a12010-09-23 18:30:22 +02001140 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001141 else if (msg_head[1] == PEER_MSG_STKT_SWITCH) {
1142 struct shared_table *st;
1143 int table_id;
Emeric Brun2b920a12010-09-23 18:30:22 +02001144
Emeric Brunb3971ab2015-05-12 18:49:09 +02001145 table_id = intdecode(&msg_cur, msg_end);
1146 if (!msg_cur) {
1147 /* malformed message */
1148 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1149 goto switchstate;
1150 }
1151 curpeer->remote_table = NULL;
1152 for (st = curpeer->tables; st; st = st->next) {
1153 if (st->remote_id == table_id) {
1154 curpeer->remote_table = st;
1155 break;
1156 }
1157 }
1158
Emeric Brun2b920a12010-09-23 18:30:22 +02001159 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001160 else if (msg_head[1] == PEER_MSG_STKT_UPDATE
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001161 || msg_head[1] == PEER_MSG_STKT_INCUPDATE
1162 || msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED
1163 || msg_head[1] == PEER_MSG_STKT_INCUPDATE_TIMED) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001164 struct shared_table *st = curpeer->remote_table;
1165 uint32_t update;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001166 int expire;
Emeric Brun94900952015-06-11 18:25:54 +02001167 unsigned int data_type;
1168 void *data_ptr;
Emeric Brun2b920a12010-09-23 18:30:22 +02001169
Emeric Brunb3971ab2015-05-12 18:49:09 +02001170 /* Here we have data message */
1171 if (!st)
1172 goto ignore_msg;
Emeric Brun2b920a12010-09-23 18:30:22 +02001173
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001174 expire = MS_TO_TICKS(st->table->expire);
1175
1176 if (msg_head[1] == PEER_MSG_STKT_UPDATE ||
1177 msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001178 if (msg_len < sizeof(update)) {
1179 /* malformed message */
1180 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1181 goto switchstate;
1182 }
1183 memcpy(&update, msg_cur, sizeof(update));
1184 msg_cur += sizeof(update);
1185 st->last_get = htonl(update);
1186 }
1187 else {
1188 st->last_get++;
1189 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001190
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001191 if (msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED ||
1192 msg_head[1] == PEER_MSG_STKT_INCUPDATE_TIMED) {
1193 size_t expire_sz = sizeof expire;
1194
1195 if (msg_cur + expire_sz > msg_end) {
1196 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1197 goto switchstate;
1198 }
1199 memcpy(&expire, msg_cur, expire_sz);
1200 msg_cur += expire_sz;
1201 expire = ntohl(expire);
1202 }
1203
Emeric Brunb3971ab2015-05-12 18:49:09 +02001204 newts = stksess_new(st->table, NULL);
1205 if (!newts)
1206 goto ignore_msg;
Emeric Brun55482912018-01-22 15:10:08 +01001207
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +02001208 if (st->table->type == SMP_T_STR) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001209 unsigned int to_read, to_store;
Emeric Brun2b920a12010-09-23 18:30:22 +02001210
Emeric Brunb3971ab2015-05-12 18:49:09 +02001211 to_read = intdecode(&msg_cur, msg_end);
1212 if (!msg_cur) {
1213 /* malformed message */
1214 stksess_free(st->table, newts);
1215 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1216 goto switchstate;
1217 }
1218 to_store = MIN(to_read, st->table->key_size - 1);
1219 if (msg_cur + to_store > msg_end) {
1220 /* malformed message */
1221 stksess_free(st->table, newts);
1222 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1223 goto switchstate;
1224 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001225
Emeric Brunb3971ab2015-05-12 18:49:09 +02001226 memcpy(newts->key.key, msg_cur, to_store);
1227 newts->key.key[to_store] = 0;
1228 msg_cur += to_read;
1229 }
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +02001230 else if (st->table->type == SMP_T_SINT) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001231 unsigned int netinteger;
Willy Tarreau72d6c162013-04-11 16:14:13 +02001232
Emeric Brunb3971ab2015-05-12 18:49:09 +02001233 if (msg_cur + sizeof(netinteger) > msg_end) {
1234 /* malformed message */
1235 stksess_free(st->table, newts);
1236 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1237 goto switchstate;
1238 }
1239 memcpy(&netinteger, msg_cur, sizeof(netinteger));
1240 netinteger = ntohl(netinteger);
1241 memcpy(newts->key.key, &netinteger, sizeof(netinteger));
1242 msg_cur += sizeof(netinteger);
1243 }
1244 else {
1245 if (msg_cur + st->table->key_size > msg_end) {
1246 /* malformed message */
1247 stksess_free(st->table, newts);
1248 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1249 goto switchstate;
1250 }
1251 memcpy(newts->key.key, msg_cur, st->table->key_size);
1252 msg_cur += st->table->key_size;
1253 }
1254
1255 /* lookup for existing entry */
Emeric Brun819fc6f2017-06-13 19:37:32 +02001256 ts = stktable_set_entry(st->table, newts);
1257 if (ts != newts) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001258 stksess_free(st->table, newts);
1259 newts = NULL;
1260 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001261
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001262 HA_RWLOCK_WRLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001263
Emeric Brun94900952015-06-11 18:25:54 +02001264 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001265
Emeric Brun94900952015-06-11 18:25:54 +02001266 if ((1 << data_type) & st->remote_data) {
1267 switch (stktable_data_types[data_type].std_type) {
1268 case STD_T_SINT: {
1269 int data;
1270
1271 data = intdecode(&msg_cur, msg_end);
1272 if (!msg_cur) {
1273 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001274 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001275 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001276 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1277 goto switchstate;
1278 }
1279
1280 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1281 if (data_ptr)
1282 stktable_data_cast(data_ptr, std_t_sint) = data;
1283 break;
1284 }
1285 case STD_T_UINT: {
1286 unsigned int data;
1287
1288 data = intdecode(&msg_cur, msg_end);
1289 if (!msg_cur) {
1290 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001291 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001292 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001293 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1294 goto switchstate;
1295 }
1296
1297 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1298 if (data_ptr)
1299 stktable_data_cast(data_ptr, std_t_uint) = data;
1300 break;
1301 }
1302 case STD_T_ULL: {
1303 unsigned long long data;
1304
1305 data = intdecode(&msg_cur, msg_end);
1306 if (!msg_cur) {
1307 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001308 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001309 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001310 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1311 goto switchstate;
1312 }
1313
1314 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1315 if (data_ptr)
1316 stktable_data_cast(data_ptr, std_t_ull) = data;
1317 break;
1318 }
1319 case STD_T_FRQP: {
1320 struct freq_ctr_period data;
1321
Emeric Brunf2fc1fd2017-11-02 17:32:43 +01001322 /* First bit is reserved for the freq_ctr_period lock
1323 Note: here we're still protected by the stksess lock
1324 so we don't need to update the update the freq_ctr_period
1325 using its internal lock */
1326
1327 data.curr_tick = tick_add(now_ms, -intdecode(&msg_cur, msg_end)) & ~0x1;
Emeric Brun94900952015-06-11 18:25:54 +02001328 if (!msg_cur) {
1329 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001330 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001331 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001332 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1333 goto switchstate;
1334 }
1335 data.curr_ctr = intdecode(&msg_cur, msg_end);
1336 if (!msg_cur) {
1337 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001338 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001339 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001340 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1341 goto switchstate;
1342 }
1343 data.prev_ctr = intdecode(&msg_cur, msg_end);
1344 if (!msg_cur) {
1345 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001346 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001347 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001348 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1349 goto switchstate;
1350 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001351
Emeric Brun94900952015-06-11 18:25:54 +02001352 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1353 if (data_ptr)
1354 stktable_data_cast(data_ptr, std_t_frqp) = data;
1355 break;
1356 }
1357 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001358 }
1359 }
Emeric Brun819fc6f2017-06-13 19:37:32 +02001360
Emeric Brun55482912018-01-22 15:10:08 +01001361 /* Force new expiration */
1362 ts->expire = tick_add(now_ms, expire);
1363
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001364 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001365 stktable_touch_remote(st->table, ts, 1);
1366
Emeric Brunb3971ab2015-05-12 18:49:09 +02001367 }
1368 else if (msg_head[1] == PEER_MSG_STKT_ACK) {
1369 /* ack message */
1370 uint32_t table_id ;
1371 uint32_t update;
1372 struct shared_table *st;
1373
1374 table_id = intdecode(&msg_cur, msg_end);
1375 if (!msg_cur || (msg_cur + sizeof(update) > msg_end)) {
1376 /* malformed message */
1377 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1378 goto switchstate;
1379 }
1380 memcpy(&update, msg_cur, sizeof(update));
1381 update = ntohl(update);
Emeric Brun2b920a12010-09-23 18:30:22 +02001382
Emeric Brunb3971ab2015-05-12 18:49:09 +02001383 for (st = curpeer->tables; st; st = st->next) {
1384 if (st->local_id == table_id) {
1385 st->update = update;
1386 break;
1387 }
1388 }
1389 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001390 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001391 else if (msg_head[0] == PEER_MSG_CLASS_RESERVED) {
1392 appctx->st0 = PEER_SESS_ST_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +02001393 goto switchstate;
1394 }
1395
Emeric Brunb3971ab2015-05-12 18:49:09 +02001396ignore_msg:
Emeric Brun2b920a12010-09-23 18:30:22 +02001397 /* skip consumed message */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001398 co_skip(si_oc(si), totl);
Emeric Brun2b920a12010-09-23 18:30:22 +02001399 /* loop on that state to peek next message */
Willy Tarreau72d6c162013-04-11 16:14:13 +02001400 goto switchstate;
1401
Emeric Brun2b920a12010-09-23 18:30:22 +02001402incomplete:
Willy Tarreau06d80a92017-10-19 14:32:15 +02001403 /* we get here when a co_getblk() returns <= 0 in reql */
Willy Tarreau9d9179b2013-04-11 16:56:44 +02001404
Willy Tarreau72d6c162013-04-11 16:14:13 +02001405 if (reql < 0) {
1406 /* there was an error */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001407 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau72d6c162013-04-11 16:14:13 +02001408 goto switchstate;
1409 }
1410
Emeric Brun2b920a12010-09-23 18:30:22 +02001411
Emeric Brun2b920a12010-09-23 18:30:22 +02001412
Emeric Brunb3971ab2015-05-12 18:49:09 +02001413
Emeric Brun2b920a12010-09-23 18:30:22 +02001414 /* Need to request a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001415 if ((curpeer->flags & PEER_F_LEARN_ASSIGN) &&
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001416 (curpeers->flags & PEERS_F_RESYNC_ASSIGN) &&
1417 !(curpeers->flags & PEERS_F_RESYNC_PROCESS)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001418 unsigned char msg[2];
Emeric Brun2b920a12010-09-23 18:30:22 +02001419
Emeric Brunb3971ab2015-05-12 18:49:09 +02001420 /* Current peer was elected to request a resync */
1421 msg[0] = PEER_MSG_CLASS_CONTROL;
1422 msg[1] = PEER_MSG_CTRL_RESYNCREQ;
Emeric Brun2b920a12010-09-23 18:30:22 +02001423
Emeric Brunb3971ab2015-05-12 18:49:09 +02001424 /* message to buffer */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001425 repl = ci_putblk(si_ic(si), (char *)msg, sizeof(msg));
Emeric Brunb3971ab2015-05-12 18:49:09 +02001426 if (repl <= 0) {
1427 /* no more write possible */
1428 if (repl == -1)
1429 goto full;
1430 appctx->st0 = PEER_SESS_ST_END;
1431 goto switchstate;
1432 }
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001433 curpeers->flags |= PEERS_F_RESYNC_PROCESS;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001434 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001435
Emeric Brunb3971ab2015-05-12 18:49:09 +02001436 /* Nothing to read, now we start to write */
Emeric Brun2b920a12010-09-23 18:30:22 +02001437
Emeric Brunb3971ab2015-05-12 18:49:09 +02001438 if (curpeer->tables) {
1439 struct shared_table *st;
1440 struct shared_table *last_local_table;
Emeric Brun2b920a12010-09-23 18:30:22 +02001441
Emeric Brunb3971ab2015-05-12 18:49:09 +02001442 last_local_table = curpeer->last_local_table;
1443 if (!last_local_table)
1444 last_local_table = curpeer->tables;
1445 st = last_local_table->next;
Emeric Brun2b920a12010-09-23 18:30:22 +02001446
Emeric Brunb3971ab2015-05-12 18:49:09 +02001447 while (1) {
1448 if (!st)
1449 st = curpeer->tables;
Emeric Brun2b920a12010-09-23 18:30:22 +02001450
Emeric Brunb3971ab2015-05-12 18:49:09 +02001451 /* It remains some updates to ack */
1452 if (st->last_get != st->last_acked) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001453 int msglen;
Emeric Brun2b920a12010-09-23 18:30:22 +02001454
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001455 msglen = peer_prepare_ackmsg(st,
1456 trash.area,
1457 trash.size);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001458 if (!msglen) {
1459 /* internal error: message does not fit in trash */
1460 appctx->st0 = PEER_SESS_ST_END;
1461 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001462 }
1463
Emeric Brunb3971ab2015-05-12 18:49:09 +02001464 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001465 repl = ci_putblk(si_ic(si),
1466 trash.area,
1467 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001468 if (repl <= 0) {
1469 /* no more write possible */
1470 if (repl == -1) {
1471 goto full;
Emeric Brun2b920a12010-09-23 18:30:22 +02001472 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001473 appctx->st0 = PEER_SESS_ST_END;
1474 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001475 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001476 st->last_acked = st->last_get;
Emeric Brun2b920a12010-09-23 18:30:22 +02001477 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001478
Emeric Brunb3971ab2015-05-12 18:49:09 +02001479 if (!(curpeer->flags & PEER_F_TEACH_PROCESS)) {
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001480 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001481 if (!(curpeer->flags & PEER_F_LEARN_ASSIGN) &&
1482 ((int)(st->last_pushed - st->table->localupdate) < 0)) {
1483 struct eb32_node *eb;
1484 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001485
Emeric Brunb3971ab2015-05-12 18:49:09 +02001486 if (st != curpeer->last_local_table) {
1487 int msglen;
Emeric Brun2b920a12010-09-23 18:30:22 +02001488
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001489 msglen = peer_prepare_switchmsg(st,
1490 trash.area,
1491 trash.size);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001492 if (!msglen) {
Emeric Brun088c9b72017-12-01 11:37:36 +01001493 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001494 /* internal error: message does not fit in trash */
1495 appctx->st0 = PEER_SESS_ST_END;
1496 goto switchstate;
1497 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001498
Emeric Brunb3971ab2015-05-12 18:49:09 +02001499 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001500 repl = ci_putblk(si_ic(si),
1501 trash.area,
1502 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001503 if (repl <= 0) {
Emeric Brun088c9b72017-12-01 11:37:36 +01001504 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001505 /* no more write possible */
1506 if (repl == -1) {
1507 goto full;
1508 }
1509 appctx->st0 = PEER_SESS_ST_END;
1510 goto switchstate;
1511 }
1512 curpeer->last_local_table = st;
Emeric Brun2b920a12010-09-23 18:30:22 +02001513 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001514
1515 /* We force new pushed to 1 to force identifier in update message */
1516 new_pushed = 1;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001517 while (1) {
1518 uint32_t msglen;
1519 struct stksess *ts;
Emeric Brun819fc6f2017-06-13 19:37:32 +02001520 unsigned updateid;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001521
1522 /* push local updates */
Emeric Brun819fc6f2017-06-13 19:37:32 +02001523 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001524 if (!eb) {
1525 eb = eb32_first(&st->table->updates);
1526 if (!eb || ((int)(eb->key - st->last_pushed) <= 0)) {
Emeric Brunaaf58602015-06-15 17:23:30 +02001527 st->table->commitupdate = st->last_pushed = st->table->localupdate;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001528 break;
1529 }
1530 }
1531
1532 if ((int)(eb->key - st->table->localupdate) > 0) {
Emeric Brunaaf58602015-06-15 17:23:30 +02001533 st->table->commitupdate = st->last_pushed = st->table->localupdate;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001534 break;
1535 }
1536
1537 ts = eb32_entry(eb, struct stksess, upd);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001538 updateid = ts->upd.key;
1539 ts->ref_cnt++;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001540 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001541
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001542 msglen = peer_prepare_updatemsg(ts, st, updateid,
1543 trash.area,
1544 trash.size,
1545 new_pushed,
1546 0);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001547 if (!msglen) {
1548 /* internal error: message does not fit in trash */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001549 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001550 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001551 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001552 appctx->st0 = PEER_SESS_ST_END;
1553 goto switchstate;
1554 }
1555
1556 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001557 repl = ci_putblk(si_ic(si),
1558 trash.area,
1559 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001560 if (repl <= 0) {
1561 /* no more write possible */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001562 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001563 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001564 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001565 if (repl == -1) {
1566 goto full;
1567 }
1568 appctx->st0 = PEER_SESS_ST_END;
1569 goto switchstate;
1570 }
Emeric Brun819fc6f2017-06-13 19:37:32 +02001571
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001572 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001573 ts->ref_cnt--;
1574 st->last_pushed = updateid;
Emeric Brunaaf58602015-06-15 17:23:30 +02001575 if ((int)(st->last_pushed - st->table->commitupdate) > 0)
1576 st->table->commitupdate = st->last_pushed;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001577 /* identifier may not needed in next update message */
1578 new_pushed = 0;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001579 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001580 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001581 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun2b920a12010-09-23 18:30:22 +02001582 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001583 else {
1584 if (!(st->flags & SHTABLE_F_TEACH_STAGE1)) {
1585 struct eb32_node *eb;
1586 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001587
Emeric Brunb3971ab2015-05-12 18:49:09 +02001588 if (st != curpeer->last_local_table) {
1589 int msglen;
1590
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001591 msglen = peer_prepare_switchmsg(st,
1592 trash.area,
1593 trash.size);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001594 if (!msglen) {
1595 /* internal error: message does not fit in trash */
1596 appctx->st0 = PEER_SESS_ST_END;
1597 goto switchstate;
1598 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001599
Emeric Brunb3971ab2015-05-12 18:49:09 +02001600 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001601 repl = ci_putblk(si_ic(si),
1602 trash.area,
1603 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001604 if (repl <= 0) {
1605 /* no more write possible */
1606 if (repl == -1) {
1607 goto full;
1608 }
1609 appctx->st0 = PEER_SESS_ST_END;
1610 goto switchstate;
1611 }
1612 curpeer->last_local_table = st;
1613 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001614
Emeric Brunb3971ab2015-05-12 18:49:09 +02001615 /* We force new pushed to 1 to force identifier in update message */
1616 new_pushed = 1;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001617 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001618 while (1) {
1619 uint32_t msglen;
1620 struct stksess *ts;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001621 int use_timed;
Emeric Brun819fc6f2017-06-13 19:37:32 +02001622 unsigned updateid;
Emeric Brun2b920a12010-09-23 18:30:22 +02001623
Emeric Brunb3971ab2015-05-12 18:49:09 +02001624 /* push local updates */
Emeric Brun819fc6f2017-06-13 19:37:32 +02001625 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001626 if (!eb) {
1627 st->flags |= SHTABLE_F_TEACH_STAGE1;
1628 eb = eb32_first(&st->table->updates);
1629 if (eb)
1630 st->last_pushed = eb->key - 1;
1631 break;
1632 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001633
Emeric Brunb3971ab2015-05-12 18:49:09 +02001634 ts = eb32_entry(eb, struct stksess, upd);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001635 updateid = ts->upd.key;
1636 ts->ref_cnt++;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001637 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001638
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001639 use_timed = !(curpeer->flags & PEER_F_DWNGRD);
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001640 msglen = peer_prepare_updatemsg(ts, st, updateid,
1641 trash.area,
1642 trash.size,
1643 new_pushed,
1644 use_timed);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001645 if (!msglen) {
1646 /* internal error: message does not fit in trash */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001647 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001648 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001649 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001650 appctx->st0 = PEER_SESS_ST_END;
1651 goto switchstate;
1652 }
1653
1654 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001655 repl = ci_putblk(si_ic(si),
1656 trash.area,
1657 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001658 if (repl <= 0) {
1659 /* no more write possible */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001660 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001661 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001662 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001663 if (repl == -1) {
1664 goto full;
1665 }
1666 appctx->st0 = PEER_SESS_ST_END;
1667 goto switchstate;
1668 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001669 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001670 ts->ref_cnt--;
1671 st->last_pushed = updateid;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001672 /* identifier may not needed in next update message */
1673 new_pushed = 0;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001674 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001675 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun2b920a12010-09-23 18:30:22 +02001676 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001677
Emeric Brunb3971ab2015-05-12 18:49:09 +02001678 if (!(st->flags & SHTABLE_F_TEACH_STAGE2)) {
1679 struct eb32_node *eb;
1680 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001681
Emeric Brunb3971ab2015-05-12 18:49:09 +02001682 if (st != curpeer->last_local_table) {
1683 int msglen;
1684
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001685 msglen = peer_prepare_switchmsg(st,
1686 trash.area,
1687 trash.size);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001688 if (!msglen) {
1689 /* internal error: message does not fit in trash */
1690 appctx->st0 = PEER_SESS_ST_END;
1691 goto switchstate;
1692 }
1693
1694 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001695 repl = ci_putblk(si_ic(si),
1696 trash.area,
1697 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001698 if (repl <= 0) {
1699 /* no more write possible */
1700 if (repl == -1) {
1701 goto full;
1702 }
1703 appctx->st0 = PEER_SESS_ST_END;
1704 goto switchstate;
1705 }
1706 curpeer->last_local_table = st;
1707 }
1708
1709 /* We force new pushed to 1 to force identifier in update message */
1710 new_pushed = 1;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001711 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001712 while (1) {
1713 uint32_t msglen;
1714 struct stksess *ts;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001715 int use_timed;
Emeric Brun819fc6f2017-06-13 19:37:32 +02001716 unsigned updateid;
1717
1718 /* push local updates */
1719 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001720
1721 /* push local updates */
1722 if (!eb || eb->key > st->teaching_origin) {
1723 st->flags |= SHTABLE_F_TEACH_STAGE2;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001724 break;
1725 }
1726
1727 ts = eb32_entry(eb, struct stksess, upd);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001728 updateid = ts->upd.key;
1729 ts->ref_cnt++;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001730 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001731
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001732 use_timed = !(curpeer->flags & PEER_F_DWNGRD);
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001733 msglen = peer_prepare_updatemsg(ts, st, updateid,
1734 trash.area,
1735 trash.size,
1736 new_pushed,
1737 use_timed);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001738 if (!msglen) {
1739 /* internal error: message does not fit in trash */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001740 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001741 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001742 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001743 appctx->st0 = PEER_SESS_ST_END;
1744 goto switchstate;
1745 }
1746
1747 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001748 repl = ci_putblk(si_ic(si),
1749 trash.area,
1750 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001751 if (repl <= 0) {
1752 /* no more write possible */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001753 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001754 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001755 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001756 if (repl == -1) {
1757 goto full;
1758 }
1759 appctx->st0 = PEER_SESS_ST_END;
1760 goto switchstate;
1761 }
Emeric Brun819fc6f2017-06-13 19:37:32 +02001762
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001763 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001764 ts->ref_cnt--;
1765 st->last_pushed = updateid;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001766 /* identifier may not needed in next update message */
1767 new_pushed = 0;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001768 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001769 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun2b920a12010-09-23 18:30:22 +02001770 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001771 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001772
1773 if (st == last_local_table)
1774 break;
1775 st = st->next;
Emeric Brun2b920a12010-09-23 18:30:22 +02001776 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001777 }
1778
1779
1780 if ((curpeer->flags & PEER_F_TEACH_PROCESS) && !(curpeer->flags & PEER_F_TEACH_FINISHED)) {
1781 unsigned char msg[2];
1782
1783 /* Current peer was elected to request a resync */
1784 msg[0] = PEER_MSG_CLASS_CONTROL;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001785 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 +02001786 /* process final lesson message */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001787 repl = ci_putblk(si_ic(si), (char *)msg, sizeof(msg));
Emeric Brunb3971ab2015-05-12 18:49:09 +02001788 if (repl <= 0) {
1789 /* no more write possible */
1790 if (repl == -1)
1791 goto full;
1792 appctx->st0 = PEER_SESS_ST_END;
1793 goto switchstate;
1794 }
1795 /* flag finished message sent */
1796 curpeer->flags |= PEER_F_TEACH_FINISHED;
1797 }
1798
Emeric Brun597b26e2016-08-12 11:23:31 +02001799 /* Confirm finished or partial messages */
1800 while (curpeer->confirm) {
1801 unsigned char msg[2];
1802
1803 /* There is a confirm messages to send */
1804 msg[0] = PEER_MSG_CLASS_CONTROL;
1805 msg[1] = PEER_MSG_CTRL_RESYNCCONFIRM;
1806
1807 /* message to buffer */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001808 repl = ci_putblk(si_ic(si), (char *)msg, sizeof(msg));
Emeric Brun597b26e2016-08-12 11:23:31 +02001809 if (repl <= 0) {
1810 /* no more write possible */
1811 if (repl == -1)
1812 goto full;
1813 appctx->st0 = PEER_SESS_ST_END;
1814 goto switchstate;
1815 }
1816 curpeer->confirm--;
1817 }
1818
Emeric Brun2b920a12010-09-23 18:30:22 +02001819 /* noting more to do */
1820 goto out;
1821 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001822 case PEER_SESS_ST_EXIT:
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001823 repl = snprintf(trash.area, trash.size,
1824 "%d\n", appctx->st1);
1825 if (ci_putblk(si_ic(si), trash.area, repl) == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +01001826 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001827 appctx->st0 = PEER_SESS_ST_END;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001828 goto switchstate;
1829 case PEER_SESS_ST_ERRSIZE: {
1830 unsigned char msg[2];
1831
1832 msg[0] = PEER_MSG_CLASS_ERROR;
1833 msg[1] = PEER_MSG_ERR_SIZELIMIT;
1834
Willy Tarreau06d80a92017-10-19 14:32:15 +02001835 if (ci_putblk(si_ic(si), (char *)msg, sizeof(msg)) == -1)
Emeric Brunb3971ab2015-05-12 18:49:09 +02001836 goto full;
1837 appctx->st0 = PEER_SESS_ST_END;
1838 goto switchstate;
1839 }
1840 case PEER_SESS_ST_ERRPROTO: {
1841 unsigned char msg[2];
1842
1843 msg[0] = PEER_MSG_CLASS_ERROR;
1844 msg[1] = PEER_MSG_ERR_PROTOCOL;
1845
Willy Tarreau06d80a92017-10-19 14:32:15 +02001846 if (ci_putblk(si_ic(si), (char *)msg, sizeof(msg)) == -1)
Emeric Brunb3971ab2015-05-12 18:49:09 +02001847 goto full;
1848 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02001849 /* fall through */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001850 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001851 case PEER_SESS_ST_END: {
Emeric Brun80527f52017-06-19 17:46:37 +02001852 if (curpeer) {
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001853 HA_SPIN_UNLOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02001854 curpeer = NULL;
1855 }
Willy Tarreau73b013b2012-05-21 16:31:45 +02001856 si_shutw(si);
1857 si_shutr(si);
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001858 si_ic(si)->flags |= CF_READ_NULL;
Willy Tarreau828824a2015-04-19 17:20:03 +02001859 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001860 }
1861 }
1862 }
1863out:
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001864 si_oc(si)->flags |= CF_READ_DONTWAIT;
Emeric Brun80527f52017-06-19 17:46:37 +02001865
1866 if (curpeer)
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001867 HA_SPIN_UNLOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun2b920a12010-09-23 18:30:22 +02001868 return;
Willy Tarreaubc18da12015-03-13 14:00:47 +01001869full:
Willy Tarreaufe127932015-04-21 19:23:39 +02001870 si_applet_cant_put(si);
Willy Tarreaubc18da12015-03-13 14:00:47 +01001871 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001872}
1873
Willy Tarreau30576452015-04-13 13:50:30 +02001874static struct applet peer_applet = {
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001875 .obj_type = OBJ_TYPE_APPLET,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001876 .name = "<PEER>", /* used for logging */
1877 .fct = peer_io_handler,
Aman Gupta9a13e842012-04-02 18:57:53 -07001878 .release = peer_session_release,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001879};
Emeric Brun2b920a12010-09-23 18:30:22 +02001880
1881/*
1882 * Use this function to force a close of a peer session
1883 */
Willy Tarreau81bc3b02016-10-31 17:37:39 +01001884static void peer_session_forceshutdown(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +02001885{
Frédéric Lécaille5df11902017-06-13 16:39:57 +02001886 /* Note that the peer sessions which have just been created
1887 * (->st0 == PEER_SESS_ST_CONNECT) must not
1888 * be shutdown, if not, the TCP session will never be closed
1889 * and stay in CLOSE_WAIT state after having been closed by
1890 * the remote side.
1891 */
1892 if (!appctx || appctx->st0 == PEER_SESS_ST_CONNECT)
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001893 return;
1894
Willy Tarreau81bc3b02016-10-31 17:37:39 +01001895 if (appctx->applet != &peer_applet)
1896 return;
1897
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001898 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau78c0c502016-10-31 17:32:20 +01001899 appctx_wakeup(appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02001900}
1901
Willy Tarreau91d96282015-03-13 15:47:26 +01001902/* Pre-configures a peers frontend to accept incoming connections */
1903void peers_setup_frontend(struct proxy *fe)
1904{
1905 fe->last_change = now.tv_sec;
1906 fe->cap = PR_CAP_FE;
1907 fe->maxconn = 0;
1908 fe->conn_retries = CONN_RETRIES;
1909 fe->timeout.client = MS_TO_TICKS(5000);
Willy Tarreaud1d48d42015-03-13 16:15:46 +01001910 fe->accept = frontend_accept;
Willy Tarreauf87ab942015-03-13 15:55:16 +01001911 fe->default_target = &peer_applet.obj_type;
Willy Tarreau91d96282015-03-13 15:47:26 +01001912 fe->options2 |= PR_O2_INDEPSTR | PR_O2_SMARTCON | PR_O2_SMARTACC;
Willy Tarreau0fca4832015-05-01 19:12:05 +02001913 fe->bind_proc = 0; /* will be filled by users */
Willy Tarreau91d96282015-03-13 15:47:26 +01001914}
1915
Emeric Brun2b920a12010-09-23 18:30:22 +02001916/*
Willy Tarreaubd55e312010-11-11 10:55:09 +01001917 * Create a new peer session in assigned state (connect will start automatically)
Emeric Brun2b920a12010-09-23 18:30:22 +02001918 */
Willy Tarreau9df94c22016-10-31 18:42:52 +01001919static struct appctx *peer_session_create(struct peers *peers, struct peer *peer)
Emeric Brun2b920a12010-09-23 18:30:22 +02001920{
Willy Tarreau04b92862017-09-15 11:01:04 +02001921 struct proxy *p = peers->peers_fe; /* attached frontend */
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001922 struct appctx *appctx;
Willy Tarreau15b5e142015-04-04 14:38:25 +02001923 struct session *sess;
Willy Tarreau87b09662015-04-03 00:22:06 +02001924 struct stream *s;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001925 struct connection *conn;
Olivier Houchard9aaf7782017-09-13 18:30:23 +02001926 struct conn_stream *cs;
Emeric Brun2b920a12010-09-23 18:30:22 +02001927
Emeric Brunb3971ab2015-05-12 18:49:09 +02001928 peer->reconnect = tick_add(now_ms, MS_TO_TICKS(5000));
1929 peer->statuscode = PEER_SESS_SC_CONNECTCODE;
Willy Tarreaud990baf2015-04-05 00:32:03 +02001930 s = NULL;
1931
Emeric Brun1138fd02017-06-19 12:38:55 +02001932 appctx = appctx_new(&peer_applet, tid_bit);
Willy Tarreaud990baf2015-04-05 00:32:03 +02001933 if (!appctx)
1934 goto out_close;
1935
1936 appctx->st0 = PEER_SESS_ST_CONNECT;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001937 appctx->ctx.peers.ptr = (void *)peer;
Willy Tarreaud990baf2015-04-05 00:32:03 +02001938
Willy Tarreau04b92862017-09-15 11:01:04 +02001939 sess = session_new(p, NULL, &appctx->obj_type);
Willy Tarreau15b5e142015-04-04 14:38:25 +02001940 if (!sess) {
Christopher Faulet767a84b2017-11-24 16:50:31 +01001941 ha_alert("out of memory in peer_session_create().\n");
Willy Tarreaud990baf2015-04-05 00:32:03 +02001942 goto out_free_appctx;
Emeric Brun2b920a12010-09-23 18:30:22 +02001943 }
1944
Willy Tarreau87787ac2017-08-28 16:22:54 +02001945 if ((s = stream_new(sess, &appctx->obj_type)) == NULL) {
Christopher Faulet767a84b2017-11-24 16:50:31 +01001946 ha_alert("Failed to initialize stream in peer_session_create().\n");
Willy Tarreau87787ac2017-08-28 16:22:54 +02001947 goto out_free_sess;
Willy Tarreau8baf9062015-04-05 00:46:36 +02001948 }
1949
Willy Tarreau342bfb12015-04-05 01:35:34 +02001950 /* The tasks below are normally what is supposed to be done by
1951 * fe->accept().
1952 */
Willy Tarreaue7dff022015-04-03 01:14:29 +02001953 s->flags = SF_ASSIGNED|SF_ADDR_SET;
Emeric Brun2b920a12010-09-23 18:30:22 +02001954
Willy Tarreau6e2979c2015-04-27 13:21:15 +02001955 /* applet is waiting for data */
1956 si_applet_cant_get(&s->si[0]);
1957 appctx_wakeup(appctx);
1958
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001959 /* initiate an outgoing connection */
Willy Tarreaudbd02672017-12-06 17:39:53 +01001960 s->si[1].flags |= SI_FL_NOLINGER;
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001961 si_set_state(&s->si[1], SI_ST_ASS);
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001962
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001963 /* automatically prepare the stream interface to connect to the
Willy Tarreaub363a1f2013-10-01 10:45:07 +02001964 * pre-initialized connection in si->conn.
1965 */
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001966 if (unlikely((conn = conn_new()) == NULL))
Willy Tarreau8baf9062015-04-05 00:46:36 +02001967 goto out_free_strm;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001968
Olivier Houchard9aaf7782017-09-13 18:30:23 +02001969 if (unlikely((cs = cs_new(conn)) == NULL))
1970 goto out_free_conn;
1971
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001972 conn_prepare(conn, peer->proto, peer->xprt);
Olivier Houchard9aaf7782017-09-13 18:30:23 +02001973 conn_install_mux(conn, &mux_pt_ops, cs);
1974 si_attach_cs(&s->si[1], cs);
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001975
1976 conn->target = s->target = &s->be->obj_type;
1977 memcpy(&conn->addr.to, &peer->addr, sizeof(conn->addr.to));
Emeric Brun2b920a12010-09-23 18:30:22 +02001978 s->do_log = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001979 s->uniq_id = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +02001980
Willy Tarreau22ec1ea2014-11-27 20:45:39 +01001981 s->res.flags |= CF_READ_DONTWAIT;
Willy Tarreau696a2912014-11-24 11:36:57 +01001982
Emeric Brunb3971ab2015-05-12 18:49:09 +02001983 peer->appctx = appctx;
Willy Tarreau87787ac2017-08-28 16:22:54 +02001984 task_wakeup(s->task, TASK_WOKEN_INIT);
Willy Tarreau9df94c22016-10-31 18:42:52 +01001985 return appctx;
Emeric Brun2b920a12010-09-23 18:30:22 +02001986
1987 /* Error unrolling */
Olivier Houchard9aaf7782017-09-13 18:30:23 +02001988 out_free_conn:
1989 conn_free(conn);
Willy Tarreau15b5e142015-04-04 14:38:25 +02001990 out_free_strm:
Emeric Brun2b920a12010-09-23 18:30:22 +02001991 LIST_DEL(&s->list);
Willy Tarreaubafbe012017-11-24 17:34:44 +01001992 pool_free(pool_head_stream, s);
Willy Tarreau15b5e142015-04-04 14:38:25 +02001993 out_free_sess:
Willy Tarreau11c36242015-04-04 15:54:03 +02001994 session_free(sess);
Willy Tarreaud990baf2015-04-05 00:32:03 +02001995 out_free_appctx:
1996 appctx_free(appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02001997 out_close:
Willy Tarreaub21d08e2016-10-31 17:46:57 +01001998 return NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001999}
2000
2001/*
2002 * Task processing function to manage re-connect and peer session
2003 * tasks wakeup on local update.
2004 */
Olivier Houchard9f6af332018-05-25 14:04:04 +02002005static struct task *process_peer_sync(struct task * task, void *context, unsigned short state)
Emeric Brun2b920a12010-09-23 18:30:22 +02002006{
Olivier Houchard9f6af332018-05-25 14:04:04 +02002007 struct peers *peers = context;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002008 struct peer *ps;
2009 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +02002010
2011 task->expire = TICK_ETERNITY;
2012
Emeric Brunb3971ab2015-05-12 18:49:09 +02002013 if (!peers->peers_fe) {
Willy Tarreau46dc1ca2015-05-01 18:32:13 +02002014 /* this one was never started, kill it */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002015 signal_unregister_handler(peers->sighandler);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002016 task_delete(peers->sync_task);
2017 task_free(peers->sync_task);
Willy Tarreau37bb7be2015-09-21 15:24:58 +02002018 peers->sync_task = NULL;
Willy Tarreau46dc1ca2015-05-01 18:32:13 +02002019 return NULL;
2020 }
2021
Emeric Brun80527f52017-06-19 17:46:37 +02002022 /* Acquire lock for all peers of the section */
2023 for (ps = peers->remote; ps; ps = ps->next)
Christopher Faulet2a944ee2017-11-07 10:42:54 +01002024 HA_SPIN_LOCK(PEER_LOCK, &ps->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02002025
Emeric Brun2b920a12010-09-23 18:30:22 +02002026 if (!stopping) {
2027 /* Normal case (not soft stop)*/
Emeric Brunb3971ab2015-05-12 18:49:09 +02002028
2029 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL) &&
2030 (!nb_oldpids || tick_is_expired(peers->resync_timeout, now_ms)) &&
2031 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002032 /* Resync from local peer needed
2033 no peer was assigned for the lesson
2034 and no old local peer found
2035 or resync timeout expire */
2036
2037 /* flag no more resync from local, to try resync from remotes */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002038 peers->flags |= PEERS_F_RESYNC_LOCAL;
Emeric Brun2b920a12010-09-23 18:30:22 +02002039
2040 /* reschedule a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002041 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
Emeric Brun2b920a12010-09-23 18:30:22 +02002042 }
2043
2044 /* For each session */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002045 for (ps = peers->remote; ps; ps = ps->next) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002046 /* For each remote peers */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002047 if (!ps->local) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002048 if (!ps->appctx) {
2049 /* no active peer connection */
Emeric Brun2b920a12010-09-23 18:30:22 +02002050 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002051 ((ps->statuscode == PEER_SESS_SC_CONNECTCODE ||
Willy Tarreaub4e34da2015-05-20 10:39:04 +02002052 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002053 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02002054 tick_is_expired(ps->reconnect, now_ms))) {
2055 /* connection never tried
Willy Tarreau9df94c22016-10-31 18:42:52 +01002056 * or previous peer connection established with success
2057 * or previous peer connection failed while connecting
Emeric Brun2b920a12010-09-23 18:30:22 +02002058 * and reconnection timer is expired */
2059
2060 /* retry a connect */
Willy Tarreau9df94c22016-10-31 18:42:52 +01002061 ps->appctx = peer_session_create(peers, ps);
Emeric Brun2b920a12010-09-23 18:30:22 +02002062 }
Willy Tarreaub4e34da2015-05-20 10:39:04 +02002063 else if (!tick_is_expired(ps->reconnect, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002064 /* If previous session failed during connection
2065 * but reconnection timer is not expired */
2066
2067 /* reschedule task for reconnect */
2068 task->expire = tick_first(task->expire, ps->reconnect);
2069 }
2070 /* else do nothing */
Willy Tarreau9df94c22016-10-31 18:42:52 +01002071 } /* !ps->appctx */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002072 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002073 /* current peer connection is active and established */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002074 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
2075 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02002076 !(ps->flags & PEER_F_LEARN_NOTUP2DATE)) {
2077 /* Resync from a remote is needed
2078 * and no peer was assigned for lesson
2079 * and current peer may be up2date */
2080
2081 /* assign peer for the lesson */
2082 ps->flags |= PEER_F_LEARN_ASSIGN;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002083 peers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brun2b920a12010-09-23 18:30:22 +02002084
Willy Tarreau9df94c22016-10-31 18:42:52 +01002085 /* wake up peer handler to handle a request of resync */
Willy Tarreaue5843b32015-04-27 18:40:14 +02002086 appctx_wakeup(ps->appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02002087 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02002088 else {
2089 /* Awake session if there is data to push */
2090 for (st = ps->tables; st ; st = st->next) {
2091 if ((int)(st->last_pushed - st->table->localupdate) < 0) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002092 /* wake up the peer handler to push local updates */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002093 appctx_wakeup(ps->appctx);
2094 break;
2095 }
2096 }
Emeric Brun2b920a12010-09-23 18:30:22 +02002097 }
2098 /* else do nothing */
2099 } /* SUCCESSCODE */
2100 } /* !ps->peer->local */
2101 } /* for */
2102
2103 /* Resync from remotes expired: consider resync is finished */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002104 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
2105 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
2106 tick_is_expired(peers->resync_timeout, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002107 /* Resync from remote peer needed
2108 * no peer was assigned for the lesson
2109 * and resync timeout expire */
2110
2111 /* flag no more resync from remote, consider resync is finished */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002112 peers->flags |= PEERS_F_RESYNC_REMOTE;
Emeric Brun2b920a12010-09-23 18:30:22 +02002113 }
2114
Emeric Brunb3971ab2015-05-12 18:49:09 +02002115 if ((peers->flags & PEERS_RESYNC_STATEMASK) != PEERS_RESYNC_FINISHED) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002116 /* Resync not finished*/
Frédéric Lécaille5d6e5f82017-05-29 13:47:16 +02002117 /* reschedule task to resync timeout if not expired, to ended resync if needed */
2118 if (!tick_is_expired(peers->resync_timeout, now_ms))
2119 task->expire = tick_first(task->expire, peers->resync_timeout);
Emeric Brun2b920a12010-09-23 18:30:22 +02002120 }
2121 } /* !stopping */
2122 else {
2123 /* soft stop case */
2124 if (task->state & TASK_WOKEN_SIGNAL) {
2125 /* We've just recieved the signal */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002126 if (!(peers->flags & PEERS_F_DONOTSTOP)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002127 /* add DO NOT STOP flag if not present */
Christopher Faulet8d8aa0d2017-05-30 15:36:50 +02002128 HA_ATOMIC_ADD(&jobs, 1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002129 peers->flags |= PEERS_F_DONOTSTOP;
2130 ps = peers->local;
2131 for (st = ps->tables; st ; st = st->next)
2132 st->table->syncing++;
Emeric Brun2b920a12010-09-23 18:30:22 +02002133 }
2134
2135 /* disconnect all connected peers */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002136 for (ps = peers->remote; ps; ps = ps->next) {
Emeric Brun80527f52017-06-19 17:46:37 +02002137 /* we're killing a connection, we must apply a random delay before
2138 * retrying otherwise the other end will do the same and we can loop
2139 * for a while.
2140 */
2141 ps->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + random() % 2000));
Willy Tarreau9df94c22016-10-31 18:42:52 +01002142 if (ps->appctx) {
Willy Tarreau81bc3b02016-10-31 17:37:39 +01002143 peer_session_forceshutdown(ps->appctx);
Willy Tarreaue5843b32015-04-27 18:40:14 +02002144 ps->appctx = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02002145 }
2146 }
2147 }
Emeric Brun2b920a12010-09-23 18:30:22 +02002148
Emeric Brunb3971ab2015-05-12 18:49:09 +02002149 ps = peers->local;
Emeric Brun2b920a12010-09-23 18:30:22 +02002150 if (ps->flags & PEER_F_TEACH_COMPLETE) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02002151 if (peers->flags & PEERS_F_DONOTSTOP) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002152 /* resync of new process was complete, current process can die now */
Willy Tarreaucea85372017-11-29 14:49:30 +01002153 HA_ATOMIC_SUB(&jobs, 1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002154 peers->flags &= ~PEERS_F_DONOTSTOP;
2155 for (st = ps->tables; st ; st = st->next)
2156 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02002157 }
2158 }
Willy Tarreau9df94c22016-10-31 18:42:52 +01002159 else if (!ps->appctx) {
2160 /* If there's no active peer connection */
Emeric Brun2b920a12010-09-23 18:30:22 +02002161 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002162 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
2163 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE ||
2164 ps->statuscode == PEER_SESS_SC_TRYAGAIN) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002165 /* connection never tried
Willy Tarreau9df94c22016-10-31 18:42:52 +01002166 * or previous peer connection was successfully established
2167 * or previous tcp connect succeeded but init state incomplete
Emeric Brun2b920a12010-09-23 18:30:22 +02002168 * or during previous connect, peer replies a try again statuscode */
2169
2170 /* connect to the peer */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002171 peer_session_create(peers, ps);
Emeric Brun2b920a12010-09-23 18:30:22 +02002172 }
2173 else {
2174 /* Other error cases */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002175 if (peers->flags & PEERS_F_DONOTSTOP) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002176 /* unable to resync new process, current process can die now */
Christopher Faulet8d8aa0d2017-05-30 15:36:50 +02002177 HA_ATOMIC_SUB(&jobs, 1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002178 peers->flags &= ~PEERS_F_DONOTSTOP;
2179 for (st = ps->tables; st ; st = st->next)
2180 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02002181 }
2182 }
2183 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02002184 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE ) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002185 /* current peer connection is active and established
2186 * wake up all peer handlers to push remaining local updates */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002187 for (st = ps->tables; st ; st = st->next) {
2188 if ((int)(st->last_pushed - st->table->localupdate) < 0) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02002189 appctx_wakeup(ps->appctx);
2190 break;
2191 }
2192 }
Emeric Brun2b920a12010-09-23 18:30:22 +02002193 }
2194 } /* stopping */
Emeric Brun80527f52017-06-19 17:46:37 +02002195
2196 /* Release lock for all peers of the section */
2197 for (ps = peers->remote; ps; ps = ps->next)
Christopher Faulet2a944ee2017-11-07 10:42:54 +01002198 HA_SPIN_UNLOCK(PEER_LOCK, &ps->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02002199
Emeric Brun2b920a12010-09-23 18:30:22 +02002200 /* Wakeup for re-connect */
2201 return task;
2202}
2203
Emeric Brunb3971ab2015-05-12 18:49:09 +02002204
Emeric Brun2b920a12010-09-23 18:30:22 +02002205/*
Emeric Brun2b920a12010-09-23 18:30:22 +02002206 *
2207 */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002208void peers_init_sync(struct peers *peers)
Emeric Brun2b920a12010-09-23 18:30:22 +02002209{
Emeric Brun2b920a12010-09-23 18:30:22 +02002210 struct peer * curpeer;
Willy Tarreau4348fad2012-09-20 16:48:07 +02002211 struct listener *listener;
Emeric Brun2b920a12010-09-23 18:30:22 +02002212
Emeric Brun2b920a12010-09-23 18:30:22 +02002213 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002214 peers->peers_fe->maxconn += 3;
2215 }
2216
Willy Tarreau4348fad2012-09-20 16:48:07 +02002217 list_for_each_entry(listener, &peers->peers_fe->conf.listeners, by_fe)
2218 listener->maxconn = peers->peers_fe->maxconn;
Emeric Brunc60def82017-09-27 14:59:38 +02002219 peers->sync_task = task_new(MAX_THREADS_MASK);
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);
2224}
2225
2226
2227
2228/*
2229 * Function used to register a table for sync on a group of peers
2230 *
2231 */
2232void peers_register_table(struct peers *peers, struct stktable *table)
2233{
2234 struct shared_table *st;
2235 struct peer * curpeer;
2236 int id = 0;
2237
2238 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
Vincent Bernat02779b62016-04-03 13:48:43 +02002239 st = calloc(1,sizeof(*st));
Emeric Brunb3971ab2015-05-12 18:49:09 +02002240 st->table = table;
2241 st->next = curpeer->tables;
2242 if (curpeer->tables)
2243 id = curpeer->tables->local_id;
2244 st->local_id = id + 1;
2245
2246 curpeer->tables = st;
2247 }
2248
2249 table->sync_task = peers->sync_task;
Emeric Brun2b920a12010-09-23 18:30:22 +02002250}
2251