blob: e61caaf93f81eea0b9abbafd877d5e620f009feb [file] [log] [blame]
Emeric Brun2b920a12010-09-23 18:30:22 +02001/*
Emeric Brunb3971ab2015-05-12 18:49:09 +02002 * Peer synchro management.
Emeric Brun2b920a12010-09-23 18:30:22 +02003 *
4 * Copyright 2010 EXCELIANCE, Emeric Brun <ebrun@exceliance.fr>
5 *
6 * This program is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU General Public License
8 * as published by the Free Software Foundation; either version
9 * 2 of the License, or (at your option) any later version.
10 *
11 */
12
13#include <errno.h>
14#include <fcntl.h>
15#include <stdio.h>
16#include <stdlib.h>
17#include <string.h>
18
19#include <sys/socket.h>
20#include <sys/stat.h>
21#include <sys/types.h>
22
23#include <common/compat.h>
24#include <common/config.h>
25#include <common/time.h>
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +020026#include <common/standard.h>
Christopher Faulet8d8aa0d2017-05-30 15:36:50 +020027#include <common/hathreads.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020028
29#include <types/global.h>
Willy Tarreau3fdb3662012-11-12 00:42:33 +010030#include <types/listener.h>
31#include <types/obj_type.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020032#include <types/peers.h>
33
34#include <proto/acl.h>
Willy Tarreau8a8d83b2015-04-13 13:24:54 +020035#include <proto/applet.h>
Willy Tarreauc7e42382012-08-24 19:22:53 +020036#include <proto/channel.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020037#include <proto/fd.h>
Willy Tarreaud1d48d42015-03-13 16:15:46 +010038#include <proto/frontend.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020039#include <proto/log.h>
40#include <proto/hdr_idx.h>
Willy Tarreau53a47662017-08-28 10:53:00 +020041#include <proto/mux_pt.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020042#include <proto/proxy.h>
Willy Tarreaufeb76402015-04-03 14:10:06 +020043#include <proto/session.h>
Willy Tarreau87b09662015-04-03 00:22:06 +020044#include <proto/stream.h>
Willy Tarreau22ec1ea2014-11-27 20:45:39 +010045#include <proto/signal.h>
46#include <proto/stick_table.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020047#include <proto/stream_interface.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020048#include <proto/task.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020049
50
51/*******************************/
52/* Current peer learning state */
53/*******************************/
54
55/******************************/
Emeric Brunb3971ab2015-05-12 18:49:09 +020056/* Current peers section resync state */
Emeric Brun2b920a12010-09-23 18:30:22 +020057/******************************/
Emeric Brunb3971ab2015-05-12 18:49:09 +020058#define PEERS_F_RESYNC_LOCAL 0x00000001 /* Learn from local finished or no more needed */
59#define PEERS_F_RESYNC_REMOTE 0x00000002 /* Learn from remote finished or no more needed */
60#define PEERS_F_RESYNC_ASSIGN 0x00000004 /* A peer was assigned to learn our lesson */
61#define PEERS_F_RESYNC_PROCESS 0x00000008 /* The assigned peer was requested for resync */
62#define PEERS_F_DONOTSTOP 0x00010000 /* Main table sync task block process during soft stop
Emeric Brun2b920a12010-09-23 18:30:22 +020063 to push data to new process */
64
Emeric Brunb3971ab2015-05-12 18:49:09 +020065#define PEERS_RESYNC_STATEMASK (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE)
66#define PEERS_RESYNC_FROMLOCAL 0x00000000
67#define PEERS_RESYNC_FROMREMOTE PEERS_F_RESYNC_LOCAL
68#define PEERS_RESYNC_FINISHED (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE)
69
70/***********************************/
71/* Current shared table sync state */
72/***********************************/
73#define SHTABLE_F_TEACH_STAGE1 0x00000001 /* Teach state 1 complete */
74#define SHTABLE_F_TEACH_STAGE2 0x00000002 /* Teach state 2 complete */
Emeric Brun2b920a12010-09-23 18:30:22 +020075
76/******************************/
77/* Remote peer teaching state */
78/******************************/
79#define PEER_F_TEACH_PROCESS 0x00000001 /* Teach a lesson to current peer */
Emeric Brun2b920a12010-09-23 18:30:22 +020080#define PEER_F_TEACH_FINISHED 0x00000008 /* Teach conclude, (wait for confirm) */
81#define PEER_F_TEACH_COMPLETE 0x00000010 /* All that we know already taught to current peer, used only for a local peer */
82#define PEER_F_LEARN_ASSIGN 0x00000100 /* Current peer was assigned for a lesson */
83#define PEER_F_LEARN_NOTUP2DATE 0x00000200 /* Learn from peer finished but peer is not up to date */
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +020084#define PEER_F_DWNGRD 0x80000000 /* When this flag is enabled, we must downgrade the supported version announced during peer sessions. */
Emeric Brun2b920a12010-09-23 18:30:22 +020085
Emeric Brunb3971ab2015-05-12 18:49:09 +020086#define PEER_TEACH_RESET ~(PEER_F_TEACH_PROCESS|PEER_F_TEACH_FINISHED) /* PEER_F_TEACH_COMPLETE should never be reset */
Emeric Brun2b920a12010-09-23 18:30:22 +020087#define PEER_LEARN_RESET ~(PEER_F_LEARN_ASSIGN|PEER_F_LEARN_NOTUP2DATE)
88
Emeric Brunb3971ab2015-05-12 18:49:09 +020089/*****************************/
90/* Sync message class */
91/*****************************/
92enum {
93 PEER_MSG_CLASS_CONTROL = 0,
94 PEER_MSG_CLASS_ERROR,
95 PEER_MSG_CLASS_STICKTABLE = 10,
96 PEER_MSG_CLASS_RESERVED = 255,
97};
98
99/*****************************/
100/* control message types */
101/*****************************/
102enum {
103 PEER_MSG_CTRL_RESYNCREQ = 0,
104 PEER_MSG_CTRL_RESYNCFINISHED,
105 PEER_MSG_CTRL_RESYNCPARTIAL,
106 PEER_MSG_CTRL_RESYNCCONFIRM,
107};
108
109/*****************************/
110/* error message types */
111/*****************************/
112enum {
113 PEER_MSG_ERR_PROTOCOL = 0,
114 PEER_MSG_ERR_SIZELIMIT,
115};
116
117
118/*******************************/
119/* stick table sync mesg types */
120/* Note: ids >= 128 contains */
121/* id message cotains data */
122/*******************************/
123enum {
124 PEER_MSG_STKT_UPDATE = 128,
125 PEER_MSG_STKT_INCUPDATE,
126 PEER_MSG_STKT_DEFINE,
127 PEER_MSG_STKT_SWITCH,
128 PEER_MSG_STKT_ACK,
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200129 PEER_MSG_STKT_UPDATE_TIMED,
130 PEER_MSG_STKT_INCUPDATE_TIMED,
Emeric Brunb3971ab2015-05-12 18:49:09 +0200131};
Emeric Brun2b920a12010-09-23 18:30:22 +0200132
133/**********************************/
134/* Peer Session IO handler states */
135/**********************************/
136
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100137enum {
138 PEER_SESS_ST_ACCEPT = 0, /* Initial state for session create by an accept, must be zero! */
139 PEER_SESS_ST_GETVERSION, /* Validate supported protocol version */
140 PEER_SESS_ST_GETHOST, /* Validate host ID correspond to local host id */
141 PEER_SESS_ST_GETPEER, /* Validate peer ID correspond to a known remote peer id */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100142 /* after this point, data were possibly exchanged */
143 PEER_SESS_ST_SENDSUCCESS, /* Send ret code 200 (success) and wait for message */
144 PEER_SESS_ST_CONNECT, /* Initial state for session create on a connect, push presentation into buffer */
145 PEER_SESS_ST_GETSTATUS, /* Wait for the welcome message */
146 PEER_SESS_ST_WAITMSG, /* Wait for data messages */
147 PEER_SESS_ST_EXIT, /* Exit with status code */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200148 PEER_SESS_ST_ERRPROTO, /* Send error proto message before exit */
149 PEER_SESS_ST_ERRSIZE, /* Send error size message before exit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100150 PEER_SESS_ST_END, /* Killed session */
151};
Emeric Brun2b920a12010-09-23 18:30:22 +0200152
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100153/***************************************************/
154/* Peer Session status code - part of the protocol */
155/***************************************************/
Emeric Brun2b920a12010-09-23 18:30:22 +0200156
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100157#define PEER_SESS_SC_CONNECTCODE 100 /* connect in progress */
158#define PEER_SESS_SC_CONNECTEDCODE 110 /* tcp connect success */
Emeric Brun2b920a12010-09-23 18:30:22 +0200159
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100160#define PEER_SESS_SC_SUCCESSCODE 200 /* accept or connect successful */
Emeric Brun2b920a12010-09-23 18:30:22 +0200161
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100162#define PEER_SESS_SC_TRYAGAIN 300 /* try again later */
Emeric Brun2b920a12010-09-23 18:30:22 +0200163
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100164#define PEER_SESS_SC_ERRPROTO 501 /* error protocol */
165#define PEER_SESS_SC_ERRVERSION 502 /* unknown protocol version */
166#define PEER_SESS_SC_ERRHOST 503 /* bad host name */
167#define PEER_SESS_SC_ERRPEER 504 /* unknown peer */
Emeric Brun2b920a12010-09-23 18:30:22 +0200168
169#define PEER_SESSION_PROTO_NAME "HAProxyS"
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200170#define PEER_MAJOR_VER 2
171#define PEER_MINOR_VER 1
172#define PEER_DWNGRD_MINOR_VER 0
Emeric Brun2b920a12010-09-23 18:30:22 +0200173
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200174struct peers *cfg_peers = NULL;
Willy Tarreau81bc3b02016-10-31 17:37:39 +0100175static void peer_session_forceshutdown(struct appctx *appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +0200176
Emeric Brun18928af2017-03-29 16:32:53 +0200177/* This function encode an uint64 to 'dynamic' length format.
178 The encoded value is written at address *str, and the
179 caller must assure that size after *str is large enought.
180 At return, the *str is set at the next Byte after then
181 encoded integer. The function returns then length of the
182 encoded integer in Bytes */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200183int intencode(uint64_t i, char **str) {
184 int idx = 0;
185 unsigned char *msg;
186
187 if (!*str)
188 return 0;
189
190 msg = (unsigned char *)*str;
191 if (i < 240) {
192 msg[0] = (unsigned char)i;
193 *str = (char *)&msg[idx+1];
194 return (idx+1);
195 }
196
197 msg[idx] =(unsigned char)i | 240;
198 i = (i - 240) >> 4;
199 while (i >= 128) {
200 msg[++idx] = (unsigned char)i | 128;
201 i = (i - 128) >> 7;
202 }
203 msg[++idx] = (unsigned char)i;
204 *str = (char *)&msg[idx+1];
205 return (idx+1);
206}
207
208
209/* This function returns the decoded integer or 0
210 if decode failed
211 *str point on the beginning of the integer to decode
212 at the end of decoding *str point on the end of the
213 encoded integer or to null if end is reached */
Emeric Brun18928af2017-03-29 16:32:53 +0200214uint64_t intdecode(char **str, char *end)
215{
Emeric Brunb3971ab2015-05-12 18:49:09 +0200216 unsigned char *msg;
Emeric Brun18928af2017-03-29 16:32:53 +0200217 uint64_t i;
218 int shift;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200219
220 if (!*str)
221 return 0;
222
223 msg = (unsigned char *)*str;
Emeric Brun18928af2017-03-29 16:32:53 +0200224 if (msg >= (unsigned char *)end)
225 goto fail;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200226
Emeric Brun18928af2017-03-29 16:32:53 +0200227 i = *(msg++);
228 if (i >= 240) {
229 shift = 4;
230 do {
231 if (msg >= (unsigned char *)end)
232 goto fail;
233 i += (uint64_t)*msg << shift;
234 shift += 7;
235 } while (*(msg++) >= 128);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200236 }
Emeric Brun18928af2017-03-29 16:32:53 +0200237 *str = (char *)msg;
238 return i;
239
240 fail:
241 *str = NULL;
242 return 0;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200243}
Emeric Brun2b920a12010-09-23 18:30:22 +0200244
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200245/* Set the stick-table UPDATE message type byte at <msg_type> address,
246 * depending on <use_identifier> and <use_timed> boolean parameters.
247 * Always successful.
248 */
249static inline void peer_set_update_msg_type(char *msg_type, int use_identifier, int use_timed)
250{
251 if (use_timed) {
252 if (use_identifier)
253 *msg_type = PEER_MSG_STKT_UPDATE_TIMED;
254 else
255 *msg_type = PEER_MSG_STKT_INCUPDATE_TIMED;
256 }
257 else {
258 if (use_identifier)
259 *msg_type = PEER_MSG_STKT_UPDATE;
260 else
261 *msg_type = PEER_MSG_STKT_INCUPDATE;
262 }
263
264}
Emeric Brun2b920a12010-09-23 18:30:22 +0200265/*
Emeric Brunb3971ab2015-05-12 18:49:09 +0200266 * This prepare the data update message on the stick session <ts>, <st> is the considered
267 * stick table.
268 * <msg> is a buffer of <size> to recieve data message content
269 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
270 * check size)
Emeric Brun2b920a12010-09-23 18:30:22 +0200271 */
Emeric Brun819fc6f2017-06-13 19:37:32 +0200272static 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 +0200273{
274 uint32_t netinteger;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200275 unsigned short datalen;
276 char *cursor, *datamsg;
Emeric Brun94900952015-06-11 18:25:54 +0200277 unsigned int data_type;
278 void *data_ptr;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200279
280 cursor = datamsg = msg + 1 + 5;
281
Emeric Brun2b920a12010-09-23 18:30:22 +0200282 /* construct message */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200283
284 /* check if we need to send the update identifer */
Emeric Brun819fc6f2017-06-13 19:37:32 +0200285 if (!st->last_pushed || updateid < st->last_pushed || ((updateid - st->last_pushed) != 1)) {
Emeric Bruna6a09982015-09-22 15:34:19 +0200286 use_identifier = 1;
Emeric Brun2b920a12010-09-23 18:30:22 +0200287 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200288
289 /* encode update identifier if needed */
290 if (use_identifier) {
Emeric Brun819fc6f2017-06-13 19:37:32 +0200291 netinteger = htonl(updateid);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200292 memcpy(cursor, &netinteger, sizeof(netinteger));
293 cursor += sizeof(netinteger);
Emeric Brun2b920a12010-09-23 18:30:22 +0200294 }
295
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200296 if (use_timed) {
297 netinteger = htonl(tick_remain(now_ms, ts->expire));
298 memcpy(cursor, &netinteger, sizeof(netinteger));
299 cursor += sizeof(netinteger);
300 }
301
Emeric Brunb3971ab2015-05-12 18:49:09 +0200302 /* encode the key */
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +0200303 if (st->table->type == SMP_T_STR) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200304 int stlen = strlen((char *)ts->key.key);
305
Emeric Brunb3971ab2015-05-12 18:49:09 +0200306 intencode(stlen, &cursor);
307 memcpy(cursor, ts->key.key, stlen);
308 cursor += stlen;
Emeric Brun2b920a12010-09-23 18:30:22 +0200309 }
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +0200310 else if (st->table->type == SMP_T_SINT) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200311 netinteger = htonl(*((uint32_t *)ts->key.key));
Emeric Brunb3971ab2015-05-12 18:49:09 +0200312 memcpy(cursor, &netinteger, sizeof(netinteger));
313 cursor += sizeof(netinteger);
Emeric Brun2b920a12010-09-23 18:30:22 +0200314 }
315 else {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200316 memcpy(cursor, ts->key.key, st->table->key_size);
317 cursor += st->table->key_size;
Emeric Brun2b920a12010-09-23 18:30:22 +0200318 }
319
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100320 HA_RWLOCK_RDLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200321 /* encode values */
Emeric Brun94900952015-06-11 18:25:54 +0200322 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200323
Emeric Brun94900952015-06-11 18:25:54 +0200324 data_ptr = stktable_data_ptr(st->table, ts, data_type);
325 if (data_ptr) {
326 switch (stktable_data_types[data_type].std_type) {
327 case STD_T_SINT: {
328 int data;
329
330 data = stktable_data_cast(data_ptr, std_t_sint);
331 intencode(data, &cursor);
332 break;
333 }
334 case STD_T_UINT: {
335 unsigned int data;
336
337 data = stktable_data_cast(data_ptr, std_t_uint);
338 intencode(data, &cursor);
339 break;
340 }
341 case STD_T_ULL: {
342 unsigned long long data;
343
344 data = stktable_data_cast(data_ptr, std_t_ull);
345 intencode(data, &cursor);
346 break;
347 }
348 case STD_T_FRQP: {
349 struct freq_ctr_period *frqp;
350
351 frqp = &stktable_data_cast(data_ptr, std_t_frqp);
352 intencode((unsigned int)(now_ms - frqp->curr_tick), &cursor);
353 intencode(frqp->curr_ctr, &cursor);
354 intencode(frqp->prev_ctr, &cursor);
355 break;
356 }
357 }
358 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200359 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100360 HA_RWLOCK_RDUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200361
362 /* Compute datalen */
363 datalen = (cursor - datamsg);
364
365 /* prepare message header */
366 msg[0] = PEER_MSG_CLASS_STICKTABLE;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200367 peer_set_update_msg_type(&msg[1], use_identifier, use_timed);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200368 cursor = &msg[2];
369 intencode(datalen, &cursor);
370
371 /* move data after header */
372 memmove(cursor, datamsg, datalen);
373
374 /* return header size + data_len */
375 return (cursor - msg) + datalen;
376}
377
378/*
379 * This prepare the switch table message to targeted share table <st>.
380 * <msg> is a buffer of <size> to recieve data message content
381 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
382 * check size)
383 */
384static int peer_prepare_switchmsg(struct shared_table *st, char *msg, size_t size)
385{
386 int len;
387 unsigned short datalen;
Willy Tarreau83061a82018-07-13 11:56:34 +0200388 struct buffer *chunk;
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200389 char *cursor, *datamsg, *chunkp, *chunkq;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200390 uint64_t data = 0;
Emeric Brun94900952015-06-11 18:25:54 +0200391 unsigned int data_type;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200392
393 cursor = datamsg = msg + 2 + 5;
394
395 /* Encode data */
396
397 /* encode local id */
398 intencode(st->local_id, &cursor);
399
400 /* encode table name */
401 len = strlen(st->table->id);
402 intencode(len, &cursor);
403 memcpy(cursor, st->table->id, len);
404 cursor += len;
405
406 /* encode table type */
407
408 intencode(st->table->type, &cursor);
409
410 /* encode table key size */
411 intencode(st->table->key_size, &cursor);
412
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200413 chunk = get_trash_chunk();
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200414 chunkp = chunkq = chunk->area;
Emeric Brun94900952015-06-11 18:25:54 +0200415 /* encode available known data types in table */
416 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
417 if (st->table->data_ofs[data_type]) {
418 switch (stktable_data_types[data_type].std_type) {
419 case STD_T_SINT:
420 case STD_T_UINT:
421 case STD_T_ULL:
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200422 data |= 1 << data_type;
423 break;
Emeric Brun94900952015-06-11 18:25:54 +0200424 case STD_T_FRQP:
425 data |= 1 << data_type;
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200426 intencode(data_type, &chunkq);
427 intencode(st->table->data_arg[data_type].u, &chunkq);
Emeric Brun94900952015-06-11 18:25:54 +0200428 break;
429 }
430 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200431 }
432 intencode(data, &cursor);
433
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200434 /* Encode stick-table entries duration. */
435 intencode(st->table->expire, &cursor);
436
437 if (chunkq > chunkp) {
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200438 chunk->data = chunkq - chunkp;
439 memcpy(cursor, chunk->area, chunk->data);
440 cursor += chunk->data;
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200441 }
442
Emeric Brunb3971ab2015-05-12 18:49:09 +0200443 /* Compute datalen */
444 datalen = (cursor - datamsg);
Emeric Brun2b920a12010-09-23 18:30:22 +0200445
Emeric Brunb3971ab2015-05-12 18:49:09 +0200446 /* prepare message header */
447 msg[0] = PEER_MSG_CLASS_STICKTABLE;
448 msg[1] = PEER_MSG_STKT_DEFINE;
449 cursor = &msg[2];
450 intencode(datalen, &cursor);
Emeric Brun2b920a12010-09-23 18:30:22 +0200451
Emeric Brunb3971ab2015-05-12 18:49:09 +0200452 /* move data after header */
453 memmove(cursor, datamsg, datalen);
454
455 /* return header size + data_len */
456 return (cursor - msg) + datalen;
Emeric Brun2b920a12010-09-23 18:30:22 +0200457}
458
Emeric Brunb3971ab2015-05-12 18:49:09 +0200459/*
460 * This prepare the acknowledge message on the stick session <ts>, <st> is the considered
461 * stick table.
462 * <msg> is a buffer of <size> to recieve data message content
463 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
464 * check size)
465 */
466static int peer_prepare_ackmsg(struct shared_table *st, char *msg, size_t size)
467{
468 unsigned short datalen;
469 char *cursor, *datamsg;
470 uint32_t netinteger;
471
Emeric Brunb058f1c2015-09-22 15:50:18 +0200472 cursor = datamsg = msg + 2 + 5;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200473
474 intencode(st->remote_id, &cursor);
475 netinteger = htonl(st->last_get);
476 memcpy(cursor, &netinteger, sizeof(netinteger));
477 cursor += sizeof(netinteger);
478
479 /* Compute datalen */
480 datalen = (cursor - datamsg);
481
482 /* prepare message header */
483 msg[0] = PEER_MSG_CLASS_STICKTABLE;
Emeric Brune1ab8082015-08-21 11:48:54 +0200484 msg[1] = PEER_MSG_STKT_ACK;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200485 cursor = &msg[2];
486 intencode(datalen, &cursor);
487
488 /* move data after header */
489 memmove(cursor, datamsg, datalen);
490
491 /* return header size + data_len */
492 return (cursor - msg) + datalen;
493}
Emeric Brun2b920a12010-09-23 18:30:22 +0200494
495/*
496 * Callback to release a session with a peer
497 */
Willy Tarreau00a37f02015-04-13 12:05:19 +0200498static void peer_session_release(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +0200499{
Willy Tarreau00a37f02015-04-13 12:05:19 +0200500 struct stream_interface *si = appctx->owner;
Willy Tarreau87b09662015-04-03 00:22:06 +0200501 struct stream *s = si_strm(si);
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200502 struct peer *peer = appctx->ctx.peers.ptr;
503 struct peers *peers = strm_fe(s)->parent;
Emeric Brun2b920a12010-09-23 18:30:22 +0200504
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100505 /* appctx->ctx.peers.ptr is not a peer session */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100506 if (appctx->st0 < PEER_SESS_ST_SENDSUCCESS)
Emeric Brun2b920a12010-09-23 18:30:22 +0200507 return;
508
509 /* peer session identified */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200510 if (peer) {
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100511 HA_SPIN_LOCK(PEER_LOCK, &peer->lock);
Willy Tarreau9df94c22016-10-31 18:42:52 +0100512 if (peer->appctx == appctx) {
Emeric Brunb157d732015-08-21 12:00:30 +0200513 /* Re-init current table pointers to force announcement on re-connect */
514 peer->remote_table = peer->last_local_table = NULL;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200515 peer->appctx = NULL;
516 if (peer->flags & PEER_F_LEARN_ASSIGN) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200517 /* unassign current peer for learning */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200518 peer->flags &= ~(PEER_F_LEARN_ASSIGN);
519 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brun2b920a12010-09-23 18:30:22 +0200520
521 /* reschedule a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200522 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
Emeric Brun2b920a12010-09-23 18:30:22 +0200523 }
524 /* reset teaching and learning flags to 0 */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200525 peer->flags &= PEER_TEACH_RESET;
526 peer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200527 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100528 HA_SPIN_UNLOCK(PEER_LOCK, &peer->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200529 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200530 }
531}
532
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200533/* Retrieve the major and minor versions of peers protocol
534 * announced by a remote peer. <str> is a null-terminated
535 * string with the following format: "<maj_ver>.<min_ver>".
536 */
537static int peer_get_version(const char *str,
538 unsigned int *maj_ver, unsigned int *min_ver)
539{
540 unsigned int majv, minv;
541 const char *pos, *saved;
542 const char *end;
543
544 saved = pos = str;
545 end = str + strlen(str);
546
547 majv = read_uint(&pos, end);
548 if (saved == pos || *pos++ != '.')
549 return -1;
550
551 saved = pos;
552 minv = read_uint(&pos, end);
553 if (saved == pos || pos != end)
554 return -1;
555
556 *maj_ver = majv;
557 *min_ver = minv;
558
559 return 0;
560}
Emeric Brun2b920a12010-09-23 18:30:22 +0200561
562/*
563 * IO Handler to handle message exchance with a peer
564 */
Willy Tarreau00a37f02015-04-13 12:05:19 +0200565static void peer_io_handler(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +0200566{
Willy Tarreau00a37f02015-04-13 12:05:19 +0200567 struct stream_interface *si = appctx->owner;
Willy Tarreau87b09662015-04-03 00:22:06 +0200568 struct stream *s = si_strm(si);
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200569 struct peers *curpeers = strm_fe(s)->parent;
Emeric Brun80527f52017-06-19 17:46:37 +0200570 struct peer *curpeer = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +0200571 int reql = 0;
572 int repl = 0;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200573 size_t proto_len = strlen(PEER_SESSION_PROTO_NAME);
574 unsigned int maj_ver, min_ver;
Emeric Brun2b920a12010-09-23 18:30:22 +0200575
Christopher Fauleta73e59b2016-12-09 17:30:18 +0100576 /* Check if the input buffer is avalaible. */
Willy Tarreauc9fa0482018-07-10 17:43:27 +0200577 if (si_ic(si)->buf.size == 0)
Christopher Fauleta73e59b2016-12-09 17:30:18 +0100578 goto full;
579
Emeric Brun2b920a12010-09-23 18:30:22 +0200580 while (1) {
581switchstate:
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200582 maj_ver = min_ver = (unsigned int)-1;
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100583 switch(appctx->st0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100584 case PEER_SESS_ST_ACCEPT:
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100585 appctx->ctx.peers.ptr = NULL;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100586 appctx->st0 = PEER_SESS_ST_GETVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200587 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100588 case PEER_SESS_ST_GETVERSION:
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200589 reql = co_getline(si_oc(si), trash.area,
590 trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200591 if (reql <= 0) { /* closed or EOL not found */
592 if (reql == 0)
593 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100594 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200595 goto switchstate;
596 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200597 if (trash.area[reql-1] != '\n') {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100598 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200599 goto switchstate;
600 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200601 else if (reql > 1 && (trash.area[reql-2] == '\r'))
602 trash.area[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200603 else
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200604 trash.area[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200605
Willy Tarreau06d80a92017-10-19 14:32:15 +0200606 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200607
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200608 /* test protocol */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200609 if (strncmp(PEER_SESSION_PROTO_NAME " ", trash.area, proto_len + 1) != 0) {
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200610 appctx->st0 = PEER_SESS_ST_EXIT;
611 appctx->st1 = PEER_SESS_SC_ERRPROTO;
612 goto switchstate;
613 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200614 if (peer_get_version(trash.area + proto_len + 1, &maj_ver, &min_ver) == -1 ||
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200615 maj_ver != PEER_MAJOR_VER || min_ver > PEER_MINOR_VER) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100616 appctx->st0 = PEER_SESS_ST_EXIT;
617 appctx->st1 = PEER_SESS_SC_ERRVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200618 goto switchstate;
619 }
620
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100621 appctx->st0 = PEER_SESS_ST_GETHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200622 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100623 case PEER_SESS_ST_GETHOST:
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200624 reql = co_getline(si_oc(si), trash.area,
625 trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200626 if (reql <= 0) { /* closed or EOL not found */
627 if (reql == 0)
628 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100629 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200630 goto switchstate;
631 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200632 if (trash.area[reql-1] != '\n') {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100633 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200634 goto switchstate;
635 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200636 else if (reql > 1 && (trash.area[reql-2] == '\r'))
637 trash.area[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200638 else
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200639 trash.area[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200640
Willy Tarreau06d80a92017-10-19 14:32:15 +0200641 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200642
643 /* test hostname match */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200644 if (strcmp(localpeer, trash.area) != 0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100645 appctx->st0 = PEER_SESS_ST_EXIT;
646 appctx->st1 = PEER_SESS_SC_ERRHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200647 goto switchstate;
648 }
649
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100650 appctx->st0 = PEER_SESS_ST_GETPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200651 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100652 case PEER_SESS_ST_GETPEER: {
Emeric Brun2b920a12010-09-23 18:30:22 +0200653 char *p;
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200654 reql = co_getline(si_oc(si), trash.area,
655 trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200656 if (reql <= 0) { /* closed or EOL not found */
657 if (reql == 0)
658 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100659 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200660 goto switchstate;
661 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200662 if (trash.area[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200663 /* Incomplete line, we quit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100664 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200665 goto switchstate;
666 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200667 else if (reql > 1 && (trash.area[reql-2] == '\r'))
668 trash.area[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200669 else
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200670 trash.area[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200671
Willy Tarreau06d80a92017-10-19 14:32:15 +0200672 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200673
Emeric Brunb3971ab2015-05-12 18:49:09 +0200674 /* parse line "<peer name> <pid> <relative_pid>" */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200675 p = strchr(trash.area, ' ');
Emeric Brun2b920a12010-09-23 18:30:22 +0200676 if (!p) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100677 appctx->st0 = PEER_SESS_ST_EXIT;
678 appctx->st1 = PEER_SESS_SC_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +0200679 goto switchstate;
680 }
681 *p = 0;
682
683 /* lookup known peer */
684 for (curpeer = curpeers->remote; curpeer; curpeer = curpeer->next) {
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200685 if (strcmp(curpeer->id, trash.area) == 0)
Emeric Brun2b920a12010-09-23 18:30:22 +0200686 break;
687 }
688
689 /* if unknown peer */
690 if (!curpeer) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100691 appctx->st0 = PEER_SESS_ST_EXIT;
692 appctx->st1 = PEER_SESS_SC_ERRPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200693 goto switchstate;
694 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200695
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100696 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Willy Tarreau9df94c22016-10-31 18:42:52 +0100697 if (curpeer->appctx && curpeer->appctx != appctx) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200698 if (curpeer->local) {
699 /* Local connection, reply a retry */
700 appctx->st0 = PEER_SESS_ST_EXIT;
701 appctx->st1 = PEER_SESS_SC_TRYAGAIN;
702 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +0200703 }
Emeric Brun80527f52017-06-19 17:46:37 +0200704
705 /* we're killing a connection, we must apply a random delay before
706 * retrying otherwise the other end will do the same and we can loop
707 * for a while.
708 */
709 curpeer->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + random() % 2000));
Willy Tarreau81bc3b02016-10-31 17:37:39 +0100710 peer_session_forceshutdown(curpeer->appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +0200711 }
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200712 if (maj_ver != (unsigned int)-1 && min_ver != (unsigned int)-1) {
713 if (min_ver == PEER_DWNGRD_MINOR_VER) {
714 curpeer->flags |= PEER_F_DWNGRD;
715 }
716 else {
717 curpeer->flags &= ~PEER_F_DWNGRD;
718 }
719 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200720 curpeer->appctx = appctx;
721 appctx->ctx.peers.ptr = curpeer;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100722 appctx->st0 = PEER_SESS_ST_SENDSUCCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200723 /* fall through */
724 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100725 case PEER_SESS_ST_SENDSUCCESS: {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200726 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +0200727
Emeric Brun80527f52017-06-19 17:46:37 +0200728 if (!curpeer) {
729 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100730 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +0200731 if (curpeer->appctx != appctx) {
732 appctx->st0 = PEER_SESS_ST_END;
733 goto switchstate;
734 }
735 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200736 repl = snprintf(trash.area, trash.size,
737 "%d\n",
738 PEER_SESS_SC_SUCCESSCODE);
739 repl = ci_putblk(si_ic(si), trash.area, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200740 if (repl <= 0) {
741 if (repl == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +0100742 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100743 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200744 goto switchstate;
745 }
746
747 /* Register status code */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200748 curpeer->statuscode = PEER_SESS_SC_SUCCESSCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200749
750 /* Awake main task */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200751 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200752
753 /* Init confirm counter */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200754 curpeer->confirm = 0;
755
756 /* Init cursors */
757 for (st = curpeer->tables; st ; st = st->next) {
758 st->last_get = st->last_acked = 0;
759 st->teaching_origin = st->last_pushed = st->update;
760 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200761
762 /* reset teaching and learning flags to 0 */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200763 curpeer->flags &= PEER_TEACH_RESET;
764 curpeer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200765
766 /* if current peer is local */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200767 if (curpeer->local) {
768 /* if current host need resyncfrom local and no process assined */
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200769 if ((curpeers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL &&
770 !(curpeers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200771 /* assign local peer for a lesson, consider lesson already requested */
772 curpeer->flags |= PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200773 curpeers->flags |= (PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200774 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200775
Emeric Brunb3971ab2015-05-12 18:49:09 +0200776 }
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200777 else if ((curpeers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
778 !(curpeers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200779 /* assign peer for a lesson */
780 curpeer->flags |= PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200781 curpeers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200782 }
783
784
Emeric Brun2b920a12010-09-23 18:30:22 +0200785 /* switch to waiting message state */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100786 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200787 goto switchstate;
788 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100789 case PEER_SESS_ST_CONNECT: {
Emeric Brun80527f52017-06-19 17:46:37 +0200790
791 if (!curpeer) {
792 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100793 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +0200794 if (curpeer->appctx != appctx) {
795 appctx->st0 = PEER_SESS_ST_END;
796 goto switchstate;
797 }
798 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200799
800 /* Send headers */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200801 repl = snprintf(trash.area, trash.size,
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200802 PEER_SESSION_PROTO_NAME " %u.%u\n%s\n%s %d %d\n",
803 PEER_MAJOR_VER,
804 (curpeer->flags & PEER_F_DWNGRD) ? PEER_DWNGRD_MINOR_VER : PEER_MINOR_VER,
Emeric Brunb3971ab2015-05-12 18:49:09 +0200805 curpeer->id,
Emeric Brun2b920a12010-09-23 18:30:22 +0200806 localpeer,
Willy Tarreau7b77c9f2012-01-07 22:52:12 +0100807 (int)getpid(),
Emeric Brunb3971ab2015-05-12 18:49:09 +0200808 relative_pid);
Emeric Brun2b920a12010-09-23 18:30:22 +0200809
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100810 if (repl >= trash.size) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100811 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200812 goto switchstate;
813 }
814
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200815 repl = ci_putblk(si_ic(si), trash.area, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200816 if (repl <= 0) {
817 if (repl == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +0100818 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100819 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200820 goto switchstate;
821 }
822
823 /* switch to the waiting statuscode state */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100824 appctx->st0 = PEER_SESS_ST_GETSTATUS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200825 /* fall through */
826 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100827 case PEER_SESS_ST_GETSTATUS: {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200828 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +0200829
Emeric Brun80527f52017-06-19 17:46:37 +0200830 if (!curpeer) {
831 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100832 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +0200833 if (curpeer->appctx != appctx) {
834 appctx->st0 = PEER_SESS_ST_END;
835 goto switchstate;
836 }
837 }
838
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100839 if (si_ic(si)->flags & CF_WRITE_PARTIAL)
Emeric Brunb3971ab2015-05-12 18:49:09 +0200840 curpeer->statuscode = PEER_SESS_SC_CONNECTEDCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200841
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200842 reql = co_getline(si_oc(si), trash.area,
843 trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200844 if (reql <= 0) { /* closed or EOL not found */
845 if (reql == 0)
846 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100847 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200848 goto switchstate;
849 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200850 if (trash.area[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200851 /* Incomplete line, we quit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100852 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200853 goto switchstate;
854 }
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200855 else if (reql > 1 && (trash.area[reql-2] == '\r'))
856 trash.area[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200857 else
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200858 trash.area[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200859
Willy Tarreau06d80a92017-10-19 14:32:15 +0200860 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200861
862 /* Register status code */
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200863 curpeer->statuscode = atoi(trash.area);
Emeric Brun2b920a12010-09-23 18:30:22 +0200864
865 /* Awake main task */
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200866 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200867
868 /* If status code is success */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200869 if (curpeer->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200870 /* Init cursors */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200871 for (st = curpeer->tables; st ; st = st->next) {
872 st->last_get = st->last_acked = 0;
873 st->teaching_origin = st->last_pushed = st->update;
874 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200875
876 /* Init confirm counter */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200877 curpeer->confirm = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200878
Emeric Brunb3971ab2015-05-12 18:49:09 +0200879 /* reset teaching and learning flags to 0 */
880 curpeer->flags &= PEER_TEACH_RESET;
881 curpeer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200882
Emeric Brunb3971ab2015-05-12 18:49:09 +0200883 /* If current peer is local */
884 if (curpeer->local) {
885 /* flag to start to teach lesson */
886 curpeer->flags |= PEER_F_TEACH_PROCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200887
Emeric Brunb3971ab2015-05-12 18:49:09 +0200888 }
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200889 else if ((curpeers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
890 !(curpeers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200891 /* If peer is remote and resync from remote is needed,
892 and no peer currently assigned */
Emeric Brun2b920a12010-09-23 18:30:22 +0200893
Emeric Brunb3971ab2015-05-12 18:49:09 +0200894 /* assign peer for a lesson */
895 curpeer->flags |= PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200896 curpeers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brun2b920a12010-09-23 18:30:22 +0200897 }
898
899 }
900 else {
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200901 if (curpeer->statuscode == PEER_SESS_SC_ERRVERSION)
902 curpeer->flags |= PEER_F_DWNGRD;
Emeric Brun2b920a12010-09-23 18:30:22 +0200903 /* Status code is not success, abort */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100904 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200905 goto switchstate;
906 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100907 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200908 /* fall through */
909 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100910 case PEER_SESS_ST_WAITMSG: {
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200911 struct stksess *ts, *newts = NULL;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200912 uint32_t msg_len = 0;
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200913 char *msg_cur = trash.area;
914 char *msg_end = trash.area;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200915 unsigned char msg_head[7];
Emeric Brun2b920a12010-09-23 18:30:22 +0200916 int totl = 0;
917
Emeric Brun80527f52017-06-19 17:46:37 +0200918 if (!curpeer) {
919 curpeer = appctx->ctx.peers.ptr;
Christopher Faulet2a944ee2017-11-07 10:42:54 +0100920 HA_SPIN_LOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +0200921 if (curpeer->appctx != appctx) {
922 appctx->st0 = PEER_SESS_ST_END;
923 goto switchstate;
924 }
925 }
926
Willy Tarreau06d80a92017-10-19 14:32:15 +0200927 reql = co_getblk(si_oc(si), (char *)msg_head, 2*sizeof(unsigned char), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200928 if (reql <= 0) /* closed or EOL not found */
929 goto incomplete;
930
Emeric Brun2b920a12010-09-23 18:30:22 +0200931 totl += reql;
932
Emeric Brunb3971ab2015-05-12 18:49:09 +0200933 if (msg_head[1] >= 128) {
934 /* Read and Decode message length */
Willy Tarreau06d80a92017-10-19 14:32:15 +0200935 reql = co_getblk(si_oc(si), (char *)&msg_head[2], sizeof(unsigned char), totl);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200936 if (reql <= 0) /* closed */
937 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +0200938
Emeric Brunb3971ab2015-05-12 18:49:09 +0200939 totl += reql;
940
941 if (msg_head[2] < 240) {
942 msg_len = msg_head[2];
Emeric Brun2b920a12010-09-23 18:30:22 +0200943 }
944 else {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200945 int i;
946 char *cur;
947 char *end;
Willy Tarreau72d6c162013-04-11 16:14:13 +0200948
Emeric Brunb3971ab2015-05-12 18:49:09 +0200949 for (i = 3 ; i < sizeof(msg_head) ; i++) {
Willy Tarreau06d80a92017-10-19 14:32:15 +0200950 reql = co_getblk(si_oc(si), (char *)&msg_head[i], sizeof(char), totl);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200951 if (reql <= 0) /* closed */
952 goto incomplete;
953
954 totl += reql;
955
956 if (!(msg_head[i] & 0x80))
957 break;
958 }
959
960 if (i == sizeof(msg_head)) {
961 /* malformed message */
962 appctx->st0 = PEER_SESS_ST_ERRPROTO;
963 goto switchstate;
964
965 }
966 end = (char *)msg_head + sizeof(msg_head);
967 cur = (char *)&msg_head[2];
968 msg_len = intdecode(&cur, end);
969 if (!cur) {
970 /* malformed message */
971 appctx->st0 = PEER_SESS_ST_ERRPROTO;
972 goto switchstate;
973 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200974 }
975
Willy Tarreau86a446e2013-11-25 23:02:37 +0100976
Emeric Brunb3971ab2015-05-12 18:49:09 +0200977 /* Read message content */
978 if (msg_len) {
979 if (msg_len > trash.size) {
980 /* Status code is not success, abort */
981 appctx->st0 = PEER_SESS_ST_ERRSIZE;
982 goto switchstate;
983 }
Willy Tarreau72d6c162013-04-11 16:14:13 +0200984
Willy Tarreau843b7cb2018-07-13 10:54:26 +0200985 reql = co_getblk(si_oc(si),
986 trash.area,
987 msg_len,
988 totl);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200989 if (reql <= 0) /* closed */
990 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +0200991 totl += reql;
Willy Tarreau86a446e2013-11-25 23:02:37 +0100992
Emeric Brunb3971ab2015-05-12 18:49:09 +0200993 msg_end += msg_len;
994 }
995 }
Willy Tarreau86a446e2013-11-25 23:02:37 +0100996
Emeric Brunb3971ab2015-05-12 18:49:09 +0200997 if (msg_head[0] == PEER_MSG_CLASS_CONTROL) {
998 if (msg_head[1] == PEER_MSG_CTRL_RESYNCREQ) {
999 struct shared_table *st;
1000 /* Reset message: remote need resync */
1001
1002 /* prepare tables fot a global push */
1003 for (st = curpeer->tables; st; st = st->next) {
1004 st->teaching_origin = st->last_pushed = st->table->update;
1005 st->flags = 0;
Willy Tarreau86a446e2013-11-25 23:02:37 +01001006 }
Willy Tarreau72d6c162013-04-11 16:14:13 +02001007
Emeric Brunb3971ab2015-05-12 18:49:09 +02001008 /* reset teaching flags to 0 */
1009 curpeer->flags &= PEER_TEACH_RESET;
Willy Tarreau9d9179b2013-04-11 16:56:44 +02001010
Emeric Brunb3971ab2015-05-12 18:49:09 +02001011 /* flag to start to teach lesson */
1012 curpeer->flags |= PEER_F_TEACH_PROCESS;
1013
1014
1015 }
1016 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCFINISHED) {
1017
1018 if (curpeer->flags & PEER_F_LEARN_ASSIGN) {
1019 curpeer->flags &= ~PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001020 curpeers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
1021 curpeers->flags |= (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE);
Willy Tarreau86a446e2013-11-25 23:02:37 +01001022 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001023 curpeer->confirm++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001024 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001025 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCPARTIAL) {
1026
1027 if (curpeer->flags & PEER_F_LEARN_ASSIGN) {
1028 curpeer->flags &= ~PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001029 curpeers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001030
1031 curpeer->flags |= PEER_F_LEARN_NOTUP2DATE;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001032 curpeers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
1033 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Cyril Bonté9a60ff92014-02-16 01:07:07 +01001034 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001035 curpeer->confirm++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001036 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001037 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCCONFIRM) {
Emeric Brun597b26e2016-08-12 11:23:31 +02001038 struct shared_table *st;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001039
1040 /* If stopping state */
1041 if (stopping) {
1042 /* Close session, push resync no more needed */
1043 curpeer->flags |= PEER_F_TEACH_COMPLETE;
1044 appctx->st0 = PEER_SESS_ST_END;
1045 goto switchstate;
1046 }
Emeric Brun597b26e2016-08-12 11:23:31 +02001047 for (st = curpeer->tables; st; st = st->next) {
1048 st->update = st->last_pushed = st->teaching_origin;
1049 st->flags = 0;
1050 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001051
1052 /* reset teaching flags to 0 */
1053 curpeer->flags &= PEER_TEACH_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +02001054 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001055 }
1056 else if (msg_head[0] == PEER_MSG_CLASS_STICKTABLE) {
1057 if (msg_head[1] == PEER_MSG_STKT_DEFINE) {
1058 int table_id_len;
1059 struct shared_table *st;
1060 int table_type;
1061 int table_keylen;
1062 int table_id;
1063 uint64_t table_data;
Emeric Brun2b920a12010-09-23 18:30:22 +02001064
Emeric Brunb3971ab2015-05-12 18:49:09 +02001065 table_id = intdecode(&msg_cur, msg_end);
1066 if (!msg_cur) {
1067 /* malformed message */
1068 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1069 goto switchstate;
1070 }
Willy Tarreau72d6c162013-04-11 16:14:13 +02001071
Emeric Brunb3971ab2015-05-12 18:49:09 +02001072 table_id_len = intdecode(&msg_cur, msg_end);
1073 if (!msg_cur) {
1074 /* malformed message */
1075 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1076 goto switchstate;
1077 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001078
Emeric Brunb3971ab2015-05-12 18:49:09 +02001079 curpeer->remote_table = NULL;
1080 if (!table_id_len || (msg_cur + table_id_len) >= msg_end) {
1081 /* malformed message */
1082 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1083 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001084 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001085
Emeric Brunb3971ab2015-05-12 18:49:09 +02001086 for (st = curpeer->tables; st; st = st->next) {
1087 /* Reset IDs */
1088 if (st->remote_id == table_id)
1089 st->remote_id = 0;
Willy Tarreau9d9179b2013-04-11 16:56:44 +02001090
Emeric Brunb3971ab2015-05-12 18:49:09 +02001091 if (!curpeer->remote_table
1092 && (table_id_len == strlen(st->table->id))
1093 && (memcmp(st->table->id, msg_cur, table_id_len) == 0)) {
1094 curpeer->remote_table = st;
Emeric Brun2b920a12010-09-23 18:30:22 +02001095 }
1096 }
1097
Emeric Brunb3971ab2015-05-12 18:49:09 +02001098 if (!curpeer->remote_table) {
1099 goto ignore_msg;
1100 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001101
Emeric Brunb3971ab2015-05-12 18:49:09 +02001102 msg_cur += table_id_len;
1103 if (msg_cur >= msg_end) {
1104 /* malformed message */
1105 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1106 goto switchstate;
1107 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001108
Emeric Brunb3971ab2015-05-12 18:49:09 +02001109 table_type = intdecode(&msg_cur, msg_end);
1110 if (!msg_cur) {
1111 /* malformed message */
1112 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1113 goto switchstate;
1114 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001115
Emeric Brunb3971ab2015-05-12 18:49:09 +02001116 table_keylen = intdecode(&msg_cur, msg_end);
1117 if (!msg_cur) {
1118 /* malformed message */
1119 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1120 goto switchstate;
1121 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001122
Emeric Brunb3971ab2015-05-12 18:49:09 +02001123 table_data = intdecode(&msg_cur, msg_end);
1124 if (!msg_cur) {
1125 /* malformed message */
1126 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1127 goto switchstate;
1128 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001129
Emeric Brunb3971ab2015-05-12 18:49:09 +02001130 if (curpeer->remote_table->table->type != table_type
1131 || curpeer->remote_table->table->key_size != table_keylen) {
1132 curpeer->remote_table = NULL;
1133 goto ignore_msg;
1134 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001135
Emeric Brunb3971ab2015-05-12 18:49:09 +02001136 curpeer->remote_table->remote_data = table_data;
1137 curpeer->remote_table->remote_id = table_id;
Emeric Brun2b920a12010-09-23 18:30:22 +02001138 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001139 else if (msg_head[1] == PEER_MSG_STKT_SWITCH) {
1140 struct shared_table *st;
1141 int table_id;
Emeric Brun2b920a12010-09-23 18:30:22 +02001142
Emeric Brunb3971ab2015-05-12 18:49:09 +02001143 table_id = intdecode(&msg_cur, msg_end);
1144 if (!msg_cur) {
1145 /* malformed message */
1146 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1147 goto switchstate;
1148 }
1149 curpeer->remote_table = NULL;
1150 for (st = curpeer->tables; st; st = st->next) {
1151 if (st->remote_id == table_id) {
1152 curpeer->remote_table = st;
1153 break;
1154 }
1155 }
1156
Emeric Brun2b920a12010-09-23 18:30:22 +02001157 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001158 else if (msg_head[1] == PEER_MSG_STKT_UPDATE
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001159 || msg_head[1] == PEER_MSG_STKT_INCUPDATE
1160 || msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED
1161 || msg_head[1] == PEER_MSG_STKT_INCUPDATE_TIMED) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001162 struct shared_table *st = curpeer->remote_table;
1163 uint32_t update;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001164 int expire;
Emeric Brun94900952015-06-11 18:25:54 +02001165 unsigned int data_type;
1166 void *data_ptr;
Emeric Brun2b920a12010-09-23 18:30:22 +02001167
Emeric Brunb3971ab2015-05-12 18:49:09 +02001168 /* Here we have data message */
1169 if (!st)
1170 goto ignore_msg;
Emeric Brun2b920a12010-09-23 18:30:22 +02001171
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001172 expire = MS_TO_TICKS(st->table->expire);
1173
1174 if (msg_head[1] == PEER_MSG_STKT_UPDATE ||
1175 msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001176 if (msg_len < sizeof(update)) {
1177 /* malformed message */
1178 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1179 goto switchstate;
1180 }
1181 memcpy(&update, msg_cur, sizeof(update));
1182 msg_cur += sizeof(update);
1183 st->last_get = htonl(update);
1184 }
1185 else {
1186 st->last_get++;
1187 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001188
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001189 if (msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED ||
1190 msg_head[1] == PEER_MSG_STKT_INCUPDATE_TIMED) {
1191 size_t expire_sz = sizeof expire;
1192
1193 if (msg_cur + expire_sz > msg_end) {
1194 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1195 goto switchstate;
1196 }
1197 memcpy(&expire, msg_cur, expire_sz);
1198 msg_cur += expire_sz;
1199 expire = ntohl(expire);
1200 }
1201
Emeric Brunb3971ab2015-05-12 18:49:09 +02001202 newts = stksess_new(st->table, NULL);
1203 if (!newts)
1204 goto ignore_msg;
Emeric Brun55482912018-01-22 15:10:08 +01001205
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +02001206 if (st->table->type == SMP_T_STR) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001207 unsigned int to_read, to_store;
Emeric Brun2b920a12010-09-23 18:30:22 +02001208
Emeric Brunb3971ab2015-05-12 18:49:09 +02001209 to_read = intdecode(&msg_cur, msg_end);
1210 if (!msg_cur) {
1211 /* malformed message */
1212 stksess_free(st->table, newts);
1213 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1214 goto switchstate;
1215 }
1216 to_store = MIN(to_read, st->table->key_size - 1);
1217 if (msg_cur + to_store > msg_end) {
1218 /* malformed message */
1219 stksess_free(st->table, newts);
1220 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1221 goto switchstate;
1222 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001223
Emeric Brunb3971ab2015-05-12 18:49:09 +02001224 memcpy(newts->key.key, msg_cur, to_store);
1225 newts->key.key[to_store] = 0;
1226 msg_cur += to_read;
1227 }
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +02001228 else if (st->table->type == SMP_T_SINT) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001229 unsigned int netinteger;
Willy Tarreau72d6c162013-04-11 16:14:13 +02001230
Emeric Brunb3971ab2015-05-12 18:49:09 +02001231 if (msg_cur + sizeof(netinteger) > msg_end) {
1232 /* malformed message */
1233 stksess_free(st->table, newts);
1234 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1235 goto switchstate;
1236 }
1237 memcpy(&netinteger, msg_cur, sizeof(netinteger));
1238 netinteger = ntohl(netinteger);
1239 memcpy(newts->key.key, &netinteger, sizeof(netinteger));
1240 msg_cur += sizeof(netinteger);
1241 }
1242 else {
1243 if (msg_cur + st->table->key_size > msg_end) {
1244 /* malformed message */
1245 stksess_free(st->table, newts);
1246 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1247 goto switchstate;
1248 }
1249 memcpy(newts->key.key, msg_cur, st->table->key_size);
1250 msg_cur += st->table->key_size;
1251 }
1252
1253 /* lookup for existing entry */
Emeric Brun819fc6f2017-06-13 19:37:32 +02001254 ts = stktable_set_entry(st->table, newts);
1255 if (ts != newts) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001256 stksess_free(st->table, newts);
1257 newts = NULL;
1258 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001259
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001260 HA_RWLOCK_WRLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001261
Emeric Brun94900952015-06-11 18:25:54 +02001262 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001263
Emeric Brun94900952015-06-11 18:25:54 +02001264 if ((1 << data_type) & st->remote_data) {
1265 switch (stktable_data_types[data_type].std_type) {
1266 case STD_T_SINT: {
1267 int data;
1268
1269 data = intdecode(&msg_cur, msg_end);
1270 if (!msg_cur) {
1271 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001272 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001273 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001274 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1275 goto switchstate;
1276 }
1277
1278 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1279 if (data_ptr)
1280 stktable_data_cast(data_ptr, std_t_sint) = data;
1281 break;
1282 }
1283 case STD_T_UINT: {
1284 unsigned int data;
1285
1286 data = intdecode(&msg_cur, msg_end);
1287 if (!msg_cur) {
1288 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001289 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001290 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001291 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1292 goto switchstate;
1293 }
1294
1295 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1296 if (data_ptr)
1297 stktable_data_cast(data_ptr, std_t_uint) = data;
1298 break;
1299 }
1300 case STD_T_ULL: {
1301 unsigned long long data;
1302
1303 data = intdecode(&msg_cur, msg_end);
1304 if (!msg_cur) {
1305 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001306 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001307 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001308 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1309 goto switchstate;
1310 }
1311
1312 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1313 if (data_ptr)
1314 stktable_data_cast(data_ptr, std_t_ull) = data;
1315 break;
1316 }
1317 case STD_T_FRQP: {
1318 struct freq_ctr_period data;
1319
Emeric Brunf2fc1fd2017-11-02 17:32:43 +01001320 /* First bit is reserved for the freq_ctr_period lock
1321 Note: here we're still protected by the stksess lock
1322 so we don't need to update the update the freq_ctr_period
1323 using its internal lock */
1324
1325 data.curr_tick = tick_add(now_ms, -intdecode(&msg_cur, msg_end)) & ~0x1;
Emeric Brun94900952015-06-11 18:25:54 +02001326 if (!msg_cur) {
1327 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001328 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001329 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001330 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1331 goto switchstate;
1332 }
1333 data.curr_ctr = intdecode(&msg_cur, msg_end);
1334 if (!msg_cur) {
1335 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001336 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001337 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001338 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1339 goto switchstate;
1340 }
1341 data.prev_ctr = intdecode(&msg_cur, msg_end);
1342 if (!msg_cur) {
1343 /* malformed message */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001344 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001345 stktable_touch_remote(st->table, ts, 1);
Emeric Brun94900952015-06-11 18:25:54 +02001346 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1347 goto switchstate;
1348 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001349
Emeric Brun94900952015-06-11 18:25:54 +02001350 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1351 if (data_ptr)
1352 stktable_data_cast(data_ptr, std_t_frqp) = data;
1353 break;
1354 }
1355 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001356 }
1357 }
Emeric Brun819fc6f2017-06-13 19:37:32 +02001358
Emeric Brun55482912018-01-22 15:10:08 +01001359 /* Force new expiration */
1360 ts->expire = tick_add(now_ms, expire);
1361
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001362 HA_RWLOCK_WRUNLOCK(STK_SESS_LOCK, &ts->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001363 stktable_touch_remote(st->table, ts, 1);
1364
Emeric Brunb3971ab2015-05-12 18:49:09 +02001365 }
1366 else if (msg_head[1] == PEER_MSG_STKT_ACK) {
1367 /* ack message */
1368 uint32_t table_id ;
1369 uint32_t update;
1370 struct shared_table *st;
1371
1372 table_id = intdecode(&msg_cur, msg_end);
1373 if (!msg_cur || (msg_cur + sizeof(update) > msg_end)) {
1374 /* malformed message */
1375 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1376 goto switchstate;
1377 }
1378 memcpy(&update, msg_cur, sizeof(update));
1379 update = ntohl(update);
Emeric Brun2b920a12010-09-23 18:30:22 +02001380
Emeric Brunb3971ab2015-05-12 18:49:09 +02001381 for (st = curpeer->tables; st; st = st->next) {
1382 if (st->local_id == table_id) {
1383 st->update = update;
1384 break;
1385 }
1386 }
1387 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001388 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001389 else if (msg_head[0] == PEER_MSG_CLASS_RESERVED) {
1390 appctx->st0 = PEER_SESS_ST_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +02001391 goto switchstate;
1392 }
1393
Emeric Brunb3971ab2015-05-12 18:49:09 +02001394ignore_msg:
Emeric Brun2b920a12010-09-23 18:30:22 +02001395 /* skip consumed message */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001396 co_skip(si_oc(si), totl);
Emeric Brun2b920a12010-09-23 18:30:22 +02001397 /* loop on that state to peek next message */
Willy Tarreau72d6c162013-04-11 16:14:13 +02001398 goto switchstate;
1399
Emeric Brun2b920a12010-09-23 18:30:22 +02001400incomplete:
Willy Tarreau06d80a92017-10-19 14:32:15 +02001401 /* we get here when a co_getblk() returns <= 0 in reql */
Willy Tarreau9d9179b2013-04-11 16:56:44 +02001402
Willy Tarreau72d6c162013-04-11 16:14:13 +02001403 if (reql < 0) {
1404 /* there was an error */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001405 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau72d6c162013-04-11 16:14:13 +02001406 goto switchstate;
1407 }
1408
Emeric Brun2b920a12010-09-23 18:30:22 +02001409
Emeric Brun2b920a12010-09-23 18:30:22 +02001410
Emeric Brunb3971ab2015-05-12 18:49:09 +02001411
Emeric Brun2b920a12010-09-23 18:30:22 +02001412 /* Need to request a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001413 if ((curpeer->flags & PEER_F_LEARN_ASSIGN) &&
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001414 (curpeers->flags & PEERS_F_RESYNC_ASSIGN) &&
1415 !(curpeers->flags & PEERS_F_RESYNC_PROCESS)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001416 unsigned char msg[2];
Emeric Brun2b920a12010-09-23 18:30:22 +02001417
Emeric Brunb3971ab2015-05-12 18:49:09 +02001418 /* Current peer was elected to request a resync */
1419 msg[0] = PEER_MSG_CLASS_CONTROL;
1420 msg[1] = PEER_MSG_CTRL_RESYNCREQ;
Emeric Brun2b920a12010-09-23 18:30:22 +02001421
Emeric Brunb3971ab2015-05-12 18:49:09 +02001422 /* message to buffer */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001423 repl = ci_putblk(si_ic(si), (char *)msg, sizeof(msg));
Emeric Brunb3971ab2015-05-12 18:49:09 +02001424 if (repl <= 0) {
1425 /* no more write possible */
1426 if (repl == -1)
1427 goto full;
1428 appctx->st0 = PEER_SESS_ST_END;
1429 goto switchstate;
1430 }
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001431 curpeers->flags |= PEERS_F_RESYNC_PROCESS;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001432 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001433
Emeric Brunb3971ab2015-05-12 18:49:09 +02001434 /* Nothing to read, now we start to write */
Emeric Brun2b920a12010-09-23 18:30:22 +02001435
Emeric Brunb3971ab2015-05-12 18:49:09 +02001436 if (curpeer->tables) {
1437 struct shared_table *st;
1438 struct shared_table *last_local_table;
Emeric Brun2b920a12010-09-23 18:30:22 +02001439
Emeric Brunb3971ab2015-05-12 18:49:09 +02001440 last_local_table = curpeer->last_local_table;
1441 if (!last_local_table)
1442 last_local_table = curpeer->tables;
1443 st = last_local_table->next;
Emeric Brun2b920a12010-09-23 18:30:22 +02001444
Emeric Brunb3971ab2015-05-12 18:49:09 +02001445 while (1) {
1446 if (!st)
1447 st = curpeer->tables;
Emeric Brun2b920a12010-09-23 18:30:22 +02001448
Emeric Brunb3971ab2015-05-12 18:49:09 +02001449 /* It remains some updates to ack */
1450 if (st->last_get != st->last_acked) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001451 int msglen;
Emeric Brun2b920a12010-09-23 18:30:22 +02001452
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001453 msglen = peer_prepare_ackmsg(st,
1454 trash.area,
1455 trash.size);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001456 if (!msglen) {
1457 /* internal error: message does not fit in trash */
1458 appctx->st0 = PEER_SESS_ST_END;
1459 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001460 }
1461
Emeric Brunb3971ab2015-05-12 18:49:09 +02001462 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001463 repl = ci_putblk(si_ic(si),
1464 trash.area,
1465 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001466 if (repl <= 0) {
1467 /* no more write possible */
1468 if (repl == -1) {
1469 goto full;
Emeric Brun2b920a12010-09-23 18:30:22 +02001470 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001471 appctx->st0 = PEER_SESS_ST_END;
1472 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001473 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001474 st->last_acked = st->last_get;
Emeric Brun2b920a12010-09-23 18:30:22 +02001475 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001476
Emeric Brunb3971ab2015-05-12 18:49:09 +02001477 if (!(curpeer->flags & PEER_F_TEACH_PROCESS)) {
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001478 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001479 if (!(curpeer->flags & PEER_F_LEARN_ASSIGN) &&
1480 ((int)(st->last_pushed - st->table->localupdate) < 0)) {
1481 struct eb32_node *eb;
1482 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001483
Emeric Brunb3971ab2015-05-12 18:49:09 +02001484 if (st != curpeer->last_local_table) {
1485 int msglen;
Emeric Brun2b920a12010-09-23 18:30:22 +02001486
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001487 msglen = peer_prepare_switchmsg(st,
1488 trash.area,
1489 trash.size);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001490 if (!msglen) {
Emeric Brun088c9b72017-12-01 11:37:36 +01001491 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001492 /* internal error: message does not fit in trash */
1493 appctx->st0 = PEER_SESS_ST_END;
1494 goto switchstate;
1495 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001496
Emeric Brunb3971ab2015-05-12 18:49:09 +02001497 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001498 repl = ci_putblk(si_ic(si),
1499 trash.area,
1500 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001501 if (repl <= 0) {
Emeric Brun088c9b72017-12-01 11:37:36 +01001502 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001503 /* no more write possible */
1504 if (repl == -1) {
1505 goto full;
1506 }
1507 appctx->st0 = PEER_SESS_ST_END;
1508 goto switchstate;
1509 }
1510 curpeer->last_local_table = st;
Emeric Brun2b920a12010-09-23 18:30:22 +02001511 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001512
1513 /* We force new pushed to 1 to force identifier in update message */
1514 new_pushed = 1;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001515 while (1) {
1516 uint32_t msglen;
1517 struct stksess *ts;
Emeric Brun819fc6f2017-06-13 19:37:32 +02001518 unsigned updateid;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001519
1520 /* push local updates */
Emeric Brun819fc6f2017-06-13 19:37:32 +02001521 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001522 if (!eb) {
1523 eb = eb32_first(&st->table->updates);
1524 if (!eb || ((int)(eb->key - st->last_pushed) <= 0)) {
Emeric Brunaaf58602015-06-15 17:23:30 +02001525 st->table->commitupdate = st->last_pushed = st->table->localupdate;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001526 break;
1527 }
1528 }
1529
1530 if ((int)(eb->key - st->table->localupdate) > 0) {
Emeric Brunaaf58602015-06-15 17:23:30 +02001531 st->table->commitupdate = st->last_pushed = st->table->localupdate;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001532 break;
1533 }
1534
1535 ts = eb32_entry(eb, struct stksess, upd);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001536 updateid = ts->upd.key;
1537 ts->ref_cnt++;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001538 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001539
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001540 msglen = peer_prepare_updatemsg(ts, st, updateid,
1541 trash.area,
1542 trash.size,
1543 new_pushed,
1544 0);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001545 if (!msglen) {
1546 /* internal error: message does not fit in trash */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001547 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001548 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001549 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001550 appctx->st0 = PEER_SESS_ST_END;
1551 goto switchstate;
1552 }
1553
1554 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001555 repl = ci_putblk(si_ic(si),
1556 trash.area,
1557 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001558 if (repl <= 0) {
1559 /* no more write possible */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001560 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001561 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001562 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001563 if (repl == -1) {
1564 goto full;
1565 }
1566 appctx->st0 = PEER_SESS_ST_END;
1567 goto switchstate;
1568 }
Emeric Brun819fc6f2017-06-13 19:37:32 +02001569
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001570 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001571 ts->ref_cnt--;
1572 st->last_pushed = updateid;
Emeric Brunaaf58602015-06-15 17:23:30 +02001573 if ((int)(st->last_pushed - st->table->commitupdate) > 0)
1574 st->table->commitupdate = st->last_pushed;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001575 /* identifier may not needed in next update message */
1576 new_pushed = 0;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001577 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001578 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001579 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun2b920a12010-09-23 18:30:22 +02001580 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001581 else {
1582 if (!(st->flags & SHTABLE_F_TEACH_STAGE1)) {
1583 struct eb32_node *eb;
1584 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001585
Emeric Brunb3971ab2015-05-12 18:49:09 +02001586 if (st != curpeer->last_local_table) {
1587 int msglen;
1588
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001589 msglen = peer_prepare_switchmsg(st,
1590 trash.area,
1591 trash.size);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001592 if (!msglen) {
1593 /* internal error: message does not fit in trash */
1594 appctx->st0 = PEER_SESS_ST_END;
1595 goto switchstate;
1596 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001597
Emeric Brunb3971ab2015-05-12 18:49:09 +02001598 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001599 repl = ci_putblk(si_ic(si),
1600 trash.area,
1601 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001602 if (repl <= 0) {
1603 /* no more write possible */
1604 if (repl == -1) {
1605 goto full;
1606 }
1607 appctx->st0 = PEER_SESS_ST_END;
1608 goto switchstate;
1609 }
1610 curpeer->last_local_table = st;
1611 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001612
Emeric Brunb3971ab2015-05-12 18:49:09 +02001613 /* We force new pushed to 1 to force identifier in update message */
1614 new_pushed = 1;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001615 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001616 while (1) {
1617 uint32_t msglen;
1618 struct stksess *ts;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001619 int use_timed;
Emeric Brun819fc6f2017-06-13 19:37:32 +02001620 unsigned updateid;
Emeric Brun2b920a12010-09-23 18:30:22 +02001621
Emeric Brunb3971ab2015-05-12 18:49:09 +02001622 /* push local updates */
Emeric Brun819fc6f2017-06-13 19:37:32 +02001623 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001624 if (!eb) {
1625 st->flags |= SHTABLE_F_TEACH_STAGE1;
1626 eb = eb32_first(&st->table->updates);
1627 if (eb)
1628 st->last_pushed = eb->key - 1;
1629 break;
1630 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001631
Emeric Brunb3971ab2015-05-12 18:49:09 +02001632 ts = eb32_entry(eb, struct stksess, upd);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001633 updateid = ts->upd.key;
1634 ts->ref_cnt++;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001635 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001636
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001637 use_timed = !(curpeer->flags & PEER_F_DWNGRD);
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001638 msglen = peer_prepare_updatemsg(ts, st, updateid,
1639 trash.area,
1640 trash.size,
1641 new_pushed,
1642 use_timed);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001643 if (!msglen) {
1644 /* internal error: message does not fit in trash */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001645 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001646 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001647 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001648 appctx->st0 = PEER_SESS_ST_END;
1649 goto switchstate;
1650 }
1651
1652 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001653 repl = ci_putblk(si_ic(si),
1654 trash.area,
1655 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001656 if (repl <= 0) {
1657 /* no more write possible */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001658 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001659 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001660 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001661 if (repl == -1) {
1662 goto full;
1663 }
1664 appctx->st0 = PEER_SESS_ST_END;
1665 goto switchstate;
1666 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001667 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001668 ts->ref_cnt--;
1669 st->last_pushed = updateid;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001670 /* identifier may not needed in next update message */
1671 new_pushed = 0;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001672 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001673 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun2b920a12010-09-23 18:30:22 +02001674 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001675
Emeric Brunb3971ab2015-05-12 18:49:09 +02001676 if (!(st->flags & SHTABLE_F_TEACH_STAGE2)) {
1677 struct eb32_node *eb;
1678 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001679
Emeric Brunb3971ab2015-05-12 18:49:09 +02001680 if (st != curpeer->last_local_table) {
1681 int msglen;
1682
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001683 msglen = peer_prepare_switchmsg(st,
1684 trash.area,
1685 trash.size);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001686 if (!msglen) {
1687 /* internal error: message does not fit in trash */
1688 appctx->st0 = PEER_SESS_ST_END;
1689 goto switchstate;
1690 }
1691
1692 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001693 repl = ci_putblk(si_ic(si),
1694 trash.area,
1695 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001696 if (repl <= 0) {
1697 /* no more write possible */
1698 if (repl == -1) {
1699 goto full;
1700 }
1701 appctx->st0 = PEER_SESS_ST_END;
1702 goto switchstate;
1703 }
1704 curpeer->last_local_table = st;
1705 }
1706
1707 /* We force new pushed to 1 to force identifier in update message */
1708 new_pushed = 1;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001709 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001710 while (1) {
1711 uint32_t msglen;
1712 struct stksess *ts;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001713 int use_timed;
Emeric Brun819fc6f2017-06-13 19:37:32 +02001714 unsigned updateid;
1715
1716 /* push local updates */
1717 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001718
1719 /* push local updates */
1720 if (!eb || eb->key > st->teaching_origin) {
1721 st->flags |= SHTABLE_F_TEACH_STAGE2;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001722 break;
1723 }
1724
1725 ts = eb32_entry(eb, struct stksess, upd);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001726 updateid = ts->upd.key;
1727 ts->ref_cnt++;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001728 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001729
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001730 use_timed = !(curpeer->flags & PEER_F_DWNGRD);
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001731 msglen = peer_prepare_updatemsg(ts, st, updateid,
1732 trash.area,
1733 trash.size,
1734 new_pushed,
1735 use_timed);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001736 if (!msglen) {
1737 /* internal error: message does not fit in trash */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001738 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001739 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001740 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001741 appctx->st0 = PEER_SESS_ST_END;
1742 goto switchstate;
1743 }
1744
1745 /* message to buffer */
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001746 repl = ci_putblk(si_ic(si),
1747 trash.area,
1748 msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001749 if (repl <= 0) {
1750 /* no more write possible */
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001751 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001752 ts->ref_cnt--;
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001753 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001754 if (repl == -1) {
1755 goto full;
1756 }
1757 appctx->st0 = PEER_SESS_ST_END;
1758 goto switchstate;
1759 }
Emeric Brun819fc6f2017-06-13 19:37:32 +02001760
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001761 HA_SPIN_LOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun819fc6f2017-06-13 19:37:32 +02001762 ts->ref_cnt--;
1763 st->last_pushed = updateid;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001764 /* identifier may not needed in next update message */
1765 new_pushed = 0;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001766 }
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001767 HA_SPIN_UNLOCK(STK_TABLE_LOCK, &st->table->lock);
Emeric Brun2b920a12010-09-23 18:30:22 +02001768 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001769 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001770
1771 if (st == last_local_table)
1772 break;
1773 st = st->next;
Emeric Brun2b920a12010-09-23 18:30:22 +02001774 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001775 }
1776
1777
1778 if ((curpeer->flags & PEER_F_TEACH_PROCESS) && !(curpeer->flags & PEER_F_TEACH_FINISHED)) {
1779 unsigned char msg[2];
1780
1781 /* Current peer was elected to request a resync */
1782 msg[0] = PEER_MSG_CLASS_CONTROL;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001783 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 +02001784 /* process final lesson message */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001785 repl = ci_putblk(si_ic(si), (char *)msg, sizeof(msg));
Emeric Brunb3971ab2015-05-12 18:49:09 +02001786 if (repl <= 0) {
1787 /* no more write possible */
1788 if (repl == -1)
1789 goto full;
1790 appctx->st0 = PEER_SESS_ST_END;
1791 goto switchstate;
1792 }
1793 /* flag finished message sent */
1794 curpeer->flags |= PEER_F_TEACH_FINISHED;
1795 }
1796
Emeric Brun597b26e2016-08-12 11:23:31 +02001797 /* Confirm finished or partial messages */
1798 while (curpeer->confirm) {
1799 unsigned char msg[2];
1800
1801 /* There is a confirm messages to send */
1802 msg[0] = PEER_MSG_CLASS_CONTROL;
1803 msg[1] = PEER_MSG_CTRL_RESYNCCONFIRM;
1804
1805 /* message to buffer */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001806 repl = ci_putblk(si_ic(si), (char *)msg, sizeof(msg));
Emeric Brun597b26e2016-08-12 11:23:31 +02001807 if (repl <= 0) {
1808 /* no more write possible */
1809 if (repl == -1)
1810 goto full;
1811 appctx->st0 = PEER_SESS_ST_END;
1812 goto switchstate;
1813 }
1814 curpeer->confirm--;
1815 }
1816
Emeric Brun2b920a12010-09-23 18:30:22 +02001817 /* noting more to do */
1818 goto out;
1819 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001820 case PEER_SESS_ST_EXIT:
Willy Tarreau843b7cb2018-07-13 10:54:26 +02001821 repl = snprintf(trash.area, trash.size,
1822 "%d\n", appctx->st1);
1823 if (ci_putblk(si_ic(si), trash.area, repl) == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +01001824 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001825 appctx->st0 = PEER_SESS_ST_END;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001826 goto switchstate;
1827 case PEER_SESS_ST_ERRSIZE: {
1828 unsigned char msg[2];
1829
1830 msg[0] = PEER_MSG_CLASS_ERROR;
1831 msg[1] = PEER_MSG_ERR_SIZELIMIT;
1832
Willy Tarreau06d80a92017-10-19 14:32:15 +02001833 if (ci_putblk(si_ic(si), (char *)msg, sizeof(msg)) == -1)
Emeric Brunb3971ab2015-05-12 18:49:09 +02001834 goto full;
1835 appctx->st0 = PEER_SESS_ST_END;
1836 goto switchstate;
1837 }
1838 case PEER_SESS_ST_ERRPROTO: {
1839 unsigned char msg[2];
1840
1841 msg[0] = PEER_MSG_CLASS_ERROR;
1842 msg[1] = PEER_MSG_ERR_PROTOCOL;
1843
Willy Tarreau06d80a92017-10-19 14:32:15 +02001844 if (ci_putblk(si_ic(si), (char *)msg, sizeof(msg)) == -1)
Emeric Brunb3971ab2015-05-12 18:49:09 +02001845 goto full;
1846 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02001847 /* fall through */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001848 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001849 case PEER_SESS_ST_END: {
Emeric Brun80527f52017-06-19 17:46:37 +02001850 if (curpeer) {
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001851 HA_SPIN_UNLOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02001852 curpeer = NULL;
1853 }
Willy Tarreau73b013b2012-05-21 16:31:45 +02001854 si_shutw(si);
1855 si_shutr(si);
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001856 si_ic(si)->flags |= CF_READ_NULL;
Willy Tarreau828824a2015-04-19 17:20:03 +02001857 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001858 }
1859 }
1860 }
1861out:
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001862 si_oc(si)->flags |= CF_READ_DONTWAIT;
Emeric Brun80527f52017-06-19 17:46:37 +02001863
1864 if (curpeer)
Christopher Faulet2a944ee2017-11-07 10:42:54 +01001865 HA_SPIN_UNLOCK(PEER_LOCK, &curpeer->lock);
Emeric Brun2b920a12010-09-23 18:30:22 +02001866 return;
Willy Tarreaubc18da12015-03-13 14:00:47 +01001867full:
Willy Tarreaufe127932015-04-21 19:23:39 +02001868 si_applet_cant_put(si);
Willy Tarreaubc18da12015-03-13 14:00:47 +01001869 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001870}
1871
Willy Tarreau30576452015-04-13 13:50:30 +02001872static struct applet peer_applet = {
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001873 .obj_type = OBJ_TYPE_APPLET,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001874 .name = "<PEER>", /* used for logging */
1875 .fct = peer_io_handler,
Aman Gupta9a13e842012-04-02 18:57:53 -07001876 .release = peer_session_release,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001877};
Emeric Brun2b920a12010-09-23 18:30:22 +02001878
1879/*
1880 * Use this function to force a close of a peer session
1881 */
Willy Tarreau81bc3b02016-10-31 17:37:39 +01001882static void peer_session_forceshutdown(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +02001883{
Frédéric Lécaille5df11902017-06-13 16:39:57 +02001884 /* Note that the peer sessions which have just been created
1885 * (->st0 == PEER_SESS_ST_CONNECT) must not
1886 * be shutdown, if not, the TCP session will never be closed
1887 * and stay in CLOSE_WAIT state after having been closed by
1888 * the remote side.
1889 */
1890 if (!appctx || appctx->st0 == PEER_SESS_ST_CONNECT)
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001891 return;
1892
Willy Tarreau81bc3b02016-10-31 17:37:39 +01001893 if (appctx->applet != &peer_applet)
1894 return;
1895
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001896 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau78c0c502016-10-31 17:32:20 +01001897 appctx_wakeup(appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02001898}
1899
Willy Tarreau91d96282015-03-13 15:47:26 +01001900/* Pre-configures a peers frontend to accept incoming connections */
1901void peers_setup_frontend(struct proxy *fe)
1902{
1903 fe->last_change = now.tv_sec;
1904 fe->cap = PR_CAP_FE;
1905 fe->maxconn = 0;
1906 fe->conn_retries = CONN_RETRIES;
1907 fe->timeout.client = MS_TO_TICKS(5000);
Willy Tarreaud1d48d42015-03-13 16:15:46 +01001908 fe->accept = frontend_accept;
Willy Tarreauf87ab942015-03-13 15:55:16 +01001909 fe->default_target = &peer_applet.obj_type;
Willy Tarreau91d96282015-03-13 15:47:26 +01001910 fe->options2 |= PR_O2_INDEPSTR | PR_O2_SMARTCON | PR_O2_SMARTACC;
Willy Tarreau0fca4832015-05-01 19:12:05 +02001911 fe->bind_proc = 0; /* will be filled by users */
Willy Tarreau91d96282015-03-13 15:47:26 +01001912}
1913
Emeric Brun2b920a12010-09-23 18:30:22 +02001914/*
Willy Tarreaubd55e312010-11-11 10:55:09 +01001915 * Create a new peer session in assigned state (connect will start automatically)
Emeric Brun2b920a12010-09-23 18:30:22 +02001916 */
Willy Tarreau9df94c22016-10-31 18:42:52 +01001917static struct appctx *peer_session_create(struct peers *peers, struct peer *peer)
Emeric Brun2b920a12010-09-23 18:30:22 +02001918{
Willy Tarreau04b92862017-09-15 11:01:04 +02001919 struct proxy *p = peers->peers_fe; /* attached frontend */
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001920 struct appctx *appctx;
Willy Tarreau15b5e142015-04-04 14:38:25 +02001921 struct session *sess;
Willy Tarreau87b09662015-04-03 00:22:06 +02001922 struct stream *s;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001923 struct connection *conn;
Olivier Houchard9aaf7782017-09-13 18:30:23 +02001924 struct conn_stream *cs;
Emeric Brun2b920a12010-09-23 18:30:22 +02001925
Emeric Brunb3971ab2015-05-12 18:49:09 +02001926 peer->reconnect = tick_add(now_ms, MS_TO_TICKS(5000));
1927 peer->statuscode = PEER_SESS_SC_CONNECTCODE;
Willy Tarreaud990baf2015-04-05 00:32:03 +02001928 s = NULL;
1929
Emeric Brun1138fd02017-06-19 12:38:55 +02001930 appctx = appctx_new(&peer_applet, tid_bit);
Willy Tarreaud990baf2015-04-05 00:32:03 +02001931 if (!appctx)
1932 goto out_close;
1933
1934 appctx->st0 = PEER_SESS_ST_CONNECT;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001935 appctx->ctx.peers.ptr = (void *)peer;
Willy Tarreaud990baf2015-04-05 00:32:03 +02001936
Willy Tarreau04b92862017-09-15 11:01:04 +02001937 sess = session_new(p, NULL, &appctx->obj_type);
Willy Tarreau15b5e142015-04-04 14:38:25 +02001938 if (!sess) {
Christopher Faulet767a84b2017-11-24 16:50:31 +01001939 ha_alert("out of memory in peer_session_create().\n");
Willy Tarreaud990baf2015-04-05 00:32:03 +02001940 goto out_free_appctx;
Emeric Brun2b920a12010-09-23 18:30:22 +02001941 }
1942
Willy Tarreau87787ac2017-08-28 16:22:54 +02001943 if ((s = stream_new(sess, &appctx->obj_type)) == NULL) {
Christopher Faulet767a84b2017-11-24 16:50:31 +01001944 ha_alert("Failed to initialize stream in peer_session_create().\n");
Willy Tarreau87787ac2017-08-28 16:22:54 +02001945 goto out_free_sess;
Willy Tarreau8baf9062015-04-05 00:46:36 +02001946 }
1947
Willy Tarreau342bfb12015-04-05 01:35:34 +02001948 /* The tasks below are normally what is supposed to be done by
1949 * fe->accept().
1950 */
Willy Tarreaue7dff022015-04-03 01:14:29 +02001951 s->flags = SF_ASSIGNED|SF_ADDR_SET;
Emeric Brun2b920a12010-09-23 18:30:22 +02001952
Willy Tarreau6e2979c2015-04-27 13:21:15 +02001953 /* applet is waiting for data */
1954 si_applet_cant_get(&s->si[0]);
1955 appctx_wakeup(appctx);
1956
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001957 /* initiate an outgoing connection */
Willy Tarreaudbd02672017-12-06 17:39:53 +01001958 s->si[1].flags |= SI_FL_NOLINGER;
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001959 si_set_state(&s->si[1], SI_ST_ASS);
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001960
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001961 /* automatically prepare the stream interface to connect to the
Willy Tarreaub363a1f2013-10-01 10:45:07 +02001962 * pre-initialized connection in si->conn.
1963 */
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001964 if (unlikely((conn = conn_new()) == NULL))
Willy Tarreau8baf9062015-04-05 00:46:36 +02001965 goto out_free_strm;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001966
Olivier Houchard9aaf7782017-09-13 18:30:23 +02001967 if (unlikely((cs = cs_new(conn)) == NULL))
1968 goto out_free_conn;
1969
Willy Tarreaube373152018-09-06 11:45:30 +02001970 conn->target = s->target = &s->be->obj_type;
1971 memcpy(&conn->addr.to, &peer->addr, sizeof(conn->addr.to));
1972
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001973 conn_prepare(conn, peer->proto, peer->xprt);
Willy Tarreau175a2bb2018-09-12 12:02:05 +02001974 conn_install_mux(conn, &mux_pt_ops, cs, s->be);
Olivier Houchard9aaf7782017-09-13 18:30:23 +02001975 si_attach_cs(&s->si[1], cs);
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001976
Emeric Brun2b920a12010-09-23 18:30:22 +02001977 s->do_log = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001978 s->uniq_id = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +02001979
Willy Tarreau22ec1ea2014-11-27 20:45:39 +01001980 s->res.flags |= CF_READ_DONTWAIT;
Willy Tarreau696a2912014-11-24 11:36:57 +01001981
Emeric Brunb3971ab2015-05-12 18:49:09 +02001982 peer->appctx = appctx;
Willy Tarreau87787ac2017-08-28 16:22:54 +02001983 task_wakeup(s->task, TASK_WOKEN_INIT);
Willy Tarreau9df94c22016-10-31 18:42:52 +01001984 return appctx;
Emeric Brun2b920a12010-09-23 18:30:22 +02001985
1986 /* Error unrolling */
Olivier Houchard9aaf7782017-09-13 18:30:23 +02001987 out_free_conn:
1988 conn_free(conn);
Willy Tarreau15b5e142015-04-04 14:38:25 +02001989 out_free_strm:
Emeric Brun2b920a12010-09-23 18:30:22 +02001990 LIST_DEL(&s->list);
Willy Tarreaubafbe012017-11-24 17:34:44 +01001991 pool_free(pool_head_stream, s);
Willy Tarreau15b5e142015-04-04 14:38:25 +02001992 out_free_sess:
Willy Tarreau11c36242015-04-04 15:54:03 +02001993 session_free(sess);
Willy Tarreaud990baf2015-04-05 00:32:03 +02001994 out_free_appctx:
1995 appctx_free(appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02001996 out_close:
Willy Tarreaub21d08e2016-10-31 17:46:57 +01001997 return NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001998}
1999
2000/*
2001 * Task processing function to manage re-connect and peer session
2002 * tasks wakeup on local update.
2003 */
Olivier Houchard9f6af332018-05-25 14:04:04 +02002004static struct task *process_peer_sync(struct task * task, void *context, unsigned short state)
Emeric Brun2b920a12010-09-23 18:30:22 +02002005{
Olivier Houchard9f6af332018-05-25 14:04:04 +02002006 struct peers *peers = context;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002007 struct peer *ps;
2008 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +02002009
2010 task->expire = TICK_ETERNITY;
2011
Emeric Brunb3971ab2015-05-12 18:49:09 +02002012 if (!peers->peers_fe) {
Willy Tarreau46dc1ca2015-05-01 18:32:13 +02002013 /* this one was never started, kill it */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002014 signal_unregister_handler(peers->sighandler);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002015 task_delete(peers->sync_task);
2016 task_free(peers->sync_task);
Willy Tarreau37bb7be2015-09-21 15:24:58 +02002017 peers->sync_task = NULL;
Willy Tarreau46dc1ca2015-05-01 18:32:13 +02002018 return NULL;
2019 }
2020
Emeric Brun80527f52017-06-19 17:46:37 +02002021 /* Acquire lock for all peers of the section */
2022 for (ps = peers->remote; ps; ps = ps->next)
Christopher Faulet2a944ee2017-11-07 10:42:54 +01002023 HA_SPIN_LOCK(PEER_LOCK, &ps->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02002024
Emeric Brun2b920a12010-09-23 18:30:22 +02002025 if (!stopping) {
2026 /* Normal case (not soft stop)*/
Emeric Brunb3971ab2015-05-12 18:49:09 +02002027
2028 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL) &&
2029 (!nb_oldpids || tick_is_expired(peers->resync_timeout, now_ms)) &&
2030 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002031 /* Resync from local peer needed
2032 no peer was assigned for the lesson
2033 and no old local peer found
2034 or resync timeout expire */
2035
2036 /* flag no more resync from local, to try resync from remotes */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002037 peers->flags |= PEERS_F_RESYNC_LOCAL;
Emeric Brun2b920a12010-09-23 18:30:22 +02002038
2039 /* reschedule a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002040 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
Emeric Brun2b920a12010-09-23 18:30:22 +02002041 }
2042
2043 /* For each session */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002044 for (ps = peers->remote; ps; ps = ps->next) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002045 /* For each remote peers */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002046 if (!ps->local) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002047 if (!ps->appctx) {
2048 /* no active peer connection */
Emeric Brun2b920a12010-09-23 18:30:22 +02002049 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002050 ((ps->statuscode == PEER_SESS_SC_CONNECTCODE ||
Willy Tarreaub4e34da2015-05-20 10:39:04 +02002051 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002052 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02002053 tick_is_expired(ps->reconnect, now_ms))) {
2054 /* connection never tried
Willy Tarreau9df94c22016-10-31 18:42:52 +01002055 * or previous peer connection established with success
2056 * or previous peer connection failed while connecting
Emeric Brun2b920a12010-09-23 18:30:22 +02002057 * and reconnection timer is expired */
2058
2059 /* retry a connect */
Willy Tarreau9df94c22016-10-31 18:42:52 +01002060 ps->appctx = peer_session_create(peers, ps);
Emeric Brun2b920a12010-09-23 18:30:22 +02002061 }
Willy Tarreaub4e34da2015-05-20 10:39:04 +02002062 else if (!tick_is_expired(ps->reconnect, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002063 /* If previous session failed during connection
2064 * but reconnection timer is not expired */
2065
2066 /* reschedule task for reconnect */
2067 task->expire = tick_first(task->expire, ps->reconnect);
2068 }
2069 /* else do nothing */
Willy Tarreau9df94c22016-10-31 18:42:52 +01002070 } /* !ps->appctx */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002071 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002072 /* current peer connection is active and established */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002073 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
2074 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02002075 !(ps->flags & PEER_F_LEARN_NOTUP2DATE)) {
2076 /* Resync from a remote is needed
2077 * and no peer was assigned for lesson
2078 * and current peer may be up2date */
2079
2080 /* assign peer for the lesson */
2081 ps->flags |= PEER_F_LEARN_ASSIGN;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002082 peers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brun2b920a12010-09-23 18:30:22 +02002083
Willy Tarreau9df94c22016-10-31 18:42:52 +01002084 /* wake up peer handler to handle a request of resync */
Willy Tarreaue5843b32015-04-27 18:40:14 +02002085 appctx_wakeup(ps->appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02002086 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02002087 else {
2088 /* Awake session if there is data to push */
2089 for (st = ps->tables; st ; st = st->next) {
2090 if ((int)(st->last_pushed - st->table->localupdate) < 0) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002091 /* wake up the peer handler to push local updates */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002092 appctx_wakeup(ps->appctx);
2093 break;
2094 }
2095 }
Emeric Brun2b920a12010-09-23 18:30:22 +02002096 }
2097 /* else do nothing */
2098 } /* SUCCESSCODE */
2099 } /* !ps->peer->local */
2100 } /* for */
2101
2102 /* Resync from remotes expired: consider resync is finished */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002103 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
2104 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
2105 tick_is_expired(peers->resync_timeout, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002106 /* Resync from remote peer needed
2107 * no peer was assigned for the lesson
2108 * and resync timeout expire */
2109
2110 /* flag no more resync from remote, consider resync is finished */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002111 peers->flags |= PEERS_F_RESYNC_REMOTE;
Emeric Brun2b920a12010-09-23 18:30:22 +02002112 }
2113
Emeric Brunb3971ab2015-05-12 18:49:09 +02002114 if ((peers->flags & PEERS_RESYNC_STATEMASK) != PEERS_RESYNC_FINISHED) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002115 /* Resync not finished*/
Frédéric Lécaille5d6e5f82017-05-29 13:47:16 +02002116 /* reschedule task to resync timeout if not expired, to ended resync if needed */
2117 if (!tick_is_expired(peers->resync_timeout, now_ms))
2118 task->expire = tick_first(task->expire, peers->resync_timeout);
Emeric Brun2b920a12010-09-23 18:30:22 +02002119 }
2120 } /* !stopping */
2121 else {
2122 /* soft stop case */
2123 if (task->state & TASK_WOKEN_SIGNAL) {
2124 /* We've just recieved the signal */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002125 if (!(peers->flags & PEERS_F_DONOTSTOP)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002126 /* add DO NOT STOP flag if not present */
Christopher Faulet8d8aa0d2017-05-30 15:36:50 +02002127 HA_ATOMIC_ADD(&jobs, 1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002128 peers->flags |= PEERS_F_DONOTSTOP;
2129 ps = peers->local;
2130 for (st = ps->tables; st ; st = st->next)
2131 st->table->syncing++;
Emeric Brun2b920a12010-09-23 18:30:22 +02002132 }
2133
2134 /* disconnect all connected peers */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002135 for (ps = peers->remote; ps; ps = ps->next) {
Emeric Brun80527f52017-06-19 17:46:37 +02002136 /* we're killing a connection, we must apply a random delay before
2137 * retrying otherwise the other end will do the same and we can loop
2138 * for a while.
2139 */
2140 ps->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + random() % 2000));
Willy Tarreau9df94c22016-10-31 18:42:52 +01002141 if (ps->appctx) {
Willy Tarreau81bc3b02016-10-31 17:37:39 +01002142 peer_session_forceshutdown(ps->appctx);
Willy Tarreaue5843b32015-04-27 18:40:14 +02002143 ps->appctx = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02002144 }
2145 }
2146 }
Emeric Brun2b920a12010-09-23 18:30:22 +02002147
Emeric Brunb3971ab2015-05-12 18:49:09 +02002148 ps = peers->local;
Emeric Brun2b920a12010-09-23 18:30:22 +02002149 if (ps->flags & PEER_F_TEACH_COMPLETE) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02002150 if (peers->flags & PEERS_F_DONOTSTOP) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002151 /* resync of new process was complete, current process can die now */
Willy Tarreaucea85372017-11-29 14:49:30 +01002152 HA_ATOMIC_SUB(&jobs, 1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002153 peers->flags &= ~PEERS_F_DONOTSTOP;
2154 for (st = ps->tables; st ; st = st->next)
2155 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02002156 }
2157 }
Willy Tarreau9df94c22016-10-31 18:42:52 +01002158 else if (!ps->appctx) {
2159 /* If there's no active peer connection */
Emeric Brun2b920a12010-09-23 18:30:22 +02002160 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002161 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
2162 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE ||
2163 ps->statuscode == PEER_SESS_SC_TRYAGAIN) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002164 /* connection never tried
Willy Tarreau9df94c22016-10-31 18:42:52 +01002165 * or previous peer connection was successfully established
2166 * or previous tcp connect succeeded but init state incomplete
Emeric Brun2b920a12010-09-23 18:30:22 +02002167 * or during previous connect, peer replies a try again statuscode */
2168
2169 /* connect to the peer */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002170 peer_session_create(peers, ps);
Emeric Brun2b920a12010-09-23 18:30:22 +02002171 }
2172 else {
2173 /* Other error cases */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002174 if (peers->flags & PEERS_F_DONOTSTOP) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002175 /* unable to resync new process, current process can die now */
Christopher Faulet8d8aa0d2017-05-30 15:36:50 +02002176 HA_ATOMIC_SUB(&jobs, 1);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002177 peers->flags &= ~PEERS_F_DONOTSTOP;
2178 for (st = ps->tables; st ; st = st->next)
2179 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02002180 }
2181 }
2182 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02002183 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE ) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002184 /* current peer connection is active and established
2185 * wake up all peer handlers to push remaining local updates */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002186 for (st = ps->tables; st ; st = st->next) {
2187 if ((int)(st->last_pushed - st->table->localupdate) < 0) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02002188 appctx_wakeup(ps->appctx);
2189 break;
2190 }
2191 }
Emeric Brun2b920a12010-09-23 18:30:22 +02002192 }
2193 } /* stopping */
Emeric Brun80527f52017-06-19 17:46:37 +02002194
2195 /* Release lock for all peers of the section */
2196 for (ps = peers->remote; ps; ps = ps->next)
Christopher Faulet2a944ee2017-11-07 10:42:54 +01002197 HA_SPIN_UNLOCK(PEER_LOCK, &ps->lock);
Emeric Brun80527f52017-06-19 17:46:37 +02002198
Emeric Brun2b920a12010-09-23 18:30:22 +02002199 /* Wakeup for re-connect */
2200 return task;
2201}
2202
Emeric Brunb3971ab2015-05-12 18:49:09 +02002203
Emeric Brun2b920a12010-09-23 18:30:22 +02002204/*
Emeric Brun2b920a12010-09-23 18:30:22 +02002205 *
2206 */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002207void peers_init_sync(struct peers *peers)
Emeric Brun2b920a12010-09-23 18:30:22 +02002208{
Emeric Brun2b920a12010-09-23 18:30:22 +02002209 struct peer * curpeer;
Willy Tarreau4348fad2012-09-20 16:48:07 +02002210 struct listener *listener;
Emeric Brun2b920a12010-09-23 18:30:22 +02002211
Emeric Brun2b920a12010-09-23 18:30:22 +02002212 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002213 peers->peers_fe->maxconn += 3;
2214 }
2215
Willy Tarreau4348fad2012-09-20 16:48:07 +02002216 list_for_each_entry(listener, &peers->peers_fe->conf.listeners, by_fe)
2217 listener->maxconn = peers->peers_fe->maxconn;
Emeric Brunc60def82017-09-27 14:59:38 +02002218 peers->sync_task = task_new(MAX_THREADS_MASK);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002219 peers->sync_task->process = process_peer_sync;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002220 peers->sync_task->context = (void *)peers;
2221 peers->sighandler = signal_register_task(0, peers->sync_task, 0);
2222 task_wakeup(peers->sync_task, TASK_WOKEN_INIT);
2223}
2224
2225
2226
2227/*
2228 * Function used to register a table for sync on a group of peers
2229 *
2230 */
2231void peers_register_table(struct peers *peers, struct stktable *table)
2232{
2233 struct shared_table *st;
2234 struct peer * curpeer;
2235 int id = 0;
2236
2237 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
Vincent Bernat02779b62016-04-03 13:48:43 +02002238 st = calloc(1,sizeof(*st));
Emeric Brunb3971ab2015-05-12 18:49:09 +02002239 st->table = table;
2240 st->next = curpeer->tables;
2241 if (curpeer->tables)
2242 id = curpeer->tables->local_id;
2243 st->local_id = id + 1;
2244
2245 curpeer->tables = st;
2246 }
2247
2248 table->sync_task = peers->sync_task;
Emeric Brun2b920a12010-09-23 18:30:22 +02002249}
2250