blob: d3fef4ae3f4b984c2b85e99a7b4aa628178c2a2c [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>
26
27#include <types/global.h>
Willy Tarreau3fdb3662012-11-12 00:42:33 +010028#include <types/listener.h>
29#include <types/obj_type.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020030#include <types/peers.h>
31
32#include <proto/acl.h>
Willy Tarreau8a8d83b2015-04-13 13:24:54 +020033#include <proto/applet.h>
Willy Tarreauc7e42382012-08-24 19:22:53 +020034#include <proto/channel.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020035#include <proto/fd.h>
Willy Tarreaud1d48d42015-03-13 16:15:46 +010036#include <proto/frontend.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020037#include <proto/log.h>
38#include <proto/hdr_idx.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020039#include <proto/proto_tcp.h>
40#include <proto/proto_http.h>
41#include <proto/proxy.h>
Willy Tarreaufeb76402015-04-03 14:10:06 +020042#include <proto/session.h>
Willy Tarreau87b09662015-04-03 00:22:06 +020043#include <proto/stream.h>
Willy Tarreau22ec1ea2014-11-27 20:45:39 +010044#include <proto/signal.h>
45#include <proto/stick_table.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020046#include <proto/stream_interface.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020047#include <proto/task.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020048
49
50/*******************************/
51/* Current peer learning state */
52/*******************************/
53
54/******************************/
Emeric Brunb3971ab2015-05-12 18:49:09 +020055/* Current peers section resync state */
Emeric Brun2b920a12010-09-23 18:30:22 +020056/******************************/
Emeric Brunb3971ab2015-05-12 18:49:09 +020057#define PEERS_F_RESYNC_LOCAL 0x00000001 /* Learn from local finished or no more needed */
58#define PEERS_F_RESYNC_REMOTE 0x00000002 /* Learn from remote finished or no more needed */
59#define PEERS_F_RESYNC_ASSIGN 0x00000004 /* A peer was assigned to learn our lesson */
60#define PEERS_F_RESYNC_PROCESS 0x00000008 /* The assigned peer was requested for resync */
61#define PEERS_F_DONOTSTOP 0x00010000 /* Main table sync task block process during soft stop
Emeric Brun2b920a12010-09-23 18:30:22 +020062 to push data to new process */
63
Emeric Brunb3971ab2015-05-12 18:49:09 +020064#define PEERS_RESYNC_STATEMASK (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE)
65#define PEERS_RESYNC_FROMLOCAL 0x00000000
66#define PEERS_RESYNC_FROMREMOTE PEERS_F_RESYNC_LOCAL
67#define PEERS_RESYNC_FINISHED (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE)
68
69/***********************************/
70/* Current shared table sync state */
71/***********************************/
72#define SHTABLE_F_TEACH_STAGE1 0x00000001 /* Teach state 1 complete */
73#define SHTABLE_F_TEACH_STAGE2 0x00000002 /* Teach state 2 complete */
Emeric Brun2b920a12010-09-23 18:30:22 +020074
75/******************************/
76/* Remote peer teaching state */
77/******************************/
78#define PEER_F_TEACH_PROCESS 0x00000001 /* Teach a lesson to current peer */
Emeric Brun2b920a12010-09-23 18:30:22 +020079#define PEER_F_TEACH_FINISHED 0x00000008 /* Teach conclude, (wait for confirm) */
80#define PEER_F_TEACH_COMPLETE 0x00000010 /* All that we know already taught to current peer, used only for a local peer */
81#define PEER_F_LEARN_ASSIGN 0x00000100 /* Current peer was assigned for a lesson */
82#define PEER_F_LEARN_NOTUP2DATE 0x00000200 /* Learn from peer finished but peer is not up to date */
83
Emeric Brunb3971ab2015-05-12 18:49:09 +020084#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 +020085#define PEER_LEARN_RESET ~(PEER_F_LEARN_ASSIGN|PEER_F_LEARN_NOTUP2DATE)
86
Emeric Brunb3971ab2015-05-12 18:49:09 +020087/*****************************/
88/* Sync message class */
89/*****************************/
90enum {
91 PEER_MSG_CLASS_CONTROL = 0,
92 PEER_MSG_CLASS_ERROR,
93 PEER_MSG_CLASS_STICKTABLE = 10,
94 PEER_MSG_CLASS_RESERVED = 255,
95};
96
97/*****************************/
98/* control message types */
99/*****************************/
100enum {
101 PEER_MSG_CTRL_RESYNCREQ = 0,
102 PEER_MSG_CTRL_RESYNCFINISHED,
103 PEER_MSG_CTRL_RESYNCPARTIAL,
104 PEER_MSG_CTRL_RESYNCCONFIRM,
105};
106
107/*****************************/
108/* error message types */
109/*****************************/
110enum {
111 PEER_MSG_ERR_PROTOCOL = 0,
112 PEER_MSG_ERR_SIZELIMIT,
113};
114
115
116/*******************************/
117/* stick table sync mesg types */
118/* Note: ids >= 128 contains */
119/* id message cotains data */
120/*******************************/
121enum {
122 PEER_MSG_STKT_UPDATE = 128,
123 PEER_MSG_STKT_INCUPDATE,
124 PEER_MSG_STKT_DEFINE,
125 PEER_MSG_STKT_SWITCH,
126 PEER_MSG_STKT_ACK,
127};
Emeric Brun2b920a12010-09-23 18:30:22 +0200128
129/**********************************/
130/* Peer Session IO handler states */
131/**********************************/
132
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100133enum {
134 PEER_SESS_ST_ACCEPT = 0, /* Initial state for session create by an accept, must be zero! */
135 PEER_SESS_ST_GETVERSION, /* Validate supported protocol version */
136 PEER_SESS_ST_GETHOST, /* Validate host ID correspond to local host id */
137 PEER_SESS_ST_GETPEER, /* Validate peer ID correspond to a known remote peer id */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100138 /* after this point, data were possibly exchanged */
139 PEER_SESS_ST_SENDSUCCESS, /* Send ret code 200 (success) and wait for message */
140 PEER_SESS_ST_CONNECT, /* Initial state for session create on a connect, push presentation into buffer */
141 PEER_SESS_ST_GETSTATUS, /* Wait for the welcome message */
142 PEER_SESS_ST_WAITMSG, /* Wait for data messages */
143 PEER_SESS_ST_EXIT, /* Exit with status code */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200144 PEER_SESS_ST_ERRPROTO, /* Send error proto message before exit */
145 PEER_SESS_ST_ERRSIZE, /* Send error size message before exit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100146 PEER_SESS_ST_END, /* Killed session */
147};
Emeric Brun2b920a12010-09-23 18:30:22 +0200148
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100149/***************************************************/
150/* Peer Session status code - part of the protocol */
151/***************************************************/
Emeric Brun2b920a12010-09-23 18:30:22 +0200152
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100153#define PEER_SESS_SC_CONNECTCODE 100 /* connect in progress */
154#define PEER_SESS_SC_CONNECTEDCODE 110 /* tcp connect success */
Emeric Brun2b920a12010-09-23 18:30:22 +0200155
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100156#define PEER_SESS_SC_SUCCESSCODE 200 /* accept or connect successful */
Emeric Brun2b920a12010-09-23 18:30:22 +0200157
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100158#define PEER_SESS_SC_TRYAGAIN 300 /* try again later */
Emeric Brun2b920a12010-09-23 18:30:22 +0200159
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100160#define PEER_SESS_SC_ERRPROTO 501 /* error protocol */
161#define PEER_SESS_SC_ERRVERSION 502 /* unknown protocol version */
162#define PEER_SESS_SC_ERRHOST 503 /* bad host name */
163#define PEER_SESS_SC_ERRPEER 504 /* unknown peer */
Emeric Brun2b920a12010-09-23 18:30:22 +0200164
165#define PEER_SESSION_PROTO_NAME "HAProxyS"
166
167struct peers *peers = NULL;
Willy Tarreau87b09662015-04-03 00:22:06 +0200168static void peer_session_forceshutdown(struct stream * stream);
Emeric Brun2b920a12010-09-23 18:30:22 +0200169
Emeric Brunb3971ab2015-05-12 18:49:09 +0200170int intencode(uint64_t i, char **str) {
171 int idx = 0;
172 unsigned char *msg;
173
174 if (!*str)
175 return 0;
176
177 msg = (unsigned char *)*str;
178 if (i < 240) {
179 msg[0] = (unsigned char)i;
180 *str = (char *)&msg[idx+1];
181 return (idx+1);
182 }
183
184 msg[idx] =(unsigned char)i | 240;
185 i = (i - 240) >> 4;
186 while (i >= 128) {
187 msg[++idx] = (unsigned char)i | 128;
188 i = (i - 128) >> 7;
189 }
190 msg[++idx] = (unsigned char)i;
191 *str = (char *)&msg[idx+1];
192 return (idx+1);
193}
194
195
196/* This function returns the decoded integer or 0
197 if decode failed
198 *str point on the beginning of the integer to decode
199 at the end of decoding *str point on the end of the
200 encoded integer or to null if end is reached */
201uint64_t intdecode(char **str, char *end) {
202 uint64_t i;
203 int idx = 0;
204 unsigned char *msg;
205
206 if (!*str)
207 return 0;
208
209 msg = (unsigned char *)*str;
210 if (msg >= (unsigned char *)end) {
211 *str = NULL;
212 return 0;
213 }
214
215 if (msg[idx] < 240) {
216 *str = (char *)&msg[idx+1];
217 return msg[idx];
218 }
219 i = msg[idx];
220 do {
221 idx++;
222 if (msg >= (unsigned char *)end) {
223 *str = NULL;
224 return 0;
225 }
226 i += (uint64_t)msg[idx] << (4 + 7*(idx-1));
227 }
228 while (msg[idx] > 128);
229 *str = (char *)&msg[idx+1];
230 return i;
231}
Emeric Brun2b920a12010-09-23 18:30:22 +0200232
233/*
Emeric Brunb3971ab2015-05-12 18:49:09 +0200234 * This prepare the data update message on the stick session <ts>, <st> is the considered
235 * stick table.
236 * <msg> is a buffer of <size> to recieve data message content
237 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
238 * check size)
Emeric Brun2b920a12010-09-23 18:30:22 +0200239 */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200240static int peer_prepare_updatemsg(struct stksess *ts, struct shared_table *st, char *msg, size_t size, int use_identifier)
Emeric Brun2b920a12010-09-23 18:30:22 +0200241{
242 uint32_t netinteger;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200243 unsigned short datalen;
244 char *cursor, *datamsg;
Emeric Brun94900952015-06-11 18:25:54 +0200245 unsigned int data_type;
246 void *data_ptr;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200247
248 cursor = datamsg = msg + 1 + 5;
249
Emeric Brun2b920a12010-09-23 18:30:22 +0200250 /* construct message */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200251
252 /* check if we need to send the update identifer */
Emeric Bruna6a09982015-09-22 15:34:19 +0200253 if (!st->last_pushed || ts->upd.key < st->last_pushed || ((ts->upd.key - st->last_pushed) != 1)) {
254 use_identifier = 1;
Emeric Brun2b920a12010-09-23 18:30:22 +0200255 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200256
257 /* encode update identifier if needed */
258 if (use_identifier) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200259 netinteger = htonl(ts->upd.key);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200260 memcpy(cursor, &netinteger, sizeof(netinteger));
261 cursor += sizeof(netinteger);
Emeric Brun2b920a12010-09-23 18:30:22 +0200262 }
263
Emeric Brunb3971ab2015-05-12 18:49:09 +0200264 /* encode the key */
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +0200265 if (st->table->type == SMP_T_STR) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200266 int stlen = strlen((char *)ts->key.key);
267
Emeric Brunb3971ab2015-05-12 18:49:09 +0200268 intencode(stlen, &cursor);
269 memcpy(cursor, ts->key.key, stlen);
270 cursor += stlen;
Emeric Brun2b920a12010-09-23 18:30:22 +0200271 }
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +0200272 else if (st->table->type == SMP_T_SINT) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200273 netinteger = htonl(*((uint32_t *)ts->key.key));
Emeric Brunb3971ab2015-05-12 18:49:09 +0200274 memcpy(cursor, &netinteger, sizeof(netinteger));
275 cursor += sizeof(netinteger);
Emeric Brun2b920a12010-09-23 18:30:22 +0200276 }
277 else {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200278 memcpy(cursor, ts->key.key, st->table->key_size);
279 cursor += st->table->key_size;
Emeric Brun2b920a12010-09-23 18:30:22 +0200280 }
281
Emeric Brunb3971ab2015-05-12 18:49:09 +0200282 /* encode values */
Emeric Brun94900952015-06-11 18:25:54 +0200283 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200284
Emeric Brun94900952015-06-11 18:25:54 +0200285 data_ptr = stktable_data_ptr(st->table, ts, data_type);
286 if (data_ptr) {
287 switch (stktable_data_types[data_type].std_type) {
288 case STD_T_SINT: {
289 int data;
290
291 data = stktable_data_cast(data_ptr, std_t_sint);
292 intencode(data, &cursor);
293 break;
294 }
295 case STD_T_UINT: {
296 unsigned int data;
297
298 data = stktable_data_cast(data_ptr, std_t_uint);
299 intencode(data, &cursor);
300 break;
301 }
302 case STD_T_ULL: {
303 unsigned long long data;
304
305 data = stktable_data_cast(data_ptr, std_t_ull);
306 intencode(data, &cursor);
307 break;
308 }
309 case STD_T_FRQP: {
310 struct freq_ctr_period *frqp;
311
312 frqp = &stktable_data_cast(data_ptr, std_t_frqp);
313 intencode((unsigned int)(now_ms - frqp->curr_tick), &cursor);
314 intencode(frqp->curr_ctr, &cursor);
315 intencode(frqp->prev_ctr, &cursor);
316 break;
317 }
318 }
319 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200320 }
321
322 /* Compute datalen */
323 datalen = (cursor - datamsg);
324
325 /* prepare message header */
326 msg[0] = PEER_MSG_CLASS_STICKTABLE;
327 if (use_identifier)
328 msg[1] = PEER_MSG_STKT_UPDATE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200329 else
Emeric Brunb3971ab2015-05-12 18:49:09 +0200330 msg[1] = PEER_MSG_STKT_INCUPDATE;
331
332 cursor = &msg[2];
333 intencode(datalen, &cursor);
334
335 /* move data after header */
336 memmove(cursor, datamsg, datalen);
337
338 /* return header size + data_len */
339 return (cursor - msg) + datalen;
340}
341
342/*
343 * This prepare the switch table message to targeted share table <st>.
344 * <msg> is a buffer of <size> to recieve data message content
345 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
346 * check size)
347 */
348static int peer_prepare_switchmsg(struct shared_table *st, char *msg, size_t size)
349{
350 int len;
351 unsigned short datalen;
352 char *cursor, *datamsg;
353 uint64_t data = 0;
Emeric Brun94900952015-06-11 18:25:54 +0200354 unsigned int data_type;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200355
356 cursor = datamsg = msg + 2 + 5;
357
358 /* Encode data */
359
360 /* encode local id */
361 intencode(st->local_id, &cursor);
362
363 /* encode table name */
364 len = strlen(st->table->id);
365 intencode(len, &cursor);
366 memcpy(cursor, st->table->id, len);
367 cursor += len;
368
369 /* encode table type */
370
371 intencode(st->table->type, &cursor);
372
373 /* encode table key size */
374 intencode(st->table->key_size, &cursor);
375
Emeric Brun94900952015-06-11 18:25:54 +0200376 /* encode available known data types in table */
377 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
378 if (st->table->data_ofs[data_type]) {
379 switch (stktable_data_types[data_type].std_type) {
380 case STD_T_SINT:
381 case STD_T_UINT:
382 case STD_T_ULL:
383 case STD_T_FRQP:
384 data |= 1 << data_type;
385 break;
386 }
387 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200388 }
389 intencode(data, &cursor);
390
391 /* Compute datalen */
392 datalen = (cursor - datamsg);
Emeric Brun2b920a12010-09-23 18:30:22 +0200393
Emeric Brunb3971ab2015-05-12 18:49:09 +0200394 /* prepare message header */
395 msg[0] = PEER_MSG_CLASS_STICKTABLE;
396 msg[1] = PEER_MSG_STKT_DEFINE;
397 cursor = &msg[2];
398 intencode(datalen, &cursor);
Emeric Brun2b920a12010-09-23 18:30:22 +0200399
Emeric Brunb3971ab2015-05-12 18:49:09 +0200400 /* move data after header */
401 memmove(cursor, datamsg, datalen);
402
403 /* return header size + data_len */
404 return (cursor - msg) + datalen;
Emeric Brun2b920a12010-09-23 18:30:22 +0200405}
406
Emeric Brunb3971ab2015-05-12 18:49:09 +0200407/*
408 * This prepare the acknowledge message on the stick session <ts>, <st> is the considered
409 * stick table.
410 * <msg> is a buffer of <size> to recieve data message content
411 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
412 * check size)
413 */
414static int peer_prepare_ackmsg(struct shared_table *st, char *msg, size_t size)
415{
416 unsigned short datalen;
417 char *cursor, *datamsg;
418 uint32_t netinteger;
419
Emeric Brunb058f1c2015-09-22 15:50:18 +0200420 cursor = datamsg = msg + 2 + 5;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200421
422 intencode(st->remote_id, &cursor);
423 netinteger = htonl(st->last_get);
424 memcpy(cursor, &netinteger, sizeof(netinteger));
425 cursor += sizeof(netinteger);
426
427 /* Compute datalen */
428 datalen = (cursor - datamsg);
429
430 /* prepare message header */
431 msg[0] = PEER_MSG_CLASS_STICKTABLE;
Emeric Brune1ab8082015-08-21 11:48:54 +0200432 msg[1] = PEER_MSG_STKT_ACK;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200433 cursor = &msg[2];
434 intencode(datalen, &cursor);
435
436 /* move data after header */
437 memmove(cursor, datamsg, datalen);
438
439 /* return header size + data_len */
440 return (cursor - msg) + datalen;
441}
Emeric Brun2b920a12010-09-23 18:30:22 +0200442
443/*
444 * Callback to release a session with a peer
445 */
Willy Tarreau00a37f02015-04-13 12:05:19 +0200446static void peer_session_release(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +0200447{
Willy Tarreau00a37f02015-04-13 12:05:19 +0200448 struct stream_interface *si = appctx->owner;
Willy Tarreau87b09662015-04-03 00:22:06 +0200449 struct stream *s = si_strm(si);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200450 struct peer *peer = (struct peer *)appctx->ctx.peers.ptr;
451 struct peers *peers = (struct peers *)strm_fe(s)->parent;
Emeric Brun2b920a12010-09-23 18:30:22 +0200452
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100453 /* appctx->ctx.peers.ptr is not a peer session */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100454 if (appctx->st0 < PEER_SESS_ST_SENDSUCCESS)
Emeric Brun2b920a12010-09-23 18:30:22 +0200455 return;
456
457 /* peer session identified */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200458 if (peer) {
459 if (peer->stream == s) {
Emeric Brunb157d732015-08-21 12:00:30 +0200460 /* Re-init current table pointers to force announcement on re-connect */
461 peer->remote_table = peer->last_local_table = NULL;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200462 peer->stream = NULL;
463 peer->appctx = NULL;
464 if (peer->flags & PEER_F_LEARN_ASSIGN) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200465 /* unassign current peer for learning */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200466 peer->flags &= ~(PEER_F_LEARN_ASSIGN);
467 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brun2b920a12010-09-23 18:30:22 +0200468
469 /* reschedule a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200470 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
Emeric Brun2b920a12010-09-23 18:30:22 +0200471 }
472 /* reset teaching and learning flags to 0 */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200473 peer->flags &= PEER_TEACH_RESET;
474 peer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200475 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200476 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200477 }
478}
479
480
481/*
482 * IO Handler to handle message exchance with a peer
483 */
Willy Tarreau00a37f02015-04-13 12:05:19 +0200484static void peer_io_handler(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +0200485{
Willy Tarreau00a37f02015-04-13 12:05:19 +0200486 struct stream_interface *si = appctx->owner;
Willy Tarreau87b09662015-04-03 00:22:06 +0200487 struct stream *s = si_strm(si);
Willy Tarreaud0d8da92015-04-04 02:10:38 +0200488 struct peers *curpeers = (struct peers *)strm_fe(s)->parent;
Emeric Brun2b920a12010-09-23 18:30:22 +0200489 int reql = 0;
490 int repl = 0;
491
492 while (1) {
493switchstate:
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100494 switch(appctx->st0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100495 case PEER_SESS_ST_ACCEPT:
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100496 appctx->ctx.peers.ptr = NULL;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100497 appctx->st0 = PEER_SESS_ST_GETVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200498 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100499 case PEER_SESS_ST_GETVERSION:
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100500 reql = bo_getline(si_oc(si), trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200501 if (reql <= 0) { /* closed or EOL not found */
502 if (reql == 0)
503 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100504 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200505 goto switchstate;
506 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100507 if (trash.str[reql-1] != '\n') {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100508 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200509 goto switchstate;
510 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100511 else if (reql > 1 && (trash.str[reql-2] == '\r'))
512 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200513 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100514 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200515
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100516 bo_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200517
518 /* test version */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200519 if (strcmp(PEER_SESSION_PROTO_NAME " 2.0", trash.str) != 0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100520 appctx->st0 = PEER_SESS_ST_EXIT;
521 appctx->st1 = PEER_SESS_SC_ERRVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200522 /* test protocol */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100523 if (strncmp(PEER_SESSION_PROTO_NAME " ", trash.str, strlen(PEER_SESSION_PROTO_NAME)+1) != 0)
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100524 appctx->st1 = PEER_SESS_SC_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +0200525 goto switchstate;
526 }
527
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100528 appctx->st0 = PEER_SESS_ST_GETHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200529 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100530 case PEER_SESS_ST_GETHOST:
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100531 reql = bo_getline(si_oc(si), trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200532 if (reql <= 0) { /* closed or EOL not found */
533 if (reql == 0)
534 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100535 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200536 goto switchstate;
537 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100538 if (trash.str[reql-1] != '\n') {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100539 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200540 goto switchstate;
541 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100542 else if (reql > 1 && (trash.str[reql-2] == '\r'))
543 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200544 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100545 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200546
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100547 bo_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200548
549 /* test hostname match */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100550 if (strcmp(localpeer, trash.str) != 0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100551 appctx->st0 = PEER_SESS_ST_EXIT;
552 appctx->st1 = PEER_SESS_SC_ERRHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200553 goto switchstate;
554 }
555
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100556 appctx->st0 = PEER_SESS_ST_GETPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200557 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100558 case PEER_SESS_ST_GETPEER: {
Emeric Brun2b920a12010-09-23 18:30:22 +0200559 struct peer *curpeer;
560 char *p;
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100561 reql = bo_getline(si_oc(si), trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200562 if (reql <= 0) { /* closed or EOL not found */
563 if (reql == 0)
564 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100565 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200566 goto switchstate;
567 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100568 if (trash.str[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200569 /* Incomplete line, we quit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100570 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200571 goto switchstate;
572 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100573 else if (reql > 1 && (trash.str[reql-2] == '\r'))
574 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200575 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100576 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200577
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100578 bo_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200579
Emeric Brunb3971ab2015-05-12 18:49:09 +0200580 /* parse line "<peer name> <pid> <relative_pid>" */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100581 p = strchr(trash.str, ' ');
Emeric Brun2b920a12010-09-23 18:30:22 +0200582 if (!p) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100583 appctx->st0 = PEER_SESS_ST_EXIT;
584 appctx->st1 = PEER_SESS_SC_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +0200585 goto switchstate;
586 }
587 *p = 0;
588
589 /* lookup known peer */
590 for (curpeer = curpeers->remote; curpeer; curpeer = curpeer->next) {
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100591 if (strcmp(curpeer->id, trash.str) == 0)
Emeric Brun2b920a12010-09-23 18:30:22 +0200592 break;
593 }
594
595 /* if unknown peer */
596 if (!curpeer) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100597 appctx->st0 = PEER_SESS_ST_EXIT;
598 appctx->st1 = PEER_SESS_SC_ERRPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200599 goto switchstate;
600 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200601
Emeric Brunb3971ab2015-05-12 18:49:09 +0200602 if (curpeer->stream && curpeer->stream != s) {
603 if (curpeer->local) {
604 /* Local connection, reply a retry */
605 appctx->st0 = PEER_SESS_ST_EXIT;
606 appctx->st1 = PEER_SESS_SC_TRYAGAIN;
607 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +0200608 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200609 peer_session_forceshutdown(curpeer->stream);
Emeric Brun2b920a12010-09-23 18:30:22 +0200610 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200611 curpeer->stream = s;
612 curpeer->appctx = appctx;
613 appctx->ctx.peers.ptr = curpeer;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100614 appctx->st0 = PEER_SESS_ST_SENDSUCCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200615 /* fall through */
616 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100617 case PEER_SESS_ST_SENDSUCCESS: {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200618 struct peer *curpeer = (struct peer *)appctx->ctx.peers.ptr;
619 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +0200620
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100621 repl = snprintf(trash.str, trash.size, "%d\n", PEER_SESS_SC_SUCCESSCODE);
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100622 repl = bi_putblk(si_ic(si), trash.str, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200623 if (repl <= 0) {
624 if (repl == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +0100625 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100626 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200627 goto switchstate;
628 }
629
630 /* Register status code */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200631 curpeer->statuscode = PEER_SESS_SC_SUCCESSCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200632
633 /* Awake main task */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200634 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200635
636 /* Init confirm counter */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200637 curpeer->confirm = 0;
638
639 /* Init cursors */
640 for (st = curpeer->tables; st ; st = st->next) {
641 st->last_get = st->last_acked = 0;
642 st->teaching_origin = st->last_pushed = st->update;
643 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200644
645 /* reset teaching and learning flags to 0 */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200646 curpeer->flags &= PEER_TEACH_RESET;
647 curpeer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200648
649 /* if current peer is local */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200650 if (curpeer->local) {
651 /* if current host need resyncfrom local and no process assined */
652 if ((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL &&
653 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
654 /* assign local peer for a lesson, consider lesson already requested */
655 curpeer->flags |= PEER_F_LEARN_ASSIGN;
656 peers->flags |= (PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
657 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200658
Emeric Brunb3971ab2015-05-12 18:49:09 +0200659 }
660 else if ((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
661 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
662 /* assign peer for a lesson */
663 curpeer->flags |= PEER_F_LEARN_ASSIGN;
664 peers->flags |= PEERS_F_RESYNC_ASSIGN;
665 }
666
667
Emeric Brun2b920a12010-09-23 18:30:22 +0200668 /* switch to waiting message state */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100669 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200670 goto switchstate;
671 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100672 case PEER_SESS_ST_CONNECT: {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200673 struct peer *curpeer = (struct peer *)appctx->ctx.peers.ptr;
Emeric Brun2b920a12010-09-23 18:30:22 +0200674
675 /* Send headers */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100676 repl = snprintf(trash.str, trash.size,
Emeric Brunb3971ab2015-05-12 18:49:09 +0200677 PEER_SESSION_PROTO_NAME " 2.0\n%s\n%s %d %d\n",
678 curpeer->id,
Emeric Brun2b920a12010-09-23 18:30:22 +0200679 localpeer,
Willy Tarreau7b77c9f2012-01-07 22:52:12 +0100680 (int)getpid(),
Emeric Brunb3971ab2015-05-12 18:49:09 +0200681 relative_pid);
Emeric Brun2b920a12010-09-23 18:30:22 +0200682
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100683 if (repl >= trash.size) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100684 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200685 goto switchstate;
686 }
687
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100688 repl = bi_putblk(si_ic(si), trash.str, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200689 if (repl <= 0) {
690 if (repl == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +0100691 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100692 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200693 goto switchstate;
694 }
695
696 /* switch to the waiting statuscode state */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100697 appctx->st0 = PEER_SESS_ST_GETSTATUS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200698 /* fall through */
699 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100700 case PEER_SESS_ST_GETSTATUS: {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200701 struct peer *curpeer = (struct peer *)appctx->ctx.peers.ptr;
702 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +0200703
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100704 if (si_ic(si)->flags & CF_WRITE_PARTIAL)
Emeric Brunb3971ab2015-05-12 18:49:09 +0200705 curpeer->statuscode = PEER_SESS_SC_CONNECTEDCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200706
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100707 reql = bo_getline(si_oc(si), trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200708 if (reql <= 0) { /* closed or EOL not found */
709 if (reql == 0)
710 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100711 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200712 goto switchstate;
713 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100714 if (trash.str[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200715 /* Incomplete line, we quit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100716 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200717 goto switchstate;
718 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100719 else if (reql > 1 && (trash.str[reql-2] == '\r'))
720 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200721 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100722 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200723
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100724 bo_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200725
726 /* Register status code */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200727 curpeer->statuscode = atoi(trash.str);
Emeric Brun2b920a12010-09-23 18:30:22 +0200728
729 /* Awake main task */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200730 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200731
732 /* If status code is success */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200733 if (curpeer->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200734 /* Init cursors */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200735 for (st = curpeer->tables; st ; st = st->next) {
736 st->last_get = st->last_acked = 0;
737 st->teaching_origin = st->last_pushed = st->update;
738 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200739
740 /* Init confirm counter */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200741 curpeer->confirm = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200742
Emeric Brunb3971ab2015-05-12 18:49:09 +0200743 /* reset teaching and learning flags to 0 */
744 curpeer->flags &= PEER_TEACH_RESET;
745 curpeer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200746
Emeric Brunb3971ab2015-05-12 18:49:09 +0200747 /* If current peer is local */
748 if (curpeer->local) {
749 /* flag to start to teach lesson */
750 curpeer->flags |= PEER_F_TEACH_PROCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200751
Emeric Brunb3971ab2015-05-12 18:49:09 +0200752 }
753 else if ((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
754 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
755 /* If peer is remote and resync from remote is needed,
756 and no peer currently assigned */
Emeric Brun2b920a12010-09-23 18:30:22 +0200757
Emeric Brunb3971ab2015-05-12 18:49:09 +0200758 /* assign peer for a lesson */
759 curpeer->flags |= PEER_F_LEARN_ASSIGN;
760 peers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brun2b920a12010-09-23 18:30:22 +0200761 }
762
763 }
764 else {
765 /* Status code is not success, abort */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100766 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200767 goto switchstate;
768 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100769 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200770 /* fall through */
771 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100772 case PEER_SESS_ST_WAITMSG: {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200773 struct peer *curpeer = (struct peer *)appctx->ctx.peers.ptr;
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200774 struct stksess *ts, *newts = NULL;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200775 uint32_t msg_len = 0;
776 char *msg_cur = trash.str;
777 char *msg_end = trash.str;
778 unsigned char msg_head[7];
Emeric Brun2b920a12010-09-23 18:30:22 +0200779 int totl = 0;
780
Emeric Brunb3971ab2015-05-12 18:49:09 +0200781 reql = bo_getblk(si_oc(si), (char *)msg_head, 2*sizeof(unsigned char), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200782 if (reql <= 0) /* closed or EOL not found */
783 goto incomplete;
784
Emeric Brun2b920a12010-09-23 18:30:22 +0200785 totl += reql;
786
Emeric Brunb3971ab2015-05-12 18:49:09 +0200787 if (msg_head[1] >= 128) {
788 /* Read and Decode message length */
789 reql = bo_getblk(si_oc(si), (char *)&msg_head[2], sizeof(unsigned char), totl);
790 if (reql <= 0) /* closed */
791 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +0200792
Emeric Brunb3971ab2015-05-12 18:49:09 +0200793 totl += reql;
794
795 if (msg_head[2] < 240) {
796 msg_len = msg_head[2];
Emeric Brun2b920a12010-09-23 18:30:22 +0200797 }
798 else {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200799 int i;
800 char *cur;
801 char *end;
Willy Tarreau72d6c162013-04-11 16:14:13 +0200802
Emeric Brunb3971ab2015-05-12 18:49:09 +0200803 for (i = 3 ; i < sizeof(msg_head) ; i++) {
804 reql = bo_getblk(si_oc(si), (char *)&msg_head[i], sizeof(char), totl);
805 if (reql <= 0) /* closed */
806 goto incomplete;
807
808 totl += reql;
809
810 if (!(msg_head[i] & 0x80))
811 break;
812 }
813
814 if (i == sizeof(msg_head)) {
815 /* malformed message */
816 appctx->st0 = PEER_SESS_ST_ERRPROTO;
817 goto switchstate;
818
819 }
820 end = (char *)msg_head + sizeof(msg_head);
821 cur = (char *)&msg_head[2];
822 msg_len = intdecode(&cur, end);
823 if (!cur) {
824 /* malformed message */
825 appctx->st0 = PEER_SESS_ST_ERRPROTO;
826 goto switchstate;
827 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200828 }
829
Willy Tarreau86a446e2013-11-25 23:02:37 +0100830
Emeric Brunb3971ab2015-05-12 18:49:09 +0200831 /* Read message content */
832 if (msg_len) {
833 if (msg_len > trash.size) {
834 /* Status code is not success, abort */
835 appctx->st0 = PEER_SESS_ST_ERRSIZE;
836 goto switchstate;
837 }
Willy Tarreau72d6c162013-04-11 16:14:13 +0200838
Emeric Brunb3971ab2015-05-12 18:49:09 +0200839 reql = bo_getblk(si_oc(si), trash.str, msg_len, totl);
840 if (reql <= 0) /* closed */
841 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +0200842 totl += reql;
Willy Tarreau86a446e2013-11-25 23:02:37 +0100843
Emeric Brunb3971ab2015-05-12 18:49:09 +0200844 msg_end += msg_len;
845 }
846 }
Willy Tarreau86a446e2013-11-25 23:02:37 +0100847
Emeric Brunb3971ab2015-05-12 18:49:09 +0200848 if (msg_head[0] == PEER_MSG_CLASS_CONTROL) {
849 if (msg_head[1] == PEER_MSG_CTRL_RESYNCREQ) {
850 struct shared_table *st;
851 /* Reset message: remote need resync */
852
853 /* prepare tables fot a global push */
854 for (st = curpeer->tables; st; st = st->next) {
855 st->teaching_origin = st->last_pushed = st->table->update;
856 st->flags = 0;
Willy Tarreau86a446e2013-11-25 23:02:37 +0100857 }
Willy Tarreau72d6c162013-04-11 16:14:13 +0200858
Emeric Brunb3971ab2015-05-12 18:49:09 +0200859 /* reset teaching flags to 0 */
860 curpeer->flags &= PEER_TEACH_RESET;
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200861
Emeric Brunb3971ab2015-05-12 18:49:09 +0200862 /* flag to start to teach lesson */
863 curpeer->flags |= PEER_F_TEACH_PROCESS;
864
865
866 }
867 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCFINISHED) {
868
869 if (curpeer->flags & PEER_F_LEARN_ASSIGN) {
870 curpeer->flags &= ~PEER_F_LEARN_ASSIGN;
871 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
872 peers->flags |= (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE);
Willy Tarreau86a446e2013-11-25 23:02:37 +0100873 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200874 curpeer->confirm++;
Emeric Brun2b920a12010-09-23 18:30:22 +0200875 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200876 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCPARTIAL) {
877
878 if (curpeer->flags & PEER_F_LEARN_ASSIGN) {
879 curpeer->flags &= ~PEER_F_LEARN_ASSIGN;
880 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
881
882 curpeer->flags |= PEER_F_LEARN_NOTUP2DATE;
883 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
884 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
Cyril Bonté9a60ff92014-02-16 01:07:07 +0100885 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200886 curpeer->confirm++;
Emeric Brun2b920a12010-09-23 18:30:22 +0200887 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200888 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCCONFIRM) {
889
890 /* If stopping state */
891 if (stopping) {
892 /* Close session, push resync no more needed */
893 curpeer->flags |= PEER_F_TEACH_COMPLETE;
894 appctx->st0 = PEER_SESS_ST_END;
895 goto switchstate;
896 }
897
898 /* reset teaching flags to 0 */
899 curpeer->flags &= PEER_TEACH_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200900 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200901 }
902 else if (msg_head[0] == PEER_MSG_CLASS_STICKTABLE) {
903 if (msg_head[1] == PEER_MSG_STKT_DEFINE) {
904 int table_id_len;
905 struct shared_table *st;
906 int table_type;
907 int table_keylen;
908 int table_id;
909 uint64_t table_data;
Emeric Brun2b920a12010-09-23 18:30:22 +0200910
Emeric Brunb3971ab2015-05-12 18:49:09 +0200911 table_id = intdecode(&msg_cur, msg_end);
912 if (!msg_cur) {
913 /* malformed message */
914 appctx->st0 = PEER_SESS_ST_ERRPROTO;
915 goto switchstate;
916 }
Willy Tarreau72d6c162013-04-11 16:14:13 +0200917
Emeric Brunb3971ab2015-05-12 18:49:09 +0200918 table_id_len = intdecode(&msg_cur, msg_end);
919 if (!msg_cur) {
920 /* malformed message */
921 appctx->st0 = PEER_SESS_ST_ERRPROTO;
922 goto switchstate;
923 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200924
Emeric Brunb3971ab2015-05-12 18:49:09 +0200925 curpeer->remote_table = NULL;
926 if (!table_id_len || (msg_cur + table_id_len) >= msg_end) {
927 /* malformed message */
928 appctx->st0 = PEER_SESS_ST_ERRPROTO;
929 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +0200930 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200931
Emeric Brunb3971ab2015-05-12 18:49:09 +0200932 for (st = curpeer->tables; st; st = st->next) {
933 /* Reset IDs */
934 if (st->remote_id == table_id)
935 st->remote_id = 0;
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200936
Emeric Brunb3971ab2015-05-12 18:49:09 +0200937 if (!curpeer->remote_table
938 && (table_id_len == strlen(st->table->id))
939 && (memcmp(st->table->id, msg_cur, table_id_len) == 0)) {
940 curpeer->remote_table = st;
Emeric Brun2b920a12010-09-23 18:30:22 +0200941 }
942 }
943
Emeric Brunb3971ab2015-05-12 18:49:09 +0200944 if (!curpeer->remote_table) {
945 goto ignore_msg;
946 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200947
Emeric Brunb3971ab2015-05-12 18:49:09 +0200948 msg_cur += table_id_len;
949 if (msg_cur >= msg_end) {
950 /* malformed message */
951 appctx->st0 = PEER_SESS_ST_ERRPROTO;
952 goto switchstate;
953 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200954
Emeric Brunb3971ab2015-05-12 18:49:09 +0200955 table_type = intdecode(&msg_cur, msg_end);
956 if (!msg_cur) {
957 /* malformed message */
958 appctx->st0 = PEER_SESS_ST_ERRPROTO;
959 goto switchstate;
960 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200961
Emeric Brunb3971ab2015-05-12 18:49:09 +0200962 table_keylen = intdecode(&msg_cur, msg_end);
963 if (!msg_cur) {
964 /* malformed message */
965 appctx->st0 = PEER_SESS_ST_ERRPROTO;
966 goto switchstate;
967 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200968
Emeric Brunb3971ab2015-05-12 18:49:09 +0200969 table_data = intdecode(&msg_cur, msg_end);
970 if (!msg_cur) {
971 /* malformed message */
972 appctx->st0 = PEER_SESS_ST_ERRPROTO;
973 goto switchstate;
974 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200975
Emeric Brunb3971ab2015-05-12 18:49:09 +0200976 if (curpeer->remote_table->table->type != table_type
977 || curpeer->remote_table->table->key_size != table_keylen) {
978 curpeer->remote_table = NULL;
979 goto ignore_msg;
980 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200981
Emeric Brunb3971ab2015-05-12 18:49:09 +0200982 curpeer->remote_table->remote_data = table_data;
983 curpeer->remote_table->remote_id = table_id;
Emeric Brun2b920a12010-09-23 18:30:22 +0200984 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200985 else if (msg_head[1] == PEER_MSG_STKT_SWITCH) {
986 struct shared_table *st;
987 int table_id;
Emeric Brun2b920a12010-09-23 18:30:22 +0200988
Emeric Brunb3971ab2015-05-12 18:49:09 +0200989 table_id = intdecode(&msg_cur, msg_end);
990 if (!msg_cur) {
991 /* malformed message */
992 appctx->st0 = PEER_SESS_ST_ERRPROTO;
993 goto switchstate;
994 }
995 curpeer->remote_table = NULL;
996 for (st = curpeer->tables; st; st = st->next) {
997 if (st->remote_id == table_id) {
998 curpeer->remote_table = st;
999 break;
1000 }
1001 }
1002
Emeric Brun2b920a12010-09-23 18:30:22 +02001003 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001004 else if (msg_head[1] == PEER_MSG_STKT_UPDATE
1005 || msg_head[1] == PEER_MSG_STKT_INCUPDATE) {
1006 struct shared_table *st = curpeer->remote_table;
1007 uint32_t update;
Emeric Brun94900952015-06-11 18:25:54 +02001008 unsigned int data_type;
1009 void *data_ptr;
Emeric Brun2b920a12010-09-23 18:30:22 +02001010
Emeric Brunb3971ab2015-05-12 18:49:09 +02001011 /* Here we have data message */
1012 if (!st)
1013 goto ignore_msg;
Emeric Brun2b920a12010-09-23 18:30:22 +02001014
Emeric Brunb3971ab2015-05-12 18:49:09 +02001015 if (msg_head[1] == PEER_MSG_STKT_UPDATE) {
1016 if (msg_len < sizeof(update)) {
1017 /* malformed message */
1018 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1019 goto switchstate;
1020 }
1021 memcpy(&update, msg_cur, sizeof(update));
1022 msg_cur += sizeof(update);
1023 st->last_get = htonl(update);
1024 }
1025 else {
1026 st->last_get++;
1027 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001028
Emeric Brunb3971ab2015-05-12 18:49:09 +02001029 newts = stksess_new(st->table, NULL);
1030 if (!newts)
1031 goto ignore_msg;
Emeric Brun2b920a12010-09-23 18:30:22 +02001032
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +02001033 if (st->table->type == SMP_T_STR) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001034 unsigned int to_read, to_store;
Emeric Brun2b920a12010-09-23 18:30:22 +02001035
Emeric Brunb3971ab2015-05-12 18:49:09 +02001036 to_read = intdecode(&msg_cur, msg_end);
1037 if (!msg_cur) {
1038 /* malformed message */
1039 stksess_free(st->table, newts);
1040 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1041 goto switchstate;
1042 }
1043 to_store = MIN(to_read, st->table->key_size - 1);
1044 if (msg_cur + to_store > msg_end) {
1045 /* malformed message */
1046 stksess_free(st->table, newts);
1047 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1048 goto switchstate;
1049 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001050
Emeric Brunb3971ab2015-05-12 18:49:09 +02001051 memcpy(newts->key.key, msg_cur, to_store);
1052 newts->key.key[to_store] = 0;
1053 msg_cur += to_read;
1054 }
Thierry FOURNIER5d24ebc2015-07-24 08:46:42 +02001055 else if (st->table->type == SMP_T_SINT) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001056 unsigned int netinteger;
Willy Tarreau72d6c162013-04-11 16:14:13 +02001057
Emeric Brunb3971ab2015-05-12 18:49:09 +02001058 if (msg_cur + sizeof(netinteger) > msg_end) {
1059 /* malformed message */
1060 stksess_free(st->table, newts);
1061 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1062 goto switchstate;
1063 }
1064 memcpy(&netinteger, msg_cur, sizeof(netinteger));
1065 netinteger = ntohl(netinteger);
1066 memcpy(newts->key.key, &netinteger, sizeof(netinteger));
1067 msg_cur += sizeof(netinteger);
1068 }
1069 else {
1070 if (msg_cur + st->table->key_size > msg_end) {
1071 /* malformed message */
1072 stksess_free(st->table, newts);
1073 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1074 goto switchstate;
1075 }
1076 memcpy(newts->key.key, msg_cur, st->table->key_size);
1077 msg_cur += st->table->key_size;
1078 }
1079
1080 /* lookup for existing entry */
1081 ts = stktable_lookup(st->table, newts);
1082 if (ts) {
1083 /* the entry already exist, we can free ours */
1084 stktable_touch(st->table, ts, 0);
1085 stksess_free(st->table, newts);
1086 newts = NULL;
1087 }
1088 else {
1089 struct eb32_node *eb;
1090
1091 /* create new entry */
1092 ts = stktable_store(st->table, newts, 0);
1093 newts = NULL; /* don't reuse it */
1094
Emeric Brun234fc3c2015-12-16 15:16:46 +01001095 ts->upd.key= (++st->table->update)+(2147483648U);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001096 eb = eb32_insert(&st->table->updates, &ts->upd);
1097 if (eb != &ts->upd) {
1098 eb32_delete(eb);
1099 eb32_insert(&st->table->updates, &ts->upd);
1100 }
1101 }
1102
Emeric Brun94900952015-06-11 18:25:54 +02001103 for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001104
Emeric Brun94900952015-06-11 18:25:54 +02001105 if ((1 << data_type) & st->remote_data) {
1106 switch (stktable_data_types[data_type].std_type) {
1107 case STD_T_SINT: {
1108 int data;
1109
1110 data = intdecode(&msg_cur, msg_end);
1111 if (!msg_cur) {
1112 /* malformed message */
1113 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1114 goto switchstate;
1115 }
1116
1117 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1118 if (data_ptr)
1119 stktable_data_cast(data_ptr, std_t_sint) = data;
1120 break;
1121 }
1122 case STD_T_UINT: {
1123 unsigned int data;
1124
1125 data = intdecode(&msg_cur, msg_end);
1126 if (!msg_cur) {
1127 /* malformed message */
1128 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1129 goto switchstate;
1130 }
1131
1132 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1133 if (data_ptr)
1134 stktable_data_cast(data_ptr, std_t_uint) = data;
1135 break;
1136 }
1137 case STD_T_ULL: {
1138 unsigned long long data;
1139
1140 data = intdecode(&msg_cur, msg_end);
1141 if (!msg_cur) {
1142 /* malformed message */
1143 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1144 goto switchstate;
1145 }
1146
1147 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1148 if (data_ptr)
1149 stktable_data_cast(data_ptr, std_t_ull) = data;
1150 break;
1151 }
1152 case STD_T_FRQP: {
1153 struct freq_ctr_period data;
1154
Willy Tarreau3bb46172016-03-25 18:17:47 +01001155 data.curr_tick = tick_add(now_ms, -intdecode(&msg_cur, msg_end));
Emeric Brun94900952015-06-11 18:25:54 +02001156 if (!msg_cur) {
1157 /* malformed message */
1158 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1159 goto switchstate;
1160 }
1161 data.curr_ctr = intdecode(&msg_cur, msg_end);
1162 if (!msg_cur) {
1163 /* malformed message */
1164 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1165 goto switchstate;
1166 }
1167 data.prev_ctr = intdecode(&msg_cur, msg_end);
1168 if (!msg_cur) {
1169 /* malformed message */
1170 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1171 goto switchstate;
1172 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001173
Emeric Brun94900952015-06-11 18:25:54 +02001174 data_ptr = stktable_data_ptr(st->table, ts, data_type);
1175 if (data_ptr)
1176 stktable_data_cast(data_ptr, std_t_frqp) = data;
1177 break;
1178 }
1179 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001180 }
1181 }
1182 }
1183 else if (msg_head[1] == PEER_MSG_STKT_ACK) {
1184 /* ack message */
1185 uint32_t table_id ;
1186 uint32_t update;
1187 struct shared_table *st;
1188
1189 table_id = intdecode(&msg_cur, msg_end);
1190 if (!msg_cur || (msg_cur + sizeof(update) > msg_end)) {
1191 /* malformed message */
1192 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1193 goto switchstate;
1194 }
1195 memcpy(&update, msg_cur, sizeof(update));
1196 update = ntohl(update);
Emeric Brun2b920a12010-09-23 18:30:22 +02001197
Emeric Brunb3971ab2015-05-12 18:49:09 +02001198 for (st = curpeer->tables; st; st = st->next) {
1199 if (st->local_id == table_id) {
1200 st->update = update;
1201 break;
1202 }
1203 }
1204 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001205 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001206 else if (msg_head[0] == PEER_MSG_CLASS_RESERVED) {
1207 appctx->st0 = PEER_SESS_ST_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +02001208 goto switchstate;
1209 }
1210
Emeric Brunb3971ab2015-05-12 18:49:09 +02001211ignore_msg:
Emeric Brun2b920a12010-09-23 18:30:22 +02001212 /* skip consumed message */
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001213 bo_skip(si_oc(si), totl);
Emeric Brun2b920a12010-09-23 18:30:22 +02001214 /* loop on that state to peek next message */
Willy Tarreau72d6c162013-04-11 16:14:13 +02001215 goto switchstate;
1216
Emeric Brun2b920a12010-09-23 18:30:22 +02001217incomplete:
Willy Tarreau9d9179b2013-04-11 16:56:44 +02001218 /* we get here when a bo_getblk() returns <= 0 in reql */
1219
Willy Tarreau72d6c162013-04-11 16:14:13 +02001220 if (reql < 0) {
1221 /* there was an error */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001222 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau72d6c162013-04-11 16:14:13 +02001223 goto switchstate;
1224 }
1225
Emeric Brun2b920a12010-09-23 18:30:22 +02001226
1227 /* Confirm finished or partial messages */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001228 while (curpeer->confirm) {
1229 unsigned char msg[2];
1230
Emeric Brun2b920a12010-09-23 18:30:22 +02001231 /* There is a confirm messages to send */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001232 msg[0] = PEER_MSG_CLASS_CONTROL;
1233 msg[1] = PEER_MSG_CTRL_RESYNCCONFIRM;
1234
1235 /* message to buffer */
1236 repl = bi_putblk(si_ic(si), (char *)msg, sizeof(msg));
Emeric Brun2b920a12010-09-23 18:30:22 +02001237 if (repl <= 0) {
1238 /* no more write possible */
1239 if (repl == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +01001240 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001241 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02001242 goto switchstate;
1243 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001244 curpeer->confirm--;
Emeric Brun2b920a12010-09-23 18:30:22 +02001245 }
1246
Emeric Brunb3971ab2015-05-12 18:49:09 +02001247
Emeric Brun2b920a12010-09-23 18:30:22 +02001248 /* Need to request a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001249 if ((curpeer->flags & PEER_F_LEARN_ASSIGN) &&
1250 (peers->flags & PEERS_F_RESYNC_ASSIGN) &&
1251 !(peers->flags & PEERS_F_RESYNC_PROCESS)) {
1252 unsigned char msg[2];
Emeric Brun2b920a12010-09-23 18:30:22 +02001253
Emeric Brunb3971ab2015-05-12 18:49:09 +02001254 /* Current peer was elected to request a resync */
1255 msg[0] = PEER_MSG_CLASS_CONTROL;
1256 msg[1] = PEER_MSG_CTRL_RESYNCREQ;
Emeric Brun2b920a12010-09-23 18:30:22 +02001257
Emeric Brunb3971ab2015-05-12 18:49:09 +02001258 /* message to buffer */
1259 repl = bi_putblk(si_ic(si), (char *)msg, sizeof(msg));
1260 if (repl <= 0) {
1261 /* no more write possible */
1262 if (repl == -1)
1263 goto full;
1264 appctx->st0 = PEER_SESS_ST_END;
1265 goto switchstate;
1266 }
1267 peers->flags |= PEERS_F_RESYNC_PROCESS;
1268 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001269
Emeric Brunb3971ab2015-05-12 18:49:09 +02001270 /* Nothing to read, now we start to write */
Emeric Brun2b920a12010-09-23 18:30:22 +02001271
Emeric Brunb3971ab2015-05-12 18:49:09 +02001272 if (curpeer->tables) {
1273 struct shared_table *st;
1274 struct shared_table *last_local_table;
Emeric Brun2b920a12010-09-23 18:30:22 +02001275
Emeric Brunb3971ab2015-05-12 18:49:09 +02001276 last_local_table = curpeer->last_local_table;
1277 if (!last_local_table)
1278 last_local_table = curpeer->tables;
1279 st = last_local_table->next;
Emeric Brun2b920a12010-09-23 18:30:22 +02001280
Emeric Brunb3971ab2015-05-12 18:49:09 +02001281 while (1) {
1282 if (!st)
1283 st = curpeer->tables;
Emeric Brun2b920a12010-09-23 18:30:22 +02001284
Emeric Brunb3971ab2015-05-12 18:49:09 +02001285 /* It remains some updates to ack */
1286 if (st->last_get != st->last_acked) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001287 int msglen;
Emeric Brun2b920a12010-09-23 18:30:22 +02001288
Emeric Brunb3971ab2015-05-12 18:49:09 +02001289 msglen = peer_prepare_ackmsg(st, trash.str, trash.size);
1290 if (!msglen) {
1291 /* internal error: message does not fit in trash */
1292 appctx->st0 = PEER_SESS_ST_END;
1293 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001294 }
1295
Emeric Brunb3971ab2015-05-12 18:49:09 +02001296 /* message to buffer */
1297 repl = bi_putblk(si_ic(si), trash.str, msglen);
1298 if (repl <= 0) {
1299 /* no more write possible */
1300 if (repl == -1) {
1301 goto full;
Emeric Brun2b920a12010-09-23 18:30:22 +02001302 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001303 appctx->st0 = PEER_SESS_ST_END;
1304 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001305 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001306 st->last_acked = st->last_get;
Emeric Brun2b920a12010-09-23 18:30:22 +02001307 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001308
Emeric Brunb3971ab2015-05-12 18:49:09 +02001309 if (!(curpeer->flags & PEER_F_TEACH_PROCESS)) {
1310 if (!(curpeer->flags & PEER_F_LEARN_ASSIGN) &&
1311 ((int)(st->last_pushed - st->table->localupdate) < 0)) {
1312 struct eb32_node *eb;
1313 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001314
Emeric Brunb3971ab2015-05-12 18:49:09 +02001315 if (st != curpeer->last_local_table) {
1316 int msglen;
Emeric Brun2b920a12010-09-23 18:30:22 +02001317
Emeric Brunb3971ab2015-05-12 18:49:09 +02001318 msglen = peer_prepare_switchmsg(st, trash.str, trash.size);
1319 if (!msglen) {
1320 /* internal error: message does not fit in trash */
1321 appctx->st0 = PEER_SESS_ST_END;
1322 goto switchstate;
1323 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001324
Emeric Brunb3971ab2015-05-12 18:49:09 +02001325 /* message to buffer */
1326 repl = bi_putblk(si_ic(si), trash.str, msglen);
1327 if (repl <= 0) {
1328 /* no more write possible */
1329 if (repl == -1) {
1330 goto full;
1331 }
1332 appctx->st0 = PEER_SESS_ST_END;
1333 goto switchstate;
1334 }
1335 curpeer->last_local_table = st;
Emeric Brun2b920a12010-09-23 18:30:22 +02001336 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001337
1338 /* We force new pushed to 1 to force identifier in update message */
1339 new_pushed = 1;
1340 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
1341 while (1) {
1342 uint32_t msglen;
1343 struct stksess *ts;
1344
1345 /* push local updates */
1346 if (!eb) {
1347 eb = eb32_first(&st->table->updates);
1348 if (!eb || ((int)(eb->key - st->last_pushed) <= 0)) {
Emeric Brunaaf58602015-06-15 17:23:30 +02001349 st->table->commitupdate = st->last_pushed = st->table->localupdate;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001350 break;
1351 }
1352 }
1353
1354 if ((int)(eb->key - st->table->localupdate) > 0) {
Emeric Brunaaf58602015-06-15 17:23:30 +02001355 st->table->commitupdate = st->last_pushed = st->table->localupdate;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001356 break;
1357 }
1358
1359 ts = eb32_entry(eb, struct stksess, upd);
1360 msglen = peer_prepare_updatemsg(ts, st, trash.str, trash.size, new_pushed);
1361 if (!msglen) {
1362 /* internal error: message does not fit in trash */
1363 appctx->st0 = PEER_SESS_ST_END;
1364 goto switchstate;
1365 }
1366
1367 /* message to buffer */
1368 repl = bi_putblk(si_ic(si), trash.str, msglen);
1369 if (repl <= 0) {
1370 /* no more write possible */
1371 if (repl == -1) {
1372 goto full;
1373 }
1374 appctx->st0 = PEER_SESS_ST_END;
1375 goto switchstate;
1376 }
1377 st->last_pushed = ts->upd.key;
Emeric Brunaaf58602015-06-15 17:23:30 +02001378 if ((int)(st->last_pushed - st->table->commitupdate) > 0)
1379 st->table->commitupdate = st->last_pushed;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001380 /* identifier may not needed in next update message */
1381 new_pushed = 0;
1382
1383 eb = eb32_next(eb);
1384 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001385 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001386 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001387 else {
1388 if (!(st->flags & SHTABLE_F_TEACH_STAGE1)) {
1389 struct eb32_node *eb;
1390 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001391
Emeric Brunb3971ab2015-05-12 18:49:09 +02001392 if (st != curpeer->last_local_table) {
1393 int msglen;
1394
1395 msglen = peer_prepare_switchmsg(st, trash.str, trash.size);
1396 if (!msglen) {
1397 /* internal error: message does not fit in trash */
1398 appctx->st0 = PEER_SESS_ST_END;
1399 goto switchstate;
1400 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001401
Emeric Brunb3971ab2015-05-12 18:49:09 +02001402 /* message to buffer */
1403 repl = bi_putblk(si_ic(si), trash.str, msglen);
1404 if (repl <= 0) {
1405 /* no more write possible */
1406 if (repl == -1) {
1407 goto full;
1408 }
1409 appctx->st0 = PEER_SESS_ST_END;
1410 goto switchstate;
1411 }
1412 curpeer->last_local_table = st;
1413 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001414
Emeric Brunb3971ab2015-05-12 18:49:09 +02001415 /* We force new pushed to 1 to force identifier in update message */
1416 new_pushed = 1;
1417 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
1418 while (1) {
1419 uint32_t msglen;
1420 struct stksess *ts;
Emeric Brun2b920a12010-09-23 18:30:22 +02001421
Emeric Brunb3971ab2015-05-12 18:49:09 +02001422 /* push local updates */
1423 if (!eb) {
1424 st->flags |= SHTABLE_F_TEACH_STAGE1;
1425 eb = eb32_first(&st->table->updates);
1426 if (eb)
1427 st->last_pushed = eb->key - 1;
1428 break;
1429 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001430
Emeric Brunb3971ab2015-05-12 18:49:09 +02001431 ts = eb32_entry(eb, struct stksess, upd);
1432 msglen = peer_prepare_updatemsg(ts, st, trash.str, trash.size, new_pushed);
1433 if (!msglen) {
1434 /* internal error: message does not fit in trash */
1435 appctx->st0 = PEER_SESS_ST_END;
1436 goto switchstate;
1437 }
1438
1439 /* message to buffer */
1440 repl = bi_putblk(si_ic(si), trash.str, msglen);
1441 if (repl <= 0) {
1442 /* no more write possible */
1443 if (repl == -1) {
1444 goto full;
1445 }
1446 appctx->st0 = PEER_SESS_ST_END;
1447 goto switchstate;
1448 }
1449 st->last_pushed = ts->upd.key;
1450 /* identifier may not needed in next update message */
1451 new_pushed = 0;
1452
1453 eb = eb32_next(eb);
1454 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001455 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001456
Emeric Brunb3971ab2015-05-12 18:49:09 +02001457 if (!(st->flags & SHTABLE_F_TEACH_STAGE2)) {
1458 struct eb32_node *eb;
1459 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001460
Emeric Brunb3971ab2015-05-12 18:49:09 +02001461 if (st != curpeer->last_local_table) {
1462 int msglen;
1463
1464 msglen = peer_prepare_switchmsg(st, trash.str, trash.size);
1465 if (!msglen) {
1466 /* internal error: message does not fit in trash */
1467 appctx->st0 = PEER_SESS_ST_END;
1468 goto switchstate;
1469 }
1470
1471 /* message to buffer */
1472 repl = bi_putblk(si_ic(si), trash.str, msglen);
1473 if (repl <= 0) {
1474 /* no more write possible */
1475 if (repl == -1) {
1476 goto full;
1477 }
1478 appctx->st0 = PEER_SESS_ST_END;
1479 goto switchstate;
1480 }
1481 curpeer->last_local_table = st;
1482 }
1483
1484 /* We force new pushed to 1 to force identifier in update message */
1485 new_pushed = 1;
1486 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
1487 while (1) {
1488 uint32_t msglen;
1489 struct stksess *ts;
1490
1491 /* push local updates */
1492 if (!eb || eb->key > st->teaching_origin) {
1493 st->flags |= SHTABLE_F_TEACH_STAGE2;
1494 eb = eb32_first(&st->table->updates);
1495 if (eb)
1496 st->last_pushed = eb->key - 1;
1497 break;
1498 }
1499
1500 ts = eb32_entry(eb, struct stksess, upd);
1501 msglen = peer_prepare_updatemsg(ts, st, trash.str, trash.size, new_pushed);
1502 if (!msglen) {
1503 /* internal error: message does not fit in trash */
1504 appctx->st0 = PEER_SESS_ST_END;
1505 goto switchstate;
1506 }
1507
1508 /* message to buffer */
1509 repl = bi_putblk(si_ic(si), trash.str, msglen);
1510 if (repl <= 0) {
1511 /* no more write possible */
1512 if (repl == -1) {
1513 goto full;
1514 }
1515 appctx->st0 = PEER_SESS_ST_END;
1516 goto switchstate;
1517 }
1518 st->last_pushed = ts->upd.key;
1519 /* identifier may not needed in next update message */
1520 new_pushed = 0;
1521
1522 eb = eb32_next(eb);
1523 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001524 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001525 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001526
1527 if (st == last_local_table)
1528 break;
1529 st = st->next;
Emeric Brun2b920a12010-09-23 18:30:22 +02001530 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001531 }
1532
1533
1534 if ((curpeer->flags & PEER_F_TEACH_PROCESS) && !(curpeer->flags & PEER_F_TEACH_FINISHED)) {
1535 unsigned char msg[2];
1536
1537 /* Current peer was elected to request a resync */
1538 msg[0] = PEER_MSG_CLASS_CONTROL;
1539 msg[1] = ((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FINISHED) ? PEER_MSG_CTRL_RESYNCFINISHED : PEER_MSG_CTRL_RESYNCPARTIAL;
1540 /* process final lesson message */
1541 repl = bi_putblk(si_ic(si), (char *)msg, sizeof(msg));
1542 if (repl <= 0) {
1543 /* no more write possible */
1544 if (repl == -1)
1545 goto full;
1546 appctx->st0 = PEER_SESS_ST_END;
1547 goto switchstate;
1548 }
1549 /* flag finished message sent */
1550 curpeer->flags |= PEER_F_TEACH_FINISHED;
1551 }
1552
Emeric Brun2b920a12010-09-23 18:30:22 +02001553 /* noting more to do */
1554 goto out;
1555 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001556 case PEER_SESS_ST_EXIT:
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001557 repl = snprintf(trash.str, trash.size, "%d\n", appctx->st1);
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001558 if (bi_putblk(si_ic(si), trash.str, repl) == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +01001559 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001560 appctx->st0 = PEER_SESS_ST_END;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001561 goto switchstate;
1562 case PEER_SESS_ST_ERRSIZE: {
1563 unsigned char msg[2];
1564
1565 msg[0] = PEER_MSG_CLASS_ERROR;
1566 msg[1] = PEER_MSG_ERR_SIZELIMIT;
1567
1568 if (bi_putblk(si_ic(si), (char *)msg, sizeof(msg)) == -1)
1569 goto full;
1570 appctx->st0 = PEER_SESS_ST_END;
1571 goto switchstate;
1572 }
1573 case PEER_SESS_ST_ERRPROTO: {
1574 unsigned char msg[2];
1575
1576 msg[0] = PEER_MSG_CLASS_ERROR;
1577 msg[1] = PEER_MSG_ERR_PROTOCOL;
1578
1579 if (bi_putblk(si_ic(si), (char *)msg, sizeof(msg)) == -1)
1580 goto full;
1581 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02001582 /* fall through */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001583 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001584 case PEER_SESS_ST_END: {
Willy Tarreau73b013b2012-05-21 16:31:45 +02001585 si_shutw(si);
1586 si_shutr(si);
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001587 si_ic(si)->flags |= CF_READ_NULL;
Willy Tarreau828824a2015-04-19 17:20:03 +02001588 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001589 }
1590 }
1591 }
1592out:
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001593 si_oc(si)->flags |= CF_READ_DONTWAIT;
Emeric Brun2b920a12010-09-23 18:30:22 +02001594 return;
Willy Tarreaubc18da12015-03-13 14:00:47 +01001595full:
Willy Tarreaufe127932015-04-21 19:23:39 +02001596 si_applet_cant_put(si);
Willy Tarreaubc18da12015-03-13 14:00:47 +01001597 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001598}
1599
Willy Tarreau30576452015-04-13 13:50:30 +02001600static struct applet peer_applet = {
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001601 .obj_type = OBJ_TYPE_APPLET,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001602 .name = "<PEER>", /* used for logging */
1603 .fct = peer_io_handler,
Aman Gupta9a13e842012-04-02 18:57:53 -07001604 .release = peer_session_release,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001605};
Emeric Brun2b920a12010-09-23 18:30:22 +02001606
1607/*
1608 * Use this function to force a close of a peer session
1609 */
Willy Tarreau87b09662015-04-03 00:22:06 +02001610static void peer_session_forceshutdown(struct stream * stream)
Emeric Brun2b920a12010-09-23 18:30:22 +02001611{
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001612 struct appctx *appctx = NULL;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001613 struct peer *ps;
1614
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001615 int i;
Emeric Brun2b920a12010-09-23 18:30:22 +02001616
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001617 for (i = 0; i <= 1; i++) {
Willy Tarreau87b09662015-04-03 00:22:06 +02001618 appctx = objt_appctx(stream->si[i].end);
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001619 if (!appctx)
1620 continue;
1621 if (appctx->applet != &peer_applet)
1622 continue;
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001623 break;
Emeric Brun2b920a12010-09-23 18:30:22 +02001624 }
1625
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001626 if (!appctx)
1627 return;
1628
Emeric Brunb3971ab2015-05-12 18:49:09 +02001629 ps = (struct peer *)appctx->ctx.peers.ptr;
Willy Tarreaub4e34da2015-05-20 10:39:04 +02001630 /* we're killing a connection, we must apply a random delay before
1631 * retrying otherwise the other end will do the same and we can loop
1632 * for a while.
1633 */
1634 if (ps)
1635 ps->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + random() % 2000));
1636
Emeric Brun2b920a12010-09-23 18:30:22 +02001637 /* call release to reinit resync states if needed */
Willy Tarreau00a37f02015-04-13 12:05:19 +02001638 peer_session_release(appctx);
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001639 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001640 appctx->ctx.peers.ptr = NULL;
Willy Tarreau87b09662015-04-03 00:22:06 +02001641 task_wakeup(stream->task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +02001642}
1643
Willy Tarreau91d96282015-03-13 15:47:26 +01001644/* Pre-configures a peers frontend to accept incoming connections */
1645void peers_setup_frontend(struct proxy *fe)
1646{
1647 fe->last_change = now.tv_sec;
1648 fe->cap = PR_CAP_FE;
1649 fe->maxconn = 0;
1650 fe->conn_retries = CONN_RETRIES;
1651 fe->timeout.client = MS_TO_TICKS(5000);
Willy Tarreaud1d48d42015-03-13 16:15:46 +01001652 fe->accept = frontend_accept;
Willy Tarreauf87ab942015-03-13 15:55:16 +01001653 fe->default_target = &peer_applet.obj_type;
Willy Tarreau91d96282015-03-13 15:47:26 +01001654 fe->options2 |= PR_O2_INDEPSTR | PR_O2_SMARTCON | PR_O2_SMARTACC;
Willy Tarreau0fca4832015-05-01 19:12:05 +02001655 fe->bind_proc = 0; /* will be filled by users */
Willy Tarreau91d96282015-03-13 15:47:26 +01001656}
1657
Emeric Brun2b920a12010-09-23 18:30:22 +02001658/*
Willy Tarreaubd55e312010-11-11 10:55:09 +01001659 * Create a new peer session in assigned state (connect will start automatically)
Emeric Brun2b920a12010-09-23 18:30:22 +02001660 */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001661static struct stream *peer_session_create(struct peers *peers, struct peer *peer)
Emeric Brun2b920a12010-09-23 18:30:22 +02001662{
Emeric Brunb3971ab2015-05-12 18:49:09 +02001663 struct listener *l = LIST_NEXT(&peers->peers_fe->conf.listeners, struct listener *, by_fe);
Emeric Brun2b920a12010-09-23 18:30:22 +02001664 struct proxy *p = (struct proxy *)l->frontend; /* attached frontend */
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001665 struct appctx *appctx;
Willy Tarreau15b5e142015-04-04 14:38:25 +02001666 struct session *sess;
Willy Tarreau87b09662015-04-03 00:22:06 +02001667 struct stream *s;
Emeric Brun2b920a12010-09-23 18:30:22 +02001668 struct task *t;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001669 struct connection *conn;
Emeric Brun2b920a12010-09-23 18:30:22 +02001670
Emeric Brunb3971ab2015-05-12 18:49:09 +02001671 peer->reconnect = tick_add(now_ms, MS_TO_TICKS(5000));
1672 peer->statuscode = PEER_SESS_SC_CONNECTCODE;
Willy Tarreaud990baf2015-04-05 00:32:03 +02001673 s = NULL;
1674
1675 appctx = appctx_new(&peer_applet);
1676 if (!appctx)
1677 goto out_close;
1678
1679 appctx->st0 = PEER_SESS_ST_CONNECT;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001680 appctx->ctx.peers.ptr = (void *)peer;
Willy Tarreaud990baf2015-04-05 00:32:03 +02001681
Willy Tarreau4099a7c2015-04-05 00:39:55 +02001682 sess = session_new(p, l, &appctx->obj_type);
Willy Tarreau15b5e142015-04-04 14:38:25 +02001683 if (!sess) {
Godbach430f2912013-06-20 13:28:38 +08001684 Alert("out of memory in peer_session_create().\n");
Willy Tarreaud990baf2015-04-05 00:32:03 +02001685 goto out_free_appctx;
Emeric Brun2b920a12010-09-23 18:30:22 +02001686 }
1687
Willy Tarreau8baf9062015-04-05 00:46:36 +02001688 if ((t = task_new()) == NULL) {
Willy Tarreau15b5e142015-04-04 14:38:25 +02001689 Alert("out of memory in peer_session_create().\n");
1690 goto out_free_sess;
1691 }
Willy Tarreau8baf9062015-04-05 00:46:36 +02001692 t->nice = l->nice;
1693
Willy Tarreau73b65ac2015-04-08 18:26:29 +02001694 if ((s = stream_new(sess, t, &appctx->obj_type)) == NULL) {
Willy Tarreau342bfb12015-04-05 01:35:34 +02001695 Alert("Failed to initialize stream in peer_session_create().\n");
Willy Tarreau8baf9062015-04-05 00:46:36 +02001696 goto out_free_task;
1697 }
1698
Willy Tarreau342bfb12015-04-05 01:35:34 +02001699 /* The tasks below are normally what is supposed to be done by
1700 * fe->accept().
1701 */
Willy Tarreaue7dff022015-04-03 01:14:29 +02001702 s->flags = SF_ASSIGNED|SF_ADDR_SET;
Emeric Brun2b920a12010-09-23 18:30:22 +02001703
Willy Tarreau6e2979c2015-04-27 13:21:15 +02001704 /* applet is waiting for data */
1705 si_applet_cant_get(&s->si[0]);
1706 appctx_wakeup(appctx);
1707
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001708 /* initiate an outgoing connection */
1709 si_set_state(&s->si[1], SI_ST_ASS);
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001710
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001711 /* automatically prepare the stream interface to connect to the
Willy Tarreaub363a1f2013-10-01 10:45:07 +02001712 * pre-initialized connection in si->conn.
1713 */
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001714 if (unlikely((conn = conn_new()) == NULL))
Willy Tarreau8baf9062015-04-05 00:46:36 +02001715 goto out_free_strm;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001716
1717 conn_prepare(conn, peer->proto, peer->xprt);
1718 si_attach_conn(&s->si[1], conn);
1719
1720 conn->target = s->target = &s->be->obj_type;
1721 memcpy(&conn->addr.to, &peer->addr, sizeof(conn->addr.to));
Emeric Brun2b920a12010-09-23 18:30:22 +02001722 s->do_log = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001723 s->uniq_id = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +02001724
Willy Tarreau22ec1ea2014-11-27 20:45:39 +01001725 s->res.flags |= CF_READ_DONTWAIT;
Willy Tarreau696a2912014-11-24 11:36:57 +01001726
Emeric Brun2b920a12010-09-23 18:30:22 +02001727 l->nbconn++; /* warning! right now, it's up to the handler to decrease this */
1728 p->feconn++;/* beconn will be increased later */
1729 jobs++;
Willy Tarreaufb0afa72015-04-03 14:46:27 +02001730 if (!(s->sess->listener->options & LI_O_UNLIMITED))
Willy Tarreau3c63fd82011-09-07 18:00:47 +02001731 actconn++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001732 totalconn++;
1733
Emeric Brunb3971ab2015-05-12 18:49:09 +02001734 peer->appctx = appctx;
1735 peer->stream = s;
Emeric Brun2b920a12010-09-23 18:30:22 +02001736 return s;
1737
1738 /* Error unrolling */
Willy Tarreau15b5e142015-04-04 14:38:25 +02001739 out_free_strm:
Emeric Brun2b920a12010-09-23 18:30:22 +02001740 LIST_DEL(&s->list);
Willy Tarreau87b09662015-04-03 00:22:06 +02001741 pool_free2(pool2_stream, s);
Willy Tarreau8baf9062015-04-05 00:46:36 +02001742 out_free_task:
1743 task_free(t);
Willy Tarreau15b5e142015-04-04 14:38:25 +02001744 out_free_sess:
Willy Tarreau11c36242015-04-04 15:54:03 +02001745 session_free(sess);
Willy Tarreaud990baf2015-04-05 00:32:03 +02001746 out_free_appctx:
1747 appctx_free(appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02001748 out_close:
1749 return s;
1750}
1751
1752/*
1753 * Task processing function to manage re-connect and peer session
1754 * tasks wakeup on local update.
1755 */
Simon Horman96553772011-06-08 09:18:51 +09001756static struct task *process_peer_sync(struct task * task)
Emeric Brun2b920a12010-09-23 18:30:22 +02001757{
Emeric Brunb3971ab2015-05-12 18:49:09 +02001758 struct peers *peers = (struct peers *)task->context;
1759 struct peer *ps;
1760 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +02001761
1762 task->expire = TICK_ETERNITY;
1763
Emeric Brunb3971ab2015-05-12 18:49:09 +02001764 if (!peers->peers_fe) {
Willy Tarreau46dc1ca2015-05-01 18:32:13 +02001765 /* this one was never started, kill it */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001766 signal_unregister_handler(peers->sighandler);
Emeric Brunb3971ab2015-05-12 18:49:09 +02001767 task_delete(peers->sync_task);
1768 task_free(peers->sync_task);
Willy Tarreau37bb7be2015-09-21 15:24:58 +02001769 peers->sync_task = NULL;
Willy Tarreau46dc1ca2015-05-01 18:32:13 +02001770 return NULL;
1771 }
1772
Emeric Brun2b920a12010-09-23 18:30:22 +02001773 if (!stopping) {
1774 /* Normal case (not soft stop)*/
Emeric Brunb3971ab2015-05-12 18:49:09 +02001775
1776 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL) &&
1777 (!nb_oldpids || tick_is_expired(peers->resync_timeout, now_ms)) &&
1778 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001779 /* Resync from local peer needed
1780 no peer was assigned for the lesson
1781 and no old local peer found
1782 or resync timeout expire */
1783
1784 /* flag no more resync from local, to try resync from remotes */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001785 peers->flags |= PEERS_F_RESYNC_LOCAL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001786
1787 /* reschedule a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001788 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
Emeric Brun2b920a12010-09-23 18:30:22 +02001789 }
1790
1791 /* For each session */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001792 for (ps = peers->remote; ps; ps = ps->next) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001793 /* For each remote peers */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001794 if (!ps->local) {
Willy Tarreau87b09662015-04-03 00:22:06 +02001795 if (!ps->stream) {
1796 /* no active stream */
Emeric Brun2b920a12010-09-23 18:30:22 +02001797 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001798 ((ps->statuscode == PEER_SESS_SC_CONNECTCODE ||
Willy Tarreaub4e34da2015-05-20 10:39:04 +02001799 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001800 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02001801 tick_is_expired(ps->reconnect, now_ms))) {
1802 /* connection never tried
Willy Tarreau87b09662015-04-03 00:22:06 +02001803 * or previous stream established with success
1804 * or previous stream failed during connection
Emeric Brun2b920a12010-09-23 18:30:22 +02001805 * and reconnection timer is expired */
1806
1807 /* retry a connect */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001808 ps->stream = peer_session_create(peers, ps);
Emeric Brun2b920a12010-09-23 18:30:22 +02001809 }
Willy Tarreaub4e34da2015-05-20 10:39:04 +02001810 else if (!tick_is_expired(ps->reconnect, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001811 /* If previous session failed during connection
1812 * but reconnection timer is not expired */
1813
1814 /* reschedule task for reconnect */
1815 task->expire = tick_first(task->expire, ps->reconnect);
1816 }
1817 /* else do nothing */
Willy Tarreau87b09662015-04-03 00:22:06 +02001818 } /* !ps->stream */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001819 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Willy Tarreau87b09662015-04-03 00:22:06 +02001820 /* current stream is active and established */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001821 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
1822 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02001823 !(ps->flags & PEER_F_LEARN_NOTUP2DATE)) {
1824 /* Resync from a remote is needed
1825 * and no peer was assigned for lesson
1826 * and current peer may be up2date */
1827
1828 /* assign peer for the lesson */
1829 ps->flags |= PEER_F_LEARN_ASSIGN;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001830 peers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brun2b920a12010-09-23 18:30:22 +02001831
Willy Tarreau87b09662015-04-03 00:22:06 +02001832 /* awake peer stream task to handle a request of resync */
Willy Tarreaue5843b32015-04-27 18:40:14 +02001833 appctx_wakeup(ps->appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02001834 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001835 else {
1836 /* Awake session if there is data to push */
1837 for (st = ps->tables; st ; st = st->next) {
1838 if ((int)(st->last_pushed - st->table->localupdate) < 0) {
1839 /* awake peer stream task to push local updates */
1840 appctx_wakeup(ps->appctx);
1841 break;
1842 }
1843 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001844 }
1845 /* else do nothing */
1846 } /* SUCCESSCODE */
1847 } /* !ps->peer->local */
1848 } /* for */
1849
1850 /* Resync from remotes expired: consider resync is finished */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001851 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
1852 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
1853 tick_is_expired(peers->resync_timeout, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001854 /* Resync from remote peer needed
1855 * no peer was assigned for the lesson
1856 * and resync timeout expire */
1857
1858 /* flag no more resync from remote, consider resync is finished */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001859 peers->flags |= PEERS_F_RESYNC_REMOTE;
Emeric Brun2b920a12010-09-23 18:30:22 +02001860 }
1861
Emeric Brunb3971ab2015-05-12 18:49:09 +02001862 if ((peers->flags & PEERS_RESYNC_STATEMASK) != PEERS_RESYNC_FINISHED) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001863 /* Resync not finished*/
1864 /* reschedule task to resync timeout, to ended resync if needed */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001865 task->expire = tick_first(task->expire, peers->resync_timeout);
Emeric Brun2b920a12010-09-23 18:30:22 +02001866 }
1867 } /* !stopping */
1868 else {
1869 /* soft stop case */
1870 if (task->state & TASK_WOKEN_SIGNAL) {
1871 /* We've just recieved the signal */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001872 if (!(peers->flags & PEERS_F_DONOTSTOP)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001873 /* add DO NOT STOP flag if not present */
1874 jobs++;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001875 peers->flags |= PEERS_F_DONOTSTOP;
1876 ps = peers->local;
1877 for (st = ps->tables; st ; st = st->next)
1878 st->table->syncing++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001879 }
1880
1881 /* disconnect all connected peers */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001882 for (ps = peers->remote; ps; ps = ps->next) {
Willy Tarreau87b09662015-04-03 00:22:06 +02001883 if (ps->stream) {
1884 peer_session_forceshutdown(ps->stream);
1885 ps->stream = NULL;
Willy Tarreaue5843b32015-04-27 18:40:14 +02001886 ps->appctx = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001887 }
1888 }
1889 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001890
Emeric Brunb3971ab2015-05-12 18:49:09 +02001891 ps = peers->local;
Emeric Brun2b920a12010-09-23 18:30:22 +02001892 if (ps->flags & PEER_F_TEACH_COMPLETE) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001893 if (peers->flags & PEERS_F_DONOTSTOP) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001894 /* resync of new process was complete, current process can die now */
1895 jobs--;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001896 peers->flags &= ~PEERS_F_DONOTSTOP;
1897 for (st = ps->tables; st ; st = st->next)
1898 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02001899 }
1900 }
Willy Tarreau87b09662015-04-03 00:22:06 +02001901 else if (!ps->stream) {
1902 /* If stream is not active */
Emeric Brun2b920a12010-09-23 18:30:22 +02001903 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001904 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
1905 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE ||
1906 ps->statuscode == PEER_SESS_SC_TRYAGAIN) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001907 /* connection never tried
Willy Tarreau87b09662015-04-03 00:22:06 +02001908 * or previous stream was successfully established
1909 * or previous stream tcp connect success but init state incomplete
Emeric Brun2b920a12010-09-23 18:30:22 +02001910 * or during previous connect, peer replies a try again statuscode */
1911
1912 /* connect to the peer */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001913 peer_session_create(peers, ps);
Emeric Brun2b920a12010-09-23 18:30:22 +02001914 }
1915 else {
1916 /* Other error cases */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001917 if (peers->flags & PEERS_F_DONOTSTOP) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001918 /* unable to resync new process, current process can die now */
1919 jobs--;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001920 peers->flags &= ~PEERS_F_DONOTSTOP;
1921 for (st = ps->tables; st ; st = st->next)
1922 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02001923 }
1924 }
1925 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001926 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE ) {
Willy Tarreau87b09662015-04-03 00:22:06 +02001927 /* current stream active and established
1928 awake stream to push remaining local updates */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001929 for (st = ps->tables; st ; st = st->next) {
1930 if ((int)(st->last_pushed - st->table->localupdate) < 0) {
1931 /* awake peer stream task to push local updates */
1932 appctx_wakeup(ps->appctx);
1933 break;
1934 }
1935 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001936 }
1937 } /* stopping */
1938 /* Wakeup for re-connect */
1939 return task;
1940}
1941
Emeric Brunb3971ab2015-05-12 18:49:09 +02001942
Emeric Brun2b920a12010-09-23 18:30:22 +02001943/*
Emeric Brun2b920a12010-09-23 18:30:22 +02001944 *
1945 */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001946void peers_init_sync(struct peers *peers)
Emeric Brun2b920a12010-09-23 18:30:22 +02001947{
Emeric Brun2b920a12010-09-23 18:30:22 +02001948 struct peer * curpeer;
Willy Tarreau4348fad2012-09-20 16:48:07 +02001949 struct listener *listener;
Emeric Brun2b920a12010-09-23 18:30:22 +02001950
Emeric Brun2b920a12010-09-23 18:30:22 +02001951 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001952 peers->peers_fe->maxconn += 3;
1953 }
1954
Willy Tarreau4348fad2012-09-20 16:48:07 +02001955 list_for_each_entry(listener, &peers->peers_fe->conf.listeners, by_fe)
1956 listener->maxconn = peers->peers_fe->maxconn;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001957 peers->sync_task = task_new();
1958 peers->sync_task->process = process_peer_sync;
1959 peers->sync_task->expire = TICK_ETERNITY;
1960 peers->sync_task->context = (void *)peers;
1961 peers->sighandler = signal_register_task(0, peers->sync_task, 0);
1962 task_wakeup(peers->sync_task, TASK_WOKEN_INIT);
1963}
1964
1965
1966
1967/*
1968 * Function used to register a table for sync on a group of peers
1969 *
1970 */
1971void peers_register_table(struct peers *peers, struct stktable *table)
1972{
1973 struct shared_table *st;
1974 struct peer * curpeer;
1975 int id = 0;
1976
1977 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
1978 st = (struct shared_table *)calloc(1,sizeof(struct shared_table));
1979 st->table = table;
1980 st->next = curpeer->tables;
1981 if (curpeer->tables)
1982 id = curpeer->tables->local_id;
1983 st->local_id = id + 1;
1984
1985 curpeer->tables = st;
1986 }
1987
1988 table->sync_task = peers->sync_task;
Emeric Brun2b920a12010-09-23 18:30:22 +02001989}
1990