blob: 351313adea28bfb6ffe82e502af4a0dacded7458 [file] [log] [blame]
Christopher Faulet010fded2016-11-03 22:49:37 +01001/*
2 * A Random IP reputation service acting as a Stream Processing Offload Agent
3 *
4 * This is a very simple service that implement a "random" ip reputation
5 * service. It will return random scores for all checked IP addresses. It only
6 * shows you how to implement a ip reputation service or such kind of services
7 * using the SPOE.
8 *
9 * Copyright 2016 HAProxy Technologies, Christopher Faulet <cfaulet@haproxy.com>
10 *
11 * This program is free software; you can redistribute it and/or
12 * modify it under the terms of the GNU General Public License
13 * as published by the Free Software Foundation; either version
14 * 2 of the License, or (at your option) any later version.
15 *
16 */
Christopher Fauletf95b1112016-12-21 08:58:16 +010017#include <unistd.h>
Christopher Faulet010fded2016-11-03 22:49:37 +010018#include <stdlib.h>
19#include <string.h>
20#include <stdbool.h>
Christopher Fauletf95b1112016-12-21 08:58:16 +010021#include <errno.h>
22#include <stdio.h>
Christopher Faulet010fded2016-11-03 22:49:37 +010023#include <signal.h>
Christopher Faulet010fded2016-11-03 22:49:37 +010024#include <netinet/in.h>
Christopher Fauletf95b1112016-12-21 08:58:16 +010025#include <sys/socket.h>
26#include <err.h>
27#include <ctype.h>
28
29#include <pthread.h>
30
31#include <event2/util.h>
32#include <event2/event.h>
33#include <event2/event_struct.h>
34#include <event2/thread.h>
Christopher Faulet010fded2016-11-03 22:49:37 +010035
Christopher Fauletf95b1112016-12-21 08:58:16 +010036#include <common/mini-clist.h>
Christopher Faulet1f40b912017-02-17 09:32:19 +010037#include <common/chunk.h>
38
Christopher Faulet4ff3e572017-02-24 14:31:11 +010039#include <proto/spoe.h>
Christopher Fauletf95b1112016-12-21 08:58:16 +010040
41#define DEFAULT_PORT 12345
42#define CONNECTION_BACKLOG 10
43#define NUM_WORKERS 10
44#define MAX_FRAME_SIZE 16384
45#define SPOP_VERSION "1.0"
Christopher Faulet010fded2016-11-03 22:49:37 +010046
47#define SLEN(str) (sizeof(str)-1)
48
Christopher Fauletf95b1112016-12-21 08:58:16 +010049#define LOG(worker, fmt, args...) \
50 do { \
51 struct timeval now; \
52 \
53 gettimeofday(&now, NULL); \
54 fprintf(stderr, "%ld.%06ld [%02d] " fmt "\n", \
55 now.tv_sec, now.tv_usec, (worker)->id, ##args); \
Christopher Faulet010fded2016-11-03 22:49:37 +010056 } while (0)
57
Christopher Fauletf95b1112016-12-21 08:58:16 +010058#define DEBUG(x...) \
Christopher Faulet010fded2016-11-03 22:49:37 +010059 do { \
60 if (debug) \
61 LOG(x); \
62 } while (0)
63
Christopher Faulet010fded2016-11-03 22:49:37 +010064
Christopher Fauletf95b1112016-12-21 08:58:16 +010065enum spoa_state {
66 SPOA_ST_CONNECTING = 0,
67 SPOA_ST_PROCESSING,
68 SPOA_ST_DISCONNECTING,
69};
70
71enum spoa_frame_type {
72 SPOA_FRM_T_UNKNOWN = 0,
73 SPOA_FRM_T_HAPROXY,
74 SPOA_FRM_T_AGENT,
75};
Christopher Faulet010fded2016-11-03 22:49:37 +010076
Christopher Fauletf95b1112016-12-21 08:58:16 +010077struct spoe_engine {
78 char *id;
79
80 struct list processing_frames;
81 struct list outgoing_frames;
82
83 struct list clients;
84 struct list list;
85};
86
87struct spoe_frame {
88 enum spoa_frame_type type;
89 char *buf;
90 unsigned int offset;
91 unsigned int len;
92
93 unsigned int stream_id;
94 unsigned int frame_id;
Christopher Faulet85010352017-02-02 10:14:36 +010095 unsigned int flags;
96 bool hcheck; /* true is the CONNECT frame is a healthcheck */
97 bool fragmented; /* true if the frame is fragmented */
98 int ip_score; /* -1 if unset, else between 0 and 100 */
Christopher Fauletf95b1112016-12-21 08:58:16 +010099
100 struct event process_frame_event;
101 struct worker *worker;
102 struct spoe_engine *engine;
103 struct client *client;
104 struct list list;
105
Christopher Faulet85010352017-02-02 10:14:36 +0100106 char *frag_buf; /* used to accumulate payload of a fragmented frame */
107 unsigned int frag_len;
108
Christopher Fauletf95b1112016-12-21 08:58:16 +0100109 char data[0];
110};
111
112struct client {
113 int fd;
114 unsigned long id;
115 enum spoa_state state;
116
117 struct event read_frame_event;
118 struct event write_frame_event;
119
120 struct spoe_frame *incoming_frame;
121 struct spoe_frame *outgoing_frame;
122
123 struct list processing_frames;
124 struct list outgoing_frames;
125
126 unsigned int max_frame_size;
127 int status_code;
128
129 char *engine_id;
130 struct spoe_engine *engine;
131 bool pipelining;
132 bool async;
Christopher Faulet85010352017-02-02 10:14:36 +0100133 bool fragmentation;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100134
135 struct worker *worker;
136 struct list by_worker;
137 struct list by_engine;
Christopher Faulet010fded2016-11-03 22:49:37 +0100138};
139
140struct worker {
Christopher Fauletf95b1112016-12-21 08:58:16 +0100141 pthread_t thread;
142 int id;
143 struct event_base *base;
144 struct event *monitor_event;
145
146 struct list engines;
147
148 unsigned int nbclients;
149 struct list clients;
150
151 struct list frames;
Christopher Faulet010fded2016-11-03 22:49:37 +0100152};
153
Christopher Fauletf95b1112016-12-21 08:58:16 +0100154
Christopher Fauletf95b1112016-12-21 08:58:16 +0100155/* Globals */
156static struct worker *workers = NULL;
157static struct worker null_worker = { .id = 0 };
158static unsigned long clicount = 0;
159static int server_port = DEFAULT_PORT;
160static int num_workers = NUM_WORKERS;
161static unsigned int max_frame_size = MAX_FRAME_SIZE;
162struct timeval processing_delay = {0, 0};
163static bool debug = false;
164static bool pipelining = false;
165static bool async = false;
Christopher Faulet85010352017-02-02 10:14:36 +0100166static bool fragmentation = false;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100167
168
169static const char *spoe_frm_err_reasons[SPOE_FRM_ERRS] = {
Christopher Faulet85010352017-02-02 10:14:36 +0100170 [SPOE_FRM_ERR_NONE] = "normal",
171 [SPOE_FRM_ERR_IO] = "I/O error",
172 [SPOE_FRM_ERR_TOUT] = "a timeout occurred",
173 [SPOE_FRM_ERR_TOO_BIG] = "frame is too big",
174 [SPOE_FRM_ERR_INVALID] = "invalid frame received",
175 [SPOE_FRM_ERR_NO_VSN] = "version value not found",
176 [SPOE_FRM_ERR_NO_FRAME_SIZE] = "max-frame-size value not found",
177 [SPOE_FRM_ERR_NO_CAP] = "capabilities value not found",
178 [SPOE_FRM_ERR_BAD_VSN] = "unsupported version",
179 [SPOE_FRM_ERR_BAD_FRAME_SIZE] = "max-frame-size too big or too small",
180 [SPOE_FRM_ERR_FRAG_NOT_SUPPORTED] = "fragmentation not supported",
181 [SPOE_FRM_ERR_INTERLACED_FRAMES] = "invalid interlaced frames",
Christopher Faulet8eda93f2017-02-09 09:44:33 +0100182 [SPOE_FRM_ERR_FRAMEID_NOTFOUND] = "frame-id not found",
Christopher Faulet85010352017-02-02 10:14:36 +0100183 [SPOE_FRM_ERR_RES] = "resource allocation error",
184 [SPOE_FRM_ERR_UNKNOWN] = "an unknown error occurred",
Christopher Fauletf95b1112016-12-21 08:58:16 +0100185};
186
187static void signal_cb(evutil_socket_t, short, void *);
188static void accept_cb(evutil_socket_t, short, void *);
189static void worker_monitor_cb(evutil_socket_t, short, void *);
190static void process_frame_cb(evutil_socket_t, short, void *);
191static void read_frame_cb(evutil_socket_t, short, void *);
192static void write_frame_cb(evutil_socket_t, short, void *);
193
194static void use_spoe_engine(struct client *);
195static void unuse_spoe_engine(struct client *);
196static void release_frame(struct spoe_frame *);
197static void release_client(struct client *);
Christopher Faulet010fded2016-11-03 22:49:37 +0100198
199static void
Christopher Fauletf95b1112016-12-21 08:58:16 +0100200check_ipv4_reputation(struct spoe_frame *frame, struct in_addr *ipv4)
Christopher Faulet010fded2016-11-03 22:49:37 +0100201{
202 char str[INET_ADDRSTRLEN];
203
204 if (inet_ntop(AF_INET, ipv4, str, INET_ADDRSTRLEN) == NULL)
205 return;
206
Christopher Fauletf95b1112016-12-21 08:58:16 +0100207 frame->ip_score = random() % 100;
Christopher Faulet010fded2016-11-03 22:49:37 +0100208
Christopher Fauletf95b1112016-12-21 08:58:16 +0100209 DEBUG(frame->worker, "IP score for %.*s is %d",
210 INET_ADDRSTRLEN, str, frame->ip_score);
Christopher Faulet010fded2016-11-03 22:49:37 +0100211}
212
213static void
Christopher Fauletf95b1112016-12-21 08:58:16 +0100214check_ipv6_reputation(struct spoe_frame *frame, struct in6_addr *ipv6)
Christopher Faulet010fded2016-11-03 22:49:37 +0100215{
216 char str[INET6_ADDRSTRLEN];
217
218 if (inet_ntop(AF_INET6, ipv6, str, INET6_ADDRSTRLEN) == NULL)
219 return;
220
Christopher Fauletf95b1112016-12-21 08:58:16 +0100221 frame->ip_score = random() % 100;
Christopher Faulet010fded2016-11-03 22:49:37 +0100222
Christopher Fauletf95b1112016-12-21 08:58:16 +0100223 DEBUG(frame->worker, "IP score for %.*s is %d",
224 INET6_ADDRSTRLEN, str, frame->ip_score);
Christopher Faulet010fded2016-11-03 22:49:37 +0100225}
226
Christopher Faulet010fded2016-11-03 22:49:37 +0100227
228/* Check the protocol version. It returns -1 if an error occurred, the number of
229 * read bytes otherwise. */
230static int
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100231check_proto_version(struct spoe_frame *frame, char **buf, char *end)
Christopher Faulet010fded2016-11-03 22:49:37 +0100232{
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100233 char *str, *p = *buf;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100234 uint64_t sz;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100235 int ret;
Christopher Faulet010fded2016-11-03 22:49:37 +0100236
237 /* Get the list of all supported versions by HAProxy */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100238 if ((*p++ & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR)
Christopher Faulet010fded2016-11-03 22:49:37 +0100239 return -1;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100240 ret = spoe_decode_buffer(&p, end, &str, &sz);
241 if (ret == -1 || !str)
Christopher Faulet010fded2016-11-03 22:49:37 +0100242 return -1;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100243
244 DEBUG(frame->worker, "<%lu> Supported versions : %.*s",
245 frame->client->id, (int)sz, str);
Christopher Faulet010fded2016-11-03 22:49:37 +0100246
247 /* TODO: Find the right verion in supported ones */
248
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100249 ret = (p - *buf);
250 *buf = p;
251 return ret;
Christopher Faulet010fded2016-11-03 22:49:37 +0100252}
253
254/* Check max frame size value. It returns -1 if an error occurred, the number of
255 * read bytes otherwise. */
256static int
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100257check_max_frame_size(struct spoe_frame *frame, char **buf, char *end)
Christopher Faulet010fded2016-11-03 22:49:37 +0100258{
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100259 char *p = *buf;
Christopher Faulet010fded2016-11-03 22:49:37 +0100260 uint64_t sz;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100261 int type, ret;
Christopher Faulet010fded2016-11-03 22:49:37 +0100262
263 /* Get the max-frame-size value of HAProxy */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100264 type = *p++;
Christopher Faulet010fded2016-11-03 22:49:37 +0100265 if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT32 &&
266 (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT64 &&
267 (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT32 &&
Christopher Fauletf95b1112016-12-21 08:58:16 +0100268 (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT64)
Christopher Faulet010fded2016-11-03 22:49:37 +0100269 return -1;
Thierry FOURNIER6ab2bae2017-04-19 11:49:44 +0200270 if (decode_varint(&p, end, &sz) == -1)
Christopher Faulet010fded2016-11-03 22:49:37 +0100271 return -1;
Christopher Faulet010fded2016-11-03 22:49:37 +0100272
273 /* Keep the lower value */
Christopher Fauletf95b1112016-12-21 08:58:16 +0100274 if (sz < frame->client->max_frame_size)
275 frame->client->max_frame_size = sz;
276
277 DEBUG(frame->worker, "<%lu> HAProxy maximum frame size : %u",
278 frame->client->id, (unsigned int)sz);
Christopher Faulet010fded2016-11-03 22:49:37 +0100279
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100280 ret = (p - *buf);
281 *buf = p;
282 return ret;
Christopher Faulet010fded2016-11-03 22:49:37 +0100283}
284
Christopher Fauletba7bc162016-11-07 21:07:38 +0100285/* Check healthcheck value. It returns -1 if an error occurred, the number of
286 * read bytes otherwise. */
287static int
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100288check_healthcheck(struct spoe_frame *frame, char **buf, char *end)
Christopher Fauletba7bc162016-11-07 21:07:38 +0100289{
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100290 char *p = *buf;
291 int type, ret;
Christopher Fauletba7bc162016-11-07 21:07:38 +0100292
Christopher Fauletf95b1112016-12-21 08:58:16 +0100293 /* Get the "healthcheck" value */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100294 type = *p++;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100295 if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_BOOL)
296 return -1;
297 frame->hcheck = ((type & SPOE_DATA_FL_TRUE) == SPOE_DATA_FL_TRUE);
298
299 DEBUG(frame->worker, "<%lu> HELLO healthcheck : %s",
300 frame->client->id, (frame->hcheck ? "true" : "false"));
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100301
302 ret = (p - *buf);
303 *buf = p;
304 return ret;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100305}
306
307/* Check capabilities value. It returns -1 if an error occurred, the number of
308 * read bytes otherwise. */
309static int
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100310check_capabilities(struct spoe_frame *frame, char **buf, char *end)
Christopher Fauletf95b1112016-12-21 08:58:16 +0100311{
312 struct client *client = frame->client;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100313 char *str, *p = *buf;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100314 uint64_t sz;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100315 int ret;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100316
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100317 if ((*p++ & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR)
Christopher Fauletba7bc162016-11-07 21:07:38 +0100318 return -1;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100319 if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
320 return -1;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100321 if (str == NULL) /* this is not an error */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100322 goto end;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100323
324 DEBUG(frame->worker, "<%lu> HAProxy capabilities : %.*s",
325 client->id, (int)sz, str);
326
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100327 while (sz) {
Christopher Fauletf95b1112016-12-21 08:58:16 +0100328 char *delim;
329
330 /* Skip leading spaces */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100331 for (; isspace(*str) && sz; sz--);
Christopher Fauletf95b1112016-12-21 08:58:16 +0100332
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100333 if (sz >= 10 && !strncmp(str, "pipelining", 10)) {
334 str += 10; sz -= 10;
335 if (!sz || isspace(*str) || *str == ',') {
Christopher Fauletf95b1112016-12-21 08:58:16 +0100336 DEBUG(frame->worker,
337 "<%lu> HAProxy supports frame pipelining",
338 client->id);
339 client->pipelining = true;
340 }
Christopher Fauletf95b1112016-12-21 08:58:16 +0100341 }
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100342 else if (sz >= 5 && !strncmp(str, "async", 5)) {
343 str += 5; sz -= 5;
344 if (!sz || isspace(*str) || *str == ',') {
Christopher Fauletf95b1112016-12-21 08:58:16 +0100345 DEBUG(frame->worker,
346 "<%lu> HAProxy supports asynchronous frame",
347 client->id);
348 client->async = true;
349 }
350 }
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100351 else if (sz >= 13 && !strncmp(str, "fragmentation", 13)) {
352 str += 13; sz -= 13;
353 if (!sz || isspace(*str) || *str == ',') {
Christopher Faulet85010352017-02-02 10:14:36 +0100354 DEBUG(frame->worker,
355 "<%lu> HAProxy supports fragmented frame",
356 client->id);
357 client->fragmentation = true;
358 }
359 }
Christopher Fauletf95b1112016-12-21 08:58:16 +0100360
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100361 if (!sz || (delim = memchr(str, ',', sz)) == NULL)
Christopher Fauletf95b1112016-12-21 08:58:16 +0100362 break;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100363 delim++;
364 sz -= (delim - str);
365 str = delim;
Christopher Fauletba7bc162016-11-07 21:07:38 +0100366 }
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100367 end:
368 ret = (p - *buf);
369 *buf = p;
370 return ret;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100371}
372
373/* Check engine-id value. It returns -1 if an error occurred, the number of
374 * read bytes otherwise. */
375static int
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100376check_engine_id(struct spoe_frame *frame, char **buf, char *end)
Christopher Fauletf95b1112016-12-21 08:58:16 +0100377{
378 struct client *client = frame->client;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100379 char *str, *p = *buf;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100380 uint64_t sz;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100381 int ret;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100382
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100383 if ((*p++ & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR)
Christopher Fauletf95b1112016-12-21 08:58:16 +0100384 return -1;
385
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100386 if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
387 return -1;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100388 if (str == NULL) /* this is not an error */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100389 goto end;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100390
391 if (client->engine != NULL)
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100392 goto end;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100393
394 DEBUG(frame->worker, "<%lu> HAProxy engine id : %.*s",
395 client->id, (int)sz, str);
396
397 client->engine_id = strndup(str, (int)sz);
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100398 end:
399 ret = (p - *buf);
400 *buf = p;
401 return ret;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100402}
403
Christopher Fauletf032c3e2017-02-17 15:18:35 +0100404static int
405acc_payload(struct spoe_frame *frame)
406{
407 struct client *client = frame->client;
408 char *buf;
409 size_t len = frame->len - frame->offset;
410 int ret = frame->offset;
411
412 /* No need to accumulation payload */
413 if (frame->fragmented == false)
414 return ret;
415
416 buf = realloc(frame->frag_buf, frame->frag_len + len);
417 if (buf == NULL) {
418 client->status_code = SPOE_FRM_ERR_RES;
419 return -1;
420 }
421 memcpy(buf + frame->frag_len, frame->buf + frame->offset, len);
422 frame->frag_buf = buf;
423 frame->frag_len += len;
424
425 if (!(frame->flags & SPOE_FRM_FL_FIN)) {
426 /* Wait for next parts */
427 frame->buf = (char *)(frame->data);
428 frame->offset = 0;
429 frame->len = 0;
430 frame->flags = 0;
431 return 1;
432 }
433
434 frame->buf = frame->frag_buf;
435 frame->len = frame->frag_len;
436 frame->offset = 0;
437 return ret;
438}
439
Christopher Fauletf95b1112016-12-21 08:58:16 +0100440/* Check disconnect status code. It returns -1 if an error occurred, the number
441 * of read bytes otherwise. */
442static int
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100443check_discon_status_code(struct spoe_frame *frame, char **buf, char *end)
Christopher Fauletf95b1112016-12-21 08:58:16 +0100444{
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100445 char *p = *buf;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100446 uint64_t sz;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100447 int type, ret;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100448
449 /* Get the "status-code" value */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100450 type = *p++;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100451 if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT32 &&
452 (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT64 &&
453 (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT32 &&
454 (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT64)
455 return -1;
Thierry FOURNIER6ab2bae2017-04-19 11:49:44 +0200456 if (decode_varint(&p, end, &sz) == -1)
Christopher Fauletf95b1112016-12-21 08:58:16 +0100457 return -1;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100458
459 frame->client->status_code = (unsigned int)sz;
460
461 DEBUG(frame->worker, "<%lu> Disconnect status code : %u",
462 frame->client->id, frame->client->status_code);
463
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100464 ret = (p - *buf);
465 *buf = p;
466 return ret;
Christopher Fauletba7bc162016-11-07 21:07:38 +0100467}
468
Christopher Fauletf95b1112016-12-21 08:58:16 +0100469/* Check the disconnect message. It returns -1 if an error occurred, the number
470 * of read bytes otherwise. */
471static int
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100472check_discon_message(struct spoe_frame *frame, char **buf, char *end)
Christopher Fauletf95b1112016-12-21 08:58:16 +0100473{
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100474 char *str, *p = *buf;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100475 uint64_t sz;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100476 int ret;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100477
478 /* Get the "message" value */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100479 if ((*p++ & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR)
Christopher Fauletf95b1112016-12-21 08:58:16 +0100480 return -1;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100481 ret = spoe_decode_buffer(&p, end, &str, &sz);
482 if (ret == -1 || !str)
Christopher Fauletf95b1112016-12-21 08:58:16 +0100483 return -1;
484
485 DEBUG(frame->worker, "<%lu> Disconnect message : %.*s",
486 frame->client->id, (int)sz, str);
487
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100488 ret = (p - *buf);
489 *buf = p;
490 return ret;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100491}
Christopher Fauletba7bc162016-11-07 21:07:38 +0100492
Christopher Fauletf032c3e2017-02-17 15:18:35 +0100493
494
Christopher Faulet010fded2016-11-03 22:49:37 +0100495/* Decode a HELLO frame received from HAProxy. It returns -1 if an error
Christopher Fauletf95b1112016-12-21 08:58:16 +0100496 * occurred, otherwise the number of read bytes. HELLO frame cannot be
497 * ignored and having another frame than a HELLO frame is an error. */
Christopher Faulet010fded2016-11-03 22:49:37 +0100498static int
Christopher Fauletf95b1112016-12-21 08:58:16 +0100499handle_hahello(struct spoe_frame *frame)
Christopher Faulet010fded2016-11-03 22:49:37 +0100500{
Christopher Fauletf95b1112016-12-21 08:58:16 +0100501 struct client *client = frame->client;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100502 char *p, *end;
503
504 p = frame->buf;
505 end = frame->buf + frame->len;
Christopher Faulet010fded2016-11-03 22:49:37 +0100506
Christopher Fauletf95b1112016-12-21 08:58:16 +0100507 /* Check frame type: we really want a HELLO frame */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100508 if (*p++ != SPOE_FRM_T_HAPROXY_HELLO)
Christopher Fauletf95b1112016-12-21 08:58:16 +0100509 goto error;
510
511 DEBUG(frame->worker, "<%lu> Decode HAProxy HELLO frame", client->id);
Christopher Faulet010fded2016-11-03 22:49:37 +0100512
Christopher Faulet85010352017-02-02 10:14:36 +0100513 /* Retrieve flags */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100514 memcpy((char *)&(frame->flags), p, 4);
515 p += 4;
Christopher Faulet010fded2016-11-03 22:49:37 +0100516
Christopher Faulet85010352017-02-02 10:14:36 +0100517 /* Fragmentation is not supported for HELLO frame */
518 if (!(frame->flags & SPOE_FRM_FL_FIN)) {
519 client->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
520 goto error;
521 }
522
Christopher Faulet010fded2016-11-03 22:49:37 +0100523 /* stream-id and frame-id must be cleared */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100524 if (*p != 0 || *(p+1) != 0) {
Christopher Fauletf95b1112016-12-21 08:58:16 +0100525 client->status_code = SPOE_FRM_ERR_INVALID;
Christopher Faulet010fded2016-11-03 22:49:37 +0100526 goto error;
527 }
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100528 p += 2;
Christopher Faulet010fded2016-11-03 22:49:37 +0100529
530 /* Loop on K/V items */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100531 while (p < end) {
Christopher Faulet010fded2016-11-03 22:49:37 +0100532 char *str;
533 uint64_t sz;
534
535 /* Decode the item name */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100536 spoe_decode_buffer(&p, end, &str, &sz);
537 if (!str) {
Christopher Fauletf95b1112016-12-21 08:58:16 +0100538 client->status_code = SPOE_FRM_ERR_INVALID;
Christopher Faulet010fded2016-11-03 22:49:37 +0100539 goto error;
540 }
541
542 /* Check "supported-versions" K/V item */
543 if (!memcmp(str, "supported-versions", sz)) {
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100544 if (check_proto_version(frame, &p, end) == -1) {
Christopher Fauletf95b1112016-12-21 08:58:16 +0100545 client->status_code = SPOE_FRM_ERR_INVALID;
Christopher Faulet010fded2016-11-03 22:49:37 +0100546 goto error;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100547 }
Christopher Faulet010fded2016-11-03 22:49:37 +0100548 }
Christopher Fauletf95b1112016-12-21 08:58:16 +0100549 /* Check "max-frame-size" K/V item */
Christopher Faulet010fded2016-11-03 22:49:37 +0100550 else if (!memcmp(str, "max-frame-size", sz)) {
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100551 if (check_max_frame_size(frame, &p, end) == -1) {
Christopher Fauletf95b1112016-12-21 08:58:16 +0100552 client->status_code = SPOE_FRM_ERR_INVALID;
Christopher Faulet010fded2016-11-03 22:49:37 +0100553 goto error;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100554 }
Christopher Faulet010fded2016-11-03 22:49:37 +0100555 }
Christopher Fauletf95b1112016-12-21 08:58:16 +0100556 /* Check "healthcheck" K/V item */
Christopher Fauletba7bc162016-11-07 21:07:38 +0100557 else if (!memcmp(str, "healthcheck", sz)) {
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100558 if (check_healthcheck(frame, &p, end) == -1) {
Christopher Fauletf95b1112016-12-21 08:58:16 +0100559 client->status_code = SPOE_FRM_ERR_INVALID;
560 goto error;
561 }
Christopher Fauletf95b1112016-12-21 08:58:16 +0100562 }
563 /* Check "capabilities" K/V item */
564 else if (!memcmp(str, "capabilities", sz)) {
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100565 if (check_capabilities(frame, &p, end) == -1) {
Christopher Fauletf95b1112016-12-21 08:58:16 +0100566 client->status_code = SPOE_FRM_ERR_INVALID;
567 goto error;
568 }
Christopher Fauletf95b1112016-12-21 08:58:16 +0100569 }
570 /* Check "engine-id" K/V item */
571 else if (!memcmp(str, "engine-id", sz)) {
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100572 if (check_engine_id(frame, &p, end) == -1) {
Christopher Fauletf95b1112016-12-21 08:58:16 +0100573 client->status_code = SPOE_FRM_ERR_INVALID;
Christopher Fauletba7bc162016-11-07 21:07:38 +0100574 goto error;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100575 }
Christopher Fauletba7bc162016-11-07 21:07:38 +0100576 }
Christopher Faulet010fded2016-11-03 22:49:37 +0100577 else {
Christopher Fauletf95b1112016-12-21 08:58:16 +0100578 DEBUG(frame->worker, "<%lu> Skip K/V item : key=%.*s",
579 client->id, (int)sz, str);
580
Christopher Faulet010fded2016-11-03 22:49:37 +0100581 /* Silently ignore unknown item */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100582 if (spoe_skip_data(&p, end) == -1) {
Christopher Fauletf95b1112016-12-21 08:58:16 +0100583 client->status_code = SPOE_FRM_ERR_INVALID;
Christopher Faulet010fded2016-11-03 22:49:37 +0100584 goto error;
585 }
Christopher Faulet010fded2016-11-03 22:49:37 +0100586 }
587 }
588
Christopher Fauletf95b1112016-12-21 08:58:16 +0100589 if (async == false || client->engine_id == NULL)
590 client->async = false;
591 if (pipelining == false)
592 client->pipelining = false;
593
594 if (client->async == true)
595 use_spoe_engine(client);
596
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100597 return (p - frame->buf);
Christopher Fauletf95b1112016-12-21 08:58:16 +0100598 error:
Christopher Faulet010fded2016-11-03 22:49:37 +0100599 return -1;
600}
601
602/* Decode a DISCONNECT frame received from HAProxy. It returns -1 if an error
Christopher Fauletf95b1112016-12-21 08:58:16 +0100603 * occurred, otherwise the number of read bytes. DISCONNECT frame cannot be
604 * ignored and having another frame than a DISCONNECT frame is an error.*/
Christopher Faulet010fded2016-11-03 22:49:37 +0100605static int
Christopher Fauletf95b1112016-12-21 08:58:16 +0100606handle_hadiscon(struct spoe_frame *frame)
Christopher Faulet010fded2016-11-03 22:49:37 +0100607{
Christopher Fauletf95b1112016-12-21 08:58:16 +0100608 struct client *client = frame->client;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100609 char *p, *end;
610
611 p = frame->buf;
612 end = frame->buf + frame->len;
Christopher Faulet010fded2016-11-03 22:49:37 +0100613
Christopher Fauletf95b1112016-12-21 08:58:16 +0100614 /* Check frame type: we really want a DISCONNECT frame */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100615 if (*p++ != SPOE_FRM_T_HAPROXY_DISCON)
Christopher Fauletf95b1112016-12-21 08:58:16 +0100616 goto error;
617
618 DEBUG(frame->worker, "<%lu> Decode HAProxy DISCONNECT frame", client->id);
Christopher Faulet010fded2016-11-03 22:49:37 +0100619
Christopher Faulet85010352017-02-02 10:14:36 +0100620 /* Retrieve flags */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100621 memcpy((char *)&(frame->flags), p, 4);
622 p += 4;
Christopher Faulet010fded2016-11-03 22:49:37 +0100623
Christopher Faulet85010352017-02-02 10:14:36 +0100624 /* Fragmentation is not supported for DISCONNECT frame */
625 if (!(frame->flags & SPOE_FRM_FL_FIN)) {
626 client->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
627 goto error;
628 }
629
Christopher Faulet010fded2016-11-03 22:49:37 +0100630 /* stream-id and frame-id must be cleared */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100631 if (*p != 0 || *(p+1) != 0) {
Christopher Fauletf95b1112016-12-21 08:58:16 +0100632 client->status_code = SPOE_FRM_ERR_INVALID;
Christopher Faulet010fded2016-11-03 22:49:37 +0100633 goto error;
634 }
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100635 p += 2;
Christopher Faulet010fded2016-11-03 22:49:37 +0100636
Christopher Fauletf95b1112016-12-21 08:58:16 +0100637 client->status_code = SPOE_FRM_ERR_NONE;
638
Christopher Faulet010fded2016-11-03 22:49:37 +0100639 /* Loop on K/V items */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100640 while (p < end) {
Christopher Faulet010fded2016-11-03 22:49:37 +0100641 char *str;
642 uint64_t sz;
643
644 /* Decode item key */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100645 spoe_decode_buffer(&p, end, &str, &sz);
646 if (!str) {
Christopher Fauletf95b1112016-12-21 08:58:16 +0100647 client->status_code = SPOE_FRM_ERR_INVALID;
Christopher Faulet010fded2016-11-03 22:49:37 +0100648 goto error;
649 }
Christopher Fauletf95b1112016-12-21 08:58:16 +0100650
651 /* Check "status-code" K/V item */
652 if (!memcmp(str, "status-code", sz)) {
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100653 if (check_discon_status_code(frame, &p, end) == -1) {
Christopher Fauletf95b1112016-12-21 08:58:16 +0100654 client->status_code = SPOE_FRM_ERR_INVALID;
655 goto error;
656 }
Christopher Fauletf95b1112016-12-21 08:58:16 +0100657 }
658 /* Check "message" K/V item */
659 else if (!memcmp(str, "message", sz)) {
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100660 if (check_discon_message(frame, &p, end) == -1) {
Christopher Fauletf95b1112016-12-21 08:58:16 +0100661 client->status_code = SPOE_FRM_ERR_INVALID;
662 goto error;
663 }
Christopher Fauletf95b1112016-12-21 08:58:16 +0100664 }
665 else {
666 DEBUG(frame->worker, "<%lu> Skip K/V item : key=%.*s",
667 client->id, (int)sz, str);
668
669 /* Silently ignore unknown item */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100670 if (spoe_skip_data(&p, end) == -1) {
Christopher Fauletf95b1112016-12-21 08:58:16 +0100671 client->status_code = SPOE_FRM_ERR_INVALID;
672 goto error;
673 }
Christopher Faulet010fded2016-11-03 22:49:37 +0100674 }
Christopher Faulet010fded2016-11-03 22:49:37 +0100675 }
676
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100677 return (p - frame->buf);
Christopher Fauletf95b1112016-12-21 08:58:16 +0100678 error:
Christopher Faulet010fded2016-11-03 22:49:37 +0100679 return -1;
680}
681
682/* Decode a NOTIFY frame received from HAProxy. It returns -1 if an error
Christopher Faulet85010352017-02-02 10:14:36 +0100683 * occurred, 0 if it must be must be ignored, otherwise the number of read
684 * bytes. */
Christopher Faulet010fded2016-11-03 22:49:37 +0100685static int
Christopher Fauletf95b1112016-12-21 08:58:16 +0100686handle_hanotify(struct spoe_frame *frame)
Christopher Faulet010fded2016-11-03 22:49:37 +0100687{
Christopher Fauletf95b1112016-12-21 08:58:16 +0100688 struct client *client = frame->client;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100689 char *p, *end;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100690 uint64_t stream_id, frame_id;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100691
692 p = frame->buf;
693 end = frame->buf + frame->len;
Christopher Faulet010fded2016-11-03 22:49:37 +0100694
695 /* Check frame type */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100696 if (*p++ != SPOE_FRM_T_HAPROXY_NOTIFY)
Christopher Fauletf95b1112016-12-21 08:58:16 +0100697 goto ignore;
698
699 DEBUG(frame->worker, "<%lu> Decode HAProxy NOTIFY frame", client->id);
Christopher Faulet010fded2016-11-03 22:49:37 +0100700
Christopher Faulet85010352017-02-02 10:14:36 +0100701 /* Retrieve flags */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100702 memcpy((char *)&(frame->flags), p, 4);
703 p += 4;
Christopher Faulet010fded2016-11-03 22:49:37 +0100704
Christopher Fauletf032c3e2017-02-17 15:18:35 +0100705 /* Fragmentation is not supported */
Christopher Faulet85010352017-02-02 10:14:36 +0100706 if (!(frame->flags & SPOE_FRM_FL_FIN) && fragmentation == false) {
707 client->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
708 goto error;
709 }
710
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100711 /* Read the stream-id and frame-id */
Thierry FOURNIER6ab2bae2017-04-19 11:49:44 +0200712 if (decode_varint(&p, end, &stream_id) == -1)
Christopher Fauletf95b1112016-12-21 08:58:16 +0100713 goto ignore;
Thierry FOURNIER6ab2bae2017-04-19 11:49:44 +0200714 if (decode_varint(&p, end, &frame_id) == -1)
Christopher Fauletf95b1112016-12-21 08:58:16 +0100715 goto ignore;
Christopher Faulet010fded2016-11-03 22:49:37 +0100716
Christopher Fauletf032c3e2017-02-17 15:18:35 +0100717 frame->stream_id = (unsigned int)stream_id;
718 frame->frame_id = (unsigned int)frame_id;
Christopher Faulet010fded2016-11-03 22:49:37 +0100719
Christopher Fauletf032c3e2017-02-17 15:18:35 +0100720 DEBUG(frame->worker, "<%lu> STREAM-ID=%u - FRAME-ID=%u"
721 " - %s frame received"
722 " - frag_len=%u - len=%u - offset=%ld",
723 client->id, frame->stream_id, frame->frame_id,
724 (frame->flags & SPOE_FRM_FL_FIN) ? "unfragmented" : "fragmented",
725 frame->frag_len, frame->len, p - frame->buf);
726
727 frame->fragmented = !(frame->flags & SPOE_FRM_FL_FIN);
728 frame->offset = (p - frame->buf);
729 return acc_payload(frame);
730
731 ignore:
732 return 0;
733
734 error:
735 return -1;
736}
737
738/* Decode next part of a fragmented frame received from HAProxy. It returns -1
739 * if an error occurred, 0 if it must be must be ignored, otherwise the number
740 * of read bytes. */
741static int
742handle_hafrag(struct spoe_frame *frame)
743{
744 struct client *client = frame->client;
745 char *p, *end;
746 uint64_t stream_id, frame_id;
747
748 p = frame->buf;
749 end = frame->buf + frame->len;
750
751 /* Check frame type */
752 if (*p++ != SPOE_FRM_T_UNSET)
753 goto ignore;
754
755 DEBUG(frame->worker, "<%lu> Decode Next part of a fragmented frame", client->id);
756
757 /* Fragmentation is not supported */
758 if (fragmentation == false) {
759 client->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
760 goto error;
Christopher Faulet85010352017-02-02 10:14:36 +0100761 }
Christopher Fauletf032c3e2017-02-17 15:18:35 +0100762
763 /* Retrieve flags */
764 memcpy((char *)&(frame->flags), p, 4);
765 p+= 4;
766
767 /* Read the stream-id and frame-id */
Thierry FOURNIER6ab2bae2017-04-19 11:49:44 +0200768 if (decode_varint(&p, end, &stream_id) == -1)
Christopher Fauletf032c3e2017-02-17 15:18:35 +0100769 goto ignore;
Thierry FOURNIER6ab2bae2017-04-19 11:49:44 +0200770 if (decode_varint(&p, end, &frame_id) == -1)
Christopher Fauletf032c3e2017-02-17 15:18:35 +0100771 goto ignore;
Christopher Faulet010fded2016-11-03 22:49:37 +0100772
Christopher Fauletf032c3e2017-02-17 15:18:35 +0100773 if (frame->fragmented == false ||
774 frame->stream_id != (unsigned int)stream_id ||
775 frame->frame_id != (unsigned int)frame_id) {
776 client->status_code = SPOE_FRM_ERR_INTERLACED_FRAMES;
777 goto error;
778 }
779
780 if (frame->flags & SPOE_FRM_FL_ABRT) {
Christopher Faulet85010352017-02-02 10:14:36 +0100781 DEBUG(frame->worker, "<%lu> STREAM-ID=%u - FRAME-ID=%u"
Christopher Fauletf032c3e2017-02-17 15:18:35 +0100782 " - Abort processing of a fragmented frame"
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100783 " - frag_len=%u - len=%u - offset=%ld",
Christopher Faulet85010352017-02-02 10:14:36 +0100784 client->id, frame->stream_id, frame->frame_id,
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100785 frame->frag_len, frame->len, p - frame->buf);
Christopher Fauletf032c3e2017-02-17 15:18:35 +0100786 goto ignore;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100787 }
Christopher Faulet010fded2016-11-03 22:49:37 +0100788
Christopher Fauletf032c3e2017-02-17 15:18:35 +0100789 DEBUG(frame->worker, "<%lu> STREAM-ID=%u - FRAME-ID=%u"
790 " - %s fragment of a fragmented frame received"
791 " - frag_len=%u - len=%u - offset=%ld",
792 client->id, frame->stream_id, frame->frame_id,
793 (frame->flags & SPOE_FRM_FL_FIN) ? "last" : "next",
794 frame->frag_len, frame->len, p - frame->buf);
795
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100796 frame->offset = (p - frame->buf);
Christopher Fauletf032c3e2017-02-17 15:18:35 +0100797 return acc_payload(frame);
Christopher Faulet010fded2016-11-03 22:49:37 +0100798
Christopher Fauletf95b1112016-12-21 08:58:16 +0100799 ignore:
Christopher Faulet85010352017-02-02 10:14:36 +0100800 return 0;
801
802 error:
Christopher Fauletf95b1112016-12-21 08:58:16 +0100803 return -1;
804}
Christopher Faulet010fded2016-11-03 22:49:37 +0100805
Christopher Fauletf95b1112016-12-21 08:58:16 +0100806/* Encode a HELLO frame to send it to HAProxy. It returns the number of written
807 * bytes. */
808static int
809prepare_agenthello(struct spoe_frame *frame)
810{
811 struct client *client = frame->client;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100812 char *p, *end;
Christopher Faulet85010352017-02-02 10:14:36 +0100813 char capabilities[64];
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100814 int n;
Christopher Faulet85010352017-02-02 10:14:36 +0100815 unsigned int flags = SPOE_FRM_FL_FIN;
Christopher Faulet010fded2016-11-03 22:49:37 +0100816
Christopher Fauletf95b1112016-12-21 08:58:16 +0100817 DEBUG(frame->worker, "<%lu> Encode Agent HELLO frame", client->id);
818 frame->type = SPOA_FRM_T_AGENT;
Christopher Faulet010fded2016-11-03 22:49:37 +0100819
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100820 p = frame->buf;
821 end = frame->buf+max_frame_size;
822
Christopher Faulet010fded2016-11-03 22:49:37 +0100823 /* Frame Type */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100824 *p++ = SPOE_FRM_T_AGENT_HELLO;
Christopher Faulet010fded2016-11-03 22:49:37 +0100825
Christopher Faulet85010352017-02-02 10:14:36 +0100826 /* Set flags */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100827 memcpy(p, (char *)&flags, 4);
828 p += 4;
Christopher Faulet010fded2016-11-03 22:49:37 +0100829
830 /* No stream-id and frame-id for HELLO frames */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100831 *p++ = 0;
832 *p++ = 0;
Christopher Faulet010fded2016-11-03 22:49:37 +0100833
834 /* "version" K/V item */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100835 spoe_encode_buffer("version", 7, &p, end);
836 *p++ = SPOE_DATA_T_STR;
837 spoe_encode_buffer(SPOP_VERSION, SLEN(SPOP_VERSION), &p, end);
Christopher Fauletf95b1112016-12-21 08:58:16 +0100838 DEBUG(frame->worker, "<%lu> Agent version : %s",
839 client->id, SPOP_VERSION);
840
Christopher Faulet010fded2016-11-03 22:49:37 +0100841
842 /* "max-frame-size" K/V item */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100843 spoe_encode_buffer("max-frame-size", 14, &p ,end);
844 *p++ = SPOE_DATA_T_UINT32;
Thierry FOURNIER6ab2bae2017-04-19 11:49:44 +0200845 encode_varint(client->max_frame_size, &p, end);
Christopher Fauletf95b1112016-12-21 08:58:16 +0100846 DEBUG(frame->worker, "<%lu> Agent maximum frame size : %u",
847 client->id, client->max_frame_size);
Christopher Faulet010fded2016-11-03 22:49:37 +0100848
849 /* "capabilities" K/V item */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100850 spoe_encode_buffer("capabilities", 12, &p, end);
851 *p++ = SPOE_DATA_T_STR;
Christopher Faulet85010352017-02-02 10:14:36 +0100852
853 memset(capabilities, 0, sizeof(capabilities));
854 n = 0;
855
856 /* 1. Fragmentation capability ? */
857 if (fragmentation == true) {
858 memcpy(capabilities, "fragmentation", 13);
859 n += 13;
860 }
861 /* 2. Pipelining capability ? */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100862 if (client->pipelining == true) {
863 if (n) capabilities[n++] = ',';
864 memcpy(capabilities + n, "pipelining", 10);
Christopher Faulet85010352017-02-02 10:14:36 +0100865 n += 10;
866 }
867 /* 3. Async capability ? */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100868 if (client->async == true) {
869 if (n) capabilities[n++] = ',';
870 memcpy(capabilities + n, "async", 5);
Christopher Faulet85010352017-02-02 10:14:36 +0100871 n += 5;
872 }
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100873 spoe_encode_buffer(capabilities, n, &p, end);
Christopher Faulet010fded2016-11-03 22:49:37 +0100874
Christopher Faulet85010352017-02-02 10:14:36 +0100875 DEBUG(frame->worker, "<%lu> Agent capabilities : %.*s",
876 client->id, n, capabilities);
Christopher Fauletf95b1112016-12-21 08:58:16 +0100877
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100878 frame->len = (p - frame->buf);
879 return frame->len;
Christopher Faulet010fded2016-11-03 22:49:37 +0100880}
881
Christopher Fauletf95b1112016-12-21 08:58:16 +0100882/* Encode a DISCONNECT frame to send it to HAProxy. It returns the number of
883 * written bytes. */
Christopher Faulet010fded2016-11-03 22:49:37 +0100884static int
Christopher Fauletf95b1112016-12-21 08:58:16 +0100885prepare_agentdicon(struct spoe_frame *frame)
Christopher Faulet010fded2016-11-03 22:49:37 +0100886{
Christopher Fauletf95b1112016-12-21 08:58:16 +0100887 struct client *client = frame->client;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100888 char *p, *end;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100889 const char *reason;
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100890 int rlen;
Christopher Faulet85010352017-02-02 10:14:36 +0100891 unsigned int flags = SPOE_FRM_FL_FIN;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100892
893 DEBUG(frame->worker, "<%lu> Encode Agent DISCONNECT frame", client->id);
894 frame->type = SPOA_FRM_T_AGENT;
895
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100896 p = frame->buf;
897 end = frame->buf+max_frame_size;
898
Christopher Fauletf95b1112016-12-21 08:58:16 +0100899 if (client->status_code >= SPOE_FRM_ERRS)
900 client->status_code = SPOE_FRM_ERR_UNKNOWN;
901 reason = spoe_frm_err_reasons[client->status_code];
902 rlen = strlen(reason);
Christopher Faulet010fded2016-11-03 22:49:37 +0100903
904 /* Frame type */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100905 *p++ = SPOE_FRM_T_AGENT_DISCON;
Christopher Faulet010fded2016-11-03 22:49:37 +0100906
Christopher Faulet85010352017-02-02 10:14:36 +0100907 /* Set flags */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100908 memcpy(p, (char *)&flags, 4);
909 p += 4;
Christopher Faulet010fded2016-11-03 22:49:37 +0100910
Christopher Fauletf95b1112016-12-21 08:58:16 +0100911 /* No stream-id and frame-id for DISCONNECT frames */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100912 *p++ = 0;
913 *p++ = 0;
Christopher Fauletf95b1112016-12-21 08:58:16 +0100914
915 /* There are 2 mandatory items: "status-code" and "message" */
916
917 /* "status-code" K/V item */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100918 spoe_encode_buffer("status-code", 11, &p, end);
919 *p++ = SPOE_DATA_T_UINT32;
Thierry FOURNIER6ab2bae2017-04-19 11:49:44 +0200920 encode_varint(client->status_code, &p, end);
Christopher Fauletf95b1112016-12-21 08:58:16 +0100921 DEBUG(frame->worker, "<%lu> Disconnect status code : %u",
922 client->id, client->status_code);
Christopher Faulet010fded2016-11-03 22:49:37 +0100923
Christopher Fauletf95b1112016-12-21 08:58:16 +0100924 /* "message" K/V item */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100925 spoe_encode_buffer("message", 7, &p, end);
926 *p++ = SPOE_DATA_T_STR;
927 spoe_encode_buffer(reason, rlen, &p, end);
Christopher Fauletf95b1112016-12-21 08:58:16 +0100928 DEBUG(frame->worker, "<%lu> Disconnect message : %s",
929 client->id, reason);
Christopher Faulet010fded2016-11-03 22:49:37 +0100930
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100931 frame->len = (p - frame->buf);
932 return frame->len;
Christopher Faulet010fded2016-11-03 22:49:37 +0100933}
934
Christopher Fauletf95b1112016-12-21 08:58:16 +0100935/* Encode a ACK frame to send it to HAProxy. It returns the number of written
936 * bytes. */
Christopher Faulet010fded2016-11-03 22:49:37 +0100937static int
Christopher Fauletf95b1112016-12-21 08:58:16 +0100938prepare_agentack(struct spoe_frame *frame)
Christopher Faulet010fded2016-11-03 22:49:37 +0100939{
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100940 char *p, *end;
Christopher Faulet85010352017-02-02 10:14:36 +0100941 unsigned int flags = SPOE_FRM_FL_FIN;
Christopher Faulet010fded2016-11-03 22:49:37 +0100942
Christopher Fauletf95b1112016-12-21 08:58:16 +0100943 /* Be careful here, in async mode, frame->client can be NULL */
944
945 DEBUG(frame->worker, "Encode Agent ACK frame");
946 frame->type = SPOA_FRM_T_AGENT;
Christopher Faulet010fded2016-11-03 22:49:37 +0100947
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100948 p = frame->buf;
949 end = frame->buf+max_frame_size;
950
Christopher Faulet010fded2016-11-03 22:49:37 +0100951 /* Frame type */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100952 *p++ = SPOE_FRM_T_AGENT_ACK;
Christopher Faulet010fded2016-11-03 22:49:37 +0100953
Christopher Faulet85010352017-02-02 10:14:36 +0100954 /* Set flags */
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100955 memcpy(p, (char *)&flags, 4);
956 p += 4;
Christopher Faulet010fded2016-11-03 22:49:37 +0100957
Christopher Fauletf95b1112016-12-21 08:58:16 +0100958 /* Set stream-id and frame-id for ACK frames */
Thierry FOURNIER6ab2bae2017-04-19 11:49:44 +0200959 encode_varint(frame->stream_id, &p, end);
960 encode_varint(frame->frame_id, &p, end);
Christopher Faulet010fded2016-11-03 22:49:37 +0100961
Christopher Fauletf95b1112016-12-21 08:58:16 +0100962 DEBUG(frame->worker, "STREAM-ID=%u - FRAME-ID=%u",
963 frame->stream_id, frame->frame_id);
Christopher Faulet010fded2016-11-03 22:49:37 +0100964
Christopher Faulet4ff3e572017-02-24 14:31:11 +0100965 frame->len = (p - frame->buf);
966 return frame->len;
Christopher Faulet010fded2016-11-03 22:49:37 +0100967}
968
969static int
Christopher Fauletf95b1112016-12-21 08:58:16 +0100970create_server_socket(void)
Christopher Faulet010fded2016-11-03 22:49:37 +0100971{
Christopher Fauletf95b1112016-12-21 08:58:16 +0100972 struct sockaddr_in listen_addr;
973 int fd, yes = 1;
974
975 fd = socket(AF_INET, SOCK_STREAM, 0);
976 if (fd < 0) {
977 LOG(&null_worker, "Failed to create service socket : %m");
978 return -1;
Christopher Faulet010fded2016-11-03 22:49:37 +0100979 }
Christopher Fauletf95b1112016-12-21 08:58:16 +0100980
981 memset(&listen_addr, 0, sizeof(listen_addr));
982 listen_addr.sin_family = AF_INET;
983 listen_addr.sin_addr.s_addr = INADDR_ANY;
984 listen_addr.sin_port = htons(server_port);
985
986 if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)) < 0 ||
987 setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes)) < 0) {
988 LOG(&null_worker, "Failed to set option on server socket : %m");
989 return -1;
Christopher Faulet010fded2016-11-03 22:49:37 +0100990 }
Christopher Fauletf95b1112016-12-21 08:58:16 +0100991
992 if (bind(fd, (struct sockaddr *) &listen_addr, sizeof(listen_addr)) < 0) {
993 LOG(&null_worker, "Failed to bind server socket : %m");
994 return -1;
Christopher Faulet010fded2016-11-03 22:49:37 +0100995 }
Christopher Fauletf95b1112016-12-21 08:58:16 +0100996
997 if (listen(fd, CONNECTION_BACKLOG) < 0) {
998 LOG(&null_worker, "Failed to listen on server socket : %m");
999 return -1;
Christopher Faulet010fded2016-11-03 22:49:37 +01001000 }
Christopher Fauletf95b1112016-12-21 08:58:16 +01001001
1002 return fd;
Christopher Faulet010fded2016-11-03 22:49:37 +01001003}
1004
Christopher Fauletf95b1112016-12-21 08:58:16 +01001005static void
1006release_frame(struct spoe_frame *frame)
1007{
1008 struct worker *worker;
1009
1010 if (frame == NULL)
1011 return;
1012
1013 if (event_pending(&frame->process_frame_event, EV_TIMEOUT, NULL))
1014 event_del(&frame->process_frame_event);
1015
1016 worker = frame->worker;
1017 LIST_DEL(&frame->list);
Christopher Faulet85010352017-02-02 10:14:36 +01001018 if (frame->frag_buf)
1019 free(frame->frag_buf);
Christopher Fauletf95b1112016-12-21 08:58:16 +01001020 memset(frame, 0, sizeof(*frame)+max_frame_size+4);
1021 LIST_ADDQ(&worker->frames, &frame->list);
1022}
1023
1024static void
1025release_client(struct client *c)
Christopher Faulet010fded2016-11-03 22:49:37 +01001026{
Christopher Fauletf95b1112016-12-21 08:58:16 +01001027 struct spoe_frame *frame, *back;
1028
1029 if (c == NULL)
1030 return;
1031
1032 DEBUG(c->worker, "<%lu> Release client", c->id);
1033
1034 LIST_DEL(&c->by_worker);
1035 c->worker->nbclients--;
1036
1037 unuse_spoe_engine(c);
1038 free(c->engine_id);
1039
1040 if (event_pending(&c->read_frame_event, EV_READ, NULL))
1041 event_del(&c->read_frame_event);
1042 if (event_pending(&c->write_frame_event, EV_WRITE, NULL))
1043 event_del(&c->write_frame_event);
1044
1045 release_frame(c->incoming_frame);
1046 release_frame(c->outgoing_frame);
1047 list_for_each_entry_safe(frame, back, &c->processing_frames, list) {
1048 release_frame(frame);
Christopher Faulet010fded2016-11-03 22:49:37 +01001049 }
Christopher Fauletf95b1112016-12-21 08:58:16 +01001050 list_for_each_entry_safe(frame, back, &c->outgoing_frames, list) {
1051 release_frame(frame);
Christopher Faulet010fded2016-11-03 22:49:37 +01001052 }
Christopher Fauletf95b1112016-12-21 08:58:16 +01001053
1054 if (c->fd >= 0)
1055 close(c->fd);
1056
1057 free(c);
1058}
1059
1060static void
1061reset_frame(struct spoe_frame *frame)
1062{
1063 if (frame == NULL)
1064 return;
1065
Christopher Faulet85010352017-02-02 10:14:36 +01001066 if (frame->frag_buf)
1067 free(frame->frag_buf);
1068
1069 frame->type = SPOA_FRM_T_UNKNOWN;
1070 frame->buf = (char *)(frame->data);
1071 frame->offset = 0;
1072 frame->len = 0;
1073 frame->stream_id = 0;
1074 frame->frame_id = 0;
1075 frame->flags = 0;
1076 frame->hcheck = false;
1077 frame->fragmented = false;
1078 frame->ip_score = -1;
1079 frame->frag_buf = NULL;
1080 frame->frag_len = 0;
Christopher Fauletf95b1112016-12-21 08:58:16 +01001081 LIST_INIT(&frame->list);
1082}
1083
1084static void
1085use_spoe_engine(struct client *client)
1086{
1087 struct spoe_engine *eng;
1088
1089 if (client->engine_id == NULL)
1090 return;
1091
1092 list_for_each_entry(eng, &client->worker->engines, list) {
1093 if (!strcmp(eng->id, client->engine_id))
1094 goto end;
1095 }
1096
1097 if ((eng = malloc(sizeof(*eng))) == NULL) {
1098 client->async = false;
1099 return;
1100 }
1101
1102 eng->id = strdup(client->engine_id);
1103 LIST_INIT(&eng->clients);
1104 LIST_INIT(&eng->processing_frames);
1105 LIST_INIT(&eng->outgoing_frames);
1106 LIST_ADDQ(&client->worker->engines, &eng->list);
1107 LOG(client->worker, "Add new SPOE engine '%s'", eng->id);
1108
1109 end:
1110 client->engine = eng;
1111 LIST_ADDQ(&eng->clients, &client->by_engine);
1112}
1113
1114static void
1115unuse_spoe_engine(struct client *client)
1116{
1117 struct spoe_engine *eng;
1118 struct spoe_frame *frame, *back;
1119
1120 if (client == NULL || client->engine == NULL)
1121 return;
1122
1123 eng = client->engine;
1124 client->engine = NULL;
1125 LIST_DEL(&client->by_engine);
1126 if (!LIST_ISEMPTY(&eng->clients))
1127 return;
1128
1129 LOG(client->worker, "Remove SPOE engine '%s'", eng->id);
1130 LIST_DEL(&eng->list);
1131
1132 list_for_each_entry_safe(frame, back, &eng->processing_frames, list) {
1133 release_frame(frame);
1134 }
1135 list_for_each_entry_safe(frame, back, &eng->outgoing_frames, list) {
1136 release_frame(frame);
1137 }
1138 free(eng->id);
1139 free(eng);
1140}
1141
1142
1143static struct spoe_frame *
1144acquire_incoming_frame(struct client *client)
1145{
1146 struct spoe_frame *frame;
1147
1148 frame = client->incoming_frame;
1149 if (frame != NULL)
1150 return frame;
1151
1152 if (LIST_ISEMPTY(&client->worker->frames)) {
1153 if ((frame = calloc(1, sizeof(*frame)+max_frame_size+4)) == NULL) {
1154 LOG(client->worker, "Failed to allocate new frame : %m");
1155 return NULL;
1156 }
Christopher Faulet010fded2016-11-03 22:49:37 +01001157 }
Christopher Fauletf95b1112016-12-21 08:58:16 +01001158 else {
1159 frame = LIST_NEXT(&client->worker->frames, typeof(frame), list);
1160 LIST_DEL(&frame->list);
Christopher Faulet010fded2016-11-03 22:49:37 +01001161 }
Christopher Fauletf95b1112016-12-21 08:58:16 +01001162
1163 reset_frame(frame);
1164 frame->worker = client->worker;
1165 frame->engine = client->engine;
1166 frame->client = client;
1167
1168 if (event_assign(&frame->process_frame_event, client->worker->base, -1,
1169 EV_TIMEOUT|EV_PERSIST, process_frame_cb, frame) < 0) {
1170 LOG(client->worker, "Failed to create frame event");
1171 return NULL;
Christopher Faulet010fded2016-11-03 22:49:37 +01001172 }
Christopher Fauletf95b1112016-12-21 08:58:16 +01001173
1174 client->incoming_frame = frame;
1175 return frame;
Christopher Faulet010fded2016-11-03 22:49:37 +01001176}
1177
Christopher Fauletf95b1112016-12-21 08:58:16 +01001178static struct spoe_frame *
1179acquire_outgoing_frame(struct client *client)
Christopher Faulet010fded2016-11-03 22:49:37 +01001180{
Christopher Fauletf95b1112016-12-21 08:58:16 +01001181 struct spoe_engine *engine = client->engine;
1182 struct spoe_frame *frame = NULL;
Christopher Faulet010fded2016-11-03 22:49:37 +01001183
Christopher Fauletf95b1112016-12-21 08:58:16 +01001184 if (client->outgoing_frame != NULL)
1185 frame = client->outgoing_frame;
1186 else if (!LIST_ISEMPTY(&client->outgoing_frames)) {
1187 frame = LIST_NEXT(&client->outgoing_frames, typeof(frame), list);
1188 LIST_DEL(&frame->list);
1189 client->outgoing_frame = frame;
1190 }
1191 else if (engine!= NULL && !LIST_ISEMPTY(&engine->outgoing_frames)) {
1192 frame = LIST_NEXT(&engine->outgoing_frames, typeof(frame), list);
1193 LIST_DEL(&frame->list);
1194 client->outgoing_frame = frame;
1195 }
1196 return frame;
1197}
1198
1199static void
1200write_frame(struct client *client, struct spoe_frame *frame)
1201{
1202 uint32_t netint;
1203
1204 LIST_DEL(&frame->list);
1205
1206 frame->buf = (char *)(frame->data);
1207 frame->offset = 0;
1208 netint = htonl(frame->len);
1209 memcpy(frame->buf, &netint, 4);
Christopher Faulet010fded2016-11-03 22:49:37 +01001210
Christopher Fauletf95b1112016-12-21 08:58:16 +01001211 if (client != NULL) { /* HELLO or DISCONNECT frames */
1212 event_add(&client->write_frame_event, NULL);
Christopher Faulet010fded2016-11-03 22:49:37 +01001213
Christopher Fauletf95b1112016-12-21 08:58:16 +01001214 /* Try to process the frame as soon as possible, and always
1215 * attach it to the client */
1216 if (client->async || client->pipelining) {
1217 if (client->outgoing_frame == NULL)
1218 client->outgoing_frame = frame;
1219 else
1220 LIST_ADD(&client->outgoing_frames, &frame->list);
1221 }
1222 else {
1223 client->outgoing_frame = frame;
1224 event_del(&client->read_frame_event);
Christopher Faulet010fded2016-11-03 22:49:37 +01001225 }
Christopher Fauletf95b1112016-12-21 08:58:16 +01001226 }
1227 else { /* for all other frames */
1228 if (frame->client == NULL) { /* async mode ! */
1229 LIST_ADDQ(&frame->engine->outgoing_frames, &frame->list);
1230 list_for_each_entry(client, &frame->engine->clients, by_engine)
1231 event_add(&client->write_frame_event, NULL);
1232 }
1233 else if (frame->client->pipelining) {
1234 LIST_ADDQ(&frame->client->outgoing_frames, &frame->list);
1235 event_add(&frame->client->write_frame_event, NULL);
1236 }
1237 else {
1238 frame->client->outgoing_frame = frame;
1239 event_add(&frame->client->write_frame_event, NULL);
1240 event_del(&frame->client->read_frame_event);
1241 }
1242 }
1243}
Christopher Faulet010fded2016-11-03 22:49:37 +01001244
Christopher Fauletf95b1112016-12-21 08:58:16 +01001245static void
1246process_incoming_frame(struct spoe_frame *frame)
1247{
1248 struct client *client = frame->client;
Christopher Faulet010fded2016-11-03 22:49:37 +01001249
Christopher Fauletf95b1112016-12-21 08:58:16 +01001250 if (event_add(&frame->process_frame_event, &processing_delay) < 0) {
1251 LOG(client->worker, "Failed to process incoming frame");
1252 release_frame(frame);
1253 return;
1254 }
1255
1256 if (client->async) {
1257 frame->client = NULL;
1258 LIST_ADDQ(&frame->engine->processing_frames, &frame->list);
1259 }
1260 else if (client->pipelining)
1261 LIST_ADDQ(&client->processing_frames, &frame->list);
1262 else
1263 event_del(&client->read_frame_event);
1264}
1265
1266static void
1267signal_cb(evutil_socket_t sig, short events, void *user_data)
1268{
1269 struct event_base *base = user_data;
1270 int i;
1271
1272 DEBUG(&null_worker, "Stopping the server");
1273
1274 event_base_loopbreak(base);
1275 DEBUG(&null_worker, "Main event loop stopped");
1276
1277 for (i = 0; i < num_workers; i++) {
1278 event_base_loopbreak(workers[i].base);
1279 DEBUG(&null_worker, "Event loop stopped for worker %02d",
1280 workers[i].id);
1281 }
1282}
1283
1284static void
1285worker_monitor_cb(evutil_socket_t fd, short events, void *arg)
1286{
1287 struct worker *worker = arg;
1288
1289 LOG(worker, "%u clients connected", worker->nbclients);
1290}
1291
1292static void
1293process_frame_cb(evutil_socket_t fd, short events, void *arg)
1294{
1295 struct spoe_frame *frame = arg;
Christopher Faulet4ff3e572017-02-24 14:31:11 +01001296 char *p, *end;
1297 int ret;
Christopher Fauletf95b1112016-12-21 08:58:16 +01001298
1299 DEBUG(frame->worker,
Christopher Faulet85010352017-02-02 10:14:36 +01001300 "Process frame messages : STREAM-ID=%u - FRAME-ID=%u - length=%u bytes",
1301 frame->stream_id, frame->frame_id, frame->len - frame->offset);
Christopher Fauletf95b1112016-12-21 08:58:16 +01001302
Christopher Faulet4ff3e572017-02-24 14:31:11 +01001303 p = frame->buf + frame->offset;
1304 end = frame->buf + frame->len;
1305
Christopher Fauletf95b1112016-12-21 08:58:16 +01001306 /* Loop on messages */
Christopher Faulet4ff3e572017-02-24 14:31:11 +01001307 while (p < end) {
Christopher Fauletf95b1112016-12-21 08:58:16 +01001308 char *str;
1309 uint64_t sz;
Christopher Faulet4ff3e572017-02-24 14:31:11 +01001310 int nbargs;
Christopher Fauletf95b1112016-12-21 08:58:16 +01001311
1312 /* Decode the message name */
Christopher Faulet4ff3e572017-02-24 14:31:11 +01001313 spoe_decode_buffer(&p, end, &str, &sz);
1314 if (!str)
Christopher Fauletf95b1112016-12-21 08:58:16 +01001315 goto stop_processing;
1316
1317 DEBUG(frame->worker, "Process SPOE Message '%.*s'", (int)sz, str);
1318
Christopher Faulet4ff3e572017-02-24 14:31:11 +01001319 nbargs = *p++; /* Get the number of arguments */
1320 frame->offset = (p - frame->buf); /* Save index to handle errors and skip args */
Christopher Fauletf95b1112016-12-21 08:58:16 +01001321 if (!memcmp(str, "check-client-ip", sz)) {
Christopher Faulet4ff3e572017-02-24 14:31:11 +01001322 struct sample smp;
Christopher Fauletf95b1112016-12-21 08:58:16 +01001323
Christopher Faulet4ff3e572017-02-24 14:31:11 +01001324 memset(&smp, 0, sizeof(smp));
Christopher Fauletf95b1112016-12-21 08:58:16 +01001325
1326 if (nbargs != 1)
1327 goto skip_message;
1328
Christopher Faulet4ff3e572017-02-24 14:31:11 +01001329 if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
Christopher Fauletf95b1112016-12-21 08:58:16 +01001330 goto stop_processing;
Christopher Faulet4ff3e572017-02-24 14:31:11 +01001331 if (spoe_decode_data(&p, end, &smp) == -1)
Christopher Fauletf95b1112016-12-21 08:58:16 +01001332 goto skip_message;
Christopher Fauletf95b1112016-12-21 08:58:16 +01001333
Christopher Faulet4ff3e572017-02-24 14:31:11 +01001334 if (smp.data.type == SMP_T_IPV4)
1335 check_ipv4_reputation(frame, &smp.data.u.ipv4);
1336 if (smp.data.type == SMP_T_IPV6)
1337 check_ipv6_reputation(frame, &smp.data.u.ipv6);
Christopher Fauletf95b1112016-12-21 08:58:16 +01001338 }
1339 else {
1340 skip_message:
Christopher Faulet4ff3e572017-02-24 14:31:11 +01001341 p = frame->buf + frame->offset; /* Restore index */
Christopher Fauletf95b1112016-12-21 08:58:16 +01001342
1343 while (nbargs-- > 0) {
1344 /* Silently ignore argument: its name and its value */
Christopher Faulet4ff3e572017-02-24 14:31:11 +01001345 if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
Christopher Fauletf95b1112016-12-21 08:58:16 +01001346 goto stop_processing;
Christopher Faulet4ff3e572017-02-24 14:31:11 +01001347 if (spoe_skip_data(&p, end) == -1)
Christopher Fauletf95b1112016-12-21 08:58:16 +01001348 goto stop_processing;
Christopher Fauletf95b1112016-12-21 08:58:16 +01001349 }
1350 }
1351 }
1352
1353 stop_processing:
1354 /* Prepare agent ACK frame */
Christopher Faulet85010352017-02-02 10:14:36 +01001355 frame->buf = (char *)(frame->data) + 4;
Christopher Fauletf95b1112016-12-21 08:58:16 +01001356 frame->offset = 0;
Christopher Faulet85010352017-02-02 10:14:36 +01001357 frame->len = 0;
1358 frame->flags = 0;
Christopher Faulet4ff3e572017-02-24 14:31:11 +01001359
1360 ret = prepare_agentack(frame);
Christopher Faulet94bb4c62017-09-26 11:49:23 +02001361 p = frame->buf + ret;
1362 end = frame->buf+max_frame_size;
Christopher Fauletf95b1112016-12-21 08:58:16 +01001363
1364 if (frame->ip_score != -1) {
1365 DEBUG(frame->worker, "Add action : set variable ip_scode=%u",
1366 frame->ip_score);
1367
Christopher Faulet4ff3e572017-02-24 14:31:11 +01001368 *p++ = SPOE_ACT_T_SET_VAR; /* Action type */
1369 *p++ = 3; /* Number of args */
1370 *p++ = SPOE_SCOPE_SESS; /* Arg 1: the scope */
1371 spoe_encode_buffer("ip_score", 8, &p, end); /* Arg 2: variable name */
1372 *p++ = SPOE_DATA_T_UINT32;
Thierry FOURNIER6ab2bae2017-04-19 11:49:44 +02001373 encode_varint(frame->ip_score, &p, end); /* Arg 3: variable value */
Christopher Faulet4ff3e572017-02-24 14:31:11 +01001374 frame->len = (p - frame->buf);
Christopher Fauletf95b1112016-12-21 08:58:16 +01001375 }
1376 write_frame(NULL, frame);
1377}
1378
1379static void
1380read_frame_cb(evutil_socket_t fd, short events, void *arg)
1381{
1382 struct client *client = arg;
1383 struct spoe_frame *frame;
1384 uint32_t netint;
1385 int n;
1386
1387 DEBUG(client->worker, "<%lu> %s", client->id, __FUNCTION__);
1388 if ((frame = acquire_incoming_frame(client)) == NULL)
1389 goto close;
1390
1391 frame->type = SPOA_FRM_T_HAPROXY;
1392 if (frame->buf == (char *)(frame->data)) {
1393 /* Read the frame length: frame->buf points on length part (frame->data) */
1394 n = read(client->fd, frame->buf+frame->offset, 4-frame->offset);
1395 if (n <= 0) {
1396 if (n < 0)
1397 LOG(client->worker, "Failed to read frame length : %m");
Christopher Fauletba7bc162016-11-07 21:07:38 +01001398 goto close;
Christopher Faulet010fded2016-11-03 22:49:37 +01001399 }
Christopher Fauletf95b1112016-12-21 08:58:16 +01001400 frame->offset += n;
1401 if (frame->offset != 4)
1402 return;
1403 memcpy(&netint, frame->buf, 4);
1404 frame->buf += 4;
1405 frame->offset = 0;
1406 frame->len = ntohl(netint);
1407 }
Christopher Faulet010fded2016-11-03 22:49:37 +01001408
Christopher Fauletf95b1112016-12-21 08:58:16 +01001409 /* Read the frame: frame->buf points on frame part (frame->data+4)*/
1410 n = read(client->fd, frame->buf + frame->offset,
1411 frame->len - frame->offset);
1412 if (n <= 0) {
1413 if (n < 0) {
1414 LOG(client->worker, "Frame to read frame : %m");
Christopher Faulet010fded2016-11-03 22:49:37 +01001415 goto close;
1416 }
Christopher Fauletf95b1112016-12-21 08:58:16 +01001417 return;
1418 }
1419 frame->offset += n;
1420 if (frame->offset != frame->len)
1421 return;
1422 frame->offset = 0;
1423
1424 DEBUG(client->worker, "<%lu> New Frame of %u bytes received",
1425 client->id, frame->len);
1426
1427 switch (client->state) {
1428 case SPOA_ST_CONNECTING:
1429 if (handle_hahello(frame) < 0) {
1430 LOG(client->worker, "Failed to decode HELLO frame");
1431 goto disconnect;
1432 }
1433 prepare_agenthello(frame);
1434 goto write_frame;
1435
1436 case SPOA_ST_PROCESSING:
Christopher Faulet85010352017-02-02 10:14:36 +01001437 if (frame->buf[0] == SPOE_FRM_T_HAPROXY_DISCON) {
Christopher Fauletf95b1112016-12-21 08:58:16 +01001438 client->state = SPOA_ST_DISCONNECTING;
1439 goto disconnecting;
1440 }
Christopher Fauletf032c3e2017-02-17 15:18:35 +01001441 if (frame->buf[0] == SPOE_FRM_T_UNSET)
1442 n = handle_hafrag(frame);
1443 else
1444 n = handle_hanotify(frame);
1445
Christopher Fauletf95b1112016-12-21 08:58:16 +01001446 if (n < 0) {
Christopher Faulet85010352017-02-02 10:14:36 +01001447 LOG(client->worker, "Failed to decode frame: %s",
1448 spoe_frm_err_reasons[client->status_code]);
1449 goto disconnect;
Christopher Fauletf95b1112016-12-21 08:58:16 +01001450 }
Christopher Faulet8eda93f2017-02-09 09:44:33 +01001451 else if (n == 0) {
Christopher Faulet85010352017-02-02 10:14:36 +01001452 LOG(client->worker, "Ignore invalid/unknown/aborted frame");
1453 goto ignore_frame;
Christopher Fauletf95b1112016-12-21 08:58:16 +01001454 }
Christopher Fauletf032c3e2017-02-17 15:18:35 +01001455 else if (n == 1)
1456 goto noop;
Christopher Fauletf95b1112016-12-21 08:58:16 +01001457 else
1458 goto process_frame;
1459
1460 case SPOA_ST_DISCONNECTING:
1461 disconnecting:
1462 if (handle_hadiscon(frame) < 0) {
1463 LOG(client->worker, "Failed to decode DISCONNECT frame");
1464 goto disconnect;
1465 }
1466 if (client->status_code != SPOE_FRM_ERR_NONE)
1467 LOG(client->worker, "<%lu> Peer closed connection: %s",
1468 client->id, spoe_frm_err_reasons[client->status_code]);
1469 client->status_code = SPOE_FRM_ERR_NONE;
1470 goto disconnect;
1471 }
1472
Christopher Fauletf032c3e2017-02-17 15:18:35 +01001473 noop:
1474 return;
1475
Christopher Fauletf95b1112016-12-21 08:58:16 +01001476 ignore_frame:
1477 reset_frame(frame);
1478 return;
1479
1480 process_frame:
1481 process_incoming_frame(frame);
1482 client->incoming_frame = NULL;
1483 return;
1484
1485 write_frame:
1486 write_frame(client, frame);
1487 client->incoming_frame = NULL;
1488 return;
1489
1490 disconnect:
1491 client->state = SPOA_ST_DISCONNECTING;
1492 if (prepare_agentdicon(frame) < 0) {
1493 LOG(client->worker, "Failed to encode DISCONNECT frame");
1494 goto close;
1495 }
1496 goto write_frame;
1497
1498 close:
1499 release_client(client);
1500}
1501
1502static void
1503write_frame_cb(evutil_socket_t fd, short events, void *arg)
1504{
1505 struct client *client = arg;
1506 struct spoe_frame *frame;
1507 int n;
1508
1509 DEBUG(client->worker, "<%lu> %s", client->id, __FUNCTION__);
1510 if ((frame = acquire_outgoing_frame(client)) == NULL) {
1511 event_del(&client->write_frame_event);
1512 return;
1513 }
1514
1515 if (frame->buf == (char *)(frame->data)) {
1516 /* Write the frame length: frame->buf points on length part (frame->data) */
1517 n = write(client->fd, frame->buf+frame->offset, 4-frame->offset);
1518 if (n <= 0) {
1519 if (n < 0)
1520 LOG(client->worker, "Failed to write frame length : %m");
Christopher Faulet010fded2016-11-03 22:49:37 +01001521 goto close;
1522 }
Christopher Fauletf95b1112016-12-21 08:58:16 +01001523 frame->offset += n;
1524 if (frame->offset != 4)
1525 return;
1526 frame->buf += 4;
1527 frame->offset = 0;
1528 }
1529
1530 /* Write the frame: frame->buf points on frame part (frame->data+4)*/
1531 n = write(client->fd, frame->buf + frame->offset,
1532 frame->len - frame->offset);
1533 if (n <= 0) {
1534 if (n < 0) {
1535 LOG(client->worker, "Failed to write frame : %m");
Christopher Faulet010fded2016-11-03 22:49:37 +01001536 goto close;
1537 }
Christopher Fauletf95b1112016-12-21 08:58:16 +01001538 return;
1539 }
1540 frame->offset += n;
1541 if (frame->offset != frame->len)
1542 return;
1543
1544 DEBUG(client->worker, "<%lu> Frame of %u bytes send",
1545 client->id, frame->len);
1546
1547 switch (client->state) {
1548 case SPOA_ST_CONNECTING:
1549 if (frame->hcheck == true) {
1550 DEBUG(client->worker,
1551 "<%lu> Close client after healthcheck",
1552 client->id);
1553 goto close;
1554 }
1555 client->state = SPOA_ST_PROCESSING;
1556 break;
1557
1558 case SPOA_ST_PROCESSING:
1559 break;
1560
1561 case SPOA_ST_DISCONNECTING:
1562 goto close;
1563 }
1564
1565 release_frame(frame);
1566 client->outgoing_frame = NULL;
1567 if (!client->async && !client->pipelining) {
1568 event_del(&client->write_frame_event);
1569 event_add(&client->read_frame_event, NULL);
1570 }
1571 return;
1572
1573 close:
1574 release_client(client);
1575}
1576
1577static void
1578accept_cb(int listener, short event, void *arg)
1579{
1580 struct worker *worker;
1581 struct client *client;
1582 int fd;
1583
1584 worker = &workers[clicount++ % num_workers];
1585
1586 if ((fd = accept(listener, NULL, NULL)) < 0) {
1587 if (errno != EAGAIN && errno != EWOULDBLOCK)
1588 LOG(worker, "Failed to accept client connection : %m");
1589 return;
1590 }
1591
1592 DEBUG(&null_worker,
1593 "<%lu> New Client connection accepted and assigned to worker %02d",
1594 clicount, worker->id);
1595
1596 if (evutil_make_socket_nonblocking(fd) < 0) {
1597 LOG(&null_worker, "Failed to set client socket to non-blocking : %m");
1598 close(fd);
1599 return;
1600 }
1601
1602 if ((client = calloc(1, sizeof(*client))) == NULL) {
1603 LOG(&null_worker, "Failed to allocate memory for client state : %m");
1604 close(fd);
1605 return;
1606 }
1607
1608 client->id = clicount;
1609 client->fd = fd;
1610 client->worker = worker;
1611 client->state = SPOA_ST_CONNECTING;
1612 client->status_code = SPOE_FRM_ERR_NONE;
1613 client->max_frame_size = max_frame_size;
1614 client->engine = NULL;
1615 client->pipelining = false;
1616 client->async = false;
1617 client->incoming_frame = NULL;
1618 client->outgoing_frame = NULL;
1619 LIST_INIT(&client->processing_frames);
1620 LIST_INIT(&client->outgoing_frames);
1621
1622 LIST_ADDQ(&worker->clients, &client->by_worker);
1623
1624 worker->nbclients++;
1625
1626 if (event_assign(&client->read_frame_event, worker->base, fd,
1627 EV_READ|EV_PERSIST, read_frame_cb, client) < 0 ||
1628 event_assign(&client->write_frame_event, worker->base, fd,
1629 EV_WRITE|EV_PERSIST, write_frame_cb, client) < 0) {
1630 LOG(&null_worker, "Failed to create client events");
1631 release_client(client);
1632 return;
1633 }
1634 event_add(&client->read_frame_event, NULL);
1635}
1636
1637static void *
1638worker_function(void *data)
1639{
1640 struct client *client, *cback;
1641 struct spoe_frame *frame, *fback;
1642 struct worker *worker = data;
1643
1644 DEBUG(worker, "Worker ready to process client messages");
1645 event_base_dispatch(worker->base);
Christopher Faulet010fded2016-11-03 22:49:37 +01001646
Christopher Fauletf95b1112016-12-21 08:58:16 +01001647 list_for_each_entry_safe(client, cback, &worker->clients, by_worker) {
1648 release_client(client);
Christopher Faulet010fded2016-11-03 22:49:37 +01001649 }
1650
Christopher Fauletf95b1112016-12-21 08:58:16 +01001651 list_for_each_entry_safe(frame, fback, &worker->frames, list) {
1652 LIST_DEL(&frame->list);
1653 free(frame);
1654 }
1655
1656 event_free(worker->monitor_event);
1657 event_base_free(worker->base);
1658 DEBUG(worker, "Worker is stopped");
1659 pthread_exit(&null_worker);
1660}
1661
1662
1663static int
1664parse_processing_delay(const char *str)
1665{
1666 unsigned long value;
1667
1668 value = 0;
1669 while (1) {
1670 unsigned int j;
1671
1672 j = *str - '0';
1673 if (j > 9)
1674 break;
1675 str++;
1676 value *= 10;
1677 value += j;
1678 }
1679
1680 switch (*str) {
1681 case '\0': /* no unit = millisecond */
1682 value *= 1000;
1683 break;
1684 case 's': /* second */
1685 value *= 1000000;
1686 str++;
1687 break;
1688 case 'm': /* millisecond : "ms" */
1689 if (str[1] != 's')
1690 return -1;
1691 value *= 1000;
1692 str += 2;
1693 break;
1694 case 'u': /* microsecond : "us" */
1695 if (str[1] != 's')
1696 return -1;
1697 str += 2;
1698 break;
1699 default:
1700 return -1;
1701 }
1702 if (*str)
1703 return -1;
1704
1705 processing_delay.tv_sec = (time_t)(value / 1000000);
1706 processing_delay.tv_usec = (suseconds_t)(value % 1000000);
1707 return 0;
Christopher Faulet010fded2016-11-03 22:49:37 +01001708}
1709
Christopher Fauletf95b1112016-12-21 08:58:16 +01001710
Christopher Faulet010fded2016-11-03 22:49:37 +01001711static void
1712usage(char *prog)
1713{
Christopher Fauletf95b1112016-12-21 08:58:16 +01001714 fprintf(stderr,
1715 "Usage : %s [OPTION]...\n"
1716 " -h Print this message\n"
1717 " -d Enable the debug mode\n"
1718 " -m <max-frame-size> Specify the maximum frame size (default : %u)\n"
1719 " -p <port> Specify the port to listen on (default : %d)\n"
1720 " -n <num-workers> Specify the number of workers (default : %d)\n"
1721 " -c <capability> Enable the support of the specified capability\n"
1722 " -t <time> Set a delay to process a message (default: 0)\n"
1723 " The value is specified in milliseconds by default,\n"
1724 " but can be in any other unit if the number is suffixed\n"
1725 " by a unit (us, ms, s)\n"
1726 "\n"
Christopher Faulet85010352017-02-02 10:14:36 +01001727 " Supported capabilities: fragmentation, pipelining, async\n",
Christopher Fauletf95b1112016-12-21 08:58:16 +01001728 prog, MAX_FRAME_SIZE, DEFAULT_PORT, NUM_WORKERS);
Christopher Faulet010fded2016-11-03 22:49:37 +01001729}
1730
1731int
1732main(int argc, char **argv)
1733{
Christopher Fauletf95b1112016-12-21 08:58:16 +01001734 struct event_base *base = NULL;
1735 struct event *signal_event = NULL, *accept_event = NULL;
1736 int opt, i, fd = -1;
Christopher Faulet010fded2016-11-03 22:49:37 +01001737
Christopher Fauletf95b1112016-12-21 08:58:16 +01001738 // TODO: add '-t <processing-time>' option
1739 while ((opt = getopt(argc, argv, "hdm:n:p:c:t:")) != -1) {
Christopher Faulet010fded2016-11-03 22:49:37 +01001740 switch (opt) {
1741 case 'h':
1742 usage(argv[0]);
1743 return EXIT_SUCCESS;
1744 case 'd':
1745 debug = true;
1746 break;
Christopher Fauletf95b1112016-12-21 08:58:16 +01001747 case 'm':
1748 max_frame_size = atoi(optarg);
1749 break;
Christopher Faulet010fded2016-11-03 22:49:37 +01001750 case 'n':
Christopher Fauletf95b1112016-12-21 08:58:16 +01001751 num_workers = atoi(optarg);
Christopher Faulet010fded2016-11-03 22:49:37 +01001752 break;
1753 case 'p':
Christopher Fauletf95b1112016-12-21 08:58:16 +01001754 server_port = atoi(optarg);
Christopher Faulet010fded2016-11-03 22:49:37 +01001755 break;
Christopher Fauletf95b1112016-12-21 08:58:16 +01001756 case 'c':
1757 if (!strcmp(optarg, "pipelining"))
1758 pipelining = true;
1759 else if (!strcmp(optarg, "async"))
1760 async = true;
Christopher Faulet85010352017-02-02 10:14:36 +01001761 else if (!strcmp(optarg, "fragmentation"))
1762 fragmentation = true;
Christopher Fauletf95b1112016-12-21 08:58:16 +01001763 else
1764 fprintf(stderr, "WARNING: unsupported capability '%s'\n", optarg);
1765 break;
1766 case 't':
1767 if (!parse_processing_delay(optarg))
1768 break;
1769 fprintf(stderr, "%s: failed to parse time '%s'.\n", argv[0], optarg);
1770 fprintf(stderr, "Try '%s -h' for more information.\n", argv[0]);
1771 return EXIT_FAILURE;
Christopher Faulet010fded2016-11-03 22:49:37 +01001772 default:
1773 usage(argv[0]);
1774 return EXIT_FAILURE;
1775 }
1776 }
1777
Christopher Fauletf95b1112016-12-21 08:58:16 +01001778 if (num_workers <= 0) {
1779 LOG(&null_worker, "%s : Invalid number of workers '%d'\n",
1780 argv[0], num_workers);
Christopher Faulet010fded2016-11-03 22:49:37 +01001781 goto error;
1782 }
Christopher Fauletf95b1112016-12-21 08:58:16 +01001783
1784 if (server_port <= 0) {
1785 LOG(&null_worker, "%s : Invalid port '%d'\n",
1786 argv[0], server_port);
Christopher Faulet010fded2016-11-03 22:49:37 +01001787 goto error;
1788 }
1789
Christopher Fauletf95b1112016-12-21 08:58:16 +01001790
1791 if (evthread_use_pthreads() < 0) {
1792 LOG(&null_worker, "No pthreads support for libevent");
Christopher Faulet010fded2016-11-03 22:49:37 +01001793 goto error;
1794 }
1795
Christopher Fauletf95b1112016-12-21 08:58:16 +01001796 if ((base = event_base_new()) == NULL) {
1797 LOG(&null_worker, "Failed to initialize libevent : %m");
1798 goto error;
1799 }
Christopher Faulet010fded2016-11-03 22:49:37 +01001800
Christopher Fauletf95b1112016-12-21 08:58:16 +01001801 signal(SIGPIPE, SIG_IGN);
Christopher Faulet010fded2016-11-03 22:49:37 +01001802
Christopher Fauletf95b1112016-12-21 08:58:16 +01001803 if ((fd = create_server_socket()) < 0) {
1804 LOG(&null_worker, "Failed to create server socket");
1805 goto error;
1806 }
1807 if (evutil_make_socket_nonblocking(fd) < 0) {
1808 LOG(&null_worker, "Failed to set server socket to non-blocking");
Christopher Faulet010fded2016-11-03 22:49:37 +01001809 goto error;
1810 }
1811
Christopher Fauletf95b1112016-12-21 08:58:16 +01001812 if ((workers = calloc(num_workers, sizeof(*workers))) == NULL) {
1813 LOG(&null_worker, "Failed to set allocate memory for workers");
Christopher Faulet010fded2016-11-03 22:49:37 +01001814 goto error;
1815 }
Christopher Fauletf95b1112016-12-21 08:58:16 +01001816
1817 for (i = 0; i < num_workers; ++i) {
1818 struct worker *w = &workers[i];
Christopher Faulet010fded2016-11-03 22:49:37 +01001819
Christopher Fauletf95b1112016-12-21 08:58:16 +01001820 w->id = i+1;
1821 w->nbclients = 0;
1822 LIST_INIT(&w->engines);
1823 LIST_INIT(&w->clients);
1824 LIST_INIT(&w->frames);
1825
1826 if ((w->base = event_base_new()) == NULL) {
1827 LOG(&null_worker,
1828 "Failed to initialize libevent for worker %02d : %m",
1829 w->id);
1830 goto error;
1831 }
Christopher Faulet010fded2016-11-03 22:49:37 +01001832
Christopher Fauletf95b1112016-12-21 08:58:16 +01001833 w->monitor_event = event_new(w->base, fd, EV_PERSIST,
1834 worker_monitor_cb, (void *)w);
1835 if (w->monitor_event == NULL ||
1836 event_add(w->monitor_event, (struct timeval[]){{5,0}}) < 0) {
1837 LOG(&null_worker,
1838 "Failed to create monitor event for worker %02d",
1839 w->id);
Christopher Faulet010fded2016-11-03 22:49:37 +01001840 goto error;
1841 }
Christopher Fauletf95b1112016-12-21 08:58:16 +01001842
1843 if (pthread_create(&w->thread, NULL, worker_function, (void *)w)) {
1844 LOG(&null_worker,
1845 "Failed to start thread for worker %02d : %m",
1846 w->id);
1847 }
1848 DEBUG(&null_worker, "Worker %02d initialized", w->id);
1849 }
1850
1851 accept_event = event_new(base, fd, EV_READ|EV_PERSIST, accept_cb,
1852 (void *)base);
1853 if (accept_event == NULL || event_add(accept_event, NULL) < 0) {
1854 LOG(&null_worker, "Failed to create accept event : %m");
1855 }
1856
1857 signal_event = evsignal_new(base, SIGINT, signal_cb, (void *)base);
1858 if (signal_event == NULL || event_add(signal_event, NULL) < 0) {
1859 LOG(&null_worker, "Failed to create signal event : %m");
Christopher Faulet010fded2016-11-03 22:49:37 +01001860 }
1861
Christopher Fauletf95b1112016-12-21 08:58:16 +01001862 DEBUG(&null_worker,
1863 "Server is ready"
Christopher Faulet85010352017-02-02 10:14:36 +01001864 " [fragmentation=%s - pipelining=%s - async=%s - debug=%s - max-frame-size=%u]",
1865 (fragmentation?"true":"false"), (pipelining?"true":"false"), (async?"true":"false"),
Christopher Fauletf95b1112016-12-21 08:58:16 +01001866 (debug?"true":"false"), max_frame_size);
1867 event_base_dispatch(base);
1868
1869 for (i = 0; i < num_workers; i++) {
1870 struct worker *w = &workers[i];
1871
1872 pthread_join(w->thread, NULL);
1873 DEBUG(&null_worker, "Worker %02d terminated", w->id);
Christopher Faulet010fded2016-11-03 22:49:37 +01001874 }
Christopher Fauletf95b1112016-12-21 08:58:16 +01001875
1876 free(workers);
1877 event_free(signal_event);
1878 event_free(accept_event);
1879 event_base_free(base);
1880 close(fd);
Christopher Faulet010fded2016-11-03 22:49:37 +01001881 return EXIT_SUCCESS;
Christopher Fauletf95b1112016-12-21 08:58:16 +01001882
1883 error:
1884 if (workers != NULL)
1885 free(workers);
1886 if (signal_event != NULL)
1887 event_free(signal_event);
1888 if (accept_event != NULL)
1889 event_free(accept_event);
1890 if (base != NULL)
1891 event_base_free(base);
1892 if (fd != -1)
1893 close(fd);
Christopher Faulet010fded2016-11-03 22:49:37 +01001894 return EXIT_FAILURE;
1895}