blob: 0c8861f3969662f9c3e906168c837519905532a2 [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
174struct peers *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;
386 char *cursor, *datamsg;
387 uint64_t data = 0;
Emeric Brun94900952015-06-11 18:25:54 +0200388 unsigned int data_type;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200389
390 cursor = datamsg = msg + 2 + 5;
391
392 /* Encode data */
393
394 /* encode local id */
395 intencode(st->local_id, &cursor);
396
397 /* encode table name */
398 len = strlen(st->table->id);
399 intencode(len, &cursor);
400 memcpy(cursor, st->table->id, len);
401 cursor += len;
402
403 /* encode table type */
404
405 intencode(st->table->type, &cursor);
406
407 /* encode table key size */
408 intencode(st->table->key_size, &cursor);
409
Emeric Brun94900952015-06-11 18:25:54 +0200410 /* encode available known data types in table */
411 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
412 if (st->table->data_ofs[data_type]) {
413 switch (stktable_data_types[data_type].std_type) {
414 case STD_T_SINT:
415 case STD_T_UINT:
416 case STD_T_ULL:
417 case STD_T_FRQP:
418 data |= 1 << data_type;
419 break;
420 }
421 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200422 }
423 intencode(data, &cursor);
424
425 /* Compute datalen */
426 datalen = (cursor - datamsg);
Emeric Brun2b920a12010-09-23 18:30:22 +0200427
Emeric Brunb3971ab2015-05-12 18:49:09 +0200428 /* prepare message header */
429 msg[0] = PEER_MSG_CLASS_STICKTABLE;
430 msg[1] = PEER_MSG_STKT_DEFINE;
431 cursor = &msg[2];
432 intencode(datalen, &cursor);
Emeric Brun2b920a12010-09-23 18:30:22 +0200433
Emeric Brunb3971ab2015-05-12 18:49:09 +0200434 /* move data after header */
435 memmove(cursor, datamsg, datalen);
436
437 /* return header size + data_len */
438 return (cursor - msg) + datalen;
Emeric Brun2b920a12010-09-23 18:30:22 +0200439}
440
Emeric Brunb3971ab2015-05-12 18:49:09 +0200441/*
442 * This prepare the acknowledge message on the stick session <ts>, <st> is the considered
443 * stick table.
444 * <msg> is a buffer of <size> to recieve data message content
445 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
446 * check size)
447 */
448static int peer_prepare_ackmsg(struct shared_table *st, char *msg, size_t size)
449{
450 unsigned short datalen;
451 char *cursor, *datamsg;
452 uint32_t netinteger;
453
Emeric Brunb058f1c2015-09-22 15:50:18 +0200454 cursor = datamsg = msg + 2 + 5;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200455
456 intencode(st->remote_id, &cursor);
457 netinteger = htonl(st->last_get);
458 memcpy(cursor, &netinteger, sizeof(netinteger));
459 cursor += sizeof(netinteger);
460
461 /* Compute datalen */
462 datalen = (cursor - datamsg);
463
464 /* prepare message header */
465 msg[0] = PEER_MSG_CLASS_STICKTABLE;
Emeric Brune1ab8082015-08-21 11:48:54 +0200466 msg[1] = PEER_MSG_STKT_ACK;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200467 cursor = &msg[2];
468 intencode(datalen, &cursor);
469
470 /* move data after header */
471 memmove(cursor, datamsg, datalen);
472
473 /* return header size + data_len */
474 return (cursor - msg) + datalen;
475}
Emeric Brun2b920a12010-09-23 18:30:22 +0200476
477/*
478 * Callback to release a session with a peer
479 */
Willy Tarreau00a37f02015-04-13 12:05:19 +0200480static void peer_session_release(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +0200481{
Willy Tarreau00a37f02015-04-13 12:05:19 +0200482 struct stream_interface *si = appctx->owner;
Willy Tarreau87b09662015-04-03 00:22:06 +0200483 struct stream *s = si_strm(si);
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200484 struct peer *peer = appctx->ctx.peers.ptr;
485 struct peers *peers = strm_fe(s)->parent;
Emeric Brun2b920a12010-09-23 18:30:22 +0200486
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100487 /* appctx->ctx.peers.ptr is not a peer session */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100488 if (appctx->st0 < PEER_SESS_ST_SENDSUCCESS)
Emeric Brun2b920a12010-09-23 18:30:22 +0200489 return;
490
491 /* peer session identified */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200492 if (peer) {
Willy Tarreau9df94c22016-10-31 18:42:52 +0100493 if (peer->appctx == appctx) {
Emeric Brunb157d732015-08-21 12:00:30 +0200494 /* Re-init current table pointers to force announcement on re-connect */
495 peer->remote_table = peer->last_local_table = NULL;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200496 peer->appctx = NULL;
497 if (peer->flags & PEER_F_LEARN_ASSIGN) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200498 /* unassign current peer for learning */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200499 peer->flags &= ~(PEER_F_LEARN_ASSIGN);
500 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brun2b920a12010-09-23 18:30:22 +0200501
502 /* reschedule a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200503 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
Emeric Brun2b920a12010-09-23 18:30:22 +0200504 }
505 /* reset teaching and learning flags to 0 */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200506 peer->flags &= PEER_TEACH_RESET;
507 peer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200508 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200509 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200510 }
511}
512
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200513/* Retrieve the major and minor versions of peers protocol
514 * announced by a remote peer. <str> is a null-terminated
515 * string with the following format: "<maj_ver>.<min_ver>".
516 */
517static int peer_get_version(const char *str,
518 unsigned int *maj_ver, unsigned int *min_ver)
519{
520 unsigned int majv, minv;
521 const char *pos, *saved;
522 const char *end;
523
524 saved = pos = str;
525 end = str + strlen(str);
526
527 majv = read_uint(&pos, end);
528 if (saved == pos || *pos++ != '.')
529 return -1;
530
531 saved = pos;
532 minv = read_uint(&pos, end);
533 if (saved == pos || pos != end)
534 return -1;
535
536 *maj_ver = majv;
537 *min_ver = minv;
538
539 return 0;
540}
Emeric Brun2b920a12010-09-23 18:30:22 +0200541
542/*
543 * IO Handler to handle message exchance with a peer
544 */
Willy Tarreau00a37f02015-04-13 12:05:19 +0200545static void peer_io_handler(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +0200546{
Willy Tarreau00a37f02015-04-13 12:05:19 +0200547 struct stream_interface *si = appctx->owner;
Willy Tarreau87b09662015-04-03 00:22:06 +0200548 struct stream *s = si_strm(si);
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200549 struct peers *curpeers = strm_fe(s)->parent;
Emeric Brun2b920a12010-09-23 18:30:22 +0200550 int reql = 0;
551 int repl = 0;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200552 size_t proto_len = strlen(PEER_SESSION_PROTO_NAME);
553 unsigned int maj_ver, min_ver;
Emeric Brun2b920a12010-09-23 18:30:22 +0200554
Christopher Fauleta73e59b2016-12-09 17:30:18 +0100555 /* Check if the input buffer is avalaible. */
556 if (si_ic(si)->buf->size == 0)
557 goto full;
558
Emeric Brun2b920a12010-09-23 18:30:22 +0200559 while (1) {
560switchstate:
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200561 maj_ver = min_ver = (unsigned int)-1;
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100562 switch(appctx->st0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100563 case PEER_SESS_ST_ACCEPT:
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100564 appctx->ctx.peers.ptr = NULL;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100565 appctx->st0 = PEER_SESS_ST_GETVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200566 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100567 case PEER_SESS_ST_GETVERSION:
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100568 reql = bo_getline(si_oc(si), trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200569 if (reql <= 0) { /* closed or EOL not found */
570 if (reql == 0)
571 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100572 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200573 goto switchstate;
574 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100575 if (trash.str[reql-1] != '\n') {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100576 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200577 goto switchstate;
578 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100579 else if (reql > 1 && (trash.str[reql-2] == '\r'))
580 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200581 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100582 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200583
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100584 bo_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200585
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200586 /* test protocol */
587 if (strncmp(PEER_SESSION_PROTO_NAME " ", trash.str, proto_len + 1) != 0) {
588 appctx->st0 = PEER_SESS_ST_EXIT;
589 appctx->st1 = PEER_SESS_SC_ERRPROTO;
590 goto switchstate;
591 }
592 if (peer_get_version(trash.str + proto_len + 1, &maj_ver, &min_ver) == -1 ||
593 maj_ver != PEER_MAJOR_VER || min_ver > PEER_MINOR_VER) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100594 appctx->st0 = PEER_SESS_ST_EXIT;
595 appctx->st1 = PEER_SESS_SC_ERRVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200596 goto switchstate;
597 }
598
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100599 appctx->st0 = PEER_SESS_ST_GETHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200600 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100601 case PEER_SESS_ST_GETHOST:
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100602 reql = bo_getline(si_oc(si), trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200603 if (reql <= 0) { /* closed or EOL not found */
604 if (reql == 0)
605 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100606 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200607 goto switchstate;
608 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100609 if (trash.str[reql-1] != '\n') {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100610 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200611 goto switchstate;
612 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100613 else if (reql > 1 && (trash.str[reql-2] == '\r'))
614 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200615 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100616 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200617
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100618 bo_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200619
620 /* test hostname match */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100621 if (strcmp(localpeer, trash.str) != 0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100622 appctx->st0 = PEER_SESS_ST_EXIT;
623 appctx->st1 = PEER_SESS_SC_ERRHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200624 goto switchstate;
625 }
626
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100627 appctx->st0 = PEER_SESS_ST_GETPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200628 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100629 case PEER_SESS_ST_GETPEER: {
Emeric Brun2b920a12010-09-23 18:30:22 +0200630 struct peer *curpeer;
631 char *p;
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100632 reql = bo_getline(si_oc(si), trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200633 if (reql <= 0) { /* closed or EOL not found */
634 if (reql == 0)
635 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100636 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200637 goto switchstate;
638 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100639 if (trash.str[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200640 /* Incomplete line, we quit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100641 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200642 goto switchstate;
643 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100644 else if (reql > 1 && (trash.str[reql-2] == '\r'))
645 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200646 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100647 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200648
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100649 bo_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200650
Emeric Brunb3971ab2015-05-12 18:49:09 +0200651 /* parse line "<peer name> <pid> <relative_pid>" */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100652 p = strchr(trash.str, ' ');
Emeric Brun2b920a12010-09-23 18:30:22 +0200653 if (!p) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100654 appctx->st0 = PEER_SESS_ST_EXIT;
655 appctx->st1 = PEER_SESS_SC_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +0200656 goto switchstate;
657 }
658 *p = 0;
659
660 /* lookup known peer */
661 for (curpeer = curpeers->remote; curpeer; curpeer = curpeer->next) {
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100662 if (strcmp(curpeer->id, trash.str) == 0)
Emeric Brun2b920a12010-09-23 18:30:22 +0200663 break;
664 }
665
666 /* if unknown peer */
667 if (!curpeer) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100668 appctx->st0 = PEER_SESS_ST_EXIT;
669 appctx->st1 = PEER_SESS_SC_ERRPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200670 goto switchstate;
671 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200672
Willy Tarreau9df94c22016-10-31 18:42:52 +0100673 if (curpeer->appctx && curpeer->appctx != appctx) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200674 if (curpeer->local) {
675 /* Local connection, reply a retry */
676 appctx->st0 = PEER_SESS_ST_EXIT;
677 appctx->st1 = PEER_SESS_SC_TRYAGAIN;
678 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +0200679 }
Willy Tarreau81bc3b02016-10-31 17:37:39 +0100680 peer_session_forceshutdown(curpeer->appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +0200681 }
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200682 if (maj_ver != (unsigned int)-1 && min_ver != (unsigned int)-1) {
683 if (min_ver == PEER_DWNGRD_MINOR_VER) {
684 curpeer->flags |= PEER_F_DWNGRD;
685 }
686 else {
687 curpeer->flags &= ~PEER_F_DWNGRD;
688 }
689 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200690 curpeer->appctx = appctx;
691 appctx->ctx.peers.ptr = curpeer;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100692 appctx->st0 = PEER_SESS_ST_SENDSUCCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200693 /* fall through */
694 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100695 case PEER_SESS_ST_SENDSUCCESS: {
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200696 struct peer *curpeer = appctx->ctx.peers.ptr;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200697 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +0200698
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100699 repl = snprintf(trash.str, trash.size, "%d\n", PEER_SESS_SC_SUCCESSCODE);
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100700 repl = bi_putblk(si_ic(si), trash.str, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200701 if (repl <= 0) {
702 if (repl == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +0100703 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100704 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200705 goto switchstate;
706 }
707
708 /* Register status code */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200709 curpeer->statuscode = PEER_SESS_SC_SUCCESSCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200710
711 /* Awake main task */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200712 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200713
714 /* Init confirm counter */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200715 curpeer->confirm = 0;
716
717 /* Init cursors */
718 for (st = curpeer->tables; st ; st = st->next) {
719 st->last_get = st->last_acked = 0;
720 st->teaching_origin = st->last_pushed = st->update;
721 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200722
723 /* reset teaching and learning flags to 0 */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200724 curpeer->flags &= PEER_TEACH_RESET;
725 curpeer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200726
727 /* if current peer is local */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200728 if (curpeer->local) {
729 /* if current host need resyncfrom local and no process assined */
730 if ((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL &&
731 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
732 /* assign local peer for a lesson, consider lesson already requested */
733 curpeer->flags |= PEER_F_LEARN_ASSIGN;
734 peers->flags |= (PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
735 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200736
Emeric Brunb3971ab2015-05-12 18:49:09 +0200737 }
738 else if ((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
739 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
740 /* assign peer for a lesson */
741 curpeer->flags |= PEER_F_LEARN_ASSIGN;
742 peers->flags |= PEERS_F_RESYNC_ASSIGN;
743 }
744
745
Emeric Brun2b920a12010-09-23 18:30:22 +0200746 /* switch to waiting message state */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100747 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200748 goto switchstate;
749 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100750 case PEER_SESS_ST_CONNECT: {
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200751 struct peer *curpeer = appctx->ctx.peers.ptr;
Emeric Brun2b920a12010-09-23 18:30:22 +0200752
753 /* Send headers */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100754 repl = snprintf(trash.str, trash.size,
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200755 PEER_SESSION_PROTO_NAME " %u.%u\n%s\n%s %d %d\n",
756 PEER_MAJOR_VER,
757 (curpeer->flags & PEER_F_DWNGRD) ? PEER_DWNGRD_MINOR_VER : PEER_MINOR_VER,
Emeric Brunb3971ab2015-05-12 18:49:09 +0200758 curpeer->id,
Emeric Brun2b920a12010-09-23 18:30:22 +0200759 localpeer,
Willy Tarreau7b77c9f2012-01-07 22:52:12 +0100760 (int)getpid(),
Emeric Brunb3971ab2015-05-12 18:49:09 +0200761 relative_pid);
Emeric Brun2b920a12010-09-23 18:30:22 +0200762
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100763 if (repl >= trash.size) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100764 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200765 goto switchstate;
766 }
767
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100768 repl = bi_putblk(si_ic(si), trash.str, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200769 if (repl <= 0) {
770 if (repl == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +0100771 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100772 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200773 goto switchstate;
774 }
775
776 /* switch to the waiting statuscode state */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100777 appctx->st0 = PEER_SESS_ST_GETSTATUS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200778 /* fall through */
779 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100780 case PEER_SESS_ST_GETSTATUS: {
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200781 struct peer *curpeer = appctx->ctx.peers.ptr;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200782 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +0200783
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100784 if (si_ic(si)->flags & CF_WRITE_PARTIAL)
Emeric Brunb3971ab2015-05-12 18:49:09 +0200785 curpeer->statuscode = PEER_SESS_SC_CONNECTEDCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200786
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100787 reql = bo_getline(si_oc(si), trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200788 if (reql <= 0) { /* closed or EOL not found */
789 if (reql == 0)
790 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100791 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200792 goto switchstate;
793 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100794 if (trash.str[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200795 /* Incomplete line, we quit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100796 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200797 goto switchstate;
798 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100799 else if (reql > 1 && (trash.str[reql-2] == '\r'))
800 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200801 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100802 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200803
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100804 bo_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200805
806 /* Register status code */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200807 curpeer->statuscode = atoi(trash.str);
Emeric Brun2b920a12010-09-23 18:30:22 +0200808
809 /* Awake main task */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200810 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200811
812 /* If status code is success */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200813 if (curpeer->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200814 /* Init cursors */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200815 for (st = curpeer->tables; st ; st = st->next) {
816 st->last_get = st->last_acked = 0;
817 st->teaching_origin = st->last_pushed = st->update;
818 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200819
820 /* Init confirm counter */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200821 curpeer->confirm = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200822
Emeric Brunb3971ab2015-05-12 18:49:09 +0200823 /* reset teaching and learning flags to 0 */
824 curpeer->flags &= PEER_TEACH_RESET;
825 curpeer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200826
Emeric Brunb3971ab2015-05-12 18:49:09 +0200827 /* If current peer is local */
828 if (curpeer->local) {
829 /* flag to start to teach lesson */
830 curpeer->flags |= PEER_F_TEACH_PROCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200831
Emeric Brunb3971ab2015-05-12 18:49:09 +0200832 }
833 else if ((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
834 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
835 /* If peer is remote and resync from remote is needed,
836 and no peer currently assigned */
Emeric Brun2b920a12010-09-23 18:30:22 +0200837
Emeric Brunb3971ab2015-05-12 18:49:09 +0200838 /* assign peer for a lesson */
839 curpeer->flags |= PEER_F_LEARN_ASSIGN;
840 peers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brun2b920a12010-09-23 18:30:22 +0200841 }
842
843 }
844 else {
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +0200845 if (curpeer->statuscode == PEER_SESS_SC_ERRVERSION)
846 curpeer->flags |= PEER_F_DWNGRD;
Emeric Brun2b920a12010-09-23 18:30:22 +0200847 /* Status code is not success, abort */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100848 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200849 goto switchstate;
850 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100851 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200852 /* fall through */
853 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100854 case PEER_SESS_ST_WAITMSG: {
Vincent Bernat3c2f2f22016-04-03 13:48:42 +0200855 struct peer *curpeer = appctx->ctx.peers.ptr;
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200856 struct stksess *ts, *newts = NULL;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200857 uint32_t msg_len = 0;
858 char *msg_cur = trash.str;
859 char *msg_end = trash.str;
860 unsigned char msg_head[7];
Emeric Brun2b920a12010-09-23 18:30:22 +0200861 int totl = 0;
862
Emeric Brunb3971ab2015-05-12 18:49:09 +0200863 reql = bo_getblk(si_oc(si), (char *)msg_head, 2*sizeof(unsigned char), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200864 if (reql <= 0) /* closed or EOL not found */
865 goto incomplete;
866
Emeric Brun2b920a12010-09-23 18:30:22 +0200867 totl += reql;
868
Emeric Brunb3971ab2015-05-12 18:49:09 +0200869 if (msg_head[1] >= 128) {
870 /* Read and Decode message length */
871 reql = bo_getblk(si_oc(si), (char *)&msg_head[2], sizeof(unsigned char), totl);
872 if (reql <= 0) /* closed */
873 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +0200874
Emeric Brunb3971ab2015-05-12 18:49:09 +0200875 totl += reql;
876
877 if (msg_head[2] < 240) {
878 msg_len = msg_head[2];
Emeric Brun2b920a12010-09-23 18:30:22 +0200879 }
880 else {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200881 int i;
882 char *cur;
883 char *end;
Willy Tarreau72d6c162013-04-11 16:14:13 +0200884
Emeric Brunb3971ab2015-05-12 18:49:09 +0200885 for (i = 3 ; i < sizeof(msg_head) ; i++) {
886 reql = bo_getblk(si_oc(si), (char *)&msg_head[i], sizeof(char), totl);
887 if (reql <= 0) /* closed */
888 goto incomplete;
889
890 totl += reql;
891
892 if (!(msg_head[i] & 0x80))
893 break;
894 }
895
896 if (i == sizeof(msg_head)) {
897 /* malformed message */
898 appctx->st0 = PEER_SESS_ST_ERRPROTO;
899 goto switchstate;
900
901 }
902 end = (char *)msg_head + sizeof(msg_head);
903 cur = (char *)&msg_head[2];
904 msg_len = intdecode(&cur, end);
905 if (!cur) {
906 /* malformed message */
907 appctx->st0 = PEER_SESS_ST_ERRPROTO;
908 goto switchstate;
909 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200910 }
911
Willy Tarreau86a446e2013-11-25 23:02:37 +0100912
Emeric Brunb3971ab2015-05-12 18:49:09 +0200913 /* Read message content */
914 if (msg_len) {
915 if (msg_len > trash.size) {
916 /* Status code is not success, abort */
917 appctx->st0 = PEER_SESS_ST_ERRSIZE;
918 goto switchstate;
919 }
Willy Tarreau72d6c162013-04-11 16:14:13 +0200920
Emeric Brunb3971ab2015-05-12 18:49:09 +0200921 reql = bo_getblk(si_oc(si), trash.str, msg_len, totl);
922 if (reql <= 0) /* closed */
923 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +0200924 totl += reql;
Willy Tarreau86a446e2013-11-25 23:02:37 +0100925
Emeric Brunb3971ab2015-05-12 18:49:09 +0200926 msg_end += msg_len;
927 }
928 }
Willy Tarreau86a446e2013-11-25 23:02:37 +0100929
Emeric Brunb3971ab2015-05-12 18:49:09 +0200930 if (msg_head[0] == PEER_MSG_CLASS_CONTROL) {
931 if (msg_head[1] == PEER_MSG_CTRL_RESYNCREQ) {
932 struct shared_table *st;
933 /* Reset message: remote need resync */
934
935 /* prepare tables fot a global push */
936 for (st = curpeer->tables; st; st = st->next) {
937 st->teaching_origin = st->last_pushed = st->table->update;
938 st->flags = 0;
Willy Tarreau86a446e2013-11-25 23:02:37 +0100939 }
Willy Tarreau72d6c162013-04-11 16:14:13 +0200940
Emeric Brunb3971ab2015-05-12 18:49:09 +0200941 /* reset teaching flags to 0 */
942 curpeer->flags &= PEER_TEACH_RESET;
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200943
Emeric Brunb3971ab2015-05-12 18:49:09 +0200944 /* flag to start to teach lesson */
945 curpeer->flags |= PEER_F_TEACH_PROCESS;
946
947
948 }
949 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCFINISHED) {
950
951 if (curpeer->flags & PEER_F_LEARN_ASSIGN) {
952 curpeer->flags &= ~PEER_F_LEARN_ASSIGN;
953 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
954 peers->flags |= (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE);
Willy Tarreau86a446e2013-11-25 23:02:37 +0100955 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200956 curpeer->confirm++;
Emeric Brun2b920a12010-09-23 18:30:22 +0200957 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200958 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCPARTIAL) {
959
960 if (curpeer->flags & PEER_F_LEARN_ASSIGN) {
961 curpeer->flags &= ~PEER_F_LEARN_ASSIGN;
962 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
963
964 curpeer->flags |= PEER_F_LEARN_NOTUP2DATE;
965 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
966 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
Cyril Bonté9a60ff92014-02-16 01:07:07 +0100967 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200968 curpeer->confirm++;
Emeric Brun2b920a12010-09-23 18:30:22 +0200969 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200970 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCCONFIRM) {
Emeric Brun597b26e2016-08-12 11:23:31 +0200971 struct shared_table *st;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200972
973 /* If stopping state */
974 if (stopping) {
975 /* Close session, push resync no more needed */
976 curpeer->flags |= PEER_F_TEACH_COMPLETE;
977 appctx->st0 = PEER_SESS_ST_END;
978 goto switchstate;
979 }
Emeric Brun597b26e2016-08-12 11:23:31 +0200980 for (st = curpeer->tables; st; st = st->next) {
981 st->update = st->last_pushed = st->teaching_origin;
982 st->flags = 0;
983 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200984
985 /* reset teaching flags to 0 */
986 curpeer->flags &= PEER_TEACH_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200987 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200988 }
989 else if (msg_head[0] == PEER_MSG_CLASS_STICKTABLE) {
990 if (msg_head[1] == PEER_MSG_STKT_DEFINE) {
991 int table_id_len;
992 struct shared_table *st;
993 int table_type;
994 int table_keylen;
995 int table_id;
996 uint64_t table_data;
Emeric Brun2b920a12010-09-23 18:30:22 +0200997
Emeric Brunb3971ab2015-05-12 18:49:09 +0200998 table_id = intdecode(&msg_cur, msg_end);
999 if (!msg_cur) {
1000 /* malformed message */
1001 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1002 goto switchstate;
1003 }
Willy Tarreau72d6c162013-04-11 16:14:13 +02001004
Emeric Brunb3971ab2015-05-12 18:49:09 +02001005 table_id_len = intdecode(&msg_cur, msg_end);
1006 if (!msg_cur) {
1007 /* malformed message */
1008 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1009 goto switchstate;
1010 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001011
Emeric Brunb3971ab2015-05-12 18:49:09 +02001012 curpeer->remote_table = NULL;
1013 if (!table_id_len || (msg_cur + table_id_len) >= msg_end) {
1014 /* malformed message */
1015 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1016 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001017 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001018
Emeric Brunb3971ab2015-05-12 18:49:09 +02001019 for (st = curpeer->tables; st; st = st->next) {
1020 /* Reset IDs */
1021 if (st->remote_id == table_id)
1022 st->remote_id = 0;
Willy Tarreau9d9179b2013-04-11 16:56:44 +02001023
Emeric Brunb3971ab2015-05-12 18:49:09 +02001024 if (!curpeer->remote_table
1025 && (table_id_len == strlen(st->table->id))
1026 && (memcmp(st->table->id, msg_cur, table_id_len) == 0)) {
1027 curpeer->remote_table = st;
Emeric Brun2b920a12010-09-23 18:30:22 +02001028 }
1029 }
1030
Emeric Brunb3971ab2015-05-12 18:49:09 +02001031 if (!curpeer->remote_table) {
1032 goto ignore_msg;
1033 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001034
Emeric Brunb3971ab2015-05-12 18:49:09 +02001035 msg_cur += table_id_len;
1036 if (msg_cur >= msg_end) {
1037 /* malformed message */
1038 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1039 goto switchstate;
1040 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001041
Emeric Brunb3971ab2015-05-12 18:49:09 +02001042 table_type = intdecode(&msg_cur, msg_end);
1043 if (!msg_cur) {
1044 /* malformed message */
1045 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1046 goto switchstate;
1047 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001048
Emeric Brunb3971ab2015-05-12 18:49:09 +02001049 table_keylen = intdecode(&msg_cur, msg_end);
1050 if (!msg_cur) {
1051 /* malformed message */
1052 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1053 goto switchstate;
1054 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001055
Emeric Brunb3971ab2015-05-12 18:49:09 +02001056 table_data = intdecode(&msg_cur, msg_end);
1057 if (!msg_cur) {
1058 /* malformed message */
1059 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1060 goto switchstate;
1061 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001062
Emeric Brunb3971ab2015-05-12 18:49:09 +02001063 if (curpeer->remote_table->table->type != table_type
1064 || curpeer->remote_table->table->key_size != table_keylen) {
1065 curpeer->remote_table = NULL;
1066 goto ignore_msg;
1067 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001068
Emeric Brunb3971ab2015-05-12 18:49:09 +02001069 curpeer->remote_table->remote_data = table_data;
1070 curpeer->remote_table->remote_id = table_id;
Emeric Brun2b920a12010-09-23 18:30:22 +02001071 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001072 else if (msg_head[1] == PEER_MSG_STKT_SWITCH) {
1073 struct shared_table *st;
1074 int table_id;
Emeric Brun2b920a12010-09-23 18:30:22 +02001075
Emeric Brunb3971ab2015-05-12 18:49:09 +02001076 table_id = intdecode(&msg_cur, msg_end);
1077 if (!msg_cur) {
1078 /* malformed message */
1079 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1080 goto switchstate;
1081 }
1082 curpeer->remote_table = NULL;
1083 for (st = curpeer->tables; st; st = st->next) {
1084 if (st->remote_id == table_id) {
1085 curpeer->remote_table = st;
1086 break;
1087 }
1088 }
1089
Emeric Brun2b920a12010-09-23 18:30:22 +02001090 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001091 else if (msg_head[1] == PEER_MSG_STKT_UPDATE
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001092 || msg_head[1] == PEER_MSG_STKT_INCUPDATE
1093 || msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED
1094 || msg_head[1] == PEER_MSG_STKT_INCUPDATE_TIMED) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001095 struct shared_table *st = curpeer->remote_table;
1096 uint32_t update;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001097 int expire;
Emeric Brun94900952015-06-11 18:25:54 +02001098 unsigned int data_type;
1099 void *data_ptr;
Emeric Brun2b920a12010-09-23 18:30:22 +02001100
Emeric Brunb3971ab2015-05-12 18:49:09 +02001101 /* Here we have data message */
1102 if (!st)
1103 goto ignore_msg;
Emeric Brun2b920a12010-09-23 18:30:22 +02001104
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001105 expire = MS_TO_TICKS(st->table->expire);
1106
1107 if (msg_head[1] == PEER_MSG_STKT_UPDATE ||
1108 msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001109 if (msg_len < sizeof(update)) {
1110 /* malformed message */
1111 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1112 goto switchstate;
1113 }
1114 memcpy(&update, msg_cur, sizeof(update));
1115 msg_cur += sizeof(update);
1116 st->last_get = htonl(update);
1117 }
1118 else {
1119 st->last_get++;
1120 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001121
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001122 if (msg_head[1] == PEER_MSG_STKT_UPDATE_TIMED ||
1123 msg_head[1] == PEER_MSG_STKT_INCUPDATE_TIMED) {
1124 size_t expire_sz = sizeof expire;
1125
1126 if (msg_cur + expire_sz > msg_end) {
1127 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1128 goto switchstate;
1129 }
1130 memcpy(&expire, msg_cur, expire_sz);
1131 msg_cur += expire_sz;
1132 expire = ntohl(expire);
1133 }
1134
Emeric Brunb3971ab2015-05-12 18:49:09 +02001135 newts = stksess_new(st->table, NULL);
1136 if (!newts)
1137 goto ignore_msg;
Emeric Brun2b920a12010-09-23 18:30:22 +02001138
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +02001139 if (st->table->type == SMP_T_STR) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001140 unsigned int to_read, to_store;
Emeric Brun2b920a12010-09-23 18:30:22 +02001141
Emeric Brunb3971ab2015-05-12 18:49:09 +02001142 to_read = intdecode(&msg_cur, msg_end);
1143 if (!msg_cur) {
1144 /* malformed message */
1145 stksess_free(st->table, newts);
1146 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1147 goto switchstate;
1148 }
1149 to_store = MIN(to_read, st->table->key_size - 1);
1150 if (msg_cur + to_store > msg_end) {
1151 /* malformed message */
1152 stksess_free(st->table, newts);
1153 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1154 goto switchstate;
1155 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001156
Emeric Brunb3971ab2015-05-12 18:49:09 +02001157 memcpy(newts->key.key, msg_cur, to_store);
1158 newts->key.key[to_store] = 0;
1159 msg_cur += to_read;
1160 }
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +02001161 else if (st->table->type == SMP_T_SINT) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001162 unsigned int netinteger;
Willy Tarreau72d6c162013-04-11 16:14:13 +02001163
Emeric Brunb3971ab2015-05-12 18:49:09 +02001164 if (msg_cur + sizeof(netinteger) > msg_end) {
1165 /* malformed message */
1166 stksess_free(st->table, newts);
1167 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1168 goto switchstate;
1169 }
1170 memcpy(&netinteger, msg_cur, sizeof(netinteger));
1171 netinteger = ntohl(netinteger);
1172 memcpy(newts->key.key, &netinteger, sizeof(netinteger));
1173 msg_cur += sizeof(netinteger);
1174 }
1175 else {
1176 if (msg_cur + st->table->key_size > msg_end) {
1177 /* malformed message */
1178 stksess_free(st->table, newts);
1179 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1180 goto switchstate;
1181 }
1182 memcpy(newts->key.key, msg_cur, st->table->key_size);
1183 msg_cur += st->table->key_size;
1184 }
1185
1186 /* lookup for existing entry */
1187 ts = stktable_lookup(st->table, newts);
1188 if (ts) {
1189 /* the entry already exist, we can free ours */
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001190 stktable_touch_with_exp(st->table, ts, 0, tick_add(now_ms, expire));
Emeric Brunb3971ab2015-05-12 18:49:09 +02001191 stksess_free(st->table, newts);
1192 newts = NULL;
1193 }
1194 else {
1195 struct eb32_node *eb;
1196
1197 /* create new entry */
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001198 ts = stktable_store_with_exp(st->table, newts, 0, tick_add(now_ms, expire));
Emeric Brunb3971ab2015-05-12 18:49:09 +02001199 newts = NULL; /* don't reuse it */
1200
Emeric Brun234fc3c2015-12-16 15:16:46 +01001201 ts->upd.key= (++st->table->update)+(2147483648U);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001202 eb = eb32_insert(&st->table->updates, &ts->upd);
1203 if (eb != &ts->upd) {
1204 eb32_delete(eb);
1205 eb32_insert(&st->table->updates, &ts->upd);
1206 }
1207 }
1208
Emeric Brun94900952015-06-11 18:25:54 +02001209 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001210
Emeric Brun94900952015-06-11 18:25:54 +02001211 if ((1 << data_type) & st->remote_data) {
1212 switch (stktable_data_types[data_type].std_type) {
1213 case STD_T_SINT: {
1214 int data;
1215
1216 data = intdecode(&msg_cur, msg_end);
1217 if (!msg_cur) {
1218 /* malformed message */
1219 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1220 goto switchstate;
1221 }
1222
1223 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1224 if (data_ptr)
1225 stktable_data_cast(data_ptr, std_t_sint) = data;
1226 break;
1227 }
1228 case STD_T_UINT: {
1229 unsigned int data;
1230
1231 data = intdecode(&msg_cur, msg_end);
1232 if (!msg_cur) {
1233 /* malformed message */
1234 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1235 goto switchstate;
1236 }
1237
1238 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1239 if (data_ptr)
1240 stktable_data_cast(data_ptr, std_t_uint) = data;
1241 break;
1242 }
1243 case STD_T_ULL: {
1244 unsigned long long data;
1245
1246 data = intdecode(&msg_cur, msg_end);
1247 if (!msg_cur) {
1248 /* malformed message */
1249 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1250 goto switchstate;
1251 }
1252
1253 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1254 if (data_ptr)
1255 stktable_data_cast(data_ptr, std_t_ull) = data;
1256 break;
1257 }
1258 case STD_T_FRQP: {
1259 struct freq_ctr_period data;
1260
Willy Tarreau3bb46172016-03-25 18:17:47 +01001261 data.curr_tick = tick_add(now_ms, -intdecode(&msg_cur, msg_end));
Emeric Brun94900952015-06-11 18:25:54 +02001262 if (!msg_cur) {
1263 /* malformed message */
1264 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1265 goto switchstate;
1266 }
1267 data.curr_ctr = intdecode(&msg_cur, msg_end);
1268 if (!msg_cur) {
1269 /* malformed message */
1270 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1271 goto switchstate;
1272 }
1273 data.prev_ctr = intdecode(&msg_cur, msg_end);
1274 if (!msg_cur) {
1275 /* malformed message */
1276 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1277 goto switchstate;
1278 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001279
Emeric Brun94900952015-06-11 18:25:54 +02001280 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1281 if (data_ptr)
1282 stktable_data_cast(data_ptr, std_t_frqp) = data;
1283 break;
1284 }
1285 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001286 }
1287 }
1288 }
1289 else if (msg_head[1] == PEER_MSG_STKT_ACK) {
1290 /* ack message */
1291 uint32_t table_id ;
1292 uint32_t update;
1293 struct shared_table *st;
1294
1295 table_id = intdecode(&msg_cur, msg_end);
1296 if (!msg_cur || (msg_cur + sizeof(update) > msg_end)) {
1297 /* malformed message */
1298 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1299 goto switchstate;
1300 }
1301 memcpy(&update, msg_cur, sizeof(update));
1302 update = ntohl(update);
Emeric Brun2b920a12010-09-23 18:30:22 +02001303
Emeric Brunb3971ab2015-05-12 18:49:09 +02001304 for (st = curpeer->tables; st; st = st->next) {
1305 if (st->local_id == table_id) {
1306 st->update = update;
1307 break;
1308 }
1309 }
1310 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001311 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001312 else if (msg_head[0] == PEER_MSG_CLASS_RESERVED) {
1313 appctx->st0 = PEER_SESS_ST_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +02001314 goto switchstate;
1315 }
1316
Emeric Brunb3971ab2015-05-12 18:49:09 +02001317ignore_msg:
Emeric Brun2b920a12010-09-23 18:30:22 +02001318 /* skip consumed message */
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001319 bo_skip(si_oc(si), totl);
Emeric Brun2b920a12010-09-23 18:30:22 +02001320 /* loop on that state to peek next message */
Willy Tarreau72d6c162013-04-11 16:14:13 +02001321 goto switchstate;
1322
Emeric Brun2b920a12010-09-23 18:30:22 +02001323incomplete:
Willy Tarreau9d9179b2013-04-11 16:56:44 +02001324 /* we get here when a bo_getblk() returns <= 0 in reql */
1325
Willy Tarreau72d6c162013-04-11 16:14:13 +02001326 if (reql < 0) {
1327 /* there was an error */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001328 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau72d6c162013-04-11 16:14:13 +02001329 goto switchstate;
1330 }
1331
Emeric Brun2b920a12010-09-23 18:30:22 +02001332
Emeric Brun2b920a12010-09-23 18:30:22 +02001333
Emeric Brunb3971ab2015-05-12 18:49:09 +02001334
Emeric Brun2b920a12010-09-23 18:30:22 +02001335 /* Need to request a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001336 if ((curpeer->flags & PEER_F_LEARN_ASSIGN) &&
1337 (peers->flags & PEERS_F_RESYNC_ASSIGN) &&
1338 !(peers->flags & PEERS_F_RESYNC_PROCESS)) {
1339 unsigned char msg[2];
Emeric Brun2b920a12010-09-23 18:30:22 +02001340
Emeric Brunb3971ab2015-05-12 18:49:09 +02001341 /* Current peer was elected to request a resync */
1342 msg[0] = PEER_MSG_CLASS_CONTROL;
1343 msg[1] = PEER_MSG_CTRL_RESYNCREQ;
Emeric Brun2b920a12010-09-23 18:30:22 +02001344
Emeric Brunb3971ab2015-05-12 18:49:09 +02001345 /* message to buffer */
1346 repl = bi_putblk(si_ic(si), (char *)msg, sizeof(msg));
1347 if (repl <= 0) {
1348 /* no more write possible */
1349 if (repl == -1)
1350 goto full;
1351 appctx->st0 = PEER_SESS_ST_END;
1352 goto switchstate;
1353 }
1354 peers->flags |= PEERS_F_RESYNC_PROCESS;
1355 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001356
Emeric Brunb3971ab2015-05-12 18:49:09 +02001357 /* Nothing to read, now we start to write */
Emeric Brun2b920a12010-09-23 18:30:22 +02001358
Emeric Brunb3971ab2015-05-12 18:49:09 +02001359 if (curpeer->tables) {
1360 struct shared_table *st;
1361 struct shared_table *last_local_table;
Emeric Brun2b920a12010-09-23 18:30:22 +02001362
Emeric Brunb3971ab2015-05-12 18:49:09 +02001363 last_local_table = curpeer->last_local_table;
1364 if (!last_local_table)
1365 last_local_table = curpeer->tables;
1366 st = last_local_table->next;
Emeric Brun2b920a12010-09-23 18:30:22 +02001367
Emeric Brunb3971ab2015-05-12 18:49:09 +02001368 while (1) {
1369 if (!st)
1370 st = curpeer->tables;
Emeric Brun2b920a12010-09-23 18:30:22 +02001371
Emeric Brunb3971ab2015-05-12 18:49:09 +02001372 /* It remains some updates to ack */
1373 if (st->last_get != st->last_acked) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001374 int msglen;
Emeric Brun2b920a12010-09-23 18:30:22 +02001375
Emeric Brunb3971ab2015-05-12 18:49:09 +02001376 msglen = peer_prepare_ackmsg(st, trash.str, trash.size);
1377 if (!msglen) {
1378 /* internal error: message does not fit in trash */
1379 appctx->st0 = PEER_SESS_ST_END;
1380 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001381 }
1382
Emeric Brunb3971ab2015-05-12 18:49:09 +02001383 /* message to buffer */
1384 repl = bi_putblk(si_ic(si), trash.str, msglen);
1385 if (repl <= 0) {
1386 /* no more write possible */
1387 if (repl == -1) {
1388 goto full;
Emeric Brun2b920a12010-09-23 18:30:22 +02001389 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001390 appctx->st0 = PEER_SESS_ST_END;
1391 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001392 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001393 st->last_acked = st->last_get;
Emeric Brun2b920a12010-09-23 18:30:22 +02001394 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001395
Emeric Brunb3971ab2015-05-12 18:49:09 +02001396 if (!(curpeer->flags & PEER_F_TEACH_PROCESS)) {
1397 if (!(curpeer->flags & PEER_F_LEARN_ASSIGN) &&
1398 ((int)(st->last_pushed - st->table->localupdate) < 0)) {
1399 struct eb32_node *eb;
1400 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001401
Emeric Brunb3971ab2015-05-12 18:49:09 +02001402 if (st != curpeer->last_local_table) {
1403 int msglen;
Emeric Brun2b920a12010-09-23 18:30:22 +02001404
Emeric Brunb3971ab2015-05-12 18:49:09 +02001405 msglen = peer_prepare_switchmsg(st, trash.str, trash.size);
1406 if (!msglen) {
1407 /* internal error: message does not fit in trash */
1408 appctx->st0 = PEER_SESS_ST_END;
1409 goto switchstate;
1410 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001411
Emeric Brunb3971ab2015-05-12 18:49:09 +02001412 /* message to buffer */
1413 repl = bi_putblk(si_ic(si), trash.str, msglen);
1414 if (repl <= 0) {
1415 /* no more write possible */
1416 if (repl == -1) {
1417 goto full;
1418 }
1419 appctx->st0 = PEER_SESS_ST_END;
1420 goto switchstate;
1421 }
1422 curpeer->last_local_table = st;
Emeric Brun2b920a12010-09-23 18:30:22 +02001423 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001424
1425 /* We force new pushed to 1 to force identifier in update message */
1426 new_pushed = 1;
1427 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
1428 while (1) {
1429 uint32_t msglen;
1430 struct stksess *ts;
1431
1432 /* push local updates */
1433 if (!eb) {
1434 eb = eb32_first(&st->table->updates);
1435 if (!eb || ((int)(eb->key - st->last_pushed) <= 0)) {
Emeric Brunaaf58602015-06-15 17:23:30 +02001436 st->table->commitupdate = st->last_pushed = st->table->localupdate;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001437 break;
1438 }
1439 }
1440
1441 if ((int)(eb->key - st->table->localupdate) > 0) {
Emeric Brunaaf58602015-06-15 17:23:30 +02001442 st->table->commitupdate = st->last_pushed = st->table->localupdate;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001443 break;
1444 }
1445
1446 ts = eb32_entry(eb, struct stksess, upd);
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001447 msglen = peer_prepare_updatemsg(ts, st, trash.str, trash.size, new_pushed, 0);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001448 if (!msglen) {
1449 /* internal error: message does not fit in trash */
1450 appctx->st0 = PEER_SESS_ST_END;
1451 goto switchstate;
1452 }
1453
1454 /* message to buffer */
1455 repl = bi_putblk(si_ic(si), trash.str, msglen);
1456 if (repl <= 0) {
1457 /* no more write possible */
1458 if (repl == -1) {
1459 goto full;
1460 }
1461 appctx->st0 = PEER_SESS_ST_END;
1462 goto switchstate;
1463 }
1464 st->last_pushed = ts->upd.key;
Emeric Brunaaf58602015-06-15 17:23:30 +02001465 if ((int)(st->last_pushed - st->table->commitupdate) > 0)
1466 st->table->commitupdate = st->last_pushed;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001467 /* identifier may not needed in next update message */
1468 new_pushed = 0;
1469
1470 eb = eb32_next(eb);
1471 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001472 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001473 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001474 else {
1475 if (!(st->flags & SHTABLE_F_TEACH_STAGE1)) {
1476 struct eb32_node *eb;
1477 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001478
Emeric Brunb3971ab2015-05-12 18:49:09 +02001479 if (st != curpeer->last_local_table) {
1480 int msglen;
1481
1482 msglen = peer_prepare_switchmsg(st, trash.str, trash.size);
1483 if (!msglen) {
1484 /* internal error: message does not fit in trash */
1485 appctx->st0 = PEER_SESS_ST_END;
1486 goto switchstate;
1487 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001488
Emeric Brunb3971ab2015-05-12 18:49:09 +02001489 /* message to buffer */
1490 repl = bi_putblk(si_ic(si), trash.str, msglen);
1491 if (repl <= 0) {
1492 /* no more write possible */
1493 if (repl == -1) {
1494 goto full;
1495 }
1496 appctx->st0 = PEER_SESS_ST_END;
1497 goto switchstate;
1498 }
1499 curpeer->last_local_table = st;
1500 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001501
Emeric Brunb3971ab2015-05-12 18:49:09 +02001502 /* We force new pushed to 1 to force identifier in update message */
1503 new_pushed = 1;
1504 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
1505 while (1) {
1506 uint32_t msglen;
1507 struct stksess *ts;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001508 int use_timed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001509
Emeric Brunb3971ab2015-05-12 18:49:09 +02001510 /* push local updates */
1511 if (!eb) {
1512 st->flags |= SHTABLE_F_TEACH_STAGE1;
1513 eb = eb32_first(&st->table->updates);
1514 if (eb)
1515 st->last_pushed = eb->key - 1;
1516 break;
1517 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001518
Emeric Brunb3971ab2015-05-12 18:49:09 +02001519 ts = eb32_entry(eb, struct stksess, upd);
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001520 use_timed = !(curpeer->flags & PEER_F_DWNGRD);
1521 msglen = peer_prepare_updatemsg(ts, st, trash.str, trash.size, new_pushed, use_timed);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001522 if (!msglen) {
1523 /* internal error: message does not fit in trash */
1524 appctx->st0 = PEER_SESS_ST_END;
1525 goto switchstate;
1526 }
1527
1528 /* message to buffer */
1529 repl = bi_putblk(si_ic(si), trash.str, msglen);
1530 if (repl <= 0) {
1531 /* no more write possible */
1532 if (repl == -1) {
1533 goto full;
1534 }
1535 appctx->st0 = PEER_SESS_ST_END;
1536 goto switchstate;
1537 }
1538 st->last_pushed = ts->upd.key;
1539 /* identifier may not needed in next update message */
1540 new_pushed = 0;
1541
1542 eb = eb32_next(eb);
1543 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001544 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001545
Emeric Brunb3971ab2015-05-12 18:49:09 +02001546 if (!(st->flags & SHTABLE_F_TEACH_STAGE2)) {
1547 struct eb32_node *eb;
1548 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001549
Emeric Brunb3971ab2015-05-12 18:49:09 +02001550 if (st != curpeer->last_local_table) {
1551 int msglen;
1552
1553 msglen = peer_prepare_switchmsg(st, trash.str, trash.size);
1554 if (!msglen) {
1555 /* internal error: message does not fit in trash */
1556 appctx->st0 = PEER_SESS_ST_END;
1557 goto switchstate;
1558 }
1559
1560 /* message to buffer */
1561 repl = bi_putblk(si_ic(si), trash.str, msglen);
1562 if (repl <= 0) {
1563 /* no more write possible */
1564 if (repl == -1) {
1565 goto full;
1566 }
1567 appctx->st0 = PEER_SESS_ST_END;
1568 goto switchstate;
1569 }
1570 curpeer->last_local_table = st;
1571 }
1572
1573 /* We force new pushed to 1 to force identifier in update message */
1574 new_pushed = 1;
1575 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
1576 while (1) {
1577 uint32_t msglen;
1578 struct stksess *ts;
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001579 int use_timed;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001580
1581 /* push local updates */
1582 if (!eb || eb->key > st->teaching_origin) {
1583 st->flags |= SHTABLE_F_TEACH_STAGE2;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001584 break;
1585 }
1586
1587 ts = eb32_entry(eb, struct stksess, upd);
Frédéric Lécaille523cc9e2016-10-12 17:30:30 +02001588 use_timed = !(curpeer->flags & PEER_F_DWNGRD);
1589 msglen = peer_prepare_updatemsg(ts, st, trash.str, trash.size, new_pushed, use_timed);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001590 if (!msglen) {
1591 /* internal error: message does not fit in trash */
1592 appctx->st0 = PEER_SESS_ST_END;
1593 goto switchstate;
1594 }
1595
1596 /* message to buffer */
1597 repl = bi_putblk(si_ic(si), trash.str, msglen);
1598 if (repl <= 0) {
1599 /* no more write possible */
1600 if (repl == -1) {
1601 goto full;
1602 }
1603 appctx->st0 = PEER_SESS_ST_END;
1604 goto switchstate;
1605 }
1606 st->last_pushed = ts->upd.key;
1607 /* identifier may not needed in next update message */
1608 new_pushed = 0;
1609
1610 eb = eb32_next(eb);
1611 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001612 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001613 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001614
1615 if (st == last_local_table)
1616 break;
1617 st = st->next;
Emeric Brun2b920a12010-09-23 18:30:22 +02001618 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001619 }
1620
1621
1622 if ((curpeer->flags & PEER_F_TEACH_PROCESS) && !(curpeer->flags & PEER_F_TEACH_FINISHED)) {
1623 unsigned char msg[2];
1624
1625 /* Current peer was elected to request a resync */
1626 msg[0] = PEER_MSG_CLASS_CONTROL;
1627 msg[1] = ((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FINISHED) ? PEER_MSG_CTRL_RESYNCFINISHED : PEER_MSG_CTRL_RESYNCPARTIAL;
1628 /* process final lesson message */
1629 repl = bi_putblk(si_ic(si), (char *)msg, sizeof(msg));
1630 if (repl <= 0) {
1631 /* no more write possible */
1632 if (repl == -1)
1633 goto full;
1634 appctx->st0 = PEER_SESS_ST_END;
1635 goto switchstate;
1636 }
1637 /* flag finished message sent */
1638 curpeer->flags |= PEER_F_TEACH_FINISHED;
1639 }
1640
Emeric Brun597b26e2016-08-12 11:23:31 +02001641 /* Confirm finished or partial messages */
1642 while (curpeer->confirm) {
1643 unsigned char msg[2];
1644
1645 /* There is a confirm messages to send */
1646 msg[0] = PEER_MSG_CLASS_CONTROL;
1647 msg[1] = PEER_MSG_CTRL_RESYNCCONFIRM;
1648
1649 /* message to buffer */
1650 repl = bi_putblk(si_ic(si), (char *)msg, sizeof(msg));
1651 if (repl <= 0) {
1652 /* no more write possible */
1653 if (repl == -1)
1654 goto full;
1655 appctx->st0 = PEER_SESS_ST_END;
1656 goto switchstate;
1657 }
1658 curpeer->confirm--;
1659 }
1660
Emeric Brun2b920a12010-09-23 18:30:22 +02001661 /* noting more to do */
1662 goto out;
1663 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001664 case PEER_SESS_ST_EXIT:
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001665 repl = snprintf(trash.str, trash.size, "%d\n", appctx->st1);
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001666 if (bi_putblk(si_ic(si), trash.str, repl) == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +01001667 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001668 appctx->st0 = PEER_SESS_ST_END;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001669 goto switchstate;
1670 case PEER_SESS_ST_ERRSIZE: {
1671 unsigned char msg[2];
1672
1673 msg[0] = PEER_MSG_CLASS_ERROR;
1674 msg[1] = PEER_MSG_ERR_SIZELIMIT;
1675
1676 if (bi_putblk(si_ic(si), (char *)msg, sizeof(msg)) == -1)
1677 goto full;
1678 appctx->st0 = PEER_SESS_ST_END;
1679 goto switchstate;
1680 }
1681 case PEER_SESS_ST_ERRPROTO: {
1682 unsigned char msg[2];
1683
1684 msg[0] = PEER_MSG_CLASS_ERROR;
1685 msg[1] = PEER_MSG_ERR_PROTOCOL;
1686
1687 if (bi_putblk(si_ic(si), (char *)msg, sizeof(msg)) == -1)
1688 goto full;
1689 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02001690 /* fall through */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001691 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001692 case PEER_SESS_ST_END: {
Willy Tarreau73b013b2012-05-21 16:31:45 +02001693 si_shutw(si);
1694 si_shutr(si);
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001695 si_ic(si)->flags |= CF_READ_NULL;
Willy Tarreau828824a2015-04-19 17:20:03 +02001696 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001697 }
1698 }
1699 }
1700out:
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001701 si_oc(si)->flags |= CF_READ_DONTWAIT;
Emeric Brun2b920a12010-09-23 18:30:22 +02001702 return;
Willy Tarreaubc18da12015-03-13 14:00:47 +01001703full:
Willy Tarreaufe127932015-04-21 19:23:39 +02001704 si_applet_cant_put(si);
Willy Tarreaubc18da12015-03-13 14:00:47 +01001705 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001706}
1707
Willy Tarreau30576452015-04-13 13:50:30 +02001708static struct applet peer_applet = {
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001709 .obj_type = OBJ_TYPE_APPLET,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001710 .name = "<PEER>", /* used for logging */
1711 .fct = peer_io_handler,
Aman Gupta9a13e842012-04-02 18:57:53 -07001712 .release = peer_session_release,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001713};
Emeric Brun2b920a12010-09-23 18:30:22 +02001714
1715/*
1716 * Use this function to force a close of a peer session
1717 */
Willy Tarreau81bc3b02016-10-31 17:37:39 +01001718static void peer_session_forceshutdown(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +02001719{
Emeric Brunb3971ab2015-05-12 18:49:09 +02001720 struct peer *ps;
1721
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001722 if (!appctx)
1723 return;
1724
Willy Tarreau81bc3b02016-10-31 17:37:39 +01001725 if (appctx->applet != &peer_applet)
1726 return;
1727
Vincent Bernat3c2f2f22016-04-03 13:48:42 +02001728 ps = appctx->ctx.peers.ptr;
Willy Tarreaub4e34da2015-05-20 10:39:04 +02001729 /* we're killing a connection, we must apply a random delay before
1730 * retrying otherwise the other end will do the same and we can loop
1731 * for a while.
1732 */
1733 if (ps)
1734 ps->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + random() % 2000));
1735
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001736 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001737 appctx->ctx.peers.ptr = NULL;
Willy Tarreau78c0c502016-10-31 17:32:20 +01001738 appctx_wakeup(appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02001739}
1740
Willy Tarreau91d96282015-03-13 15:47:26 +01001741/* Pre-configures a peers frontend to accept incoming connections */
1742void peers_setup_frontend(struct proxy *fe)
1743{
1744 fe->last_change = now.tv_sec;
1745 fe->cap = PR_CAP_FE;
1746 fe->maxconn = 0;
1747 fe->conn_retries = CONN_RETRIES;
1748 fe->timeout.client = MS_TO_TICKS(5000);
Willy Tarreaud1d48d42015-03-13 16:15:46 +01001749 fe->accept = frontend_accept;
Willy Tarreauf87ab942015-03-13 15:55:16 +01001750 fe->default_target = &peer_applet.obj_type;
Willy Tarreau91d96282015-03-13 15:47:26 +01001751 fe->options2 |= PR_O2_INDEPSTR | PR_O2_SMARTCON | PR_O2_SMARTACC;
Willy Tarreau0fca4832015-05-01 19:12:05 +02001752 fe->bind_proc = 0; /* will be filled by users */
Willy Tarreau91d96282015-03-13 15:47:26 +01001753}
1754
Emeric Brun2b920a12010-09-23 18:30:22 +02001755/*
Willy Tarreaubd55e312010-11-11 10:55:09 +01001756 * Create a new peer session in assigned state (connect will start automatically)
Emeric Brun2b920a12010-09-23 18:30:22 +02001757 */
Willy Tarreau9df94c22016-10-31 18:42:52 +01001758static struct appctx *peer_session_create(struct peers *peers, struct peer *peer)
Emeric Brun2b920a12010-09-23 18:30:22 +02001759{
Emeric Brunb3971ab2015-05-12 18:49:09 +02001760 struct listener *l = LIST_NEXT(&peers->peers_fe->conf.listeners, struct listener *, by_fe);
Willy Tarreauc95bad52016-12-22 00:13:31 +01001761 struct proxy *p = l->bind_conf->frontend; /* attached frontend */
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001762 struct appctx *appctx;
Willy Tarreau15b5e142015-04-04 14:38:25 +02001763 struct session *sess;
Willy Tarreau87b09662015-04-03 00:22:06 +02001764 struct stream *s;
Emeric Brun2b920a12010-09-23 18:30:22 +02001765 struct task *t;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001766 struct connection *conn;
Emeric Brun2b920a12010-09-23 18:30:22 +02001767
Emeric Brunb3971ab2015-05-12 18:49:09 +02001768 peer->reconnect = tick_add(now_ms, MS_TO_TICKS(5000));
1769 peer->statuscode = PEER_SESS_SC_CONNECTCODE;
Willy Tarreaud990baf2015-04-05 00:32:03 +02001770 s = NULL;
1771
1772 appctx = appctx_new(&peer_applet);
1773 if (!appctx)
1774 goto out_close;
1775
1776 appctx->st0 = PEER_SESS_ST_CONNECT;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001777 appctx->ctx.peers.ptr = (void *)peer;
Willy Tarreaud990baf2015-04-05 00:32:03 +02001778
Willy Tarreau4099a7c2015-04-05 00:39:55 +02001779 sess = session_new(p, l, &appctx->obj_type);
Willy Tarreau15b5e142015-04-04 14:38:25 +02001780 if (!sess) {
Godbach430f2912013-06-20 13:28:38 +08001781 Alert("out of memory in peer_session_create().\n");
Willy Tarreaud990baf2015-04-05 00:32:03 +02001782 goto out_free_appctx;
Emeric Brun2b920a12010-09-23 18:30:22 +02001783 }
1784
Willy Tarreau8baf9062015-04-05 00:46:36 +02001785 if ((t = task_new()) == NULL) {
Willy Tarreau15b5e142015-04-04 14:38:25 +02001786 Alert("out of memory in peer_session_create().\n");
1787 goto out_free_sess;
1788 }
Willy Tarreau8baf9062015-04-05 00:46:36 +02001789 t->nice = l->nice;
1790
Willy Tarreau73b65ac2015-04-08 18:26:29 +02001791 if ((s = stream_new(sess, t, &appctx->obj_type)) == NULL) {
Willy Tarreau342bfb12015-04-05 01:35:34 +02001792 Alert("Failed to initialize stream in peer_session_create().\n");
Willy Tarreau8baf9062015-04-05 00:46:36 +02001793 goto out_free_task;
1794 }
1795
Willy Tarreau342bfb12015-04-05 01:35:34 +02001796 /* The tasks below are normally what is supposed to be done by
1797 * fe->accept().
1798 */
Willy Tarreaue7dff022015-04-03 01:14:29 +02001799 s->flags = SF_ASSIGNED|SF_ADDR_SET;
Emeric Brun2b920a12010-09-23 18:30:22 +02001800
Willy Tarreau6e2979c2015-04-27 13:21:15 +02001801 /* applet is waiting for data */
1802 si_applet_cant_get(&s->si[0]);
1803 appctx_wakeup(appctx);
1804
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001805 /* initiate an outgoing connection */
1806 si_set_state(&s->si[1], SI_ST_ASS);
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001807
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001808 /* automatically prepare the stream interface to connect to the
Willy Tarreaub363a1f2013-10-01 10:45:07 +02001809 * pre-initialized connection in si->conn.
1810 */
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001811 if (unlikely((conn = conn_new()) == NULL))
Willy Tarreau8baf9062015-04-05 00:46:36 +02001812 goto out_free_strm;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001813
1814 conn_prepare(conn, peer->proto, peer->xprt);
1815 si_attach_conn(&s->si[1], conn);
1816
1817 conn->target = s->target = &s->be->obj_type;
1818 memcpy(&conn->addr.to, &peer->addr, sizeof(conn->addr.to));
Emeric Brun2b920a12010-09-23 18:30:22 +02001819 s->do_log = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001820 s->uniq_id = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +02001821
Willy Tarreau22ec1ea2014-11-27 20:45:39 +01001822 s->res.flags |= CF_READ_DONTWAIT;
Willy Tarreau696a2912014-11-24 11:36:57 +01001823
Emeric Brun2b920a12010-09-23 18:30:22 +02001824 l->nbconn++; /* warning! right now, it's up to the handler to decrease this */
1825 p->feconn++;/* beconn will be increased later */
1826 jobs++;
Willy Tarreaufb0afa72015-04-03 14:46:27 +02001827 if (!(s->sess->listener->options & LI_O_UNLIMITED))
Willy Tarreau3c63fd82011-09-07 18:00:47 +02001828 actconn++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001829 totalconn++;
1830
Emeric Brunb3971ab2015-05-12 18:49:09 +02001831 peer->appctx = appctx;
Willy Tarreau9df94c22016-10-31 18:42:52 +01001832 return appctx;
Emeric Brun2b920a12010-09-23 18:30:22 +02001833
1834 /* Error unrolling */
Willy Tarreau15b5e142015-04-04 14:38:25 +02001835 out_free_strm:
Emeric Brun2b920a12010-09-23 18:30:22 +02001836 LIST_DEL(&s->list);
Willy Tarreau87b09662015-04-03 00:22:06 +02001837 pool_free2(pool2_stream, s);
Willy Tarreau8baf9062015-04-05 00:46:36 +02001838 out_free_task:
1839 task_free(t);
Willy Tarreau15b5e142015-04-04 14:38:25 +02001840 out_free_sess:
Willy Tarreau11c36242015-04-04 15:54:03 +02001841 session_free(sess);
Willy Tarreaud990baf2015-04-05 00:32:03 +02001842 out_free_appctx:
1843 appctx_free(appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02001844 out_close:
Willy Tarreaub21d08e2016-10-31 17:46:57 +01001845 return NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001846}
1847
1848/*
1849 * Task processing function to manage re-connect and peer session
1850 * tasks wakeup on local update.
1851 */
Simon Horman96553772011-06-08 09:18:51 +09001852static struct task *process_peer_sync(struct task * task)
Emeric Brun2b920a12010-09-23 18:30:22 +02001853{
Vincent Bernat3c2f2f22016-04-03 13:48:42 +02001854 struct peers *peers = task->context;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001855 struct peer *ps;
1856 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +02001857
1858 task->expire = TICK_ETERNITY;
1859
Emeric Brunb3971ab2015-05-12 18:49:09 +02001860 if (!peers->peers_fe) {
Willy Tarreau46dc1ca2015-05-01 18:32:13 +02001861 /* this one was never started, kill it */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001862 signal_unregister_handler(peers->sighandler);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001863 task_delete(peers->sync_task);
1864 task_free(peers->sync_task);
Willy Tarreau37bb7be2015-09-21 15:24:58 +02001865 peers->sync_task = NULL;
Willy Tarreau46dc1ca2015-05-01 18:32:13 +02001866 return NULL;
1867 }
1868
Emeric Brun2b920a12010-09-23 18:30:22 +02001869 if (!stopping) {
1870 /* Normal case (not soft stop)*/
Emeric Brunb3971ab2015-05-12 18:49:09 +02001871
1872 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL) &&
1873 (!nb_oldpids || tick_is_expired(peers->resync_timeout, now_ms)) &&
1874 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001875 /* Resync from local peer needed
1876 no peer was assigned for the lesson
1877 and no old local peer found
1878 or resync timeout expire */
1879
1880 /* flag no more resync from local, to try resync from remotes */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001881 peers->flags |= PEERS_F_RESYNC_LOCAL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001882
1883 /* reschedule a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001884 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
Emeric Brun2b920a12010-09-23 18:30:22 +02001885 }
1886
1887 /* For each session */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001888 for (ps = peers->remote; ps; ps = ps->next) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001889 /* For each remote peers */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001890 if (!ps->local) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01001891 if (!ps->appctx) {
1892 /* no active peer connection */
Emeric Brun2b920a12010-09-23 18:30:22 +02001893 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001894 ((ps->statuscode == PEER_SESS_SC_CONNECTCODE ||
Willy Tarreaub4e34da2015-05-20 10:39:04 +02001895 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001896 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02001897 tick_is_expired(ps->reconnect, now_ms))) {
1898 /* connection never tried
Willy Tarreau9df94c22016-10-31 18:42:52 +01001899 * or previous peer connection established with success
1900 * or previous peer connection failed while connecting
Emeric Brun2b920a12010-09-23 18:30:22 +02001901 * and reconnection timer is expired */
1902
1903 /* retry a connect */
Willy Tarreau9df94c22016-10-31 18:42:52 +01001904 ps->appctx = peer_session_create(peers, ps);
Emeric Brun2b920a12010-09-23 18:30:22 +02001905 }
Willy Tarreaub4e34da2015-05-20 10:39:04 +02001906 else if (!tick_is_expired(ps->reconnect, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001907 /* If previous session failed during connection
1908 * but reconnection timer is not expired */
1909
1910 /* reschedule task for reconnect */
1911 task->expire = tick_first(task->expire, ps->reconnect);
1912 }
1913 /* else do nothing */
Willy Tarreau9df94c22016-10-31 18:42:52 +01001914 } /* !ps->appctx */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001915 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01001916 /* current peer connection is active and established */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001917 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
1918 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02001919 !(ps->flags & PEER_F_LEARN_NOTUP2DATE)) {
1920 /* Resync from a remote is needed
1921 * and no peer was assigned for lesson
1922 * and current peer may be up2date */
1923
1924 /* assign peer for the lesson */
1925 ps->flags |= PEER_F_LEARN_ASSIGN;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001926 peers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brun2b920a12010-09-23 18:30:22 +02001927
Willy Tarreau9df94c22016-10-31 18:42:52 +01001928 /* wake up peer handler to handle a request of resync */
Willy Tarreaue5843b32015-04-27 18:40:14 +02001929 appctx_wakeup(ps->appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02001930 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001931 else {
1932 /* Awake session if there is data to push */
1933 for (st = ps->tables; st ; st = st->next) {
1934 if ((int)(st->last_pushed - st->table->localupdate) < 0) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01001935 /* wake up the peer handler to push local updates */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001936 appctx_wakeup(ps->appctx);
1937 break;
1938 }
1939 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001940 }
1941 /* else do nothing */
1942 } /* SUCCESSCODE */
1943 } /* !ps->peer->local */
1944 } /* for */
1945
1946 /* Resync from remotes expired: consider resync is finished */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001947 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
1948 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
1949 tick_is_expired(peers->resync_timeout, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001950 /* Resync from remote peer needed
1951 * no peer was assigned for the lesson
1952 * and resync timeout expire */
1953
1954 /* flag no more resync from remote, consider resync is finished */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001955 peers->flags |= PEERS_F_RESYNC_REMOTE;
Emeric Brun2b920a12010-09-23 18:30:22 +02001956 }
1957
Emeric Brunb3971ab2015-05-12 18:49:09 +02001958 if ((peers->flags & PEERS_RESYNC_STATEMASK) != PEERS_RESYNC_FINISHED) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001959 /* Resync not finished*/
1960 /* reschedule task to resync timeout, to ended resync if needed */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001961 task->expire = tick_first(task->expire, peers->resync_timeout);
Emeric Brun2b920a12010-09-23 18:30:22 +02001962 }
1963 } /* !stopping */
1964 else {
1965 /* soft stop case */
1966 if (task->state & TASK_WOKEN_SIGNAL) {
1967 /* We've just recieved the signal */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001968 if (!(peers->flags & PEERS_F_DONOTSTOP)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001969 /* add DO NOT STOP flag if not present */
1970 jobs++;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001971 peers->flags |= PEERS_F_DONOTSTOP;
1972 ps = peers->local;
1973 for (st = ps->tables; st ; st = st->next)
1974 st->table->syncing++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001975 }
1976
1977 /* disconnect all connected peers */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001978 for (ps = peers->remote; ps; ps = ps->next) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01001979 if (ps->appctx) {
Willy Tarreau81bc3b02016-10-31 17:37:39 +01001980 peer_session_forceshutdown(ps->appctx);
Willy Tarreaue5843b32015-04-27 18:40:14 +02001981 ps->appctx = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001982 }
1983 }
1984 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001985
Emeric Brunb3971ab2015-05-12 18:49:09 +02001986 ps = peers->local;
Emeric Brun2b920a12010-09-23 18:30:22 +02001987 if (ps->flags & PEER_F_TEACH_COMPLETE) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001988 if (peers->flags & PEERS_F_DONOTSTOP) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001989 /* resync of new process was complete, current process can die now */
1990 jobs--;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001991 peers->flags &= ~PEERS_F_DONOTSTOP;
1992 for (st = ps->tables; st ; st = st->next)
1993 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02001994 }
1995 }
Willy Tarreau9df94c22016-10-31 18:42:52 +01001996 else if (!ps->appctx) {
1997 /* If there's no active peer connection */
Emeric Brun2b920a12010-09-23 18:30:22 +02001998 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001999 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
2000 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE ||
2001 ps->statuscode == PEER_SESS_SC_TRYAGAIN) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002002 /* connection never tried
Willy Tarreau9df94c22016-10-31 18:42:52 +01002003 * or previous peer connection was successfully established
2004 * or previous tcp connect succeeded but init state incomplete
Emeric Brun2b920a12010-09-23 18:30:22 +02002005 * or during previous connect, peer replies a try again statuscode */
2006
2007 /* connect to the peer */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002008 peer_session_create(peers, ps);
Emeric Brun2b920a12010-09-23 18:30:22 +02002009 }
2010 else {
2011 /* Other error cases */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002012 if (peers->flags & PEERS_F_DONOTSTOP) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002013 /* unable to resync new process, current process can die now */
2014 jobs--;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002015 peers->flags &= ~PEERS_F_DONOTSTOP;
2016 for (st = ps->tables; st ; st = st->next)
2017 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02002018 }
2019 }
2020 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02002021 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE ) {
Willy Tarreau9df94c22016-10-31 18:42:52 +01002022 /* current peer connection is active and established
2023 * wake up all peer handlers to push remaining local updates */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002024 for (st = ps->tables; st ; st = st->next) {
2025 if ((int)(st->last_pushed - st->table->localupdate) < 0) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02002026 appctx_wakeup(ps->appctx);
2027 break;
2028 }
2029 }
Emeric Brun2b920a12010-09-23 18:30:22 +02002030 }
2031 } /* stopping */
2032 /* Wakeup for re-connect */
2033 return task;
2034}
2035
Emeric Brunb3971ab2015-05-12 18:49:09 +02002036
Emeric Brun2b920a12010-09-23 18:30:22 +02002037/*
Emeric Brun2b920a12010-09-23 18:30:22 +02002038 *
2039 */
Emeric Brunb3971ab2015-05-12 18:49:09 +02002040void peers_init_sync(struct peers *peers)
Emeric Brun2b920a12010-09-23 18:30:22 +02002041{
Emeric Brun2b920a12010-09-23 18:30:22 +02002042 struct peer * curpeer;
Willy Tarreau4348fad2012-09-20 16:48:07 +02002043 struct listener *listener;
Emeric Brun2b920a12010-09-23 18:30:22 +02002044
Emeric Brun2b920a12010-09-23 18:30:22 +02002045 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
Emeric Brun2b920a12010-09-23 18:30:22 +02002046 peers->peers_fe->maxconn += 3;
2047 }
2048
Willy Tarreau4348fad2012-09-20 16:48:07 +02002049 list_for_each_entry(listener, &peers->peers_fe->conf.listeners, by_fe)
2050 listener->maxconn = peers->peers_fe->maxconn;
Emeric Brunb3971ab2015-05-12 18:49:09 +02002051 peers->sync_task = task_new();
2052 peers->sync_task->process = process_peer_sync;
2053 peers->sync_task->expire = TICK_ETERNITY;
2054 peers->sync_task->context = (void *)peers;
2055 peers->sighandler = signal_register_task(0, peers->sync_task, 0);
2056 task_wakeup(peers->sync_task, TASK_WOKEN_INIT);
2057}
2058
2059
2060
2061/*
2062 * Function used to register a table for sync on a group of peers
2063 *
2064 */
2065void peers_register_table(struct peers *peers, struct stktable *table)
2066{
2067 struct shared_table *st;
2068 struct peer * curpeer;
2069 int id = 0;
2070
2071 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
Vincent Bernat02779b62016-04-03 13:48:43 +02002072 st = calloc(1,sizeof(*st));
Emeric Brunb3971ab2015-05-12 18:49:09 +02002073 st->table = table;
2074 st->next = curpeer->tables;
2075 if (curpeer->tables)
2076 id = curpeer->tables->local_id;
2077 st->local_id = id + 1;
2078
2079 curpeer->tables = st;
2080 }
2081
2082 table->sync_task = peers->sync_task;
Emeric Brun2b920a12010-09-23 18:30:22 +02002083}
2084