blob: 2c8d9cc6c0b3f70db1a625bb55ce80317b361179 [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) {
Willy Tarreau86a446e2013-11-25 23:02:37 +0100393 /* Check key size mismatches, except for strings
394 * which may be truncated as long as they fit in
395 * a buffer.
396 */
397 if (key_size != st->table->key_size &&
398 (key_type != STKTABLE_TYPE_STRING ||
399 1 + 4 + 4 + key_size - 1 >= trash.size)) {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100400 si->applet.st0 = PEER_SESSION_EXIT;
401 si->applet.st1 = PEER_SESSION_ERRSIZE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200402 goto switchstate;
403 }
404
405 /* If key type mismatches */
406 if (key_type != st->table->type) {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100407 si->applet.st0 = PEER_SESSION_EXIT;
408 si->applet.st1 = PEER_SESSION_ERRTYPE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200409 goto switchstate;
410 }
411
412 /* lookup peer session of current peer */
413 for (ps = st->sessions; ps; ps = ps->next) {
414 if (ps->peer == curpeer) {
415 /* If session already active, replaced by new one */
416 if (ps->session && ps->session != s) {
417 if (ps->peer->local) {
418 /* Local connection, reply a retry */
Willy Tarreaubc4af052011-02-13 13:25:14 +0100419 si->applet.st0 = PEER_SESSION_EXIT;
420 si->applet.st1 = PEER_SESSION_TRYAGAIN;
Emeric Brun2b920a12010-09-23 18:30:22 +0200421 goto switchstate;
422 }
423 peer_session_forceshutdown(ps->session);
424 }
425 ps->session = s;
426 break;
427 }
428 }
429 break;
430 }
431 }
432
433 /* If table not found */
434 if (!st){
Willy Tarreaubc4af052011-02-13 13:25:14 +0100435 si->applet.st0 = PEER_SESSION_EXIT;
436 si->applet.st1 = PEER_SESSION_ERRTABLE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200437 goto switchstate;
438 }
439
440 /* If no peer session for current peer */
441 if (!ps) {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100442 si->applet.st0 = PEER_SESSION_EXIT;
443 si->applet.st1 = PEER_SESSION_ERRPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200444 goto switchstate;
445 }
446
Willy Tarreauf2943dc2012-10-26 20:10:28 +0200447 si->conn->xprt_ctx = ps;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100448 si->applet.st0 = PEER_SESSION_SENDSUCCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200449 /* fall through */
450 }
451 case PEER_SESSION_SENDSUCCESS:{
Willy Tarreauf2943dc2012-10-26 20:10:28 +0200452 struct peer_session *ps = (struct peer_session *)si->conn->xprt_ctx;
Emeric Brun2b920a12010-09-23 18:30:22 +0200453
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100454 repl = snprintf(trash.str, trash.size, "%d\n", PEER_SESSION_SUCCESSCODE);
455 repl = bi_putblk(si->ib, trash.str, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200456 if (repl <= 0) {
457 if (repl == -1)
458 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100459 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200460 goto switchstate;
461 }
462
463 /* Register status code */
464 ps->statuscode = PEER_SESSION_SUCCESSCODE;
465
466 /* Awake main task */
467 task_wakeup(ps->table->sync_task, TASK_WOKEN_MSG);
468
469 /* Init cursors */
470 ps->teaching_origin =ps->lastpush = ps->lastack = ps->pushack = 0;
471 ps->pushed = ps->update;
472
473 /* Init confirm counter */
474 ps->confirm = 0;
475
476 /* reset teaching and learning flags to 0 */
477 ps->flags &= PEER_TEACH_RESET;
478 ps->flags &= PEER_LEARN_RESET;
479
480 /* if current peer is local */
481 if (ps->peer->local) {
482 /* if table need resyncfrom local and no process assined */
483 if ((ps->table->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMLOCAL &&
484 !(ps->table->flags & SHTABLE_F_RESYNC_ASSIGN)) {
485 /* assign local peer for a lesson, consider lesson already requested */
486 ps->flags |= PEER_F_LEARN_ASSIGN;
487 ps->table->flags |= (SHTABLE_F_RESYNC_ASSIGN|SHTABLE_F_RESYNC_PROCESS);
488 }
489
490 }
491 else if ((ps->table->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMREMOTE &&
492 !(ps->table->flags & SHTABLE_F_RESYNC_ASSIGN)) {
493 /* assign peer for a lesson */
494 ps->flags |= PEER_F_LEARN_ASSIGN;
495 ps->table->flags |= SHTABLE_F_RESYNC_ASSIGN;
496 }
497 /* switch to waiting message state */
Willy Tarreaubc4af052011-02-13 13:25:14 +0100498 si->applet.st0 = PEER_SESSION_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200499 goto switchstate;
500 }
501 case PEER_SESSION_CONNECT: {
Willy Tarreauf2943dc2012-10-26 20:10:28 +0200502 struct peer_session *ps = (struct peer_session *)si->conn->xprt_ctx;
Emeric Brun2b920a12010-09-23 18:30:22 +0200503
504 /* Send headers */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100505 repl = snprintf(trash.str, trash.size,
Emeric Brun2b920a12010-09-23 18:30:22 +0200506 PEER_SESSION_PROTO_NAME " 1.0\n%s\n%s %d\n%s %lu %d\n",
507 ps->peer->id,
508 localpeer,
Willy Tarreau7b77c9f2012-01-07 22:52:12 +0100509 (int)getpid(),
Emeric Brun2b920a12010-09-23 18:30:22 +0200510 ps->table->table->id,
511 ps->table->table->type,
Willy Tarreaubd55e312010-11-11 10:55:09 +0100512 (int)ps->table->table->key_size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200513
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100514 if (repl >= trash.size) {
Willy Tarreaubc4af052011-02-13 13:25:14 +0100515 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200516 goto switchstate;
517 }
518
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100519 repl = bi_putblk(si->ib, trash.str, repl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200520 if (repl <= 0) {
521 if (repl == -1)
522 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100523 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200524 goto switchstate;
525 }
526
527 /* switch to the waiting statuscode state */
Willy Tarreaubc4af052011-02-13 13:25:14 +0100528 si->applet.st0 = PEER_SESSION_GETSTATUS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200529 /* fall through */
530 }
531 case PEER_SESSION_GETSTATUS: {
Willy Tarreauf2943dc2012-10-26 20:10:28 +0200532 struct peer_session *ps = (struct peer_session *)si->conn->xprt_ctx;
Emeric Brun2b920a12010-09-23 18:30:22 +0200533
Willy Tarreau03cdb7c2012-08-27 23:14:58 +0200534 if (si->ib->flags & CF_WRITE_PARTIAL)
Emeric Brun2b920a12010-09-23 18:30:22 +0200535 ps->statuscode = PEER_SESSION_CONNECTEDCODE;
536
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100537 reql = bo_getline(si->ob, trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200538 if (reql <= 0) { /* closed or EOL not found */
539 if (reql == 0)
540 goto out;
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 if (trash.str[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200545 /* Incomplete line, we quit */
Willy Tarreaubc4af052011-02-13 13:25:14 +0100546 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200547 goto switchstate;
548 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100549 else if (reql > 1 && (trash.str[reql-2] == '\r'))
550 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200551 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100552 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200553
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200554 bo_skip(si->ob, reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200555
556 /* Register status code */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100557 ps->statuscode = atoi(trash.str);
Emeric Brun2b920a12010-09-23 18:30:22 +0200558
559 /* Awake main task */
560 task_wakeup(ps->table->sync_task, TASK_WOKEN_MSG);
561
562 /* If status code is success */
563 if (ps->statuscode == PEER_SESSION_SUCCESSCODE) {
564 /* Init cursors */
565 ps->teaching_origin = ps->lastpush = ps->lastack = ps->pushack = 0;
566 ps->pushed = ps->update;
567
568 /* Init confirm counter */
569 ps->confirm = 0;
570
571 /* reset teaching and learning flags to 0 */
572 ps->flags &= PEER_TEACH_RESET;
573 ps->flags &= PEER_LEARN_RESET;
574
575 /* If current peer is local */
576 if (ps->peer->local) {
577 /* Init cursors to push a resync */
578 ps->teaching_origin = ps->pushed = ps->table->table->update;
579 /* flag to start to teach lesson */
580 ps->flags |= PEER_F_TEACH_PROCESS;
581
582 }
583 else if ((ps->table->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMREMOTE &&
584 !(ps->table->flags & SHTABLE_F_RESYNC_ASSIGN)) {
585 /* If peer is remote and resync from remote is needed,
586 and no peer currently assigned */
587
588 /* assign peer for a lesson */
589 ps->flags |= PEER_F_LEARN_ASSIGN;
590 ps->table->flags |= SHTABLE_F_RESYNC_ASSIGN;
591 }
592
593 }
594 else {
595 /* Status code is not success, abort */
Willy Tarreaubc4af052011-02-13 13:25:14 +0100596 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200597 goto switchstate;
598 }
Willy Tarreaubc4af052011-02-13 13:25:14 +0100599 si->applet.st0 = PEER_SESSION_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200600 /* fall through */
601 }
602 case PEER_SESSION_WAITMSG: {
Willy Tarreauf2943dc2012-10-26 20:10:28 +0200603 struct peer_session *ps = (struct peer_session *)si->conn->xprt_ctx;
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200604 struct stksess *ts, *newts = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +0200605 char c;
606 int totl = 0;
607
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200608 reql = bo_getblk(si->ob, (char *)&c, sizeof(c), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200609 if (reql <= 0) /* closed or EOL not found */
610 goto incomplete;
611
Emeric Brun2b920a12010-09-23 18:30:22 +0200612 totl += reql;
613
614 if ((c & 0x80) || (c == 'D')) {
615 /* Here we have data message */
616 unsigned int pushack;
Emeric Brun2b920a12010-09-23 18:30:22 +0200617 int srvid;
618 uint32_t netinteger;
619
620 /* Compute update remote version */
621 if (c & 0x80) {
622 pushack = ps->pushack + (unsigned int)(c & 0x7F);
623 }
624 else {
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200625 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200626 if (reql <= 0) /* closed or EOL not found */
627 goto incomplete;
628
Emeric Brun2b920a12010-09-23 18:30:22 +0200629 totl += reql;
630 pushack = ntohl(netinteger);
631 }
632
Willy Tarreau86a446e2013-11-25 23:02:37 +0100633 /* Read key. The string keys are read in two steps, the first step
634 * consists in reading whatever fits into the table directly into
635 * the pre-allocated key. The second step consists in simply
636 * draining all exceeding data. This can happen for example after a
637 * config reload with a smaller key size for the stick table than
638 * what was previously set, or when facing the impossibility to
639 * allocate a new stksess (for example when the table is full with
640 * "nopurge").
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200641 */
Emeric Brun2b920a12010-09-23 18:30:22 +0200642 if (ps->table->table->type == STKTABLE_TYPE_STRING) {
Willy Tarreau86a446e2013-11-25 23:02:37 +0100643 unsigned int to_read, to_store;
644
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200645 /* read size first */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200646 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200647 if (reql <= 0) /* closed or EOL not found */
648 goto incomplete;
649
Emeric Brun2b920a12010-09-23 18:30:22 +0200650 totl += reql;
Willy Tarreau86a446e2013-11-25 23:02:37 +0100651
652 to_store = 0;
653 to_read = ntohl(netinteger);
654
655 if (to_read + totl > si->ob->buf->size) {
656 /* impossible to read a key this large, abort */
657 reql = -1;
Willy Tarreau72d6c162013-04-11 16:14:13 +0200658 goto incomplete;
Willy Tarreau86a446e2013-11-25 23:02:37 +0100659 }
Willy Tarreau72d6c162013-04-11 16:14:13 +0200660
Willy Tarreau86a446e2013-11-25 23:02:37 +0100661 newts = stksess_new(ps->table->table, NULL);
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200662 if (newts)
Willy Tarreau86a446e2013-11-25 23:02:37 +0100663 to_store = MIN(to_read, ps->table->table->key_size - 1);
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200664
Willy Tarreau86a446e2013-11-25 23:02:37 +0100665 /* we read up to two blocks, the first one goes into the key,
666 * the rest is drained into the trash.
667 */
668 if (to_store) {
669 reql = bo_getblk(si->ob, (char *)newts->key.key, to_store, totl);
670 if (reql <= 0) /* closed or incomplete */
671 goto incomplete;
672 newts->key.key[reql] = 0;
673 totl += reql;
674 to_read -= reql;
675 }
676 if (to_read) {
677 reql = bo_getblk(si->ob, trash.str, to_read, totl);
678 if (reql <= 0) /* closed or incomplete */
679 goto incomplete;
680 totl += reql;
681 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200682 }
683 else if (ps->table->table->type == STKTABLE_TYPE_INTEGER) {
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200684 newts = stksess_new(ps->table->table, NULL);
685 reql = bo_getblk(si->ob, newts ? (char *)newts->key.key : trash.str, sizeof(netinteger), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200686 if (reql <= 0) /* closed or EOL not found */
687 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +0200688 totl += reql;
Emeric Brun2b920a12010-09-23 18:30:22 +0200689 }
690 else {
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200691 /* type ip or binary */
692 newts = stksess_new(ps->table->table, NULL);
693 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 +0200694 if (reql <= 0) /* closed or EOL not found */
695 goto incomplete;
Willy Tarreau72d6c162013-04-11 16:14:13 +0200696 totl += reql;
Emeric Brun2b920a12010-09-23 18:30:22 +0200697 }
698
699 /* read server id */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200700 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200701 if (reql <= 0) /* closed or EOL not found */
702 goto incomplete;
703
Emeric Brun2b920a12010-09-23 18:30:22 +0200704 totl += reql;
705 srvid = ntohl(netinteger);
706
707 /* update entry */
Emeric Brun2b920a12010-09-23 18:30:22 +0200708 if (newts) {
709 /* lookup for existing entry */
710 ts = stktable_lookup(ps->table->table, newts);
711 if (ts) {
712 /* the entry already exist, we can free ours */
713 stktable_touch(ps->table->table, ts, 0);
714 stksess_free(ps->table->table, newts);
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200715 newts = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +0200716 }
717 else {
718 struct eb32_node *eb;
719
720 /* create new entry */
721 ts = stktable_store(ps->table->table, newts, 0);
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200722 newts = NULL; /* don't reuse it */
723
Emeric Brun2b920a12010-09-23 18:30:22 +0200724 ts->upd.key= (++ps->table->table->update)+(2^31);
725 eb = eb32_insert(&ps->table->table->updates, &ts->upd);
726 if (eb != &ts->upd) {
727 eb32_delete(eb);
728 eb32_insert(&ps->table->table->updates, &ts->upd);
729 }
730 }
731
732 /* update entry */
733 if (srvid && stktable_data_ptr(ps->table->table, ts, STKTABLE_DT_SERVER_ID))
734 stktable_data_cast(stktable_data_ptr(ps->table->table, ts, STKTABLE_DT_SERVER_ID), server_id) = srvid;
735 ps->pushack = pushack;
736 }
737
738 }
739 else if (c == 'R') {
740 /* Reset message: remote need resync */
741
742 /* reinit counters for a resync */
743 ps->lastpush = 0;
744 ps->teaching_origin = ps->pushed = ps->table->table->update;
745
746 /* reset teaching flags to 0 */
747 ps->flags &= PEER_TEACH_RESET;
748
749 /* flag to start to teach lesson */
750 ps->flags |= PEER_F_TEACH_PROCESS;
751 }
752 else if (c == 'F') {
753 /* Finish message, all known updates have been pushed by remote */
754 /* and remote is up to date */
755
756 /* If resync is in progress with remote peer */
757 if (ps->flags & PEER_F_LEARN_ASSIGN) {
758
759 /* unassign current peer for learning */
760 ps->flags &= ~PEER_F_LEARN_ASSIGN;
761 ps->table->flags &= ~(SHTABLE_F_RESYNC_ASSIGN|SHTABLE_F_RESYNC_PROCESS);
762
763 /* Consider table is now up2date, resync resync no more needed from local neither remote */
764 ps->table->flags |= (SHTABLE_F_RESYNC_LOCAL|SHTABLE_F_RESYNC_REMOTE);
765 }
766 /* Increase confirm counter to launch a confirm message */
767 ps->confirm++;
768 }
769 else if (c == 'c') {
770 /* confirm message, remote peer is now up to date with us */
771
772 /* If stopping state */
773 if (stopping) {
774 /* Close session, push resync no more needed */
775 ps->flags |= PEER_F_TEACH_COMPLETE;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100776 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200777 goto switchstate;
778 }
779
780 /* reset teaching flags to 0 */
781 ps->flags &= PEER_TEACH_RESET;
782 }
783 else if (c == 'C') {
784 /* Continue message, all known updates have been pushed by remote */
785 /* but remote is not up to date */
786
787 /* If resync is in progress with current peer */
788 if (ps->flags & PEER_F_LEARN_ASSIGN) {
789
790 /* unassign current peer */
791 ps->flags &= ~PEER_F_LEARN_ASSIGN;
792 ps->table->flags &= ~(SHTABLE_F_RESYNC_ASSIGN|SHTABLE_F_RESYNC_PROCESS);
793
794 /* flag current peer is not up 2 date to try from an other */
795 ps->flags |= PEER_F_LEARN_NOTUP2DATE;
796
797 /* reschedule a resync */
798 ps->table->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
799 task_wakeup(ps->table->sync_task, TASK_WOKEN_MSG);
800 }
801 ps->confirm++;
802 }
803 else if (c == 'A') {
804 /* ack message */
805 uint32_t netinteger;
806
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200807 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200808 if (reql <= 0) /* closed or EOL not found */
809 goto incomplete;
810
Emeric Brun2b920a12010-09-23 18:30:22 +0200811 totl += reql;
812
813 /* Consider remote is up to date with "acked" version */
814 ps->update = ntohl(netinteger);
815 }
816 else {
817 /* Unknown message */
Willy Tarreaubc4af052011-02-13 13:25:14 +0100818 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200819 goto switchstate;
820 }
821
822 /* skip consumed message */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200823 bo_skip(si->ob, totl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200824
825 /* loop on that state to peek next message */
Willy Tarreau72d6c162013-04-11 16:14:13 +0200826 goto switchstate;
827
Emeric Brun2b920a12010-09-23 18:30:22 +0200828incomplete:
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200829 /* we get here when a bo_getblk() returns <= 0 in reql */
830
831 /* first, we may have to release newts */
832 if (newts) {
833 stksess_free(ps->table->table, newts);
834 newts = NULL;
835 }
836
Willy Tarreau72d6c162013-04-11 16:14:13 +0200837 if (reql < 0) {
838 /* there was an error */
839 si->applet.st0 = PEER_SESSION_END;
840 goto switchstate;
841 }
842
Emeric Brun2b920a12010-09-23 18:30:22 +0200843 /* Nothing to read, now we start to write */
844
845 /* Confirm finished or partial messages */
846 while (ps->confirm) {
847 /* There is a confirm messages to send */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200848 repl = bi_putchr(si->ib, 'c');
Emeric Brun2b920a12010-09-23 18:30:22 +0200849 if (repl <= 0) {
850 /* no more write possible */
851 if (repl == -1)
852 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100853 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200854 goto switchstate;
855 }
856 ps->confirm--;
857 }
858
859 /* Need to request a resync */
860 if ((ps->flags & PEER_F_LEARN_ASSIGN) &&
861 (ps->table->flags & SHTABLE_F_RESYNC_ASSIGN) &&
862 !(ps->table->flags & SHTABLE_F_RESYNC_PROCESS)) {
863 /* Current peer was elected to request a resync */
864
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200865 repl = bi_putchr(si->ib, 'R');
Emeric Brun2b920a12010-09-23 18:30:22 +0200866 if (repl <= 0) {
867 /* no more write possible */
868 if (repl == -1)
869 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100870 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200871 goto switchstate;
872 }
873 ps->table->flags |= SHTABLE_F_RESYNC_PROCESS;
874 }
875
876 /* It remains some updates to ack */
877 if (ps->pushack != ps->lastack) {
878 uint32_t netinteger;
879
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100880 trash.str[0] = 'A';
Emeric Brun2b920a12010-09-23 18:30:22 +0200881 netinteger = htonl(ps->pushack);
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100882 memcpy(&trash.str[1], &netinteger, sizeof(netinteger));
Emeric Brun2b920a12010-09-23 18:30:22 +0200883
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100884 repl = bi_putblk(si->ib, trash.str, 1+sizeof(netinteger));
Emeric Brun2b920a12010-09-23 18:30:22 +0200885 if (repl <= 0) {
886 /* no more write possible */
887 if (repl == -1)
888 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100889 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200890 goto switchstate;
891 }
892 ps->lastack = ps->pushack;
893 }
894
895 if (ps->flags & PEER_F_TEACH_PROCESS) {
896 /* current peer was requested for a lesson */
897
898 if (!(ps->flags & PEER_F_TEACH_STAGE1)) {
899 /* lesson stage 1 not complete */
900 struct eb32_node *eb;
901
902 eb = eb32_lookup_ge(&ps->table->table->updates, ps->pushed+1);
903 while (1) {
904 int msglen;
905 struct stksess *ts;
906
907 if (!eb) {
908 /* flag lesson stage1 complete */
909 ps->flags |= PEER_F_TEACH_STAGE1;
910 eb = eb32_first(&ps->table->table->updates);
911 if (eb)
912 ps->pushed = eb->key - 1;
913 break;
914 }
915
916 ts = eb32_entry(eb, struct stksess, upd);
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100917 msglen = peer_prepare_datamsg(ts, ps, trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200918 if (msglen) {
919 /* message to buffer */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100920 repl = bi_putblk(si->ib, trash.str, msglen);
Emeric Brun2b920a12010-09-23 18:30:22 +0200921 if (repl <= 0) {
922 /* no more write possible */
923 if (repl == -1)
924 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100925 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200926 goto switchstate;
927 }
928 ps->lastpush = ps->pushed = ts->upd.key;
929 }
930 eb = eb32_next(eb);
931 }
932 } /* !TEACH_STAGE1 */
933
934 if (!(ps->flags & PEER_F_TEACH_STAGE2)) {
935 /* lesson stage 2 not complete */
936 struct eb32_node *eb;
937
938 eb = eb32_lookup_ge(&ps->table->table->updates, ps->pushed+1);
939 while (1) {
940 int msglen;
941 struct stksess *ts;
942
943 if (!eb || eb->key > ps->teaching_origin) {
944 /* flag lesson stage1 complete */
945 ps->flags |= PEER_F_TEACH_STAGE2;
946 ps->pushed = ps->teaching_origin;
947 break;
948 }
949
950 ts = eb32_entry(eb, struct stksess, upd);
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100951 msglen = peer_prepare_datamsg(ts, ps, trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200952 if (msglen) {
953 /* message to buffer */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100954 repl = bi_putblk(si->ib, trash.str, msglen);
Emeric Brun2b920a12010-09-23 18:30:22 +0200955 if (repl <= 0) {
956 /* no more write possible */
957 if (repl == -1)
958 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100959 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200960 goto switchstate;
961 }
962 ps->lastpush = ps->pushed = ts->upd.key;
963 }
964 eb = eb32_next(eb);
965 }
966 } /* !TEACH_STAGE2 */
967
968 if (!(ps->flags & PEER_F_TEACH_FINISHED)) {
969 /* process final lesson message */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200970 repl = bi_putchr(si->ib, ((ps->table->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FINISHED) ? 'F' : 'C');
Emeric Brun2b920a12010-09-23 18:30:22 +0200971 if (repl <= 0) {
972 /* no more write possible */
973 if (repl == -1)
974 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +0100975 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200976 goto switchstate;
977 }
978
979 /* flag finished message sent */
980 ps->flags |= PEER_F_TEACH_FINISHED;
981 } /* !TEACH_FINISHED */
982 } /* TEACH_PROCESS */
983
984 if (!(ps->flags & PEER_F_LEARN_ASSIGN) &&
985 (int)(ps->pushed - ps->table->table->localupdate) < 0) {
986 /* Push local updates, only if no learning in progress (to avoid ping-pong effects) */
987 struct eb32_node *eb;
988
989 eb = eb32_lookup_ge(&ps->table->table->updates, ps->pushed+1);
990 while (1) {
991 int msglen;
992 struct stksess *ts;
993
994 /* push local updates */
995 if (!eb) {
996 eb = eb32_first(&ps->table->table->updates);
997 if (!eb || ((int)(eb->key - ps->pushed) <= 0)) {
998 ps->pushed = ps->table->table->localupdate;
999 break;
1000 }
1001 }
1002
1003 if ((int)(eb->key - ps->table->table->localupdate) > 0) {
1004 ps->pushed = ps->table->table->localupdate;
1005 break;
1006 }
1007
1008 ts = eb32_entry(eb, struct stksess, upd);
Willy Tarreau19d14ef2012-10-29 16:51:55 +01001009 msglen = peer_prepare_datamsg(ts, ps, trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +02001010 if (msglen) {
1011 /* message to buffer */
Willy Tarreau19d14ef2012-10-29 16:51:55 +01001012 repl = bi_putblk(si->ib, trash.str, msglen);
Emeric Brun2b920a12010-09-23 18:30:22 +02001013 if (repl <= 0) {
1014 /* no more write possible */
1015 if (repl == -1)
1016 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +01001017 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02001018 goto switchstate;
1019 }
1020 ps->lastpush = ps->pushed = ts->upd.key;
1021 }
1022 eb = eb32_next(eb);
1023 }
1024 } /* ! LEARN_ASSIGN */
1025 /* noting more to do */
1026 goto out;
1027 }
1028 case PEER_SESSION_EXIT:
Willy Tarreau19d14ef2012-10-29 16:51:55 +01001029 repl = snprintf(trash.str, trash.size, "%d\n", si->applet.st1);
Emeric Brun2b920a12010-09-23 18:30:22 +02001030
Willy Tarreau19d14ef2012-10-29 16:51:55 +01001031 if (bi_putblk(si->ib, trash.str, repl) == -1)
Emeric Brun2b920a12010-09-23 18:30:22 +02001032 goto out;
Willy Tarreaubc4af052011-02-13 13:25:14 +01001033 si->applet.st0 = PEER_SESSION_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02001034 /* fall through */
1035 case PEER_SESSION_END: {
Willy Tarreau73b013b2012-05-21 16:31:45 +02001036 si_shutw(si);
1037 si_shutr(si);
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001038 si->ib->flags |= CF_READ_NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001039 goto quit;
1040 }
1041 }
1042 }
1043out:
Willy Tarreau73b013b2012-05-21 16:31:45 +02001044 si_update(si);
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001045 si->ob->flags |= CF_READ_DONTWAIT;
Emeric Brun2b920a12010-09-23 18:30:22 +02001046 /* we don't want to expire timeouts while we're processing requests */
1047 si->ib->rex = TICK_ETERNITY;
1048 si->ob->wex = TICK_ETERNITY;
1049quit:
1050 return;
1051}
1052
Willy Tarreaub24281b2011-02-13 13:16:36 +01001053static struct si_applet peer_applet = {
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001054 .obj_type = OBJ_TYPE_APPLET,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001055 .name = "<PEER>", /* used for logging */
1056 .fct = peer_io_handler,
Aman Gupta9a13e842012-04-02 18:57:53 -07001057 .release = peer_session_release,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001058};
Emeric Brun2b920a12010-09-23 18:30:22 +02001059
1060/*
1061 * Use this function to force a close of a peer session
1062 */
Simon Horman96553772011-06-08 09:18:51 +09001063static void peer_session_forceshutdown(struct session * session)
Emeric Brun2b920a12010-09-23 18:30:22 +02001064{
1065 struct stream_interface *oldsi;
1066
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001067 if (objt_applet(session->si[0].conn->target) == &peer_applet) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001068 oldsi = &session->si[0];
1069 }
1070 else {
1071 oldsi = &session->si[1];
1072 }
1073
1074 /* call release to reinit resync states if needed */
1075 peer_session_release(oldsi);
Willy Tarreaubc4af052011-02-13 13:25:14 +01001076 oldsi->applet.st0 = PEER_SESSION_END;
Willy Tarreauf2943dc2012-10-26 20:10:28 +02001077 oldsi->conn->xprt_ctx = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001078 task_wakeup(session->task, TASK_WOKEN_MSG);
1079}
1080
1081/*
1082 * this function is called on a read event from a listen socket, corresponding
1083 * to an accept. It tries to accept as many connections as possible.
Willy Tarreaubd55e312010-11-11 10:55:09 +01001084 * It returns a positive value upon success, 0 if the connection needs to be
1085 * closed and ignored, or a negative value upon critical failure.
Emeric Brun2b920a12010-09-23 18:30:22 +02001086 */
1087int peer_accept(struct session *s)
1088{
1089 /* we have a dedicated I/O handler for the stats */
Willy Tarreaub24281b2011-02-13 13:16:36 +01001090 stream_int_register_handler(&s->si[1], &peer_applet);
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001091 s->target = s->si[1].conn->target; // for logging only
Willy Tarreauf2943dc2012-10-26 20:10:28 +02001092 s->si[1].conn->xprt_ctx = s;
Willy Tarreaubc4af052011-02-13 13:25:14 +01001093 s->si[1].applet.st0 = PEER_SESSION_ACCEPT;
Emeric Brun2b920a12010-09-23 18:30:22 +02001094
1095 tv_zero(&s->logs.tv_request);
1096 s->logs.t_queue = 0;
1097 s->logs.t_connect = 0;
1098 s->logs.t_data = 0;
1099 s->logs.t_close = 0;
1100 s->logs.bytes_in = s->logs.bytes_out = 0;
1101 s->logs.prx_queue_size = 0;/* we get the number of pending conns before us */
1102 s->logs.srv_queue_size = 0; /* we will get this number soon */
1103
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001104 s->req->flags |= CF_READ_DONTWAIT; /* we plan to read small requests */
Emeric Brun2b920a12010-09-23 18:30:22 +02001105
1106 if (s->listener->timeout) {
1107 s->req->rto = *s->listener->timeout;
1108 s->rep->wto = *s->listener->timeout;
1109 }
1110 return 1;
1111}
1112
1113/*
Willy Tarreaubd55e312010-11-11 10:55:09 +01001114 * Create a new peer session in assigned state (connect will start automatically)
Emeric Brun2b920a12010-09-23 18:30:22 +02001115 */
Simon Horman96553772011-06-08 09:18:51 +09001116static struct session *peer_session_create(struct peer *peer, struct peer_session *ps)
Emeric Brun2b920a12010-09-23 18:30:22 +02001117{
Willy Tarreau4348fad2012-09-20 16:48:07 +02001118 struct listener *l = LIST_NEXT(&peer->peers->peers_fe->conf.listeners, struct listener *, by_fe);
Emeric Brun2b920a12010-09-23 18:30:22 +02001119 struct proxy *p = (struct proxy *)l->frontend; /* attached frontend */
1120 struct session *s;
1121 struct http_txn *txn;
1122 struct task *t;
1123
1124 if ((s = pool_alloc2(pool2_session)) == NULL) { /* disable this proxy for a while */
Godbach430f2912013-06-20 13:28:38 +08001125 Alert("out of memory in peer_session_create().\n");
Emeric Brun2b920a12010-09-23 18:30:22 +02001126 goto out_close;
1127 }
1128
Willy Tarreauf2943dc2012-10-26 20:10:28 +02001129 if (unlikely((s->si[0].conn = pool_alloc2(pool2_connection)) == NULL))
1130 goto out_fail_conn0;
1131
1132 if (unlikely((s->si[1].conn = pool_alloc2(pool2_connection)) == NULL))
1133 goto out_fail_conn1;
1134
Emeric Brun2b920a12010-09-23 18:30:22 +02001135 LIST_ADDQ(&sessions, &s->list);
1136 LIST_INIT(&s->back_refs);
1137
1138 s->flags = SN_ASSIGNED|SN_ADDR_SET;
Emeric Brun2b920a12010-09-23 18:30:22 +02001139
1140 /* if this session comes from a known monitoring system, we want to ignore
1141 * it as soon as possible, which means closing it immediately for TCP.
1142 */
1143 if ((t = task_new()) == NULL) { /* disable this proxy for a while */
Godbach430f2912013-06-20 13:28:38 +08001144 Alert("out of memory in peer_session_create().\n");
Emeric Brun2b920a12010-09-23 18:30:22 +02001145 goto out_free_session;
1146 }
1147
1148 ps->reconnect = tick_add(now_ms, MS_TO_TICKS(5000));
1149 ps->statuscode = PEER_SESSION_CONNECTCODE;
1150
1151 t->process = l->handler;
1152 t->context = s;
1153 t->nice = l->nice;
1154
Willy Tarreauf2943dc2012-10-26 20:10:28 +02001155 memcpy(&s->si[1].conn->addr.to, &peer->addr, sizeof(s->si[1].conn->addr.to));
Emeric Brun2b920a12010-09-23 18:30:22 +02001156 s->task = t;
1157 s->listener = l;
1158
1159 /* Note: initially, the session's backend points to the frontend.
1160 * This changes later when switching rules are executed or
1161 * when the default backend is assigned.
1162 */
1163 s->be = s->fe = p;
1164
1165 s->req = s->rep = NULL; /* will be allocated later */
1166
Willy Tarreauf2943dc2012-10-26 20:10:28 +02001167 s->si[0].conn->t.sock.fd = -1;
1168 s->si[0].conn->flags = CO_FL_NONE;
Willy Tarreau14cba4b2012-11-30 17:33:05 +01001169 s->si[0].conn->err_code = CO_ER_NONE;
Emeric Brun2b920a12010-09-23 18:30:22 +02001170 s->si[0].owner = t;
1171 s->si[0].state = s->si[0].prev_state = SI_ST_EST;
1172 s->si[0].err_type = SI_ET_NONE;
1173 s->si[0].err_loc = NULL;
Willy Tarreau26d8c592012-05-07 18:12:14 +02001174 s->si[0].release = NULL;
Willy Tarreau63e7fe32012-05-08 15:20:43 +02001175 s->si[0].send_proxy_ofs = 0;
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001176 s->si[0].conn->target = &l->obj_type;
Emeric Brun2b920a12010-09-23 18:30:22 +02001177 s->si[0].exp = TICK_ETERNITY;
1178 s->si[0].flags = SI_FL_NONE;
1179 if (s->fe->options2 & PR_O2_INDEPSTR)
1180 s->si[0].flags |= SI_FL_INDEP_STR;
Emeric Brun2b920a12010-09-23 18:30:22 +02001181
Willy Tarreaub24281b2011-02-13 13:16:36 +01001182 stream_int_register_handler(&s->si[0], &peer_applet);
Willy Tarreaufa6bac62012-05-31 14:16:59 +02001183 s->si[0].applet.st0 = PEER_SESSION_CONNECT;
Willy Tarreauf2943dc2012-10-26 20:10:28 +02001184 s->si[0].conn->xprt_ctx = (void *)ps;
Emeric Brun2b920a12010-09-23 18:30:22 +02001185
Willy Tarreauf2943dc2012-10-26 20:10:28 +02001186 s->si[1].conn->t.sock.fd = -1; /* just to help with debugging */
1187 s->si[1].conn->flags = CO_FL_NONE;
Willy Tarreau14cba4b2012-11-30 17:33:05 +01001188 s->si[1].conn->err_code = CO_ER_NONE;
Emeric Brun2b920a12010-09-23 18:30:22 +02001189 s->si[1].owner = t;
1190 s->si[1].state = s->si[1].prev_state = SI_ST_ASS;
1191 s->si[1].conn_retries = p->conn_retries;
1192 s->si[1].err_type = SI_ET_NONE;
1193 s->si[1].err_loc = NULL;
Willy Tarreau26d8c592012-05-07 18:12:14 +02001194 s->si[1].release = NULL;
Willy Tarreau63e7fe32012-05-08 15:20:43 +02001195 s->si[1].send_proxy_ofs = 0;
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001196 s->si[1].conn->target = &s->be->obj_type;
Willy Tarreauf7bc57c2012-10-03 00:19:48 +02001197 si_prepare_conn(&s->si[1], peer->proto, peer->xprt);
Emeric Brun2b920a12010-09-23 18:30:22 +02001198 s->si[1].exp = TICK_ETERNITY;
1199 s->si[1].flags = SI_FL_NONE;
1200 if (s->be->options2 & PR_O2_INDEPSTR)
1201 s->si[1].flags |= SI_FL_INDEP_STR;
1202
Willy Tarreau9bd0d742011-07-20 00:17:39 +02001203 session_init_srv_conn(s);
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001204 s->target = &s->be->obj_type;
Emeric Brun2b920a12010-09-23 18:30:22 +02001205 s->pend_pos = NULL;
1206
1207 /* init store persistence */
1208 s->store_count = 0;
Willy Tarreaud5ca9ab2013-05-28 17:40:25 +02001209 memset(s->stkctr, 0, sizeof(s->stkctr));
Emeric Brun2b920a12010-09-23 18:30:22 +02001210
1211 /* FIXME: the logs are horribly complicated now, because they are
Willy Tarreauae727bf2013-10-01 17:06:10 +02001212 * defined in <p>, <p>, and later <be> and <be>. We still initialize
1213 * a few of them to help troubleshooting (eg: show sess shows them).
Emeric Brun2b920a12010-09-23 18:30:22 +02001214 */
1215
1216 s->logs.logwait = 0;
Willy Tarreauabcd5142013-06-11 17:18:02 +02001217 s->logs.level = 0;
Willy Tarreauae727bf2013-10-01 17:06:10 +02001218 s->logs.accept_date = date; /* user-visible date for logging */
1219 s->logs.tv_accept = now; /* corrected date for internal use */
Emeric Brun2b920a12010-09-23 18:30:22 +02001220 s->do_log = NULL;
1221
1222 /* default error reporting function, may be changed by analysers */
1223 s->srv_error = default_srv_error;
1224
Emeric Brun2b920a12010-09-23 18:30:22 +02001225 s->uniq_id = 0;
Willy Tarreaubd833142012-05-08 15:51:44 +02001226 s->unique_id = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001227
1228 txn = &s->txn;
1229 /* Those variables will be checked and freed if non-NULL in
1230 * session.c:session_free(). It is important that they are
1231 * properly initialized.
1232 */
1233 txn->sessid = NULL;
1234 txn->srv_cookie = NULL;
1235 txn->cli_cookie = NULL;
1236 txn->uri = NULL;
1237 txn->req.cap = NULL;
1238 txn->rsp.cap = NULL;
1239 txn->hdr_idx.v = NULL;
1240 txn->hdr_idx.size = txn->hdr_idx.used = 0;
1241
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001242 if ((s->req = pool_alloc2(pool2_channel)) == NULL)
Emeric Brun2b920a12010-09-23 18:30:22 +02001243 goto out_fail_req; /* no memory */
1244
Willy Tarreau9b28e032012-10-12 23:49:43 +02001245 if ((s->req->buf = pool_alloc2(pool2_buffer)) == NULL)
1246 goto out_fail_req_buf; /* no memory */
1247
Willy Tarreau19d14ef2012-10-29 16:51:55 +01001248 s->req->buf->size = trash.size;
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001249 channel_init(s->req);
Emeric Brun2b920a12010-09-23 18:30:22 +02001250 s->req->prod = &s->si[0];
1251 s->req->cons = &s->si[1];
1252 s->si[0].ib = s->si[1].ob = s->req;
1253
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001254 s->req->flags |= CF_READ_ATTACHED; /* the producer is already connected */
Emeric Brun2b920a12010-09-23 18:30:22 +02001255
1256 /* activate default analysers enabled for this listener */
1257 s->req->analysers = l->analysers;
1258
1259 /* note: this should not happen anymore since there's always at least the switching rules */
1260 if (!s->req->analysers) {
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001261 channel_auto_connect(s->req);/* don't wait to establish connection */
1262 channel_auto_close(s->req);/* let the producer forward close requests */
Emeric Brun2b920a12010-09-23 18:30:22 +02001263 }
1264
1265 s->req->rto = s->fe->timeout.client;
1266 s->req->wto = s->be->timeout.server;
1267
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001268 if ((s->rep = pool_alloc2(pool2_channel)) == NULL)
Emeric Brun2b920a12010-09-23 18:30:22 +02001269 goto out_fail_rep; /* no memory */
1270
Willy Tarreau9b28e032012-10-12 23:49:43 +02001271 if ((s->rep->buf = pool_alloc2(pool2_buffer)) == NULL)
1272 goto out_fail_rep_buf; /* no memory */
1273
Willy Tarreau19d14ef2012-10-29 16:51:55 +01001274 s->rep->buf->size = trash.size;
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001275 channel_init(s->rep);
Emeric Brun2b920a12010-09-23 18:30:22 +02001276 s->rep->prod = &s->si[1];
1277 s->rep->cons = &s->si[0];
1278 s->si[0].ob = s->si[1].ib = s->rep;
1279
1280 s->rep->rto = s->be->timeout.server;
1281 s->rep->wto = s->fe->timeout.client;
1282
1283 s->req->rex = TICK_ETERNITY;
1284 s->req->wex = TICK_ETERNITY;
1285 s->req->analyse_exp = TICK_ETERNITY;
1286 s->rep->rex = TICK_ETERNITY;
1287 s->rep->wex = TICK_ETERNITY;
1288 s->rep->analyse_exp = TICK_ETERNITY;
1289 t->expire = TICK_ETERNITY;
1290
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001291 s->rep->flags |= CF_READ_DONTWAIT;
Emeric Brun2b920a12010-09-23 18:30:22 +02001292 /* it is important not to call the wakeup function directly but to
1293 * pass through task_wakeup(), because this one knows how to apply
1294 * priorities to tasks.
1295 */
1296 task_wakeup(t, TASK_WOKEN_INIT);
1297
1298 l->nbconn++; /* warning! right now, it's up to the handler to decrease this */
1299 p->feconn++;/* beconn will be increased later */
1300 jobs++;
Willy Tarreau3c63fd82011-09-07 18:00:47 +02001301 if (!(s->listener->options & LI_O_UNLIMITED))
1302 actconn++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001303 totalconn++;
1304
1305 return s;
1306
1307 /* Error unrolling */
Willy Tarreau9b28e032012-10-12 23:49:43 +02001308 out_fail_rep_buf:
1309 pool_free2(pool2_channel, s->rep);
Emeric Brun2b920a12010-09-23 18:30:22 +02001310 out_fail_rep:
Willy Tarreau9b28e032012-10-12 23:49:43 +02001311 pool_free2(pool2_buffer, s->req->buf);
1312 out_fail_req_buf:
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001313 pool_free2(pool2_channel, s->req);
Emeric Brun2b920a12010-09-23 18:30:22 +02001314 out_fail_req:
1315 task_free(t);
1316 out_free_session:
1317 LIST_DEL(&s->list);
Willy Tarreauf2943dc2012-10-26 20:10:28 +02001318 pool_free2(pool2_connection, s->si[1].conn);
1319 out_fail_conn1:
1320 pool_free2(pool2_connection, s->si[0].conn);
1321 out_fail_conn0:
Emeric Brun2b920a12010-09-23 18:30:22 +02001322 pool_free2(pool2_session, s);
1323 out_close:
1324 return s;
1325}
1326
1327/*
1328 * Task processing function to manage re-connect and peer session
1329 * tasks wakeup on local update.
1330 */
Simon Horman96553772011-06-08 09:18:51 +09001331static struct task *process_peer_sync(struct task * task)
Emeric Brun2b920a12010-09-23 18:30:22 +02001332{
1333 struct shared_table *st = (struct shared_table *)task->context;
1334 struct peer_session *ps;
1335
1336 task->expire = TICK_ETERNITY;
1337
1338 if (!stopping) {
1339 /* Normal case (not soft stop)*/
1340 if (((st->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMLOCAL) &&
1341 (!nb_oldpids || tick_is_expired(st->resync_timeout, now_ms)) &&
1342 !(st->flags & SHTABLE_F_RESYNC_ASSIGN)) {
1343 /* Resync from local peer needed
1344 no peer was assigned for the lesson
1345 and no old local peer found
1346 or resync timeout expire */
1347
1348 /* flag no more resync from local, to try resync from remotes */
1349 st->flags |= SHTABLE_F_RESYNC_LOCAL;
1350
1351 /* reschedule a resync */
1352 st->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
1353 }
1354
1355 /* For each session */
1356 for (ps = st->sessions; ps; ps = ps->next) {
1357 /* For each remote peers */
1358 if (!ps->peer->local) {
1359 if (!ps->session) {
1360 /* no active session */
1361 if (ps->statuscode == 0 ||
1362 ps->statuscode == PEER_SESSION_SUCCESSCODE ||
1363 ((ps->statuscode == PEER_SESSION_CONNECTCODE ||
1364 ps->statuscode == PEER_SESSION_CONNECTEDCODE) &&
1365 tick_is_expired(ps->reconnect, now_ms))) {
1366 /* connection never tried
1367 * or previous session established with success
1368 * or previous session failed during connection
1369 * and reconnection timer is expired */
1370
1371 /* retry a connect */
1372 ps->session = peer_session_create(ps->peer, ps);
1373 }
1374 else if (ps->statuscode == PEER_SESSION_CONNECTCODE ||
1375 ps->statuscode == PEER_SESSION_CONNECTEDCODE) {
1376 /* If previous session failed during connection
1377 * but reconnection timer is not expired */
1378
1379 /* reschedule task for reconnect */
1380 task->expire = tick_first(task->expire, ps->reconnect);
1381 }
1382 /* else do nothing */
1383 } /* !ps->session */
1384 else if (ps->statuscode == PEER_SESSION_SUCCESSCODE) {
1385 /* current session is active and established */
1386 if (((st->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMREMOTE) &&
1387 !(st->flags & SHTABLE_F_RESYNC_ASSIGN) &&
1388 !(ps->flags & PEER_F_LEARN_NOTUP2DATE)) {
1389 /* Resync from a remote is needed
1390 * and no peer was assigned for lesson
1391 * and current peer may be up2date */
1392
1393 /* assign peer for the lesson */
1394 ps->flags |= PEER_F_LEARN_ASSIGN;
1395 st->flags |= SHTABLE_F_RESYNC_ASSIGN;
1396
1397 /* awake peer session task to handle a request of resync */
1398 task_wakeup(ps->session->task, TASK_WOKEN_MSG);
1399 }
1400 else if ((int)(ps->pushed - ps->table->table->localupdate) < 0) {
1401 /* awake peer session task to push local updates */
1402 task_wakeup(ps->session->task, TASK_WOKEN_MSG);
1403 }
1404 /* else do nothing */
1405 } /* SUCCESSCODE */
1406 } /* !ps->peer->local */
1407 } /* for */
1408
1409 /* Resync from remotes expired: consider resync is finished */
1410 if (((st->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMREMOTE) &&
1411 !(st->flags & SHTABLE_F_RESYNC_ASSIGN) &&
1412 tick_is_expired(st->resync_timeout, now_ms)) {
1413 /* Resync from remote peer needed
1414 * no peer was assigned for the lesson
1415 * and resync timeout expire */
1416
1417 /* flag no more resync from remote, consider resync is finished */
1418 st->flags |= SHTABLE_F_RESYNC_REMOTE;
1419 }
1420
1421 if ((st->flags & SHTABLE_RESYNC_STATEMASK) != SHTABLE_RESYNC_FINISHED) {
1422 /* Resync not finished*/
1423 /* reschedule task to resync timeout, to ended resync if needed */
1424 task->expire = tick_first(task->expire, st->resync_timeout);
1425 }
1426 } /* !stopping */
1427 else {
1428 /* soft stop case */
1429 if (task->state & TASK_WOKEN_SIGNAL) {
1430 /* We've just recieved the signal */
1431 if (!(st->flags & SHTABLE_F_DONOTSTOP)) {
1432 /* add DO NOT STOP flag if not present */
1433 jobs++;
1434 st->flags |= SHTABLE_F_DONOTSTOP;
Willy Tarreau3a925c12013-09-04 17:54:01 +02001435 st->table->syncing++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001436 }
1437
1438 /* disconnect all connected peers */
1439 for (ps = st->sessions; ps; ps = ps->next) {
1440 if (ps->session) {
1441 peer_session_forceshutdown(ps->session);
1442 ps->session = NULL;
1443 }
1444 }
1445 }
1446 ps = st->local_session;
1447
1448 if (ps->flags & PEER_F_TEACH_COMPLETE) {
1449 if (st->flags & SHTABLE_F_DONOTSTOP) {
1450 /* resync of new process was complete, current process can die now */
1451 jobs--;
1452 st->flags &= ~SHTABLE_F_DONOTSTOP;
Willy Tarreau3a925c12013-09-04 17:54:01 +02001453 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02001454 }
1455 }
1456 else if (!ps->session) {
1457 /* If session is not active */
1458 if (ps->statuscode == 0 ||
1459 ps->statuscode == PEER_SESSION_SUCCESSCODE ||
1460 ps->statuscode == PEER_SESSION_CONNECTEDCODE ||
1461 ps->statuscode == PEER_SESSION_TRYAGAIN) {
1462 /* connection never tried
1463 * or previous session was successfully established
1464 * or previous session tcp connect success but init state incomplete
1465 * or during previous connect, peer replies a try again statuscode */
1466
1467 /* connect to the peer */
1468 ps->session = peer_session_create(ps->peer, ps);
1469 }
1470 else {
1471 /* Other error cases */
1472 if (st->flags & SHTABLE_F_DONOTSTOP) {
1473 /* unable to resync new process, current process can die now */
1474 jobs--;
1475 st->flags &= ~SHTABLE_F_DONOTSTOP;
Willy Tarreau3a925c12013-09-04 17:54:01 +02001476 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02001477 }
1478 }
1479 }
1480 else if (ps->statuscode == PEER_SESSION_SUCCESSCODE &&
1481 (int)(ps->pushed - ps->table->table->localupdate) < 0) {
1482 /* current session active and established
1483 awake session to push remaining local updates */
1484 task_wakeup(ps->session->task, TASK_WOKEN_MSG);
1485 }
1486 } /* stopping */
1487 /* Wakeup for re-connect */
1488 return task;
1489}
1490
1491/*
1492 * Function used to register a table for sync on a group of peers
1493 *
1494 */
1495void peers_register_table(struct peers *peers, struct stktable *table)
1496{
1497 struct shared_table *st;
1498 struct peer * curpeer;
1499 struct peer_session *ps;
Willy Tarreau4348fad2012-09-20 16:48:07 +02001500 struct listener *listener;
Emeric Brun2b920a12010-09-23 18:30:22 +02001501
1502 st = (struct shared_table *)calloc(1,sizeof(struct shared_table));
1503 st->table = table;
1504 st->next = peers->tables;
1505 st->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
1506 peers->tables = st;
1507
1508 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
1509 ps = (struct peer_session *)calloc(1,sizeof(struct peer_session));
1510 ps->table = st;
1511 ps->peer = curpeer;
1512 if (curpeer->local)
1513 st->local_session = ps;
1514 ps->next = st->sessions;
1515 ps->reconnect = now_ms;
1516 st->sessions = ps;
1517 peers->peers_fe->maxconn += 3;
1518 }
1519
Willy Tarreau4348fad2012-09-20 16:48:07 +02001520 list_for_each_entry(listener, &peers->peers_fe->conf.listeners, by_fe)
1521 listener->maxconn = peers->peers_fe->maxconn;
Emeric Brun2b920a12010-09-23 18:30:22 +02001522 st->sync_task = task_new();
1523 st->sync_task->process = process_peer_sync;
1524 st->sync_task->expire = TICK_ETERNITY;
1525 st->sync_task->context = (void *)st;
1526 table->sync_task =st->sync_task;
1527 signal_register_task(0, table->sync_task, 0);
1528 task_wakeup(st->sync_task, TASK_WOKEN_INIT);
1529}
1530