blob: 0d06505d618d1789bc1b3f6453cd91b34d0ad1d3 [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
Willy Tarreaue4d927a2013-12-01 12:47:35 +010085enum {
86 PEER_SESS_ST_ACCEPT = 0, /* Initial state for session create by an accept, must be zero! */
87 PEER_SESS_ST_GETVERSION, /* Validate supported protocol version */
88 PEER_SESS_ST_GETHOST, /* Validate host ID correspond to local host id */
89 PEER_SESS_ST_GETPEER, /* Validate peer ID correspond to a known remote peer id */
90 PEER_SESS_ST_GETTABLE, /* Search into registered table for a table with same id and validate type and size */
91 /* after this point, data were possibly exchanged */
92 PEER_SESS_ST_SENDSUCCESS, /* Send ret code 200 (success) and wait for message */
93 PEER_SESS_ST_CONNECT, /* Initial state for session create on a connect, push presentation into buffer */
94 PEER_SESS_ST_GETSTATUS, /* Wait for the welcome message */
95 PEER_SESS_ST_WAITMSG, /* Wait for data messages */
96 PEER_SESS_ST_EXIT, /* Exit with status code */
97 PEER_SESS_ST_END, /* Killed session */
98};
Emeric Brun2b920a12010-09-23 18:30:22 +020099
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100100/***************************************************/
101/* Peer Session status code - part of the protocol */
102/***************************************************/
Emeric Brun2b920a12010-09-23 18:30:22 +0200103
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100104#define PEER_SESS_SC_CONNECTCODE 100 /* connect in progress */
105#define PEER_SESS_SC_CONNECTEDCODE 110 /* tcp connect success */
Emeric Brun2b920a12010-09-23 18:30:22 +0200106
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100107#define PEER_SESS_SC_SUCCESSCODE 200 /* accept or connect successful */
Emeric Brun2b920a12010-09-23 18:30:22 +0200108
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100109#define PEER_SESS_SC_TRYAGAIN 300 /* try again later */
Emeric Brun2b920a12010-09-23 18:30:22 +0200110
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100111#define PEER_SESS_SC_ERRPROTO 501 /* error protocol */
112#define PEER_SESS_SC_ERRVERSION 502 /* unknown protocol version */
113#define PEER_SESS_SC_ERRHOST 503 /* bad host name */
114#define PEER_SESS_SC_ERRPEER 504 /* unknown peer */
115#define PEER_SESS_SC_ERRTYPE 505 /* table key type mismatch */
116#define PEER_SESS_SC_ERRSIZE 506 /* table key size mismatch */
117#define PEER_SESS_SC_ERRTABLE 507 /* unknown table */
Emeric Brun2b920a12010-09-23 18:30:22 +0200118
119#define PEER_SESSION_PROTO_NAME "HAProxyS"
120
121struct peers *peers = NULL;
Simon Horman96553772011-06-08 09:18:51 +0900122static void peer_session_forceshutdown(struct session * session);
Emeric Brun2b920a12010-09-23 18:30:22 +0200123
124
125/*
126 * This prepare the data update message of the stick session <ts>, <ps> is the the peer session
127 * where the data going to be pushed, <msg> is a buffer of <size> to recieve data message content
128 */
Simon Horman96553772011-06-08 09:18:51 +0900129static int peer_prepare_datamsg(struct stksess *ts, struct peer_session *ps, char *msg, size_t size)
Emeric Brun2b920a12010-09-23 18:30:22 +0200130{
131 uint32_t netinteger;
132 int len;
133 /* construct message */
134 if (ps->lastpush && ts->upd.key > ps->lastpush && (ts->upd.key - ps->lastpush) <= 127) {
135 msg[0] = 0x80 + ts->upd.key - ps->lastpush;
136 len = sizeof(char);
137 }
138 else {
139 msg[0] = 'D';
140 netinteger = htonl(ts->upd.key);
141 memcpy(&msg[sizeof(char)], &netinteger, sizeof(netinteger));
142 len = sizeof(char) + sizeof(netinteger);
143 }
144
145 if (ps->table->table->type == STKTABLE_TYPE_STRING) {
146 int stlen = strlen((char *)ts->key.key);
147
148 netinteger = htonl(strlen((char *)ts->key.key));
149 memcpy(&msg[len], &netinteger, sizeof(netinteger));
150 memcpy(&msg[len+sizeof(netinteger)], ts->key.key, stlen);
151 len += sizeof(netinteger) + stlen;
152
153 }
154 else if (ps->table->table->type == STKTABLE_TYPE_INTEGER) {
155 netinteger = htonl(*((uint32_t *)ts->key.key));
156 memcpy(&msg[len], &netinteger, sizeof(netinteger));
157 len += sizeof(netinteger);
158 }
159 else {
160 memcpy(&msg[len], ts->key.key, ps->table->table->key_size);
161 len += ps->table->table->key_size;
162 }
163
164 if (stktable_data_ptr(ps->table->table, ts, STKTABLE_DT_SERVER_ID))
165 netinteger = htonl(stktable_data_cast(stktable_data_ptr(ps->table->table, ts, STKTABLE_DT_SERVER_ID), server_id));
166 else
167 netinteger = 0;
168
169 memcpy(&msg[len], &netinteger , sizeof(netinteger));
170 len += sizeof(netinteger);
171
172 return len;
173}
174
175
176/*
177 * Callback to release a session with a peer
178 */
Simon Horman96553772011-06-08 09:18:51 +0900179static void peer_session_release(struct stream_interface *si)
Emeric Brun2b920a12010-09-23 18:30:22 +0200180{
Willy Tarreau9f681482013-07-08 16:05:07 +0200181 struct session *s = session_from_task(si->owner);
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100182 struct appctx *appctx = objt_appctx(si->end);
183 struct peer_session *ps = (struct peer_session *)appctx->ctx.peers.ptr;
Emeric Brun2b920a12010-09-23 18:30:22 +0200184
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100185 /* appctx->ctx.peers.ptr is not a peer session */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100186 if (appctx->st0 < PEER_SESS_ST_SENDSUCCESS)
Emeric Brun2b920a12010-09-23 18:30:22 +0200187 return;
188
189 /* peer session identified */
190 if (ps) {
191 if (ps->session == s) {
192 ps->session = NULL;
193 if (ps->flags & PEER_F_LEARN_ASSIGN) {
194 /* unassign current peer for learning */
195 ps->flags &= ~(PEER_F_LEARN_ASSIGN);
196 ps->table->flags &= ~(SHTABLE_F_RESYNC_ASSIGN|SHTABLE_F_RESYNC_PROCESS);
197
198 /* reschedule a resync */
199 ps->table->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
200 }
201 /* reset teaching and learning flags to 0 */
202 ps->flags &= PEER_TEACH_RESET;
203 ps->flags &= PEER_LEARN_RESET;
204 }
205 task_wakeup(ps->table->sync_task, TASK_WOKEN_MSG);
206 }
207}
208
209
210/*
211 * IO Handler to handle message exchance with a peer
212 */
Willy Tarreaub24281b2011-02-13 13:16:36 +0100213static void peer_io_handler(struct stream_interface *si)
Emeric Brun2b920a12010-09-23 18:30:22 +0200214{
Willy Tarreau9f681482013-07-08 16:05:07 +0200215 struct session *s = session_from_task(si->owner);
Emeric Brun2b920a12010-09-23 18:30:22 +0200216 struct peers *curpeers = (struct peers *)s->fe->parent;
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100217 struct appctx *appctx = objt_appctx(si->end);
Emeric Brun2b920a12010-09-23 18:30:22 +0200218 int reql = 0;
219 int repl = 0;
220
221 while (1) {
222switchstate:
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100223 switch(appctx->st0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100224 case PEER_SESS_ST_ACCEPT:
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100225 appctx->ctx.peers.ptr = NULL;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100226 appctx->st0 = PEER_SESS_ST_GETVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200227 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100228 case PEER_SESS_ST_GETVERSION:
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100229 reql = bo_getline(si->ob, trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200230 if (reql <= 0) { /* closed or EOL not found */
231 if (reql == 0)
232 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100233 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200234 goto switchstate;
235 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100236 if (trash.str[reql-1] != '\n') {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100237 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200238 goto switchstate;
239 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100240 else if (reql > 1 && (trash.str[reql-2] == '\r'))
241 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200242 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100243 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200244
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200245 bo_skip(si->ob, reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200246
247 /* test version */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100248 if (strcmp(PEER_SESSION_PROTO_NAME " 1.0", trash.str) != 0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100249 appctx->st0 = PEER_SESS_ST_EXIT;
250 appctx->st1 = PEER_SESS_SC_ERRVERSION;
Emeric Brun2b920a12010-09-23 18:30:22 +0200251 /* test protocol */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100252 if (strncmp(PEER_SESSION_PROTO_NAME " ", trash.str, strlen(PEER_SESSION_PROTO_NAME)+1) != 0)
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100253 appctx->st1 = PEER_SESS_SC_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +0200254 goto switchstate;
255 }
256
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100257 appctx->st0 = PEER_SESS_ST_GETHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200258 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100259 case PEER_SESS_ST_GETHOST:
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100260 reql = bo_getline(si->ob, trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200261 if (reql <= 0) { /* closed or EOL not found */
262 if (reql == 0)
263 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100264 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200265 goto switchstate;
266 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100267 if (trash.str[reql-1] != '\n') {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100268 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200269 goto switchstate;
270 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100271 else if (reql > 1 && (trash.str[reql-2] == '\r'))
272 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200273 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100274 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200275
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200276 bo_skip(si->ob, reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200277
278 /* test hostname match */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100279 if (strcmp(localpeer, trash.str) != 0) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100280 appctx->st0 = PEER_SESS_ST_EXIT;
281 appctx->st1 = PEER_SESS_SC_ERRHOST;
Emeric Brun2b920a12010-09-23 18:30:22 +0200282 goto switchstate;
283 }
284
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100285 appctx->st0 = PEER_SESS_ST_GETPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200286 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100287 case PEER_SESS_ST_GETPEER: {
Emeric Brun2b920a12010-09-23 18:30:22 +0200288 struct peer *curpeer;
289 char *p;
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100290 reql = bo_getline(si->ob, trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200291 if (reql <= 0) { /* closed or EOL not found */
292 if (reql == 0)
293 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100294 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200295 goto switchstate;
296 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100297 if (trash.str[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200298 /* Incomplete line, we quit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100299 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200300 goto switchstate;
301 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100302 else if (reql > 1 && (trash.str[reql-2] == '\r'))
303 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200304 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100305 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200306
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200307 bo_skip(si->ob, reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200308
309 /* parse line "<peer name> <pid>" */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100310 p = strchr(trash.str, ' ');
Emeric Brun2b920a12010-09-23 18:30:22 +0200311 if (!p) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100312 appctx->st0 = PEER_SESS_ST_EXIT;
313 appctx->st1 = PEER_SESS_SC_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +0200314 goto switchstate;
315 }
316 *p = 0;
317
318 /* lookup known peer */
319 for (curpeer = curpeers->remote; curpeer; curpeer = curpeer->next) {
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100320 if (strcmp(curpeer->id, trash.str) == 0)
Emeric Brun2b920a12010-09-23 18:30:22 +0200321 break;
322 }
323
324 /* if unknown peer */
325 if (!curpeer) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100326 appctx->st0 = PEER_SESS_ST_EXIT;
327 appctx->st1 = PEER_SESS_SC_ERRPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200328 goto switchstate;
329 }
330
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100331 appctx->ctx.peers.ptr = curpeer;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100332 appctx->st0 = PEER_SESS_ST_GETTABLE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200333 /* fall through */
334 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100335 case PEER_SESS_ST_GETTABLE: {
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100336 struct peer *curpeer = (struct peer *)appctx->ctx.peers.ptr;
Emeric Brun2b920a12010-09-23 18:30:22 +0200337 struct shared_table *st;
338 struct peer_session *ps = NULL;
339 unsigned long key_type;
340 size_t key_size;
341 char *p;
342
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100343 reql = bo_getline(si->ob, trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200344 if (reql <= 0) { /* closed or EOL not found */
345 if (reql == 0)
346 goto out;
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100347 appctx->ctx.peers.ptr = NULL;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100348 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200349 goto switchstate;
350 }
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100351 /* Re init appctx->ctx.peers.ptr to null, to handle correctly a release case */
352 appctx->ctx.peers.ptr = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +0200353
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100354 if (trash.str[reql-1] != '\n') {
Emeric Brun2b920a12010-09-23 18:30:22 +0200355 /* Incomplete line, we quit */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100356 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200357 goto switchstate;
358 }
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100359 else if (reql > 1 && (trash.str[reql-2] == '\r'))
360 trash.str[reql-2] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200361 else
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100362 trash.str[reql-1] = 0;
Emeric Brun2b920a12010-09-23 18:30:22 +0200363
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200364 bo_skip(si->ob, reql);
Emeric Brun2b920a12010-09-23 18:30:22 +0200365
366 /* Parse line "<table name> <type> <size>" */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100367 p = strchr(trash.str, ' ');
Emeric Brun2b920a12010-09-23 18:30:22 +0200368 if (!p) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100369 appctx->st0 = PEER_SESS_ST_EXIT;
370 appctx->st1 = PEER_SESS_SC_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +0200371 goto switchstate;
372 }
373 *p = 0;
374 key_type = (unsigned long)atol(p+1);
375
376 p = strchr(p+1, ' ');
377 if (!p) {
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100378 appctx->ctx.peers.ptr = NULL;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100379 appctx->st0 = PEER_SESS_ST_EXIT;
380 appctx->st1 = PEER_SESS_SC_ERRPROTO;
Emeric Brun2b920a12010-09-23 18:30:22 +0200381 goto switchstate;
382 }
383
384 key_size = (size_t)atoi(p);
385 for (st = curpeers->tables; st; st = st->next) {
386 /* If table name matches */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100387 if (strcmp(st->table->id, trash.str) == 0) {
Willy Tarreau86a446e2013-11-25 23:02:37 +0100388 /* Check key size mismatches, except for strings
389 * which may be truncated as long as they fit in
390 * a buffer.
391 */
392 if (key_size != st->table->key_size &&
393 (key_type != STKTABLE_TYPE_STRING ||
394 1 + 4 + 4 + key_size - 1 >= trash.size)) {
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100395 appctx->st0 = PEER_SESS_ST_EXIT;
396 appctx->st1 = PEER_SESS_SC_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 Tarreaue4d927a2013-12-01 12:47:35 +0100402 appctx->st0 = PEER_SESS_ST_EXIT;
403 appctx->st1 = PEER_SESS_SC_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 Tarreaue4d927a2013-12-01 12:47:35 +0100414 appctx->st0 = PEER_SESS_ST_EXIT;
415 appctx->st1 = PEER_SESS_SC_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 Tarreaue4d927a2013-12-01 12:47:35 +0100430 appctx->st0 = PEER_SESS_ST_EXIT;
431 appctx->st1 = PEER_SESS_SC_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 Tarreaue4d927a2013-12-01 12:47:35 +0100437 appctx->st0 = PEER_SESS_ST_EXIT;
438 appctx->st1 = PEER_SESS_SC_ERRPEER;
Emeric Brun2b920a12010-09-23 18:30:22 +0200439 goto switchstate;
440 }
441
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100442 appctx->ctx.peers.ptr = ps;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100443 appctx->st0 = PEER_SESS_ST_SENDSUCCESS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200444 /* fall through */
445 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100446 case PEER_SESS_ST_SENDSUCCESS: {
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100447 struct peer_session *ps = (struct peer_session *)appctx->ctx.peers.ptr;
Emeric Brun2b920a12010-09-23 18:30:22 +0200448
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100449 repl = snprintf(trash.str, trash.size, "%d\n", PEER_SESS_SC_SUCCESSCODE);
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100450 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 Tarreaue4d927a2013-12-01 12:47:35 +0100454 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200455 goto switchstate;
456 }
457
458 /* Register status code */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100459 ps->statuscode = PEER_SESS_SC_SUCCESSCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200460
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 Tarreaue4d927a2013-12-01 12:47:35 +0100493 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200494 goto switchstate;
495 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100496 case PEER_SESS_ST_CONNECT: {
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100497 struct peer_session *ps = (struct peer_session *)appctx->ctx.peers.ptr;
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 Tarreaue4d927a2013-12-01 12:47:35 +0100510 appctx->st0 = PEER_SESS_ST_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 Tarreaue4d927a2013-12-01 12:47:35 +0100518 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200519 goto switchstate;
520 }
521
522 /* switch to the waiting statuscode state */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100523 appctx->st0 = PEER_SESS_ST_GETSTATUS;
Emeric Brun2b920a12010-09-23 18:30:22 +0200524 /* fall through */
525 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100526 case PEER_SESS_ST_GETSTATUS: {
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100527 struct peer_session *ps = (struct peer_session *)appctx->ctx.peers.ptr;
Emeric Brun2b920a12010-09-23 18:30:22 +0200528
Willy Tarreau03cdb7c2012-08-27 23:14:58 +0200529 if (si->ib->flags & CF_WRITE_PARTIAL)
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100530 ps->statuscode = PEER_SESS_SC_CONNECTEDCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +0200531
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 Tarreaue4d927a2013-12-01 12:47:35 +0100536 appctx->st0 = PEER_SESS_ST_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 Tarreaue4d927a2013-12-01 12:47:35 +0100541 appctx->st0 = PEER_SESS_ST_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 */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100558 if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Emeric Brun2b920a12010-09-23 18:30:22 +0200559 /* 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 Tarreaue4d927a2013-12-01 12:47:35 +0100591 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200592 goto switchstate;
593 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100594 appctx->st0 = PEER_SESS_ST_WAITMSG;
Emeric Brun2b920a12010-09-23 18:30:22 +0200595 /* fall through */
596 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100597 case PEER_SESS_ST_WAITMSG: {
Willy Tarreau7b4b4992013-12-01 09:15:12 +0100598 struct peer_session *ps = (struct peer_session *)appctx->ctx.peers.ptr;
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 Tarreau86a446e2013-11-25 23:02:37 +0100628 /* Read key. The string keys are read in two steps, the first step
629 * consists in reading whatever fits into the table directly into
630 * the pre-allocated key. The second step consists in simply
631 * draining all exceeding data. This can happen for example after a
632 * config reload with a smaller key size for the stick table than
633 * what was previously set, or when facing the impossibility to
634 * allocate a new stksess (for example when the table is full with
635 * "nopurge").
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200636 */
Emeric Brun2b920a12010-09-23 18:30:22 +0200637 if (ps->table->table->type == STKTABLE_TYPE_STRING) {
Willy Tarreau86a446e2013-11-25 23:02:37 +0100638 unsigned int to_read, to_store;
639
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200640 /* read size first */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200641 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200642 if (reql <= 0) /* closed or EOL not found */
643 goto incomplete;
644
Emeric Brun2b920a12010-09-23 18:30:22 +0200645 totl += reql;
Willy Tarreau86a446e2013-11-25 23:02:37 +0100646
647 to_store = 0;
648 to_read = ntohl(netinteger);
649
650 if (to_read + totl > si->ob->buf->size) {
651 /* impossible to read a key this large, abort */
652 reql = -1;
Willy Tarreau72d6c162013-04-11 16:14:13 +0200653 goto incomplete;
Willy Tarreau86a446e2013-11-25 23:02:37 +0100654 }
Willy Tarreau72d6c162013-04-11 16:14:13 +0200655
Willy Tarreau86a446e2013-11-25 23:02:37 +0100656 newts = stksess_new(ps->table->table, NULL);
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200657 if (newts)
Willy Tarreau86a446e2013-11-25 23:02:37 +0100658 to_store = MIN(to_read, ps->table->table->key_size - 1);
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200659
Willy Tarreau86a446e2013-11-25 23:02:37 +0100660 /* we read up to two blocks, the first one goes into the key,
661 * the rest is drained into the trash.
662 */
663 if (to_store) {
664 reql = bo_getblk(si->ob, (char *)newts->key.key, to_store, totl);
665 if (reql <= 0) /* closed or incomplete */
666 goto incomplete;
667 newts->key.key[reql] = 0;
668 totl += reql;
669 to_read -= reql;
670 }
671 if (to_read) {
672 reql = bo_getblk(si->ob, trash.str, to_read, totl);
673 if (reql <= 0) /* closed or incomplete */
674 goto incomplete;
675 totl += reql;
676 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200677 }
678 else if (ps->table->table->type == STKTABLE_TYPE_INTEGER) {
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200679 newts = stksess_new(ps->table->table, NULL);
680 reql = bo_getblk(si->ob, newts ? (char *)newts->key.key : trash.str, sizeof(netinteger), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200681 if (reql <= 0) /* closed or EOL not found */
682 goto incomplete;
Emeric Brun2b920a12010-09-23 18:30:22 +0200683 totl += reql;
Emeric Brun2b920a12010-09-23 18:30:22 +0200684 }
685 else {
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200686 /* type ip or binary */
687 newts = stksess_new(ps->table->table, NULL);
688 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 +0200689 if (reql <= 0) /* closed or EOL not found */
690 goto incomplete;
Willy Tarreau72d6c162013-04-11 16:14:13 +0200691 totl += reql;
Emeric Brun2b920a12010-09-23 18:30:22 +0200692 }
693
694 /* read server id */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200695 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200696 if (reql <= 0) /* closed or EOL not found */
697 goto incomplete;
698
Emeric Brun2b920a12010-09-23 18:30:22 +0200699 totl += reql;
700 srvid = ntohl(netinteger);
701
702 /* update entry */
Emeric Brun2b920a12010-09-23 18:30:22 +0200703 if (newts) {
704 /* lookup for existing entry */
705 ts = stktable_lookup(ps->table->table, newts);
706 if (ts) {
707 /* the entry already exist, we can free ours */
708 stktable_touch(ps->table->table, ts, 0);
709 stksess_free(ps->table->table, newts);
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200710 newts = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +0200711 }
712 else {
713 struct eb32_node *eb;
714
715 /* create new entry */
716 ts = stktable_store(ps->table->table, newts, 0);
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200717 newts = NULL; /* don't reuse it */
718
Emeric Brun2b920a12010-09-23 18:30:22 +0200719 ts->upd.key= (++ps->table->table->update)+(2^31);
720 eb = eb32_insert(&ps->table->table->updates, &ts->upd);
721 if (eb != &ts->upd) {
722 eb32_delete(eb);
723 eb32_insert(&ps->table->table->updates, &ts->upd);
724 }
725 }
726
727 /* update entry */
728 if (srvid && stktable_data_ptr(ps->table->table, ts, STKTABLE_DT_SERVER_ID))
729 stktable_data_cast(stktable_data_ptr(ps->table->table, ts, STKTABLE_DT_SERVER_ID), server_id) = srvid;
730 ps->pushack = pushack;
731 }
732
733 }
734 else if (c == 'R') {
735 /* Reset message: remote need resync */
736
737 /* reinit counters for a resync */
738 ps->lastpush = 0;
739 ps->teaching_origin = ps->pushed = ps->table->table->update;
740
741 /* reset teaching flags to 0 */
742 ps->flags &= PEER_TEACH_RESET;
743
744 /* flag to start to teach lesson */
745 ps->flags |= PEER_F_TEACH_PROCESS;
746 }
747 else if (c == 'F') {
748 /* Finish message, all known updates have been pushed by remote */
749 /* and remote is up to date */
750
751 /* If resync is in progress with remote peer */
752 if (ps->flags & PEER_F_LEARN_ASSIGN) {
753
754 /* unassign current peer for learning */
755 ps->flags &= ~PEER_F_LEARN_ASSIGN;
756 ps->table->flags &= ~(SHTABLE_F_RESYNC_ASSIGN|SHTABLE_F_RESYNC_PROCESS);
757
758 /* Consider table is now up2date, resync resync no more needed from local neither remote */
759 ps->table->flags |= (SHTABLE_F_RESYNC_LOCAL|SHTABLE_F_RESYNC_REMOTE);
760 }
761 /* Increase confirm counter to launch a confirm message */
762 ps->confirm++;
763 }
764 else if (c == 'c') {
765 /* confirm message, remote peer is now up to date with us */
766
767 /* If stopping state */
768 if (stopping) {
769 /* Close session, push resync no more needed */
770 ps->flags |= PEER_F_TEACH_COMPLETE;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100771 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200772 goto switchstate;
773 }
774
775 /* reset teaching flags to 0 */
776 ps->flags &= PEER_TEACH_RESET;
777 }
778 else if (c == 'C') {
779 /* Continue message, all known updates have been pushed by remote */
780 /* but remote is not up to date */
781
782 /* If resync is in progress with current peer */
783 if (ps->flags & PEER_F_LEARN_ASSIGN) {
784
785 /* unassign current peer */
786 ps->flags &= ~PEER_F_LEARN_ASSIGN;
787 ps->table->flags &= ~(SHTABLE_F_RESYNC_ASSIGN|SHTABLE_F_RESYNC_PROCESS);
788
789 /* flag current peer is not up 2 date to try from an other */
790 ps->flags |= PEER_F_LEARN_NOTUP2DATE;
791
792 /* reschedule a resync */
793 ps->table->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
794 task_wakeup(ps->table->sync_task, TASK_WOKEN_MSG);
795 }
796 ps->confirm++;
797 }
798 else if (c == 'A') {
799 /* ack message */
800 uint32_t netinteger;
801
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200802 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200803 if (reql <= 0) /* closed or EOL not found */
804 goto incomplete;
805
Emeric Brun2b920a12010-09-23 18:30:22 +0200806 totl += reql;
807
808 /* Consider remote is up to date with "acked" version */
809 ps->update = ntohl(netinteger);
810 }
811 else {
812 /* Unknown message */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100813 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200814 goto switchstate;
815 }
816
817 /* skip consumed message */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200818 bo_skip(si->ob, totl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200819
820 /* loop on that state to peek next message */
Willy Tarreau72d6c162013-04-11 16:14:13 +0200821 goto switchstate;
822
Emeric Brun2b920a12010-09-23 18:30:22 +0200823incomplete:
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200824 /* we get here when a bo_getblk() returns <= 0 in reql */
825
826 /* first, we may have to release newts */
827 if (newts) {
828 stksess_free(ps->table->table, newts);
829 newts = NULL;
830 }
831
Willy Tarreau72d6c162013-04-11 16:14:13 +0200832 if (reql < 0) {
833 /* there was an error */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100834 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau72d6c162013-04-11 16:14:13 +0200835 goto switchstate;
836 }
837
Emeric Brun2b920a12010-09-23 18:30:22 +0200838 /* Nothing to read, now we start to write */
839
840 /* Confirm finished or partial messages */
841 while (ps->confirm) {
842 /* There is a confirm messages to send */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200843 repl = bi_putchr(si->ib, 'c');
Emeric Brun2b920a12010-09-23 18:30:22 +0200844 if (repl <= 0) {
845 /* no more write possible */
846 if (repl == -1)
847 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100848 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200849 goto switchstate;
850 }
851 ps->confirm--;
852 }
853
854 /* Need to request a resync */
855 if ((ps->flags & PEER_F_LEARN_ASSIGN) &&
856 (ps->table->flags & SHTABLE_F_RESYNC_ASSIGN) &&
857 !(ps->table->flags & SHTABLE_F_RESYNC_PROCESS)) {
858 /* Current peer was elected to request a resync */
859
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200860 repl = bi_putchr(si->ib, 'R');
Emeric Brun2b920a12010-09-23 18:30:22 +0200861 if (repl <= 0) {
862 /* no more write possible */
863 if (repl == -1)
864 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100865 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200866 goto switchstate;
867 }
868 ps->table->flags |= SHTABLE_F_RESYNC_PROCESS;
869 }
870
871 /* It remains some updates to ack */
872 if (ps->pushack != ps->lastack) {
873 uint32_t netinteger;
874
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100875 trash.str[0] = 'A';
Emeric Brun2b920a12010-09-23 18:30:22 +0200876 netinteger = htonl(ps->pushack);
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100877 memcpy(&trash.str[1], &netinteger, sizeof(netinteger));
Emeric Brun2b920a12010-09-23 18:30:22 +0200878
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100879 repl = bi_putblk(si->ib, trash.str, 1+sizeof(netinteger));
Emeric Brun2b920a12010-09-23 18:30:22 +0200880 if (repl <= 0) {
881 /* no more write possible */
882 if (repl == -1)
883 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100884 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200885 goto switchstate;
886 }
887 ps->lastack = ps->pushack;
888 }
889
890 if (ps->flags & PEER_F_TEACH_PROCESS) {
891 /* current peer was requested for a lesson */
892
893 if (!(ps->flags & PEER_F_TEACH_STAGE1)) {
894 /* lesson stage 1 not complete */
895 struct eb32_node *eb;
896
897 eb = eb32_lookup_ge(&ps->table->table->updates, ps->pushed+1);
898 while (1) {
899 int msglen;
900 struct stksess *ts;
901
902 if (!eb) {
903 /* flag lesson stage1 complete */
904 ps->flags |= PEER_F_TEACH_STAGE1;
905 eb = eb32_first(&ps->table->table->updates);
906 if (eb)
907 ps->pushed = eb->key - 1;
908 break;
909 }
910
911 ts = eb32_entry(eb, struct stksess, upd);
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100912 msglen = peer_prepare_datamsg(ts, ps, trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200913 if (msglen) {
914 /* message to buffer */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100915 repl = bi_putblk(si->ib, trash.str, msglen);
Emeric Brun2b920a12010-09-23 18:30:22 +0200916 if (repl <= 0) {
917 /* no more write possible */
918 if (repl == -1)
919 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100920 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200921 goto switchstate;
922 }
923 ps->lastpush = ps->pushed = ts->upd.key;
924 }
925 eb = eb32_next(eb);
926 }
927 } /* !TEACH_STAGE1 */
928
929 if (!(ps->flags & PEER_F_TEACH_STAGE2)) {
930 /* lesson stage 2 not complete */
931 struct eb32_node *eb;
932
933 eb = eb32_lookup_ge(&ps->table->table->updates, ps->pushed+1);
934 while (1) {
935 int msglen;
936 struct stksess *ts;
937
938 if (!eb || eb->key > ps->teaching_origin) {
939 /* flag lesson stage1 complete */
940 ps->flags |= PEER_F_TEACH_STAGE2;
941 ps->pushed = ps->teaching_origin;
942 break;
943 }
944
945 ts = eb32_entry(eb, struct stksess, upd);
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100946 msglen = peer_prepare_datamsg(ts, ps, trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200947 if (msglen) {
948 /* message to buffer */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100949 repl = bi_putblk(si->ib, trash.str, msglen);
Emeric Brun2b920a12010-09-23 18:30:22 +0200950 if (repl <= 0) {
951 /* no more write possible */
952 if (repl == -1)
953 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100954 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200955 goto switchstate;
956 }
957 ps->lastpush = ps->pushed = ts->upd.key;
958 }
959 eb = eb32_next(eb);
960 }
961 } /* !TEACH_STAGE2 */
962
963 if (!(ps->flags & PEER_F_TEACH_FINISHED)) {
964 /* process final lesson message */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200965 repl = bi_putchr(si->ib, ((ps->table->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FINISHED) ? 'F' : 'C');
Emeric Brun2b920a12010-09-23 18:30:22 +0200966 if (repl <= 0) {
967 /* no more write possible */
968 if (repl == -1)
969 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100970 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200971 goto switchstate;
972 }
973
974 /* flag finished message sent */
975 ps->flags |= PEER_F_TEACH_FINISHED;
976 } /* !TEACH_FINISHED */
977 } /* TEACH_PROCESS */
978
979 if (!(ps->flags & PEER_F_LEARN_ASSIGN) &&
980 (int)(ps->pushed - ps->table->table->localupdate) < 0) {
981 /* Push local updates, only if no learning in progress (to avoid ping-pong effects) */
982 struct eb32_node *eb;
983
984 eb = eb32_lookup_ge(&ps->table->table->updates, ps->pushed+1);
985 while (1) {
986 int msglen;
987 struct stksess *ts;
988
989 /* push local updates */
990 if (!eb) {
991 eb = eb32_first(&ps->table->table->updates);
992 if (!eb || ((int)(eb->key - ps->pushed) <= 0)) {
993 ps->pushed = ps->table->table->localupdate;
994 break;
995 }
996 }
997
998 if ((int)(eb->key - ps->table->table->localupdate) > 0) {
999 ps->pushed = ps->table->table->localupdate;
1000 break;
1001 }
1002
1003 ts = eb32_entry(eb, struct stksess, upd);
Willy Tarreau19d14ef2012-10-29 16:51:55 +01001004 msglen = peer_prepare_datamsg(ts, ps, trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +02001005 if (msglen) {
1006 /* message to buffer */
Willy Tarreau19d14ef2012-10-29 16:51:55 +01001007 repl = bi_putblk(si->ib, trash.str, msglen);
Emeric Brun2b920a12010-09-23 18:30:22 +02001008 if (repl <= 0) {
1009 /* no more write possible */
1010 if (repl == -1)
1011 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001012 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02001013 goto switchstate;
1014 }
1015 ps->lastpush = ps->pushed = ts->upd.key;
1016 }
1017 eb = eb32_next(eb);
1018 }
1019 } /* ! LEARN_ASSIGN */
1020 /* noting more to do */
1021 goto out;
1022 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001023 case PEER_SESS_ST_EXIT:
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001024 repl = snprintf(trash.str, trash.size, "%d\n", appctx->st1);
Emeric Brun2b920a12010-09-23 18:30:22 +02001025
Willy Tarreau19d14ef2012-10-29 16:51:55 +01001026 if (bi_putblk(si->ib, trash.str, repl) == -1)
Emeric Brun2b920a12010-09-23 18:30:22 +02001027 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001028 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02001029 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001030 case PEER_SESS_ST_END: {
Willy Tarreau73b013b2012-05-21 16:31:45 +02001031 si_shutw(si);
1032 si_shutr(si);
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001033 si->ib->flags |= CF_READ_NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001034 goto quit;
1035 }
1036 }
1037 }
1038out:
Willy Tarreau73b013b2012-05-21 16:31:45 +02001039 si_update(si);
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001040 si->ob->flags |= CF_READ_DONTWAIT;
Emeric Brun2b920a12010-09-23 18:30:22 +02001041 /* we don't want to expire timeouts while we're processing requests */
1042 si->ib->rex = TICK_ETERNITY;
1043 si->ob->wex = TICK_ETERNITY;
1044quit:
1045 return;
1046}
1047
Willy Tarreaub24281b2011-02-13 13:16:36 +01001048static struct si_applet peer_applet = {
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001049 .obj_type = OBJ_TYPE_APPLET,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001050 .name = "<PEER>", /* used for logging */
1051 .fct = peer_io_handler,
Aman Gupta9a13e842012-04-02 18:57:53 -07001052 .release = peer_session_release,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001053};
Emeric Brun2b920a12010-09-23 18:30:22 +02001054
1055/*
1056 * Use this function to force a close of a peer session
1057 */
Simon Horman96553772011-06-08 09:18:51 +09001058static void peer_session_forceshutdown(struct session * session)
Emeric Brun2b920a12010-09-23 18:30:22 +02001059{
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001060 struct stream_interface *oldsi = NULL;
1061 struct appctx *appctx = NULL;
1062 int i;
Emeric Brun2b920a12010-09-23 18:30:22 +02001063
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001064 for (i = 0; i <= 1; i++) {
1065 appctx = objt_appctx(session->si[i].end);
1066 if (!appctx)
1067 continue;
1068 if (appctx->applet != &peer_applet)
1069 continue;
1070
1071 oldsi = &session->si[i];
1072 break;
Emeric Brun2b920a12010-09-23 18:30:22 +02001073 }
1074
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001075 if (!appctx)
1076 return;
1077
Emeric Brun2b920a12010-09-23 18:30:22 +02001078 /* call release to reinit resync states if needed */
1079 peer_session_release(oldsi);
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001080 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001081 appctx->ctx.peers.ptr = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001082 task_wakeup(session->task, TASK_WOKEN_MSG);
1083}
1084
Willy Tarreaub363a1f2013-10-01 10:45:07 +02001085/* Finish a session accept() for a peer. It returns a negative value in case of
1086 * a critical failure which must cause the listener to be disabled, a positive
1087 * value in case of success, or zero if it is a success but the session must be
1088 * closed ASAP and ignored.
Emeric Brun2b920a12010-09-23 18:30:22 +02001089 */
1090int peer_accept(struct session *s)
1091{
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001092 s->target = &peer_applet.obj_type;
Willy Tarreau6d7f8f72013-12-01 12:54:55 +01001093 /* no need to initialize the applet, it will start with st0=st1 = 0 */
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 */
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001120 struct appctx *appctx;
Emeric Brun2b920a12010-09-23 18:30:22 +02001121 struct session *s;
1122 struct http_txn *txn;
1123 struct task *t;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001124 struct connection *conn;
Emeric Brun2b920a12010-09-23 18:30:22 +02001125
1126 if ((s = pool_alloc2(pool2_session)) == NULL) { /* disable this proxy for a while */
Godbach430f2912013-06-20 13:28:38 +08001127 Alert("out of memory in peer_session_create().\n");
Emeric Brun2b920a12010-09-23 18:30:22 +02001128 goto out_close;
1129 }
1130
1131 LIST_ADDQ(&sessions, &s->list);
1132 LIST_INIT(&s->back_refs);
1133
1134 s->flags = SN_ASSIGNED|SN_ADDR_SET;
Emeric Brun2b920a12010-09-23 18:30:22 +02001135
1136 /* if this session comes from a known monitoring system, we want to ignore
1137 * it as soon as possible, which means closing it immediately for TCP.
1138 */
1139 if ((t = task_new()) == NULL) { /* disable this proxy for a while */
Godbach430f2912013-06-20 13:28:38 +08001140 Alert("out of memory in peer_session_create().\n");
Emeric Brun2b920a12010-09-23 18:30:22 +02001141 goto out_free_session;
1142 }
1143
1144 ps->reconnect = tick_add(now_ms, MS_TO_TICKS(5000));
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001145 ps->statuscode = PEER_SESS_SC_CONNECTCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +02001146
1147 t->process = l->handler;
1148 t->context = s;
1149 t->nice = l->nice;
1150
Emeric Brun2b920a12010-09-23 18:30:22 +02001151 s->task = t;
1152 s->listener = l;
1153
1154 /* Note: initially, the session's backend points to the frontend.
1155 * This changes later when switching rules are executed or
1156 * when the default backend is assigned.
1157 */
1158 s->be = s->fe = p;
1159
1160 s->req = s->rep = NULL; /* will be allocated later */
1161
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001162 si_reset(&s->si[0], t);
1163 si_set_state(&s->si[0], SI_ST_EST);
1164
Emeric Brun2b920a12010-09-23 18:30:22 +02001165 if (s->fe->options2 & PR_O2_INDEPSTR)
1166 s->si[0].flags |= SI_FL_INDEP_STR;
Emeric Brun2b920a12010-09-23 18:30:22 +02001167
Willy Tarreau1fbe1c92013-12-01 09:35:41 +01001168 appctx = stream_int_register_handler(&s->si[0], &peer_applet);
1169 if (!appctx)
1170 goto out_fail_conn1;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001171 appctx->st0 = PEER_SESS_ST_CONNECT;
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001172 appctx->ctx.peers.ptr = (void *)ps;
Emeric Brun2b920a12010-09-23 18:30:22 +02001173
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001174 si_reset(&s->si[1], t);
1175
1176 /* initiate an outgoing connection */
1177 si_set_state(&s->si[1], SI_ST_ASS);
Emeric Brun2b920a12010-09-23 18:30:22 +02001178 s->si[1].conn_retries = p->conn_retries;
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001179
Emeric Brun2b920a12010-09-23 18:30:22 +02001180 if (s->be->options2 & PR_O2_INDEPSTR)
1181 s->si[1].flags |= SI_FL_INDEP_STR;
1182
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001183 /* automatically prepare the stream interface to connect to the
Willy Tarreaub363a1f2013-10-01 10:45:07 +02001184 * pre-initialized connection in si->conn.
1185 */
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001186 if (unlikely((conn = conn_new()) == NULL))
1187 goto out_fail_conn1;
1188
1189 conn_prepare(conn, peer->proto, peer->xprt);
1190 si_attach_conn(&s->si[1], conn);
1191
1192 conn->target = s->target = &s->be->obj_type;
1193 memcpy(&conn->addr.to, &peer->addr, sizeof(conn->addr.to));
Willy Tarreaub363a1f2013-10-01 10:45:07 +02001194
Willy Tarreau9bd0d742011-07-20 00:17:39 +02001195 session_init_srv_conn(s);
Emeric Brun2b920a12010-09-23 18:30:22 +02001196 s->pend_pos = NULL;
1197
1198 /* init store persistence */
1199 s->store_count = 0;
Willy Tarreaud5ca9ab2013-05-28 17:40:25 +02001200 memset(s->stkctr, 0, sizeof(s->stkctr));
Emeric Brun2b920a12010-09-23 18:30:22 +02001201
1202 /* FIXME: the logs are horribly complicated now, because they are
Willy Tarreauae727bf2013-10-01 17:06:10 +02001203 * defined in <p>, <p>, and later <be> and <be>. We still initialize
1204 * a few of them to help troubleshooting (eg: show sess shows them).
Emeric Brun2b920a12010-09-23 18:30:22 +02001205 */
1206
1207 s->logs.logwait = 0;
Willy Tarreauabcd5142013-06-11 17:18:02 +02001208 s->logs.level = 0;
Willy Tarreauae727bf2013-10-01 17:06:10 +02001209 s->logs.accept_date = date; /* user-visible date for logging */
1210 s->logs.tv_accept = now; /* corrected date for internal use */
Emeric Brun2b920a12010-09-23 18:30:22 +02001211 s->do_log = NULL;
1212
1213 /* default error reporting function, may be changed by analysers */
1214 s->srv_error = default_srv_error;
1215
Emeric Brun2b920a12010-09-23 18:30:22 +02001216 s->uniq_id = 0;
Willy Tarreaubd833142012-05-08 15:51:44 +02001217 s->unique_id = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001218
1219 txn = &s->txn;
1220 /* Those variables will be checked and freed if non-NULL in
1221 * session.c:session_free(). It is important that they are
1222 * properly initialized.
1223 */
1224 txn->sessid = NULL;
1225 txn->srv_cookie = NULL;
1226 txn->cli_cookie = NULL;
1227 txn->uri = NULL;
1228 txn->req.cap = NULL;
1229 txn->rsp.cap = NULL;
1230 txn->hdr_idx.v = NULL;
1231 txn->hdr_idx.size = txn->hdr_idx.used = 0;
1232
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001233 if ((s->req = pool_alloc2(pool2_channel)) == NULL)
Emeric Brun2b920a12010-09-23 18:30:22 +02001234 goto out_fail_req; /* no memory */
1235
Willy Tarreau9b28e032012-10-12 23:49:43 +02001236 if ((s->req->buf = pool_alloc2(pool2_buffer)) == NULL)
1237 goto out_fail_req_buf; /* no memory */
1238
Willy Tarreau19d14ef2012-10-29 16:51:55 +01001239 s->req->buf->size = trash.size;
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001240 channel_init(s->req);
Emeric Brun2b920a12010-09-23 18:30:22 +02001241 s->req->prod = &s->si[0];
1242 s->req->cons = &s->si[1];
1243 s->si[0].ib = s->si[1].ob = s->req;
1244
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001245 s->req->flags |= CF_READ_ATTACHED; /* the producer is already connected */
Emeric Brun2b920a12010-09-23 18:30:22 +02001246
1247 /* activate default analysers enabled for this listener */
1248 s->req->analysers = l->analysers;
1249
1250 /* note: this should not happen anymore since there's always at least the switching rules */
1251 if (!s->req->analysers) {
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001252 channel_auto_connect(s->req);/* don't wait to establish connection */
1253 channel_auto_close(s->req);/* let the producer forward close requests */
Emeric Brun2b920a12010-09-23 18:30:22 +02001254 }
1255
1256 s->req->rto = s->fe->timeout.client;
1257 s->req->wto = s->be->timeout.server;
1258
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001259 if ((s->rep = pool_alloc2(pool2_channel)) == NULL)
Emeric Brun2b920a12010-09-23 18:30:22 +02001260 goto out_fail_rep; /* no memory */
1261
Willy Tarreau9b28e032012-10-12 23:49:43 +02001262 if ((s->rep->buf = pool_alloc2(pool2_buffer)) == NULL)
1263 goto out_fail_rep_buf; /* no memory */
1264
Willy Tarreau19d14ef2012-10-29 16:51:55 +01001265 s->rep->buf->size = trash.size;
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001266 channel_init(s->rep);
Emeric Brun2b920a12010-09-23 18:30:22 +02001267 s->rep->prod = &s->si[1];
1268 s->rep->cons = &s->si[0];
1269 s->si[0].ob = s->si[1].ib = s->rep;
1270
1271 s->rep->rto = s->be->timeout.server;
1272 s->rep->wto = s->fe->timeout.client;
1273
1274 s->req->rex = TICK_ETERNITY;
1275 s->req->wex = TICK_ETERNITY;
1276 s->req->analyse_exp = TICK_ETERNITY;
1277 s->rep->rex = TICK_ETERNITY;
1278 s->rep->wex = TICK_ETERNITY;
1279 s->rep->analyse_exp = TICK_ETERNITY;
1280 t->expire = TICK_ETERNITY;
1281
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001282 s->rep->flags |= CF_READ_DONTWAIT;
Emeric Brun2b920a12010-09-23 18:30:22 +02001283 /* it is important not to call the wakeup function directly but to
1284 * pass through task_wakeup(), because this one knows how to apply
1285 * priorities to tasks.
1286 */
1287 task_wakeup(t, TASK_WOKEN_INIT);
1288
1289 l->nbconn++; /* warning! right now, it's up to the handler to decrease this */
1290 p->feconn++;/* beconn will be increased later */
1291 jobs++;
Willy Tarreau3c63fd82011-09-07 18:00:47 +02001292 if (!(s->listener->options & LI_O_UNLIMITED))
1293 actconn++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001294 totalconn++;
1295
1296 return s;
1297
1298 /* Error unrolling */
Willy Tarreau9b28e032012-10-12 23:49:43 +02001299 out_fail_rep_buf:
1300 pool_free2(pool2_channel, s->rep);
Emeric Brun2b920a12010-09-23 18:30:22 +02001301 out_fail_rep:
Willy Tarreau9b28e032012-10-12 23:49:43 +02001302 pool_free2(pool2_buffer, s->req->buf);
1303 out_fail_req_buf:
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001304 pool_free2(pool2_channel, s->req);
Emeric Brun2b920a12010-09-23 18:30:22 +02001305 out_fail_req:
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001306 conn_free(conn);
1307 out_fail_conn1:
Emeric Brun2b920a12010-09-23 18:30:22 +02001308 task_free(t);
1309 out_free_session:
1310 LIST_DEL(&s->list);
1311 pool_free2(pool2_session, s);
1312 out_close:
1313 return s;
1314}
1315
1316/*
1317 * Task processing function to manage re-connect and peer session
1318 * tasks wakeup on local update.
1319 */
Simon Horman96553772011-06-08 09:18:51 +09001320static struct task *process_peer_sync(struct task * task)
Emeric Brun2b920a12010-09-23 18:30:22 +02001321{
1322 struct shared_table *st = (struct shared_table *)task->context;
1323 struct peer_session *ps;
1324
1325 task->expire = TICK_ETERNITY;
1326
1327 if (!stopping) {
1328 /* Normal case (not soft stop)*/
1329 if (((st->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMLOCAL) &&
1330 (!nb_oldpids || tick_is_expired(st->resync_timeout, now_ms)) &&
1331 !(st->flags & SHTABLE_F_RESYNC_ASSIGN)) {
1332 /* Resync from local peer needed
1333 no peer was assigned for the lesson
1334 and no old local peer found
1335 or resync timeout expire */
1336
1337 /* flag no more resync from local, to try resync from remotes */
1338 st->flags |= SHTABLE_F_RESYNC_LOCAL;
1339
1340 /* reschedule a resync */
1341 st->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
1342 }
1343
1344 /* For each session */
1345 for (ps = st->sessions; ps; ps = ps->next) {
1346 /* For each remote peers */
1347 if (!ps->peer->local) {
1348 if (!ps->session) {
1349 /* no active session */
1350 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001351 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
1352 ((ps->statuscode == PEER_SESS_SC_CONNECTCODE ||
1353 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02001354 tick_is_expired(ps->reconnect, now_ms))) {
1355 /* connection never tried
1356 * or previous session established with success
1357 * or previous session failed during connection
1358 * and reconnection timer is expired */
1359
1360 /* retry a connect */
1361 ps->session = peer_session_create(ps->peer, ps);
1362 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001363 else if (ps->statuscode == PEER_SESS_SC_CONNECTCODE ||
1364 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001365 /* If previous session failed during connection
1366 * but reconnection timer is not expired */
1367
1368 /* reschedule task for reconnect */
1369 task->expire = tick_first(task->expire, ps->reconnect);
1370 }
1371 /* else do nothing */
1372 } /* !ps->session */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001373 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001374 /* current session is active and established */
1375 if (((st->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMREMOTE) &&
1376 !(st->flags & SHTABLE_F_RESYNC_ASSIGN) &&
1377 !(ps->flags & PEER_F_LEARN_NOTUP2DATE)) {
1378 /* Resync from a remote is needed
1379 * and no peer was assigned for lesson
1380 * and current peer may be up2date */
1381
1382 /* assign peer for the lesson */
1383 ps->flags |= PEER_F_LEARN_ASSIGN;
1384 st->flags |= SHTABLE_F_RESYNC_ASSIGN;
1385
1386 /* awake peer session task to handle a request of resync */
1387 task_wakeup(ps->session->task, TASK_WOKEN_MSG);
1388 }
1389 else if ((int)(ps->pushed - ps->table->table->localupdate) < 0) {
1390 /* awake peer session task to push local updates */
1391 task_wakeup(ps->session->task, TASK_WOKEN_MSG);
1392 }
1393 /* else do nothing */
1394 } /* SUCCESSCODE */
1395 } /* !ps->peer->local */
1396 } /* for */
1397
1398 /* Resync from remotes expired: consider resync is finished */
1399 if (((st->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMREMOTE) &&
1400 !(st->flags & SHTABLE_F_RESYNC_ASSIGN) &&
1401 tick_is_expired(st->resync_timeout, now_ms)) {
1402 /* Resync from remote peer needed
1403 * no peer was assigned for the lesson
1404 * and resync timeout expire */
1405
1406 /* flag no more resync from remote, consider resync is finished */
1407 st->flags |= SHTABLE_F_RESYNC_REMOTE;
1408 }
1409
1410 if ((st->flags & SHTABLE_RESYNC_STATEMASK) != SHTABLE_RESYNC_FINISHED) {
1411 /* Resync not finished*/
1412 /* reschedule task to resync timeout, to ended resync if needed */
1413 task->expire = tick_first(task->expire, st->resync_timeout);
1414 }
1415 } /* !stopping */
1416 else {
1417 /* soft stop case */
1418 if (task->state & TASK_WOKEN_SIGNAL) {
1419 /* We've just recieved the signal */
1420 if (!(st->flags & SHTABLE_F_DONOTSTOP)) {
1421 /* add DO NOT STOP flag if not present */
1422 jobs++;
1423 st->flags |= SHTABLE_F_DONOTSTOP;
Willy Tarreau3a925c12013-09-04 17:54:01 +02001424 st->table->syncing++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001425 }
1426
1427 /* disconnect all connected peers */
1428 for (ps = st->sessions; ps; ps = ps->next) {
1429 if (ps->session) {
1430 peer_session_forceshutdown(ps->session);
1431 ps->session = NULL;
1432 }
1433 }
1434 }
1435 ps = st->local_session;
1436
1437 if (ps->flags & PEER_F_TEACH_COMPLETE) {
1438 if (st->flags & SHTABLE_F_DONOTSTOP) {
1439 /* resync of new process was complete, current process can die now */
1440 jobs--;
1441 st->flags &= ~SHTABLE_F_DONOTSTOP;
Willy Tarreau3a925c12013-09-04 17:54:01 +02001442 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02001443 }
1444 }
1445 else if (!ps->session) {
1446 /* If session is not active */
1447 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001448 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
1449 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE ||
1450 ps->statuscode == PEER_SESS_SC_TRYAGAIN) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001451 /* connection never tried
1452 * or previous session was successfully established
1453 * or previous session tcp connect success but init state incomplete
1454 * or during previous connect, peer replies a try again statuscode */
1455
1456 /* connect to the peer */
1457 ps->session = peer_session_create(ps->peer, ps);
1458 }
1459 else {
1460 /* Other error cases */
1461 if (st->flags & SHTABLE_F_DONOTSTOP) {
1462 /* unable to resync new process, current process can die now */
1463 jobs--;
1464 st->flags &= ~SHTABLE_F_DONOTSTOP;
Willy Tarreau3a925c12013-09-04 17:54:01 +02001465 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02001466 }
1467 }
1468 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001469 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE &&
Emeric Brun2b920a12010-09-23 18:30:22 +02001470 (int)(ps->pushed - ps->table->table->localupdate) < 0) {
1471 /* current session active and established
1472 awake session to push remaining local updates */
1473 task_wakeup(ps->session->task, TASK_WOKEN_MSG);
1474 }
1475 } /* stopping */
1476 /* Wakeup for re-connect */
1477 return task;
1478}
1479
1480/*
1481 * Function used to register a table for sync on a group of peers
1482 *
1483 */
1484void peers_register_table(struct peers *peers, struct stktable *table)
1485{
1486 struct shared_table *st;
1487 struct peer * curpeer;
1488 struct peer_session *ps;
Willy Tarreau4348fad2012-09-20 16:48:07 +02001489 struct listener *listener;
Emeric Brun2b920a12010-09-23 18:30:22 +02001490
1491 st = (struct shared_table *)calloc(1,sizeof(struct shared_table));
1492 st->table = table;
1493 st->next = peers->tables;
1494 st->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
1495 peers->tables = st;
1496
1497 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
1498 ps = (struct peer_session *)calloc(1,sizeof(struct peer_session));
1499 ps->table = st;
1500 ps->peer = curpeer;
1501 if (curpeer->local)
1502 st->local_session = ps;
1503 ps->next = st->sessions;
1504 ps->reconnect = now_ms;
1505 st->sessions = ps;
1506 peers->peers_fe->maxconn += 3;
1507 }
1508
Willy Tarreau4348fad2012-09-20 16:48:07 +02001509 list_for_each_entry(listener, &peers->peers_fe->conf.listeners, by_fe)
1510 listener->maxconn = peers->peers_fe->maxconn;
Emeric Brun2b920a12010-09-23 18:30:22 +02001511 st->sync_task = task_new();
1512 st->sync_task->process = process_peer_sync;
1513 st->sync_task->expire = TICK_ETERNITY;
1514 st->sync_task->context = (void *)st;
1515 table->sync_task =st->sync_task;
1516 signal_register_task(0, table->sync_task, 0);
1517 task_wakeup(st->sync_task, TASK_WOKEN_INIT);
1518}
1519