1 /*
2 * Listener management functions.
3 *
4 * Copyright 2000-2013 Willy Tarreau <w@1wt.eu>
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 <ctype.h>
14 #include <errno.h>
15 #include <stdio.h>
16 #include <string.h>
17 #include <unistd.h>
18 #include <fcntl.h>
19
20 #include <haproxy/acl.h>
21 #include <haproxy/api.h>
22 #include <haproxy/cfgparse.h>
23 #include <haproxy/connection.h>
24 #include <haproxy/errors.h>
25 #include <haproxy/fd.h>
26 #include <haproxy/freq_ctr.h>
27 #include <haproxy/global.h>
28 #include <haproxy/list.h>
29 #include <haproxy/listener.h>
30 #include <haproxy/log.h>
31 #include <haproxy/protocol.h>
32 #include <haproxy/proxy.h>
33 #include <haproxy/sample.h>
34 #include <haproxy/stream.h>
35 #include <haproxy/task.h>
36 #include <haproxy/time.h>
37 #include <haproxy/tools.h>
38
39
40 /* List head of all known bind keywords */
41 static struct bind_kw_list bind_keywords = {
42 .list = LIST_HEAD_INIT(bind_keywords.list)
43 };
44
45 /* list of the temporarily limited listeners because of lack of resource */
46 static struct mt_list global_listener_queue = MT_LIST_HEAD_INIT(global_listener_queue);
47 static struct task *global_listener_queue_task;
48
49 /* listener status for stats */
50 const char* li_status_st[LI_STATE_COUNT] = {
51 [LI_STATUS_WAITING] = "WAITING",
52 [LI_STATUS_OPEN] = "OPEN",
53 [LI_STATUS_FULL] = "FULL",
54 };
55
56 #if defined(USE_THREAD)
57
58 struct accept_queue_ring accept_queue_rings[MAX_THREADS] __attribute__((aligned(64))) = { };
59
60 /* dequeue and process a pending connection from the local accept queue (single
61 * consumer). Returns the accepted connection or NULL if none was found.
62 */
accept_queue_pop_sc(struct accept_queue_ring * ring)63 struct connection *accept_queue_pop_sc(struct accept_queue_ring *ring)
64 {
65 unsigned int pos, next;
66 struct connection *ptr;
67 struct connection **e;
68
69 pos = ring->head;
70
71 if (pos == ring->tail)
72 return NULL;
73
74 next = pos + 1;
75 if (next >= ACCEPT_QUEUE_SIZE)
76 next = 0;
77
78 e = &ring->entry[pos];
79
80 /* wait for the producer to update the listener's pointer */
81 while (1) {
82 ptr = *e;
83 __ha_barrier_load();
84 if (ptr)
85 break;
86 pl_cpu_relax();
87 }
88
89 /* release the entry */
90 *e = NULL;
91
92 __ha_barrier_store();
93 ring->head = next;
94 return ptr;
95 }
96
97
98 /* tries to push a new accepted connection <conn> into ring <ring>. Returns
99 * non-zero if it succeeds, or zero if the ring is full. Supports multiple
100 * producers.
101 */
accept_queue_push_mp(struct accept_queue_ring * ring,struct connection * conn)102 int accept_queue_push_mp(struct accept_queue_ring *ring, struct connection *conn)
103 {
104 unsigned int pos, next;
105
106 pos = ring->tail;
107 do {
108 next = pos + 1;
109 if (next >= ACCEPT_QUEUE_SIZE)
110 next = 0;
111 if (next == ring->head)
112 return 0; // ring full
113 } while (unlikely(!_HA_ATOMIC_CAS(&ring->tail, &pos, next)));
114
115 ring->entry[pos] = conn;
116 __ha_barrier_store();
117 return 1;
118 }
119
120 /* proceed with accepting new connections. Don't mark it static so that it appears
121 * in task dumps.
122 */
accept_queue_process(struct task * t,void * context,unsigned int state)123 struct task *accept_queue_process(struct task *t, void *context, unsigned int state)
124 {
125 struct accept_queue_ring *ring = context;
126 struct connection *conn;
127 struct listener *li;
128 unsigned int max_accept;
129 int ret;
130
131 /* if global.tune.maxaccept is -1, then max_accept is UINT_MAX. It
132 * is not really illimited, but it is probably enough.
133 */
134 max_accept = global.tune.maxaccept ? global.tune.maxaccept : MAX_ACCEPT;
135 for (; max_accept; max_accept--) {
136 conn = accept_queue_pop_sc(ring);
137 if (!conn)
138 break;
139
140 li = __objt_listener(conn->target);
141 _HA_ATOMIC_INC(&li->thr_conn[tid]);
142 ret = li->accept(conn);
143 if (ret <= 0) {
144 /* connection was terminated by the application */
145 continue;
146 }
147
148 /* increase the per-process number of cumulated sessions, this
149 * may only be done once l->accept() has accepted the connection.
150 */
151 if (!(li->options & LI_O_UNLIMITED)) {
152 HA_ATOMIC_UPDATE_MAX(&global.sps_max,
153 update_freq_ctr(&global.sess_per_sec, 1));
154 if (li->bind_conf && li->bind_conf->is_ssl) {
155 HA_ATOMIC_UPDATE_MAX(&global.ssl_max,
156 update_freq_ctr(&global.ssl_per_sec, 1));
157 }
158 }
159 }
160
161 /* ran out of budget ? Let's come here ASAP */
162 if (!max_accept)
163 tasklet_wakeup(ring->tasklet);
164
165 return NULL;
166 }
167
168 /* Initializes the accept-queues. Returns 0 on success, otherwise ERR_* flags */
accept_queue_init()169 static int accept_queue_init()
170 {
171 struct tasklet *t;
172 int i;
173
174 for (i = 0; i < global.nbthread; i++) {
175 t = tasklet_new();
176 if (!t) {
177 ha_alert("Out of memory while initializing accept queue for thread %d\n", i);
178 return ERR_FATAL|ERR_ABORT;
179 }
180 t->tid = i;
181 t->process = accept_queue_process;
182 t->context = &accept_queue_rings[i];
183 accept_queue_rings[i].tasklet = t;
184 }
185 return 0;
186 }
187
188 REGISTER_CONFIG_POSTPARSER("multi-threaded accept queue", accept_queue_init);
189
190 #endif // USE_THREAD
191
192 /* helper to get listener status for stats */
get_li_status(struct listener * l)193 enum li_status get_li_status(struct listener *l)
194 {
195 if (!l->maxconn || l->nbconn < l->maxconn) {
196 if (l->state == LI_LIMITED)
197 return LI_STATUS_WAITING;
198 else
199 return LI_STATUS_OPEN;
200 }
201 return LI_STATUS_FULL;
202 }
203
204 /* adjust the listener's state and its proxy's listener counters if needed.
205 * It must be called under the listener's lock, but uses atomic ops to change
206 * the proxy's counters so that the proxy lock is not needed.
207 */
listener_set_state(struct listener * l,enum li_state st)208 void listener_set_state(struct listener *l, enum li_state st)
209 {
210 struct proxy *px = l->bind_conf->frontend;
211
212 if (px) {
213 /* from state */
214 switch (l->state) {
215 case LI_NEW: /* first call */
216 _HA_ATOMIC_INC(&px->li_all);
217 break;
218 case LI_INIT:
219 case LI_ASSIGNED:
220 break;
221 case LI_PAUSED:
222 _HA_ATOMIC_DEC(&px->li_paused);
223 break;
224 case LI_LISTEN:
225 _HA_ATOMIC_DEC(&px->li_bound);
226 break;
227 case LI_READY:
228 case LI_FULL:
229 case LI_LIMITED:
230 _HA_ATOMIC_DEC(&px->li_ready);
231 break;
232 }
233
234 /* to state */
235 switch (st) {
236 case LI_NEW:
237 case LI_INIT:
238 case LI_ASSIGNED:
239 break;
240 case LI_PAUSED:
241 BUG_ON(l->rx.fd == -1);
242 _HA_ATOMIC_INC(&px->li_paused);
243 break;
244 case LI_LISTEN:
245 BUG_ON(l->rx.fd == -1);
246 _HA_ATOMIC_INC(&px->li_bound);
247 break;
248 case LI_READY:
249 case LI_FULL:
250 case LI_LIMITED:
251 BUG_ON(l->rx.fd == -1);
252 _HA_ATOMIC_INC(&px->li_ready);
253 break;
254 }
255 }
256 l->state = st;
257 }
258
259 /* This function adds the specified listener's file descriptor to the polling
260 * lists if it is in the LI_LISTEN state. The listener enters LI_READY or
261 * LI_FULL state depending on its number of connections. In daemon mode, we
262 * also support binding only the relevant processes to their respective
263 * listeners. We don't do that in debug mode however.
264 */
enable_listener(struct listener * listener)265 void enable_listener(struct listener *listener)
266 {
267 HA_SPIN_LOCK(LISTENER_LOCK, &listener->lock);
268
269 /* If this listener is supposed to be only in the master, close it in
270 * the workers. Conversely, if it's supposed to be only in the workers
271 * close it in the master.
272 */
273 if (!!master != !!(listener->rx.flags & RX_F_MWORKER))
274 do_unbind_listener(listener);
275
276 if (listener->state == LI_LISTEN) {
277 BUG_ON(listener->rx.fd == -1);
278 if ((global.mode & (MODE_DAEMON | MODE_MWORKER)) &&
279 (!!master != !!(listener->rx.flags & RX_F_MWORKER) ||
280 !(proc_mask(listener->rx.settings->bind_proc) & pid_bit))) {
281 /* we don't want to enable this listener and don't
282 * want any fd event to reach it.
283 */
284 do_unbind_listener(listener);
285 }
286 else if (!listener->maxconn || listener->nbconn < listener->maxconn) {
287 listener->rx.proto->enable(listener);
288 listener_set_state(listener, LI_READY);
289 }
290 else {
291 listener_set_state(listener, LI_FULL);
292 }
293 }
294
295 HA_SPIN_UNLOCK(LISTENER_LOCK, &listener->lock);
296 }
297
298 /*
299 * This function completely stops a listener. It will need to operate under the
300 * proxy's lock, the protocol's lock, and the listener's lock. The caller is
301 * responsible for indicating in lpx, lpr, lli whether the respective locks are
302 * already held (non-zero) or not (zero) so that the function picks the missing
303 * ones, in this order. The proxy's listeners count is updated and the proxy is
304 * disabled and woken up after the last one is gone.
305 */
stop_listener(struct listener * l,int lpx,int lpr,int lli)306 void stop_listener(struct listener *l, int lpx, int lpr, int lli)
307 {
308 struct proxy *px = l->bind_conf->frontend;
309
310 if (l->options & LI_O_NOSTOP) {
311 /* master-worker sockpairs are never closed but don't count as a
312 * job.
313 */
314 return;
315 }
316
317 if (!lpx)
318 HA_RWLOCK_WRLOCK(PROXY_LOCK, &px->lock);
319
320 if (!lpr)
321 HA_SPIN_LOCK(PROTO_LOCK, &proto_lock);
322
323 if (!lli)
324 HA_SPIN_LOCK(LISTENER_LOCK, &l->lock);
325
326 if (l->state > LI_INIT) {
327 do_unbind_listener(l);
328
329 if (l->state >= LI_ASSIGNED)
330 __delete_listener(l);
331
332 proxy_cond_disable(px);
333 }
334
335 if (!lli)
336 HA_SPIN_UNLOCK(LISTENER_LOCK, &l->lock);
337
338 if (!lpr)
339 HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock);
340
341 if (!lpx)
342 HA_RWLOCK_WRUNLOCK(PROXY_LOCK, &px->lock);
343 }
344
345 /* This function adds the specified <listener> to the protocol <proto>. It
346 * does nothing if the protocol was already added. The listener's state is
347 * automatically updated from LI_INIT to LI_ASSIGNED. The number of listeners
348 * for the protocol is updated. This must be called with the proto lock held.
349 */
default_add_listener(struct protocol * proto,struct listener * listener)350 void default_add_listener(struct protocol *proto, struct listener *listener)
351 {
352 if (listener->state != LI_INIT)
353 return;
354 listener_set_state(listener, LI_ASSIGNED);
355 listener->rx.proto = proto;
356 LIST_APPEND(&proto->receivers, &listener->rx.proto_list);
357 proto->nb_receivers++;
358 }
359
360 /* default function called to suspend a listener: it simply passes the call to
361 * the underlying receiver. This is find for most socket-based protocols. This
362 * must be called under the listener's lock. It will return non-zero on success,
363 * 0 on failure. If no receiver-level suspend is provided, the operation is
364 * assumed to succeed.
365 */
default_suspend_listener(struct listener * l)366 int default_suspend_listener(struct listener *l)
367 {
368 int ret = 1;
369
370 if (!l->rx.proto->rx_suspend)
371 return 1;
372
373 ret = l->rx.proto->rx_suspend(&l->rx);
374 return ret > 0 ? ret : 0;
375 }
376
377
378 /* Tries to resume a suspended listener, and returns non-zero on success or
379 * zero on failure. On certain errors, an alert or a warning might be displayed.
380 * It must be called with the listener's lock held. Depending on the listener's
381 * state and protocol, a listen() call might be used to resume operations, or a
382 * call to the receiver's resume() function might be used as well. This is
383 * suitable as a default function for TCP and UDP. This must be called with the
384 * listener's lock held.
385 */
default_resume_listener(struct listener * l)386 int default_resume_listener(struct listener *l)
387 {
388 int ret = 1;
389
390 if (l->state == LI_ASSIGNED) {
391 char msg[100];
392 int err;
393
394 err = l->rx.proto->listen(l, msg, sizeof(msg));
395 if (err & ERR_ALERT)
396 ha_alert("Resuming listener: %s\n", msg);
397 else if (err & ERR_WARN)
398 ha_warning("Resuming listener: %s\n", msg);
399
400 if (err & (ERR_FATAL | ERR_ABORT)) {
401 ret = 0;
402 goto end;
403 }
404 }
405
406 if (l->state < LI_PAUSED) {
407 ret = 0;
408 goto end;
409 }
410
411 if (l->state == LI_PAUSED && l->rx.proto->rx_resume &&
412 l->rx.proto->rx_resume(&l->rx) <= 0)
413 ret = 0;
414 end:
415 return ret;
416 }
417
418
419 /* This function tries to temporarily disable a listener, depending on the OS
420 * capabilities. Linux unbinds the listen socket after a SHUT_RD, and ignores
421 * SHUT_WR. Solaris refuses either shutdown(). OpenBSD ignores SHUT_RD but
422 * closes upon SHUT_WR and refuses to rebind. So a common validation path
423 * involves SHUT_WR && listen && SHUT_RD. In case of success, the FD's polling
424 * is disabled. It normally returns non-zero, unless an error is reported.
425 */
pause_listener(struct listener * l)426 int pause_listener(struct listener *l)
427 {
428 struct proxy *px = l->bind_conf->frontend;
429 int ret = 1;
430
431 HA_SPIN_LOCK(LISTENER_LOCK, &l->lock);
432
433 if ((global.mode & (MODE_DAEMON | MODE_MWORKER)) &&
434 !(proc_mask(l->rx.settings->bind_proc) & pid_bit))
435 goto end;
436
437 if (l->state <= LI_PAUSED)
438 goto end;
439
440 if (l->rx.proto->suspend)
441 ret = l->rx.proto->suspend(l);
442
443 MT_LIST_DELETE(&l->wait_queue);
444
445 listener_set_state(l, LI_PAUSED);
446
447 if (px && !px->li_ready) {
448 ha_warning("Paused %s %s.\n", proxy_cap_str(px->cap), px->id);
449 send_log(px, LOG_WARNING, "Paused %s %s.\n", proxy_cap_str(px->cap), px->id);
450 }
451 end:
452 HA_SPIN_UNLOCK(LISTENER_LOCK, &l->lock);
453 return ret;
454 }
455
456 /* This function tries to resume a temporarily disabled listener. Paused, full,
457 * limited and disabled listeners are handled, which means that this function
458 * may replace enable_listener(). The resulting state will either be LI_READY
459 * or LI_FULL. 0 is returned in case of failure to resume (eg: dead socket).
460 * Listeners bound to a different process are not woken up unless we're in
461 * foreground mode, and are ignored. If the listener was only in the assigned
462 * state, it's totally rebound. This can happen if a pause() has completely
463 * stopped it. If the resume fails, 0 is returned and an error might be
464 * displayed.
465 */
resume_listener(struct listener * l)466 int resume_listener(struct listener *l)
467 {
468 struct proxy *px = l->bind_conf->frontend;
469 int was_paused = px && px->li_paused;
470 int ret = 1;
471
472 HA_SPIN_LOCK(LISTENER_LOCK, &l->lock);
473
474 /* check that another thread didn't to the job in parallel (e.g. at the
475 * end of listen_accept() while we'd come from dequeue_all_listeners().
476 */
477 if (MT_LIST_INLIST(&l->wait_queue))
478 goto end;
479
480 if ((global.mode & (MODE_DAEMON | MODE_MWORKER)) &&
481 !(proc_mask(l->rx.settings->bind_proc) & pid_bit))
482 goto end;
483
484 if (l->state == LI_READY)
485 goto end;
486
487 if (l->rx.proto->resume)
488 ret = l->rx.proto->resume(l);
489
490 if (l->maxconn && l->nbconn >= l->maxconn) {
491 l->rx.proto->disable(l);
492 listener_set_state(l, LI_FULL);
493 goto done;
494 }
495
496 l->rx.proto->enable(l);
497 listener_set_state(l, LI_READY);
498
499 done:
500 if (was_paused && !px->li_paused) {
501 ha_warning("Resumed %s %s.\n", proxy_cap_str(px->cap), px->id);
502 send_log(px, LOG_WARNING, "Resumed %s %s.\n", proxy_cap_str(px->cap), px->id);
503 }
504 end:
505 HA_SPIN_UNLOCK(LISTENER_LOCK, &l->lock);
506 return ret;
507 }
508
509 /* Marks a ready listener as full so that the stream code tries to re-enable
510 * it upon next close() using resume_listener().
511 */
listener_full(struct listener * l)512 static void listener_full(struct listener *l)
513 {
514 HA_SPIN_LOCK(LISTENER_LOCK, &l->lock);
515 if (l->state >= LI_READY) {
516 MT_LIST_DELETE(&l->wait_queue);
517 if (l->state != LI_FULL) {
518 l->rx.proto->disable(l);
519 listener_set_state(l, LI_FULL);
520 }
521 }
522 HA_SPIN_UNLOCK(LISTENER_LOCK, &l->lock);
523 }
524
525 /* Marks a ready listener as limited so that we only try to re-enable it when
526 * resources are free again. It will be queued into the specified queue.
527 */
limit_listener(struct listener * l,struct mt_list * list)528 static void limit_listener(struct listener *l, struct mt_list *list)
529 {
530 HA_SPIN_LOCK(LISTENER_LOCK, &l->lock);
531 if (l->state == LI_READY) {
532 MT_LIST_TRY_APPEND(list, &l->wait_queue);
533 l->rx.proto->disable(l);
534 listener_set_state(l, LI_LIMITED);
535 }
536 HA_SPIN_UNLOCK(LISTENER_LOCK, &l->lock);
537 }
538
539 /* Dequeues all listeners waiting for a resource the global wait queue */
dequeue_all_listeners()540 void dequeue_all_listeners()
541 {
542 struct listener *listener;
543
544 while ((listener = MT_LIST_POP(&global_listener_queue, struct listener *, wait_queue))) {
545 /* This cannot fail because the listeners are by definition in
546 * the LI_LIMITED state.
547 */
548 resume_listener(listener);
549 }
550 }
551
552 /* Dequeues all listeners waiting for a resource in proxy <px>'s queue */
dequeue_proxy_listeners(struct proxy * px)553 void dequeue_proxy_listeners(struct proxy *px)
554 {
555 struct listener *listener;
556
557 while ((listener = MT_LIST_POP(&px->listener_queue, struct listener *, wait_queue))) {
558 /* This cannot fail because the listeners are by definition in
559 * the LI_LIMITED state.
560 */
561 resume_listener(listener);
562 }
563 }
564
565
566 /* default function used to unbind a listener. This is for use by standard
567 * protocols working on top of accepted sockets. The receiver's rx_unbind()
568 * will automatically be used after the listener is disabled if the socket is
569 * still bound. This must be used under the listener's lock.
570 */
default_unbind_listener(struct listener * listener)571 void default_unbind_listener(struct listener *listener)
572 {
573 if (listener->state <= LI_ASSIGNED)
574 goto out_close;
575
576 if (listener->rx.fd == -1) {
577 listener_set_state(listener, LI_ASSIGNED);
578 goto out_close;
579 }
580
581 if (listener->state >= LI_READY) {
582 listener->rx.proto->disable(listener);
583 if (listener->rx.flags & RX_F_BOUND)
584 listener_set_state(listener, LI_LISTEN);
585 }
586
587 out_close:
588 if (listener->rx.flags & RX_F_BOUND)
589 listener->rx.proto->rx_unbind(&listener->rx);
590 }
591
592 /* This function closes the listening socket for the specified listener,
593 * provided that it's already in a listening state. The protocol's unbind()
594 * is called to put the listener into LI_ASSIGNED or LI_LISTEN and handle
595 * the unbinding tasks. The listener enters then the LI_ASSIGNED state if
596 * the receiver is unbound. Must be called with the lock held.
597 */
do_unbind_listener(struct listener * listener)598 void do_unbind_listener(struct listener *listener)
599 {
600 MT_LIST_DELETE(&listener->wait_queue);
601
602 if (listener->rx.proto->unbind)
603 listener->rx.proto->unbind(listener);
604
605 /* we may have to downgrade the listener if the rx was closed */
606 if (!(listener->rx.flags & RX_F_BOUND) && listener->state > LI_ASSIGNED)
607 listener_set_state(listener, LI_ASSIGNED);
608 }
609
610 /* This function closes the listening socket for the specified listener,
611 * provided that it's already in a listening state. The listener enters the
612 * LI_ASSIGNED state, except if the FD is not closed, in which case it may
613 * remain in LI_LISTEN. This function is intended to be used as a generic
614 * function for standard protocols.
615 */
unbind_listener(struct listener * listener)616 void unbind_listener(struct listener *listener)
617 {
618 HA_SPIN_LOCK(LISTENER_LOCK, &listener->lock);
619 do_unbind_listener(listener);
620 HA_SPIN_UNLOCK(LISTENER_LOCK, &listener->lock);
621 }
622
623 /* creates one or multiple listeners for bind_conf <bc> on sockaddr <ss> on port
624 * range <portl> to <porth>, and possibly attached to fd <fd> (or -1 for auto
625 * allocation). The address family is taken from ss->ss_family, and the protocol
626 * passed in <proto> must be usable on this family. The protocol's default iocb
627 * is automatically preset as the receivers' iocb. The number of jobs and
628 * listeners is automatically increased by the number of listeners created. It
629 * returns non-zero on success, zero on error with the error message set in <err>.
630 */
create_listeners(struct bind_conf * bc,const struct sockaddr_storage * ss,int portl,int porth,int fd,struct protocol * proto,char ** err)631 int create_listeners(struct bind_conf *bc, const struct sockaddr_storage *ss,
632 int portl, int porth, int fd, struct protocol *proto, char **err)
633 {
634 struct listener *l;
635 int port;
636
637 for (port = portl; port <= porth; port++) {
638 l = calloc(1, sizeof(*l));
639 if (!l) {
640 memprintf(err, "out of memory");
641 return 0;
642 }
643 l->obj_type = OBJ_TYPE_LISTENER;
644 LIST_APPEND(&bc->frontend->conf.listeners, &l->by_fe);
645 LIST_APPEND(&bc->listeners, &l->by_bind);
646 l->bind_conf = bc;
647 l->rx.settings = &bc->settings;
648 l->rx.owner = l;
649 l->rx.iocb = proto->default_iocb;
650 l->rx.fd = fd;
651
652 memcpy(&l->rx.addr, ss, sizeof(*ss));
653 if (proto->fam->set_port)
654 proto->fam->set_port(&l->rx.addr, port);
655
656 MT_LIST_INIT(&l->wait_queue);
657 listener_set_state(l, LI_INIT);
658
659 proto->add(proto, l);
660
661 if (fd != -1)
662 l->rx.flags |= RX_F_INHERITED;
663
664 l->extra_counters = NULL;
665
666 HA_SPIN_INIT(&l->lock);
667 _HA_ATOMIC_INC(&jobs);
668 _HA_ATOMIC_INC(&listeners);
669 }
670 return 1;
671 }
672
673 /* Delete a listener from its protocol's list of listeners. The listener's
674 * state is automatically updated from LI_ASSIGNED to LI_INIT. The protocol's
675 * number of listeners is updated, as well as the global number of listeners
676 * and jobs. Note that the listener must have previously been unbound. This
677 * is a low-level function expected to be called with the proto_lock and the
678 * listener's lock held.
679 */
__delete_listener(struct listener * listener)680 void __delete_listener(struct listener *listener)
681 {
682 if (listener->state == LI_ASSIGNED) {
683 listener_set_state(listener, LI_INIT);
684 LIST_DELETE(&listener->rx.proto_list);
685 listener->rx.proto->nb_receivers--;
686 _HA_ATOMIC_DEC(&jobs);
687 _HA_ATOMIC_DEC(&listeners);
688 }
689 }
690
691 /* Delete a listener from its protocol's list of listeners (please check
692 * __delete_listener() above). The proto_lock and the listener's lock will
693 * be grabbed in this order.
694 */
delete_listener(struct listener * listener)695 void delete_listener(struct listener *listener)
696 {
697 HA_SPIN_LOCK(PROTO_LOCK, &proto_lock);
698 HA_SPIN_LOCK(LISTENER_LOCK, &listener->lock);
699 __delete_listener(listener);
700 HA_SPIN_UNLOCK(LISTENER_LOCK, &listener->lock);
701 HA_SPIN_UNLOCK(PROTO_LOCK, &proto_lock);
702 }
703
704 /* Returns a suitable value for a listener's backlog. It uses the listener's,
705 * otherwise the frontend's backlog, otherwise the listener's maxconn,
706 * otherwise the frontend's maxconn, otherwise 1024.
707 */
listener_backlog(const struct listener * l)708 int listener_backlog(const struct listener *l)
709 {
710 if (l->backlog)
711 return l->backlog;
712
713 if (l->bind_conf->frontend->backlog)
714 return l->bind_conf->frontend->backlog;
715
716 if (l->maxconn)
717 return l->maxconn;
718
719 if (l->bind_conf->frontend->maxconn)
720 return l->bind_conf->frontend->maxconn;
721
722 return 1024;
723 }
724
725 /* This function is called on a read event from a listening socket, corresponding
726 * to an accept. It tries to accept as many connections as possible, and for each
727 * calls the listener's accept handler (generally the frontend's accept handler).
728 */
listener_accept(struct listener * l)729 void listener_accept(struct listener *l)
730 {
731 struct connection *cli_conn;
732 struct proxy *p;
733 unsigned int max_accept;
734 int next_conn = 0;
735 int next_feconn = 0;
736 int next_actconn = 0;
737 int expire;
738 int ret;
739
740 p = l->bind_conf->frontend;
741
742 /* if l->maxaccept is -1, then max_accept is UINT_MAX. It is not really
743 * illimited, but it is probably enough.
744 */
745 max_accept = l->maxaccept ? l->maxaccept : 1;
746
747 if (!(l->options & LI_O_UNLIMITED) && global.sps_lim) {
748 int max = freq_ctr_remain(&global.sess_per_sec, global.sps_lim, 0);
749
750 if (unlikely(!max)) {
751 /* frontend accept rate limit was reached */
752 expire = tick_add(now_ms, next_event_delay(&global.sess_per_sec, global.sps_lim, 0));
753 goto limit_global;
754 }
755
756 if (max_accept > max)
757 max_accept = max;
758 }
759
760 if (!(l->options & LI_O_UNLIMITED) && global.cps_lim) {
761 int max = freq_ctr_remain(&global.conn_per_sec, global.cps_lim, 0);
762
763 if (unlikely(!max)) {
764 /* frontend accept rate limit was reached */
765 expire = tick_add(now_ms, next_event_delay(&global.conn_per_sec, global.cps_lim, 0));
766 goto limit_global;
767 }
768
769 if (max_accept > max)
770 max_accept = max;
771 }
772 #ifdef USE_OPENSSL
773 if (!(l->options & LI_O_UNLIMITED) && global.ssl_lim && l->bind_conf && l->bind_conf->is_ssl) {
774 int max = freq_ctr_remain(&global.ssl_per_sec, global.ssl_lim, 0);
775
776 if (unlikely(!max)) {
777 /* frontend accept rate limit was reached */
778 expire = tick_add(now_ms, next_event_delay(&global.ssl_per_sec, global.ssl_lim, 0));
779 goto limit_global;
780 }
781
782 if (max_accept > max)
783 max_accept = max;
784 }
785 #endif
786 if (p && p->fe_sps_lim) {
787 int max = freq_ctr_remain(&p->fe_sess_per_sec, p->fe_sps_lim, 0);
788
789 if (unlikely(!max)) {
790 /* frontend accept rate limit was reached */
791 expire = tick_add(now_ms, next_event_delay(&p->fe_sess_per_sec, p->fe_sps_lim, 0));
792 goto limit_proxy;
793 }
794
795 if (max_accept > max)
796 max_accept = max;
797 }
798
799 /* Note: if we fail to allocate a connection because of configured
800 * limits, we'll schedule a new attempt worst 1 second later in the
801 * worst case. If we fail due to system limits or temporary resource
802 * shortage, we try again 100ms later in the worst case.
803 */
804 for (; max_accept; next_conn = next_feconn = next_actconn = 0, max_accept--) {
805 unsigned int count;
806 int status;
807 __decl_thread(unsigned long mask);
808
809 /* pre-increase the number of connections without going too far.
810 * We process the listener, then the proxy, then the process.
811 * We know which ones to unroll based on the next_xxx value.
812 */
813 do {
814 count = l->nbconn;
815 if (unlikely(l->maxconn && count >= l->maxconn)) {
816 /* the listener was marked full or another
817 * thread is going to do it.
818 */
819 next_conn = 0;
820 listener_full(l);
821 goto end;
822 }
823 next_conn = count + 1;
824 } while (!_HA_ATOMIC_CAS(&l->nbconn, (int *)(&count), next_conn));
825
826 if (p) {
827 do {
828 count = p->feconn;
829 if (unlikely(count >= p->maxconn)) {
830 /* the frontend was marked full or another
831 * thread is going to do it.
832 */
833 next_feconn = 0;
834 expire = TICK_ETERNITY;
835 goto limit_proxy;
836 }
837 next_feconn = count + 1;
838 } while (!_HA_ATOMIC_CAS(&p->feconn, &count, next_feconn));
839 }
840
841 if (!(l->options & LI_O_UNLIMITED)) {
842 do {
843 count = actconn;
844 if (unlikely(count >= global.maxconn)) {
845 /* the process was marked full or another
846 * thread is going to do it.
847 */
848 next_actconn = 0;
849 expire = tick_add(now_ms, 1000); /* try again in 1 second */
850 goto limit_global;
851 }
852 next_actconn = count + 1;
853 } while (!_HA_ATOMIC_CAS(&actconn, (int *)(&count), next_actconn));
854 }
855
856 cli_conn = l->rx.proto->accept_conn(l, &status);
857 if (!cli_conn) {
858 switch (status) {
859 case CO_AC_DONE:
860 goto end;
861
862 case CO_AC_RETRY: /* likely a signal */
863 _HA_ATOMIC_DEC(&l->nbconn);
864 if (p)
865 _HA_ATOMIC_DEC(&p->feconn);
866 if (!(l->options & LI_O_UNLIMITED))
867 _HA_ATOMIC_DEC(&actconn);
868 continue;
869
870 case CO_AC_YIELD:
871 max_accept = 0;
872 goto end;
873
874 default:
875 goto transient_error;
876 }
877 }
878
879 /* The connection was accepted, it must be counted as such */
880 if (l->counters)
881 HA_ATOMIC_UPDATE_MAX(&l->counters->conn_max, next_conn);
882
883 if (p)
884 HA_ATOMIC_UPDATE_MAX(&p->fe_counters.conn_max, next_feconn);
885
886 proxy_inc_fe_conn_ctr(l, p);
887
888 if (!(l->options & LI_O_UNLIMITED)) {
889 count = update_freq_ctr(&global.conn_per_sec, 1);
890 HA_ATOMIC_UPDATE_MAX(&global.cps_max, count);
891 }
892
893 _HA_ATOMIC_INC(&activity[tid].accepted);
894
895 if (unlikely(cli_conn->handle.fd >= global.maxsock)) {
896 send_log(p, LOG_EMERG,
897 "Proxy %s reached the configured maximum connection limit. Please check the global 'maxconn' value.\n",
898 p->id);
899 close(cli_conn->handle.fd);
900 conn_free(cli_conn);
901 expire = tick_add(now_ms, 1000); /* try again in 1 second */
902 goto limit_global;
903 }
904
905 /* past this point, l->accept() will automatically decrement
906 * l->nbconn, feconn and actconn once done. Setting next_*conn=0
907 * allows the error path not to rollback on nbconn. It's more
908 * convenient than duplicating all exit labels.
909 */
910 next_conn = 0;
911 next_feconn = 0;
912 next_actconn = 0;
913
914
915 #if defined(USE_THREAD)
916 mask = thread_mask(l->rx.settings->bind_thread) & all_threads_mask;
917 if (atleast2(mask) && (global.tune.options & GTUNE_LISTENER_MQ) && !stopping) {
918 struct accept_queue_ring *ring;
919 unsigned int t, t0, t1, t2;
920
921 /* The principle is that we have two running indexes,
922 * each visiting in turn all threads bound to this
923 * listener. The connection will be assigned to the one
924 * with the least connections, and the other one will
925 * be updated. This provides a good fairness on short
926 * connections (round robin) and on long ones (conn
927 * count), without ever missing any idle thread.
928 */
929
930 /* keep a copy for the final update. thr_idx is composite
931 * and made of (t2<<16) + t1.
932 */
933 t0 = l->thr_idx;
934 do {
935 unsigned long m1, m2;
936 int q1, q2;
937
938 t2 = t1 = t0;
939 t2 >>= 16;
940 t1 &= 0xFFFF;
941
942 /* t1 walks low to high bits ;
943 * t2 walks high to low.
944 */
945 m1 = mask >> t1;
946 m2 = mask & (t2 ? nbits(t2 + 1) : ~0UL);
947
948 if (unlikely(!(m1 & 1))) {
949 m1 &= ~1UL;
950 if (!m1) {
951 m1 = mask;
952 t1 = 0;
953 }
954 t1 += my_ffsl(m1) - 1;
955 }
956
957 if (unlikely(!(m2 & (1UL << t2)) || t1 == t2)) {
958 /* highest bit not set */
959 if (!m2)
960 m2 = mask;
961
962 t2 = my_flsl(m2) - 1;
963 }
964
965 /* now we have two distinct thread IDs belonging to the mask */
966 q1 = accept_queue_rings[t1].tail - accept_queue_rings[t1].head + ACCEPT_QUEUE_SIZE;
967 if (q1 >= ACCEPT_QUEUE_SIZE)
968 q1 -= ACCEPT_QUEUE_SIZE;
969
970 q2 = accept_queue_rings[t2].tail - accept_queue_rings[t2].head + ACCEPT_QUEUE_SIZE;
971 if (q2 >= ACCEPT_QUEUE_SIZE)
972 q2 -= ACCEPT_QUEUE_SIZE;
973
974 /* we have 3 possibilities now :
975 * q1 < q2 : t1 is less loaded than t2, so we pick it
976 * and update t2 (since t1 might still be
977 * lower than another thread)
978 * q1 > q2 : t2 is less loaded than t1, so we pick it
979 * and update t1 (since t2 might still be
980 * lower than another thread)
981 * q1 = q2 : both are equally loaded, thus we pick t1
982 * and update t1 as it will become more loaded
983 * than t2.
984 */
985
986 q1 += l->thr_conn[t1];
987 q2 += l->thr_conn[t2];
988
989 if (q1 - q2 < 0) {
990 t = t1;
991 t2 = t2 ? t2 - 1 : LONGBITS - 1;
992 }
993 else if (q1 - q2 > 0) {
994 t = t2;
995 t1++;
996 if (t1 >= LONGBITS)
997 t1 = 0;
998 }
999 else {
1000 t = t1;
1001 t1++;
1002 if (t1 >= LONGBITS)
1003 t1 = 0;
1004 }
1005
1006 /* new value for thr_idx */
1007 t1 += (t2 << 16);
1008 } while (unlikely(!_HA_ATOMIC_CAS(&l->thr_idx, &t0, t1)));
1009
1010 /* We successfully selected the best thread "t" for this
1011 * connection. We use deferred accepts even if it's the
1012 * local thread because tests show that it's the best
1013 * performing model, likely due to better cache locality
1014 * when processing this loop.
1015 */
1016 ring = &accept_queue_rings[t];
1017 if (accept_queue_push_mp(ring, cli_conn)) {
1018 _HA_ATOMIC_INC(&activity[t].accq_pushed);
1019 tasklet_wakeup(ring->tasklet);
1020 continue;
1021 }
1022 /* If the ring is full we do a synchronous accept on
1023 * the local thread here.
1024 */
1025 _HA_ATOMIC_INC(&activity[t].accq_full);
1026 }
1027 #endif // USE_THREAD
1028
1029 _HA_ATOMIC_INC(&l->thr_conn[tid]);
1030 ret = l->accept(cli_conn);
1031 if (unlikely(ret <= 0)) {
1032 /* The connection was closed by stream_accept(). Either
1033 * we just have to ignore it (ret == 0) or it's a critical
1034 * error due to a resource shortage, and we must stop the
1035 * listener (ret < 0).
1036 */
1037 if (ret == 0) /* successful termination */
1038 continue;
1039
1040 goto transient_error;
1041 }
1042
1043 /* increase the per-process number of cumulated sessions, this
1044 * may only be done once l->accept() has accepted the connection.
1045 */
1046 if (!(l->options & LI_O_UNLIMITED)) {
1047 count = update_freq_ctr(&global.sess_per_sec, 1);
1048 HA_ATOMIC_UPDATE_MAX(&global.sps_max, count);
1049 }
1050 #ifdef USE_OPENSSL
1051 if (!(l->options & LI_O_UNLIMITED) && l->bind_conf && l->bind_conf->is_ssl) {
1052 count = update_freq_ctr(&global.ssl_per_sec, 1);
1053 HA_ATOMIC_UPDATE_MAX(&global.ssl_max, count);
1054 }
1055 #endif
1056
1057 ti->flags &= ~TI_FL_STUCK; // this thread is still running
1058 } /* end of for (max_accept--) */
1059
1060 end:
1061 if (next_conn)
1062 _HA_ATOMIC_DEC(&l->nbconn);
1063
1064 if (p && next_feconn)
1065 _HA_ATOMIC_DEC(&p->feconn);
1066
1067 if (next_actconn)
1068 _HA_ATOMIC_DEC(&actconn);
1069
1070 if ((l->state == LI_FULL && (!l->maxconn || l->nbconn < l->maxconn)) ||
1071 (l->state == LI_LIMITED &&
1072 ((!p || p->feconn < p->maxconn) && (actconn < global.maxconn) &&
1073 (!tick_isset(global_listener_queue_task->expire) ||
1074 tick_is_expired(global_listener_queue_task->expire, now_ms))))) {
1075 /* at least one thread has to this when quitting */
1076 resume_listener(l);
1077
1078 /* Dequeues all of the listeners waiting for a resource */
1079 dequeue_all_listeners();
1080
1081 if (p && !MT_LIST_ISEMPTY(&p->listener_queue) &&
1082 (!p->fe_sps_lim || freq_ctr_remain(&p->fe_sess_per_sec, p->fe_sps_lim, 0) > 0))
1083 dequeue_proxy_listeners(p);
1084 }
1085 return;
1086
1087 transient_error:
1088 /* pause the listener for up to 100 ms */
1089 expire = tick_add(now_ms, 100);
1090
1091 /* This may be a shared socket that was paused by another process.
1092 * Let's put it to pause in this case.
1093 */
1094 if (l->rx.proto && l->rx.proto->rx_listening(&l->rx) == 0) {
1095 pause_listener(l);
1096 goto end;
1097 }
1098
1099 limit_global:
1100 /* (re-)queue the listener to the global queue and set it to expire no
1101 * later than <expire> ahead. The listener turns to LI_LIMITED.
1102 */
1103 limit_listener(l, &global_listener_queue);
1104 task_schedule(global_listener_queue_task, expire);
1105 goto end;
1106
1107 limit_proxy:
1108 /* (re-)queue the listener to the proxy's queue and set it to expire no
1109 * later than <expire> ahead. The listener turns to LI_LIMITED.
1110 */
1111 limit_listener(l, &p->listener_queue);
1112 if (p->task && tick_isset(expire))
1113 task_schedule(p->task, expire);
1114 goto end;
1115 }
1116
1117 /* Notify the listener that a connection initiated from it was released. This
1118 * is used to keep the connection count consistent and to possibly re-open
1119 * listening when it was limited.
1120 */
listener_release(struct listener * l)1121 void listener_release(struct listener *l)
1122 {
1123 struct proxy *fe = l->bind_conf->frontend;
1124
1125 if (!(l->options & LI_O_UNLIMITED))
1126 _HA_ATOMIC_DEC(&actconn);
1127 if (fe)
1128 _HA_ATOMIC_DEC(&fe->feconn);
1129 _HA_ATOMIC_DEC(&l->nbconn);
1130 _HA_ATOMIC_DEC(&l->thr_conn[tid]);
1131
1132 if (l->state == LI_FULL || l->state == LI_LIMITED)
1133 resume_listener(l);
1134
1135 /* Dequeues all of the listeners waiting for a resource */
1136 dequeue_all_listeners();
1137
1138 if (!MT_LIST_ISEMPTY(&fe->listener_queue) &&
1139 (!fe->fe_sps_lim || freq_ctr_remain(&fe->fe_sess_per_sec, fe->fe_sps_lim, 0) > 0))
1140 dequeue_proxy_listeners(fe);
1141 }
1142
1143 /* Initializes the listener queues. Returns 0 on success, otherwise ERR_* flags */
listener_queue_init()1144 static int listener_queue_init()
1145 {
1146 global_listener_queue_task = task_new(MAX_THREADS_MASK);
1147 if (!global_listener_queue_task) {
1148 ha_alert("Out of memory when initializing global listener queue\n");
1149 return ERR_FATAL|ERR_ABORT;
1150 }
1151 /* very simple initialization, users will queue the task if needed */
1152 global_listener_queue_task->context = NULL; /* not even a context! */
1153 global_listener_queue_task->process = manage_global_listener_queue;
1154
1155 return 0;
1156 }
1157
listener_queue_deinit()1158 static void listener_queue_deinit()
1159 {
1160 task_destroy(global_listener_queue_task);
1161 global_listener_queue_task = NULL;
1162 }
1163
1164 REGISTER_CONFIG_POSTPARSER("multi-threaded listener queue", listener_queue_init);
1165 REGISTER_POST_DEINIT(listener_queue_deinit);
1166
1167
1168 /* This is the global management task for listeners. It enables listeners waiting
1169 * for global resources when there are enough free resource, or at least once in
1170 * a while. It is designed to be called as a task. It's exported so that it's easy
1171 * to spot in "show tasks" or "show profiling".
1172 */
manage_global_listener_queue(struct task * t,void * context,unsigned int state)1173 struct task *manage_global_listener_queue(struct task *t, void *context, unsigned int state)
1174 {
1175 /* If there are still too many concurrent connections, let's wait for
1176 * some of them to go away. We don't need to re-arm the timer because
1177 * each of them will scan the queue anyway.
1178 */
1179 if (unlikely(actconn >= global.maxconn))
1180 goto out;
1181
1182 /* We should periodically try to enable listeners waiting for a global
1183 * resource here, because it is possible, though very unlikely, that
1184 * they have been blocked by a temporary lack of global resource such
1185 * as a file descriptor or memory and that the temporary condition has
1186 * disappeared.
1187 */
1188 dequeue_all_listeners();
1189
1190 out:
1191 t->expire = TICK_ETERNITY;
1192 task_queue(t);
1193 return t;
1194 }
1195
1196 /*
1197 * Registers the bind keyword list <kwl> as a list of valid keywords for next
1198 * parsing sessions.
1199 */
bind_register_keywords(struct bind_kw_list * kwl)1200 void bind_register_keywords(struct bind_kw_list *kwl)
1201 {
1202 LIST_APPEND(&bind_keywords.list, &kwl->list);
1203 }
1204
1205 /* Return a pointer to the bind keyword <kw>, or NULL if not found. If the
1206 * keyword is found with a NULL ->parse() function, then an attempt is made to
1207 * find one with a valid ->parse() function. This way it is possible to declare
1208 * platform-dependant, known keywords as NULL, then only declare them as valid
1209 * if some options are met. Note that if the requested keyword contains an
1210 * opening parenthesis, everything from this point is ignored.
1211 */
bind_find_kw(const char * kw)1212 struct bind_kw *bind_find_kw(const char *kw)
1213 {
1214 int index;
1215 const char *kwend;
1216 struct bind_kw_list *kwl;
1217 struct bind_kw *ret = NULL;
1218
1219 kwend = strchr(kw, '(');
1220 if (!kwend)
1221 kwend = kw + strlen(kw);
1222
1223 list_for_each_entry(kwl, &bind_keywords.list, list) {
1224 for (index = 0; kwl->kw[index].kw != NULL; index++) {
1225 if ((strncmp(kwl->kw[index].kw, kw, kwend - kw) == 0) &&
1226 kwl->kw[index].kw[kwend-kw] == 0) {
1227 if (kwl->kw[index].parse)
1228 return &kwl->kw[index]; /* found it !*/
1229 else
1230 ret = &kwl->kw[index]; /* may be OK */
1231 }
1232 }
1233 }
1234 return ret;
1235 }
1236
1237 /* Dumps all registered "bind" keywords to the <out> string pointer. The
1238 * unsupported keywords are only dumped if their supported form was not
1239 * found.
1240 */
bind_dump_kws(char ** out)1241 void bind_dump_kws(char **out)
1242 {
1243 struct bind_kw_list *kwl;
1244 int index;
1245
1246 if (!out)
1247 return;
1248
1249 *out = NULL;
1250 list_for_each_entry(kwl, &bind_keywords.list, list) {
1251 for (index = 0; kwl->kw[index].kw != NULL; index++) {
1252 if (kwl->kw[index].parse ||
1253 bind_find_kw(kwl->kw[index].kw) == &kwl->kw[index]) {
1254 memprintf(out, "%s[%4s] %s%s%s\n", *out ? *out : "",
1255 kwl->scope,
1256 kwl->kw[index].kw,
1257 kwl->kw[index].skip ? " <arg>" : "",
1258 kwl->kw[index].parse ? "" : " (not supported)");
1259 }
1260 }
1261 }
1262 }
1263
1264 /* Try to find in srv_keyword the word that looks closest to <word> by counting
1265 * transitions between letters, digits and other characters. Will return the
1266 * best matching word if found, otherwise NULL.
1267 */
bind_find_best_kw(const char * word)1268 const char *bind_find_best_kw(const char *word)
1269 {
1270 uint8_t word_sig[1024];
1271 uint8_t list_sig[1024];
1272 const struct bind_kw_list *kwl;
1273 const char *best_ptr = NULL;
1274 int dist, best_dist = INT_MAX;
1275 int index;
1276
1277 make_word_fingerprint(word_sig, word);
1278 list_for_each_entry(kwl, &bind_keywords.list, list) {
1279 for (index = 0; kwl->kw[index].kw != NULL; index++) {
1280 make_word_fingerprint(list_sig, kwl->kw[index].kw);
1281 dist = word_fingerprint_distance(word_sig, list_sig);
1282 if (dist < best_dist) {
1283 best_dist = dist;
1284 best_ptr = kwl->kw[index].kw;
1285 }
1286 }
1287 }
1288
1289 if (best_dist > 2 * strlen(word) || (best_ptr && best_dist > 2 * strlen(best_ptr)))
1290 best_ptr = NULL;
1291
1292 return best_ptr;
1293 }
1294
1295 /************************************************************************/
1296 /* All supported sample and ACL keywords must be declared here. */
1297 /************************************************************************/
1298
1299 /* set temp integer to the number of connexions to the same listening socket */
1300 static int
smp_fetch_dconn(const struct arg * args,struct sample * smp,const char * kw,void * private)1301 smp_fetch_dconn(const struct arg *args, struct sample *smp, const char *kw, void *private)
1302 {
1303 smp->data.type = SMP_T_SINT;
1304 smp->data.u.sint = smp->sess->listener->nbconn;
1305 return 1;
1306 }
1307
1308 /* set temp integer to the id of the socket (listener) */
1309 static int
smp_fetch_so_id(const struct arg * args,struct sample * smp,const char * kw,void * private)1310 smp_fetch_so_id(const struct arg *args, struct sample *smp, const char *kw, void *private)
1311 {
1312 smp->data.type = SMP_T_SINT;
1313 smp->data.u.sint = smp->sess->listener->luid;
1314 return 1;
1315 }
1316 static int
smp_fetch_so_name(const struct arg * args,struct sample * smp,const char * kw,void * private)1317 smp_fetch_so_name(const struct arg *args, struct sample *smp, const char *kw, void *private)
1318 {
1319 smp->data.u.str.area = smp->sess->listener->name;
1320 if (!smp->data.u.str.area)
1321 return 0;
1322
1323 smp->data.type = SMP_T_STR;
1324 smp->flags = SMP_F_CONST;
1325 smp->data.u.str.data = strlen(smp->data.u.str.area);
1326 return 1;
1327 }
1328
1329 /* parse the "accept-proxy" bind keyword */
bind_parse_accept_proxy(char ** args,int cur_arg,struct proxy * px,struct bind_conf * conf,char ** err)1330 static int bind_parse_accept_proxy(char **args, int cur_arg, struct proxy *px, struct bind_conf *conf, char **err)
1331 {
1332 struct listener *l;
1333
1334 list_for_each_entry(l, &conf->listeners, by_bind)
1335 l->options |= LI_O_ACC_PROXY;
1336
1337 return 0;
1338 }
1339
1340 /* parse the "accept-netscaler-cip" bind keyword */
bind_parse_accept_netscaler_cip(char ** args,int cur_arg,struct proxy * px,struct bind_conf * conf,char ** err)1341 static int bind_parse_accept_netscaler_cip(char **args, int cur_arg, struct proxy *px, struct bind_conf *conf, char **err)
1342 {
1343 struct listener *l;
1344 uint32_t val;
1345
1346 if (!*args[cur_arg + 1]) {
1347 memprintf(err, "'%s' : missing value", args[cur_arg]);
1348 return ERR_ALERT | ERR_FATAL;
1349 }
1350
1351 val = atol(args[cur_arg + 1]);
1352 if (val <= 0) {
1353 memprintf(err, "'%s' : invalid value %d, must be >= 0", args[cur_arg], val);
1354 return ERR_ALERT | ERR_FATAL;
1355 }
1356
1357 list_for_each_entry(l, &conf->listeners, by_bind) {
1358 l->options |= LI_O_ACC_CIP;
1359 conf->ns_cip_magic = val;
1360 }
1361
1362 return 0;
1363 }
1364
1365 /* parse the "backlog" bind keyword */
bind_parse_backlog(char ** args,int cur_arg,struct proxy * px,struct bind_conf * conf,char ** err)1366 static int bind_parse_backlog(char **args, int cur_arg, struct proxy *px, struct bind_conf *conf, char **err)
1367 {
1368 struct listener *l;
1369 int val;
1370
1371 if (!*args[cur_arg + 1]) {
1372 memprintf(err, "'%s' : missing value", args[cur_arg]);
1373 return ERR_ALERT | ERR_FATAL;
1374 }
1375
1376 val = atol(args[cur_arg + 1]);
1377 if (val < 0) {
1378 memprintf(err, "'%s' : invalid value %d, must be > 0", args[cur_arg], val);
1379 return ERR_ALERT | ERR_FATAL;
1380 }
1381
1382 list_for_each_entry(l, &conf->listeners, by_bind)
1383 l->backlog = val;
1384
1385 return 0;
1386 }
1387
1388 /* parse the "id" bind keyword */
bind_parse_id(char ** args,int cur_arg,struct proxy * px,struct bind_conf * conf,char ** err)1389 static int bind_parse_id(char **args, int cur_arg, struct proxy *px, struct bind_conf *conf, char **err)
1390 {
1391 struct eb32_node *node;
1392 struct listener *l, *new;
1393 char *error;
1394
1395 if (conf->listeners.n != conf->listeners.p) {
1396 memprintf(err, "'%s' can only be used with a single socket", args[cur_arg]);
1397 return ERR_ALERT | ERR_FATAL;
1398 }
1399
1400 if (!*args[cur_arg + 1]) {
1401 memprintf(err, "'%s' : expects an integer argument", args[cur_arg]);
1402 return ERR_ALERT | ERR_FATAL;
1403 }
1404
1405 new = LIST_NEXT(&conf->listeners, struct listener *, by_bind);
1406 new->luid = strtol(args[cur_arg + 1], &error, 10);
1407 if (*error != '\0') {
1408 memprintf(err, "'%s' : expects an integer argument, found '%s'", args[cur_arg], args[cur_arg + 1]);
1409 return ERR_ALERT | ERR_FATAL;
1410 }
1411 new->conf.id.key = new->luid;
1412
1413 if (new->luid <= 0) {
1414 memprintf(err, "'%s' : custom id has to be > 0", args[cur_arg]);
1415 return ERR_ALERT | ERR_FATAL;
1416 }
1417
1418 node = eb32_lookup(&px->conf.used_listener_id, new->luid);
1419 if (node) {
1420 l = container_of(node, struct listener, conf.id);
1421 memprintf(err, "'%s' : custom id %d already used at %s:%d ('bind %s')",
1422 args[cur_arg], l->luid, l->bind_conf->file, l->bind_conf->line,
1423 l->bind_conf->arg);
1424 return ERR_ALERT | ERR_FATAL;
1425 }
1426
1427 eb32_insert(&px->conf.used_listener_id, &new->conf.id);
1428 return 0;
1429 }
1430
1431 /* parse the "maxconn" bind keyword */
bind_parse_maxconn(char ** args,int cur_arg,struct proxy * px,struct bind_conf * conf,char ** err)1432 static int bind_parse_maxconn(char **args, int cur_arg, struct proxy *px, struct bind_conf *conf, char **err)
1433 {
1434 struct listener *l;
1435 int val;
1436
1437 if (!*args[cur_arg + 1]) {
1438 memprintf(err, "'%s' : missing value", args[cur_arg]);
1439 return ERR_ALERT | ERR_FATAL;
1440 }
1441
1442 val = atol(args[cur_arg + 1]);
1443 if (val < 0) {
1444 memprintf(err, "'%s' : invalid value %d, must be >= 0", args[cur_arg], val);
1445 return ERR_ALERT | ERR_FATAL;
1446 }
1447
1448 list_for_each_entry(l, &conf->listeners, by_bind)
1449 l->maxconn = val;
1450
1451 return 0;
1452 }
1453
1454 /* parse the "name" bind keyword */
bind_parse_name(char ** args,int cur_arg,struct proxy * px,struct bind_conf * conf,char ** err)1455 static int bind_parse_name(char **args, int cur_arg, struct proxy *px, struct bind_conf *conf, char **err)
1456 {
1457 struct listener *l;
1458
1459 if (!*args[cur_arg + 1]) {
1460 memprintf(err, "'%s' : missing name", args[cur_arg]);
1461 return ERR_ALERT | ERR_FATAL;
1462 }
1463
1464 list_for_each_entry(l, &conf->listeners, by_bind)
1465 l->name = strdup(args[cur_arg + 1]);
1466
1467 return 0;
1468 }
1469
1470 /* parse the "nice" bind keyword */
bind_parse_nice(char ** args,int cur_arg,struct proxy * px,struct bind_conf * conf,char ** err)1471 static int bind_parse_nice(char **args, int cur_arg, struct proxy *px, struct bind_conf *conf, char **err)
1472 {
1473 struct listener *l;
1474 int val;
1475
1476 if (!*args[cur_arg + 1]) {
1477 memprintf(err, "'%s' : missing value", args[cur_arg]);
1478 return ERR_ALERT | ERR_FATAL;
1479 }
1480
1481 val = atol(args[cur_arg + 1]);
1482 if (val < -1024 || val > 1024) {
1483 memprintf(err, "'%s' : invalid value %d, allowed range is -1024..1024", args[cur_arg], val);
1484 return ERR_ALERT | ERR_FATAL;
1485 }
1486
1487 list_for_each_entry(l, &conf->listeners, by_bind)
1488 l->nice = val;
1489
1490 return 0;
1491 }
1492
1493 /* parse the "process" bind keyword */
bind_parse_process(char ** args,int cur_arg,struct proxy * px,struct bind_conf * conf,char ** err)1494 static int bind_parse_process(char **args, int cur_arg, struct proxy *px, struct bind_conf *conf, char **err)
1495 {
1496 char *slash;
1497 unsigned long proc = 0, thread = 0;
1498
1499 if ((slash = strchr(args[cur_arg + 1], '/')) != NULL)
1500 *slash = 0;
1501
1502 if (parse_process_number(args[cur_arg + 1], &proc, MAX_PROCS, NULL, err)) {
1503 memprintf(err, "'%s' : %s", args[cur_arg], *err);
1504 return ERR_ALERT | ERR_FATAL;
1505 }
1506
1507 if (slash) {
1508 if (parse_process_number(slash+1, &thread, MAX_THREADS, NULL, err)) {
1509 memprintf(err, "'%s' : %s", args[cur_arg], *err);
1510 return ERR_ALERT | ERR_FATAL;
1511 }
1512 *slash = '/';
1513 }
1514
1515 conf->settings.bind_proc |= proc;
1516 conf->settings.bind_thread |= thread;
1517 return 0;
1518 }
1519
1520 /* parse the "proto" bind keyword */
bind_parse_proto(char ** args,int cur_arg,struct proxy * px,struct bind_conf * conf,char ** err)1521 static int bind_parse_proto(char **args, int cur_arg, struct proxy *px, struct bind_conf *conf, char **err)
1522 {
1523 struct ist proto;
1524
1525 if (!*args[cur_arg + 1]) {
1526 memprintf(err, "'%s' : missing value", args[cur_arg]);
1527 return ERR_ALERT | ERR_FATAL;
1528 }
1529
1530 proto = ist(args[cur_arg + 1]);
1531 conf->mux_proto = get_mux_proto(proto);
1532 if (!conf->mux_proto) {
1533 memprintf(err, "'%s' : unknown MUX protocol '%s'", args[cur_arg], args[cur_arg+1]);
1534 return ERR_ALERT | ERR_FATAL;
1535 }
1536 return 0;
1537 }
1538
1539 /* config parser for global "tune.listener.multi-queue", accepts "on" or "off" */
cfg_parse_tune_listener_mq(char ** args,int section_type,struct proxy * curpx,const struct proxy * defpx,const char * file,int line,char ** err)1540 static int cfg_parse_tune_listener_mq(char **args, int section_type, struct proxy *curpx,
1541 const struct proxy *defpx, const char *file, int line,
1542 char **err)
1543 {
1544 if (too_many_args(1, args, err, NULL))
1545 return -1;
1546
1547 if (strcmp(args[1], "on") == 0)
1548 global.tune.options |= GTUNE_LISTENER_MQ;
1549 else if (strcmp(args[1], "off") == 0)
1550 global.tune.options &= ~GTUNE_LISTENER_MQ;
1551 else {
1552 memprintf(err, "'%s' expects either 'on' or 'off' but got '%s'.", args[0], args[1]);
1553 return -1;
1554 }
1555 return 0;
1556 }
1557
1558 /* Note: must not be declared <const> as its list will be overwritten.
1559 * Please take care of keeping this list alphabetically sorted.
1560 */
1561 static struct sample_fetch_kw_list smp_kws = {ILH, {
1562 { "dst_conn", smp_fetch_dconn, 0, NULL, SMP_T_SINT, SMP_USE_FTEND, },
1563 { "so_id", smp_fetch_so_id, 0, NULL, SMP_T_SINT, SMP_USE_FTEND, },
1564 { "so_name", smp_fetch_so_name, 0, NULL, SMP_T_STR, SMP_USE_FTEND, },
1565 { /* END */ },
1566 }};
1567
1568 INITCALL1(STG_REGISTER, sample_register_fetches, &smp_kws);
1569
1570 /* Note: must not be declared <const> as its list will be overwritten.
1571 * Please take care of keeping this list alphabetically sorted.
1572 */
1573 static struct acl_kw_list acl_kws = {ILH, {
1574 { /* END */ },
1575 }};
1576
1577 INITCALL1(STG_REGISTER, acl_register_keywords, &acl_kws);
1578
1579 /* Note: must not be declared <const> as its list will be overwritten.
1580 * Please take care of keeping this list alphabetically sorted, doing so helps
1581 * all code contributors.
1582 * Optional keywords are also declared with a NULL ->parse() function so that
1583 * the config parser can report an appropriate error when a known keyword was
1584 * not enabled.
1585 */
1586 static struct bind_kw_list bind_kws = { "ALL", { }, {
1587 { "accept-netscaler-cip", bind_parse_accept_netscaler_cip, 1 }, /* enable NetScaler Client IP insertion protocol */
1588 { "accept-proxy", bind_parse_accept_proxy, 0 }, /* enable PROXY protocol */
1589 { "backlog", bind_parse_backlog, 1 }, /* set backlog of listening socket */
1590 { "id", bind_parse_id, 1 }, /* set id of listening socket */
1591 { "maxconn", bind_parse_maxconn, 1 }, /* set maxconn of listening socket */
1592 { "name", bind_parse_name, 1 }, /* set name of listening socket */
1593 { "nice", bind_parse_nice, 1 }, /* set nice of listening socket */
1594 { "process", bind_parse_process, 1 }, /* set list of allowed process for this socket */
1595 { "proto", bind_parse_proto, 1 }, /* set the proto to use for all incoming connections */
1596 { /* END */ },
1597 }};
1598
1599 INITCALL1(STG_REGISTER, bind_register_keywords, &bind_kws);
1600
1601 /* config keyword parsers */
1602 static struct cfg_kw_list cfg_kws = {ILH, {
1603 { CFG_GLOBAL, "tune.listener.multi-queue", cfg_parse_tune_listener_mq },
1604 { 0, NULL, NULL }
1605 }};
1606
1607 INITCALL1(STG_REGISTER, cfg_register_keywords, &cfg_kws);
1608
1609 /*
1610 * Local variables:
1611 * c-indent-level: 8
1612 * c-basic-offset: 8
1613 * End:
1614 */
1615