blob: 5456410b58dd281fbcf96c21d685cf946e39f1d6 [file] [log] [blame]
Emeric Brun2b920a12010-09-23 18:30:22 +02001/*
Willy Tarreaubd55e312010-11-11 10:55:09 +01002 * Stick table 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>
28#include <types/peers.h>
29
30#include <proto/acl.h>
31#include <proto/buffers.h>
32#include <proto/fd.h>
33#include <proto/log.h>
34#include <proto/hdr_idx.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020035#include <proto/protocols.h>
36#include <proto/proto_tcp.h>
37#include <proto/proto_http.h>
38#include <proto/proxy.h>
39#include <proto/session.h>
Willy Tarreauc63190d2012-05-11 14:23:52 +020040#include <proto/sock_raw.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020041#include <proto/stream_interface.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020042#include <proto/task.h>
43#include <proto/stick_table.h>
44#include <proto/signal.h>
45
46
47/*******************************/
48/* Current peer learning state */
49/*******************************/
50
51/******************************/
52/* Current table resync state */
53/******************************/
54#define SHTABLE_F_RESYNC_LOCAL 0x00000001 /* Learn from local finished or no more needed */
55#define SHTABLE_F_RESYNC_REMOTE 0x00000002 /* Learn from remote finished or no more needed */
56#define SHTABLE_F_RESYNC_ASSIGN 0x00000004 /* A peer was assigned to learn our lesson */
57#define SHTABLE_F_RESYNC_PROCESS 0x00000008 /* The assigned peer was requested for resync */
58#define SHTABLE_F_DONOTSTOP 0x00010000 /* Main table sync task block process during soft stop
59 to push data to new process */
60
61#define SHTABLE_RESYNC_STATEMASK (SHTABLE_F_RESYNC_LOCAL|SHTABLE_F_RESYNC_REMOTE)
62#define SHTABLE_RESYNC_FROMLOCAL 0x00000000
63#define SHTABLE_RESYNC_FROMREMOTE SHTABLE_F_RESYNC_LOCAL
64#define SHTABLE_RESYNC_FINISHED (SHTABLE_F_RESYNC_LOCAL|SHTABLE_F_RESYNC_REMOTE)
65
66/******************************/
67/* Remote peer teaching state */
68/******************************/
69#define PEER_F_TEACH_PROCESS 0x00000001 /* Teach a lesson to current peer */
70#define PEER_F_TEACH_STAGE1 0x00000002 /* Teach state 1 complete */
71#define PEER_F_TEACH_STAGE2 0x00000004 /* Teach stage 2 complete */
72#define PEER_F_TEACH_FINISHED 0x00000008 /* Teach conclude, (wait for confirm) */
73#define PEER_F_TEACH_COMPLETE 0x00000010 /* All that we know already taught to current peer, used only for a local peer */
74#define PEER_F_LEARN_ASSIGN 0x00000100 /* Current peer was assigned for a lesson */
75#define PEER_F_LEARN_NOTUP2DATE 0x00000200 /* Learn from peer finished but peer is not up to date */
76
77#define PEER_TEACH_RESET ~(PEER_F_TEACH_PROCESS|PEER_F_TEACH_STAGE1|PEER_F_TEACH_STAGE2|PEER_F_TEACH_FINISHED) /* PEER_F_TEACH_COMPLETE should never be reset */
78#define PEER_LEARN_RESET ~(PEER_F_LEARN_ASSIGN|PEER_F_LEARN_NOTUP2DATE)
79
80
81/**********************************/
82/* Peer Session IO handler states */
83/**********************************/
84
85#define PEER_SESSION_ACCEPT 1000 /* Initial state for session create by an accept */
86#define PEER_SESSION_GETVERSION 1001 /* Validate supported protocol version*/
87#define PEER_SESSION_GETHOST 1002 /* Validate host ID correspond to local host id */
88#define PEER_SESSION_GETPEER 1003 /* Validate peer ID correspond to a known remote peer id */
89#define PEER_SESSION_GETTABLE 1004 /* Search into registered table for a table with same id and
90 validate type and size */
91#define PEER_SESSION_SENDSUCCESS 1005 /* Send ret code 200 (success) and wait for message */
92/* next state is WAITMSG */
93
94#define PEER_SESSION_CONNECT 2000 /* Initial state for session create on a connect,
95 push presentation into buffer */
96#define PEER_SESSION_GETSTATUS 2001 /* Wait for the welcome message */
97#define PEER_SESSION_WAITMSG 2002 /* Wait for datamessages*/
98/* loop on WAITMSG */
99
100#define PEER_SESSION_EXIT 10000 /* Exit with status code */
101#define PEER_SESSION_END 10001 /* Killed session */
102/* session ended */
103
104
105/**********************************/
106/* Peer Session status code */
107/**********************************/
108
109#define PEER_SESSION_CONNECTCODE 100 /* connect in progress */
110#define PEER_SESSION_CONNECTEDCODE 110 /* tcp connect success */
111
112#define PEER_SESSION_SUCCESSCODE 200 /* accept or connect successful */
113
114#define PEER_SESSION_TRYAGAIN 300 /* try again later */
115
116#define PEER_SESSION_ERRPROTO 501 /* error protocol */
117#define PEER_SESSION_ERRVERSION 502 /* unknown protocol version */
118#define PEER_SESSION_ERRHOST 503 /* bad host name */
119#define PEER_SESSION_ERRPEER 504 /* unknown peer */
120#define PEER_SESSION_ERRTYPE 505 /* table key type mismatch */
121#define PEER_SESSION_ERRSIZE 506 /* table key size mismatch */
122#define PEER_SESSION_ERRTABLE 507 /* unknown table */
123
124#define PEER_SESSION_PROTO_NAME "HAProxyS"
125
126struct peers *peers = NULL;
Simon Horman96553772011-06-08 09:18:51 +0900127static void peer_session_forceshutdown(struct session * session);
Emeric Brun2b920a12010-09-23 18:30:22 +0200128
129
130/*
131 * This prepare the data update message of the stick session <ts>, <ps> is the the peer session
132 * where the data going to be pushed, <msg> is a buffer of <size> to recieve data message content
133 */
Simon Horman96553772011-06-08 09:18:51 +0900134static int peer_prepare_datamsg(struct stksess *ts, struct peer_session *ps, char *msg, size_t size)
Emeric Brun2b920a12010-09-23 18:30:22 +0200135{
136 uint32_t netinteger;
137 int len;
138 /* construct message */
139 if (ps->lastpush && ts->upd.key > ps->lastpush && (ts->upd.key - ps->lastpush) <= 127) {
140 msg[0] = 0x80 + ts->upd.key - ps->lastpush;
141 len = sizeof(char);
142 }
143 else {
144 msg[0] = 'D';
145 netinteger = htonl(ts->upd.key);
146 memcpy(&msg[sizeof(char)], &netinteger, sizeof(netinteger));
147 len = sizeof(char) + sizeof(netinteger);
148 }
149
150 if (ps->table->table->type == STKTABLE_TYPE_STRING) {
151 int stlen = strlen((char *)ts->key.key);
152
153 netinteger = htonl(strlen((char *)ts->key.key));
154 memcpy(&msg[len], &netinteger, sizeof(netinteger));
155 memcpy(&msg[len+sizeof(netinteger)], ts->key.key, stlen);
156 len += sizeof(netinteger) + stlen;
157
158 }
159 else if (ps->table->table->type == STKTABLE_TYPE_INTEGER) {
160 netinteger = htonl(*((uint32_t *)ts->key.key));
161 memcpy(&msg[len], &netinteger, sizeof(netinteger));
162 len += sizeof(netinteger);
163 }
164 else {
165 memcpy(&msg[len], ts->key.key, ps->table->table->key_size);
166 len += ps->table->table->key_size;
167 }
168
169 if (stktable_data_ptr(ps->table->table, ts, STKTABLE_DT_SERVER_ID))
170 netinteger = htonl(stktable_data_cast(stktable_data_ptr(ps->table->table, ts, STKTABLE_DT_SERVER_ID), server_id));
171 else
172 netinteger = 0;
173
174 memcpy(&msg[len], &netinteger , sizeof(netinteger));
175 len += sizeof(netinteger);
176
177 return len;
178}
179
180
181/*
182 * Callback to release a session with a peer
183 */
Simon Horman96553772011-06-08 09:18:51 +0900184static void peer_session_release(struct stream_interface *si)
Emeric Brun2b920a12010-09-23 18:30:22 +0200185{
Aman Guptad94991d2012-04-06 17:39:26 -0700186 struct task *t = (struct task *)si->owner;
Emeric Brun2b920a12010-09-23 18:30:22 +0200187 struct session *s = (struct session *)t->context;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100188 struct peer_session *ps = (struct peer_session *)si->applet.private;
Emeric Brun2b920a12010-09-23 18:30:22 +0200189
Willy Tarreaubc4af052011-02-13 13:25:14 +0100190 /* si->applet.private is not a peer session */
191 if (si->applet.st0 < PEER_SESSION_SENDSUCCESS)
Emeric Brun2b920a12010-09-23 18:30:22 +0200192 return;
193
194 /* peer session identified */
195 if (ps) {
196 if (ps->session == s) {
197 ps->session = NULL;
198 if (ps->flags & PEER_F_LEARN_ASSIGN) {
199 /* unassign current peer for learning */
200 ps->flags &= ~(PEER_F_LEARN_ASSIGN);
201 ps->table->flags &= ~(SHTABLE_F_RESYNC_ASSIGN|SHTABLE_F_RESYNC_PROCESS);
202
203 /* reschedule a resync */
204 ps->table->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
205 }
206 /* reset teaching and learning flags to 0 */
207 ps->flags &= PEER_TEACH_RESET;
208 ps->flags &= PEER_LEARN_RESET;
209 }
210 task_wakeup(ps->table->sync_task, TASK_WOKEN_MSG);
211 }
212}
213
214
215/*
216 * IO Handler to handle message exchance with a peer
217 */
Willy Tarreaub24281b2011-02-13 13:16:36 +0100218static void peer_io_handler(struct stream_interface *si)
Emeric Brun2b920a12010-09-23 18:30:22 +0200219{
220 struct task *t= (struct task *)si->owner;
221 struct session *s = (struct session *)t->context;
222 struct peers *curpeers = (struct peers *)s->fe->parent;
223 int reql = 0;
224 int repl = 0;
225
226 while (1) {
227switchstate:
Willy Tarreaubc4af052011-02-13 13:25:14 +0100228 switch(si->applet.st0) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200229 case PEER_SESSION_ACCEPT:
Willy Tarreaubc4af052011-02-13 13:25:14 +0100230 si->applet.private = NULL;
231 si->applet.st0 = PEER_SESSION_GETVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200232 /* fall through */
233 case PEER_SESSION_GETVERSION:
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200234 reql = bo_getline(si->ob, trash, sizeof(trash));
Emeric Brun2b920a12010-09-23 18:30:22 +0200235 if (reql <= 0) { /* closed or EOL not found */
236 if (reql == 0)
237 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100238 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200239 goto switchstate;
240 }
241 if (trash[reql-1] != '\n') {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100242 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200243 goto switchstate;
244 }
245 else if (reql > 1 && (trash[reql-2] == '\r'))
246 trash[reql-2] = 0;
247 else
248 trash[reql-1] = 0;
249
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200250 bo_skip(si->ob, reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200251
252 /* test version */
253 if (strcmp(PEER_SESSION_PROTO_NAME " 1.0", trash) != 0) {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100254 si->applet.st0 = PEER_SESSION_EXIT;
255 si->applet.st1 = PEER_SESSION_ERRVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200256 /* test protocol */
257 if (strncmp(PEER_SESSION_PROTO_NAME " ", trash, strlen(PEER_SESSION_PROTO_NAME)+1) != 0)
Willy Tarreaubc4af052011-02-13 13:25:14 +0100258 si->applet.st1 = PEER_SESSION_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +0200259 goto switchstate;
260 }
261
Willy Tarreaubc4af052011-02-13 13:25:14 +0100262 si->applet.st0 = PEER_SESSION_GETHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200263 /* fall through */
264 case PEER_SESSION_GETHOST:
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200265 reql = bo_getline(si->ob, trash, sizeof(trash));
Emeric Brun2b920a12010-09-23 18:30:22 +0200266 if (reql <= 0) { /* closed or EOL not found */
267 if (reql == 0)
268 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100269 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200270 goto switchstate;
271 }
272 if (trash[reql-1] != '\n') {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100273 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200274 goto switchstate;
275 }
276 else if (reql > 1 && (trash[reql-2] == '\r'))
277 trash[reql-2] = 0;
278 else
279 trash[reql-1] = 0;
280
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200281 bo_skip(si->ob, reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200282
283 /* test hostname match */
284 if (strcmp(localpeer, trash) != 0) {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100285 si->applet.st0 = PEER_SESSION_EXIT;
286 si->applet.st1 = PEER_SESSION_ERRHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200287 goto switchstate;
288 }
289
Willy Tarreaubc4af052011-02-13 13:25:14 +0100290 si->applet.st0 = PEER_SESSION_GETPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200291 /* fall through */
292 case PEER_SESSION_GETPEER: {
293 struct peer *curpeer;
294 char *p;
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200295 reql = bo_getline(si->ob, trash, sizeof(trash));
Emeric Brun2b920a12010-09-23 18:30:22 +0200296 if (reql <= 0) { /* closed or EOL not found */
297 if (reql == 0)
298 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100299 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200300 goto switchstate;
301 }
302 if (trash[reql-1] != '\n') {
303 /* Incomplete line, we quit */
Willy Tarreaubc4af052011-02-13 13:25:14 +0100304 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200305 goto switchstate;
306 }
307 else if (reql > 1 && (trash[reql-2] == '\r'))
308 trash[reql-2] = 0;
309 else
310 trash[reql-1] = 0;
311
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200312 bo_skip(si->ob, reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200313
314 /* parse line "<peer name> <pid>" */
315 p = strchr(trash, ' ');
316 if (!p) {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100317 si->applet.st0 = PEER_SESSION_EXIT;
318 si->applet.st1 = PEER_SESSION_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +0200319 goto switchstate;
320 }
321 *p = 0;
322
323 /* lookup known peer */
324 for (curpeer = curpeers->remote; curpeer; curpeer = curpeer->next) {
325 if (strcmp(curpeer->id, trash) == 0)
326 break;
327 }
328
329 /* if unknown peer */
330 if (!curpeer) {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100331 si->applet.st0 = PEER_SESSION_EXIT;
332 si->applet.st1 = PEER_SESSION_ERRPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200333 goto switchstate;
334 }
335
Willy Tarreaubc4af052011-02-13 13:25:14 +0100336 si->applet.private = curpeer;
337 si->applet.st0 = PEER_SESSION_GETTABLE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200338 /* fall through */
339 }
340 case PEER_SESSION_GETTABLE: {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100341 struct peer *curpeer = (struct peer *)si->applet.private;
Emeric Brun2b920a12010-09-23 18:30:22 +0200342 struct shared_table *st;
343 struct peer_session *ps = NULL;
344 unsigned long key_type;
345 size_t key_size;
346 char *p;
347
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200348 reql = bo_getline(si->ob, trash, sizeof(trash));
Emeric Brun2b920a12010-09-23 18:30:22 +0200349 if (reql <= 0) { /* closed or EOL not found */
350 if (reql == 0)
351 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100352 si->applet.private = NULL;
353 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200354 goto switchstate;
355 }
Willy Tarreaubc4af052011-02-13 13:25:14 +0100356 /* Re init si->applet.private to null, to handle correctly a release case */
357 si->applet.private = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +0200358
359 if (trash[reql-1] != '\n') {
360 /* Incomplete line, we quit */
Willy Tarreaubc4af052011-02-13 13:25:14 +0100361 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200362 goto switchstate;
363 }
364 else if (reql > 1 && (trash[reql-2] == '\r'))
365 trash[reql-2] = 0;
366 else
367 trash[reql-1] = 0;
368
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200369 bo_skip(si->ob, reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200370
371 /* Parse line "<table name> <type> <size>" */
372 p = strchr(trash, ' ');
373 if (!p) {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100374 si->applet.st0 = PEER_SESSION_EXIT;
375 si->applet.st1 = PEER_SESSION_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +0200376 goto switchstate;
377 }
378 *p = 0;
379 key_type = (unsigned long)atol(p+1);
380
381 p = strchr(p+1, ' ');
382 if (!p) {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100383 si->applet.private = NULL;
384 si->applet.st0 = PEER_SESSION_EXIT;
385 si->applet.st1 = PEER_SESSION_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +0200386 goto switchstate;
387 }
388
389 key_size = (size_t)atoi(p);
390 for (st = curpeers->tables; st; st = st->next) {
391 /* If table name matches */
392 if (strcmp(st->table->id, trash) == 0) {
393 /* If key size mismatches */
394 if (key_size != st->table->key_size) {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100395 si->applet.st0 = PEER_SESSION_EXIT;
396 si->applet.st1 = PEER_SESSION_ERRSIZE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200397 goto switchstate;
398 }
399
400 /* If key type mismatches */
401 if (key_type != st->table->type) {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100402 si->applet.st0 = PEER_SESSION_EXIT;
403 si->applet.st1 = PEER_SESSION_ERRTYPE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200404 goto switchstate;
405 }
406
407 /* lookup peer session of current peer */
408 for (ps = st->sessions; ps; ps = ps->next) {
409 if (ps->peer == curpeer) {
410 /* If session already active, replaced by new one */
411 if (ps->session && ps->session != s) {
412 if (ps->peer->local) {
413 /* Local connection, reply a retry */
Willy Tarreaubc4af052011-02-13 13:25:14 +0100414 si->applet.st0 = PEER_SESSION_EXIT;
415 si->applet.st1 = PEER_SESSION_TRYAGAIN;
Emeric Brun2b920a12010-09-23 18:30:22 +0200416 goto switchstate;
417 }
418 peer_session_forceshutdown(ps->session);
419 }
420 ps->session = s;
421 break;
422 }
423 }
424 break;
425 }
426 }
427
428 /* If table not found */
429 if (!st){
Willy Tarreaubc4af052011-02-13 13:25:14 +0100430 si->applet.st0 = PEER_SESSION_EXIT;
431 si->applet.st1 = PEER_SESSION_ERRTABLE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200432 goto switchstate;
433 }
434
435 /* If no peer session for current peer */
436 if (!ps) {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100437 si->applet.st0 = PEER_SESSION_EXIT;
438 si->applet.st1 = PEER_SESSION_ERRPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200439 goto switchstate;
440 }
441
Willy Tarreaubc4af052011-02-13 13:25:14 +0100442 si->applet.private = ps;
443 si->applet.st0 = PEER_SESSION_SENDSUCCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200444 /* fall through */
445 }
446 case PEER_SESSION_SENDSUCCESS:{
Willy Tarreaubc4af052011-02-13 13:25:14 +0100447 struct peer_session *ps = (struct peer_session *)si->applet.private;
Emeric Brun2b920a12010-09-23 18:30:22 +0200448
449 repl = snprintf(trash, sizeof(trash), "%d\n", PEER_SESSION_SUCCESSCODE);
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200450 repl = bi_putblk(si->ib, trash, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200451 if (repl <= 0) {
452 if (repl == -1)
453 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100454 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200455 goto switchstate;
456 }
457
458 /* Register status code */
459 ps->statuscode = PEER_SESSION_SUCCESSCODE;
460
461 /* Awake main task */
462 task_wakeup(ps->table->sync_task, TASK_WOKEN_MSG);
463
464 /* Init cursors */
465 ps->teaching_origin =ps->lastpush = ps->lastack = ps->pushack = 0;
466 ps->pushed = ps->update;
467
468 /* Init confirm counter */
469 ps->confirm = 0;
470
471 /* reset teaching and learning flags to 0 */
472 ps->flags &= PEER_TEACH_RESET;
473 ps->flags &= PEER_LEARN_RESET;
474
475 /* if current peer is local */
476 if (ps->peer->local) {
477 /* if table need resyncfrom local and no process assined */
478 if ((ps->table->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMLOCAL &&
479 !(ps->table->flags & SHTABLE_F_RESYNC_ASSIGN)) {
480 /* assign local peer for a lesson, consider lesson already requested */
481 ps->flags |= PEER_F_LEARN_ASSIGN;
482 ps->table->flags |= (SHTABLE_F_RESYNC_ASSIGN|SHTABLE_F_RESYNC_PROCESS);
483 }
484
485 }
486 else if ((ps->table->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMREMOTE &&
487 !(ps->table->flags & SHTABLE_F_RESYNC_ASSIGN)) {
488 /* assign peer for a lesson */
489 ps->flags |= PEER_F_LEARN_ASSIGN;
490 ps->table->flags |= SHTABLE_F_RESYNC_ASSIGN;
491 }
492 /* switch to waiting message state */
Willy Tarreaubc4af052011-02-13 13:25:14 +0100493 si->applet.st0 = PEER_SESSION_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200494 goto switchstate;
495 }
496 case PEER_SESSION_CONNECT: {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100497 struct peer_session *ps = (struct peer_session *)si->applet.private;
Emeric Brun2b920a12010-09-23 18:30:22 +0200498
499 /* Send headers */
500 repl = snprintf(trash, sizeof(trash),
501 PEER_SESSION_PROTO_NAME " 1.0\n%s\n%s %d\n%s %lu %d\n",
502 ps->peer->id,
503 localpeer,
Willy Tarreau7b77c9f2012-01-07 22:52:12 +0100504 (int)getpid(),
Emeric Brun2b920a12010-09-23 18:30:22 +0200505 ps->table->table->id,
506 ps->table->table->type,
Willy Tarreaubd55e312010-11-11 10:55:09 +0100507 (int)ps->table->table->key_size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200508
509 if (repl >= sizeof(trash)) {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100510 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200511 goto switchstate;
512 }
513
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200514 repl = bi_putblk(si->ib, trash, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200515 if (repl <= 0) {
516 if (repl == -1)
517 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100518 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200519 goto switchstate;
520 }
521
522 /* switch to the waiting statuscode state */
Willy Tarreaubc4af052011-02-13 13:25:14 +0100523 si->applet.st0 = PEER_SESSION_GETSTATUS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200524 /* fall through */
525 }
526 case PEER_SESSION_GETSTATUS: {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100527 struct peer_session *ps = (struct peer_session *)si->applet.private;
Emeric Brun2b920a12010-09-23 18:30:22 +0200528
529 if (si->ib->flags & BF_WRITE_PARTIAL)
530 ps->statuscode = PEER_SESSION_CONNECTEDCODE;
531
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200532 reql = bo_getline(si->ob, trash, sizeof(trash));
Emeric Brun2b920a12010-09-23 18:30:22 +0200533 if (reql <= 0) { /* closed or EOL not found */
534 if (reql == 0)
535 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100536 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200537 goto switchstate;
538 }
539 if (trash[reql-1] != '\n') {
540 /* Incomplete line, we quit */
Willy Tarreaubc4af052011-02-13 13:25:14 +0100541 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200542 goto switchstate;
543 }
544 else if (reql > 1 && (trash[reql-2] == '\r'))
545 trash[reql-2] = 0;
546 else
547 trash[reql-1] = 0;
548
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200549 bo_skip(si->ob, reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200550
551 /* Register status code */
552 ps->statuscode = atoi(trash);
553
554 /* Awake main task */
555 task_wakeup(ps->table->sync_task, TASK_WOKEN_MSG);
556
557 /* If status code is success */
558 if (ps->statuscode == PEER_SESSION_SUCCESSCODE) {
559 /* Init cursors */
560 ps->teaching_origin = ps->lastpush = ps->lastack = ps->pushack = 0;
561 ps->pushed = ps->update;
562
563 /* Init confirm counter */
564 ps->confirm = 0;
565
566 /* reset teaching and learning flags to 0 */
567 ps->flags &= PEER_TEACH_RESET;
568 ps->flags &= PEER_LEARN_RESET;
569
570 /* If current peer is local */
571 if (ps->peer->local) {
572 /* Init cursors to push a resync */
573 ps->teaching_origin = ps->pushed = ps->table->table->update;
574 /* flag to start to teach lesson */
575 ps->flags |= PEER_F_TEACH_PROCESS;
576
577 }
578 else if ((ps->table->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMREMOTE &&
579 !(ps->table->flags & SHTABLE_F_RESYNC_ASSIGN)) {
580 /* If peer is remote and resync from remote is needed,
581 and no peer currently assigned */
582
583 /* assign peer for a lesson */
584 ps->flags |= PEER_F_LEARN_ASSIGN;
585 ps->table->flags |= SHTABLE_F_RESYNC_ASSIGN;
586 }
587
588 }
589 else {
590 /* Status code is not success, abort */
Willy Tarreaubc4af052011-02-13 13:25:14 +0100591 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200592 goto switchstate;
593 }
Willy Tarreaubc4af052011-02-13 13:25:14 +0100594 si->applet.st0 = PEER_SESSION_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200595 /* fall through */
596 }
597 case PEER_SESSION_WAITMSG: {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100598 struct peer_session *ps = (struct peer_session *)si->applet.private;
Emeric Brun2b920a12010-09-23 18:30:22 +0200599 char c;
600 int totl = 0;
601
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200602 reql = bo_getblk(si->ob, (char *)&c, sizeof(c), totl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200603 if (reql <= 0) { /* closed or EOL not found */
604 if (reql == 0) {
605 /* nothing to read */
606 goto incomplete;
607 }
Willy Tarreaubc4af052011-02-13 13:25:14 +0100608 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200609 goto switchstate;
610 }
611 totl += reql;
612
613 if ((c & 0x80) || (c == 'D')) {
614 /* Here we have data message */
615 unsigned int pushack;
616 struct stksess *ts;
617 struct stksess *newts;
618 struct stktable_key stkey;
619 int srvid;
620 uint32_t netinteger;
621
622 /* Compute update remote version */
623 if (c & 0x80) {
624 pushack = ps->pushack + (unsigned int)(c & 0x7F);
625 }
626 else {
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200627 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200628 if (reql <= 0) { /* closed or EOL not found */
629 if (reql == 0) {
630 goto incomplete;
631 }
Willy Tarreaubc4af052011-02-13 13:25:14 +0100632 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200633 goto switchstate;
634 }
635 totl += reql;
636 pushack = ntohl(netinteger);
637 }
638
639 /* read key */
640 if (ps->table->table->type == STKTABLE_TYPE_STRING) {
641 /* type string */
642 stkey.key = stkey.data.buf;
643
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200644 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200645 if (reql <= 0) { /* closed or EOL not found */
646 if (reql == 0) {
647 goto incomplete;
648 }
Willy Tarreaubc4af052011-02-13 13:25:14 +0100649 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200650 goto switchstate;
651 }
652 totl += reql;
653 stkey.key_len = ntohl(netinteger);
654
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200655 reql = bo_getblk(si->ob, stkey.key, stkey.key_len, totl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200656 if (reql <= 0) { /* closed or EOL not found */
657 if (reql == 0) {
658 goto incomplete;
659 }
Willy Tarreaubc4af052011-02-13 13:25:14 +0100660 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200661 goto switchstate;
662 }
663 totl += reql;
664 }
665 else if (ps->table->table->type == STKTABLE_TYPE_INTEGER) {
666 /* type integer */
667 stkey.key_len = (size_t)-1;
668 stkey.key = &stkey.data.integer;
669
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200670 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200671 if (reql <= 0) { /* closed or EOL not found */
672 if (reql == 0) {
673 goto incomplete;
674 }
Willy Tarreaubc4af052011-02-13 13:25:14 +0100675 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200676 goto switchstate;
677 }
678 totl += reql;
679 stkey.data.integer = ntohl(netinteger);
680 }
681 else {
682 /* type ip */
683 stkey.key_len = (size_t)-1;
684 stkey.key = stkey.data.buf;
685
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200686 reql = bo_getblk(si->ob, (char *)&stkey.data.buf, ps->table->table->key_size, totl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200687 if (reql <= 0) { /* closed or EOL not found */
688 if (reql == 0) {
689 goto incomplete;
690 }
Willy Tarreaubc4af052011-02-13 13:25:14 +0100691 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200692 goto switchstate;
693 }
694 totl += reql;
695
696 }
697
698 /* read server id */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200699 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200700 if (reql <= 0) { /* closed or EOL not found */
701 if (reql == 0) {
702 goto incomplete;
703 }
Willy Tarreaubc4af052011-02-13 13:25:14 +0100704 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200705 goto switchstate;
706 }
707 totl += reql;
708 srvid = ntohl(netinteger);
709
710 /* update entry */
711 newts = stksess_new(ps->table->table, &stkey);
712 if (newts) {
713 /* lookup for existing entry */
714 ts = stktable_lookup(ps->table->table, newts);
715 if (ts) {
716 /* the entry already exist, we can free ours */
717 stktable_touch(ps->table->table, ts, 0);
718 stksess_free(ps->table->table, newts);
719 }
720 else {
721 struct eb32_node *eb;
722
723 /* create new entry */
724 ts = stktable_store(ps->table->table, newts, 0);
725 ts->upd.key= (++ps->table->table->update)+(2^31);
726 eb = eb32_insert(&ps->table->table->updates, &ts->upd);
727 if (eb != &ts->upd) {
728 eb32_delete(eb);
729 eb32_insert(&ps->table->table->updates, &ts->upd);
730 }
731 }
732
733 /* update entry */
734 if (srvid && stktable_data_ptr(ps->table->table, ts, STKTABLE_DT_SERVER_ID))
735 stktable_data_cast(stktable_data_ptr(ps->table->table, ts, STKTABLE_DT_SERVER_ID), server_id) = srvid;
736 ps->pushack = pushack;
737 }
738
739 }
740 else if (c == 'R') {
741 /* Reset message: remote need resync */
742
743 /* reinit counters for a resync */
744 ps->lastpush = 0;
745 ps->teaching_origin = ps->pushed = ps->table->table->update;
746
747 /* reset teaching flags to 0 */
748 ps->flags &= PEER_TEACH_RESET;
749
750 /* flag to start to teach lesson */
751 ps->flags |= PEER_F_TEACH_PROCESS;
752 }
753 else if (c == 'F') {
754 /* Finish message, all known updates have been pushed by remote */
755 /* and remote is up to date */
756
757 /* If resync is in progress with remote peer */
758 if (ps->flags & PEER_F_LEARN_ASSIGN) {
759
760 /* unassign current peer for learning */
761 ps->flags &= ~PEER_F_LEARN_ASSIGN;
762 ps->table->flags &= ~(SHTABLE_F_RESYNC_ASSIGN|SHTABLE_F_RESYNC_PROCESS);
763
764 /* Consider table is now up2date, resync resync no more needed from local neither remote */
765 ps->table->flags |= (SHTABLE_F_RESYNC_LOCAL|SHTABLE_F_RESYNC_REMOTE);
766 }
767 /* Increase confirm counter to launch a confirm message */
768 ps->confirm++;
769 }
770 else if (c == 'c') {
771 /* confirm message, remote peer is now up to date with us */
772
773 /* If stopping state */
774 if (stopping) {
775 /* Close session, push resync no more needed */
776 ps->flags |= PEER_F_TEACH_COMPLETE;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100777 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200778 goto switchstate;
779 }
780
781 /* reset teaching flags to 0 */
782 ps->flags &= PEER_TEACH_RESET;
783 }
784 else if (c == 'C') {
785 /* Continue message, all known updates have been pushed by remote */
786 /* but remote is not up to date */
787
788 /* If resync is in progress with current peer */
789 if (ps->flags & PEER_F_LEARN_ASSIGN) {
790
791 /* unassign current peer */
792 ps->flags &= ~PEER_F_LEARN_ASSIGN;
793 ps->table->flags &= ~(SHTABLE_F_RESYNC_ASSIGN|SHTABLE_F_RESYNC_PROCESS);
794
795 /* flag current peer is not up 2 date to try from an other */
796 ps->flags |= PEER_F_LEARN_NOTUP2DATE;
797
798 /* reschedule a resync */
799 ps->table->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
800 task_wakeup(ps->table->sync_task, TASK_WOKEN_MSG);
801 }
802 ps->confirm++;
803 }
804 else if (c == 'A') {
805 /* ack message */
806 uint32_t netinteger;
807
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200808 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200809 if (reql <= 0) { /* closed or EOL not found */
810 if (reql == 0) {
811 goto incomplete;
812 }
Willy Tarreaubc4af052011-02-13 13:25:14 +0100813 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200814 goto switchstate;
815 }
816 totl += reql;
817
818 /* Consider remote is up to date with "acked" version */
819 ps->update = ntohl(netinteger);
820 }
821 else {
822 /* Unknown message */
Willy Tarreaubc4af052011-02-13 13:25:14 +0100823 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200824 goto switchstate;
825 }
826
827 /* skip consumed message */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200828 bo_skip(si->ob, totl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200829
830 /* loop on that state to peek next message */
831 continue;
832incomplete:
833 /* Nothing to read, now we start to write */
834
835 /* Confirm finished or partial messages */
836 while (ps->confirm) {
837 /* There is a confirm messages to send */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200838 repl = bi_putchr(si->ib, 'c');
Emeric Brun2b920a12010-09-23 18:30:22 +0200839 if (repl <= 0) {
840 /* no more write possible */
841 if (repl == -1)
842 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100843 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200844 goto switchstate;
845 }
846 ps->confirm--;
847 }
848
849 /* Need to request a resync */
850 if ((ps->flags & PEER_F_LEARN_ASSIGN) &&
851 (ps->table->flags & SHTABLE_F_RESYNC_ASSIGN) &&
852 !(ps->table->flags & SHTABLE_F_RESYNC_PROCESS)) {
853 /* Current peer was elected to request a resync */
854
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200855 repl = bi_putchr(si->ib, 'R');
Emeric Brun2b920a12010-09-23 18:30:22 +0200856 if (repl <= 0) {
857 /* no more write possible */
858 if (repl == -1)
859 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100860 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200861 goto switchstate;
862 }
863 ps->table->flags |= SHTABLE_F_RESYNC_PROCESS;
864 }
865
866 /* It remains some updates to ack */
867 if (ps->pushack != ps->lastack) {
868 uint32_t netinteger;
869
870 trash[0] = 'A';
871 netinteger = htonl(ps->pushack);
872 memcpy(&trash[1], &netinteger, sizeof(netinteger));
873
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200874 repl = bi_putblk(si->ib, trash, 1+sizeof(netinteger));
Emeric Brun2b920a12010-09-23 18:30:22 +0200875 if (repl <= 0) {
876 /* no more write possible */
877 if (repl == -1)
878 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100879 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200880 goto switchstate;
881 }
882 ps->lastack = ps->pushack;
883 }
884
885 if (ps->flags & PEER_F_TEACH_PROCESS) {
886 /* current peer was requested for a lesson */
887
888 if (!(ps->flags & PEER_F_TEACH_STAGE1)) {
889 /* lesson stage 1 not complete */
890 struct eb32_node *eb;
891
892 eb = eb32_lookup_ge(&ps->table->table->updates, ps->pushed+1);
893 while (1) {
894 int msglen;
895 struct stksess *ts;
896
897 if (!eb) {
898 /* flag lesson stage1 complete */
899 ps->flags |= PEER_F_TEACH_STAGE1;
900 eb = eb32_first(&ps->table->table->updates);
901 if (eb)
902 ps->pushed = eb->key - 1;
903 break;
904 }
905
906 ts = eb32_entry(eb, struct stksess, upd);
907 msglen = peer_prepare_datamsg(ts, ps, trash, sizeof(trash));
908 if (msglen) {
909 /* message to buffer */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200910 repl = bi_putblk(si->ib, trash, msglen);
Emeric Brun2b920a12010-09-23 18:30:22 +0200911 if (repl <= 0) {
912 /* no more write possible */
913 if (repl == -1)
914 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100915 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200916 goto switchstate;
917 }
918 ps->lastpush = ps->pushed = ts->upd.key;
919 }
920 eb = eb32_next(eb);
921 }
922 } /* !TEACH_STAGE1 */
923
924 if (!(ps->flags & PEER_F_TEACH_STAGE2)) {
925 /* lesson stage 2 not complete */
926 struct eb32_node *eb;
927
928 eb = eb32_lookup_ge(&ps->table->table->updates, ps->pushed+1);
929 while (1) {
930 int msglen;
931 struct stksess *ts;
932
933 if (!eb || eb->key > ps->teaching_origin) {
934 /* flag lesson stage1 complete */
935 ps->flags |= PEER_F_TEACH_STAGE2;
936 ps->pushed = ps->teaching_origin;
937 break;
938 }
939
940 ts = eb32_entry(eb, struct stksess, upd);
941 msglen = peer_prepare_datamsg(ts, ps, trash, sizeof(trash));
942 if (msglen) {
943 /* message to buffer */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200944 repl = bi_putblk(si->ib, trash, msglen);
Emeric Brun2b920a12010-09-23 18:30:22 +0200945 if (repl <= 0) {
946 /* no more write possible */
947 if (repl == -1)
948 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100949 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200950 goto switchstate;
951 }
952 ps->lastpush = ps->pushed = ts->upd.key;
953 }
954 eb = eb32_next(eb);
955 }
956 } /* !TEACH_STAGE2 */
957
958 if (!(ps->flags & PEER_F_TEACH_FINISHED)) {
959 /* process final lesson message */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200960 repl = bi_putchr(si->ib, ((ps->table->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FINISHED) ? 'F' : 'C');
Emeric Brun2b920a12010-09-23 18:30:22 +0200961 if (repl <= 0) {
962 /* no more write possible */
963 if (repl == -1)
964 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100965 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200966 goto switchstate;
967 }
968
969 /* flag finished message sent */
970 ps->flags |= PEER_F_TEACH_FINISHED;
971 } /* !TEACH_FINISHED */
972 } /* TEACH_PROCESS */
973
974 if (!(ps->flags & PEER_F_LEARN_ASSIGN) &&
975 (int)(ps->pushed - ps->table->table->localupdate) < 0) {
976 /* Push local updates, only if no learning in progress (to avoid ping-pong effects) */
977 struct eb32_node *eb;
978
979 eb = eb32_lookup_ge(&ps->table->table->updates, ps->pushed+1);
980 while (1) {
981 int msglen;
982 struct stksess *ts;
983
984 /* push local updates */
985 if (!eb) {
986 eb = eb32_first(&ps->table->table->updates);
987 if (!eb || ((int)(eb->key - ps->pushed) <= 0)) {
988 ps->pushed = ps->table->table->localupdate;
989 break;
990 }
991 }
992
993 if ((int)(eb->key - ps->table->table->localupdate) > 0) {
994 ps->pushed = ps->table->table->localupdate;
995 break;
996 }
997
998 ts = eb32_entry(eb, struct stksess, upd);
999 msglen = peer_prepare_datamsg(ts, ps, trash, sizeof(trash));
1000 if (msglen) {
1001 /* message to buffer */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +02001002 repl = bi_putblk(si->ib, trash, msglen);
Emeric Brun2b920a12010-09-23 18:30:22 +02001003 if (repl <= 0) {
1004 /* no more write possible */
1005 if (repl == -1)
1006 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +01001007 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02001008 goto switchstate;
1009 }
1010 ps->lastpush = ps->pushed = ts->upd.key;
1011 }
1012 eb = eb32_next(eb);
1013 }
1014 } /* ! LEARN_ASSIGN */
1015 /* noting more to do */
1016 goto out;
1017 }
1018 case PEER_SESSION_EXIT:
Willy Tarreaubc4af052011-02-13 13:25:14 +01001019 repl = snprintf(trash, sizeof(trash), "%d\n", si->applet.st1);
Emeric Brun2b920a12010-09-23 18:30:22 +02001020
Willy Tarreau9dab5fc2012-05-07 11:56:55 +02001021 if (bi_putblk(si->ib, trash, repl) == -1)
Emeric Brun2b920a12010-09-23 18:30:22 +02001022 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +01001023 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02001024 /* fall through */
1025 case PEER_SESSION_END: {
Willy Tarreau060781f2012-05-07 16:50:03 +02001026 si->sock.shutw(si);
1027 si->sock.shutr(si);
Emeric Brun2b920a12010-09-23 18:30:22 +02001028 si->ib->flags |= BF_READ_NULL;
1029 goto quit;
1030 }
1031 }
1032 }
1033out:
Willy Tarreau060781f2012-05-07 16:50:03 +02001034 si->sock.update(si);
Emeric Brun2b920a12010-09-23 18:30:22 +02001035 si->ob->flags |= BF_READ_DONTWAIT;
1036 /* we don't want to expire timeouts while we're processing requests */
1037 si->ib->rex = TICK_ETERNITY;
1038 si->ob->wex = TICK_ETERNITY;
1039quit:
1040 return;
1041}
1042
Willy Tarreaub24281b2011-02-13 13:16:36 +01001043static struct si_applet peer_applet = {
1044 .name = "<PEER>", /* used for logging */
1045 .fct = peer_io_handler,
Aman Gupta9a13e842012-04-02 18:57:53 -07001046 .release = peer_session_release,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001047};
Emeric Brun2b920a12010-09-23 18:30:22 +02001048
1049/*
1050 * Use this function to force a close of a peer session
1051 */
Simon Horman96553772011-06-08 09:18:51 +09001052static void peer_session_forceshutdown(struct session * session)
Emeric Brun2b920a12010-09-23 18:30:22 +02001053{
1054 struct stream_interface *oldsi;
1055
Willy Tarreau7c0a1512011-03-10 11:17:02 +01001056 if (session->si[0].target.type == TARG_TYPE_APPLET &&
1057 session->si[0].target.ptr.a == &peer_applet) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001058 oldsi = &session->si[0];
1059 }
1060 else {
1061 oldsi = &session->si[1];
1062 }
1063
1064 /* call release to reinit resync states if needed */
1065 peer_session_release(oldsi);
Willy Tarreaubc4af052011-02-13 13:25:14 +01001066 oldsi->applet.st0 = PEER_SESSION_END;
1067 oldsi->applet.private = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001068 task_wakeup(session->task, TASK_WOKEN_MSG);
1069}
1070
1071/*
1072 * this function is called on a read event from a listen socket, corresponding
1073 * to an accept. It tries to accept as many connections as possible.
Willy Tarreaubd55e312010-11-11 10:55:09 +01001074 * It returns a positive value upon success, 0 if the connection needs to be
1075 * closed and ignored, or a negative value upon critical failure.
Emeric Brun2b920a12010-09-23 18:30:22 +02001076 */
1077int peer_accept(struct session *s)
1078{
1079 /* we have a dedicated I/O handler for the stats */
Willy Tarreaub24281b2011-02-13 13:16:36 +01001080 stream_int_register_handler(&s->si[1], &peer_applet);
Willy Tarreau7b7a8e92011-03-27 19:53:06 +02001081 copy_target(&s->target, &s->si[1].target); // for logging only
Willy Tarreaubc4af052011-02-13 13:25:14 +01001082 s->si[1].applet.private = s;
1083 s->si[1].applet.st0 = PEER_SESSION_ACCEPT;
Emeric Brun2b920a12010-09-23 18:30:22 +02001084
1085 tv_zero(&s->logs.tv_request);
1086 s->logs.t_queue = 0;
1087 s->logs.t_connect = 0;
1088 s->logs.t_data = 0;
1089 s->logs.t_close = 0;
1090 s->logs.bytes_in = s->logs.bytes_out = 0;
1091 s->logs.prx_queue_size = 0;/* we get the number of pending conns before us */
1092 s->logs.srv_queue_size = 0; /* we will get this number soon */
1093
1094 s->req->flags |= BF_READ_DONTWAIT; /* we plan to read small requests */
1095
1096 if (s->listener->timeout) {
1097 s->req->rto = *s->listener->timeout;
1098 s->rep->wto = *s->listener->timeout;
1099 }
1100 return 1;
1101}
1102
1103/*
Willy Tarreaubd55e312010-11-11 10:55:09 +01001104 * Create a new peer session in assigned state (connect will start automatically)
Emeric Brun2b920a12010-09-23 18:30:22 +02001105 */
Simon Horman96553772011-06-08 09:18:51 +09001106static struct session *peer_session_create(struct peer *peer, struct peer_session *ps)
Emeric Brun2b920a12010-09-23 18:30:22 +02001107{
1108 struct listener *l = ((struct proxy *)peer->peers->peers_fe)->listen;
1109 struct proxy *p = (struct proxy *)l->frontend; /* attached frontend */
1110 struct session *s;
1111 struct http_txn *txn;
1112 struct task *t;
1113
1114 if ((s = pool_alloc2(pool2_session)) == NULL) { /* disable this proxy for a while */
1115 Alert("out of memory in event_accept().\n");
Emeric Brun2b920a12010-09-23 18:30:22 +02001116 goto out_close;
1117 }
1118
1119 LIST_ADDQ(&sessions, &s->list);
1120 LIST_INIT(&s->back_refs);
1121
1122 s->flags = SN_ASSIGNED|SN_ADDR_SET;
1123 s->term_trace = 0;
1124
1125 /* if this session comes from a known monitoring system, we want to ignore
1126 * it as soon as possible, which means closing it immediately for TCP.
1127 */
1128 if ((t = task_new()) == NULL) { /* disable this proxy for a while */
1129 Alert("out of memory in event_accept().\n");
Emeric Brun2b920a12010-09-23 18:30:22 +02001130 goto out_free_session;
1131 }
1132
1133 ps->reconnect = tick_add(now_ms, MS_TO_TICKS(5000));
1134 ps->statuscode = PEER_SESSION_CONNECTCODE;
1135
1136 t->process = l->handler;
1137 t->context = s;
1138 t->nice = l->nice;
1139
Willy Tarreau6471afb2011-09-23 10:54:59 +02001140 memcpy(&s->si[1].addr.to, &peer->addr, sizeof(s->si[1].addr.to));
Emeric Brun2b920a12010-09-23 18:30:22 +02001141 s->task = t;
1142 s->listener = l;
1143
1144 /* Note: initially, the session's backend points to the frontend.
1145 * This changes later when switching rules are executed or
1146 * when the default backend is assigned.
1147 */
1148 s->be = s->fe = p;
1149
1150 s->req = s->rep = NULL; /* will be allocated later */
1151
1152 s->si[0].fd = -1;
1153 s->si[0].owner = t;
1154 s->si[0].state = s->si[0].prev_state = SI_ST_EST;
1155 s->si[0].err_type = SI_ET_NONE;
1156 s->si[0].err_loc = NULL;
Willy Tarreau26d8c592012-05-07 18:12:14 +02001157 s->si[0].proto = NULL;
1158 s->si[0].release = NULL;
Willy Tarreau63e7fe32012-05-08 15:20:43 +02001159 s->si[0].send_proxy_ofs = 0;
Willy Tarreau1539a012012-05-11 14:47:34 +02001160 set_target_client(&s->si[0].target);
Emeric Brun2b920a12010-09-23 18:30:22 +02001161 s->si[0].exp = TICK_ETERNITY;
1162 s->si[0].flags = SI_FL_NONE;
1163 if (s->fe->options2 & PR_O2_INDEPSTR)
1164 s->si[0].flags |= SI_FL_INDEP_STR;
Willy Tarreaubc4af052011-02-13 13:25:14 +01001165 s->si[0].applet.private = (void *)ps;
1166 s->si[0].applet.st0 = PEER_SESSION_CONNECT;
Emeric Brun2b920a12010-09-23 18:30:22 +02001167
Willy Tarreaub24281b2011-02-13 13:16:36 +01001168 stream_int_register_handler(&s->si[0], &peer_applet);
Emeric Brun2b920a12010-09-23 18:30:22 +02001169
1170 s->si[1].fd = -1; /* just to help with debugging */
1171 s->si[1].owner = t;
1172 s->si[1].state = s->si[1].prev_state = SI_ST_ASS;
1173 s->si[1].conn_retries = p->conn_retries;
1174 s->si[1].err_type = SI_ET_NONE;
1175 s->si[1].err_loc = NULL;
Willy Tarreau26d8c592012-05-07 18:12:14 +02001176 s->si[1].proto = peer->proto;
1177 s->si[1].release = NULL;
Willy Tarreau63e7fe32012-05-08 15:20:43 +02001178 s->si[1].send_proxy_ofs = 0;
Willy Tarreau9e000c62011-03-10 14:03:36 +01001179 set_target_proxy(&s->si[1].target, s->be);
Willy Tarreauf873d752012-05-11 17:47:17 +02001180 stream_interface_prepare(&s->si[1], &sock_raw);
Emeric Brun2b920a12010-09-23 18:30:22 +02001181 s->si[1].exp = TICK_ETERNITY;
1182 s->si[1].flags = SI_FL_NONE;
1183 if (s->be->options2 & PR_O2_INDEPSTR)
1184 s->si[1].flags |= SI_FL_INDEP_STR;
1185
Willy Tarreau9bd0d742011-07-20 00:17:39 +02001186 session_init_srv_conn(s);
Simon Horman8b7b05a2011-08-13 08:03:48 +09001187 set_target_proxy(&s->target, s->be);
Emeric Brun2b920a12010-09-23 18:30:22 +02001188 s->pend_pos = NULL;
1189
1190 /* init store persistence */
1191 s->store_count = 0;
1192 s->stkctr1_entry = NULL;
1193 s->stkctr2_entry = NULL;
1194
1195 /* FIXME: the logs are horribly complicated now, because they are
1196 * defined in <p>, <p>, and later <be> and <be>.
1197 */
1198
1199 s->logs.logwait = 0;
1200 s->do_log = NULL;
1201
1202 /* default error reporting function, may be changed by analysers */
1203 s->srv_error = default_srv_error;
1204
Emeric Brun2b920a12010-09-23 18:30:22 +02001205 s->uniq_id = 0;
Willy Tarreaubd833142012-05-08 15:51:44 +02001206 s->unique_id = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001207
1208 txn = &s->txn;
1209 /* Those variables will be checked and freed if non-NULL in
1210 * session.c:session_free(). It is important that they are
1211 * properly initialized.
1212 */
1213 txn->sessid = NULL;
1214 txn->srv_cookie = NULL;
1215 txn->cli_cookie = NULL;
1216 txn->uri = NULL;
1217 txn->req.cap = NULL;
1218 txn->rsp.cap = NULL;
1219 txn->hdr_idx.v = NULL;
1220 txn->hdr_idx.size = txn->hdr_idx.used = 0;
1221
1222 if ((s->req = pool_alloc2(pool2_buffer)) == NULL)
1223 goto out_fail_req; /* no memory */
1224
1225 s->req->size = global.tune.bufsize;
1226 buffer_init(s->req);
1227 s->req->prod = &s->si[0];
1228 s->req->cons = &s->si[1];
1229 s->si[0].ib = s->si[1].ob = s->req;
1230
1231 s->req->flags |= BF_READ_ATTACHED; /* the producer is already connected */
1232
1233 /* activate default analysers enabled for this listener */
1234 s->req->analysers = l->analysers;
1235
1236 /* note: this should not happen anymore since there's always at least the switching rules */
1237 if (!s->req->analysers) {
1238 buffer_auto_connect(s->req);/* don't wait to establish connection */
1239 buffer_auto_close(s->req);/* let the producer forward close requests */
1240 }
1241
1242 s->req->rto = s->fe->timeout.client;
1243 s->req->wto = s->be->timeout.server;
1244
1245 if ((s->rep = pool_alloc2(pool2_buffer)) == NULL)
1246 goto out_fail_rep; /* no memory */
1247
1248 s->rep->size = global.tune.bufsize;
1249 buffer_init(s->rep);
1250 s->rep->prod = &s->si[1];
1251 s->rep->cons = &s->si[0];
1252 s->si[0].ob = s->si[1].ib = s->rep;
1253
1254 s->rep->rto = s->be->timeout.server;
1255 s->rep->wto = s->fe->timeout.client;
1256
1257 s->req->rex = TICK_ETERNITY;
1258 s->req->wex = TICK_ETERNITY;
1259 s->req->analyse_exp = TICK_ETERNITY;
1260 s->rep->rex = TICK_ETERNITY;
1261 s->rep->wex = TICK_ETERNITY;
1262 s->rep->analyse_exp = TICK_ETERNITY;
1263 t->expire = TICK_ETERNITY;
1264
1265 s->rep->flags |= BF_READ_DONTWAIT;
1266 /* it is important not to call the wakeup function directly but to
1267 * pass through task_wakeup(), because this one knows how to apply
1268 * priorities to tasks.
1269 */
1270 task_wakeup(t, TASK_WOKEN_INIT);
1271
1272 l->nbconn++; /* warning! right now, it's up to the handler to decrease this */
1273 p->feconn++;/* beconn will be increased later */
1274 jobs++;
Willy Tarreau3c63fd82011-09-07 18:00:47 +02001275 if (!(s->listener->options & LI_O_UNLIMITED))
1276 actconn++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001277 totalconn++;
1278
1279 return s;
1280
1281 /* Error unrolling */
1282 out_fail_rep:
1283 pool_free2(pool2_buffer, s->req);
1284 out_fail_req:
1285 task_free(t);
1286 out_free_session:
1287 LIST_DEL(&s->list);
1288 pool_free2(pool2_session, s);
1289 out_close:
1290 return s;
1291}
1292
1293/*
1294 * Task processing function to manage re-connect and peer session
1295 * tasks wakeup on local update.
1296 */
Simon Horman96553772011-06-08 09:18:51 +09001297static struct task *process_peer_sync(struct task * task)
Emeric Brun2b920a12010-09-23 18:30:22 +02001298{
1299 struct shared_table *st = (struct shared_table *)task->context;
1300 struct peer_session *ps;
1301
1302 task->expire = TICK_ETERNITY;
1303
1304 if (!stopping) {
1305 /* Normal case (not soft stop)*/
1306 if (((st->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMLOCAL) &&
1307 (!nb_oldpids || tick_is_expired(st->resync_timeout, now_ms)) &&
1308 !(st->flags & SHTABLE_F_RESYNC_ASSIGN)) {
1309 /* Resync from local peer needed
1310 no peer was assigned for the lesson
1311 and no old local peer found
1312 or resync timeout expire */
1313
1314 /* flag no more resync from local, to try resync from remotes */
1315 st->flags |= SHTABLE_F_RESYNC_LOCAL;
1316
1317 /* reschedule a resync */
1318 st->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
1319 }
1320
1321 /* For each session */
1322 for (ps = st->sessions; ps; ps = ps->next) {
1323 /* For each remote peers */
1324 if (!ps->peer->local) {
1325 if (!ps->session) {
1326 /* no active session */
1327 if (ps->statuscode == 0 ||
1328 ps->statuscode == PEER_SESSION_SUCCESSCODE ||
1329 ((ps->statuscode == PEER_SESSION_CONNECTCODE ||
1330 ps->statuscode == PEER_SESSION_CONNECTEDCODE) &&
1331 tick_is_expired(ps->reconnect, now_ms))) {
1332 /* connection never tried
1333 * or previous session established with success
1334 * or previous session failed during connection
1335 * and reconnection timer is expired */
1336
1337 /* retry a connect */
1338 ps->session = peer_session_create(ps->peer, ps);
1339 }
1340 else if (ps->statuscode == PEER_SESSION_CONNECTCODE ||
1341 ps->statuscode == PEER_SESSION_CONNECTEDCODE) {
1342 /* If previous session failed during connection
1343 * but reconnection timer is not expired */
1344
1345 /* reschedule task for reconnect */
1346 task->expire = tick_first(task->expire, ps->reconnect);
1347 }
1348 /* else do nothing */
1349 } /* !ps->session */
1350 else if (ps->statuscode == PEER_SESSION_SUCCESSCODE) {
1351 /* current session is active and established */
1352 if (((st->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMREMOTE) &&
1353 !(st->flags & SHTABLE_F_RESYNC_ASSIGN) &&
1354 !(ps->flags & PEER_F_LEARN_NOTUP2DATE)) {
1355 /* Resync from a remote is needed
1356 * and no peer was assigned for lesson
1357 * and current peer may be up2date */
1358
1359 /* assign peer for the lesson */
1360 ps->flags |= PEER_F_LEARN_ASSIGN;
1361 st->flags |= SHTABLE_F_RESYNC_ASSIGN;
1362
1363 /* awake peer session task to handle a request of resync */
1364 task_wakeup(ps->session->task, TASK_WOKEN_MSG);
1365 }
1366 else if ((int)(ps->pushed - ps->table->table->localupdate) < 0) {
1367 /* awake peer session task to push local updates */
1368 task_wakeup(ps->session->task, TASK_WOKEN_MSG);
1369 }
1370 /* else do nothing */
1371 } /* SUCCESSCODE */
1372 } /* !ps->peer->local */
1373 } /* for */
1374
1375 /* Resync from remotes expired: consider resync is finished */
1376 if (((st->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMREMOTE) &&
1377 !(st->flags & SHTABLE_F_RESYNC_ASSIGN) &&
1378 tick_is_expired(st->resync_timeout, now_ms)) {
1379 /* Resync from remote peer needed
1380 * no peer was assigned for the lesson
1381 * and resync timeout expire */
1382
1383 /* flag no more resync from remote, consider resync is finished */
1384 st->flags |= SHTABLE_F_RESYNC_REMOTE;
1385 }
1386
1387 if ((st->flags & SHTABLE_RESYNC_STATEMASK) != SHTABLE_RESYNC_FINISHED) {
1388 /* Resync not finished*/
1389 /* reschedule task to resync timeout, to ended resync if needed */
1390 task->expire = tick_first(task->expire, st->resync_timeout);
1391 }
1392 } /* !stopping */
1393 else {
1394 /* soft stop case */
1395 if (task->state & TASK_WOKEN_SIGNAL) {
1396 /* We've just recieved the signal */
1397 if (!(st->flags & SHTABLE_F_DONOTSTOP)) {
1398 /* add DO NOT STOP flag if not present */
1399 jobs++;
1400 st->flags |= SHTABLE_F_DONOTSTOP;
1401 }
1402
1403 /* disconnect all connected peers */
1404 for (ps = st->sessions; ps; ps = ps->next) {
1405 if (ps->session) {
1406 peer_session_forceshutdown(ps->session);
1407 ps->session = NULL;
1408 }
1409 }
1410 }
1411 ps = st->local_session;
1412
1413 if (ps->flags & PEER_F_TEACH_COMPLETE) {
1414 if (st->flags & SHTABLE_F_DONOTSTOP) {
1415 /* resync of new process was complete, current process can die now */
1416 jobs--;
1417 st->flags &= ~SHTABLE_F_DONOTSTOP;
1418 }
1419 }
1420 else if (!ps->session) {
1421 /* If session is not active */
1422 if (ps->statuscode == 0 ||
1423 ps->statuscode == PEER_SESSION_SUCCESSCODE ||
1424 ps->statuscode == PEER_SESSION_CONNECTEDCODE ||
1425 ps->statuscode == PEER_SESSION_TRYAGAIN) {
1426 /* connection never tried
1427 * or previous session was successfully established
1428 * or previous session tcp connect success but init state incomplete
1429 * or during previous connect, peer replies a try again statuscode */
1430
1431 /* connect to the peer */
1432 ps->session = peer_session_create(ps->peer, ps);
1433 }
1434 else {
1435 /* Other error cases */
1436 if (st->flags & SHTABLE_F_DONOTSTOP) {
1437 /* unable to resync new process, current process can die now */
1438 jobs--;
1439 st->flags &= ~SHTABLE_F_DONOTSTOP;
1440 }
1441 }
1442 }
1443 else if (ps->statuscode == PEER_SESSION_SUCCESSCODE &&
1444 (int)(ps->pushed - ps->table->table->localupdate) < 0) {
1445 /* current session active and established
1446 awake session to push remaining local updates */
1447 task_wakeup(ps->session->task, TASK_WOKEN_MSG);
1448 }
1449 } /* stopping */
1450 /* Wakeup for re-connect */
1451 return task;
1452}
1453
1454/*
1455 * Function used to register a table for sync on a group of peers
1456 *
1457 */
1458void peers_register_table(struct peers *peers, struct stktable *table)
1459{
1460 struct shared_table *st;
1461 struct peer * curpeer;
1462 struct peer_session *ps;
1463
1464 st = (struct shared_table *)calloc(1,sizeof(struct shared_table));
1465 st->table = table;
1466 st->next = peers->tables;
1467 st->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
1468 peers->tables = st;
1469
1470 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
1471 ps = (struct peer_session *)calloc(1,sizeof(struct peer_session));
1472 ps->table = st;
1473 ps->peer = curpeer;
1474 if (curpeer->local)
1475 st->local_session = ps;
1476 ps->next = st->sessions;
1477 ps->reconnect = now_ms;
1478 st->sessions = ps;
1479 peers->peers_fe->maxconn += 3;
1480 }
1481
1482 peers->peers_fe->listen->maxconn = peers->peers_fe->maxconn;
1483 st->sync_task = task_new();
1484 st->sync_task->process = process_peer_sync;
1485 st->sync_task->expire = TICK_ETERNITY;
1486 st->sync_task->context = (void *)st;
1487 table->sync_task =st->sync_task;
1488 signal_register_task(0, table->sync_task, 0);
1489 task_wakeup(st->sync_task, TASK_WOKEN_INIT);
1490}
1491