blob: 3174929227769e6c8dcaa94b719660a057c50929 [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;
245
246 cursor = datamsg = msg + 1 + 5;
247
Emeric Brun2b920a12010-09-23 18:30:22 +0200248 /* construct message */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200249
250 /* check if we need to send the update identifer */
251 if (st->last_pushed && ts->upd.key > st->last_pushed && (ts->upd.key - st->last_pushed) == 1) {
252 use_identifier = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200253 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200254
255 /* encode update identifier if needed */
256 if (use_identifier) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200257 netinteger = htonl(ts->upd.key);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200258 memcpy(cursor, &netinteger, sizeof(netinteger));
259 cursor += sizeof(netinteger);
Emeric Brun2b920a12010-09-23 18:30:22 +0200260 }
261
Emeric Brunb3971ab2015-05-12 18:49:09 +0200262 /* encode the key */
263 if (st->table->type == STKTABLE_TYPE_STRING) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200264 int stlen = strlen((char *)ts->key.key);
265
Emeric Brunb3971ab2015-05-12 18:49:09 +0200266 intencode(stlen, &cursor);
267 memcpy(cursor, ts->key.key, stlen);
268 cursor += stlen;
Emeric Brun2b920a12010-09-23 18:30:22 +0200269 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200270 else if (st->table->type == STKTABLE_TYPE_INTEGER) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200271 netinteger = htonl(*((uint32_t *)ts->key.key));
Emeric Brunb3971ab2015-05-12 18:49:09 +0200272 memcpy(cursor, &netinteger, sizeof(netinteger));
273 cursor += sizeof(netinteger);
Emeric Brun2b920a12010-09-23 18:30:22 +0200274 }
275 else {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200276 memcpy(cursor, ts->key.key, st->table->key_size);
277 cursor += st->table->key_size;
Emeric Brun2b920a12010-09-23 18:30:22 +0200278 }
279
Emeric Brunb3971ab2015-05-12 18:49:09 +0200280 /* encode values */
281 if (stktable_data_ptr(st->table, ts, STKTABLE_DT_SERVER_ID)) {
282 int srvid;
283
284 srvid = stktable_data_cast(stktable_data_ptr(st->table, ts, STKTABLE_DT_SERVER_ID), server_id);
285 intencode(srvid, &cursor);
286 }
287
288 /* Compute datalen */
289 datalen = (cursor - datamsg);
290
291 /* prepare message header */
292 msg[0] = PEER_MSG_CLASS_STICKTABLE;
293 if (use_identifier)
294 msg[1] = PEER_MSG_STKT_UPDATE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200295 else
Emeric Brunb3971ab2015-05-12 18:49:09 +0200296 msg[1] = PEER_MSG_STKT_INCUPDATE;
297
298 cursor = &msg[2];
299 intencode(datalen, &cursor);
300
301 /* move data after header */
302 memmove(cursor, datamsg, datalen);
303
304 /* return header size + data_len */
305 return (cursor - msg) + datalen;
306}
307
308/*
309 * This prepare the switch table message to targeted share table <st>.
310 * <msg> is a buffer of <size> to recieve data message content
311 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
312 * check size)
313 */
314static int peer_prepare_switchmsg(struct shared_table *st, char *msg, size_t size)
315{
316 int len;
317 unsigned short datalen;
318 char *cursor, *datamsg;
319 uint64_t data = 0;
320
321 cursor = datamsg = msg + 2 + 5;
322
323 /* Encode data */
324
325 /* encode local id */
326 intencode(st->local_id, &cursor);
327
328 /* encode table name */
329 len = strlen(st->table->id);
330 intencode(len, &cursor);
331 memcpy(cursor, st->table->id, len);
332 cursor += len;
333
334 /* encode table type */
335
336 intencode(st->table->type, &cursor);
337
338 /* encode table key size */
339 intencode(st->table->key_size, &cursor);
340
341 /* encode available data types in table */
342 if (st->table->data_ofs[STKTABLE_DT_SERVER_ID]) {
343 data |= 1 << STKTABLE_DT_SERVER_ID;
344 }
345 intencode(data, &cursor);
346
347 /* Compute datalen */
348 datalen = (cursor - datamsg);
Emeric Brun2b920a12010-09-23 18:30:22 +0200349
Emeric Brunb3971ab2015-05-12 18:49:09 +0200350 /* prepare message header */
351 msg[0] = PEER_MSG_CLASS_STICKTABLE;
352 msg[1] = PEER_MSG_STKT_DEFINE;
353 cursor = &msg[2];
354 intencode(datalen, &cursor);
Emeric Brun2b920a12010-09-23 18:30:22 +0200355
Emeric Brunb3971ab2015-05-12 18:49:09 +0200356 /* move data after header */
357 memmove(cursor, datamsg, datalen);
358
359 /* return header size + data_len */
360 return (cursor - msg) + datalen;
Emeric Brun2b920a12010-09-23 18:30:22 +0200361}
362
Emeric Brunb3971ab2015-05-12 18:49:09 +0200363/*
364 * This prepare the acknowledge message on the stick session <ts>, <st> is the considered
365 * stick table.
366 * <msg> is a buffer of <size> to recieve data message content
367 * If function returns 0, the caller should consider we were unable to encode this message (TODO:
368 * check size)
369 */
370static int peer_prepare_ackmsg(struct shared_table *st, char *msg, size_t size)
371{
372 unsigned short datalen;
373 char *cursor, *datamsg;
374 uint32_t netinteger;
375
376 cursor = datamsg = trash.str + 2 + 5;
377
378 intencode(st->remote_id, &cursor);
379 netinteger = htonl(st->last_get);
380 memcpy(cursor, &netinteger, sizeof(netinteger));
381 cursor += sizeof(netinteger);
382
383 /* Compute datalen */
384 datalen = (cursor - datamsg);
385
386 /* prepare message header */
387 msg[0] = PEER_MSG_CLASS_STICKTABLE;
388 msg[1] = PEER_MSG_STKT_DEFINE;
389 cursor = &msg[2];
390 intencode(datalen, &cursor);
391
392 /* move data after header */
393 memmove(cursor, datamsg, datalen);
394
395 /* return header size + data_len */
396 return (cursor - msg) + datalen;
397}
Emeric Brun2b920a12010-09-23 18:30:22 +0200398
399/*
400 * Callback to release a session with a peer
401 */
Willy Tarreau00a37f02015-04-13 12:05:19 +0200402static void peer_session_release(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +0200403{
Willy Tarreau00a37f02015-04-13 12:05:19 +0200404 struct stream_interface *si = appctx->owner;
Willy Tarreau87b09662015-04-03 00:22:06 +0200405 struct stream *s = si_strm(si);
Emeric Brunb3971ab2015-05-12 18:49:09 +0200406 struct peer *peer = (struct peer *)appctx->ctx.peers.ptr;
407 struct peers *peers = (struct peers *)strm_fe(s)->parent;
Emeric Brun2b920a12010-09-23 18:30:22 +0200408
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100409 /* appctx->ctx.peers.ptr is not a peer session */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100410 if (appctx->st0 < PEER_SESS_ST_SENDSUCCESS)
Emeric Brun2b920a12010-09-23 18:30:22 +0200411 return;
412
413 /* peer session identified */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200414 if (peer) {
415 if (peer->stream == s) {
416 peer->stream = NULL;
417 peer->appctx = NULL;
418 if (peer->flags & PEER_F_LEARN_ASSIGN) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200419 /* unassign current peer for learning */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200420 peer->flags &= ~(PEER_F_LEARN_ASSIGN);
421 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
Emeric Brun2b920a12010-09-23 18:30:22 +0200422
423 /* reschedule a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200424 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
Emeric Brun2b920a12010-09-23 18:30:22 +0200425 }
426 /* reset teaching and learning flags to 0 */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200427 peer->flags &= PEER_TEACH_RESET;
428 peer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200429 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200430 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200431 }
432}
433
434
435/*
436 * IO Handler to handle message exchance with a peer
437 */
Willy Tarreau00a37f02015-04-13 12:05:19 +0200438static void peer_io_handler(struct appctx *appctx)
Emeric Brun2b920a12010-09-23 18:30:22 +0200439{
Willy Tarreau00a37f02015-04-13 12:05:19 +0200440 struct stream_interface *si = appctx->owner;
Willy Tarreau87b09662015-04-03 00:22:06 +0200441 struct stream *s = si_strm(si);
Willy Tarreaud0d8da92015-04-04 02:10:38 +0200442 struct peers *curpeers = (struct peers *)strm_fe(s)->parent;
Emeric Brun2b920a12010-09-23 18:30:22 +0200443 int reql = 0;
444 int repl = 0;
445
446 while (1) {
447switchstate:
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100448 switch(appctx->st0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100449 case PEER_SESS_ST_ACCEPT:
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100450 appctx->ctx.peers.ptr = NULL;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100451 appctx->st0 = PEER_SESS_ST_GETVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200452 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100453 case PEER_SESS_ST_GETVERSION:
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100454 reql = bo_getline(si_oc(si), trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200455 if (reql <= 0) { /* closed or EOL not found */
456 if (reql == 0)
457 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100458 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200459 goto switchstate;
460 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100461 if (trash.str[reql-1] != '\n') {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100462 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200463 goto switchstate;
464 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100465 else if (reql > 1 && (trash.str[reql-2] == '\r'))
466 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200467 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100468 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200469
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100470 bo_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200471
472 /* test version */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200473 if (strcmp(PEER_SESSION_PROTO_NAME " 2.0", trash.str) != 0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100474 appctx->st0 = PEER_SESS_ST_EXIT;
475 appctx->st1 = PEER_SESS_SC_ERRVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200476 /* test protocol */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100477 if (strncmp(PEER_SESSION_PROTO_NAME " ", trash.str, strlen(PEER_SESSION_PROTO_NAME)+1) != 0)
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100478 appctx->st1 = PEER_SESS_SC_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +0200479 goto switchstate;
480 }
481
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100482 appctx->st0 = PEER_SESS_ST_GETHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200483 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100484 case PEER_SESS_ST_GETHOST:
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100485 reql = bo_getline(si_oc(si), trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200486 if (reql <= 0) { /* closed or EOL not found */
487 if (reql == 0)
488 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100489 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200490 goto switchstate;
491 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100492 if (trash.str[reql-1] != '\n') {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100493 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200494 goto switchstate;
495 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100496 else if (reql > 1 && (trash.str[reql-2] == '\r'))
497 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200498 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100499 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200500
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100501 bo_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200502
503 /* test hostname match */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100504 if (strcmp(localpeer, trash.str) != 0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100505 appctx->st0 = PEER_SESS_ST_EXIT;
506 appctx->st1 = PEER_SESS_SC_ERRHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200507 goto switchstate;
508 }
509
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100510 appctx->st0 = PEER_SESS_ST_GETPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200511 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100512 case PEER_SESS_ST_GETPEER: {
Emeric Brun2b920a12010-09-23 18:30:22 +0200513 struct peer *curpeer;
514 char *p;
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100515 reql = bo_getline(si_oc(si), trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200516 if (reql <= 0) { /* closed or EOL not found */
517 if (reql == 0)
518 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100519 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200520 goto switchstate;
521 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100522 if (trash.str[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200523 /* Incomplete line, we quit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100524 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200525 goto switchstate;
526 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100527 else if (reql > 1 && (trash.str[reql-2] == '\r'))
528 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200529 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100530 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200531
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100532 bo_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200533
Emeric Brunb3971ab2015-05-12 18:49:09 +0200534 /* parse line "<peer name> <pid> <relative_pid>" */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100535 p = strchr(trash.str, ' ');
Emeric Brun2b920a12010-09-23 18:30:22 +0200536 if (!p) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100537 appctx->st0 = PEER_SESS_ST_EXIT;
538 appctx->st1 = PEER_SESS_SC_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +0200539 goto switchstate;
540 }
541 *p = 0;
542
543 /* lookup known peer */
544 for (curpeer = curpeers->remote; curpeer; curpeer = curpeer->next) {
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100545 if (strcmp(curpeer->id, trash.str) == 0)
Emeric Brun2b920a12010-09-23 18:30:22 +0200546 break;
547 }
548
549 /* if unknown peer */
550 if (!curpeer) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100551 appctx->st0 = PEER_SESS_ST_EXIT;
552 appctx->st1 = PEER_SESS_SC_ERRPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200553 goto switchstate;
554 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200555
Emeric Brunb3971ab2015-05-12 18:49:09 +0200556 if (curpeer->stream && curpeer->stream != s) {
557 if (curpeer->local) {
558 /* Local connection, reply a retry */
559 appctx->st0 = PEER_SESS_ST_EXIT;
560 appctx->st1 = PEER_SESS_SC_TRYAGAIN;
561 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +0200562 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200563 peer_session_forceshutdown(curpeer->stream);
Emeric Brun2b920a12010-09-23 18:30:22 +0200564 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200565 curpeer->stream = s;
566 curpeer->appctx = appctx;
567 appctx->ctx.peers.ptr = curpeer;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100568 appctx->st0 = PEER_SESS_ST_SENDSUCCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200569 /* fall through */
570 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100571 case PEER_SESS_ST_SENDSUCCESS: {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200572 struct peer *curpeer = (struct peer *)appctx->ctx.peers.ptr;
573 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +0200574
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100575 repl = snprintf(trash.str, trash.size, "%d\n", PEER_SESS_SC_SUCCESSCODE);
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100576 repl = bi_putblk(si_ic(si), trash.str, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200577 if (repl <= 0) {
578 if (repl == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +0100579 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100580 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200581 goto switchstate;
582 }
583
584 /* Register status code */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200585 curpeer->statuscode = PEER_SESS_SC_SUCCESSCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200586
587 /* Awake main task */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200588 task_wakeup(curpeers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200589
590 /* Init confirm counter */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200591 curpeer->confirm = 0;
592
593 /* Init cursors */
594 for (st = curpeer->tables; st ; st = st->next) {
595 st->last_get = st->last_acked = 0;
596 st->teaching_origin = st->last_pushed = st->update;
597 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200598
599 /* reset teaching and learning flags to 0 */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200600 curpeer->flags &= PEER_TEACH_RESET;
601 curpeer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200602
603 /* if current peer is local */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200604 if (curpeer->local) {
605 /* if current host need resyncfrom local and no process assined */
606 if ((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL &&
607 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
608 /* assign local peer for a lesson, consider lesson already requested */
609 curpeer->flags |= PEER_F_LEARN_ASSIGN;
610 peers->flags |= (PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
611 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200612
Emeric Brunb3971ab2015-05-12 18:49:09 +0200613 }
614 else if ((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
615 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
616 /* assign peer for a lesson */
617 curpeer->flags |= PEER_F_LEARN_ASSIGN;
618 peers->flags |= PEERS_F_RESYNC_ASSIGN;
619 }
620
621
Emeric Brun2b920a12010-09-23 18:30:22 +0200622 /* switch to waiting message state */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100623 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200624 goto switchstate;
625 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100626 case PEER_SESS_ST_CONNECT: {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200627 struct peer *curpeer = (struct peer *)appctx->ctx.peers.ptr;
Emeric Brun2b920a12010-09-23 18:30:22 +0200628
629 /* Send headers */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100630 repl = snprintf(trash.str, trash.size,
Emeric Brunb3971ab2015-05-12 18:49:09 +0200631 PEER_SESSION_PROTO_NAME " 2.0\n%s\n%s %d %d\n",
632 curpeer->id,
Emeric Brun2b920a12010-09-23 18:30:22 +0200633 localpeer,
Willy Tarreau7b77c9f2012-01-07 22:52:12 +0100634 (int)getpid(),
Emeric Brunb3971ab2015-05-12 18:49:09 +0200635 relative_pid);
Emeric Brun2b920a12010-09-23 18:30:22 +0200636
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100637 if (repl >= trash.size) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100638 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200639 goto switchstate;
640 }
641
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100642 repl = bi_putblk(si_ic(si), trash.str, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200643 if (repl <= 0) {
644 if (repl == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +0100645 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100646 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200647 goto switchstate;
648 }
649
650 /* switch to the waiting statuscode state */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100651 appctx->st0 = PEER_SESS_ST_GETSTATUS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200652 /* fall through */
653 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100654 case PEER_SESS_ST_GETSTATUS: {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200655 struct peer *curpeer = (struct peer *)appctx->ctx.peers.ptr;
656 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +0200657
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100658 if (si_ic(si)->flags & CF_WRITE_PARTIAL)
Emeric Brunb3971ab2015-05-12 18:49:09 +0200659 curpeer->statuscode = PEER_SESS_SC_CONNECTEDCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200660
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100661 reql = bo_getline(si_oc(si), trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200662 if (reql <= 0) { /* closed or EOL not found */
663 if (reql == 0)
664 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100665 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200666 goto switchstate;
667 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100668 if (trash.str[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200669 /* Incomplete line, we quit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100670 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200671 goto switchstate;
672 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100673 else if (reql > 1 && (trash.str[reql-2] == '\r'))
674 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200675 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100676 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200677
Willy Tarreau2bb4a962014-11-28 11:11:05 +0100678 bo_skip(si_oc(si), reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200679
680 /* Register status code */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200681 curpeer->statuscode = atoi(trash.str);
Emeric Brun2b920a12010-09-23 18:30:22 +0200682
683 /* Awake main task */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200684 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +0200685
686 /* If status code is success */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200687 if (curpeer->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200688 /* Init cursors */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200689 for (st = curpeer->tables; st ; st = st->next) {
690 st->last_get = st->last_acked = 0;
691 st->teaching_origin = st->last_pushed = st->update;
692 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200693
694 /* Init confirm counter */
Emeric Brunb3971ab2015-05-12 18:49:09 +0200695 curpeer->confirm = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200696
Emeric Brunb3971ab2015-05-12 18:49:09 +0200697 /* reset teaching and learning flags to 0 */
698 curpeer->flags &= PEER_TEACH_RESET;
699 curpeer->flags &= PEER_LEARN_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200700
Emeric Brunb3971ab2015-05-12 18:49:09 +0200701 /* If current peer is local */
702 if (curpeer->local) {
703 /* flag to start to teach lesson */
704 curpeer->flags |= PEER_F_TEACH_PROCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200705
Emeric Brunb3971ab2015-05-12 18:49:09 +0200706 }
707 else if ((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE &&
708 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
709 /* If peer is remote and resync from remote is needed,
710 and no peer currently assigned */
Emeric Brun2b920a12010-09-23 18:30:22 +0200711
Emeric Brunb3971ab2015-05-12 18:49:09 +0200712 /* assign peer for a lesson */
713 curpeer->flags |= PEER_F_LEARN_ASSIGN;
714 peers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brun2b920a12010-09-23 18:30:22 +0200715 }
716
717 }
718 else {
719 /* Status code is not success, abort */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100720 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200721 goto switchstate;
722 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100723 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200724 /* fall through */
725 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100726 case PEER_SESS_ST_WAITMSG: {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200727 struct peer *curpeer = (struct peer *)appctx->ctx.peers.ptr;
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200728 struct stksess *ts, *newts = NULL;
Emeric Brunb3971ab2015-05-12 18:49:09 +0200729 uint32_t msg_len = 0;
730 char *msg_cur = trash.str;
731 char *msg_end = trash.str;
732 unsigned char msg_head[7];
Emeric Brun2b920a12010-09-23 18:30:22 +0200733 int totl = 0;
734
Emeric Brunb3971ab2015-05-12 18:49:09 +0200735 reql = bo_getblk(si_oc(si), (char *)msg_head, 2*sizeof(unsigned char), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200736 if (reql <= 0) /* closed or EOL not found */
737 goto incomplete;
738
Emeric Brun2b920a12010-09-23 18:30:22 +0200739 totl += reql;
740
Emeric Brunb3971ab2015-05-12 18:49:09 +0200741 if (msg_head[1] >= 128) {
742 /* Read and Decode message length */
743 reql = bo_getblk(si_oc(si), (char *)&msg_head[2], sizeof(unsigned char), totl);
744 if (reql <= 0) /* closed */
745 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +0200746
Emeric Brunb3971ab2015-05-12 18:49:09 +0200747 totl += reql;
748
749 if (msg_head[2] < 240) {
750 msg_len = msg_head[2];
Emeric Brun2b920a12010-09-23 18:30:22 +0200751 }
752 else {
Emeric Brunb3971ab2015-05-12 18:49:09 +0200753 int i;
754 char *cur;
755 char *end;
Willy Tarreau72d6c162013-04-11 16:14:13 +0200756
Emeric Brunb3971ab2015-05-12 18:49:09 +0200757 for (i = 3 ; i < sizeof(msg_head) ; i++) {
758 reql = bo_getblk(si_oc(si), (char *)&msg_head[i], sizeof(char), totl);
759 if (reql <= 0) /* closed */
760 goto incomplete;
761
762 totl += reql;
763
764 if (!(msg_head[i] & 0x80))
765 break;
766 }
767
768 if (i == sizeof(msg_head)) {
769 /* malformed message */
770 appctx->st0 = PEER_SESS_ST_ERRPROTO;
771 goto switchstate;
772
773 }
774 end = (char *)msg_head + sizeof(msg_head);
775 cur = (char *)&msg_head[2];
776 msg_len = intdecode(&cur, end);
777 if (!cur) {
778 /* malformed message */
779 appctx->st0 = PEER_SESS_ST_ERRPROTO;
780 goto switchstate;
781 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200782 }
783
Willy Tarreau86a446e2013-11-25 23:02:37 +0100784
Emeric Brunb3971ab2015-05-12 18:49:09 +0200785 /* Read message content */
786 if (msg_len) {
787 if (msg_len > trash.size) {
788 /* Status code is not success, abort */
789 appctx->st0 = PEER_SESS_ST_ERRSIZE;
790 goto switchstate;
791 }
Willy Tarreau72d6c162013-04-11 16:14:13 +0200792
Emeric Brunb3971ab2015-05-12 18:49:09 +0200793 reql = bo_getblk(si_oc(si), trash.str, msg_len, totl);
794 if (reql <= 0) /* closed */
795 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +0200796 totl += reql;
Willy Tarreau86a446e2013-11-25 23:02:37 +0100797
Emeric Brunb3971ab2015-05-12 18:49:09 +0200798 msg_end += msg_len;
799 }
800 }
Willy Tarreau86a446e2013-11-25 23:02:37 +0100801
Emeric Brunb3971ab2015-05-12 18:49:09 +0200802 if (msg_head[0] == PEER_MSG_CLASS_CONTROL) {
803 if (msg_head[1] == PEER_MSG_CTRL_RESYNCREQ) {
804 struct shared_table *st;
805 /* Reset message: remote need resync */
806
807 /* prepare tables fot a global push */
808 for (st = curpeer->tables; st; st = st->next) {
809 st->teaching_origin = st->last_pushed = st->table->update;
810 st->flags = 0;
Willy Tarreau86a446e2013-11-25 23:02:37 +0100811 }
Willy Tarreau72d6c162013-04-11 16:14:13 +0200812
Emeric Brunb3971ab2015-05-12 18:49:09 +0200813 /* reset teaching flags to 0 */
814 curpeer->flags &= PEER_TEACH_RESET;
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200815
Emeric Brunb3971ab2015-05-12 18:49:09 +0200816 /* flag to start to teach lesson */
817 curpeer->flags |= PEER_F_TEACH_PROCESS;
818
819
820 }
821 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCFINISHED) {
822
823 if (curpeer->flags & PEER_F_LEARN_ASSIGN) {
824 curpeer->flags &= ~PEER_F_LEARN_ASSIGN;
825 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
826 peers->flags |= (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE);
Willy Tarreau86a446e2013-11-25 23:02:37 +0100827 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200828 curpeer->confirm++;
Emeric Brun2b920a12010-09-23 18:30:22 +0200829 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200830 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCPARTIAL) {
831
832 if (curpeer->flags & PEER_F_LEARN_ASSIGN) {
833 curpeer->flags &= ~PEER_F_LEARN_ASSIGN;
834 peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
835
836 curpeer->flags |= PEER_F_LEARN_NOTUP2DATE;
837 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
838 task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
Cyril Bonté9a60ff92014-02-16 01:07:07 +0100839 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200840 curpeer->confirm++;
Emeric Brun2b920a12010-09-23 18:30:22 +0200841 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200842 else if (msg_head[1] == PEER_MSG_CTRL_RESYNCCONFIRM) {
843
844 /* If stopping state */
845 if (stopping) {
846 /* Close session, push resync no more needed */
847 curpeer->flags |= PEER_F_TEACH_COMPLETE;
848 appctx->st0 = PEER_SESS_ST_END;
849 goto switchstate;
850 }
851
852 /* reset teaching flags to 0 */
853 curpeer->flags &= PEER_TEACH_RESET;
Emeric Brun2b920a12010-09-23 18:30:22 +0200854 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200855 }
856 else if (msg_head[0] == PEER_MSG_CLASS_STICKTABLE) {
857 if (msg_head[1] == PEER_MSG_STKT_DEFINE) {
858 int table_id_len;
859 struct shared_table *st;
860 int table_type;
861 int table_keylen;
862 int table_id;
863 uint64_t table_data;
Emeric Brun2b920a12010-09-23 18:30:22 +0200864
Emeric Brunb3971ab2015-05-12 18:49:09 +0200865 table_id = intdecode(&msg_cur, msg_end);
866 if (!msg_cur) {
867 /* malformed message */
868 appctx->st0 = PEER_SESS_ST_ERRPROTO;
869 goto switchstate;
870 }
Willy Tarreau72d6c162013-04-11 16:14:13 +0200871
Emeric Brunb3971ab2015-05-12 18:49:09 +0200872 table_id_len = intdecode(&msg_cur, msg_end);
873 if (!msg_cur) {
874 /* malformed message */
875 appctx->st0 = PEER_SESS_ST_ERRPROTO;
876 goto switchstate;
877 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200878
Emeric Brunb3971ab2015-05-12 18:49:09 +0200879 curpeer->remote_table = NULL;
880 if (!table_id_len || (msg_cur + table_id_len) >= msg_end) {
881 /* malformed message */
882 appctx->st0 = PEER_SESS_ST_ERRPROTO;
883 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +0200884 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200885
Emeric Brunb3971ab2015-05-12 18:49:09 +0200886 for (st = curpeer->tables; st; st = st->next) {
887 /* Reset IDs */
888 if (st->remote_id == table_id)
889 st->remote_id = 0;
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200890
Emeric Brunb3971ab2015-05-12 18:49:09 +0200891 if (!curpeer->remote_table
892 && (table_id_len == strlen(st->table->id))
893 && (memcmp(st->table->id, msg_cur, table_id_len) == 0)) {
894 curpeer->remote_table = st;
Emeric Brun2b920a12010-09-23 18:30:22 +0200895 }
896 }
897
Emeric Brunb3971ab2015-05-12 18:49:09 +0200898 if (!curpeer->remote_table) {
899 goto ignore_msg;
900 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200901
Emeric Brunb3971ab2015-05-12 18:49:09 +0200902 msg_cur += table_id_len;
903 if (msg_cur >= msg_end) {
904 /* malformed message */
905 appctx->st0 = PEER_SESS_ST_ERRPROTO;
906 goto switchstate;
907 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200908
Emeric Brunb3971ab2015-05-12 18:49:09 +0200909 table_type = intdecode(&msg_cur, msg_end);
910 if (!msg_cur) {
911 /* malformed message */
912 appctx->st0 = PEER_SESS_ST_ERRPROTO;
913 goto switchstate;
914 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200915
Emeric Brunb3971ab2015-05-12 18:49:09 +0200916 table_keylen = intdecode(&msg_cur, msg_end);
917 if (!msg_cur) {
918 /* malformed message */
919 appctx->st0 = PEER_SESS_ST_ERRPROTO;
920 goto switchstate;
921 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200922
Emeric Brunb3971ab2015-05-12 18:49:09 +0200923 table_data = intdecode(&msg_cur, msg_end);
924 if (!msg_cur) {
925 /* malformed message */
926 appctx->st0 = PEER_SESS_ST_ERRPROTO;
927 goto switchstate;
928 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200929
Emeric Brunb3971ab2015-05-12 18:49:09 +0200930 if (curpeer->remote_table->table->type != table_type
931 || curpeer->remote_table->table->key_size != table_keylen) {
932 curpeer->remote_table = NULL;
933 goto ignore_msg;
934 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200935
Emeric Brunb3971ab2015-05-12 18:49:09 +0200936 curpeer->remote_table->remote_data = table_data;
937 curpeer->remote_table->remote_id = table_id;
Emeric Brun2b920a12010-09-23 18:30:22 +0200938 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200939 else if (msg_head[1] == PEER_MSG_STKT_SWITCH) {
940 struct shared_table *st;
941 int table_id;
Emeric Brun2b920a12010-09-23 18:30:22 +0200942
Emeric Brunb3971ab2015-05-12 18:49:09 +0200943 table_id = intdecode(&msg_cur, msg_end);
944 if (!msg_cur) {
945 /* malformed message */
946 appctx->st0 = PEER_SESS_ST_ERRPROTO;
947 goto switchstate;
948 }
949 curpeer->remote_table = NULL;
950 for (st = curpeer->tables; st; st = st->next) {
951 if (st->remote_id == table_id) {
952 curpeer->remote_table = st;
953 break;
954 }
955 }
956
Emeric Brun2b920a12010-09-23 18:30:22 +0200957 }
Emeric Brunb3971ab2015-05-12 18:49:09 +0200958 else if (msg_head[1] == PEER_MSG_STKT_UPDATE
959 || msg_head[1] == PEER_MSG_STKT_INCUPDATE) {
960 struct shared_table *st = curpeer->remote_table;
961 uint32_t update;
Emeric Brun2b920a12010-09-23 18:30:22 +0200962
Emeric Brunb3971ab2015-05-12 18:49:09 +0200963 /* Here we have data message */
964 if (!st)
965 goto ignore_msg;
Emeric Brun2b920a12010-09-23 18:30:22 +0200966
Emeric Brunb3971ab2015-05-12 18:49:09 +0200967 if (msg_head[1] == PEER_MSG_STKT_UPDATE) {
968 if (msg_len < sizeof(update)) {
969 /* malformed message */
970 appctx->st0 = PEER_SESS_ST_ERRPROTO;
971 goto switchstate;
972 }
973 memcpy(&update, msg_cur, sizeof(update));
974 msg_cur += sizeof(update);
975 st->last_get = htonl(update);
976 }
977 else {
978 st->last_get++;
979 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200980
Emeric Brunb3971ab2015-05-12 18:49:09 +0200981 newts = stksess_new(st->table, NULL);
982 if (!newts)
983 goto ignore_msg;
Emeric Brun2b920a12010-09-23 18:30:22 +0200984
Emeric Brunb3971ab2015-05-12 18:49:09 +0200985 if (st->table->type == STKTABLE_TYPE_STRING) {
986 unsigned int to_read, to_store;
Emeric Brun2b920a12010-09-23 18:30:22 +0200987
Emeric Brunb3971ab2015-05-12 18:49:09 +0200988 to_read = intdecode(&msg_cur, msg_end);
989 if (!msg_cur) {
990 /* malformed message */
991 stksess_free(st->table, newts);
992 appctx->st0 = PEER_SESS_ST_ERRPROTO;
993 goto switchstate;
994 }
995 to_store = MIN(to_read, st->table->key_size - 1);
996 if (msg_cur + to_store > msg_end) {
997 /* malformed message */
998 stksess_free(st->table, newts);
999 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1000 goto switchstate;
1001 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001002
Emeric Brunb3971ab2015-05-12 18:49:09 +02001003 memcpy(newts->key.key, msg_cur, to_store);
1004 newts->key.key[to_store] = 0;
1005 msg_cur += to_read;
1006 }
1007 else if (st->table->type == STKTABLE_TYPE_INTEGER) {
1008 unsigned int netinteger;
Willy Tarreau72d6c162013-04-11 16:14:13 +02001009
Emeric Brunb3971ab2015-05-12 18:49:09 +02001010 if (msg_cur + sizeof(netinteger) > msg_end) {
1011 /* malformed message */
1012 stksess_free(st->table, newts);
1013 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1014 goto switchstate;
1015 }
1016 memcpy(&netinteger, msg_cur, sizeof(netinteger));
1017 netinteger = ntohl(netinteger);
1018 memcpy(newts->key.key, &netinteger, sizeof(netinteger));
1019 msg_cur += sizeof(netinteger);
1020 }
1021 else {
1022 if (msg_cur + st->table->key_size > msg_end) {
1023 /* malformed message */
1024 stksess_free(st->table, newts);
1025 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1026 goto switchstate;
1027 }
1028 memcpy(newts->key.key, msg_cur, st->table->key_size);
1029 msg_cur += st->table->key_size;
1030 }
1031
1032 /* lookup for existing entry */
1033 ts = stktable_lookup(st->table, newts);
1034 if (ts) {
1035 /* the entry already exist, we can free ours */
1036 stktable_touch(st->table, ts, 0);
1037 stksess_free(st->table, newts);
1038 newts = NULL;
1039 }
1040 else {
1041 struct eb32_node *eb;
1042
1043 /* create new entry */
1044 ts = stktable_store(st->table, newts, 0);
1045 newts = NULL; /* don't reuse it */
1046
1047 ts->upd.key= (++st->table->update)+(2^31);
1048 eb = eb32_insert(&st->table->updates, &ts->upd);
1049 if (eb != &ts->upd) {
1050 eb32_delete(eb);
1051 eb32_insert(&st->table->updates, &ts->upd);
1052 }
1053 }
1054
1055 if ((1 << STKTABLE_DT_SERVER_ID) & st->remote_data) {
1056 int srvid;
1057
1058 srvid = intdecode(&msg_cur, msg_end);
1059 if (!msg_cur) {
1060 /* malformed message */
1061 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1062 goto switchstate;
1063 }
1064
1065 if (stktable_data_ptr(st->table, ts, STKTABLE_DT_SERVER_ID)) {
1066 stktable_data_cast(stktable_data_ptr(st->table, ts, STKTABLE_DT_SERVER_ID), server_id) = srvid;
1067 }
1068 }
1069 }
1070 else if (msg_head[1] == PEER_MSG_STKT_ACK) {
1071 /* ack message */
1072 uint32_t table_id ;
1073 uint32_t update;
1074 struct shared_table *st;
1075
1076 table_id = intdecode(&msg_cur, msg_end);
1077 if (!msg_cur || (msg_cur + sizeof(update) > msg_end)) {
1078 /* malformed message */
1079 appctx->st0 = PEER_SESS_ST_ERRPROTO;
1080 goto switchstate;
1081 }
1082 memcpy(&update, msg_cur, sizeof(update));
1083 update = ntohl(update);
Emeric Brun2b920a12010-09-23 18:30:22 +02001084
Emeric Brunb3971ab2015-05-12 18:49:09 +02001085 for (st = curpeer->tables; st; st = st->next) {
1086 if (st->local_id == table_id) {
1087 st->update = update;
1088 break;
1089 }
1090 }
1091 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001092 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001093 else if (msg_head[0] == PEER_MSG_CLASS_RESERVED) {
1094 appctx->st0 = PEER_SESS_ST_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +02001095 goto switchstate;
1096 }
1097
Emeric Brunb3971ab2015-05-12 18:49:09 +02001098ignore_msg:
Emeric Brun2b920a12010-09-23 18:30:22 +02001099 /* skip consumed message */
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001100 bo_skip(si_oc(si), totl);
Emeric Brun2b920a12010-09-23 18:30:22 +02001101 /* loop on that state to peek next message */
Willy Tarreau72d6c162013-04-11 16:14:13 +02001102 goto switchstate;
1103
Emeric Brun2b920a12010-09-23 18:30:22 +02001104incomplete:
Willy Tarreau9d9179b2013-04-11 16:56:44 +02001105 /* we get here when a bo_getblk() returns <= 0 in reql */
1106
Willy Tarreau72d6c162013-04-11 16:14:13 +02001107 if (reql < 0) {
1108 /* there was an error */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001109 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau72d6c162013-04-11 16:14:13 +02001110 goto switchstate;
1111 }
1112
Emeric Brun2b920a12010-09-23 18:30:22 +02001113
1114 /* Confirm finished or partial messages */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001115 while (curpeer->confirm) {
1116 unsigned char msg[2];
1117
Emeric Brun2b920a12010-09-23 18:30:22 +02001118 /* There is a confirm messages to send */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001119 msg[0] = PEER_MSG_CLASS_CONTROL;
1120 msg[1] = PEER_MSG_CTRL_RESYNCCONFIRM;
1121
1122 /* message to buffer */
1123 repl = bi_putblk(si_ic(si), (char *)msg, sizeof(msg));
Emeric Brun2b920a12010-09-23 18:30:22 +02001124 if (repl <= 0) {
1125 /* no more write possible */
1126 if (repl == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +01001127 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001128 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02001129 goto switchstate;
1130 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001131 curpeer->confirm--;
Emeric Brun2b920a12010-09-23 18:30:22 +02001132 }
1133
Emeric Brunb3971ab2015-05-12 18:49:09 +02001134
Emeric Brun2b920a12010-09-23 18:30:22 +02001135 /* Need to request a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001136 if ((curpeer->flags & PEER_F_LEARN_ASSIGN) &&
1137 (peers->flags & PEERS_F_RESYNC_ASSIGN) &&
1138 !(peers->flags & PEERS_F_RESYNC_PROCESS)) {
1139 unsigned char msg[2];
Emeric Brun2b920a12010-09-23 18:30:22 +02001140
Emeric Brunb3971ab2015-05-12 18:49:09 +02001141 /* Current peer was elected to request a resync */
1142 msg[0] = PEER_MSG_CLASS_CONTROL;
1143 msg[1] = PEER_MSG_CTRL_RESYNCREQ;
Emeric Brun2b920a12010-09-23 18:30:22 +02001144
Emeric Brunb3971ab2015-05-12 18:49:09 +02001145 /* message to buffer */
1146 repl = bi_putblk(si_ic(si), (char *)msg, sizeof(msg));
1147 if (repl <= 0) {
1148 /* no more write possible */
1149 if (repl == -1)
1150 goto full;
1151 appctx->st0 = PEER_SESS_ST_END;
1152 goto switchstate;
1153 }
1154 peers->flags |= PEERS_F_RESYNC_PROCESS;
1155 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001156
Emeric Brunb3971ab2015-05-12 18:49:09 +02001157 /* Nothing to read, now we start to write */
Emeric Brun2b920a12010-09-23 18:30:22 +02001158
Emeric Brunb3971ab2015-05-12 18:49:09 +02001159 if (curpeer->tables) {
1160 struct shared_table *st;
1161 struct shared_table *last_local_table;
Emeric Brun2b920a12010-09-23 18:30:22 +02001162
Emeric Brunb3971ab2015-05-12 18:49:09 +02001163 last_local_table = curpeer->last_local_table;
1164 if (!last_local_table)
1165 last_local_table = curpeer->tables;
1166 st = last_local_table->next;
Emeric Brun2b920a12010-09-23 18:30:22 +02001167
Emeric Brunb3971ab2015-05-12 18:49:09 +02001168 while (1) {
1169 if (!st)
1170 st = curpeer->tables;
Emeric Brun2b920a12010-09-23 18:30:22 +02001171
Emeric Brunb3971ab2015-05-12 18:49:09 +02001172 /* It remains some updates to ack */
1173 if (st->last_get != st->last_acked) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001174 int msglen;
Emeric Brun2b920a12010-09-23 18:30:22 +02001175
Emeric Brunb3971ab2015-05-12 18:49:09 +02001176 msglen = peer_prepare_ackmsg(st, trash.str, trash.size);
1177 if (!msglen) {
1178 /* internal error: message does not fit in trash */
1179 appctx->st0 = PEER_SESS_ST_END;
1180 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001181 }
1182
Emeric Brunb3971ab2015-05-12 18:49:09 +02001183 /* message to buffer */
1184 repl = bi_putblk(si_ic(si), trash.str, msglen);
1185 if (repl <= 0) {
1186 /* no more write possible */
1187 if (repl == -1) {
1188 goto full;
Emeric Brun2b920a12010-09-23 18:30:22 +02001189 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001190 appctx->st0 = PEER_SESS_ST_END;
1191 goto switchstate;
Emeric Brun2b920a12010-09-23 18:30:22 +02001192 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001193 st->last_acked = st->last_get;
Emeric Brun2b920a12010-09-23 18:30:22 +02001194 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001195
Emeric Brunb3971ab2015-05-12 18:49:09 +02001196 if (!(curpeer->flags & PEER_F_TEACH_PROCESS)) {
1197 if (!(curpeer->flags & PEER_F_LEARN_ASSIGN) &&
1198 ((int)(st->last_pushed - st->table->localupdate) < 0)) {
1199 struct eb32_node *eb;
1200 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001201
Emeric Brunb3971ab2015-05-12 18:49:09 +02001202 if (st != curpeer->last_local_table) {
1203 int msglen;
Emeric Brun2b920a12010-09-23 18:30:22 +02001204
Emeric Brunb3971ab2015-05-12 18:49:09 +02001205 msglen = peer_prepare_switchmsg(st, trash.str, trash.size);
1206 if (!msglen) {
1207 /* internal error: message does not fit in trash */
1208 appctx->st0 = PEER_SESS_ST_END;
1209 goto switchstate;
1210 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001211
Emeric Brunb3971ab2015-05-12 18:49:09 +02001212 /* message to buffer */
1213 repl = bi_putblk(si_ic(si), trash.str, msglen);
1214 if (repl <= 0) {
1215 /* no more write possible */
1216 if (repl == -1) {
1217 goto full;
1218 }
1219 appctx->st0 = PEER_SESS_ST_END;
1220 goto switchstate;
1221 }
1222 curpeer->last_local_table = st;
Emeric Brun2b920a12010-09-23 18:30:22 +02001223 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001224
1225 /* We force new pushed to 1 to force identifier in update message */
1226 new_pushed = 1;
1227 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
1228 while (1) {
1229 uint32_t msglen;
1230 struct stksess *ts;
1231
1232 /* push local updates */
1233 if (!eb) {
1234 eb = eb32_first(&st->table->updates);
1235 if (!eb || ((int)(eb->key - st->last_pushed) <= 0)) {
1236 st->last_pushed = st->table->localupdate;
1237 break;
1238 }
1239 }
1240
1241 if ((int)(eb->key - st->table->localupdate) > 0) {
1242 st->last_pushed = st->table->localupdate;
1243 break;
1244 }
1245
1246 ts = eb32_entry(eb, struct stksess, upd);
1247 msglen = peer_prepare_updatemsg(ts, st, trash.str, trash.size, new_pushed);
1248 if (!msglen) {
1249 /* internal error: message does not fit in trash */
1250 appctx->st0 = PEER_SESS_ST_END;
1251 goto switchstate;
1252 }
1253
1254 /* message to buffer */
1255 repl = bi_putblk(si_ic(si), trash.str, msglen);
1256 if (repl <= 0) {
1257 /* no more write possible */
1258 if (repl == -1) {
1259 goto full;
1260 }
1261 appctx->st0 = PEER_SESS_ST_END;
1262 goto switchstate;
1263 }
1264 st->last_pushed = ts->upd.key;
1265 /* identifier may not needed in next update message */
1266 new_pushed = 0;
1267
1268 eb = eb32_next(eb);
1269 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001270 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001271 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001272 else {
1273 if (!(st->flags & SHTABLE_F_TEACH_STAGE1)) {
1274 struct eb32_node *eb;
1275 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001276
Emeric Brunb3971ab2015-05-12 18:49:09 +02001277 if (st != curpeer->last_local_table) {
1278 int msglen;
1279
1280 msglen = peer_prepare_switchmsg(st, trash.str, trash.size);
1281 if (!msglen) {
1282 /* internal error: message does not fit in trash */
1283 appctx->st0 = PEER_SESS_ST_END;
1284 goto switchstate;
1285 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001286
Emeric Brunb3971ab2015-05-12 18:49:09 +02001287 /* message to buffer */
1288 repl = bi_putblk(si_ic(si), trash.str, msglen);
1289 if (repl <= 0) {
1290 /* no more write possible */
1291 if (repl == -1) {
1292 goto full;
1293 }
1294 appctx->st0 = PEER_SESS_ST_END;
1295 goto switchstate;
1296 }
1297 curpeer->last_local_table = st;
1298 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001299
Emeric Brunb3971ab2015-05-12 18:49:09 +02001300 /* We force new pushed to 1 to force identifier in update message */
1301 new_pushed = 1;
1302 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
1303 while (1) {
1304 uint32_t msglen;
1305 struct stksess *ts;
Emeric Brun2b920a12010-09-23 18:30:22 +02001306
Emeric Brunb3971ab2015-05-12 18:49:09 +02001307 /* push local updates */
1308 if (!eb) {
1309 st->flags |= SHTABLE_F_TEACH_STAGE1;
1310 eb = eb32_first(&st->table->updates);
1311 if (eb)
1312 st->last_pushed = eb->key - 1;
1313 break;
1314 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001315
Emeric Brunb3971ab2015-05-12 18:49:09 +02001316 ts = eb32_entry(eb, struct stksess, upd);
1317 msglen = peer_prepare_updatemsg(ts, st, trash.str, trash.size, new_pushed);
1318 if (!msglen) {
1319 /* internal error: message does not fit in trash */
1320 appctx->st0 = PEER_SESS_ST_END;
1321 goto switchstate;
1322 }
1323
1324 /* message to buffer */
1325 repl = bi_putblk(si_ic(si), trash.str, msglen);
1326 if (repl <= 0) {
1327 /* no more write possible */
1328 if (repl == -1) {
1329 goto full;
1330 }
1331 appctx->st0 = PEER_SESS_ST_END;
1332 goto switchstate;
1333 }
1334 st->last_pushed = ts->upd.key;
1335 /* identifier may not needed in next update message */
1336 new_pushed = 0;
1337
1338 eb = eb32_next(eb);
1339 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001340 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001341
Emeric Brunb3971ab2015-05-12 18:49:09 +02001342 if (!(st->flags & SHTABLE_F_TEACH_STAGE2)) {
1343 struct eb32_node *eb;
1344 int new_pushed;
Emeric Brun2b920a12010-09-23 18:30:22 +02001345
Emeric Brunb3971ab2015-05-12 18:49:09 +02001346 if (st != curpeer->last_local_table) {
1347 int msglen;
1348
1349 msglen = peer_prepare_switchmsg(st, trash.str, trash.size);
1350 if (!msglen) {
1351 /* internal error: message does not fit in trash */
1352 appctx->st0 = PEER_SESS_ST_END;
1353 goto switchstate;
1354 }
1355
1356 /* message to buffer */
1357 repl = bi_putblk(si_ic(si), trash.str, msglen);
1358 if (repl <= 0) {
1359 /* no more write possible */
1360 if (repl == -1) {
1361 goto full;
1362 }
1363 appctx->st0 = PEER_SESS_ST_END;
1364 goto switchstate;
1365 }
1366 curpeer->last_local_table = st;
1367 }
1368
1369 /* We force new pushed to 1 to force identifier in update message */
1370 new_pushed = 1;
1371 eb = eb32_lookup_ge(&st->table->updates, st->last_pushed+1);
1372 while (1) {
1373 uint32_t msglen;
1374 struct stksess *ts;
1375
1376 /* push local updates */
1377 if (!eb || eb->key > st->teaching_origin) {
1378 st->flags |= SHTABLE_F_TEACH_STAGE2;
1379 eb = eb32_first(&st->table->updates);
1380 if (eb)
1381 st->last_pushed = eb->key - 1;
1382 break;
1383 }
1384
1385 ts = eb32_entry(eb, struct stksess, upd);
1386 msglen = peer_prepare_updatemsg(ts, st, trash.str, trash.size, new_pushed);
1387 if (!msglen) {
1388 /* internal error: message does not fit in trash */
1389 appctx->st0 = PEER_SESS_ST_END;
1390 goto switchstate;
1391 }
1392
1393 /* message to buffer */
1394 repl = bi_putblk(si_ic(si), trash.str, msglen);
1395 if (repl <= 0) {
1396 /* no more write possible */
1397 if (repl == -1) {
1398 goto full;
1399 }
1400 appctx->st0 = PEER_SESS_ST_END;
1401 goto switchstate;
1402 }
1403 st->last_pushed = ts->upd.key;
1404 /* identifier may not needed in next update message */
1405 new_pushed = 0;
1406
1407 eb = eb32_next(eb);
1408 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001409 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001410 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001411
1412 if (st == last_local_table)
1413 break;
1414 st = st->next;
Emeric Brun2b920a12010-09-23 18:30:22 +02001415 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001416 }
1417
1418
1419 if ((curpeer->flags & PEER_F_TEACH_PROCESS) && !(curpeer->flags & PEER_F_TEACH_FINISHED)) {
1420 unsigned char msg[2];
1421
1422 /* Current peer was elected to request a resync */
1423 msg[0] = PEER_MSG_CLASS_CONTROL;
1424 msg[1] = ((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FINISHED) ? PEER_MSG_CTRL_RESYNCFINISHED : PEER_MSG_CTRL_RESYNCPARTIAL;
1425 /* process final lesson message */
1426 repl = bi_putblk(si_ic(si), (char *)msg, sizeof(msg));
1427 if (repl <= 0) {
1428 /* no more write possible */
1429 if (repl == -1)
1430 goto full;
1431 appctx->st0 = PEER_SESS_ST_END;
1432 goto switchstate;
1433 }
1434 /* flag finished message sent */
1435 curpeer->flags |= PEER_F_TEACH_FINISHED;
1436 }
1437
Emeric Brun2b920a12010-09-23 18:30:22 +02001438 /* noting more to do */
1439 goto out;
1440 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001441 case PEER_SESS_ST_EXIT:
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001442 repl = snprintf(trash.str, trash.size, "%d\n", appctx->st1);
Emeric Brun2b920a12010-09-23 18:30:22 +02001443
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001444 if (bi_putblk(si_ic(si), trash.str, repl) == -1)
Willy Tarreaubc18da12015-03-13 14:00:47 +01001445 goto full;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001446 appctx->st0 = PEER_SESS_ST_END;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001447 goto switchstate;
1448 case PEER_SESS_ST_ERRSIZE: {
1449 unsigned char msg[2];
1450
1451 msg[0] = PEER_MSG_CLASS_ERROR;
1452 msg[1] = PEER_MSG_ERR_SIZELIMIT;
1453
1454 if (bi_putblk(si_ic(si), (char *)msg, sizeof(msg)) == -1)
1455 goto full;
1456 appctx->st0 = PEER_SESS_ST_END;
1457 goto switchstate;
1458 }
1459 case PEER_SESS_ST_ERRPROTO: {
1460 unsigned char msg[2];
1461
1462 msg[0] = PEER_MSG_CLASS_ERROR;
1463 msg[1] = PEER_MSG_ERR_PROTOCOL;
1464
1465 if (bi_putblk(si_ic(si), (char *)msg, sizeof(msg)) == -1)
1466 goto full;
1467 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02001468 /* fall through */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001469 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001470 case PEER_SESS_ST_END: {
Willy Tarreau73b013b2012-05-21 16:31:45 +02001471 si_shutw(si);
1472 si_shutr(si);
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001473 si_ic(si)->flags |= CF_READ_NULL;
Willy Tarreau828824a2015-04-19 17:20:03 +02001474 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001475 }
1476 }
1477 }
1478out:
Willy Tarreau2bb4a962014-11-28 11:11:05 +01001479 si_oc(si)->flags |= CF_READ_DONTWAIT;
Emeric Brun2b920a12010-09-23 18:30:22 +02001480 return;
Willy Tarreaubc18da12015-03-13 14:00:47 +01001481full:
Willy Tarreaufe127932015-04-21 19:23:39 +02001482 si_applet_cant_put(si);
Willy Tarreaubc18da12015-03-13 14:00:47 +01001483 goto out;
Emeric Brun2b920a12010-09-23 18:30:22 +02001484}
1485
Willy Tarreau30576452015-04-13 13:50:30 +02001486static struct applet peer_applet = {
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001487 .obj_type = OBJ_TYPE_APPLET,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001488 .name = "<PEER>", /* used for logging */
1489 .fct = peer_io_handler,
Aman Gupta9a13e842012-04-02 18:57:53 -07001490 .release = peer_session_release,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001491};
Emeric Brun2b920a12010-09-23 18:30:22 +02001492
1493/*
1494 * Use this function to force a close of a peer session
1495 */
Willy Tarreau87b09662015-04-03 00:22:06 +02001496static void peer_session_forceshutdown(struct stream * stream)
Emeric Brun2b920a12010-09-23 18:30:22 +02001497{
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001498 struct appctx *appctx = NULL;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001499 struct peer *ps;
1500
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001501 int i;
Emeric Brun2b920a12010-09-23 18:30:22 +02001502
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001503 for (i = 0; i <= 1; i++) {
Willy Tarreau87b09662015-04-03 00:22:06 +02001504 appctx = objt_appctx(stream->si[i].end);
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001505 if (!appctx)
1506 continue;
1507 if (appctx->applet != &peer_applet)
1508 continue;
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001509 break;
Emeric Brun2b920a12010-09-23 18:30:22 +02001510 }
1511
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001512 if (!appctx)
1513 return;
1514
Emeric Brunb3971ab2015-05-12 18:49:09 +02001515 ps = (struct peer *)appctx->ctx.peers.ptr;
Willy Tarreaub4e34da2015-05-20 10:39:04 +02001516 /* we're killing a connection, we must apply a random delay before
1517 * retrying otherwise the other end will do the same and we can loop
1518 * for a while.
1519 */
1520 if (ps)
1521 ps->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + random() % 2000));
1522
Emeric Brun2b920a12010-09-23 18:30:22 +02001523 /* call release to reinit resync states if needed */
Willy Tarreau00a37f02015-04-13 12:05:19 +02001524 peer_session_release(appctx);
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001525 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001526 appctx->ctx.peers.ptr = NULL;
Willy Tarreau87b09662015-04-03 00:22:06 +02001527 task_wakeup(stream->task, TASK_WOKEN_MSG);
Emeric Brun2b920a12010-09-23 18:30:22 +02001528}
1529
Willy Tarreau91d96282015-03-13 15:47:26 +01001530/* Pre-configures a peers frontend to accept incoming connections */
1531void peers_setup_frontend(struct proxy *fe)
1532{
1533 fe->last_change = now.tv_sec;
1534 fe->cap = PR_CAP_FE;
1535 fe->maxconn = 0;
1536 fe->conn_retries = CONN_RETRIES;
1537 fe->timeout.client = MS_TO_TICKS(5000);
Willy Tarreaud1d48d42015-03-13 16:15:46 +01001538 fe->accept = frontend_accept;
Willy Tarreauf87ab942015-03-13 15:55:16 +01001539 fe->default_target = &peer_applet.obj_type;
Willy Tarreau91d96282015-03-13 15:47:26 +01001540 fe->options2 |= PR_O2_INDEPSTR | PR_O2_SMARTCON | PR_O2_SMARTACC;
Willy Tarreau0fca4832015-05-01 19:12:05 +02001541 fe->bind_proc = 0; /* will be filled by users */
Willy Tarreau91d96282015-03-13 15:47:26 +01001542}
1543
Emeric Brun2b920a12010-09-23 18:30:22 +02001544/*
Willy Tarreaubd55e312010-11-11 10:55:09 +01001545 * Create a new peer session in assigned state (connect will start automatically)
Emeric Brun2b920a12010-09-23 18:30:22 +02001546 */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001547static struct stream *peer_session_create(struct peers *peers, struct peer *peer)
Emeric Brun2b920a12010-09-23 18:30:22 +02001548{
Emeric Brunb3971ab2015-05-12 18:49:09 +02001549 struct listener *l = LIST_NEXT(&peers->peers_fe->conf.listeners, struct listener *, by_fe);
Emeric Brun2b920a12010-09-23 18:30:22 +02001550 struct proxy *p = (struct proxy *)l->frontend; /* attached frontend */
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001551 struct appctx *appctx;
Willy Tarreau15b5e142015-04-04 14:38:25 +02001552 struct session *sess;
Willy Tarreau87b09662015-04-03 00:22:06 +02001553 struct stream *s;
Emeric Brun2b920a12010-09-23 18:30:22 +02001554 struct task *t;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001555 struct connection *conn;
Emeric Brun2b920a12010-09-23 18:30:22 +02001556
Emeric Brunb3971ab2015-05-12 18:49:09 +02001557 peer->reconnect = tick_add(now_ms, MS_TO_TICKS(5000));
1558 peer->statuscode = PEER_SESS_SC_CONNECTCODE;
Willy Tarreaud990baf2015-04-05 00:32:03 +02001559 s = NULL;
1560
1561 appctx = appctx_new(&peer_applet);
1562 if (!appctx)
1563 goto out_close;
1564
1565 appctx->st0 = PEER_SESS_ST_CONNECT;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001566 appctx->ctx.peers.ptr = (void *)peer;
Willy Tarreaud990baf2015-04-05 00:32:03 +02001567
Willy Tarreau4099a7c2015-04-05 00:39:55 +02001568 sess = session_new(p, l, &appctx->obj_type);
Willy Tarreau15b5e142015-04-04 14:38:25 +02001569 if (!sess) {
Godbach430f2912013-06-20 13:28:38 +08001570 Alert("out of memory in peer_session_create().\n");
Willy Tarreaud990baf2015-04-05 00:32:03 +02001571 goto out_free_appctx;
Emeric Brun2b920a12010-09-23 18:30:22 +02001572 }
1573
Willy Tarreau8baf9062015-04-05 00:46:36 +02001574 if ((t = task_new()) == NULL) {
Willy Tarreau15b5e142015-04-04 14:38:25 +02001575 Alert("out of memory in peer_session_create().\n");
1576 goto out_free_sess;
1577 }
Willy Tarreau8baf9062015-04-05 00:46:36 +02001578 t->nice = l->nice;
1579
Willy Tarreau73b65ac2015-04-08 18:26:29 +02001580 if ((s = stream_new(sess, t, &appctx->obj_type)) == NULL) {
Willy Tarreau342bfb12015-04-05 01:35:34 +02001581 Alert("Failed to initialize stream in peer_session_create().\n");
Willy Tarreau8baf9062015-04-05 00:46:36 +02001582 goto out_free_task;
1583 }
1584
Willy Tarreau342bfb12015-04-05 01:35:34 +02001585 /* The tasks below are normally what is supposed to be done by
1586 * fe->accept().
1587 */
Willy Tarreaue7dff022015-04-03 01:14:29 +02001588 s->flags = SF_ASSIGNED|SF_ADDR_SET;
Emeric Brun2b920a12010-09-23 18:30:22 +02001589
Willy Tarreau6e2979c2015-04-27 13:21:15 +02001590 /* applet is waiting for data */
1591 si_applet_cant_get(&s->si[0]);
1592 appctx_wakeup(appctx);
1593
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001594 /* initiate an outgoing connection */
1595 si_set_state(&s->si[1], SI_ST_ASS);
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001596
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001597 /* automatically prepare the stream interface to connect to the
Willy Tarreaub363a1f2013-10-01 10:45:07 +02001598 * pre-initialized connection in si->conn.
1599 */
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001600 if (unlikely((conn = conn_new()) == NULL))
Willy Tarreau8baf9062015-04-05 00:46:36 +02001601 goto out_free_strm;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001602
1603 conn_prepare(conn, peer->proto, peer->xprt);
1604 si_attach_conn(&s->si[1], conn);
1605
1606 conn->target = s->target = &s->be->obj_type;
1607 memcpy(&conn->addr.to, &peer->addr, sizeof(conn->addr.to));
Emeric Brun2b920a12010-09-23 18:30:22 +02001608 s->do_log = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001609 s->uniq_id = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +02001610
Willy Tarreau22ec1ea2014-11-27 20:45:39 +01001611 s->res.flags |= CF_READ_DONTWAIT;
Willy Tarreau696a2912014-11-24 11:36:57 +01001612
Emeric Brun2b920a12010-09-23 18:30:22 +02001613 l->nbconn++; /* warning! right now, it's up to the handler to decrease this */
1614 p->feconn++;/* beconn will be increased later */
1615 jobs++;
Willy Tarreaufb0afa72015-04-03 14:46:27 +02001616 if (!(s->sess->listener->options & LI_O_UNLIMITED))
Willy Tarreau3c63fd82011-09-07 18:00:47 +02001617 actconn++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001618 totalconn++;
1619
Emeric Brunb3971ab2015-05-12 18:49:09 +02001620 peer->appctx = appctx;
1621 peer->stream = s;
Emeric Brun2b920a12010-09-23 18:30:22 +02001622 return s;
1623
1624 /* Error unrolling */
Willy Tarreau15b5e142015-04-04 14:38:25 +02001625 out_free_strm:
Emeric Brun2b920a12010-09-23 18:30:22 +02001626 LIST_DEL(&s->list);
Willy Tarreau87b09662015-04-03 00:22:06 +02001627 pool_free2(pool2_stream, s);
Willy Tarreau8baf9062015-04-05 00:46:36 +02001628 out_free_task:
1629 task_free(t);
Willy Tarreau15b5e142015-04-04 14:38:25 +02001630 out_free_sess:
Willy Tarreau11c36242015-04-04 15:54:03 +02001631 session_free(sess);
Willy Tarreaud990baf2015-04-05 00:32:03 +02001632 out_free_appctx:
1633 appctx_free(appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02001634 out_close:
1635 return s;
1636}
1637
1638/*
1639 * Task processing function to manage re-connect and peer session
1640 * tasks wakeup on local update.
1641 */
Simon Horman96553772011-06-08 09:18:51 +09001642static struct task *process_peer_sync(struct task * task)
Emeric Brun2b920a12010-09-23 18:30:22 +02001643{
Emeric Brunb3971ab2015-05-12 18:49:09 +02001644 struct peers *peers = (struct peers *)task->context;
1645 struct peer *ps;
1646 struct shared_table *st;
Emeric Brun2b920a12010-09-23 18:30:22 +02001647
1648 task->expire = TICK_ETERNITY;
1649
Emeric Brunb3971ab2015-05-12 18:49:09 +02001650 if (!peers->peers_fe) {
Willy Tarreau46dc1ca2015-05-01 18:32:13 +02001651 /* this one was never started, kill it */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001652 signal_unregister_handler(peers->sighandler);
1653 peers->sync_task = NULL;
1654 task_delete(peers->sync_task);
1655 task_free(peers->sync_task);
Willy Tarreau46dc1ca2015-05-01 18:32:13 +02001656 return NULL;
1657 }
1658
Emeric Brun2b920a12010-09-23 18:30:22 +02001659 if (!stopping) {
1660 /* Normal case (not soft stop)*/
Emeric Brunb3971ab2015-05-12 18:49:09 +02001661
1662 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMLOCAL) &&
1663 (!nb_oldpids || tick_is_expired(peers->resync_timeout, now_ms)) &&
1664 !(peers->flags & PEERS_F_RESYNC_ASSIGN)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001665 /* Resync from local peer needed
1666 no peer was assigned for the lesson
1667 and no old local peer found
1668 or resync timeout expire */
1669
1670 /* flag no more resync from local, to try resync from remotes */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001671 peers->flags |= PEERS_F_RESYNC_LOCAL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001672
1673 /* reschedule a resync */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001674 peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
Emeric Brun2b920a12010-09-23 18:30:22 +02001675 }
1676
1677 /* For each session */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001678 for (ps = peers->remote; ps; ps = ps->next) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001679 /* For each remote peers */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001680 if (!ps->local) {
Willy Tarreau87b09662015-04-03 00:22:06 +02001681 if (!ps->stream) {
1682 /* no active stream */
Emeric Brun2b920a12010-09-23 18:30:22 +02001683 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001684 ((ps->statuscode == PEER_SESS_SC_CONNECTCODE ||
Willy Tarreaub4e34da2015-05-20 10:39:04 +02001685 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001686 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02001687 tick_is_expired(ps->reconnect, now_ms))) {
1688 /* connection never tried
Willy Tarreau87b09662015-04-03 00:22:06 +02001689 * or previous stream established with success
1690 * or previous stream failed during connection
Emeric Brun2b920a12010-09-23 18:30:22 +02001691 * and reconnection timer is expired */
1692
1693 /* retry a connect */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001694 ps->stream = peer_session_create(peers, ps);
Emeric Brun2b920a12010-09-23 18:30:22 +02001695 }
Willy Tarreaub4e34da2015-05-20 10:39:04 +02001696 else if (!tick_is_expired(ps->reconnect, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001697 /* If previous session failed during connection
1698 * but reconnection timer is not expired */
1699
1700 /* reschedule task for reconnect */
1701 task->expire = tick_first(task->expire, ps->reconnect);
1702 }
1703 /* else do nothing */
Willy Tarreau87b09662015-04-03 00:22:06 +02001704 } /* !ps->stream */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001705 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Willy Tarreau87b09662015-04-03 00:22:06 +02001706 /* current stream is active and established */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001707 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
1708 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02001709 !(ps->flags & PEER_F_LEARN_NOTUP2DATE)) {
1710 /* Resync from a remote is needed
1711 * and no peer was assigned for lesson
1712 * and current peer may be up2date */
1713
1714 /* assign peer for the lesson */
1715 ps->flags |= PEER_F_LEARN_ASSIGN;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001716 peers->flags |= PEERS_F_RESYNC_ASSIGN;
Emeric Brun2b920a12010-09-23 18:30:22 +02001717
Willy Tarreau87b09662015-04-03 00:22:06 +02001718 /* awake peer stream task to handle a request of resync */
Willy Tarreaue5843b32015-04-27 18:40:14 +02001719 appctx_wakeup(ps->appctx);
Emeric Brun2b920a12010-09-23 18:30:22 +02001720 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001721 else {
1722 /* Awake session if there is data to push */
1723 for (st = ps->tables; st ; st = st->next) {
1724 if ((int)(st->last_pushed - st->table->localupdate) < 0) {
1725 /* awake peer stream task to push local updates */
1726 appctx_wakeup(ps->appctx);
1727 break;
1728 }
1729 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001730 }
1731 /* else do nothing */
1732 } /* SUCCESSCODE */
1733 } /* !ps->peer->local */
1734 } /* for */
1735
1736 /* Resync from remotes expired: consider resync is finished */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001737 if (((peers->flags & PEERS_RESYNC_STATEMASK) == PEERS_RESYNC_FROMREMOTE) &&
1738 !(peers->flags & PEERS_F_RESYNC_ASSIGN) &&
1739 tick_is_expired(peers->resync_timeout, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001740 /* Resync from remote peer needed
1741 * no peer was assigned for the lesson
1742 * and resync timeout expire */
1743
1744 /* flag no more resync from remote, consider resync is finished */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001745 peers->flags |= PEERS_F_RESYNC_REMOTE;
Emeric Brun2b920a12010-09-23 18:30:22 +02001746 }
1747
Emeric Brunb3971ab2015-05-12 18:49:09 +02001748 if ((peers->flags & PEERS_RESYNC_STATEMASK) != PEERS_RESYNC_FINISHED) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001749 /* Resync not finished*/
1750 /* reschedule task to resync timeout, to ended resync if needed */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001751 task->expire = tick_first(task->expire, peers->resync_timeout);
Emeric Brun2b920a12010-09-23 18:30:22 +02001752 }
1753 } /* !stopping */
1754 else {
1755 /* soft stop case */
1756 if (task->state & TASK_WOKEN_SIGNAL) {
1757 /* We've just recieved the signal */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001758 if (!(peers->flags & PEERS_F_DONOTSTOP)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001759 /* add DO NOT STOP flag if not present */
1760 jobs++;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001761 peers->flags |= PEERS_F_DONOTSTOP;
1762 ps = peers->local;
1763 for (st = ps->tables; st ; st = st->next)
1764 st->table->syncing++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001765 }
1766
1767 /* disconnect all connected peers */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001768 for (ps = peers->remote; ps; ps = ps->next) {
Willy Tarreau87b09662015-04-03 00:22:06 +02001769 if (ps->stream) {
1770 peer_session_forceshutdown(ps->stream);
1771 ps->stream = NULL;
Willy Tarreaue5843b32015-04-27 18:40:14 +02001772 ps->appctx = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001773 }
1774 }
1775 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001776
Emeric Brunb3971ab2015-05-12 18:49:09 +02001777 ps = peers->local;
Emeric Brun2b920a12010-09-23 18:30:22 +02001778 if (ps->flags & PEER_F_TEACH_COMPLETE) {
Emeric Brunb3971ab2015-05-12 18:49:09 +02001779 if (peers->flags & PEERS_F_DONOTSTOP) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001780 /* resync of new process was complete, current process can die now */
1781 jobs--;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001782 peers->flags &= ~PEERS_F_DONOTSTOP;
1783 for (st = ps->tables; st ; st = st->next)
1784 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02001785 }
1786 }
Willy Tarreau87b09662015-04-03 00:22:06 +02001787 else if (!ps->stream) {
1788 /* If stream is not active */
Emeric Brun2b920a12010-09-23 18:30:22 +02001789 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001790 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
1791 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE ||
1792 ps->statuscode == PEER_SESS_SC_TRYAGAIN) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001793 /* connection never tried
Willy Tarreau87b09662015-04-03 00:22:06 +02001794 * or previous stream was successfully established
1795 * or previous stream tcp connect success but init state incomplete
Emeric Brun2b920a12010-09-23 18:30:22 +02001796 * or during previous connect, peer replies a try again statuscode */
1797
1798 /* connect to the peer */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001799 peer_session_create(peers, ps);
Emeric Brun2b920a12010-09-23 18:30:22 +02001800 }
1801 else {
1802 /* Other error cases */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001803 if (peers->flags & PEERS_F_DONOTSTOP) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001804 /* unable to resync new process, current process can die now */
1805 jobs--;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001806 peers->flags &= ~PEERS_F_DONOTSTOP;
1807 for (st = ps->tables; st ; st = st->next)
1808 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02001809 }
1810 }
1811 }
Emeric Brunb3971ab2015-05-12 18:49:09 +02001812 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE ) {
Willy Tarreau87b09662015-04-03 00:22:06 +02001813 /* current stream active and established
1814 awake stream to push remaining local updates */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001815 for (st = ps->tables; st ; st = st->next) {
1816 if ((int)(st->last_pushed - st->table->localupdate) < 0) {
1817 /* awake peer stream task to push local updates */
1818 appctx_wakeup(ps->appctx);
1819 break;
1820 }
1821 }
Emeric Brun2b920a12010-09-23 18:30:22 +02001822 }
1823 } /* stopping */
1824 /* Wakeup for re-connect */
1825 return task;
1826}
1827
Emeric Brunb3971ab2015-05-12 18:49:09 +02001828
Emeric Brun2b920a12010-09-23 18:30:22 +02001829/*
Emeric Brun2b920a12010-09-23 18:30:22 +02001830 *
1831 */
Emeric Brunb3971ab2015-05-12 18:49:09 +02001832void peers_init_sync(struct peers *peers)
Emeric Brun2b920a12010-09-23 18:30:22 +02001833{
Emeric Brun2b920a12010-09-23 18:30:22 +02001834 struct peer * curpeer;
Willy Tarreau4348fad2012-09-20 16:48:07 +02001835 struct listener *listener;
Emeric Brun2b920a12010-09-23 18:30:22 +02001836
Emeric Brun2b920a12010-09-23 18:30:22 +02001837 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001838 peers->peers_fe->maxconn += 3;
1839 }
1840
Willy Tarreau4348fad2012-09-20 16:48:07 +02001841 list_for_each_entry(listener, &peers->peers_fe->conf.listeners, by_fe)
1842 listener->maxconn = peers->peers_fe->maxconn;
Emeric Brunb3971ab2015-05-12 18:49:09 +02001843 peers->sync_task = task_new();
1844 peers->sync_task->process = process_peer_sync;
1845 peers->sync_task->expire = TICK_ETERNITY;
1846 peers->sync_task->context = (void *)peers;
1847 peers->sighandler = signal_register_task(0, peers->sync_task, 0);
1848 task_wakeup(peers->sync_task, TASK_WOKEN_INIT);
1849}
1850
1851
1852
1853/*
1854 * Function used to register a table for sync on a group of peers
1855 *
1856 */
1857void peers_register_table(struct peers *peers, struct stktable *table)
1858{
1859 struct shared_table *st;
1860 struct peer * curpeer;
1861 int id = 0;
1862
1863 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
1864 st = (struct shared_table *)calloc(1,sizeof(struct shared_table));
1865 st->table = table;
1866 st->next = curpeer->tables;
1867 if (curpeer->tables)
1868 id = curpeer->tables->local_id;
1869 st->local_id = id + 1;
1870
1871 curpeer->tables = st;
1872 }
1873
1874 table->sync_task = peers->sync_task;
Emeric Brun2b920a12010-09-23 18:30:22 +02001875}
1876