blob: 0564d3dcfffae2a2186bb122382df5aab47b856d [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) {
Cyril Bonté9a60ff92014-02-16 01:07:07 +0100679 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200680 if (reql <= 0) /* closed or EOL not found */
681 goto incomplete;
Cyril Bonté9a60ff92014-02-16 01:07:07 +0100682 newts = stksess_new(ps->table->table, NULL);
683 if (newts) {
684 netinteger = ntohl(netinteger);
685 memcpy(newts->key.key, &netinteger, sizeof(netinteger));
686 }
Emeric Brun2b920a12010-09-23 18:30:22 +0200687 totl += reql;
Emeric Brun2b920a12010-09-23 18:30:22 +0200688 }
689 else {
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200690 /* type ip or binary */
691 newts = stksess_new(ps->table->table, NULL);
692 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 +0200693 if (reql <= 0) /* closed or EOL not found */
694 goto incomplete;
Willy Tarreau72d6c162013-04-11 16:14:13 +0200695 totl += reql;
Emeric Brun2b920a12010-09-23 18:30:22 +0200696 }
697
698 /* read server id */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200699 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200700 if (reql <= 0) /* closed or EOL not found */
701 goto incomplete;
702
Emeric Brun2b920a12010-09-23 18:30:22 +0200703 totl += reql;
704 srvid = ntohl(netinteger);
705
706 /* update entry */
Emeric Brun2b920a12010-09-23 18:30:22 +0200707 if (newts) {
708 /* lookup for existing entry */
709 ts = stktable_lookup(ps->table->table, newts);
710 if (ts) {
711 /* the entry already exist, we can free ours */
712 stktable_touch(ps->table->table, ts, 0);
713 stksess_free(ps->table->table, newts);
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200714 newts = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +0200715 }
716 else {
717 struct eb32_node *eb;
718
719 /* create new entry */
720 ts = stktable_store(ps->table->table, newts, 0);
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200721 newts = NULL; /* don't reuse it */
722
Emeric Brun2b920a12010-09-23 18:30:22 +0200723 ts->upd.key= (++ps->table->table->update)+(2^31);
724 eb = eb32_insert(&ps->table->table->updates, &ts->upd);
725 if (eb != &ts->upd) {
726 eb32_delete(eb);
727 eb32_insert(&ps->table->table->updates, &ts->upd);
728 }
729 }
730
731 /* update entry */
732 if (srvid && stktable_data_ptr(ps->table->table, ts, STKTABLE_DT_SERVER_ID))
733 stktable_data_cast(stktable_data_ptr(ps->table->table, ts, STKTABLE_DT_SERVER_ID), server_id) = srvid;
734 ps->pushack = pushack;
735 }
736
737 }
738 else if (c == 'R') {
739 /* Reset message: remote need resync */
740
741 /* reinit counters for a resync */
742 ps->lastpush = 0;
743 ps->teaching_origin = ps->pushed = ps->table->table->update;
744
745 /* reset teaching flags to 0 */
746 ps->flags &= PEER_TEACH_RESET;
747
748 /* flag to start to teach lesson */
749 ps->flags |= PEER_F_TEACH_PROCESS;
750 }
751 else if (c == 'F') {
752 /* Finish message, all known updates have been pushed by remote */
753 /* and remote is up to date */
754
755 /* If resync is in progress with remote peer */
756 if (ps->flags & PEER_F_LEARN_ASSIGN) {
757
758 /* unassign current peer for learning */
759 ps->flags &= ~PEER_F_LEARN_ASSIGN;
760 ps->table->flags &= ~(SHTABLE_F_RESYNC_ASSIGN|SHTABLE_F_RESYNC_PROCESS);
761
762 /* Consider table is now up2date, resync resync no more needed from local neither remote */
763 ps->table->flags |= (SHTABLE_F_RESYNC_LOCAL|SHTABLE_F_RESYNC_REMOTE);
764 }
765 /* Increase confirm counter to launch a confirm message */
766 ps->confirm++;
767 }
768 else if (c == 'c') {
769 /* confirm message, remote peer is now up to date with us */
770
771 /* If stopping state */
772 if (stopping) {
773 /* Close session, push resync no more needed */
774 ps->flags |= PEER_F_TEACH_COMPLETE;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100775 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200776 goto switchstate;
777 }
778
779 /* reset teaching flags to 0 */
780 ps->flags &= PEER_TEACH_RESET;
781 }
782 else if (c == 'C') {
783 /* Continue message, all known updates have been pushed by remote */
784 /* but remote is not up to date */
785
786 /* If resync is in progress with current peer */
787 if (ps->flags & PEER_F_LEARN_ASSIGN) {
788
789 /* unassign current peer */
790 ps->flags &= ~PEER_F_LEARN_ASSIGN;
791 ps->table->flags &= ~(SHTABLE_F_RESYNC_ASSIGN|SHTABLE_F_RESYNC_PROCESS);
792
793 /* flag current peer is not up 2 date to try from an other */
794 ps->flags |= PEER_F_LEARN_NOTUP2DATE;
795
796 /* reschedule a resync */
797 ps->table->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
798 task_wakeup(ps->table->sync_task, TASK_WOKEN_MSG);
799 }
800 ps->confirm++;
801 }
802 else if (c == 'A') {
803 /* ack message */
804 uint32_t netinteger;
805
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200806 reql = bo_getblk(si->ob, (char *)&netinteger, sizeof(netinteger), totl);
Willy Tarreau72d6c162013-04-11 16:14:13 +0200807 if (reql <= 0) /* closed or EOL not found */
808 goto incomplete;
809
Emeric Brun2b920a12010-09-23 18:30:22 +0200810 totl += reql;
811
812 /* Consider remote is up to date with "acked" version */
813 ps->update = ntohl(netinteger);
814 }
815 else {
816 /* Unknown message */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100817 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200818 goto switchstate;
819 }
820
821 /* skip consumed message */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200822 bo_skip(si->ob, totl);
Emeric Brun2b920a12010-09-23 18:30:22 +0200823
824 /* loop on that state to peek next message */
Willy Tarreau72d6c162013-04-11 16:14:13 +0200825 goto switchstate;
826
Emeric Brun2b920a12010-09-23 18:30:22 +0200827incomplete:
Willy Tarreau9d9179b2013-04-11 16:56:44 +0200828 /* we get here when a bo_getblk() returns <= 0 in reql */
829
830 /* first, we may have to release newts */
831 if (newts) {
832 stksess_free(ps->table->table, newts);
833 newts = NULL;
834 }
835
Willy Tarreau72d6c162013-04-11 16:14:13 +0200836 if (reql < 0) {
837 /* there was an error */
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100838 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau72d6c162013-04-11 16:14:13 +0200839 goto switchstate;
840 }
841
Emeric Brun2b920a12010-09-23 18:30:22 +0200842 /* Nothing to read, now we start to write */
843
844 /* Confirm finished or partial messages */
845 while (ps->confirm) {
846 /* There is a confirm messages to send */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200847 repl = bi_putchr(si->ib, 'c');
Emeric Brun2b920a12010-09-23 18:30:22 +0200848 if (repl <= 0) {
849 /* no more write possible */
850 if (repl == -1)
851 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100852 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200853 goto switchstate;
854 }
855 ps->confirm--;
856 }
857
858 /* Need to request a resync */
859 if ((ps->flags & PEER_F_LEARN_ASSIGN) &&
860 (ps->table->flags & SHTABLE_F_RESYNC_ASSIGN) &&
861 !(ps->table->flags & SHTABLE_F_RESYNC_PROCESS)) {
862 /* Current peer was elected to request a resync */
863
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200864 repl = bi_putchr(si->ib, 'R');
Emeric Brun2b920a12010-09-23 18:30:22 +0200865 if (repl <= 0) {
866 /* no more write possible */
867 if (repl == -1)
868 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100869 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200870 goto switchstate;
871 }
872 ps->table->flags |= SHTABLE_F_RESYNC_PROCESS;
873 }
874
875 /* It remains some updates to ack */
876 if (ps->pushack != ps->lastack) {
877 uint32_t netinteger;
878
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100879 trash.str[0] = 'A';
Emeric Brun2b920a12010-09-23 18:30:22 +0200880 netinteger = htonl(ps->pushack);
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100881 memcpy(&trash.str[1], &netinteger, sizeof(netinteger));
Emeric Brun2b920a12010-09-23 18:30:22 +0200882
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100883 repl = bi_putblk(si->ib, trash.str, 1+sizeof(netinteger));
Emeric Brun2b920a12010-09-23 18:30:22 +0200884 if (repl <= 0) {
885 /* no more write possible */
886 if (repl == -1)
887 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100888 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200889 goto switchstate;
890 }
891 ps->lastack = ps->pushack;
892 }
893
894 if (ps->flags & PEER_F_TEACH_PROCESS) {
895 /* current peer was requested for a lesson */
896
897 if (!(ps->flags & PEER_F_TEACH_STAGE1)) {
898 /* lesson stage 1 not complete */
899 struct eb32_node *eb;
900
901 eb = eb32_lookup_ge(&ps->table->table->updates, ps->pushed+1);
902 while (1) {
903 int msglen;
904 struct stksess *ts;
905
906 if (!eb) {
907 /* flag lesson stage1 complete */
908 ps->flags |= PEER_F_TEACH_STAGE1;
909 eb = eb32_first(&ps->table->table->updates);
910 if (eb)
911 ps->pushed = eb->key - 1;
912 break;
913 }
914
915 ts = eb32_entry(eb, struct stksess, upd);
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100916 msglen = peer_prepare_datamsg(ts, ps, trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200917 if (msglen) {
918 /* message to buffer */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100919 repl = bi_putblk(si->ib, trash.str, msglen);
Emeric Brun2b920a12010-09-23 18:30:22 +0200920 if (repl <= 0) {
921 /* no more write possible */
922 if (repl == -1)
923 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100924 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200925 goto switchstate;
926 }
927 ps->lastpush = ps->pushed = ts->upd.key;
928 }
929 eb = eb32_next(eb);
930 }
931 } /* !TEACH_STAGE1 */
932
933 if (!(ps->flags & PEER_F_TEACH_STAGE2)) {
934 /* lesson stage 2 not complete */
935 struct eb32_node *eb;
936
937 eb = eb32_lookup_ge(&ps->table->table->updates, ps->pushed+1);
938 while (1) {
939 int msglen;
940 struct stksess *ts;
941
942 if (!eb || eb->key > ps->teaching_origin) {
943 /* flag lesson stage1 complete */
944 ps->flags |= PEER_F_TEACH_STAGE2;
945 ps->pushed = ps->teaching_origin;
946 break;
947 }
948
949 ts = eb32_entry(eb, struct stksess, upd);
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100950 msglen = peer_prepare_datamsg(ts, ps, trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +0200951 if (msglen) {
952 /* message to buffer */
Willy Tarreau19d14ef2012-10-29 16:51:55 +0100953 repl = bi_putblk(si->ib, trash.str, msglen);
Emeric Brun2b920a12010-09-23 18:30:22 +0200954 if (repl <= 0) {
955 /* no more write possible */
956 if (repl == -1)
957 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100958 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200959 goto switchstate;
960 }
961 ps->lastpush = ps->pushed = ts->upd.key;
962 }
963 eb = eb32_next(eb);
964 }
965 } /* !TEACH_STAGE2 */
966
967 if (!(ps->flags & PEER_F_TEACH_FINISHED)) {
968 /* process final lesson message */
Willy Tarreau9dab5fc2012-05-07 11:56:55 +0200969 repl = bi_putchr(si->ib, ((ps->table->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FINISHED) ? 'F' : 'C');
Emeric Brun2b920a12010-09-23 18:30:22 +0200970 if (repl <= 0) {
971 /* no more write possible */
972 if (repl == -1)
973 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +0100974 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +0200975 goto switchstate;
976 }
977
978 /* flag finished message sent */
979 ps->flags |= PEER_F_TEACH_FINISHED;
980 } /* !TEACH_FINISHED */
981 } /* TEACH_PROCESS */
982
983 if (!(ps->flags & PEER_F_LEARN_ASSIGN) &&
984 (int)(ps->pushed - ps->table->table->localupdate) < 0) {
985 /* Push local updates, only if no learning in progress (to avoid ping-pong effects) */
986 struct eb32_node *eb;
987
988 eb = eb32_lookup_ge(&ps->table->table->updates, ps->pushed+1);
989 while (1) {
990 int msglen;
991 struct stksess *ts;
992
993 /* push local updates */
994 if (!eb) {
995 eb = eb32_first(&ps->table->table->updates);
996 if (!eb || ((int)(eb->key - ps->pushed) <= 0)) {
997 ps->pushed = ps->table->table->localupdate;
998 break;
999 }
1000 }
1001
1002 if ((int)(eb->key - ps->table->table->localupdate) > 0) {
1003 ps->pushed = ps->table->table->localupdate;
1004 break;
1005 }
1006
1007 ts = eb32_entry(eb, struct stksess, upd);
Willy Tarreau19d14ef2012-10-29 16:51:55 +01001008 msglen = peer_prepare_datamsg(ts, ps, trash.str, trash.size);
Emeric Brun2b920a12010-09-23 18:30:22 +02001009 if (msglen) {
1010 /* message to buffer */
Willy Tarreau19d14ef2012-10-29 16:51:55 +01001011 repl = bi_putblk(si->ib, trash.str, msglen);
Emeric Brun2b920a12010-09-23 18:30:22 +02001012 if (repl <= 0) {
1013 /* no more write possible */
1014 if (repl == -1)
1015 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001016 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02001017 goto switchstate;
1018 }
1019 ps->lastpush = ps->pushed = ts->upd.key;
1020 }
1021 eb = eb32_next(eb);
1022 }
1023 } /* ! LEARN_ASSIGN */
1024 /* noting more to do */
1025 goto out;
1026 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001027 case PEER_SESS_ST_EXIT:
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001028 repl = snprintf(trash.str, trash.size, "%d\n", appctx->st1);
Emeric Brun2b920a12010-09-23 18:30:22 +02001029
Willy Tarreau19d14ef2012-10-29 16:51:55 +01001030 if (bi_putblk(si->ib, trash.str, repl) == -1)
Emeric Brun2b920a12010-09-23 18:30:22 +02001031 goto out;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001032 appctx->st0 = PEER_SESS_ST_END;
Emeric Brun2b920a12010-09-23 18:30:22 +02001033 /* fall through */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001034 case PEER_SESS_ST_END: {
Willy Tarreau73b013b2012-05-21 16:31:45 +02001035 si_shutw(si);
1036 si_shutr(si);
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001037 si->ib->flags |= CF_READ_NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001038 goto quit;
1039 }
1040 }
1041 }
1042out:
Willy Tarreau73b013b2012-05-21 16:31:45 +02001043 si_update(si);
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001044 si->ob->flags |= CF_READ_DONTWAIT;
Emeric Brun2b920a12010-09-23 18:30:22 +02001045 /* we don't want to expire timeouts while we're processing requests */
1046 si->ib->rex = TICK_ETERNITY;
1047 si->ob->wex = TICK_ETERNITY;
1048quit:
1049 return;
1050}
1051
Willy Tarreaub24281b2011-02-13 13:16:36 +01001052static struct si_applet peer_applet = {
Willy Tarreau3fdb3662012-11-12 00:42:33 +01001053 .obj_type = OBJ_TYPE_APPLET,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001054 .name = "<PEER>", /* used for logging */
1055 .fct = peer_io_handler,
Aman Gupta9a13e842012-04-02 18:57:53 -07001056 .release = peer_session_release,
Willy Tarreaub24281b2011-02-13 13:16:36 +01001057};
Emeric Brun2b920a12010-09-23 18:30:22 +02001058
1059/*
1060 * Use this function to force a close of a peer session
1061 */
Simon Horman96553772011-06-08 09:18:51 +09001062static void peer_session_forceshutdown(struct session * session)
Emeric Brun2b920a12010-09-23 18:30:22 +02001063{
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001064 struct stream_interface *oldsi = NULL;
1065 struct appctx *appctx = NULL;
Willy Tarreau76a06b22015-05-20 10:39:04 +02001066 struct peer_session *ps;
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001067 int i;
Emeric Brun2b920a12010-09-23 18:30:22 +02001068
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001069 for (i = 0; i <= 1; i++) {
1070 appctx = objt_appctx(session->si[i].end);
1071 if (!appctx)
1072 continue;
1073 if (appctx->applet != &peer_applet)
1074 continue;
1075
1076 oldsi = &session->si[i];
1077 break;
Emeric Brun2b920a12010-09-23 18:30:22 +02001078 }
1079
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001080 if (!appctx)
1081 return;
1082
Willy Tarreau76a06b22015-05-20 10:39:04 +02001083 ps = (struct peer_session *)appctx->ctx.peers.ptr;
1084 /* we're killing a connection, we must apply a random delay before
1085 * retrying otherwise the other end will do the same and we can loop
1086 * for a while.
1087 */
1088 if (ps)
1089 ps->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + random() % 2000));
1090
Emeric Brun2b920a12010-09-23 18:30:22 +02001091 /* call release to reinit resync states if needed */
1092 peer_session_release(oldsi);
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001093 appctx->st0 = PEER_SESS_ST_END;
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001094 appctx->ctx.peers.ptr = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001095 task_wakeup(session->task, TASK_WOKEN_MSG);
1096}
1097
Willy Tarreaub363a1f2013-10-01 10:45:07 +02001098/* Finish a session accept() for a peer. It returns a negative value in case of
1099 * a critical failure which must cause the listener to be disabled, a positive
1100 * value in case of success, or zero if it is a success but the session must be
1101 * closed ASAP and ignored.
Emeric Brun2b920a12010-09-23 18:30:22 +02001102 */
1103int peer_accept(struct session *s)
1104{
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001105 s->target = &peer_applet.obj_type;
Willy Tarreau6d7f8f72013-12-01 12:54:55 +01001106 /* no need to initialize the applet, it will start with st0=st1 = 0 */
Emeric Brun2b920a12010-09-23 18:30:22 +02001107
1108 tv_zero(&s->logs.tv_request);
1109 s->logs.t_queue = 0;
1110 s->logs.t_connect = 0;
1111 s->logs.t_data = 0;
1112 s->logs.t_close = 0;
1113 s->logs.bytes_in = s->logs.bytes_out = 0;
1114 s->logs.prx_queue_size = 0;/* we get the number of pending conns before us */
1115 s->logs.srv_queue_size = 0; /* we will get this number soon */
1116
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001117 s->req->flags |= CF_READ_DONTWAIT; /* we plan to read small requests */
Emeric Brun2b920a12010-09-23 18:30:22 +02001118
1119 if (s->listener->timeout) {
1120 s->req->rto = *s->listener->timeout;
1121 s->rep->wto = *s->listener->timeout;
1122 }
1123 return 1;
1124}
1125
1126/*
Willy Tarreaubd55e312010-11-11 10:55:09 +01001127 * Create a new peer session in assigned state (connect will start automatically)
Emeric Brun2b920a12010-09-23 18:30:22 +02001128 */
Simon Horman96553772011-06-08 09:18:51 +09001129static struct session *peer_session_create(struct peer *peer, struct peer_session *ps)
Emeric Brun2b920a12010-09-23 18:30:22 +02001130{
Willy Tarreau4348fad2012-09-20 16:48:07 +02001131 struct listener *l = LIST_NEXT(&peer->peers->peers_fe->conf.listeners, struct listener *, by_fe);
Emeric Brun2b920a12010-09-23 18:30:22 +02001132 struct proxy *p = (struct proxy *)l->frontend; /* attached frontend */
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001133 struct appctx *appctx;
Emeric Brun2b920a12010-09-23 18:30:22 +02001134 struct session *s;
1135 struct http_txn *txn;
1136 struct task *t;
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001137 struct connection *conn;
Emeric Brun2b920a12010-09-23 18:30:22 +02001138
1139 if ((s = pool_alloc2(pool2_session)) == 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_close;
1142 }
1143
1144 LIST_ADDQ(&sessions, &s->list);
1145 LIST_INIT(&s->back_refs);
1146
1147 s->flags = SN_ASSIGNED|SN_ADDR_SET;
Emeric Brun2b920a12010-09-23 18:30:22 +02001148
1149 /* if this session comes from a known monitoring system, we want to ignore
1150 * it as soon as possible, which means closing it immediately for TCP.
1151 */
1152 if ((t = task_new()) == NULL) { /* disable this proxy for a while */
Godbach430f2912013-06-20 13:28:38 +08001153 Alert("out of memory in peer_session_create().\n");
Emeric Brun2b920a12010-09-23 18:30:22 +02001154 goto out_free_session;
1155 }
1156
1157 ps->reconnect = tick_add(now_ms, MS_TO_TICKS(5000));
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001158 ps->statuscode = PEER_SESS_SC_CONNECTCODE;
Emeric Brun2b920a12010-09-23 18:30:22 +02001159
1160 t->process = l->handler;
1161 t->context = s;
1162 t->nice = l->nice;
1163
Emeric Brun2b920a12010-09-23 18:30:22 +02001164 s->task = t;
1165 s->listener = l;
1166
1167 /* Note: initially, the session's backend points to the frontend.
1168 * This changes later when switching rules are executed or
1169 * when the default backend is assigned.
1170 */
1171 s->be = s->fe = p;
1172
1173 s->req = s->rep = NULL; /* will be allocated later */
1174
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001175 si_reset(&s->si[0], t);
1176 si_set_state(&s->si[0], SI_ST_EST);
1177
Emeric Brun2b920a12010-09-23 18:30:22 +02001178 if (s->fe->options2 & PR_O2_INDEPSTR)
1179 s->si[0].flags |= SI_FL_INDEP_STR;
Emeric Brun2b920a12010-09-23 18:30:22 +02001180
Willy Tarreau1fbe1c92013-12-01 09:35:41 +01001181 appctx = stream_int_register_handler(&s->si[0], &peer_applet);
1182 if (!appctx)
1183 goto out_fail_conn1;
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001184 appctx->st0 = PEER_SESS_ST_CONNECT;
Willy Tarreau7b4b4992013-12-01 09:15:12 +01001185 appctx->ctx.peers.ptr = (void *)ps;
Emeric Brun2b920a12010-09-23 18:30:22 +02001186
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001187 si_reset(&s->si[1], t);
1188
1189 /* initiate an outgoing connection */
1190 si_set_state(&s->si[1], SI_ST_ASS);
Emeric Brun2b920a12010-09-23 18:30:22 +02001191 s->si[1].conn_retries = p->conn_retries;
Willy Tarreau3ed35ef2013-10-24 11:51:38 +02001192
Emeric Brun2b920a12010-09-23 18:30:22 +02001193 if (s->be->options2 & PR_O2_INDEPSTR)
1194 s->si[1].flags |= SI_FL_INDEP_STR;
1195
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001196 /* automatically prepare the stream interface to connect to the
Willy Tarreaub363a1f2013-10-01 10:45:07 +02001197 * pre-initialized connection in si->conn.
1198 */
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001199 if (unlikely((conn = conn_new()) == NULL))
1200 goto out_fail_conn1;
1201
1202 conn_prepare(conn, peer->proto, peer->xprt);
1203 si_attach_conn(&s->si[1], conn);
1204
1205 conn->target = s->target = &s->be->obj_type;
1206 memcpy(&conn->addr.to, &peer->addr, sizeof(conn->addr.to));
Willy Tarreaub363a1f2013-10-01 10:45:07 +02001207
Willy Tarreau9bd0d742011-07-20 00:17:39 +02001208 session_init_srv_conn(s);
Emeric Brun2b920a12010-09-23 18:30:22 +02001209 s->pend_pos = NULL;
1210
1211 /* init store persistence */
1212 s->store_count = 0;
Willy Tarreaud5ca9ab2013-05-28 17:40:25 +02001213 memset(s->stkctr, 0, sizeof(s->stkctr));
Emeric Brun2b920a12010-09-23 18:30:22 +02001214
1215 /* FIXME: the logs are horribly complicated now, because they are
Willy Tarreauae727bf2013-10-01 17:06:10 +02001216 * defined in <p>, <p>, and later <be> and <be>. We still initialize
1217 * a few of them to help troubleshooting (eg: show sess shows them).
Emeric Brun2b920a12010-09-23 18:30:22 +02001218 */
1219
1220 s->logs.logwait = 0;
Willy Tarreauabcd5142013-06-11 17:18:02 +02001221 s->logs.level = 0;
Willy Tarreauae727bf2013-10-01 17:06:10 +02001222 s->logs.accept_date = date; /* user-visible date for logging */
1223 s->logs.tv_accept = now; /* corrected date for internal use */
Emeric Brun2b920a12010-09-23 18:30:22 +02001224 s->do_log = NULL;
1225
1226 /* default error reporting function, may be changed by analysers */
1227 s->srv_error = default_srv_error;
1228
Emeric Brun2b920a12010-09-23 18:30:22 +02001229 s->uniq_id = 0;
Willy Tarreaubd833142012-05-08 15:51:44 +02001230 s->unique_id = NULL;
Emeric Brun2b920a12010-09-23 18:30:22 +02001231
1232 txn = &s->txn;
1233 /* Those variables will be checked and freed if non-NULL in
1234 * session.c:session_free(). It is important that they are
1235 * properly initialized.
1236 */
1237 txn->sessid = NULL;
1238 txn->srv_cookie = NULL;
1239 txn->cli_cookie = NULL;
1240 txn->uri = NULL;
1241 txn->req.cap = NULL;
1242 txn->rsp.cap = NULL;
1243 txn->hdr_idx.v = NULL;
1244 txn->hdr_idx.size = txn->hdr_idx.used = 0;
1245
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001246 if ((s->req = pool_alloc2(pool2_channel)) == NULL)
Emeric Brun2b920a12010-09-23 18:30:22 +02001247 goto out_fail_req; /* no memory */
1248
Willy Tarreau9b28e032012-10-12 23:49:43 +02001249 if ((s->req->buf = pool_alloc2(pool2_buffer)) == NULL)
1250 goto out_fail_req_buf; /* no memory */
1251
Willy Tarreau5ccff3a2014-11-24 10:47:35 +01001252 s->req->buf->size = global.tune.bufsize;
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001253 channel_init(s->req);
Emeric Brun2b920a12010-09-23 18:30:22 +02001254 s->req->prod = &s->si[0];
1255 s->req->cons = &s->si[1];
1256 s->si[0].ib = s->si[1].ob = s->req;
1257
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001258 s->req->flags |= CF_READ_ATTACHED; /* the producer is already connected */
Emeric Brun2b920a12010-09-23 18:30:22 +02001259
1260 /* activate default analysers enabled for this listener */
1261 s->req->analysers = l->analysers;
1262
1263 /* note: this should not happen anymore since there's always at least the switching rules */
1264 if (!s->req->analysers) {
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001265 channel_auto_connect(s->req);/* don't wait to establish connection */
1266 channel_auto_close(s->req);/* let the producer forward close requests */
Emeric Brun2b920a12010-09-23 18:30:22 +02001267 }
1268
1269 s->req->rto = s->fe->timeout.client;
1270 s->req->wto = s->be->timeout.server;
1271
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001272 if ((s->rep = pool_alloc2(pool2_channel)) == NULL)
Emeric Brun2b920a12010-09-23 18:30:22 +02001273 goto out_fail_rep; /* no memory */
1274
Willy Tarreau9b28e032012-10-12 23:49:43 +02001275 if ((s->rep->buf = pool_alloc2(pool2_buffer)) == NULL)
1276 goto out_fail_rep_buf; /* no memory */
1277
Willy Tarreau5ccff3a2014-11-24 10:47:35 +01001278 s->rep->buf->size = global.tune.bufsize;
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001279 channel_init(s->rep);
Emeric Brun2b920a12010-09-23 18:30:22 +02001280 s->rep->prod = &s->si[1];
1281 s->rep->cons = &s->si[0];
1282 s->si[0].ob = s->si[1].ib = s->rep;
1283
1284 s->rep->rto = s->be->timeout.server;
1285 s->rep->wto = s->fe->timeout.client;
1286
1287 s->req->rex = TICK_ETERNITY;
1288 s->req->wex = TICK_ETERNITY;
1289 s->req->analyse_exp = TICK_ETERNITY;
1290 s->rep->rex = TICK_ETERNITY;
1291 s->rep->wex = TICK_ETERNITY;
1292 s->rep->analyse_exp = TICK_ETERNITY;
1293 t->expire = TICK_ETERNITY;
1294
Willy Tarreau03cdb7c2012-08-27 23:14:58 +02001295 s->rep->flags |= CF_READ_DONTWAIT;
Emeric Brun2b920a12010-09-23 18:30:22 +02001296 /* it is important not to call the wakeup function directly but to
1297 * pass through task_wakeup(), because this one knows how to apply
1298 * priorities to tasks.
1299 */
1300 task_wakeup(t, TASK_WOKEN_INIT);
1301
1302 l->nbconn++; /* warning! right now, it's up to the handler to decrease this */
1303 p->feconn++;/* beconn will be increased later */
1304 jobs++;
Willy Tarreau3c63fd82011-09-07 18:00:47 +02001305 if (!(s->listener->options & LI_O_UNLIMITED))
1306 actconn++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001307 totalconn++;
1308
1309 return s;
1310
1311 /* Error unrolling */
Willy Tarreau9b28e032012-10-12 23:49:43 +02001312 out_fail_rep_buf:
1313 pool_free2(pool2_channel, s->rep);
Emeric Brun2b920a12010-09-23 18:30:22 +02001314 out_fail_rep:
Willy Tarreau9b28e032012-10-12 23:49:43 +02001315 pool_free2(pool2_buffer, s->req->buf);
1316 out_fail_req_buf:
Willy Tarreau8263d2b2012-08-28 00:06:31 +02001317 pool_free2(pool2_channel, s->req);
Emeric Brun2b920a12010-09-23 18:30:22 +02001318 out_fail_req:
Willy Tarreau32e3c6a2013-10-11 19:34:20 +02001319 conn_free(conn);
1320 out_fail_conn1:
Emeric Brun2b920a12010-09-23 18:30:22 +02001321 task_free(t);
1322 out_free_session:
1323 LIST_DEL(&s->list);
1324 pool_free2(pool2_session, s);
1325 out_close:
1326 return s;
1327}
1328
1329/*
1330 * Task processing function to manage re-connect and peer session
1331 * tasks wakeup on local update.
1332 */
Simon Horman96553772011-06-08 09:18:51 +09001333static struct task *process_peer_sync(struct task * task)
Emeric Brun2b920a12010-09-23 18:30:22 +02001334{
1335 struct shared_table *st = (struct shared_table *)task->context;
1336 struct peer_session *ps;
1337
1338 task->expire = TICK_ETERNITY;
1339
Willy Tarreauaa070682015-05-01 18:32:13 +02001340 if (!st->sessions->peer->peers->peers_fe) {
1341 /* this one was never started, kill it */
1342 signal_unregister_handler(st->sighandler);
1343 st->table->sync_task = NULL;
1344 task_delete(st->sync_task);
1345 task_free(st->sync_task);
1346 return NULL;
1347 }
1348
Emeric Brun2b920a12010-09-23 18:30:22 +02001349 if (!stopping) {
1350 /* Normal case (not soft stop)*/
1351 if (((st->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMLOCAL) &&
1352 (!nb_oldpids || tick_is_expired(st->resync_timeout, now_ms)) &&
1353 !(st->flags & SHTABLE_F_RESYNC_ASSIGN)) {
1354 /* Resync from local peer needed
1355 no peer was assigned for the lesson
1356 and no old local peer found
1357 or resync timeout expire */
1358
1359 /* flag no more resync from local, to try resync from remotes */
1360 st->flags |= SHTABLE_F_RESYNC_LOCAL;
1361
1362 /* reschedule a resync */
1363 st->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
1364 }
1365
1366 /* For each session */
1367 for (ps = st->sessions; ps; ps = ps->next) {
1368 /* For each remote peers */
1369 if (!ps->peer->local) {
1370 if (!ps->session) {
1371 /* no active session */
1372 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001373 ((ps->statuscode == PEER_SESS_SC_CONNECTCODE ||
Willy Tarreau76a06b22015-05-20 10:39:04 +02001374 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001375 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE) &&
Emeric Brun2b920a12010-09-23 18:30:22 +02001376 tick_is_expired(ps->reconnect, now_ms))) {
1377 /* connection never tried
1378 * or previous session established with success
1379 * or previous session failed during connection
1380 * and reconnection timer is expired */
1381
1382 /* retry a connect */
1383 ps->session = peer_session_create(ps->peer, ps);
1384 }
Willy Tarreau76a06b22015-05-20 10:39:04 +02001385 else if (!tick_is_expired(ps->reconnect, now_ms)) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001386 /* If previous session failed during connection
1387 * but reconnection timer is not expired */
1388
1389 /* reschedule task for reconnect */
1390 task->expire = tick_first(task->expire, ps->reconnect);
1391 }
1392 /* else do nothing */
1393 } /* !ps->session */
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001394 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001395 /* current session is active and established */
1396 if (((st->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMREMOTE) &&
1397 !(st->flags & SHTABLE_F_RESYNC_ASSIGN) &&
1398 !(ps->flags & PEER_F_LEARN_NOTUP2DATE)) {
1399 /* Resync from a remote is needed
1400 * and no peer was assigned for lesson
1401 * and current peer may be up2date */
1402
1403 /* assign peer for the lesson */
1404 ps->flags |= PEER_F_LEARN_ASSIGN;
1405 st->flags |= SHTABLE_F_RESYNC_ASSIGN;
1406
1407 /* awake peer session task to handle a request of resync */
1408 task_wakeup(ps->session->task, TASK_WOKEN_MSG);
1409 }
1410 else if ((int)(ps->pushed - ps->table->table->localupdate) < 0) {
1411 /* awake peer session task to push local updates */
1412 task_wakeup(ps->session->task, TASK_WOKEN_MSG);
1413 }
1414 /* else do nothing */
1415 } /* SUCCESSCODE */
1416 } /* !ps->peer->local */
1417 } /* for */
1418
1419 /* Resync from remotes expired: consider resync is finished */
1420 if (((st->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMREMOTE) &&
1421 !(st->flags & SHTABLE_F_RESYNC_ASSIGN) &&
1422 tick_is_expired(st->resync_timeout, now_ms)) {
1423 /* Resync from remote peer needed
1424 * no peer was assigned for the lesson
1425 * and resync timeout expire */
1426
1427 /* flag no more resync from remote, consider resync is finished */
1428 st->flags |= SHTABLE_F_RESYNC_REMOTE;
1429 }
1430
1431 if ((st->flags & SHTABLE_RESYNC_STATEMASK) != SHTABLE_RESYNC_FINISHED) {
1432 /* Resync not finished*/
1433 /* reschedule task to resync timeout, to ended resync if needed */
1434 task->expire = tick_first(task->expire, st->resync_timeout);
1435 }
1436 } /* !stopping */
1437 else {
1438 /* soft stop case */
1439 if (task->state & TASK_WOKEN_SIGNAL) {
1440 /* We've just recieved the signal */
1441 if (!(st->flags & SHTABLE_F_DONOTSTOP)) {
1442 /* add DO NOT STOP flag if not present */
1443 jobs++;
1444 st->flags |= SHTABLE_F_DONOTSTOP;
Willy Tarreau3a925c12013-09-04 17:54:01 +02001445 st->table->syncing++;
Emeric Brun2b920a12010-09-23 18:30:22 +02001446 }
1447
1448 /* disconnect all connected peers */
1449 for (ps = st->sessions; ps; ps = ps->next) {
1450 if (ps->session) {
1451 peer_session_forceshutdown(ps->session);
1452 ps->session = NULL;
1453 }
1454 }
1455 }
1456 ps = st->local_session;
1457
1458 if (ps->flags & PEER_F_TEACH_COMPLETE) {
1459 if (st->flags & SHTABLE_F_DONOTSTOP) {
1460 /* resync of new process was complete, current process can die now */
1461 jobs--;
1462 st->flags &= ~SHTABLE_F_DONOTSTOP;
Willy Tarreau3a925c12013-09-04 17:54:01 +02001463 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02001464 }
1465 }
1466 else if (!ps->session) {
1467 /* If session is not active */
1468 if (ps->statuscode == 0 ||
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001469 ps->statuscode == PEER_SESS_SC_SUCCESSCODE ||
1470 ps->statuscode == PEER_SESS_SC_CONNECTEDCODE ||
1471 ps->statuscode == PEER_SESS_SC_TRYAGAIN) {
Emeric Brun2b920a12010-09-23 18:30:22 +02001472 /* connection never tried
1473 * or previous session was successfully established
1474 * or previous session tcp connect success but init state incomplete
1475 * or during previous connect, peer replies a try again statuscode */
1476
1477 /* connect to the peer */
1478 ps->session = peer_session_create(ps->peer, ps);
1479 }
1480 else {
1481 /* Other error cases */
1482 if (st->flags & SHTABLE_F_DONOTSTOP) {
1483 /* unable to resync new process, current process can die now */
1484 jobs--;
1485 st->flags &= ~SHTABLE_F_DONOTSTOP;
Willy Tarreau3a925c12013-09-04 17:54:01 +02001486 st->table->syncing--;
Emeric Brun2b920a12010-09-23 18:30:22 +02001487 }
1488 }
1489 }
Willy Tarreaue4d927a2013-12-01 12:47:35 +01001490 else if (ps->statuscode == PEER_SESS_SC_SUCCESSCODE &&
Emeric Brun2b920a12010-09-23 18:30:22 +02001491 (int)(ps->pushed - ps->table->table->localupdate) < 0) {
1492 /* current session active and established
1493 awake session to push remaining local updates */
1494 task_wakeup(ps->session->task, TASK_WOKEN_MSG);
1495 }
1496 } /* stopping */
1497 /* Wakeup for re-connect */
1498 return task;
1499}
1500
1501/*
1502 * Function used to register a table for sync on a group of peers
1503 *
1504 */
1505void peers_register_table(struct peers *peers, struct stktable *table)
1506{
1507 struct shared_table *st;
1508 struct peer * curpeer;
1509 struct peer_session *ps;
Willy Tarreau4348fad2012-09-20 16:48:07 +02001510 struct listener *listener;
Emeric Brun2b920a12010-09-23 18:30:22 +02001511
1512 st = (struct shared_table *)calloc(1,sizeof(struct shared_table));
1513 st->table = table;
1514 st->next = peers->tables;
1515 st->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
1516 peers->tables = st;
1517
1518 for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {
1519 ps = (struct peer_session *)calloc(1,sizeof(struct peer_session));
1520 ps->table = st;
1521 ps->peer = curpeer;
1522 if (curpeer->local)
1523 st->local_session = ps;
1524 ps->next = st->sessions;
1525 ps->reconnect = now_ms;
1526 st->sessions = ps;
1527 peers->peers_fe->maxconn += 3;
1528 }
1529
Willy Tarreau4348fad2012-09-20 16:48:07 +02001530 list_for_each_entry(listener, &peers->peers_fe->conf.listeners, by_fe)
1531 listener->maxconn = peers->peers_fe->maxconn;
Emeric Brun2b920a12010-09-23 18:30:22 +02001532 st->sync_task = task_new();
1533 st->sync_task->process = process_peer_sync;
1534 st->sync_task->expire = TICK_ETERNITY;
1535 st->sync_task->context = (void *)st;
Willy Tarreaue0603ae2015-05-01 18:29:09 +02001536 table->sync_task = st->sync_task;
1537 st->sighandler = signal_register_task(0, table->sync_task, 0);
Emeric Brun2b920a12010-09-23 18:30:22 +02001538 task_wakeup(st->sync_task, TASK_WOKEN_INIT);
1539}
1540