blob: b98ec61a8607dab60e8153b47d735b94cb80621f [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>
Emeric Brun2b920a12010-09-23 18:30:22 +020027
28#include <types/global.h>
Willy Tarreau3fdb3662012-11-12 00:42:33 +010029#include <types/listener.h>
30#include <types/obj_type.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020031#include <types/peers.h>
32
33#include <proto/acl.h>
Willy Tarreau8a8d83b2015-04-13 13:24:54 +020034#include <proto/applet.h>
Willy Tarreauc7e42382012-08-24 19:22:53 +020035#include <proto/channel.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020036#include <proto/fd.h>
Willy Tarreaud1d48d42015-03-13 16:15:46 +010037#include <proto/frontend.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020038#include <proto/log.h>
39#include <proto/hdr_idx.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020040#include <proto/proto_tcp.h>
41#include <proto/proto_http.h>
42#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 */
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200272static int peer_prepare_updatemsg(struct stksess *ts, struct shared_table *st, 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 Bruna6a09982015-09-22 15:34:19 +0200285 if (!st->last_pushed || ts->upd.key < st->last_pushed || ((ts->upd.key - st->last_pushed) != 1)) {
286 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 Brun2b920a12010-09-23 18:30:22 +0200291 netinteger = htonl(ts->upd.key);
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
Emeric Brunb3971ab2015-05-12 18:49:09 +0200320 /* encode values */
Emeric Brun94900952015-06-11 18:25:54 +0200321 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200322
Emeric Brun94900952015-06-11 18:25:54 +0200323 data_ptr = stktable_data_ptr(st->table, ts, data_type);
324 if (data_ptr) {
325 switch (stktable_data_types[data_type].std_type) {
326 case STD_T_SINT: {
327 int data;
328
329 data = stktable_data_cast(data_ptr, std_t_sint);
330 intencode(data, &cursor);
331 break;
332 }
333 case STD_T_UINT: {
334 unsigned int data;
335
336 data = stktable_data_cast(data_ptr, std_t_uint);
337 intencode(data, &cursor);
338 break;
339 }
340 case STD_T_ULL: {
341 unsigned long long data;
342
343 data = stktable_data_cast(data_ptr, std_t_ull);
344 intencode(data, &cursor);
345 break;
346 }
347 case STD_T_FRQP: {
348 struct freq_ctr_period *frqp;
349
350 frqp = &stktable_data_cast(data_ptr, std_t_frqp);
351 intencode((unsigned int)(now_ms - frqp->curr_tick), &cursor);
352 intencode(frqp->curr_ctr, &cursor);
353 intencode(frqp->prev_ctr, &cursor);
354 break;
355 }
356 }
357 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200358 }
359
360 /* Compute datalen */
361 datalen = (cursor - datamsg);
362
363 /* prepare message header */
364 msg[0] = PEER_MSG_CLASS_STICKTABLE;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200365 peer_set_update_msg_type(&msg[1], use_identifier, use_timed);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200366 cursor = &msg[2];
367 intencode(datalen, &cursor);
368
369 /* move data after header */
370 memmove(cursor, datamsg, datalen);
371
372 /* return header size + data_len */
373 return (cursor - msg) + datalen;
374}
375
376/*
377 * This prepare the switch table message to targeted share table <st>.
378 * <msg> is a buffer of <size> to recieve data message content
379 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
380 * check size)
381 */
382static int peer_prepare_switchmsg(struct shared_table *st, char *msg, size_t size)
383{
384 int len;
385 unsigned short datalen;
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200386 struct chunk *chunk;
387 char *cursor, *datamsg, *chunkp, *chunkq;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200388 uint64_t data = 0;
Emeric Brun94900952015-06-11 18:25:54 +0200389 unsigned int data_type;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200390
391 cursor = datamsg = msg + 2 + 5;
392
393 /* Encode data */
394
395 /* encode local id */
396 intencode(st->local_id, &cursor);
397
398 /* encode table name */
399 len = strlen(st->table->id);
400 intencode(len, &cursor);
401 memcpy(cursor, st->table->id, len);
402 cursor += len;
403
404 /* encode table type */
405
406 intencode(st->table->type, &cursor);
407
408 /* encode table key size */
409 intencode(st->table->key_size, &cursor);
410
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200411 chunk = get_trash_chunk();
412 chunkp = chunkq = chunk->str;
Emeric Brun94900952015-06-11 18:25:54 +0200413 /* encode available known data types in table */
414 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
415 if (st->table->data_ofs[data_type]) {
416 switch (stktable_data_types[data_type].std_type) {
417 case STD_T_SINT:
418 case STD_T_UINT:
419 case STD_T_ULL:
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200420 data |= 1 << data_type;
421 break;
Emeric Brun94900952015-06-11 18:25:54 +0200422 case STD_T_FRQP:
423 data |= 1 << data_type;
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200424 intencode(data_type, &chunkq);
425 intencode(st->table->data_arg[data_type].u, &chunkq);
Emeric Brun94900952015-06-11 18:25:54 +0200426 break;
427 }
428 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200429 }
430 intencode(data, &cursor);
431
Frédéric Lécaille37a72542017-07-06 15:02:16 +0200432 /* Encode stick-table entries duration. */
433 intencode(st->table->expire, &cursor);
434
435 if (chunkq > chunkp) {
436 chunk->len = chunkq - chunkp;
437 memcpy(cursor, chunk->str, chunk->len);
438 cursor += chunk->len;
439 }
440
Emeric Brunb3971ab2015-05-12 18:49:09 +0200441 /* Compute datalen */
442 datalen = (cursor - datamsg);
Emeric Brun2b920a12010-09-23 18:30:22 +0200443
Emeric Brunb3971ab2015-05-12 18:49:09 +0200444 /* prepare message header */
445 msg[0] = PEER_MSG_CLASS_STICKTABLE;
446 msg[1] = PEER_MSG_STKT_DEFINE;
447 cursor = &msg[2];
448 intencode(datalen, &cursor);
Emeric Brun2b920a12010-09-23 18:30:22 +0200449
Emeric Brunb3971ab2015-05-12 18:49:09 +0200450 /* move data after header */
451 memmove(cursor, datamsg, datalen);
452
453 /* return header size + data_len */
454 return (cursor - msg) + datalen;
Emeric Brun2b920a12010-09-23 18:30:22 +0200455}
456
Emeric Brunb3971ab2015-05-12 18:49:09 +0200457/*
458 * This prepare the acknowledge message on the stick session <ts>, <st> is the considered
459 * stick table.
460 * <msg> is a buffer of <size> to recieve data message content
461 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
462 * check size)
463 */
464static int peer_prepare_ackmsg(struct shared_table *st, char *msg, size_t size)
465{
466 unsigned short datalen;
467 char *cursor, *datamsg;
468 uint32_t netinteger;
469
Emeric Brunb058f1c2015-09-22 15:50:18 +0200470 cursor = datamsg = msg + 2 + 5;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200471
472 intencode(st->remote_id, &cursor);
473 netinteger = htonl(st->last_get);
474 memcpy(cursor, &netinteger, sizeof(netinteger));
475 cursor += sizeof(netinteger);
476
477 /* Compute datalen */
478 datalen = (cursor - datamsg);
479
480 /* prepare message header */
481 msg[0] = PEER_MSG_CLASS_STICKTABLE;
Emeric Brune1ab8082015-08-21 11:48:54 +0200482 msg[1] = PEER_MSG_STKT_ACK;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200483 cursor = &msg[2];
484 intencode(datalen, &cursor);
485
486 /* move data after header */
487 memmove(cursor, datamsg, datalen);
488
489 /* return header size + data_len */
490 return (cursor - msg) + datalen;
491}
Emeric Brun2b920a12010-09-23 18:30:22 +0200492
493/*
494 * Callback to release a session with a peer
495 */
Willy Tarreau00a37f02015-04-13 12:05:19 +0200496static void peer_session_release(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +0200497{
Willy Tarreau00a37f02015-04-13 12:05:19 +0200498 struct stream_interface *si = appctx->owner;
Willy Tarreau87b09662015-04-03 00:22:06 +0200499 struct stream *s = si_strm(si);
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200500 struct peer *peer = appctx->ctx.peers.ptr;
501 struct peers *peers = strm_fe(s)->parent;
Emeric Brun2b920a12010-09-23 18:30:22 +0200502
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100503 /* appctx->ctx.peers.ptr is not a peer session */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100504 if (appctx->st0 < PEER_SESS_ST_SENDSUCCESS)
Emeric Brun2b920a12010-09-23 18:30:22 +0200505 return;
506
507 /* peer session identified */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200508 if (peer) {
Willy Tarreau9df94c22016-10-31 18:42:52 +0100509 if (peer->appctx == appctx) {
Emeric Brunb157d732015-08-21 12:00:30 +0200510 /* Re-init current table pointers to force announcement on re-connect */
511 peer->remote_table = peer->last_local_table = NULL;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200512 peer->appctx = NULL;
513 if (peer->flags & PEER_F_LEARN_ASSIGN) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200514 /* unassign current peer for learning */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200515 peer->flags &= ~(PEER_F_LEARN_ASSIGN);
516 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brun2b920a12010-09-23 18:30:22 +0200517
518 /* reschedule a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200519 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
Emeric Brun2b920a12010-09-23 18:30:22 +0200520 }
521 /* reset teaching and learning flags to 0 */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200522 peer->flags &= PEER_TEACH_RESET;
523 peer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200524 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200525 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200526 }
527}
528
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200529/* Retrieve the major and minor versions of peers protocol
530 * announced by a remote peer. <str> is a null-terminated
531 * string with the following format: "<maj_ver>.<min_ver>".
532 */
533static int peer_get_version(const char *str,
534 unsigned int *maj_ver, unsigned int *min_ver)
535{
536 unsigned int majv, minv;
537 const char *pos, *saved;
538 const char *end;
539
540 saved = pos = str;
541 end = str + strlen(str);
542
543 majv = read_uint(&pos, end);
544 if (saved == pos || *pos++ != '.')
545 return -1;
546
547 saved = pos;
548 minv = read_uint(&pos, end);
549 if (saved == pos || pos != end)
550 return -1;
551
552 *maj_ver = majv;
553 *min_ver = minv;
554
555 return 0;
556}
Emeric Brun2b920a12010-09-23 18:30:22 +0200557
558/*
559 * IO Handler to handle message exchance with a peer
560 */
Willy Tarreau00a37f02015-04-13 12:05:19 +0200561static void peer_io_handler(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +0200562{
Willy Tarreau00a37f02015-04-13 12:05:19 +0200563 struct stream_interface *si = appctx->owner;
Willy Tarreau87b09662015-04-03 00:22:06 +0200564 struct stream *s = si_strm(si);
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200565 struct peers *curpeers = strm_fe(s)->parent;
Emeric Brun2b920a12010-09-23 18:30:22 +0200566 int reql = 0;
567 int repl = 0;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200568 size_t proto_len = strlen(PEER_SESSION_PROTO_NAME);
569 unsigned int maj_ver, min_ver;
Emeric Brun2b920a12010-09-23 18:30:22 +0200570
Christopher Fauleta73e59b2016-12-09 17:30:18 +0100571 /* Check if the input buffer is avalaible. */
572 if (si_ic(si)->buf->size == 0)
573 goto full;
574
Emeric Brun2b920a12010-09-23 18:30:22 +0200575 while (1) {
576switchstate:
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200577 maj_ver = min_ver = (unsigned int)-1;
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100578 switch(appctx->st0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100579 case PEER_SESS_ST_ACCEPT:
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100580 appctx->ctx.peers.ptr = NULL;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100581 appctx->st0 = PEER_SESS_ST_GETVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200582 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100583 case PEER_SESS_ST_GETVERSION:
Willy Tarreau06d80a92017-10-19 14:32:15 +0200584 reql = co_getline(si_oc(si), trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200585 if (reql <= 0) { /* closed or EOL not found */
586 if (reql == 0)
587 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100588 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200589 goto switchstate;
590 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100591 if (trash.str[reql-1] != '\n') {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100592 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200593 goto switchstate;
594 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100595 else if (reql > 1 && (trash.str[reql-2] == '\r'))
596 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200597 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100598 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200599
Willy Tarreau06d80a92017-10-19 14:32:15 +0200600 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200601
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200602 /* test protocol */
603 if (strncmp(PEER_SESSION_PROTO_NAME " ", trash.str, proto_len + 1) != 0) {
604 appctx->st0 = PEER_SESS_ST_EXIT;
605 appctx->st1 = PEER_SESS_SC_ERRPROTO;
606 goto switchstate;
607 }
608 if (peer_get_version(trash.str + proto_len + 1, &maj_ver, &min_ver) == -1 ||
609 maj_ver != PEER_MAJOR_VER || min_ver > PEER_MINOR_VER) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100610 appctx->st0 = PEER_SESS_ST_EXIT;
611 appctx->st1 = PEER_SESS_SC_ERRVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200612 goto switchstate;
613 }
614
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100615 appctx->st0 = PEER_SESS_ST_GETHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200616 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100617 case PEER_SESS_ST_GETHOST:
Willy Tarreau06d80a92017-10-19 14:32:15 +0200618 reql = co_getline(si_oc(si), trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200619 if (reql <= 0) { /* closed or EOL not found */
620 if (reql == 0)
621 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100622 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200623 goto switchstate;
624 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100625 if (trash.str[reql-1] != '\n') {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100626 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200627 goto switchstate;
628 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100629 else if (reql > 1 && (trash.str[reql-2] == '\r'))
630 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200631 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100632 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200633
Willy Tarreau06d80a92017-10-19 14:32:15 +0200634 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200635
636 /* test hostname match */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100637 if (strcmp(localpeer, trash.str) != 0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100638 appctx->st0 = PEER_SESS_ST_EXIT;
639 appctx->st1 = PEER_SESS_SC_ERRHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200640 goto switchstate;
641 }
642
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100643 appctx->st0 = PEER_SESS_ST_GETPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200644 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100645 case PEER_SESS_ST_GETPEER: {
Emeric Brun2b920a12010-09-23 18:30:22 +0200646 struct peer *curpeer;
647 char *p;
Willy Tarreau06d80a92017-10-19 14:32:15 +0200648 reql = co_getline(si_oc(si), trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200649 if (reql <= 0) { /* closed or EOL not found */
650 if (reql == 0)
651 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100652 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200653 goto switchstate;
654 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100655 if (trash.str[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200656 /* Incomplete line, we quit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100657 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200658 goto switchstate;
659 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100660 else if (reql > 1 && (trash.str[reql-2] == '\r'))
661 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200662 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100663 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200664
Willy Tarreau06d80a92017-10-19 14:32:15 +0200665 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200666
Emeric Brunb3971ab2015-05-12 18:49:09 +0200667 /* parse line "<peer name> <pid> <relative_pid>" */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100668 p = strchr(trash.str, ' ');
Emeric Brun2b920a12010-09-23 18:30:22 +0200669 if (!p) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100670 appctx->st0 = PEER_SESS_ST_EXIT;
671 appctx->st1 = PEER_SESS_SC_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +0200672 goto switchstate;
673 }
674 *p = 0;
675
676 /* lookup known peer */
677 for (curpeer = curpeers->remote; curpeer; curpeer = curpeer->next) {
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100678 if (strcmp(curpeer->id, trash.str) == 0)
Emeric Brun2b920a12010-09-23 18:30:22 +0200679 break;
680 }
681
682 /* if unknown peer */
683 if (!curpeer) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100684 appctx->st0 = PEER_SESS_ST_EXIT;
685 appctx->st1 = PEER_SESS_SC_ERRPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200686 goto switchstate;
687 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200688
Willy Tarreau9df94c22016-10-31 18:42:52 +0100689 if (curpeer->appctx && curpeer->appctx != appctx) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200690 if (curpeer->local) {
691 /* Local connection, reply a retry */
692 appctx->st0 = PEER_SESS_ST_EXIT;
693 appctx->st1 = PEER_SESS_SC_TRYAGAIN;
694 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +0200695 }
Willy Tarreau81bc3b02016-10-31 17:37:39 +0100696 peer_session_forceshutdown(curpeer->appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +0200697 }
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200698 if (maj_ver != (unsigned int)-1 && min_ver != (unsigned int)-1) {
699 if (min_ver == PEER_DWNGRD_MINOR_VER) {
700 curpeer->flags |= PEER_F_DWNGRD;
701 }
702 else {
703 curpeer->flags &= ~PEER_F_DWNGRD;
704 }
705 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200706 curpeer->appctx = appctx;
707 appctx->ctx.peers.ptr = curpeer;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100708 appctx->st0 = PEER_SESS_ST_SENDSUCCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200709 /* fall through */
710 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100711 case PEER_SESS_ST_SENDSUCCESS: {
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200712 struct peer *curpeer = appctx->ctx.peers.ptr;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200713 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +0200714
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100715 repl = snprintf(trash.str, trash.size, "%d\n", PEER_SESS_SC_SUCCESSCODE);
Willy Tarreau06d80a92017-10-19 14:32:15 +0200716 repl = ci_putblk(si_ic(si), trash.str, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200717 if (repl <= 0) {
718 if (repl == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +0100719 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100720 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200721 goto switchstate;
722 }
723
724 /* Register status code */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200725 curpeer->statuscode = PEER_SESS_SC_SUCCESSCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200726
727 /* Awake main task */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200728 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200729
730 /* Init confirm counter */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200731 curpeer->confirm = 0;
732
733 /* Init cursors */
734 for (st = curpeer->tables; st ; st = st->next) {
735 st->last_get = st->last_acked = 0;
736 st->teaching_origin = st->last_pushed = st->update;
737 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200738
739 /* reset teaching and learning flags to 0 */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200740 curpeer->flags &= PEER_TEACH_RESET;
741 curpeer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200742
743 /* if current peer is local */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200744 if (curpeer->local) {
745 /* if current host need resyncfrom local and no process assined */
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200746 if ((curpeers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL &&
747 !(curpeers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200748 /* assign local peer for a lesson, consider lesson already requested */
749 curpeer->flags |= PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200750 curpeers->flags |= (PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200751 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200752
Emeric Brunb3971ab2015-05-12 18:49:09 +0200753 }
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200754 else if ((curpeers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
755 !(curpeers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200756 /* assign peer for a lesson */
757 curpeer->flags |= PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200758 curpeers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200759 }
760
761
Emeric Brun2b920a12010-09-23 18:30:22 +0200762 /* switch to waiting message state */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100763 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200764 goto switchstate;
765 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100766 case PEER_SESS_ST_CONNECT: {
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200767 struct peer *curpeer = appctx->ctx.peers.ptr;
Emeric Brun2b920a12010-09-23 18:30:22 +0200768
769 /* Send headers */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100770 repl = snprintf(trash.str, trash.size,
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200771 PEER_SESSION_PROTO_NAME " %u.%u\n%s\n%s %d %d\n",
772 PEER_MAJOR_VER,
773 (curpeer->flags & PEER_F_DWNGRD) ? PEER_DWNGRD_MINOR_VER : PEER_MINOR_VER,
Emeric Brunb3971ab2015-05-12 18:49:09 +0200774 curpeer->id,
Emeric Brun2b920a12010-09-23 18:30:22 +0200775 localpeer,
Willy Tarreau7b77c9f2012-01-07 22:52:12 +0100776 (int)getpid(),
Emeric Brunb3971ab2015-05-12 18:49:09 +0200777 relative_pid);
Emeric Brun2b920a12010-09-23 18:30:22 +0200778
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100779 if (repl >= trash.size) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100780 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200781 goto switchstate;
782 }
783
Willy Tarreau06d80a92017-10-19 14:32:15 +0200784 repl = ci_putblk(si_ic(si), trash.str, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200785 if (repl <= 0) {
786 if (repl == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +0100787 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100788 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200789 goto switchstate;
790 }
791
792 /* switch to the waiting statuscode state */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100793 appctx->st0 = PEER_SESS_ST_GETSTATUS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200794 /* fall through */
795 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100796 case PEER_SESS_ST_GETSTATUS: {
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200797 struct peer *curpeer = appctx->ctx.peers.ptr;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200798 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +0200799
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100800 if (si_ic(si)->flags & CF_WRITE_PARTIAL)
Emeric Brunb3971ab2015-05-12 18:49:09 +0200801 curpeer->statuscode = PEER_SESS_SC_CONNECTEDCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200802
Willy Tarreau06d80a92017-10-19 14:32:15 +0200803 reql = co_getline(si_oc(si), trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200804 if (reql <= 0) { /* closed or EOL not found */
805 if (reql == 0)
806 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100807 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200808 goto switchstate;
809 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100810 if (trash.str[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200811 /* Incomplete line, we quit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100812 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200813 goto switchstate;
814 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100815 else if (reql > 1 && (trash.str[reql-2] == '\r'))
816 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200817 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100818 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200819
Willy Tarreau06d80a92017-10-19 14:32:15 +0200820 co_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200821
822 /* Register status code */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200823 curpeer->statuscode = atoi(trash.str);
Emeric Brun2b920a12010-09-23 18:30:22 +0200824
825 /* Awake main task */
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200826 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200827
828 /* If status code is success */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200829 if (curpeer->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200830 /* Init cursors */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200831 for (st = curpeer->tables; st ; st = st->next) {
832 st->last_get = st->last_acked = 0;
833 st->teaching_origin = st->last_pushed = st->update;
834 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200835
836 /* Init confirm counter */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200837 curpeer->confirm = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200838
Emeric Brunb3971ab2015-05-12 18:49:09 +0200839 /* reset teaching and learning flags to 0 */
840 curpeer->flags &= PEER_TEACH_RESET;
841 curpeer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200842
Emeric Brunb3971ab2015-05-12 18:49:09 +0200843 /* If current peer is local */
844 if (curpeer->local) {
845 /* flag to start to teach lesson */
846 curpeer->flags |= PEER_F_TEACH_PROCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200847
Emeric Brunb3971ab2015-05-12 18:49:09 +0200848 }
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200849 else if ((curpeers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
850 !(curpeers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200851 /* If peer is remote and resync from remote is needed,
852 and no peer currently assigned */
Emeric Brun2b920a12010-09-23 18:30:22 +0200853
Emeric Brunb3971ab2015-05-12 18:49:09 +0200854 /* assign peer for a lesson */
855 curpeer->flags |= PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200856 curpeers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brun2b920a12010-09-23 18:30:22 +0200857 }
858
859 }
860 else {
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200861 if (curpeer->statuscode == PEER_SESS_SC_ERRVERSION)
862 curpeer->flags |= PEER_F_DWNGRD;
Emeric Brun2b920a12010-09-23 18:30:22 +0200863 /* Status code is not success, abort */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100864 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200865 goto switchstate;
866 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100867 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200868 /* fall through */
869 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100870 case PEER_SESS_ST_WAITMSG: {
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200871 struct peer *curpeer = appctx->ctx.peers.ptr;
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200872 struct stksess *ts, *newts = NULL;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200873 uint32_t msg_len = 0;
874 char *msg_cur = trash.str;
875 char *msg_end = trash.str;
876 unsigned char msg_head[7];
Emeric Brun2b920a12010-09-23 18:30:22 +0200877 int totl = 0;
878
Willy Tarreau06d80a92017-10-19 14:32:15 +0200879 reql = co_getblk(si_oc(si), (char *)msg_head, 2*sizeof(unsigned char), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200880 if (reql <= 0) /* closed or EOL not found */
881 goto incomplete;
882
Emeric Brun2b920a12010-09-23 18:30:22 +0200883 totl += reql;
884
Emeric Brunb3971ab2015-05-12 18:49:09 +0200885 if (msg_head[1] >= 128) {
886 /* Read and Decode message length */
Willy Tarreau06d80a92017-10-19 14:32:15 +0200887 reql = co_getblk(si_oc(si), (char *)&msg_head[2], sizeof(unsigned char), totl);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200888 if (reql <= 0) /* closed */
889 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +0200890
Emeric Brunb3971ab2015-05-12 18:49:09 +0200891 totl += reql;
892
893 if (msg_head[2] < 240) {
894 msg_len = msg_head[2];
Emeric Brun2b920a12010-09-23 18:30:22 +0200895 }
896 else {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200897 int i;
898 char *cur;
899 char *end;
Willy Tarreau72d6c162013-04-11 16:14:13 +0200900
Emeric Brunb3971ab2015-05-12 18:49:09 +0200901 for (i = 3 ; i < sizeof(msg_head) ; i++) {
Willy Tarreau06d80a92017-10-19 14:32:15 +0200902 reql = co_getblk(si_oc(si), (char *)&msg_head[i], sizeof(char), totl);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200903 if (reql <= 0) /* closed */
904 goto incomplete;
905
906 totl += reql;
907
908 if (!(msg_head[i] & 0x80))
909 break;
910 }
911
912 if (i == sizeof(msg_head)) {
913 /* malformed message */
914 appctx->st0 = PEER_SESS_ST_ERRPROTO;
915 goto switchstate;
916
917 }
918 end = (char *)msg_head + sizeof(msg_head);
919 cur = (char *)&msg_head[2];
920 msg_len = intdecode(&cur, end);
921 if (!cur) {
922 /* malformed message */
923 appctx->st0 = PEER_SESS_ST_ERRPROTO;
924 goto switchstate;
925 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200926 }
927
Willy Tarreau86a446e2013-11-25 23:02:37 +0100928
Emeric Brunb3971ab2015-05-12 18:49:09 +0200929 /* Read message content */
930 if (msg_len) {
931 if (msg_len > trash.size) {
932 /* Status code is not success, abort */
933 appctx->st0 = PEER_SESS_ST_ERRSIZE;
934 goto switchstate;
935 }
Willy Tarreau72d6c162013-04-11 16:14:13 +0200936
Willy Tarreau06d80a92017-10-19 14:32:15 +0200937 reql = co_getblk(si_oc(si), trash.str, msg_len, totl);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200938 if (reql <= 0) /* closed */
939 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +0200940 totl += reql;
Willy Tarreau86a446e2013-11-25 23:02:37 +0100941
Emeric Brunb3971ab2015-05-12 18:49:09 +0200942 msg_end += msg_len;
943 }
944 }
Willy Tarreau86a446e2013-11-25 23:02:37 +0100945
Emeric Brunb3971ab2015-05-12 18:49:09 +0200946 if (msg_head[0] == PEER_MSG_CLASS_CONTROL) {
947 if (msg_head[1] == PEER_MSG_CTRL_RESYNCREQ) {
948 struct shared_table *st;
949 /* Reset message: remote need resync */
950
951 /* prepare tables fot a global push */
952 for (st = curpeer->tables; st; st = st->next) {
953 st->teaching_origin = st->last_pushed = st->table->update;
954 st->flags = 0;
Willy Tarreau86a446e2013-11-25 23:02:37 +0100955 }
Willy Tarreau72d6c162013-04-11 16:14:13 +0200956
Emeric Brunb3971ab2015-05-12 18:49:09 +0200957 /* reset teaching flags to 0 */
958 curpeer->flags &= PEER_TEACH_RESET;
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200959
Emeric Brunb3971ab2015-05-12 18:49:09 +0200960 /* flag to start to teach lesson */
961 curpeer->flags |= PEER_F_TEACH_PROCESS;
962
963
964 }
965 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCFINISHED) {
966
967 if (curpeer->flags & PEER_F_LEARN_ASSIGN) {
968 curpeer->flags &= ~PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200969 curpeers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
970 curpeers->flags |= (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE);
Willy Tarreau86a446e2013-11-25 23:02:37 +0100971 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200972 curpeer->confirm++;
Emeric Brun2b920a12010-09-23 18:30:22 +0200973 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200974 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCPARTIAL) {
975
976 if (curpeer->flags & PEER_F_LEARN_ASSIGN) {
977 curpeer->flags &= ~PEER_F_LEARN_ASSIGN;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200978 curpeers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200979
980 curpeer->flags |= PEER_F_LEARN_NOTUP2DATE;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +0200981 curpeers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
982 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Cyril Bonté9a60ff92014-02-16 01:07:07 +0100983 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200984 curpeer->confirm++;
Emeric Brun2b920a12010-09-23 18:30:22 +0200985 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200986 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCCONFIRM) {
Emeric Brun597b26e2016-08-12 11:23:31 +0200987 struct shared_table *st;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200988
989 /* If stopping state */
990 if (stopping) {
991 /* Close session, push resync no more needed */
992 curpeer->flags |= PEER_F_TEACH_COMPLETE;
993 appctx->st0 = PEER_SESS_ST_END;
994 goto switchstate;
995 }
Emeric Brun597b26e2016-08-12 11:23:31 +0200996 for (st = curpeer->tables; st; st = st->next) {
997 st->update = st->last_pushed = st->teaching_origin;
998 st->flags = 0;
999 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001000
1001 /* reset teaching flags to 0 */
1002 curpeer->flags &= PEER_TEACH_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +02001003 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001004 }
1005 else if (msg_head[0] == PEER_MSG_CLASS_STICKTABLE) {
1006 if (msg_head[1] == PEER_MSG_STKT_DEFINE) {
1007 int table_id_len;
1008 struct shared_table *st;
1009 int table_type;
1010 int table_keylen;
1011 int table_id;
1012 uint64_t table_data;
Emeric Brun2b920a12010-09-23 18:30:22 +02001013
Emeric Brunb3971ab2015-05-12 18:49:09 +02001014 table_id = intdecode(&msg_cur, msg_end);
1015 if (!msg_cur) {
1016 /* malformed message */
1017 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1018 goto switchstate;
1019 }
Willy Tarreau72d6c162013-04-11 16:14:13 +02001020
Emeric Brunb3971ab2015-05-12 18:49:09 +02001021 table_id_len = intdecode(&msg_cur, msg_end);
1022 if (!msg_cur) {
1023 /* malformed message */
1024 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1025 goto switchstate;
1026 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001027
Emeric Brunb3971ab2015-05-12 18:49:09 +02001028 curpeer->remote_table = NULL;
1029 if (!table_id_len || (msg_cur + table_id_len) >= msg_end) {
1030 /* malformed message */
1031 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1032 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001033 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001034
Emeric Brunb3971ab2015-05-12 18:49:09 +02001035 for (st = curpeer->tables; st; st = st->next) {
1036 /* Reset IDs */
1037 if (st->remote_id == table_id)
1038 st->remote_id = 0;
Willy Tarreau9d9179b2013-04-11 16:56:44 +02001039
Emeric Brunb3971ab2015-05-12 18:49:09 +02001040 if (!curpeer->remote_table
1041 && (table_id_len == strlen(st->table->id))
1042 && (memcmp(st->table->id, msg_cur, table_id_len) == 0)) {
1043 curpeer->remote_table = st;
Emeric Brun2b920a12010-09-23 18:30:22 +02001044 }
1045 }
1046
Emeric Brunb3971ab2015-05-12 18:49:09 +02001047 if (!curpeer->remote_table) {
1048 goto ignore_msg;
1049 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001050
Emeric Brunb3971ab2015-05-12 18:49:09 +02001051 msg_cur += table_id_len;
1052 if (msg_cur >= msg_end) {
1053 /* malformed message */
1054 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1055 goto switchstate;
1056 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001057
Emeric Brunb3971ab2015-05-12 18:49:09 +02001058 table_type = intdecode(&msg_cur, msg_end);
1059 if (!msg_cur) {
1060 /* malformed message */
1061 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1062 goto switchstate;
1063 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001064
Emeric Brunb3971ab2015-05-12 18:49:09 +02001065 table_keylen = intdecode(&msg_cur, msg_end);
1066 if (!msg_cur) {
1067 /* malformed message */
1068 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1069 goto switchstate;
1070 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001071
Emeric Brunb3971ab2015-05-12 18:49:09 +02001072 table_data = 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 if (curpeer->remote_table->table->type != table_type
1080 || curpeer->remote_table->table->key_size != table_keylen) {
1081 curpeer->remote_table = NULL;
1082 goto ignore_msg;
1083 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001084
Emeric Brunb3971ab2015-05-12 18:49:09 +02001085 curpeer->remote_table->remote_data = table_data;
1086 curpeer->remote_table->remote_id = table_id;
Emeric Brun2b920a12010-09-23 18:30:22 +02001087 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001088 else if (msg_head[1] == PEER_MSG_STKT_SWITCH) {
1089 struct shared_table *st;
1090 int table_id;
Emeric Brun2b920a12010-09-23 18:30:22 +02001091
Emeric Brunb3971ab2015-05-12 18:49:09 +02001092 table_id = intdecode(&msg_cur, msg_end);
1093 if (!msg_cur) {
1094 /* malformed message */
1095 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1096 goto switchstate;
1097 }
1098 curpeer->remote_table = NULL;
1099 for (st = curpeer->tables; st; st = st->next) {
1100 if (st->remote_id == table_id) {
1101 curpeer->remote_table = st;
1102 break;
1103 }
1104 }
1105
Emeric Brun2b920a12010-09-23 18:30:22 +02001106 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001107 else if (msg_head[1] == PEER_MSG_STKT_UPDATE
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001108 || msg_head[1] == PEER_MSG_STKT_INCUPDATE
1109 || msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED
1110 || msg_head[1] == PEER_MSG_STKT_INCUPDATE_TIMED) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001111 struct shared_table *st = curpeer->remote_table;
1112 uint32_t update;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001113 int expire;
Emeric Brun94900952015-06-11 18:25:54 +02001114 unsigned int data_type;
1115 void *data_ptr;
Emeric Brun2b920a12010-09-23 18:30:22 +02001116
Emeric Brunb3971ab2015-05-12 18:49:09 +02001117 /* Here we have data message */
1118 if (!st)
1119 goto ignore_msg;
Emeric Brun2b920a12010-09-23 18:30:22 +02001120
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001121 expire = MS_TO_TICKS(st->table->expire);
1122
1123 if (msg_head[1] == PEER_MSG_STKT_UPDATE ||
1124 msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001125 if (msg_len < sizeof(update)) {
1126 /* malformed message */
1127 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1128 goto switchstate;
1129 }
1130 memcpy(&update, msg_cur, sizeof(update));
1131 msg_cur += sizeof(update);
1132 st->last_get = htonl(update);
1133 }
1134 else {
1135 st->last_get++;
1136 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001137
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001138 if (msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED ||
1139 msg_head[1] == PEER_MSG_STKT_INCUPDATE_TIMED) {
1140 size_t expire_sz = sizeof expire;
1141
1142 if (msg_cur + expire_sz > msg_end) {
1143 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1144 goto switchstate;
1145 }
1146 memcpy(&expire, msg_cur, expire_sz);
1147 msg_cur += expire_sz;
1148 expire = ntohl(expire);
1149 }
1150
Emeric Brunb3971ab2015-05-12 18:49:09 +02001151 newts = stksess_new(st->table, NULL);
1152 if (!newts)
1153 goto ignore_msg;
Emeric Brun2b920a12010-09-23 18:30:22 +02001154
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +02001155 if (st->table->type == SMP_T_STR) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001156 unsigned int to_read, to_store;
Emeric Brun2b920a12010-09-23 18:30:22 +02001157
Emeric Brunb3971ab2015-05-12 18:49:09 +02001158 to_read = intdecode(&msg_cur, msg_end);
1159 if (!msg_cur) {
1160 /* malformed message */
1161 stksess_free(st->table, newts);
1162 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1163 goto switchstate;
1164 }
1165 to_store = MIN(to_read, st->table->key_size - 1);
1166 if (msg_cur + to_store > msg_end) {
1167 /* malformed message */
1168 stksess_free(st->table, newts);
1169 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1170 goto switchstate;
1171 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001172
Emeric Brunb3971ab2015-05-12 18:49:09 +02001173 memcpy(newts->key.key, msg_cur, to_store);
1174 newts->key.key[to_store] = 0;
1175 msg_cur += to_read;
1176 }
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +02001177 else if (st->table->type == SMP_T_SINT) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001178 unsigned int netinteger;
Willy Tarreau72d6c162013-04-11 16:14:13 +02001179
Emeric Brunb3971ab2015-05-12 18:49:09 +02001180 if (msg_cur + sizeof(netinteger) > msg_end) {
1181 /* malformed message */
1182 stksess_free(st->table, newts);
1183 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1184 goto switchstate;
1185 }
1186 memcpy(&netinteger, msg_cur, sizeof(netinteger));
1187 netinteger = ntohl(netinteger);
1188 memcpy(newts->key.key, &netinteger, sizeof(netinteger));
1189 msg_cur += sizeof(netinteger);
1190 }
1191 else {
1192 if (msg_cur + st->table->key_size > msg_end) {
1193 /* malformed message */
1194 stksess_free(st->table, newts);
1195 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1196 goto switchstate;
1197 }
1198 memcpy(newts->key.key, msg_cur, st->table->key_size);
1199 msg_cur += st->table->key_size;
1200 }
1201
1202 /* lookup for existing entry */
1203 ts = stktable_lookup(st->table, newts);
1204 if (ts) {
1205 /* the entry already exist, we can free ours */
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001206 stktable_touch_with_exp(st->table, ts, 0, tick_add(now_ms, expire));
Emeric Brunb3971ab2015-05-12 18:49:09 +02001207 stksess_free(st->table, newts);
1208 newts = NULL;
1209 }
1210 else {
1211 struct eb32_node *eb;
1212
1213 /* create new entry */
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001214 ts = stktable_store_with_exp(st->table, newts, 0, tick_add(now_ms, expire));
Emeric Brunb3971ab2015-05-12 18:49:09 +02001215 newts = NULL; /* don't reuse it */
1216
Emeric Brun234fc3c2015-12-16 15:16:46 +01001217 ts->upd.key= (++st->table->update)+(2147483648U);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001218 eb = eb32_insert(&st->table->updates, &ts->upd);
1219 if (eb != &ts->upd) {
1220 eb32_delete(eb);
1221 eb32_insert(&st->table->updates, &ts->upd);
1222 }
1223 }
1224
Emeric Brun94900952015-06-11 18:25:54 +02001225 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001226
Emeric Brun94900952015-06-11 18:25:54 +02001227 if ((1 << data_type) & st->remote_data) {
1228 switch (stktable_data_types[data_type].std_type) {
1229 case STD_T_SINT: {
1230 int data;
1231
1232 data = intdecode(&msg_cur, msg_end);
1233 if (!msg_cur) {
1234 /* malformed message */
1235 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1236 goto switchstate;
1237 }
1238
1239 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1240 if (data_ptr)
1241 stktable_data_cast(data_ptr, std_t_sint) = data;
1242 break;
1243 }
1244 case STD_T_UINT: {
1245 unsigned int data;
1246
1247 data = intdecode(&msg_cur, msg_end);
1248 if (!msg_cur) {
1249 /* malformed message */
1250 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1251 goto switchstate;
1252 }
1253
1254 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1255 if (data_ptr)
1256 stktable_data_cast(data_ptr, std_t_uint) = data;
1257 break;
1258 }
1259 case STD_T_ULL: {
1260 unsigned long long data;
1261
1262 data = intdecode(&msg_cur, msg_end);
1263 if (!msg_cur) {
1264 /* malformed message */
1265 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1266 goto switchstate;
1267 }
1268
1269 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1270 if (data_ptr)
1271 stktable_data_cast(data_ptr, std_t_ull) = data;
1272 break;
1273 }
1274 case STD_T_FRQP: {
1275 struct freq_ctr_period data;
1276
Willy Tarreau3bb46172016-03-25 18:17:47 +01001277 data.curr_tick = tick_add(now_ms, -intdecode(&msg_cur, msg_end));
Emeric Brun94900952015-06-11 18:25:54 +02001278 if (!msg_cur) {
1279 /* malformed message */
1280 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1281 goto switchstate;
1282 }
1283 data.curr_ctr = intdecode(&msg_cur, msg_end);
1284 if (!msg_cur) {
1285 /* malformed message */
1286 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1287 goto switchstate;
1288 }
1289 data.prev_ctr = intdecode(&msg_cur, msg_end);
1290 if (!msg_cur) {
1291 /* malformed message */
1292 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1293 goto switchstate;
1294 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001295
Emeric Brun94900952015-06-11 18:25:54 +02001296 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1297 if (data_ptr)
1298 stktable_data_cast(data_ptr, std_t_frqp) = data;
1299 break;
1300 }
1301 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001302 }
1303 }
1304 }
1305 else if (msg_head[1] == PEER_MSG_STKT_ACK) {
1306 /* ack message */
1307 uint32_t table_id ;
1308 uint32_t update;
1309 struct shared_table *st;
1310
1311 table_id = intdecode(&msg_cur, msg_end);
1312 if (!msg_cur || (msg_cur + sizeof(update) > msg_end)) {
1313 /* malformed message */
1314 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1315 goto switchstate;
1316 }
1317 memcpy(&update, msg_cur, sizeof(update));
1318 update = ntohl(update);
Emeric Brun2b920a12010-09-23 18:30:22 +02001319
Emeric Brunb3971ab2015-05-12 18:49:09 +02001320 for (st = curpeer->tables; st; st = st->next) {
1321 if (st->local_id == table_id) {
1322 st->update = update;
1323 break;
1324 }
1325 }
1326 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001327 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001328 else if (msg_head[0] == PEER_MSG_CLASS_RESERVED) {
1329 appctx->st0 = PEER_SESS_ST_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +02001330 goto switchstate;
1331 }
1332
Emeric Brunb3971ab2015-05-12 18:49:09 +02001333ignore_msg:
Emeric Brun2b920a12010-09-23 18:30:22 +02001334 /* skip consumed message */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001335 co_skip(si_oc(si), totl);
Emeric Brun2b920a12010-09-23 18:30:22 +02001336 /* loop on that state to peek next message */
Willy Tarreau72d6c162013-04-11 16:14:13 +02001337 goto switchstate;
1338
Emeric Brun2b920a12010-09-23 18:30:22 +02001339incomplete:
Willy Tarreau06d80a92017-10-19 14:32:15 +02001340 /* we get here when a co_getblk() returns <= 0 in reql */
Willy Tarreau9d9179b2013-04-11 16:56:44 +02001341
Willy Tarreau72d6c162013-04-11 16:14:13 +02001342 if (reql < 0) {
1343 /* there was an error */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001344 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau72d6c162013-04-11 16:14:13 +02001345 goto switchstate;
1346 }
1347
Emeric Brun2b920a12010-09-23 18:30:22 +02001348
Emeric Brun2b920a12010-09-23 18:30:22 +02001349
Emeric Brunb3971ab2015-05-12 18:49:09 +02001350
Emeric Brun2b920a12010-09-23 18:30:22 +02001351 /* Need to request a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001352 if ((curpeer->flags & PEER_F_LEARN_ASSIGN) &&
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001353 (curpeers->flags & PEERS_F_RESYNC_ASSIGN) &&
1354 !(curpeers->flags & PEERS_F_RESYNC_PROCESS)) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001355 unsigned char msg[2];
Emeric Brun2b920a12010-09-23 18:30:22 +02001356
Emeric Brunb3971ab2015-05-12 18:49:09 +02001357 /* Current peer was elected to request a resync */
1358 msg[0] = PEER_MSG_CLASS_CONTROL;
1359 msg[1] = PEER_MSG_CTRL_RESYNCREQ;
Emeric Brun2b920a12010-09-23 18:30:22 +02001360
Emeric Brunb3971ab2015-05-12 18:49:09 +02001361 /* message to buffer */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001362 repl = ci_putblk(si_ic(si), (char *)msg, sizeof(msg));
Emeric Brunb3971ab2015-05-12 18:49:09 +02001363 if (repl <= 0) {
1364 /* no more write possible */
1365 if (repl == -1)
1366 goto full;
1367 appctx->st0 = PEER_SESS_ST_END;
1368 goto switchstate;
1369 }
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001370 curpeers->flags |= PEERS_F_RESYNC_PROCESS;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001371 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001372
Emeric Brunb3971ab2015-05-12 18:49:09 +02001373 /* Nothing to read, now we start to write */
Emeric Brun2b920a12010-09-23 18:30:22 +02001374
Emeric Brunb3971ab2015-05-12 18:49:09 +02001375 if (curpeer->tables) {
1376 struct shared_table *st;
1377 struct shared_table *last_local_table;
Emeric Brun2b920a12010-09-23 18:30:22 +02001378
Emeric Brunb3971ab2015-05-12 18:49:09 +02001379 last_local_table = curpeer->last_local_table;
1380 if (!last_local_table)
1381 last_local_table = curpeer->tables;
1382 st = last_local_table->next;
Emeric Brun2b920a12010-09-23 18:30:22 +02001383
Emeric Brunb3971ab2015-05-12 18:49:09 +02001384 while (1) {
1385 if (!st)
1386 st = curpeer->tables;
Emeric Brun2b920a12010-09-23 18:30:22 +02001387
Emeric Brunb3971ab2015-05-12 18:49:09 +02001388 /* It remains some updates to ack */
1389 if (st->last_get != st->last_acked) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001390 int msglen;
Emeric Brun2b920a12010-09-23 18:30:22 +02001391
Emeric Brunb3971ab2015-05-12 18:49:09 +02001392 msglen = peer_prepare_ackmsg(st, trash.str, trash.size);
1393 if (!msglen) {
1394 /* internal error: message does not fit in trash */
1395 appctx->st0 = PEER_SESS_ST_END;
1396 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001397 }
1398
Emeric Brunb3971ab2015-05-12 18:49:09 +02001399 /* message to buffer */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001400 repl = ci_putblk(si_ic(si), trash.str, msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001401 if (repl <= 0) {
1402 /* no more write possible */
1403 if (repl == -1) {
1404 goto full;
Emeric Brun2b920a12010-09-23 18:30:22 +02001405 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001406 appctx->st0 = PEER_SESS_ST_END;
1407 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001408 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001409 st->last_acked = st->last_get;
Emeric Brun2b920a12010-09-23 18:30:22 +02001410 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001411
Emeric Brunb3971ab2015-05-12 18:49:09 +02001412 if (!(curpeer->flags & PEER_F_TEACH_PROCESS)) {
1413 if (!(curpeer->flags & PEER_F_LEARN_ASSIGN) &&
1414 ((int)(st->last_pushed - st->table->localupdate) < 0)) {
1415 struct eb32_node *eb;
1416 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001417
Emeric Brunb3971ab2015-05-12 18:49:09 +02001418 if (st != curpeer->last_local_table) {
1419 int msglen;
Emeric Brun2b920a12010-09-23 18:30:22 +02001420
Emeric Brunb3971ab2015-05-12 18:49:09 +02001421 msglen = peer_prepare_switchmsg(st, trash.str, trash.size);
1422 if (!msglen) {
1423 /* internal error: message does not fit in trash */
1424 appctx->st0 = PEER_SESS_ST_END;
1425 goto switchstate;
1426 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001427
Emeric Brunb3971ab2015-05-12 18:49:09 +02001428 /* message to buffer */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001429 repl = ci_putblk(si_ic(si), trash.str, msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001430 if (repl <= 0) {
1431 /* no more write possible */
1432 if (repl == -1) {
1433 goto full;
1434 }
1435 appctx->st0 = PEER_SESS_ST_END;
1436 goto switchstate;
1437 }
1438 curpeer->last_local_table = st;
Emeric Brun2b920a12010-09-23 18:30:22 +02001439 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001440
1441 /* We force new pushed to 1 to force identifier in update message */
1442 new_pushed = 1;
1443 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
1444 while (1) {
1445 uint32_t msglen;
1446 struct stksess *ts;
1447
1448 /* push local updates */
1449 if (!eb) {
1450 eb = eb32_first(&st->table->updates);
1451 if (!eb || ((int)(eb->key - st->last_pushed) <= 0)) {
Emeric Brunaaf58602015-06-15 17:23:30 +02001452 st->table->commitupdate = st->last_pushed = st->table->localupdate;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001453 break;
1454 }
1455 }
1456
1457 if ((int)(eb->key - st->table->localupdate) > 0) {
Emeric Brunaaf58602015-06-15 17:23:30 +02001458 st->table->commitupdate = st->last_pushed = st->table->localupdate;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001459 break;
1460 }
1461
1462 ts = eb32_entry(eb, struct stksess, upd);
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001463 msglen = peer_prepare_updatemsg(ts, st, trash.str, trash.size, new_pushed, 0);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001464 if (!msglen) {
1465 /* internal error: message does not fit in trash */
1466 appctx->st0 = PEER_SESS_ST_END;
1467 goto switchstate;
1468 }
1469
1470 /* message to buffer */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001471 repl = ci_putblk(si_ic(si), trash.str, msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001472 if (repl <= 0) {
1473 /* no more write possible */
1474 if (repl == -1) {
1475 goto full;
1476 }
1477 appctx->st0 = PEER_SESS_ST_END;
1478 goto switchstate;
1479 }
1480 st->last_pushed = ts->upd.key;
Emeric Brunaaf58602015-06-15 17:23:30 +02001481 if ((int)(st->last_pushed - st->table->commitupdate) > 0)
1482 st->table->commitupdate = st->last_pushed;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001483 /* identifier may not needed in next update message */
1484 new_pushed = 0;
1485
1486 eb = eb32_next(eb);
1487 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001488 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001489 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001490 else {
1491 if (!(st->flags & SHTABLE_F_TEACH_STAGE1)) {
1492 struct eb32_node *eb;
1493 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001494
Emeric Brunb3971ab2015-05-12 18:49:09 +02001495 if (st != curpeer->last_local_table) {
1496 int msglen;
1497
1498 msglen = peer_prepare_switchmsg(st, trash.str, trash.size);
1499 if (!msglen) {
1500 /* internal error: message does not fit in trash */
1501 appctx->st0 = PEER_SESS_ST_END;
1502 goto switchstate;
1503 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001504
Emeric Brunb3971ab2015-05-12 18:49:09 +02001505 /* message to buffer */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001506 repl = ci_putblk(si_ic(si), trash.str, msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001507 if (repl <= 0) {
1508 /* no more write possible */
1509 if (repl == -1) {
1510 goto full;
1511 }
1512 appctx->st0 = PEER_SESS_ST_END;
1513 goto switchstate;
1514 }
1515 curpeer->last_local_table = st;
1516 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001517
Emeric Brunb3971ab2015-05-12 18:49:09 +02001518 /* We force new pushed to 1 to force identifier in update message */
1519 new_pushed = 1;
1520 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
1521 while (1) {
1522 uint32_t msglen;
1523 struct stksess *ts;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001524 int use_timed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001525
Emeric Brunb3971ab2015-05-12 18:49:09 +02001526 /* push local updates */
1527 if (!eb) {
1528 st->flags |= SHTABLE_F_TEACH_STAGE1;
1529 eb = eb32_first(&st->table->updates);
1530 if (eb)
1531 st->last_pushed = eb->key - 1;
1532 break;
1533 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001534
Emeric Brunb3971ab2015-05-12 18:49:09 +02001535 ts = eb32_entry(eb, struct stksess, upd);
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001536 use_timed = !(curpeer->flags & PEER_F_DWNGRD);
1537 msglen = peer_prepare_updatemsg(ts, st, trash.str, trash.size, new_pushed, use_timed);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001538 if (!msglen) {
1539 /* internal error: message does not fit in trash */
1540 appctx->st0 = PEER_SESS_ST_END;
1541 goto switchstate;
1542 }
1543
1544 /* message to buffer */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001545 repl = ci_putblk(si_ic(si), trash.str, msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001546 if (repl <= 0) {
1547 /* no more write possible */
1548 if (repl == -1) {
1549 goto full;
1550 }
1551 appctx->st0 = PEER_SESS_ST_END;
1552 goto switchstate;
1553 }
1554 st->last_pushed = ts->upd.key;
1555 /* identifier may not needed in next update message */
1556 new_pushed = 0;
1557
1558 eb = eb32_next(eb);
1559 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001560 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001561
Emeric Brunb3971ab2015-05-12 18:49:09 +02001562 if (!(st->flags & SHTABLE_F_TEACH_STAGE2)) {
1563 struct eb32_node *eb;
1564 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001565
Emeric Brunb3971ab2015-05-12 18:49:09 +02001566 if (st != curpeer->last_local_table) {
1567 int msglen;
1568
1569 msglen = peer_prepare_switchmsg(st, trash.str, trash.size);
1570 if (!msglen) {
1571 /* internal error: message does not fit in trash */
1572 appctx->st0 = PEER_SESS_ST_END;
1573 goto switchstate;
1574 }
1575
1576 /* message to buffer */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001577 repl = ci_putblk(si_ic(si), trash.str, msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001578 if (repl <= 0) {
1579 /* no more write possible */
1580 if (repl == -1) {
1581 goto full;
1582 }
1583 appctx->st0 = PEER_SESS_ST_END;
1584 goto switchstate;
1585 }
1586 curpeer->last_local_table = st;
1587 }
1588
1589 /* We force new pushed to 1 to force identifier in update message */
1590 new_pushed = 1;
1591 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
1592 while (1) {
1593 uint32_t msglen;
1594 struct stksess *ts;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001595 int use_timed;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001596
1597 /* push local updates */
1598 if (!eb || eb->key > st->teaching_origin) {
1599 st->flags |= SHTABLE_F_TEACH_STAGE2;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001600 break;
1601 }
1602
1603 ts = eb32_entry(eb, struct stksess, upd);
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001604 use_timed = !(curpeer->flags & PEER_F_DWNGRD);
1605 msglen = peer_prepare_updatemsg(ts, st, trash.str, trash.size, new_pushed, use_timed);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001606 if (!msglen) {
1607 /* internal error: message does not fit in trash */
1608 appctx->st0 = PEER_SESS_ST_END;
1609 goto switchstate;
1610 }
1611
1612 /* message to buffer */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001613 repl = ci_putblk(si_ic(si), trash.str, msglen);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001614 if (repl <= 0) {
1615 /* no more write possible */
1616 if (repl == -1) {
1617 goto full;
1618 }
1619 appctx->st0 = PEER_SESS_ST_END;
1620 goto switchstate;
1621 }
1622 st->last_pushed = ts->upd.key;
1623 /* identifier may not needed in next update message */
1624 new_pushed = 0;
1625
1626 eb = eb32_next(eb);
1627 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001628 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001629 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001630
1631 if (st == last_local_table)
1632 break;
1633 st = st->next;
Emeric Brun2b920a12010-09-23 18:30:22 +02001634 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001635 }
1636
1637
1638 if ((curpeer->flags & PEER_F_TEACH_PROCESS) && !(curpeer->flags & PEER_F_TEACH_FINISHED)) {
1639 unsigned char msg[2];
1640
1641 /* Current peer was elected to request a resync */
1642 msg[0] = PEER_MSG_CLASS_CONTROL;
Frédéric Lécailleed2b4a62017-07-13 09:07:09 +02001643 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 +02001644 /* process final lesson message */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001645 repl = ci_putblk(si_ic(si), (char *)msg, sizeof(msg));
Emeric Brunb3971ab2015-05-12 18:49:09 +02001646 if (repl <= 0) {
1647 /* no more write possible */
1648 if (repl == -1)
1649 goto full;
1650 appctx->st0 = PEER_SESS_ST_END;
1651 goto switchstate;
1652 }
1653 /* flag finished message sent */
1654 curpeer->flags |= PEER_F_TEACH_FINISHED;
1655 }
1656
Emeric Brun597b26e2016-08-12 11:23:31 +02001657 /* Confirm finished or partial messages */
1658 while (curpeer->confirm) {
1659 unsigned char msg[2];
1660
1661 /* There is a confirm messages to send */
1662 msg[0] = PEER_MSG_CLASS_CONTROL;
1663 msg[1] = PEER_MSG_CTRL_RESYNCCONFIRM;
1664
1665 /* message to buffer */
Willy Tarreau06d80a92017-10-19 14:32:15 +02001666 repl = ci_putblk(si_ic(si), (char *)msg, sizeof(msg));
Emeric Brun597b26e2016-08-12 11:23:31 +02001667 if (repl <= 0) {
1668 /* no more write possible */
1669 if (repl == -1)
1670 goto full;
1671 appctx->st0 = PEER_SESS_ST_END;
1672 goto switchstate;
1673 }
1674 curpeer->confirm--;
1675 }
1676
Emeric Brun2b920a12010-09-23 18:30:22 +02001677 /* noting more to do */
1678 goto out;
1679 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001680 case PEER_SESS_ST_EXIT:
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001681 repl = snprintf(trash.str, trash.size, "%d\n", appctx->st1);
Willy Tarreau06d80a92017-10-19 14:32:15 +02001682 if (ci_putblk(si_ic(si), trash.str, repl) == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +01001683 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001684 appctx->st0 = PEER_SESS_ST_END;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001685 goto switchstate;
1686 case PEER_SESS_ST_ERRSIZE: {
1687 unsigned char msg[2];
1688
1689 msg[0] = PEER_MSG_CLASS_ERROR;
1690 msg[1] = PEER_MSG_ERR_SIZELIMIT;
1691
Willy Tarreau06d80a92017-10-19 14:32:15 +02001692 if (ci_putblk(si_ic(si), (char *)msg, sizeof(msg)) == -1)
Emeric Brunb3971ab2015-05-12 18:49:09 +02001693 goto full;
1694 appctx->st0 = PEER_SESS_ST_END;
1695 goto switchstate;
1696 }
1697 case PEER_SESS_ST_ERRPROTO: {
1698 unsigned char msg[2];
1699
1700 msg[0] = PEER_MSG_CLASS_ERROR;
1701 msg[1] = PEER_MSG_ERR_PROTOCOL;
1702
Willy Tarreau06d80a92017-10-19 14:32:15 +02001703 if (ci_putblk(si_ic(si), (char *)msg, sizeof(msg)) == -1)
Emeric Brunb3971ab2015-05-12 18:49:09 +02001704 goto full;
1705 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02001706 /* fall through */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001707 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001708 case PEER_SESS_ST_END: {
Willy Tarreau73b013b2012-05-21 16:31:45 +02001709 si_shutw(si);
1710 si_shutr(si);
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001711 si_ic(si)->flags |= CF_READ_NULL;
Willy Tarreau828824a2015-04-19 17:20:03 +02001712 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001713 }
1714 }
1715 }
1716out:
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001717 si_oc(si)->flags |= CF_READ_DONTWAIT;
Emeric Brun2b920a12010-09-23 18:30:22 +02001718 return;
Willy Tarreaubc18da12015-03-13 14:00:47 +01001719full:
Willy Tarreaufe127932015-04-21 19:23:39 +02001720 si_applet_cant_put(si);
Willy Tarreaubc18da12015-03-13 14:00:47 +01001721 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001722}
1723
Willy Tarreau30576452015-04-13 13:50:30 +02001724static struct applet peer_applet = {
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001725 .obj_type = OBJ_TYPE_APPLET,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001726 .name = "<PEER>", /* used for logging */
1727 .fct = peer_io_handler,
Aman Gupta9a13e842012-04-02 18:57:53 -07001728 .release = peer_session_release,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001729};
Emeric Brun2b920a12010-09-23 18:30:22 +02001730
1731/*
1732 * Use this function to force a close of a peer session
1733 */
Willy Tarreau81bc3b02016-10-31 17:37:39 +01001734static void peer_session_forceshutdown(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +02001735{
Emeric Brunb3971ab2015-05-12 18:49:09 +02001736 struct peer *ps;
1737
Frédéric Lécaille5df11902017-06-13 16:39:57 +02001738 /* Note that the peer sessions which have just been created
1739 * (->st0 == PEER_SESS_ST_CONNECT) must not
1740 * be shutdown, if not, the TCP session will never be closed
1741 * and stay in CLOSE_WAIT state after having been closed by
1742 * the remote side.
1743 */
1744 if (!appctx || appctx->st0 == PEER_SESS_ST_CONNECT)
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001745 return;
1746
Willy Tarreau81bc3b02016-10-31 17:37:39 +01001747 if (appctx->applet != &peer_applet)
1748 return;
1749
Vincent Bernat3c2f2f22016-04-03 13:48:42 +02001750 ps = appctx->ctx.peers.ptr;
Willy Tarreaub4e34da2015-05-20 10:39:04 +02001751 /* we're killing a connection, we must apply a random delay before
1752 * retrying otherwise the other end will do the same and we can loop
1753 * for a while.
1754 */
1755 if (ps)
1756 ps->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + random() % 2000));
1757
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001758 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001759 appctx->ctx.peers.ptr = NULL;
Willy Tarreau78c0c502016-10-31 17:32:20 +01001760 appctx_wakeup(appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02001761}
1762
Willy Tarreau91d96282015-03-13 15:47:26 +01001763/* Pre-configures a peers frontend to accept incoming connections */
1764void peers_setup_frontend(struct proxy *fe)
1765{
1766 fe->last_change = now.tv_sec;
1767 fe->cap = PR_CAP_FE;
1768 fe->maxconn = 0;
1769 fe->conn_retries = CONN_RETRIES;
1770 fe->timeout.client = MS_TO_TICKS(5000);
Willy Tarreaud1d48d42015-03-13 16:15:46 +01001771 fe->accept = frontend_accept;
Willy Tarreauf87ab942015-03-13 15:55:16 +01001772 fe->default_target = &peer_applet.obj_type;
Willy Tarreau91d96282015-03-13 15:47:26 +01001773 fe->options2 |= PR_O2_INDEPSTR | PR_O2_SMARTCON | PR_O2_SMARTACC;
Willy Tarreau0fca4832015-05-01 19:12:05 +02001774 fe->bind_proc = 0; /* will be filled by users */
Willy Tarreau91d96282015-03-13 15:47:26 +01001775}
1776
Emeric Brun2b920a12010-09-23 18:30:22 +02001777/*
Willy Tarreaubd55e312010-11-11 10:55:09 +01001778 * Create a new peer session in assigned state (connect will start automatically)
Emeric Brun2b920a12010-09-23 18:30:22 +02001779 */
Willy Tarreau9df94c22016-10-31 18:42:52 +01001780static struct appctx *peer_session_create(struct peers *peers, struct peer *peer)
Emeric Brun2b920a12010-09-23 18:30:22 +02001781{
Willy Tarreau04b92862017-09-15 11:01:04 +02001782 struct proxy *p = peers->peers_fe; /* attached frontend */
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001783 struct appctx *appctx;
Willy Tarreau15b5e142015-04-04 14:38:25 +02001784 struct session *sess;
Willy Tarreau87b09662015-04-03 00:22:06 +02001785 struct stream *s;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001786 struct connection *conn;
Emeric Brun2b920a12010-09-23 18:30:22 +02001787
Emeric Brunb3971ab2015-05-12 18:49:09 +02001788 peer->reconnect = tick_add(now_ms, MS_TO_TICKS(5000));
1789 peer->statuscode = PEER_SESS_SC_CONNECTCODE;
Willy Tarreaud990baf2015-04-05 00:32:03 +02001790 s = NULL;
1791
1792 appctx = appctx_new(&peer_applet);
1793 if (!appctx)
1794 goto out_close;
1795
1796 appctx->st0 = PEER_SESS_ST_CONNECT;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001797 appctx->ctx.peers.ptr = (void *)peer;
Willy Tarreaud990baf2015-04-05 00:32:03 +02001798
Willy Tarreau04b92862017-09-15 11:01:04 +02001799 sess = session_new(p, NULL, &appctx->obj_type);
Willy Tarreau15b5e142015-04-04 14:38:25 +02001800 if (!sess) {
Godbach430f2912013-06-20 13:28:38 +08001801 Alert("out of memory in peer_session_create().\n");
Willy Tarreaud990baf2015-04-05 00:32:03 +02001802 goto out_free_appctx;
Emeric Brun2b920a12010-09-23 18:30:22 +02001803 }
1804
Willy Tarreau87787ac2017-08-28 16:22:54 +02001805 if ((s = stream_new(sess, &appctx->obj_type)) == NULL) {
Willy Tarreau342bfb12015-04-05 01:35:34 +02001806 Alert("Failed to initialize stream in peer_session_create().\n");
Willy Tarreau87787ac2017-08-28 16:22:54 +02001807 goto out_free_sess;
Willy Tarreau8baf9062015-04-05 00:46:36 +02001808 }
1809
Willy Tarreau342bfb12015-04-05 01:35:34 +02001810 /* The tasks below are normally what is supposed to be done by
1811 * fe->accept().
1812 */
Willy Tarreaue7dff022015-04-03 01:14:29 +02001813 s->flags = SF_ASSIGNED|SF_ADDR_SET;
Emeric Brun2b920a12010-09-23 18:30:22 +02001814
Willy Tarreau6e2979c2015-04-27 13:21:15 +02001815 /* applet is waiting for data */
1816 si_applet_cant_get(&s->si[0]);
1817 appctx_wakeup(appctx);
1818
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001819 /* initiate an outgoing connection */
1820 si_set_state(&s->si[1], SI_ST_ASS);
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001821
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001822 /* automatically prepare the stream interface to connect to the
Willy Tarreaub363a1f2013-10-01 10:45:07 +02001823 * pre-initialized connection in si->conn.
1824 */
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001825 if (unlikely((conn = conn_new()) == NULL))
Willy Tarreau8baf9062015-04-05 00:46:36 +02001826 goto out_free_strm;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001827
1828 conn_prepare(conn, peer->proto, peer->xprt);
1829 si_attach_conn(&s->si[1], conn);
1830
1831 conn->target = s->target = &s->be->obj_type;
1832 memcpy(&conn->addr.to, &peer->addr, sizeof(conn->addr.to));
Emeric Brun2b920a12010-09-23 18:30:22 +02001833 s->do_log = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001834 s->uniq_id = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +02001835
Willy Tarreau22ec1ea2014-11-27 20:45:39 +01001836 s->res.flags |= CF_READ_DONTWAIT;
Willy Tarreau696a2912014-11-24 11:36:57 +01001837
Emeric Brunb3971ab2015-05-12 18:49:09 +02001838 peer->appctx = appctx;
Willy Tarreau87787ac2017-08-28 16:22:54 +02001839 task_wakeup(s->task, TASK_WOKEN_INIT);
Willy Tarreau9df94c22016-10-31 18:42:52 +01001840 return appctx;
Emeric Brun2b920a12010-09-23 18:30:22 +02001841
1842 /* Error unrolling */
Willy Tarreau15b5e142015-04-04 14:38:25 +02001843 out_free_strm:
Emeric Brun2b920a12010-09-23 18:30:22 +02001844 LIST_DEL(&s->list);
Willy Tarreau87b09662015-04-03 00:22:06 +02001845 pool_free2(pool2_stream, s);
Willy Tarreau15b5e142015-04-04 14:38:25 +02001846 out_free_sess:
Willy Tarreau11c36242015-04-04 15:54:03 +02001847 session_free(sess);
Willy Tarreaud990baf2015-04-05 00:32:03 +02001848 out_free_appctx:
1849 appctx_free(appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02001850 out_close:
Willy Tarreaub21d08e2016-10-31 17:46:57 +01001851 return NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001852}
1853
1854/*
1855 * Task processing function to manage re-connect and peer session
1856 * tasks wakeup on local update.
1857 */
Simon Horman96553772011-06-08 09:18:51 +09001858static struct task *process_peer_sync(struct task * task)
Emeric Brun2b920a12010-09-23 18:30:22 +02001859{
Vincent Bernat3c2f2f22016-04-03 13:48:42 +02001860 struct peers *peers = task->context;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001861 struct peer *ps;
1862 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +02001863
1864 task->expire = TICK_ETERNITY;
1865
Emeric Brunb3971ab2015-05-12 18:49:09 +02001866 if (!peers->peers_fe) {
Willy Tarreau46dc1ca2015-05-01 18:32:13 +02001867 /* this one was never started, kill it */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001868 signal_unregister_handler(peers->sighandler);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001869 task_delete(peers->sync_task);
1870 task_free(peers->sync_task);
Willy Tarreau37bb7be2015-09-21 15:24:58 +02001871 peers->sync_task = NULL;
Willy Tarreau46dc1ca2015-05-01 18:32:13 +02001872 return NULL;
1873 }
1874
Emeric Brun2b920a12010-09-23 18:30:22 +02001875 if (!stopping) {
1876 /* Normal case (not soft stop)*/
Emeric Brunb3971ab2015-05-12 18:49:09 +02001877
1878 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL) &&
1879 (!nb_oldpids || tick_is_expired(peers->resync_timeout, now_ms)) &&
1880 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001881 /* Resync from local peer needed
1882 no peer was assigned for the lesson
1883 and no old local peer found
1884 or resync timeout expire */
1885
1886 /* flag no more resync from local, to try resync from remotes */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001887 peers->flags |= PEERS_F_RESYNC_LOCAL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001888
1889 /* reschedule a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001890 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
Emeric Brun2b920a12010-09-23 18:30:22 +02001891 }
1892
1893 /* For each session */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001894 for (ps = peers->remote; ps; ps = ps->next) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001895 /* For each remote peers */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001896 if (!ps->local) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01001897 if (!ps->appctx) {
1898 /* no active peer connection */
Emeric Brun2b920a12010-09-23 18:30:22 +02001899 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001900 ((ps->statuscode == PEER_SESS_SC_CONNECTCODE ||
Willy Tarreaub4e34da2015-05-20 10:39:04 +02001901 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001902 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02001903 tick_is_expired(ps->reconnect, now_ms))) {
1904 /* connection never tried
Willy Tarreau9df94c22016-10-31 18:42:52 +01001905 * or previous peer connection established with success
1906 * or previous peer connection failed while connecting
Emeric Brun2b920a12010-09-23 18:30:22 +02001907 * and reconnection timer is expired */
1908
1909 /* retry a connect */
Willy Tarreau9df94c22016-10-31 18:42:52 +01001910 ps->appctx = peer_session_create(peers, ps);
Emeric Brun2b920a12010-09-23 18:30:22 +02001911 }
Willy Tarreaub4e34da2015-05-20 10:39:04 +02001912 else if (!tick_is_expired(ps->reconnect, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001913 /* If previous session failed during connection
1914 * but reconnection timer is not expired */
1915
1916 /* reschedule task for reconnect */
1917 task->expire = tick_first(task->expire, ps->reconnect);
1918 }
1919 /* else do nothing */
Willy Tarreau9df94c22016-10-31 18:42:52 +01001920 } /* !ps->appctx */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001921 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01001922 /* current peer connection is active and established */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001923 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
1924 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02001925 !(ps->flags & PEER_F_LEARN_NOTUP2DATE)) {
1926 /* Resync from a remote is needed
1927 * and no peer was assigned for lesson
1928 * and current peer may be up2date */
1929
1930 /* assign peer for the lesson */
1931 ps->flags |= PEER_F_LEARN_ASSIGN;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001932 peers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brun2b920a12010-09-23 18:30:22 +02001933
Willy Tarreau9df94c22016-10-31 18:42:52 +01001934 /* wake up peer handler to handle a request of resync */
Willy Tarreaue5843b32015-04-27 18:40:14 +02001935 appctx_wakeup(ps->appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02001936 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001937 else {
1938 /* Awake session if there is data to push */
1939 for (st = ps->tables; st ; st = st->next) {
1940 if ((int)(st->last_pushed - st->table->localupdate) < 0) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01001941 /* wake up the peer handler to push local updates */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001942 appctx_wakeup(ps->appctx);
1943 break;
1944 }
1945 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001946 }
1947 /* else do nothing */
1948 } /* SUCCESSCODE */
1949 } /* !ps->peer->local */
1950 } /* for */
1951
1952 /* Resync from remotes expired: consider resync is finished */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001953 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
1954 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
1955 tick_is_expired(peers->resync_timeout, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001956 /* Resync from remote peer needed
1957 * no peer was assigned for the lesson
1958 * and resync timeout expire */
1959
1960 /* flag no more resync from remote, consider resync is finished */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001961 peers->flags |= PEERS_F_RESYNC_REMOTE;
Emeric Brun2b920a12010-09-23 18:30:22 +02001962 }
1963
Emeric Brunb3971ab2015-05-12 18:49:09 +02001964 if ((peers->flags & PEERS_RESYNC_STATEMASK) != PEERS_RESYNC_FINISHED) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001965 /* Resync not finished*/
Frédéric Lécaille5d6e5f82017-05-29 13:47:16 +02001966 /* reschedule task to resync timeout if not expired, to ended resync if needed */
1967 if (!tick_is_expired(peers->resync_timeout, now_ms))
1968 task->expire = tick_first(task->expire, peers->resync_timeout);
Emeric Brun2b920a12010-09-23 18:30:22 +02001969 }
1970 } /* !stopping */
1971 else {
1972 /* soft stop case */
1973 if (task->state & TASK_WOKEN_SIGNAL) {
1974 /* We've just recieved the signal */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001975 if (!(peers->flags & PEERS_F_DONOTSTOP)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001976 /* add DO NOT STOP flag if not present */
1977 jobs++;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001978 peers->flags |= PEERS_F_DONOTSTOP;
1979 ps = peers->local;
1980 for (st = ps->tables; st ; st = st->next)
1981 st->table->syncing++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001982 }
1983
1984 /* disconnect all connected peers */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001985 for (ps = peers->remote; ps; ps = ps->next) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01001986 if (ps->appctx) {
Willy Tarreau81bc3b02016-10-31 17:37:39 +01001987 peer_session_forceshutdown(ps->appctx);
Willy Tarreaue5843b32015-04-27 18:40:14 +02001988 ps->appctx = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001989 }
1990 }
1991 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001992
Emeric Brunb3971ab2015-05-12 18:49:09 +02001993 ps = peers->local;
Emeric Brun2b920a12010-09-23 18:30:22 +02001994 if (ps->flags & PEER_F_TEACH_COMPLETE) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001995 if (peers->flags & PEERS_F_DONOTSTOP) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001996 /* resync of new process was complete, current process can die now */
1997 jobs--;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001998 peers->flags &= ~PEERS_F_DONOTSTOP;
1999 for (st = ps->tables; st ; st = st->next)
2000 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02002001 }
2002 }
Willy Tarreau9df94c22016-10-31 18:42:52 +01002003 else if (!ps->appctx) {
2004 /* If there's no active peer connection */
Emeric Brun2b920a12010-09-23 18:30:22 +02002005 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01002006 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
2007 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE ||
2008 ps->statuscode == PEER_SESS_SC_TRYAGAIN) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002009 /* connection never tried
Willy Tarreau9df94c22016-10-31 18:42:52 +01002010 * or previous peer connection was successfully established
2011 * or previous tcp connect succeeded but init state incomplete
Emeric Brun2b920a12010-09-23 18:30:22 +02002012 * or during previous connect, peer replies a try again statuscode */
2013
2014 /* connect to the peer */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002015 peer_session_create(peers, ps);
Emeric Brun2b920a12010-09-23 18:30:22 +02002016 }
2017 else {
2018 /* Other error cases */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002019 if (peers->flags & PEERS_F_DONOTSTOP) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002020 /* unable to resync new process, current process can die now */
2021 jobs--;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002022 peers->flags &= ~PEERS_F_DONOTSTOP;
2023 for (st = ps->tables; st ; st = st->next)
2024 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02002025 }
2026 }
2027 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02002028 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE ) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002029 /* current peer connection is active and established
2030 * wake up all peer handlers to push remaining local updates */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002031 for (st = ps->tables; st ; st = st->next) {
2032 if ((int)(st->last_pushed - st->table->localupdate) < 0) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02002033 appctx_wakeup(ps->appctx);
2034 break;
2035 }
2036 }
Emeric Brun2b920a12010-09-23 18:30:22 +02002037 }
2038 } /* stopping */
2039 /* Wakeup for re-connect */
2040 return task;
2041}
2042
Emeric Brunb3971ab2015-05-12 18:49:09 +02002043
Emeric Brun2b920a12010-09-23 18:30:22 +02002044/*
Emeric Brun2b920a12010-09-23 18:30:22 +02002045 *
2046 */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002047void peers_init_sync(struct peers *peers)
Emeric Brun2b920a12010-09-23 18:30:22 +02002048{
Emeric Brun2b920a12010-09-23 18:30:22 +02002049 struct peer * curpeer;
Willy Tarreau4348fad2012-09-20 16:48:07 +02002050 struct listener *listener;
Emeric Brun2b920a12010-09-23 18:30:22 +02002051
Emeric Brun2b920a12010-09-23 18:30:22 +02002052 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002053 peers->peers_fe->maxconn += 3;
2054 }
2055
Willy Tarreau4348fad2012-09-20 16:48:07 +02002056 list_for_each_entry(listener, &peers->peers_fe->conf.listeners, by_fe)
2057 listener->maxconn = peers->peers_fe->maxconn;
Emeric Brunc60def82017-09-27 14:59:38 +02002058 peers->sync_task = task_new(MAX_THREADS_MASK);
Emeric Brunb3971ab2015-05-12 18:49:09 +02002059 peers->sync_task->process = process_peer_sync;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002060 peers->sync_task->context = (void *)peers;
2061 peers->sighandler = signal_register_task(0, peers->sync_task, 0);
2062 task_wakeup(peers->sync_task, TASK_WOKEN_INIT);
2063}
2064
2065
2066
2067/*
2068 * Function used to register a table for sync on a group of peers
2069 *
2070 */
2071void peers_register_table(struct peers *peers, struct stktable *table)
2072{
2073 struct shared_table *st;
2074 struct peer * curpeer;
2075 int id = 0;
2076
2077 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
Vincent Bernat02779b62016-04-03 13:48:43 +02002078 st = calloc(1,sizeof(*st));
Emeric Brunb3971ab2015-05-12 18:49:09 +02002079 st->table = table;
2080 st->next = curpeer->tables;
2081 if (curpeer->tables)
2082 id = curpeer->tables->local_id;
2083 st->local_id = id + 1;
2084
2085 curpeer->tables = st;
2086 }
2087
2088 table->sync_task = peers->sync_task;
Emeric Brun2b920a12010-09-23 18:30:22 +02002089}
2090