blob: 0516d0cc93c7abad7fb1ee3ded0f7a839c013d28 [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>
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 Tarreauc7e42382012-08-24 19:22:53 +020033#include <proto/channel.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020034#include <proto/fd.h>
35#include <proto/log.h>
36#include <proto/hdr_idx.h>
Emeric Brun2b920a12010-09-23 18:30:22 +020037#include <proto/proto_tcp.h>
38#include <proto/proto_http.h>
39#include <proto/proxy.h>
40#include <proto/session.h>
41#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 Tarreauf2943dc2012-10-26 20:10:28 +0200188 struct peer_session *ps = (struct peer_session *)si->conn->xprt_ctx;
Emeric Brun2b920a12010-09-23 18:30:22 +0200189
Willy Tarreauf2943dc2012-10-26 20:10:28 +0200190 /* si->conn->xprt_ctx is not a peer session */
Willy Tarreaubc4af052011-02-13 13:25:14 +0100191 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 Tarreauf2943dc2012-10-26 20:10:28 +0200230 si->conn->xprt_ctx = NULL;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100231 si->applet.st0 = PEER_SESSION_GETVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200232 /* fall through */
233 case PEER_SESSION_GETVERSION:
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100234 reql = bo_getline(si->ob, trash.str, trash.size);
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 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100241 if (trash.str[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 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100245 else if (reql > 1 && (trash.str[reql-2] == '\r'))
246 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200247 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100248 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200249
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200250 bo_skip(si->ob, reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200251
252 /* test version */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100253 if (strcmp(PEER_SESSION_PROTO_NAME " 1.0", trash.str) != 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 */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100257 if (strncmp(PEER_SESSION_PROTO_NAME " ", trash.str, 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 Tarreau19d14ef2012-10-29 16:51:55 +0100265 reql = bo_getline(si->ob, trash.str, trash.size);
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 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100272 if (trash.str[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 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100276 else if (reql > 1 && (trash.str[reql-2] == '\r'))
277 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200278 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100279 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200280
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 */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100284 if (strcmp(localpeer, trash.str) != 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 Tarreau19d14ef2012-10-29 16:51:55 +0100295 reql = bo_getline(si->ob, trash.str, trash.size);
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 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100302 if (trash.str[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200303 /* 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 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100307 else if (reql > 1 && (trash.str[reql-2] == '\r'))
308 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200309 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100310 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200311
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>" */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100315 p = strchr(trash.str, ' ');
Emeric Brun2b920a12010-09-23 18:30:22 +0200316 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) {
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100325 if (strcmp(curpeer->id, trash.str) == 0)
Emeric Brun2b920a12010-09-23 18:30:22 +0200326 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 Tarreauf2943dc2012-10-26 20:10:28 +0200336 si->conn->xprt_ctx = curpeer;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100337 si->applet.st0 = PEER_SESSION_GETTABLE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200338 /* fall through */
339 }
340 case PEER_SESSION_GETTABLE: {
Willy Tarreauf2943dc2012-10-26 20:10:28 +0200341 struct peer *curpeer = (struct peer *)si->conn->xprt_ctx;
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 Tarreau19d14ef2012-10-29 16:51:55 +0100348 reql = bo_getline(si->ob, trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200349 if (reql <= 0) { /* closed or EOL not found */
350 if (reql == 0)
351 goto out;
Willy Tarreauf2943dc2012-10-26 20:10:28 +0200352 si->conn->xprt_ctx = NULL;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100353 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200354 goto switchstate;
355 }
Willy Tarreauf2943dc2012-10-26 20:10:28 +0200356 /* Re init si->conn->xprt_ctx to null, to handle correctly a release case */
357 si->conn->xprt_ctx = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +0200358
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100359 if (trash.str[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200360 /* 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 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100364 else if (reql > 1 && (trash.str[reql-2] == '\r'))
365 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200366 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100367 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200368
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>" */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100372 p = strchr(trash.str, ' ');
Emeric Brun2b920a12010-09-23 18:30:22 +0200373 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 Tarreauf2943dc2012-10-26 20:10:28 +0200383 si->conn->xprt_ctx = NULL;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100384 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 */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100392 if (strcmp(st->table->id, trash.str) == 0) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200393 /* 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 Tarreauf2943dc2012-10-26 20:10:28 +0200442 si->conn->xprt_ctx = ps;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100443 si->applet.st0 = PEER_SESSION_SENDSUCCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200444 /* fall through */
445 }
446 case PEER_SESSION_SENDSUCCESS:{
Willy Tarreauf2943dc2012-10-26 20:10:28 +0200447 struct peer_session *ps = (struct peer_session *)si->conn->xprt_ctx;
Emeric Brun2b920a12010-09-23 18:30:22 +0200448
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100449 repl = snprintf(trash.str, trash.size, "%d\n", PEER_SESSION_SUCCESSCODE);
450 repl = bi_putblk(si->ib, trash.str, 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 Tarreauf2943dc2012-10-26 20:10:28 +0200497 struct peer_session *ps = (struct peer_session *)si->conn->xprt_ctx;
Emeric Brun2b920a12010-09-23 18:30:22 +0200498
499 /* Send headers */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100500 repl = snprintf(trash.str, trash.size,
Emeric Brun2b920a12010-09-23 18:30:22 +0200501 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
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100509 if (repl >= trash.size) {
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 Tarreau19d14ef2012-10-29 16:51:55 +0100514 repl = bi_putblk(si->ib, trash.str, 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 Tarreauf2943dc2012-10-26 20:10:28 +0200527 struct peer_session *ps = (struct peer_session *)si->conn->xprt_ctx;
Emeric Brun2b920a12010-09-23 18:30:22 +0200528
Willy Tarreau03cdb7c2012-08-27 23:14:58 +0200529 if (si->ib->flags & CF_WRITE_PARTIAL)
Emeric Brun2b920a12010-09-23 18:30:22 +0200530 ps->statuscode = PEER_SESSION_CONNECTEDCODE;
531
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100532 reql = bo_getline(si->ob, trash.str, trash.size);
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 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100539 if (trash.str[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200540 /* 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 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100544 else if (reql > 1 && (trash.str[reql-2] == '\r'))
545 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200546 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100547 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200548
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 */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100552 ps->statuscode = atoi(trash.str);
Emeric Brun2b920a12010-09-23 18:30:22 +0200553
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 Tarreauf2943dc2012-10-26 20:10:28 +0200598 struct peer_session *ps = (struct peer_session *)si->conn->xprt_ctx;
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200599 struct stksess *ts, *newts = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +0200600 char c;
601 int totl = 0;
602
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200603 reql = bo_getblk(si->ob, (char *)&c, sizeof(c), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200604 if (reql <= 0) /* closed or EOL not found */
605 goto incomplete;
606
Emeric Brun2b920a12010-09-23 18:30:22 +0200607 totl += reql;
608
609 if ((c & 0x80) || (c == 'D')) {
610 /* Here we have data message */
611 unsigned int pushack;
Emeric Brun2b920a12010-09-23 18:30:22 +0200612 int srvid;
613 uint32_t netinteger;
614
615 /* Compute update remote version */
616 if (c & 0x80) {
617 pushack = ps->pushack + (unsigned int)(c & 0x7F);
618 }
619 else {
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200620 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200621 if (reql <= 0) /* closed or EOL not found */
622 goto incomplete;
623
Emeric Brun2b920a12010-09-23 18:30:22 +0200624 totl += reql;
625 pushack = ntohl(netinteger);
626 }
627
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200628 /* read key. We try to read it directly
629 * to the target memory location so that
630 * we are certain there is always enough
631 * space. However, if the allocation fails,
632 * we fall back to trash and we ignore the
633 * input data not to block the sync of other
634 * tables (think about a full table with
635 * "nopurge" for example).
636 */
Emeric Brun2b920a12010-09-23 18:30:22 +0200637 if (ps->table->table->type == STKTABLE_TYPE_STRING) {
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200638 /* read size first */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200639 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200640 if (reql <= 0) /* closed or EOL not found */
641 goto incomplete;
642
Emeric Brun2b920a12010-09-23 18:30:22 +0200643 totl += reql;
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200644 newts = stksess_new(ps->table->table, NULL);
645 reql = bo_getblk(si->ob, newts ? (char *)newts->key.key : trash.str, ntohl(netinteger), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200646 if (reql <= 0) /* closed or EOL not found */
647 goto incomplete;
648
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200649 /* always truncate the string to the approprite size */
650 if (newts)
651 newts->key.key[MIN(ntohl(netinteger), ps->table->table->key_size)] = 0;
652
Emeric Brun2b920a12010-09-23 18:30:22 +0200653 totl += reql;
654 }
655 else if (ps->table->table->type == STKTABLE_TYPE_INTEGER) {
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200656 newts = stksess_new(ps->table->table, NULL);
657 reql = bo_getblk(si->ob, newts ? (char *)newts->key.key : trash.str, sizeof(netinteger), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200658 if (reql <= 0) /* closed or EOL not found */
659 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +0200660 totl += reql;
Emeric Brun2b920a12010-09-23 18:30:22 +0200661 }
662 else {
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200663 /* type ip or binary */
664 newts = stksess_new(ps->table->table, NULL);
665 reql = bo_getblk(si->ob, newts ? (char *)newts->key.key : trash.str, ps->table->table->key_size, totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200666 if (reql <= 0) /* closed or EOL not found */
667 goto incomplete;
Willy Tarreau72d6c162013-04-11 16:14:13 +0200668 totl += reql;
Emeric Brun2b920a12010-09-23 18:30:22 +0200669 }
670
671 /* read server id */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200672 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200673 if (reql <= 0) /* closed or EOL not found */
674 goto incomplete;
675
Emeric Brun2b920a12010-09-23 18:30:22 +0200676 totl += reql;
677 srvid = ntohl(netinteger);
678
679 /* update entry */
Emeric Brun2b920a12010-09-23 18:30:22 +0200680 if (newts) {
681 /* lookup for existing entry */
682 ts = stktable_lookup(ps->table->table, newts);
683 if (ts) {
684 /* the entry already exist, we can free ours */
685 stktable_touch(ps->table->table, ts, 0);
686 stksess_free(ps->table->table, newts);
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200687 newts = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +0200688 }
689 else {
690 struct eb32_node *eb;
691
692 /* create new entry */
693 ts = stktable_store(ps->table->table, newts, 0);
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200694 newts = NULL; /* don't reuse it */
695
Emeric Brun2b920a12010-09-23 18:30:22 +0200696 ts->upd.key= (++ps->table->table->update)+(2^31);
697 eb = eb32_insert(&ps->table->table->updates, &ts->upd);
698 if (eb != &ts->upd) {
699 eb32_delete(eb);
700 eb32_insert(&ps->table->table->updates, &ts->upd);
701 }
702 }
703
704 /* update entry */
705 if (srvid && stktable_data_ptr(ps->table->table, ts, STKTABLE_DT_SERVER_ID))
706 stktable_data_cast(stktable_data_ptr(ps->table->table, ts, STKTABLE_DT_SERVER_ID), server_id) = srvid;
707 ps->pushack = pushack;
708 }
709
710 }
711 else if (c == 'R') {
712 /* Reset message: remote need resync */
713
714 /* reinit counters for a resync */
715 ps->lastpush = 0;
716 ps->teaching_origin = ps->pushed = ps->table->table->update;
717
718 /* reset teaching flags to 0 */
719 ps->flags &= PEER_TEACH_RESET;
720
721 /* flag to start to teach lesson */
722 ps->flags |= PEER_F_TEACH_PROCESS;
723 }
724 else if (c == 'F') {
725 /* Finish message, all known updates have been pushed by remote */
726 /* and remote is up to date */
727
728 /* If resync is in progress with remote peer */
729 if (ps->flags & PEER_F_LEARN_ASSIGN) {
730
731 /* unassign current peer for learning */
732 ps->flags &= ~PEER_F_LEARN_ASSIGN;
733 ps->table->flags &= ~(SHTABLE_F_RESYNC_ASSIGN|SHTABLE_F_RESYNC_PROCESS);
734
735 /* Consider table is now up2date, resync resync no more needed from local neither remote */
736 ps->table->flags |= (SHTABLE_F_RESYNC_LOCAL|SHTABLE_F_RESYNC_REMOTE);
737 }
738 /* Increase confirm counter to launch a confirm message */
739 ps->confirm++;
740 }
741 else if (c == 'c') {
742 /* confirm message, remote peer is now up to date with us */
743
744 /* If stopping state */
745 if (stopping) {
746 /* Close session, push resync no more needed */
747 ps->flags |= PEER_F_TEACH_COMPLETE;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100748 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200749 goto switchstate;
750 }
751
752 /* reset teaching flags to 0 */
753 ps->flags &= PEER_TEACH_RESET;
754 }
755 else if (c == 'C') {
756 /* Continue message, all known updates have been pushed by remote */
757 /* but remote is not up to date */
758
759 /* If resync is in progress with current peer */
760 if (ps->flags & PEER_F_LEARN_ASSIGN) {
761
762 /* unassign current peer */
763 ps->flags &= ~PEER_F_LEARN_ASSIGN;
764 ps->table->flags &= ~(SHTABLE_F_RESYNC_ASSIGN|SHTABLE_F_RESYNC_PROCESS);
765
766 /* flag current peer is not up 2 date to try from an other */
767 ps->flags |= PEER_F_LEARN_NOTUP2DATE;
768
769 /* reschedule a resync */
770 ps->table->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
771 task_wakeup(ps->table->sync_task, TASK_WOKEN_MSG);
772 }
773 ps->confirm++;
774 }
775 else if (c == 'A') {
776 /* ack message */
777 uint32_t netinteger;
778
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200779 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200780 if (reql <= 0) /* closed or EOL not found */
781 goto incomplete;
782
Emeric Brun2b920a12010-09-23 18:30:22 +0200783 totl += reql;
784
785 /* Consider remote is up to date with "acked" version */
786 ps->update = ntohl(netinteger);
787 }
788 else {
789 /* Unknown message */
Willy Tarreaubc4af052011-02-13 13:25:14 +0100790 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200791 goto switchstate;
792 }
793
794 /* skip consumed message */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200795 bo_skip(si->ob, totl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200796
797 /* loop on that state to peek next message */
Willy Tarreau72d6c162013-04-11 16:14:13 +0200798 goto switchstate;
799
Emeric Brun2b920a12010-09-23 18:30:22 +0200800incomplete:
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200801 /* we get here when a bo_getblk() returns <= 0 in reql */
802
803 /* first, we may have to release newts */
804 if (newts) {
805 stksess_free(ps->table->table, newts);
806 newts = NULL;
807 }
808
Willy Tarreau72d6c162013-04-11 16:14:13 +0200809 if (reql < 0) {
810 /* there was an error */
811 si->applet.st0 = PEER_SESSION_END;
812 goto switchstate;
813 }
814
Emeric Brun2b920a12010-09-23 18:30:22 +0200815 /* Nothing to read, now we start to write */
816
817 /* Confirm finished or partial messages */
818 while (ps->confirm) {
819 /* There is a confirm messages to send */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200820 repl = bi_putchr(si->ib, 'c');
Emeric Brun2b920a12010-09-23 18:30:22 +0200821 if (repl <= 0) {
822 /* no more write possible */
823 if (repl == -1)
824 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100825 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200826 goto switchstate;
827 }
828 ps->confirm--;
829 }
830
831 /* Need to request a resync */
832 if ((ps->flags & PEER_F_LEARN_ASSIGN) &&
833 (ps->table->flags & SHTABLE_F_RESYNC_ASSIGN) &&
834 !(ps->table->flags & SHTABLE_F_RESYNC_PROCESS)) {
835 /* Current peer was elected to request a resync */
836
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200837 repl = bi_putchr(si->ib, 'R');
Emeric Brun2b920a12010-09-23 18:30:22 +0200838 if (repl <= 0) {
839 /* no more write possible */
840 if (repl == -1)
841 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100842 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200843 goto switchstate;
844 }
845 ps->table->flags |= SHTABLE_F_RESYNC_PROCESS;
846 }
847
848 /* It remains some updates to ack */
849 if (ps->pushack != ps->lastack) {
850 uint32_t netinteger;
851
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100852 trash.str[0] = 'A';
Emeric Brun2b920a12010-09-23 18:30:22 +0200853 netinteger = htonl(ps->pushack);
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100854 memcpy(&trash.str[1], &netinteger, sizeof(netinteger));
Emeric Brun2b920a12010-09-23 18:30:22 +0200855
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100856 repl = bi_putblk(si->ib, trash.str, 1+sizeof(netinteger));
Emeric Brun2b920a12010-09-23 18:30:22 +0200857 if (repl <= 0) {
858 /* no more write possible */
859 if (repl == -1)
860 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100861 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200862 goto switchstate;
863 }
864 ps->lastack = ps->pushack;
865 }
866
867 if (ps->flags & PEER_F_TEACH_PROCESS) {
868 /* current peer was requested for a lesson */
869
870 if (!(ps->flags & PEER_F_TEACH_STAGE1)) {
871 /* lesson stage 1 not complete */
872 struct eb32_node *eb;
873
874 eb = eb32_lookup_ge(&ps->table->table->updates, ps->pushed+1);
875 while (1) {
876 int msglen;
877 struct stksess *ts;
878
879 if (!eb) {
880 /* flag lesson stage1 complete */
881 ps->flags |= PEER_F_TEACH_STAGE1;
882 eb = eb32_first(&ps->table->table->updates);
883 if (eb)
884 ps->pushed = eb->key - 1;
885 break;
886 }
887
888 ts = eb32_entry(eb, struct stksess, upd);
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100889 msglen = peer_prepare_datamsg(ts, ps, trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200890 if (msglen) {
891 /* message to buffer */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100892 repl = bi_putblk(si->ib, trash.str, msglen);
Emeric Brun2b920a12010-09-23 18:30:22 +0200893 if (repl <= 0) {
894 /* no more write possible */
895 if (repl == -1)
896 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100897 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200898 goto switchstate;
899 }
900 ps->lastpush = ps->pushed = ts->upd.key;
901 }
902 eb = eb32_next(eb);
903 }
904 } /* !TEACH_STAGE1 */
905
906 if (!(ps->flags & PEER_F_TEACH_STAGE2)) {
907 /* lesson stage 2 not complete */
908 struct eb32_node *eb;
909
910 eb = eb32_lookup_ge(&ps->table->table->updates, ps->pushed+1);
911 while (1) {
912 int msglen;
913 struct stksess *ts;
914
915 if (!eb || eb->key > ps->teaching_origin) {
916 /* flag lesson stage1 complete */
917 ps->flags |= PEER_F_TEACH_STAGE2;
918 ps->pushed = ps->teaching_origin;
919 break;
920 }
921
922 ts = eb32_entry(eb, struct stksess, upd);
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100923 msglen = peer_prepare_datamsg(ts, ps, trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200924 if (msglen) {
925 /* message to buffer */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100926 repl = bi_putblk(si->ib, trash.str, msglen);
Emeric Brun2b920a12010-09-23 18:30:22 +0200927 if (repl <= 0) {
928 /* no more write possible */
929 if (repl == -1)
930 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100931 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200932 goto switchstate;
933 }
934 ps->lastpush = ps->pushed = ts->upd.key;
935 }
936 eb = eb32_next(eb);
937 }
938 } /* !TEACH_STAGE2 */
939
940 if (!(ps->flags & PEER_F_TEACH_FINISHED)) {
941 /* process final lesson message */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200942 repl = bi_putchr(si->ib, ((ps->table->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FINISHED) ? 'F' : 'C');
Emeric Brun2b920a12010-09-23 18:30:22 +0200943 if (repl <= 0) {
944 /* no more write possible */
945 if (repl == -1)
946 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100947 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200948 goto switchstate;
949 }
950
951 /* flag finished message sent */
952 ps->flags |= PEER_F_TEACH_FINISHED;
953 } /* !TEACH_FINISHED */
954 } /* TEACH_PROCESS */
955
956 if (!(ps->flags & PEER_F_LEARN_ASSIGN) &&
957 (int)(ps->pushed - ps->table->table->localupdate) < 0) {
958 /* Push local updates, only if no learning in progress (to avoid ping-pong effects) */
959 struct eb32_node *eb;
960
961 eb = eb32_lookup_ge(&ps->table->table->updates, ps->pushed+1);
962 while (1) {
963 int msglen;
964 struct stksess *ts;
965
966 /* push local updates */
967 if (!eb) {
968 eb = eb32_first(&ps->table->table->updates);
969 if (!eb || ((int)(eb->key - ps->pushed) <= 0)) {
970 ps->pushed = ps->table->table->localupdate;
971 break;
972 }
973 }
974
975 if ((int)(eb->key - ps->table->table->localupdate) > 0) {
976 ps->pushed = ps->table->table->localupdate;
977 break;
978 }
979
980 ts = eb32_entry(eb, struct stksess, upd);
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100981 msglen = peer_prepare_datamsg(ts, ps, trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200982 if (msglen) {
983 /* message to buffer */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100984 repl = bi_putblk(si->ib, trash.str, msglen);
Emeric Brun2b920a12010-09-23 18:30:22 +0200985 if (repl <= 0) {
986 /* no more write possible */
987 if (repl == -1)
988 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100989 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200990 goto switchstate;
991 }
992 ps->lastpush = ps->pushed = ts->upd.key;
993 }
994 eb = eb32_next(eb);
995 }
996 } /* ! LEARN_ASSIGN */
997 /* noting more to do */
998 goto out;
999 }
1000 case PEER_SESSION_EXIT:
Willy Tarreau19d14ef2012-10-29 16:51:55 +01001001 repl = snprintf(trash.str, trash.size, "%d\n", si->applet.st1);
Emeric Brun2b920a12010-09-23 18:30:22 +02001002
Willy Tarreau19d14ef2012-10-29 16:51:55 +01001003 if (bi_putblk(si->ib, trash.str, repl) == -1)
Emeric Brun2b920a12010-09-23 18:30:22 +02001004 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +01001005 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02001006 /* fall through */
1007 case PEER_SESSION_END: {
Willy Tarreau73b013b2012-05-21 16:31:45 +02001008 si_shutw(si);
1009 si_shutr(si);
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001010 si->ib->flags |= CF_READ_NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001011 goto quit;
1012 }
1013 }
1014 }
1015out:
Willy Tarreau73b013b2012-05-21 16:31:45 +02001016 si_update(si);
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001017 si->ob->flags |= CF_READ_DONTWAIT;
Emeric Brun2b920a12010-09-23 18:30:22 +02001018 /* we don't want to expire timeouts while we're processing requests */
1019 si->ib->rex = TICK_ETERNITY;
1020 si->ob->wex = TICK_ETERNITY;
1021quit:
1022 return;
1023}
1024
Willy Tarreaub24281b2011-02-13 13:16:36 +01001025static struct si_applet peer_applet = {
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001026 .obj_type = OBJ_TYPE_APPLET,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001027 .name = "<PEER>", /* used for logging */
1028 .fct = peer_io_handler,
Aman Gupta9a13e842012-04-02 18:57:53 -07001029 .release = peer_session_release,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001030};
Emeric Brun2b920a12010-09-23 18:30:22 +02001031
1032/*
1033 * Use this function to force a close of a peer session
1034 */
Simon Horman96553772011-06-08 09:18:51 +09001035static void peer_session_forceshutdown(struct session * session)
Emeric Brun2b920a12010-09-23 18:30:22 +02001036{
1037 struct stream_interface *oldsi;
1038
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001039 if (objt_applet(session->si[0].conn->target) == &peer_applet) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001040 oldsi = &session->si[0];
1041 }
1042 else {
1043 oldsi = &session->si[1];
1044 }
1045
1046 /* call release to reinit resync states if needed */
1047 peer_session_release(oldsi);
Willy Tarreaubc4af052011-02-13 13:25:14 +01001048 oldsi->applet.st0 = PEER_SESSION_END;
Willy Tarreauf2943dc2012-10-26 20:10:28 +02001049 oldsi->conn->xprt_ctx = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001050 task_wakeup(session->task, TASK_WOKEN_MSG);
1051}
1052
1053/*
1054 * this function is called on a read event from a listen socket, corresponding
1055 * to an accept. It tries to accept as many connections as possible.
Willy Tarreaubd55e312010-11-11 10:55:09 +01001056 * It returns a positive value upon success, 0 if the connection needs to be
1057 * closed and ignored, or a negative value upon critical failure.
Emeric Brun2b920a12010-09-23 18:30:22 +02001058 */
1059int peer_accept(struct session *s)
1060{
1061 /* we have a dedicated I/O handler for the stats */
Willy Tarreaub24281b2011-02-13 13:16:36 +01001062 stream_int_register_handler(&s->si[1], &peer_applet);
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001063 s->target = s->si[1].conn->target; // for logging only
Willy Tarreauf2943dc2012-10-26 20:10:28 +02001064 s->si[1].conn->xprt_ctx = s;
Willy Tarreaubc4af052011-02-13 13:25:14 +01001065 s->si[1].applet.st0 = PEER_SESSION_ACCEPT;
Emeric Brun2b920a12010-09-23 18:30:22 +02001066
1067 tv_zero(&s->logs.tv_request);
1068 s->logs.t_queue = 0;
1069 s->logs.t_connect = 0;
1070 s->logs.t_data = 0;
1071 s->logs.t_close = 0;
1072 s->logs.bytes_in = s->logs.bytes_out = 0;
1073 s->logs.prx_queue_size = 0;/* we get the number of pending conns before us */
1074 s->logs.srv_queue_size = 0; /* we will get this number soon */
1075
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001076 s->req->flags |= CF_READ_DONTWAIT; /* we plan to read small requests */
Emeric Brun2b920a12010-09-23 18:30:22 +02001077
1078 if (s->listener->timeout) {
1079 s->req->rto = *s->listener->timeout;
1080 s->rep->wto = *s->listener->timeout;
1081 }
1082 return 1;
1083}
1084
1085/*
Willy Tarreaubd55e312010-11-11 10:55:09 +01001086 * Create a new peer session in assigned state (connect will start automatically)
Emeric Brun2b920a12010-09-23 18:30:22 +02001087 */
Simon Horman96553772011-06-08 09:18:51 +09001088static struct session *peer_session_create(struct peer *peer, struct peer_session *ps)
Emeric Brun2b920a12010-09-23 18:30:22 +02001089{
Willy Tarreau4348fad2012-09-20 16:48:07 +02001090 struct listener *l = LIST_NEXT(&peer->peers->peers_fe->conf.listeners, struct listener *, by_fe);
Emeric Brun2b920a12010-09-23 18:30:22 +02001091 struct proxy *p = (struct proxy *)l->frontend; /* attached frontend */
1092 struct session *s;
1093 struct http_txn *txn;
1094 struct task *t;
1095
1096 if ((s = pool_alloc2(pool2_session)) == NULL) { /* disable this proxy for a while */
Godbach430f2912013-06-20 13:28:38 +08001097 Alert("out of memory in peer_session_create().\n");
Emeric Brun2b920a12010-09-23 18:30:22 +02001098 goto out_close;
1099 }
1100
Willy Tarreauf2943dc2012-10-26 20:10:28 +02001101 if (unlikely((s->si[0].conn = pool_alloc2(pool2_connection)) == NULL))
1102 goto out_fail_conn0;
1103
1104 if (unlikely((s->si[1].conn = pool_alloc2(pool2_connection)) == NULL))
1105 goto out_fail_conn1;
1106
Emeric Brun2b920a12010-09-23 18:30:22 +02001107 LIST_ADDQ(&sessions, &s->list);
1108 LIST_INIT(&s->back_refs);
1109
1110 s->flags = SN_ASSIGNED|SN_ADDR_SET;
Emeric Brun2b920a12010-09-23 18:30:22 +02001111
1112 /* if this session comes from a known monitoring system, we want to ignore
1113 * it as soon as possible, which means closing it immediately for TCP.
1114 */
1115 if ((t = task_new()) == NULL) { /* disable this proxy for a while */
Godbach430f2912013-06-20 13:28:38 +08001116 Alert("out of memory in peer_session_create().\n");
Emeric Brun2b920a12010-09-23 18:30:22 +02001117 goto out_free_session;
1118 }
1119
1120 ps->reconnect = tick_add(now_ms, MS_TO_TICKS(5000));
1121 ps->statuscode = PEER_SESSION_CONNECTCODE;
1122
1123 t->process = l->handler;
1124 t->context = s;
1125 t->nice = l->nice;
1126
Willy Tarreauf2943dc2012-10-26 20:10:28 +02001127 memcpy(&s->si[1].conn->addr.to, &peer->addr, sizeof(s->si[1].conn->addr.to));
Emeric Brun2b920a12010-09-23 18:30:22 +02001128 s->task = t;
1129 s->listener = l;
1130
1131 /* Note: initially, the session's backend points to the frontend.
1132 * This changes later when switching rules are executed or
1133 * when the default backend is assigned.
1134 */
1135 s->be = s->fe = p;
1136
1137 s->req = s->rep = NULL; /* will be allocated later */
1138
Willy Tarreauf2943dc2012-10-26 20:10:28 +02001139 s->si[0].conn->t.sock.fd = -1;
1140 s->si[0].conn->flags = CO_FL_NONE;
Willy Tarreau14cba4b2012-11-30 17:33:05 +01001141 s->si[0].conn->err_code = CO_ER_NONE;
Emeric Brun2b920a12010-09-23 18:30:22 +02001142 s->si[0].owner = t;
1143 s->si[0].state = s->si[0].prev_state = SI_ST_EST;
1144 s->si[0].err_type = SI_ET_NONE;
1145 s->si[0].err_loc = NULL;
Willy Tarreau26d8c592012-05-07 18:12:14 +02001146 s->si[0].release = NULL;
Willy Tarreau63e7fe32012-05-08 15:20:43 +02001147 s->si[0].send_proxy_ofs = 0;
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001148 s->si[0].conn->target = &l->obj_type;
Emeric Brun2b920a12010-09-23 18:30:22 +02001149 s->si[0].exp = TICK_ETERNITY;
1150 s->si[0].flags = SI_FL_NONE;
1151 if (s->fe->options2 & PR_O2_INDEPSTR)
1152 s->si[0].flags |= SI_FL_INDEP_STR;
Emeric Brun2b920a12010-09-23 18:30:22 +02001153
Willy Tarreaub24281b2011-02-13 13:16:36 +01001154 stream_int_register_handler(&s->si[0], &peer_applet);
Willy Tarreaufa6bac62012-05-31 14:16:59 +02001155 s->si[0].applet.st0 = PEER_SESSION_CONNECT;
Willy Tarreauf2943dc2012-10-26 20:10:28 +02001156 s->si[0].conn->xprt_ctx = (void *)ps;
Emeric Brun2b920a12010-09-23 18:30:22 +02001157
Willy Tarreauf2943dc2012-10-26 20:10:28 +02001158 s->si[1].conn->t.sock.fd = -1; /* just to help with debugging */
1159 s->si[1].conn->flags = CO_FL_NONE;
Willy Tarreau14cba4b2012-11-30 17:33:05 +01001160 s->si[1].conn->err_code = CO_ER_NONE;
Emeric Brun2b920a12010-09-23 18:30:22 +02001161 s->si[1].owner = t;
1162 s->si[1].state = s->si[1].prev_state = SI_ST_ASS;
1163 s->si[1].conn_retries = p->conn_retries;
1164 s->si[1].err_type = SI_ET_NONE;
1165 s->si[1].err_loc = NULL;
Willy Tarreau26d8c592012-05-07 18:12:14 +02001166 s->si[1].release = NULL;
Willy Tarreau63e7fe32012-05-08 15:20:43 +02001167 s->si[1].send_proxy_ofs = 0;
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001168 s->si[1].conn->target = &s->be->obj_type;
Willy Tarreauf7bc57c2012-10-03 00:19:48 +02001169 si_prepare_conn(&s->si[1], peer->proto, peer->xprt);
Emeric Brun2b920a12010-09-23 18:30:22 +02001170 s->si[1].exp = TICK_ETERNITY;
1171 s->si[1].flags = SI_FL_NONE;
1172 if (s->be->options2 & PR_O2_INDEPSTR)
1173 s->si[1].flags |= SI_FL_INDEP_STR;
1174
Willy Tarreau9bd0d742011-07-20 00:17:39 +02001175 session_init_srv_conn(s);
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001176 s->target = &s->be->obj_type;
Emeric Brun2b920a12010-09-23 18:30:22 +02001177 s->pend_pos = NULL;
1178
1179 /* init store persistence */
1180 s->store_count = 0;
Willy Tarreaud5ca9ab2013-05-28 17:40:25 +02001181 memset(s->stkctr, 0, sizeof(s->stkctr));
Emeric Brun2b920a12010-09-23 18:30:22 +02001182
1183 /* FIXME: the logs are horribly complicated now, because they are
Willy Tarreauae727bf2013-10-01 17:06:10 +02001184 * defined in <p>, <p>, and later <be> and <be>. We still initialize
1185 * a few of them to help troubleshooting (eg: show sess shows them).
Emeric Brun2b920a12010-09-23 18:30:22 +02001186 */
1187
1188 s->logs.logwait = 0;
Willy Tarreauabcd5142013-06-11 17:18:02 +02001189 s->logs.level = 0;
Willy Tarreauae727bf2013-10-01 17:06:10 +02001190 s->logs.accept_date = date; /* user-visible date for logging */
1191 s->logs.tv_accept = now; /* corrected date for internal use */
Emeric Brun2b920a12010-09-23 18:30:22 +02001192 s->do_log = NULL;
1193
1194 /* default error reporting function, may be changed by analysers */
1195 s->srv_error = default_srv_error;
1196
Emeric Brun2b920a12010-09-23 18:30:22 +02001197 s->uniq_id = 0;
Willy Tarreaubd833142012-05-08 15:51:44 +02001198 s->unique_id = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001199
1200 txn = &s->txn;
1201 /* Those variables will be checked and freed if non-NULL in
1202 * session.c:session_free(). It is important that they are
1203 * properly initialized.
1204 */
1205 txn->sessid = NULL;
1206 txn->srv_cookie = NULL;
1207 txn->cli_cookie = NULL;
1208 txn->uri = NULL;
1209 txn->req.cap = NULL;
1210 txn->rsp.cap = NULL;
1211 txn->hdr_idx.v = NULL;
1212 txn->hdr_idx.size = txn->hdr_idx.used = 0;
1213
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001214 if ((s->req = pool_alloc2(pool2_channel)) == NULL)
Emeric Brun2b920a12010-09-23 18:30:22 +02001215 goto out_fail_req; /* no memory */
1216
Willy Tarreau9b28e032012-10-12 23:49:43 +02001217 if ((s->req->buf = pool_alloc2(pool2_buffer)) == NULL)
1218 goto out_fail_req_buf; /* no memory */
1219
Willy Tarreau19d14ef2012-10-29 16:51:55 +01001220 s->req->buf->size = trash.size;
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001221 channel_init(s->req);
Emeric Brun2b920a12010-09-23 18:30:22 +02001222 s->req->prod = &s->si[0];
1223 s->req->cons = &s->si[1];
1224 s->si[0].ib = s->si[1].ob = s->req;
1225
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001226 s->req->flags |= CF_READ_ATTACHED; /* the producer is already connected */
Emeric Brun2b920a12010-09-23 18:30:22 +02001227
1228 /* activate default analysers enabled for this listener */
1229 s->req->analysers = l->analysers;
1230
1231 /* note: this should not happen anymore since there's always at least the switching rules */
1232 if (!s->req->analysers) {
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001233 channel_auto_connect(s->req);/* don't wait to establish connection */
1234 channel_auto_close(s->req);/* let the producer forward close requests */
Emeric Brun2b920a12010-09-23 18:30:22 +02001235 }
1236
1237 s->req->rto = s->fe->timeout.client;
1238 s->req->wto = s->be->timeout.server;
1239
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001240 if ((s->rep = pool_alloc2(pool2_channel)) == NULL)
Emeric Brun2b920a12010-09-23 18:30:22 +02001241 goto out_fail_rep; /* no memory */
1242
Willy Tarreau9b28e032012-10-12 23:49:43 +02001243 if ((s->rep->buf = pool_alloc2(pool2_buffer)) == NULL)
1244 goto out_fail_rep_buf; /* no memory */
1245
Willy Tarreau19d14ef2012-10-29 16:51:55 +01001246 s->rep->buf->size = trash.size;
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001247 channel_init(s->rep);
Emeric Brun2b920a12010-09-23 18:30:22 +02001248 s->rep->prod = &s->si[1];
1249 s->rep->cons = &s->si[0];
1250 s->si[0].ob = s->si[1].ib = s->rep;
1251
1252 s->rep->rto = s->be->timeout.server;
1253 s->rep->wto = s->fe->timeout.client;
1254
1255 s->req->rex = TICK_ETERNITY;
1256 s->req->wex = TICK_ETERNITY;
1257 s->req->analyse_exp = TICK_ETERNITY;
1258 s->rep->rex = TICK_ETERNITY;
1259 s->rep->wex = TICK_ETERNITY;
1260 s->rep->analyse_exp = TICK_ETERNITY;
1261 t->expire = TICK_ETERNITY;
1262
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001263 s->rep->flags |= CF_READ_DONTWAIT;
Emeric Brun2b920a12010-09-23 18:30:22 +02001264 /* it is important not to call the wakeup function directly but to
1265 * pass through task_wakeup(), because this one knows how to apply
1266 * priorities to tasks.
1267 */
1268 task_wakeup(t, TASK_WOKEN_INIT);
1269
1270 l->nbconn++; /* warning! right now, it's up to the handler to decrease this */
1271 p->feconn++;/* beconn will be increased later */
1272 jobs++;
Willy Tarreau3c63fd82011-09-07 18:00:47 +02001273 if (!(s->listener->options & LI_O_UNLIMITED))
1274 actconn++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001275 totalconn++;
1276
1277 return s;
1278
1279 /* Error unrolling */
Willy Tarreau9b28e032012-10-12 23:49:43 +02001280 out_fail_rep_buf:
1281 pool_free2(pool2_channel, s->rep);
Emeric Brun2b920a12010-09-23 18:30:22 +02001282 out_fail_rep:
Willy Tarreau9b28e032012-10-12 23:49:43 +02001283 pool_free2(pool2_buffer, s->req->buf);
1284 out_fail_req_buf:
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001285 pool_free2(pool2_channel, s->req);
Emeric Brun2b920a12010-09-23 18:30:22 +02001286 out_fail_req:
1287 task_free(t);
1288 out_free_session:
1289 LIST_DEL(&s->list);
Willy Tarreauf2943dc2012-10-26 20:10:28 +02001290 pool_free2(pool2_connection, s->si[1].conn);
1291 out_fail_conn1:
1292 pool_free2(pool2_connection, s->si[0].conn);
1293 out_fail_conn0:
Emeric Brun2b920a12010-09-23 18:30:22 +02001294 pool_free2(pool2_session, s);
1295 out_close:
1296 return s;
1297}
1298
1299/*
1300 * Task processing function to manage re-connect and peer session
1301 * tasks wakeup on local update.
1302 */
Simon Horman96553772011-06-08 09:18:51 +09001303static struct task *process_peer_sync(struct task * task)
Emeric Brun2b920a12010-09-23 18:30:22 +02001304{
1305 struct shared_table *st = (struct shared_table *)task->context;
1306 struct peer_session *ps;
1307
1308 task->expire = TICK_ETERNITY;
1309
1310 if (!stopping) {
1311 /* Normal case (not soft stop)*/
1312 if (((st->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMLOCAL) &&
1313 (!nb_oldpids || tick_is_expired(st->resync_timeout, now_ms)) &&
1314 !(st->flags & SHTABLE_F_RESYNC_ASSIGN)) {
1315 /* Resync from local peer needed
1316 no peer was assigned for the lesson
1317 and no old local peer found
1318 or resync timeout expire */
1319
1320 /* flag no more resync from local, to try resync from remotes */
1321 st->flags |= SHTABLE_F_RESYNC_LOCAL;
1322
1323 /* reschedule a resync */
1324 st->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
1325 }
1326
1327 /* For each session */
1328 for (ps = st->sessions; ps; ps = ps->next) {
1329 /* For each remote peers */
1330 if (!ps->peer->local) {
1331 if (!ps->session) {
1332 /* no active session */
1333 if (ps->statuscode == 0 ||
1334 ps->statuscode == PEER_SESSION_SUCCESSCODE ||
1335 ((ps->statuscode == PEER_SESSION_CONNECTCODE ||
1336 ps->statuscode == PEER_SESSION_CONNECTEDCODE) &&
1337 tick_is_expired(ps->reconnect, now_ms))) {
1338 /* connection never tried
1339 * or previous session established with success
1340 * or previous session failed during connection
1341 * and reconnection timer is expired */
1342
1343 /* retry a connect */
1344 ps->session = peer_session_create(ps->peer, ps);
1345 }
1346 else if (ps->statuscode == PEER_SESSION_CONNECTCODE ||
1347 ps->statuscode == PEER_SESSION_CONNECTEDCODE) {
1348 /* If previous session failed during connection
1349 * but reconnection timer is not expired */
1350
1351 /* reschedule task for reconnect */
1352 task->expire = tick_first(task->expire, ps->reconnect);
1353 }
1354 /* else do nothing */
1355 } /* !ps->session */
1356 else if (ps->statuscode == PEER_SESSION_SUCCESSCODE) {
1357 /* current session is active and established */
1358 if (((st->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMREMOTE) &&
1359 !(st->flags & SHTABLE_F_RESYNC_ASSIGN) &&
1360 !(ps->flags & PEER_F_LEARN_NOTUP2DATE)) {
1361 /* Resync from a remote is needed
1362 * and no peer was assigned for lesson
1363 * and current peer may be up2date */
1364
1365 /* assign peer for the lesson */
1366 ps->flags |= PEER_F_LEARN_ASSIGN;
1367 st->flags |= SHTABLE_F_RESYNC_ASSIGN;
1368
1369 /* awake peer session task to handle a request of resync */
1370 task_wakeup(ps->session->task, TASK_WOKEN_MSG);
1371 }
1372 else if ((int)(ps->pushed - ps->table->table->localupdate) < 0) {
1373 /* awake peer session task to push local updates */
1374 task_wakeup(ps->session->task, TASK_WOKEN_MSG);
1375 }
1376 /* else do nothing */
1377 } /* SUCCESSCODE */
1378 } /* !ps->peer->local */
1379 } /* for */
1380
1381 /* Resync from remotes expired: consider resync is finished */
1382 if (((st->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMREMOTE) &&
1383 !(st->flags & SHTABLE_F_RESYNC_ASSIGN) &&
1384 tick_is_expired(st->resync_timeout, now_ms)) {
1385 /* Resync from remote peer needed
1386 * no peer was assigned for the lesson
1387 * and resync timeout expire */
1388
1389 /* flag no more resync from remote, consider resync is finished */
1390 st->flags |= SHTABLE_F_RESYNC_REMOTE;
1391 }
1392
1393 if ((st->flags & SHTABLE_RESYNC_STATEMASK) != SHTABLE_RESYNC_FINISHED) {
1394 /* Resync not finished*/
1395 /* reschedule task to resync timeout, to ended resync if needed */
1396 task->expire = tick_first(task->expire, st->resync_timeout);
1397 }
1398 } /* !stopping */
1399 else {
1400 /* soft stop case */
1401 if (task->state & TASK_WOKEN_SIGNAL) {
1402 /* We've just recieved the signal */
1403 if (!(st->flags & SHTABLE_F_DONOTSTOP)) {
1404 /* add DO NOT STOP flag if not present */
1405 jobs++;
1406 st->flags |= SHTABLE_F_DONOTSTOP;
Willy Tarreau3a925c12013-09-04 17:54:01 +02001407 st->table->syncing++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001408 }
1409
1410 /* disconnect all connected peers */
1411 for (ps = st->sessions; ps; ps = ps->next) {
1412 if (ps->session) {
1413 peer_session_forceshutdown(ps->session);
1414 ps->session = NULL;
1415 }
1416 }
1417 }
1418 ps = st->local_session;
1419
1420 if (ps->flags & PEER_F_TEACH_COMPLETE) {
1421 if (st->flags & SHTABLE_F_DONOTSTOP) {
1422 /* resync of new process was complete, current process can die now */
1423 jobs--;
1424 st->flags &= ~SHTABLE_F_DONOTSTOP;
Willy Tarreau3a925c12013-09-04 17:54:01 +02001425 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02001426 }
1427 }
1428 else if (!ps->session) {
1429 /* If session is not active */
1430 if (ps->statuscode == 0 ||
1431 ps->statuscode == PEER_SESSION_SUCCESSCODE ||
1432 ps->statuscode == PEER_SESSION_CONNECTEDCODE ||
1433 ps->statuscode == PEER_SESSION_TRYAGAIN) {
1434 /* connection never tried
1435 * or previous session was successfully established
1436 * or previous session tcp connect success but init state incomplete
1437 * or during previous connect, peer replies a try again statuscode */
1438
1439 /* connect to the peer */
1440 ps->session = peer_session_create(ps->peer, ps);
1441 }
1442 else {
1443 /* Other error cases */
1444 if (st->flags & SHTABLE_F_DONOTSTOP) {
1445 /* unable to resync new process, current process can die now */
1446 jobs--;
1447 st->flags &= ~SHTABLE_F_DONOTSTOP;
Willy Tarreau3a925c12013-09-04 17:54:01 +02001448 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02001449 }
1450 }
1451 }
1452 else if (ps->statuscode == PEER_SESSION_SUCCESSCODE &&
1453 (int)(ps->pushed - ps->table->table->localupdate) < 0) {
1454 /* current session active and established
1455 awake session to push remaining local updates */
1456 task_wakeup(ps->session->task, TASK_WOKEN_MSG);
1457 }
1458 } /* stopping */
1459 /* Wakeup for re-connect */
1460 return task;
1461}
1462
1463/*
1464 * Function used to register a table for sync on a group of peers
1465 *
1466 */
1467void peers_register_table(struct peers *peers, struct stktable *table)
1468{
1469 struct shared_table *st;
1470 struct peer * curpeer;
1471 struct peer_session *ps;
Willy Tarreau4348fad2012-09-20 16:48:07 +02001472 struct listener *listener;
Emeric Brun2b920a12010-09-23 18:30:22 +02001473
1474 st = (struct shared_table *)calloc(1,sizeof(struct shared_table));
1475 st->table = table;
1476 st->next = peers->tables;
1477 st->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
1478 peers->tables = st;
1479
1480 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
1481 ps = (struct peer_session *)calloc(1,sizeof(struct peer_session));
1482 ps->table = st;
1483 ps->peer = curpeer;
1484 if (curpeer->local)
1485 st->local_session = ps;
1486 ps->next = st->sessions;
1487 ps->reconnect = now_ms;
1488 st->sessions = ps;
1489 peers->peers_fe->maxconn += 3;
1490 }
1491
Willy Tarreau4348fad2012-09-20 16:48:07 +02001492 list_for_each_entry(listener, &peers->peers_fe->conf.listeners, by_fe)
1493 listener->maxconn = peers->peers_fe->maxconn;
Emeric Brun2b920a12010-09-23 18:30:22 +02001494 st->sync_task = task_new();
1495 st->sync_task->process = process_peer_sync;
1496 st->sync_task->expire = TICK_ETERNITY;
1497 st->sync_task->context = (void *)st;
1498 table->sync_task =st->sync_task;
1499 signal_register_task(0, table->sync_task, 0);
1500 task_wakeup(st->sync_task, TASK_WOKEN_INIT);
1501}
1502