Coverage Report

Created: 2026-10-16 06:09

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
/src/haproxy/src/fd.c
Line
Count
Source
1
/*
2
 * File descriptors management functions.
3
 *
4
 * Copyright 2000-2014 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
 * There is no direct link between the FD and the updates list. There is only a
12
 * bit in the fdtab[] to indicate than a file descriptor is already present in
13
 * the updates list. Once an fd is present in the updates list, it will have to
14
 * be considered even if its changes are reverted in the middle or if the fd is
15
 * replaced.
16
 *
17
 * The event state for an FD, as found in fdtab[].state, is maintained for each
18
 * direction. The state field is built this way, with R bits in the low nibble
19
 * and W bits in the high nibble for ease of access and debugging :
20
 *
21
 *               7    6    5    4   3    2    1    0
22
 *             [ 0 |  0 | RW | AW | 0 |  0 | RR | AR ]
23
 *
24
 *                   A* = active     *R = read
25
 *                   R* = ready      *W = write
26
 *
27
 * An FD is marked "active" when there is a desire to use it.
28
 * An FD is marked "ready" when it has not faced a new EAGAIN since last wake-up
29
 * (it is a cache of the last EAGAIN regardless of polling changes). Each poller
30
 * has its own "polled" state for the same fd, as stored in the polled_mask.
31
 *
32
 * We have 4 possible states for each direction based on these 2 flags :
33
 *
34
 *   +---+---+----------+---------------------------------------------+
35
 *   | R | A | State    | Description                                 |
36
 *   +---+---+----------+---------------------------------------------+
37
 *   | 0 | 0 | DISABLED | No activity desired, not ready.             |
38
 *   | 0 | 1 | ACTIVE   | Activity desired.                           |
39
 *   | 1 | 0 | STOPPED  | End of activity.                            |
40
 *   | 1 | 1 | READY    | Activity desired and reported.              |
41
 *   +---+---+----------+---------------------------------------------+
42
 *
43
 * The transitions are pretty simple :
44
 *   - fd_want_*() : set flag A
45
 *   - fd_stop_*() : clear flag A
46
 *   - fd_cant_*() : clear flag R (when facing EAGAIN)
47
 *   - fd_may_*()  : set flag R (upon return from poll())
48
 *
49
 * Each poller then computes its own polled state :
50
 *     if (A) { if (!R) P := 1 } else { P := 0 }
51
 *
52
 * The state transitions look like the diagram below.
53
 *
54
 *     may  +----------+
55
 *     ,----| DISABLED |    (READY=0, ACTIVE=0)
56
 *     |    +----------+
57
 *     |  want |  ^
58
 *     |       |  |
59
 *     |       v  | stop
60
 *     |    +----------+
61
 *     |    |  ACTIVE  |    (READY=0, ACTIVE=1)
62
 *     |    +----------+
63
 *     |       |  ^
64
 *     |  may  |  |
65
 *     |       v  | EAGAIN (can't)
66
 *     |     +--------+
67
 *     |     | READY  |     (READY=1, ACTIVE=1)
68
 *     |     +--------+
69
 *     |  stop |  ^
70
 *     |       |  |
71
 *     |       v  | want
72
 *     |    +---------+
73
 *     `--->| STOPPED |     (READY=1, ACTIVE=0)
74
 *          +---------+
75
 */
76
77
#include <stdio.h>
78
#include <string.h>
79
#include <unistd.h>
80
#include <fcntl.h>
81
#include <sys/types.h>
82
#include <sys/resource.h>
83
#include <sys/uio.h>
84
85
#if defined(USE_POLL)
86
#include <poll.h>
87
#endif
88
#include <errno.h>
89
90
#include <haproxy/api.h>
91
#include <haproxy/activity.h>
92
#include <haproxy/cfgparse.h>
93
#include <haproxy/fd.h>
94
#include <haproxy/global.h>
95
#include <haproxy/log.h>
96
#include <haproxy/port_range.h>
97
#include <haproxy/ticks.h>
98
#include <haproxy/tools.h>
99
100
101
THREAD_LOCAL struct fdtab *fdtab = NULL;  /* array of all the file descriptors */
102
static struct fdtab *_fdtab;
103
THREAD_LOCAL struct polled_mask *polled_mask = NULL;  /* Array for the polled_mask of each fd */
104
static struct polled_mask *_polled_mask;
105
int totalconn;                  /* total # of terminated sessions */
106
int actconn;                    /* # of active sessions */
107
108
struct poller pollers[MAX_POLLERS] __read_mostly;
109
struct poller cur_poller           __read_mostly;
110
int nbpollers = 0;
111
112
volatile struct fdlist update_list[MAX_TGROUPS]; // Global update list
113
114
THREAD_LOCAL int *fd_updt  = NULL;  // FD updates list
115
THREAD_LOCAL int  fd_nbupdt = 0;   // number of updates in the list
116
THREAD_LOCAL int  fd_highest = -1; // highest FD known by the current thread
117
THREAD_LOCAL int poller_rd_pipe = -1; // Pipe to wake the thread
118
int poller_wr_pipe[MAX_THREADS] __read_mostly; // Pipe to wake the threads
119
120
volatile int ha_used_fds = 0; // Number of FD we're currently using
121
122
/* adds fd <fd> to fd list <list> if it was not yet in it */
123
void fd_add_to_fd_list(volatile struct fdlist *list, int fd)
124
0
{
125
0
  int next;
126
0
  int new;
127
0
  int old;
128
0
  int last;
129
130
0
redo_next:
131
0
  next = HA_ATOMIC_LOAD(&fdtab[fd].update.next);
132
  /* Check that we're not already in the cache, and if not, lock us. */
133
0
  if (next > -2)
134
0
    goto done;
135
0
  if (next == -2)
136
0
    goto redo_next;
137
0
  if (!_HA_ATOMIC_CAS(&fdtab[fd].update.next, &next, -2))
138
0
    goto redo_next;
139
0
  __ha_barrier_atomic_store();
140
141
0
  new = fd;
142
0
redo_last:
143
  /* First, insert in the linked list */
144
0
  last = list->last;
145
0
  old = -1;
146
147
0
  fdtab[fd].update.prev = -2;
148
  /* Make sure the "prev" store is visible before we update the last entry */
149
0
  __ha_barrier_store();
150
151
0
  if (unlikely(last == -1)) {
152
    /* list is empty, try to add ourselves alone so that list->last=fd */
153
0
    if (unlikely(!_HA_ATOMIC_CAS(&list->last, &old, new)))
154
0
          goto redo_last;
155
156
    /* list->first was necessary -1, we're guaranteed to be alone here */
157
0
    list->first = fd;
158
0
  } else {
159
    /* adding ourselves past the last element
160
     * The CAS will only succeed if its next is -1,
161
     * which means it's in the cache, and the last element.
162
     */
163
0
    if (unlikely(!_HA_ATOMIC_CAS(&fdtab[last].update.next, &old, new)))
164
0
      goto redo_last;
165
166
    /* Then, update the last entry */
167
0
    list->last = fd;
168
0
  }
169
0
  __ha_barrier_store();
170
  /* since we're alone at the end of the list and still locked(-2),
171
   * we know no one tried to add past us. Mark the end of list.
172
   */
173
0
  fdtab[fd].update.prev = last;
174
0
  fdtab[fd].update.next = -1;
175
0
  __ha_barrier_store();
176
0
done:
177
0
  return;
178
0
}
179
180
/* removes fd <fd> from fd list <list> */
181
void fd_rm_from_fd_list(volatile struct fdlist *list, int fd)
182
0
{
183
0
#if defined(HA_HAVE_CAS_DW) || defined(HA_CAS_IS_8B)
184
0
  volatile union {
185
0
    struct fdlist_entry ent;
186
0
    uint64_t u64;
187
0
    uint32_t u32[2];
188
0
  } cur_list, next_list;
189
0
#endif
190
0
  int old;
191
0
  int new = -2;
192
0
  int prev;
193
0
  int next;
194
0
  int last;
195
0
lock_self:
196
0
#if (defined(HA_CAS_IS_8B) || defined(HA_HAVE_CAS_DW))
197
0
  next_list.ent.next = next_list.ent.prev = -2;
198
0
  cur_list.ent = *(volatile typeof(fdtab->update)*)&fdtab[fd].update;
199
  /* First, attempt to lock our own entries */
200
0
  do {
201
    /* The FD is not in the FD cache, give up */
202
0
    if (unlikely(cur_list.ent.next <= -3))
203
0
      return;
204
0
    if (unlikely(cur_list.ent.prev == -2 || cur_list.ent.next == -2))
205
0
      goto lock_self;
206
0
  } while (
207
0
#ifdef HA_CAS_IS_8B
208
0
     unlikely(!_HA_ATOMIC_CAS(((uint64_t *)&fdtab[fd].update), (uint64_t *)&cur_list.u64, next_list.u64))
209
#else
210
     unlikely(!_HA_ATOMIC_DWCAS(((long *)&fdtab[fd].update), (uint32_t *)&cur_list.u32, (const uint32_t *)&next_list.u32))
211
#endif
212
0
      );
213
0
  next = cur_list.ent.next;
214
0
  prev = cur_list.ent.prev;
215
216
#else
217
lock_self_next:
218
  next = HA_ATOMIC_LOAD(&fdtab[fd].update.next);
219
  if (next == -2)
220
    goto lock_self_next;
221
  if (next <= -3)
222
    goto done;
223
  if (unlikely(!_HA_ATOMIC_CAS(&fdtab[fd].update.next, &next, -2)))
224
    goto lock_self_next;
225
lock_self_prev:
226
  prev = HA_ATOMIC_LOAD(&fdtab[fd].update.prev);
227
  if (prev == -2)
228
    goto lock_self_prev;
229
  if (unlikely(!_HA_ATOMIC_CAS(&fdtab[fd].update.prev, &prev, -2)))
230
    goto lock_self_prev;
231
#endif
232
0
  __ha_barrier_atomic_store();
233
234
  /* Now, lock the entries of our neighbours */
235
0
  if (likely(prev != -1)) {
236
0
redo_prev:
237
0
    old = fd;
238
239
0
    if (unlikely(!_HA_ATOMIC_CAS(&fdtab[prev].update.next, &old, new))) {
240
0
      if (unlikely(old == -2)) {
241
        /* Neighbour already locked, give up and
242
         * retry again once he's done
243
         */
244
0
        fdtab[fd].update.prev = prev;
245
0
        __ha_barrier_store();
246
0
        fdtab[fd].update.next = next;
247
0
        __ha_barrier_store();
248
0
        goto lock_self;
249
0
      }
250
0
      goto redo_prev;
251
0
    }
252
0
  }
253
0
  if (likely(next != -1)) {
254
0
redo_next:
255
0
    old = fd;
256
0
    if (unlikely(!_HA_ATOMIC_CAS(&fdtab[next].update.prev, &old, new))) {
257
0
      if (unlikely(old == -2)) {
258
        /* Neighbour already locked, give up and
259
         * retry again once he's done
260
         */
261
0
        if (prev != -1) {
262
0
          fdtab[prev].update.next = fd;
263
0
          __ha_barrier_store();
264
0
        }
265
0
        fdtab[fd].update.prev = prev;
266
0
        __ha_barrier_store();
267
0
        fdtab[fd].update.next = next;
268
0
        __ha_barrier_store();
269
0
        goto lock_self;
270
0
      }
271
0
      goto redo_next;
272
0
    }
273
0
  }
274
0
  if (list->first == fd)
275
0
    list->first = next;
276
0
  __ha_barrier_store();
277
0
  last = list->last;
278
0
  while (unlikely(last == fd && (!_HA_ATOMIC_CAS(&list->last, &last, prev))))
279
0
    __ha_compiler_barrier();
280
  /* Make sure we let other threads know we're no longer in cache,
281
   * before releasing our neighbours.
282
   */
283
0
  __ha_barrier_store();
284
0
  if (likely(prev != -1))
285
0
    fdtab[prev].update.next = next;
286
0
  __ha_barrier_store();
287
0
  if (likely(next != -1))
288
0
    fdtab[next].update.prev = prev;
289
0
  __ha_barrier_store();
290
  /* Ok, now we're out of the fd cache */
291
0
  fdtab[fd].update.next = -(next + 4);
292
0
  __ha_barrier_store();
293
0
done:
294
0
  return;
295
0
}
296
297
/* deletes the FD once nobody uses it anymore, as detected by the caller by its
298
 * thread_mask being zero and its running mask turning to zero. There is no
299
 * protection against concurrent accesses, it's up to the caller to make sure
300
 * only the last thread will call it. If called under isolation, it is safe to
301
 * call this from another group than the FD's. This is only for internal use,
302
 * please use fd_delete() instead.
303
 */
304
void _fd_delete_orphan(int fd)
305
0
{
306
0
  int tgrp = fd_tgid(fd);
307
0
  uint fd_disown;
308
309
0
  fd_disown = fdtab[fd].state & FD_DISOWN;
310
0
  if (fdtab[fd].state & FD_LINGER_RISK) {
311
    /* this is generally set when connecting to servers */
312
0
    DISGUISE(setsockopt(fd, SOL_SOCKET, SO_LINGER,
313
0
         (struct linger *) &nolinger, sizeof(struct linger)));
314
0
  }
315
316
  /* It's expected that a close() will result in the FD disappearing from
317
   * pollers, but some pollers may have some internal bookkeeping to be
318
   * done prior to the call (e.g. remove references from internal tables).
319
   */
320
0
  if (cur_poller.clo)
321
0
    cur_poller.clo(fd);
322
323
  /* now we're about to reset some of this FD's fields. We don't want
324
   * anyone to grab it anymore and we need to make sure those which could
325
   * possibly have stumbled upon it right now are leaving before we
326
   * proceed. This is done in two steps. First we reset the tgid so that
327
   * fd_take_tgid() and fd_grab_tgid() fail, then we wait for existing
328
   * ref counts to drop. Past this point we're alone dealing with the
329
   * FD's thead/running/update/polled masks.
330
   */
331
0
  fd_reset_tgid(fd);
332
333
0
  while (_HA_ATOMIC_LOAD(&fdtab[fd].refc_tgid) != 0) // refc==0 ?
334
0
    __ha_cpu_relax();
335
336
  /* we don't want this FD anymore in the global list */
337
0
  fd_rm_from_fd_list(&update_list[tgrp - 1], fd);
338
339
  /* no more updates on this FD are relevant anymore */
340
0
  HA_ATOMIC_STORE(&fdtab[fd].update_mask, 0);
341
0
  if (fd_nbupdt > 0 && fd_updt[fd_nbupdt - 1] == fd)
342
0
    fd_nbupdt--;
343
344
0
  if (unlikely(fdtab[fd].state & FD_HAS_PORT)) {
345
0
    struct sockaddr_storage sa;
346
0
    socklen_t addrlen = sizeof(sa);
347
0
    int port;
348
349
0
    BUG_ON(!(fdtab[fd].state & FD_OWNER_PR));
350
    /*
351
     * We have to release the port, let's use getsockname()
352
     * to figure out what the port was.
353
     */
354
0
    BUG_ON(getsockname(fd, (struct sockaddr *)&sa, &addrlen) != 0);
355
0
    port = get_host_port(&sa);
356
0
    if (!port)
357
0
      ABORT_NOW();
358
0
    port_range_release_port(fdtab[fd].owner, port);
359
0
  }
360
361
0
  polled_mask[fd].poll_recv = polled_mask[fd].poll_send = 0;
362
363
0
  fdtab[fd].state = 0;
364
365
#ifdef DEBUG_FD
366
  fdtab[fd].event_count = 0;
367
#endif
368
0
  fdtab[fd].owner = NULL;
369
370
  /* perform the close() call last as it's what unlocks the instant reuse
371
   * of this FD by any other thread.
372
   */
373
0
  if (!fd_disown) {
374
0
    fdtab[fd].generation++;
375
0
    close(fd);
376
0
  }
377
0
  _HA_ATOMIC_DEC(&ha_used_fds);
378
0
}
379
380
/* Deletes an FD from the fdsets. The file descriptor is also closed, possibly
381
 * asynchronously. It is safe to call it from another thread from the same
382
 * group as the FD's or from a thread from a different group.
383
 * Calling from a thread from another group is handled by grabbing the
384
 * reference count, adding the FD_MUST_CLOSE and doing the close ourselves
385
 * only if there is no other thread currently using the fd.
386
 */
387
void fd_delete(int fd)
388
0
{
389
  /* This must never happen and would definitely indicate a bug, in
390
   * addition to overwriting some unexpected memory areas.
391
   */
392
0
  BUG_ON(fd < 0 || fd >= global.maxsock);
393
394
  /* NOTE: The master when going into reexec mode re-closes all FDs after
395
   * they were already dispatched. But we know we didn't start the polling
396
   * threads so we can still close them. The masks will probably not match
397
   * however so we force the value and erase the refcount if any.
398
   */
399
0
  if (unlikely(global.mode & MODE_STARTING))
400
0
    fdtab[fd].refc_tgid = ti->tgid;
401
402
  /* the tgid cannot change before a complete close so we should never
403
   * face the situation where we try to close an fd that was reassigned.
404
   * However there is one corner case where this happens, it's when an
405
   * attempt to pause a listener fails (e.g. abns), leaving the listener
406
   * in fault state and it is forcefully stopped.
407
   * To make it work when the fd does not belong to our thread group,
408
   * first increase the tgid refcount, so that we're sure nobody will
409
   * close it before we figure out if we should, then set the
410
   * FD_MUST_CLOSE bit, so that any running thread will take care
411
   * of closing it.
412
   * Once we set the thread_mask to 0, we know for sure that if
413
   * nobody is running, nobody ever will, so at this point, if
414
   * there is no running thread, and the FD_MUST_CLOSE bit is still
415
   * set, we know it is safe to actually close the fd.
416
   */
417
0
  if (unlikely(fd_tgid(fd) != ti->tgid)) {
418
0
    uint tgrp;
419
420
0
    tgrp = fd_take_tgid(fd);
421
422
0
    HA_ATOMIC_OR(&fdtab[fd].state, FD_MUST_CLOSE);
423
0
    HA_ATOMIC_STORE(&fdtab[fd].thread_mask, 0);
424
425
0
    if (!HA_ATOMIC_LOAD(&fdtab[fd].running_mask) &&
426
0
        HA_ATOMIC_BTR(&fdtab[fd].state, FD_MUST_CLOSE_BIT)) {
427
0
      if (tgrp)
428
0
        fd_drop_tgid(fd);
429
0
      _fd_delete_orphan(fd);
430
0
    }
431
0
    else if (tgrp)
432
0
      fd_drop_tgid(fd);
433
0
    return;
434
0
  }
435
436
  /* we must postpone removal of an FD that may currently be in use
437
   * by another thread. This can happen in the following two situations:
438
   *   - after a takeover, the owning thread closes the connection but
439
   *     the previous one just woke up from the poller and entered
440
   *     the FD handler iocb. That thread holds an entry in running_mask
441
   *     and requires removal protection.
442
   *   - multiple threads are accepting connections on a listener, and
443
   *     one of them (or even an separate one) decides to unbind the
444
   *     listener under the listener's lock while other ones still hold
445
   *     the running bit.
446
   * In both situations the FD is marked as unused (thread_mask = 0) and
447
   * will not take new bits in its running_mask so we have the guarantee
448
   * that the last thread eliminating running_mask is the one allowed to
449
   * safely delete the FD. Most of the time it will be the current thread.
450
   * We still need to set and check the one-shot flag FD_MUST_CLOSE
451
   * to take care of the rare cases where a thread wakes up on late I/O
452
   * before the thread_mask is zero, and sets its bit in the running_mask
453
   * just after the current thread finishes clearing its own bit, hence
454
   * the two threads see themselves as last ones (which they really are).
455
   */
456
457
0
  HA_ATOMIC_OR(&fdtab[fd].running_mask, ti->ltid_bit);
458
0
  HA_ATOMIC_OR(&fdtab[fd].state, FD_MUST_CLOSE);
459
0
  HA_ATOMIC_STORE(&fdtab[fd].thread_mask, 0);
460
0
  if (fd_clr_running(fd) == ti->ltid_bit) {
461
0
    if (HA_ATOMIC_BTR(&fdtab[fd].state, FD_MUST_CLOSE_BIT)) {
462
0
      _fd_delete_orphan(fd);
463
0
    }
464
0
  }
465
0
}
466
467
/* makes the new fd non-blocking and clears all other O_* flags; this is meant
468
 * to be used on new FDs. Returns -1 on failure. The result is disguised at the
469
 * end because some callers need to be able to ignore it regardless of the libc
470
 * attributes.
471
 */
472
int fd_set_nonblock(int fd)
473
0
{
474
0
  int ret = fcntl(fd, F_SETFL, O_NONBLOCK);
475
476
0
  return DISGUISE(ret);
477
0
}
478
479
/* sets the close-on-exec flag on fd; returns -1 on failure. The result is
480
 * disguised at the end because some callers need to be able to ignore it
481
 * regardless of the libc attributes.
482
 */
483
int fd_set_cloexec(int fd)
484
0
{
485
0
  int flags, ret;
486
487
0
  flags = fcntl(fd, F_GETFD);
488
0
  flags |= FD_CLOEXEC;
489
0
  ret = fcntl(fd, F_SETFD, flags);
490
0
  return DISGUISE(ret);
491
0
}
492
493
/* Migrate a FD to a new thread <new_tid>. It is explicitly permitted to
494
 * migrate to another thread group, the function takes the necessary locking
495
 * for this. It is even permitted to migrate from a foreign group to another,
496
 * but the calling thread must be certain that the FD is not about to close
497
 * when doing so, reason why it is highly recommended that only one of the
498
 * FD's owners performs this operation. The polling is completely disabled.
499
 * The operation never fails.
500
 */
501
void fd_migrate_on(int fd, uint new_tid)
502
0
{
503
0
  struct thread_info *new_ti = &ha_thread_info[new_tid];
504
505
  /* we must be alone to work on this idle FD. If not, it means that its
506
   * poller is currently waking up and is about to use it, likely to
507
   * close it on shut/error, but maybe also to process any unexpectedly
508
   * pending data. It's also possible that the FD was closed and
509
   * reassigned to another thread group, so let's be careful.
510
   */
511
0
  fd_lock_tgid(fd, new_ti->tgid);
512
513
  /* now we have exclusive access to it. From now FD belongs to tid_bit
514
   * for this tgid.
515
   */
516
0
  HA_ATOMIC_STORE(&fdtab[fd].thread_mask, new_ti->ltid_bit);
517
518
  /* Make sure the FD doesn't have the active bit. It is possible that
519
   * the fd is polled by the thread that used to own it, the new thread
520
   * is supposed to call subscribe() later, to activate polling.
521
   */
522
0
  fd_stop_both(fd);
523
524
  /* we're done with it. As soon as we unlock it, other threads from the
525
   * target group can manipulate it. However it may only disappear once
526
   * we drop the reference.
527
   */
528
0
  fd_unlock_tgid(fd);
529
0
  fd_drop_tgid(fd);
530
0
}
531
532
/*
533
 * Take over a FD belonging to another thread.
534
 * unexpected_conn is the expected owner of the fd.
535
 * Returns 0 on success, and -1 on failure.
536
 */
537
int fd_takeover(int fd, void *expected_owner)
538
0
{
539
0
  unsigned long old;
540
0
  int changing_tgid = 0;
541
0
  int old_ltid, old_tgid;
542
543
  /* protect ourself against a delete then an insert for the same fd,
544
   * if it happens, then the owner will no longer be the expected
545
   * connection.
546
   */
547
0
  if (fdtab[fd].owner != expected_owner)
548
0
    return -1;
549
550
  /* We're taking a connection from a different thread group */
551
0
  if ((fdtab[fd].refc_tgid & 0x7fff) != tgid) {
552
0
    BUG_ON(global.tune.options & GTUNE_NO_TG_FD_SHARING);
553
554
0
    changing_tgid = 1;
555
556
0
    old_tgid = fd_tgid(fd);
557
0
    BUG_ON(atleast2(fdtab[fd].thread_mask));
558
0
    old_ltid = my_ffsl(fdtab[fd].thread_mask) - 1;
559
560
0
    if (unlikely(!fd_lock_tgid_cur(fd)))
561
0
      return -1;
562
0
  } else {
563
    /* we must be alone to work on this idle FD. If not, it means that its
564
     * poller is currently waking up and is about to use it, likely to
565
     * close it on shut/error, but maybe also to process any unexpectedly
566
     * pending data. It's also possible that the FD was closed and
567
     * reassigned to another thread group, so let's be careful.
568
     */
569
0
    if (unlikely(!fd_grab_tgid(fd, ti->tgid)))
570
0
      return -1;
571
0
  }
572
573
0
  old = 0;
574
0
  if (!HA_ATOMIC_CAS(&fdtab[fd].running_mask, &old, ti->ltid_bit)) {
575
0
    if (changing_tgid)
576
0
      fd_unlock_tgid(fd);
577
0
    fd_drop_tgid(fd);
578
0
    return -1;
579
0
  }
580
581
  /* success, from now on it's ours */
582
0
  HA_ATOMIC_STORE(&fdtab[fd].thread_mask, ti->ltid_bit);
583
584
  /*
585
   * Change the tgid to our own tgid.
586
   * This removes the lock, we don't need it anymore, but we keep
587
   * the refcount.
588
   */
589
0
  if (changing_tgid) {
590
0
    fd_update_tgid(fd, tgid);
591
0
    if (cur_poller.fixup_tgid_takeover)
592
0
      cur_poller.fixup_tgid_takeover(&cur_poller, fd, old_ltid, old_tgid);
593
0
  }
594
595
  /* Make sure the FD doesn't have the active bit. It is possible that
596
   * the fd is polled by the thread that used to own it, the new thread
597
   * is supposed to call subscribe() later, to activate polling.
598
   */
599
0
  fd_stop_recv(fd);
600
601
  /* essentially for debugging */
602
0
  fdtab[fd].nb_takeover++;
603
604
  /* we're done with it */
605
0
  HA_ATOMIC_AND(&fdtab[fd].running_mask, ~ti->ltid_bit);
606
607
  /* no more changes planned */
608
0
  fd_drop_tgid(fd);
609
0
  return 0;
610
0
}
611
612
void updt_fd_polling(const int fd)
613
0
{
614
0
  uint tgrp = fd_take_tgid(fd);
615
616
  /* closed ? may happen */
617
0
  if (!tgrp)
618
0
    return;
619
620
0
  if (unlikely(tgrp != tgid && tgrp <= MAX_TGROUPS)) {
621
    /* Hmmm delivered an update for another group... That may
622
     * happen on suspend/resume of a listener for example when
623
     * the FD was not even marked for running. Let's broadcast
624
     * the update.
625
     */
626
0
    unsigned long update_mask = fdtab[fd].update_mask;
627
0
    int thr;
628
629
0
    while (!_HA_ATOMIC_CAS(&fdtab[fd].update_mask, &update_mask,
630
0
                           _HA_ATOMIC_LOAD(&ha_tgroup_info[tgrp - 1].threads_enabled)))
631
0
      __ha_cpu_relax();
632
633
0
    fd_add_to_fd_list(&update_list[tgrp - 1], fd);
634
635
0
    thr = one_among_mask(fdtab[fd].thread_mask & ha_tgroup_info[tgrp - 1].threads_enabled,
636
0
                         statistical_prng_range(ha_tgroup_info[tgrp - 1].count));
637
0
    thr += ha_tgroup_info[tgrp - 1].base;
638
0
    wake_thread(thr);
639
640
0
    fd_drop_tgid(fd);
641
0
    return;
642
0
  }
643
644
0
  fd_drop_tgid(fd);
645
646
0
  if (tg->threads_enabled == 1UL || (fdtab[fd].thread_mask & tg->threads_enabled) == ti->ltid_bit) {
647
0
    if (HA_ATOMIC_BTS(&fdtab[fd].update_mask, ti->ltid))
648
0
      return;
649
650
0
    fd_updt[fd_nbupdt++] = fd;
651
0
  } else {
652
0
    unsigned long update_mask = fdtab[fd].update_mask;
653
0
    do {
654
0
      if (update_mask == fdtab[fd].thread_mask) // FIXME: this works only on thread-groups 1
655
0
        return;
656
0
    } while (!_HA_ATOMIC_CAS(&fdtab[fd].update_mask, &update_mask, fdtab[fd].thread_mask));
657
658
0
    fd_add_to_fd_list(&update_list[tgid - 1], fd);
659
660
0
    if (fd_active(fd) && !(fdtab[fd].thread_mask & ti->ltid_bit)) {
661
      /* we need to wake up another thread to handle it immediately, any will fit,
662
       * so let's pick a random one so that it doesn't always end up on the same.
663
       */
664
0
      int thr = one_among_mask(fdtab[fd].thread_mask & tg->threads_enabled,
665
0
                               statistical_prng_range(tg->count));
666
0
      thr += tg->base;
667
0
      wake_thread(thr);
668
0
    }
669
0
  }
670
0
}
671
672
/* Update events seen for FD <fd> and its state if needed. This should be
673
 * called by the poller, passing FD_EV_*_{R,W,RW} in <evts>. FD_EV_ERR_*
674
 * doesn't need to also pass FD_EV_SHUT_*, it's implied. ERR and SHUT are
675
 * allowed to be reported regardless of R/W readiness. Returns one of
676
 * FD_UPDT_*.
677
 */
678
int fd_update_events(int fd, uint evts)
679
0
{
680
0
  unsigned long locked;
681
0
  uint old, new;
682
0
  uint new_flags, must_stop;
683
0
  ulong rmask, tmask;
684
685
0
  _HA_ATOMIC_AND(&th_ctx->flags, ~TH_FL_STUCK); // this thread is still running
686
687
0
  if (unlikely(!fd_grab_tgid(fd, ti->tgid))) {
688
    /* the FD changed to another tgid, we can't safely
689
     * check it anymore. The bits in the masks are not
690
     * ours anymore and we're not allowed to touch them.
691
     * Ours have already been cleared and the FD was
692
     * closed in between so we can safely leave now.
693
     */
694
0
    activity[tid].poll_drop_fd++;
695
0
    return FD_UPDT_CLOSED;
696
0
  }
697
698
  /* Do not take running_mask if not strictly needed (will trigger a
699
   * cosmetic BUG_ON() in fd_insert() anyway if done).
700
   */
701
0
  tmask = _HA_ATOMIC_LOAD(&fdtab[fd].thread_mask);
702
0
  if (!(tmask & ti->ltid_bit))
703
0
    goto do_update;
704
705
0
  HA_ATOMIC_OR(&fdtab[fd].running_mask, ti->ltid_bit);
706
707
  /* From this point, our bit may possibly be in thread_mask, but it may
708
   * still vanish, either because a takeover completed just before taking
709
   * the bit above with the new owner deleting the FD, or because a
710
   * takeover started just before taking the bit. In order to make sure a
711
   * started takeover is complete, we need to verify that all bits of
712
   * running_mask are present in thread_mask, since takeover first takes
713
   * running then atomically replaces thread_mask. Once it's stable, if
714
   * our bit remains there, no further takeover may happen because we
715
   * hold running, but if our bit is not there it means we've lost the
716
   * takeover race and have to decline touching the FD. Regarding the
717
   * risk of deletion, our bit in running_mask prevents fd_delete() from
718
   * finalizing the close, and the caller will leave the FD with a zero
719
   * thread_mask and the FD_MUST_CLOSE flag set. It will then be our
720
   * responsibility to close it.
721
   */
722
0
  do {
723
0
    rmask = _HA_ATOMIC_LOAD(&fdtab[fd].running_mask);
724
0
    tmask = _HA_ATOMIC_LOAD(&fdtab[fd].thread_mask);
725
0
    rmask &= ~ti->ltid_bit;
726
0
  } while ((rmask & ~tmask) && (tmask & ti->ltid_bit));
727
728
  /* Now tmask is stable. Do nothing if the FD was taken over under us */
729
730
0
  if (!(tmask & ti->ltid_bit)) {
731
    /* a takeover has started */
732
0
    activity[tid].poll_skip_fd++;
733
734
0
    if (fd_clr_running(fd) == ti->ltid_bit)
735
0
      goto closed_or_migrated;
736
737
0
    goto do_update;
738
0
  }
739
740
  /* with running we're safe now, we can drop the reference */
741
0
  fd_drop_tgid(fd);
742
743
0
  locked = (tmask != ti->ltid_bit);
744
745
  /* OK now we are guaranteed that our thread_mask was present and
746
   * that we're allowed to update the FD.
747
   */
748
749
0
  new_flags =
750
0
        ((evts & FD_EV_READY_R) ? FD_POLL_IN  : 0) |
751
0
        ((evts & FD_EV_READY_W) ? FD_POLL_OUT : 0) |
752
0
        ((evts & FD_EV_SHUT_R)  ? FD_POLL_HUP : 0) |
753
0
        ((evts & FD_EV_ERR_RW)  ? FD_POLL_ERR : 0);
754
755
  /* SHUTW reported while FD was active for writes is an error */
756
0
  if ((fdtab[fd].state & FD_EV_ACTIVE_W) && (evts & FD_EV_SHUT_W))
757
0
    new_flags |= FD_POLL_ERR;
758
759
  /* compute the inactive events reported late that must be stopped */
760
0
  must_stop = 0;
761
0
  if (unlikely(!fd_active(fd))) {
762
    /* both sides stopped */
763
0
    must_stop = FD_POLL_IN | FD_POLL_OUT;
764
0
  }
765
0
  else if (unlikely(!fd_recv_active(fd) && (evts & (FD_EV_READY_R | FD_EV_SHUT_R | FD_EV_ERR_RW)))) {
766
    /* only send remains */
767
0
    must_stop = FD_POLL_IN;
768
0
  }
769
0
  else if (unlikely(!fd_send_active(fd) && (evts & (FD_EV_READY_W | FD_EV_SHUT_W | FD_EV_ERR_RW)))) {
770
    /* only recv remains */
771
0
    must_stop = FD_POLL_OUT;
772
0
  }
773
774
0
  if (new_flags & (FD_POLL_IN | FD_POLL_HUP | FD_POLL_ERR))
775
0
    new_flags |= FD_EV_READY_R;
776
777
0
  if (new_flags & (FD_POLL_OUT | FD_POLL_ERR))
778
0
    new_flags |= FD_EV_READY_W;
779
780
0
  old = fdtab[fd].state;
781
0
  new = (old & ~FD_POLL_UPDT_MASK) | new_flags;
782
783
0
  if (unlikely(locked)) {
784
    /* Locked FDs (those with more than 2 threads) are atomically updated */
785
0
    while (unlikely(new != old && !_HA_ATOMIC_CAS(&fdtab[fd].state, &old, new)))
786
0
      new = (old & ~FD_POLL_UPDT_MASK) | new_flags;
787
0
  } else {
788
0
    if (new != old)
789
0
      fdtab[fd].state = new;
790
0
  }
791
792
0
  if (fdtab[fd].iocb && fd_active(fd)) {
793
0
    fdtab[fd].iocb(fd);
794
0
  }
795
796
  /*
797
   * We entered iocb with running set and with the valid tgid.
798
   * Since then, this is what could have happened:
799
   *   - another thread tried to close the FD (e.g. timeout task from
800
   *     another one that owns it). We still have running set, but not
801
   *     tmask. We must call fd_clr_running() then _fd_delete_orphan()
802
   *     if we were the last one.
803
   *
804
   *   - the iocb tried to close the FD => bit no more present in running,
805
   *     nothing to do. If it managed to close it, the poller's ->clo()
806
   *     has already been called.
807
   *
808
   *   - after we closed, the FD was reassigned to another thread in
809
   *     another group => running not present, tgid differs, nothing to
810
   *     do because if it got reassigned it indicates it was already
811
   *     closed.
812
   *
813
   * There's no risk of takeover of the valid FD here during this period.
814
   * Also if we still have running, immediately after we release it, the
815
   * events above might instantly happen due to another thread taking
816
   * over.
817
   *
818
   * As such, the only cases where the FD is still relevant are:
819
   *   - tgid still set and running still set (most common)
820
   *   - tgid still valid but running cleared due to fd_delete(): we may
821
   *     still need to stop polling otherwise we may keep it enabled
822
   *     while waiting for other threads to close it.
823
   * And given that we may need to program a tentative update in case we
824
   * don't immediately close, it's easier to grab the tgid during the
825
   * whole check.
826
   */
827
828
0
  if (!fd_grab_tgid(fd, tgid))
829
0
    return FD_UPDT_CLOSED;
830
831
0
  tmask = _HA_ATOMIC_LOAD(&fdtab[fd].thread_mask);
832
833
  /* another thread might have attempted to close this FD in the mean
834
   * time (e.g. timeout task) striking on a previous thread and closing.
835
   * This is detected by us being the last owners of a running_mask bit,
836
   * and the thread_mask being zero. At the moment we release the running
837
   * bit, a takeover may also happen, so in practice we check for our loss
838
   * of the thread_mask bitboth thread_mask and running_mask being 0 after
839
   * we remove ourselves last. There is no risk the FD gets reassigned
840
   * to a different group since it's not released until the real close()
841
   * in _fd_delete_orphan().
842
   */
843
0
  if (fd_clr_running(fd) == ti->ltid_bit && !(tmask & ti->ltid_bit))
844
0
    goto closed_or_migrated;
845
846
  /* we had to stop this FD and it still must be stopped after the I/O
847
   * cb's changes, so let's program an update for this.
848
   */
849
0
  if (must_stop && !(fdtab[fd].update_mask & ti->ltid_bit)) {
850
0
    if (((must_stop & FD_POLL_IN)  && !fd_recv_active(fd)) ||
851
0
        ((must_stop & FD_POLL_OUT) && !fd_send_active(fd)))
852
0
      if (!HA_ATOMIC_BTS(&fdtab[fd].update_mask, ti->ltid))
853
0
        fd_updt[fd_nbupdt++] = fd;
854
0
  }
855
856
0
  fd_drop_tgid(fd);
857
0
  return FD_UPDT_DONE;
858
859
0
 closed_or_migrated:
860
  /* We only come here once we've last dropped running and the FD is
861
   * not for us as per !(tmask & tid_bit). It may imply we're
862
   * responsible for closing it. Otherwise it's just a migration.
863
   */
864
0
  if (HA_ATOMIC_BTR(&fdtab[fd].state, FD_MUST_CLOSE_BIT)) {
865
0
    fd_drop_tgid(fd);
866
0
    _fd_delete_orphan(fd);
867
0
    return FD_UPDT_CLOSED;
868
0
  }
869
870
  /* So we were alone, no close bit, at best the FD was migrated, at
871
   * worst it's in the process of being closed by another thread. We must
872
   * be ultra-careful as it can be re-inserted by yet another thread as
873
   * the result of socket() or accept(). Let's just tell the poller the
874
   * FD was lost. If it was closed it was already removed and this will
875
   * only cost an update for nothing.
876
   */
877
878
0
 do_update:
879
  /* The FD is not closed but we don't want the poller to wake up for
880
   * it anymore.
881
   */
882
0
  if (!HA_ATOMIC_BTS(&fdtab[fd].update_mask, ti->ltid))
883
0
    fd_updt[fd_nbupdt++] = fd;
884
885
0
  fd_drop_tgid(fd);
886
0
  return FD_UPDT_MIGRATED;
887
0
}
888
889
/* This is used by pollers at boot time to re-register desired events for
890
 * all FDs after new pollers have been created. It doesn't do much, it checks
891
 * that their thread group matches the one in argument, and that the thread
892
 * mask matches at least one of the bits in the mask, and if so, marks the FD
893
 * as updated.
894
 */
895
void fd_reregister_all(int tgrp, ulong mask)
896
0
{
897
0
  int fd;
898
899
0
  for (fd = 0; fd < fd_highest; fd++) {
900
0
    if (!fdtab[fd].owner)
901
0
      continue;
902
903
    /* make sure we don't register other tgroups' FDs. We just
904
     * avoid needlessly taking the lock if not needed.
905
     */
906
0
    if (!(_HA_ATOMIC_LOAD(&fdtab[fd].thread_mask) & mask) ||
907
0
        !fd_grab_tgid(fd, tgrp))
908
0
      continue;  // was not for us anyway
909
910
0
    if (_HA_ATOMIC_LOAD(&fdtab[fd].thread_mask) & mask)
911
0
      updt_fd_polling(fd);
912
0
    fd_drop_tgid(fd);
913
0
  }
914
0
}
915
916
/* Tries to send <npfx> parts from <prefix> followed by <nmsg> parts from <msg>
917
 * optionally followed by a newline if <nl> is non-null, to file descriptor
918
 * <fd>. The message is sent atomically using writev(). It may be truncated to
919
 * <maxlen> bytes if <maxlen> is non-null. There is no distinction between the
920
 * two lists, it's just a convenience to help the caller prepend some prefixes
921
 * when necessary. It takes the fd's lock to make sure no other thread will
922
 * write to the same fd in parallel. Returns the number of bytes sent, or <=0
923
 * on failure. A limit to 31 total non-empty segments is enforced. The caller
924
 * is responsible for taking care of making the fd non-blocking.
925
 */
926
ssize_t fd_write_frag_line(int fd, size_t maxlen, const struct ist pfx[], size_t npfx, const struct ist msg[], size_t nmsg, int nl)
927
0
{
928
0
  struct iovec iovec[32];
929
0
  size_t sent = 0;
930
0
  int vec = 0;
931
0
  int attempts = 0;
932
933
0
  if (!maxlen)
934
0
    maxlen = ~0;
935
936
  /* keep one char for a possible trailing '\n' in any case */
937
0
  maxlen--;
938
939
  /* make an iovec from the concatenation of all parts of the original
940
   * message. Skip empty fields and truncate the whole message to maxlen,
941
   * leaving one spare iovec for the '\n'.
942
   */
943
0
  while (vec < (sizeof(iovec) / sizeof(iovec[0]) - 1)) {
944
0
    if (!npfx) {
945
0
      pfx = msg;
946
0
      npfx = nmsg;
947
0
      nmsg = 0;
948
0
      if (!npfx)
949
0
        break;
950
0
    }
951
952
0
    iovec[vec].iov_base = pfx->ptr;
953
0
    iovec[vec].iov_len  = MIN(maxlen, pfx->len);
954
0
    maxlen -= iovec[vec].iov_len;
955
0
    if (iovec[vec].iov_len)
956
0
      vec++;
957
0
    pfx++; npfx--;
958
0
  };
959
960
0
  if (nl) {
961
0
    iovec[vec].iov_base = "\n";
962
0
    iovec[vec].iov_len  = 1;
963
0
    vec++;
964
0
  }
965
966
  /* make sure we never interleave writes and we never block. This means
967
   * we prefer to fail on collision than to block. But we don't want to
968
   * lose too many logs so we just perform a few lock attempts then give
969
   * up.
970
   */
971
972
0
  while (HA_ATOMIC_BTS(&fdtab[fd].state, FD_EXCL_SYSCALL_BIT)) {
973
0
    if (++attempts >= 200) {
974
      /* so that the caller knows the message couldn't be delivered */
975
0
      sent = -1;
976
0
      errno = EAGAIN;
977
0
      goto leave;
978
0
    }
979
0
    ha_thread_relax();
980
0
  }
981
982
0
  if (unlikely(!(fdtab[fd].state & FD_INITIALIZED))) {
983
0
    HA_ATOMIC_OR(&fdtab[fd].state, FD_INITIALIZED);
984
0
    if (!isatty(fd))
985
0
      fd_set_nonblock(fd);
986
0
  }
987
0
  sent = writev(fd, iovec, vec);
988
0
  HA_ATOMIC_BTR(&fdtab[fd].state, FD_EXCL_SYSCALL_BIT);
989
990
0
 leave:
991
  /* sent > 0 if the message was delivered */
992
0
  return sent;
993
0
}
994
995
#if defined(USE_CLOSEFROM)
996
void my_closefrom(int start)
997
{
998
  closefrom(start);
999
}
1000
1001
#elif defined(USE_POLL)
1002
/* This is a portable implementation of closefrom(). It closes all open file
1003
 * descriptors starting at <start> and above. It relies on the fact that poll()
1004
 * will return POLLNVAL for each invalid (hence close) file descriptor passed
1005
 * in argument in order to skip them. It acts with batches of FDs and will
1006
 * typically perform one poll() call per 1024 FDs so the overhead is low in
1007
 * case all FDs have to be closed.
1008
 */
1009
void my_closefrom(int start)
1010
0
{
1011
0
  struct pollfd poll_events[1024];
1012
0
  struct rlimit limit;
1013
0
  int nbfds, fd, ret, idx;
1014
0
  int step, next;
1015
1016
0
  if (getrlimit(RLIMIT_NOFILE, &limit) == 0)
1017
0
    step = nbfds = limit.rlim_cur;
1018
0
  else
1019
0
    step = nbfds = 0;
1020
1021
0
  if (nbfds <= 0) {
1022
    /* set safe limit */
1023
0
    nbfds = 1024;
1024
0
    step = 256;
1025
0
  }
1026
1027
0
  if (step > sizeof(poll_events) / sizeof(poll_events[0]))
1028
0
    step = sizeof(poll_events) / sizeof(poll_events[0]);
1029
1030
0
  while (start < nbfds) {
1031
0
    next = (start / step + 1) * step;
1032
1033
0
    for (fd = start; fd < next && fd < nbfds; fd++) {
1034
0
      poll_events[fd - start].fd = fd;
1035
0
      poll_events[fd - start].events = 0;
1036
0
    }
1037
1038
0
    do {
1039
0
      ret = poll(poll_events, fd - start, 0);
1040
0
      if (ret >= 0)
1041
0
        break;
1042
0
    } while (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR || errno == ENOMEM);
1043
1044
    /* always check the whole range */
1045
0
    ret = fd - start;
1046
1047
0
    for (idx = 0; idx < ret; idx++) {
1048
0
      if (poll_events[idx].revents & POLLNVAL)
1049
0
        continue; /* already closed */
1050
1051
0
      fd = poll_events[idx].fd;
1052
0
      close(fd);
1053
0
    }
1054
0
    start = next;
1055
0
  }
1056
0
}
1057
1058
#else // defined(USE_POLL)
1059
1060
/* This is a portable implementation of closefrom(). It closes all open file
1061
 * descriptors starting at <start> and above. This is a naive version for use
1062
 * when the operating system provides no alternative.
1063
 */
1064
void my_closefrom(int start)
1065
{
1066
  struct rlimit limit;
1067
  int nbfds;
1068
1069
  if (getrlimit(RLIMIT_NOFILE, &limit) == 0)
1070
    nbfds = limit.rlim_cur;
1071
  else
1072
    nbfds = 0;
1073
1074
  if (nbfds <= 0)
1075
    nbfds = 1024; /* safe limit */
1076
1077
  while (start < nbfds)
1078
    close(start++);
1079
}
1080
#endif // defined(USE_POLL)
1081
1082
1083
/* Computes the bounded poll() timeout based on the next expiration timer <next>
1084
 * by bounding it to MAX_DELAY_MS. <next> may equal TICK_ETERNITY. The pollers
1085
 * just needs to call this function right before polling to get their timeout
1086
 * value. Timeouts that are already expired (possibly due to a pending event)
1087
 * are accounted for in activity.poll_exp.
1088
 */
1089
int compute_poll_timeout(int next)
1090
0
{
1091
0
  int wait_time;
1092
1093
0
  if (!tick_isset(next))
1094
0
    wait_time = MAX_DELAY_MS;
1095
0
  else if (tick_is_expired(next, now_ms)) {
1096
0
    activity[tid].poll_exp++;
1097
0
    wait_time = 0;
1098
0
  }
1099
0
  else {
1100
0
    wait_time = TICKS_TO_MS(tick_remain(now_ms, next));
1101
0
    if (wait_time > MAX_DELAY_MS)
1102
0
      wait_time = MAX_DELAY_MS;
1103
0
  }
1104
0
  return wait_time;
1105
0
}
1106
1107
/* Handle the return of the poller, which consists in calculating the idle
1108
 * time, saving a few clocks, marking the thread harmful again etc. All that
1109
 * is some boring stuff that all pollers have to do anyway.
1110
 */
1111
void fd_leaving_poll(int wait_time, int status)
1112
0
{
1113
0
  clock_leaving_poll(wait_time, status);
1114
1115
0
  thread_harmless_end();
1116
0
  thread_idle_end();
1117
1118
0
  _HA_ATOMIC_AND(&th_ctx->flags, ~TH_FL_SLEEPING);
1119
0
}
1120
1121
/* disable the specified poller */
1122
void disable_poller(const char *poller_name)
1123
0
{
1124
0
  int p;
1125
1126
0
  for (p = 0; p < nbpollers; p++)
1127
0
    if (strcmp(pollers[p].name, poller_name) == 0)
1128
0
      pollers[p].pref = 0;
1129
0
}
1130
1131
void poller_pipe_io_handler(int fd)
1132
0
{
1133
0
  char buf[1024];
1134
  /* Flush the pipe */
1135
0
  while (read(fd, buf, sizeof(buf)) > 0);
1136
0
  fd_cant_recv(fd);
1137
0
}
1138
1139
/* allocate the per-thread fd_updt thus needs to be called early after
1140
 * thread creation.
1141
 */
1142
static int alloc_pollers_per_thread()
1143
0
{
1144
0
  fd_updt = calloc(global.maxsock, sizeof(*fd_updt));
1145
0
  vma_set_name_id(fd_updt, global.maxsock * sizeof(*fd_updt), "fd", "fd_updt", tid + 1);
1146
0
  return fd_updt != NULL;
1147
0
}
1148
1149
static int precreated_poller_pipes[MAX_THREADS][2];
1150
1151
/*
1152
 * Create the pipes before we create the threads, as they may not share
1153
 * their file descriptor tables.
1154
 */
1155
int fd_precreate_poller_pipes(void)
1156
0
{
1157
0
  int i;
1158
1159
0
  for (i = 0; i < global.nbthread; i++) {
1160
0
    if (pipe(precreated_poller_pipes[i]) < 0)
1161
0
      return 0;
1162
0
  }
1163
0
  return 1;
1164
0
}
1165
1166
/* Initialize the pollers per thread.*/
1167
static int init_pollers_per_thread()
1168
0
{
1169
0
  int mypipe[2];
1170
1171
  /*
1172
   * The master does not get to pre-create pipes, so do it
1173
   * now.
1174
   */
1175
0
  if (!master) {
1176
0
    mypipe[0] = precreated_poller_pipes[tid][0];
1177
0
    mypipe[1] = precreated_poller_pipes[tid][1];
1178
0
  }
1179
0
  else if (pipe(mypipe) < 0)
1180
0
    return 0;
1181
1182
0
  poller_rd_pipe = mypipe[0];
1183
0
  poller_wr_pipe[tid] = mypipe[1];
1184
0
  fd_set_nonblock(poller_rd_pipe);
1185
0
  fd_insert(poller_rd_pipe, poller_pipe_io_handler, poller_pipe_io_handler, tgid, ti->ltid_bit);
1186
0
  fd_insert(poller_wr_pipe[tid], poller_pipe_io_handler, poller_pipe_io_handler, tgid, ti->ltid_bit);
1187
0
  fd_want_recv(poller_rd_pipe);
1188
0
  fd_stop_both(poller_wr_pipe[tid]);
1189
0
  return 1;
1190
0
}
1191
1192
/* Deinitialize the pollers per thread */
1193
static void deinit_pollers_per_thread()
1194
0
{
1195
  /* rd and wr are init at the same place, but only rd is init to -1, so
1196
    we rely to rd to close.   */
1197
0
  if (poller_rd_pipe > -1) {
1198
0
    fd_delete(poller_rd_pipe);
1199
0
    poller_rd_pipe = -1;
1200
0
    fd_delete(poller_wr_pipe[tid]);
1201
0
    poller_wr_pipe[tid] = -1;
1202
0
  }
1203
0
}
1204
1205
/* Release the pollers per thread, to be called late */
1206
static void free_pollers_per_thread()
1207
0
{
1208
0
  fd_nbupdt = 0;
1209
0
  ha_free(&fd_updt);
1210
0
}
1211
1212
/*
1213
 * Initialize the pollers till the best one is found.
1214
 * If none works, returns 0, otherwise 1.
1215
 */
1216
int init_pollers()
1217
0
{
1218
0
  int p;
1219
0
  struct poller *bp;
1220
0
  int nbfdtab;
1221
1222
1223
0
  if (global.tune.options & GTUNE_NO_TG_FD_SHARING) {
1224
0
    nbfdtab = global.nbtgroups;
1225
0
  } else
1226
0
    nbfdtab = 1;
1227
  /* always provide an aligned fdtab */
1228
0
  if ((_fdtab = ha_aligned_zalloc(64, array_size_or_fail(global.maxsock, nbfdtab * sizeof(*fdtab)))) == NULL) {
1229
0
    ha_alert("Not enough memory to allocate %d entries for fdtab!\n", global.maxsock);
1230
0
    goto fail_tab;
1231
0
  }
1232
1233
0
  for (p = 0; p < global.nbtgroups; p++) {
1234
0
    if (global.tune.options & GTUNE_NO_TG_FD_SHARING) {
1235
0
      ha_tgroup_ctx[p].fdtab = (void *)((char *)(void *)_fdtab + p * global.maxsock * sizeof(*fdtab));
1236
0
    } else
1237
0
      ha_tgroup_ctx[p].fdtab = _fdtab;
1238
0
  }
1239
  /*
1240
   * Immediately set the fdtab for our thread, as we'll need it soon
1241
   */
1242
0
  fdtab = ha_tgroup_ctx[0].fdtab;
1243
0
  vma_set_name(_fdtab, global.maxsock * sizeof(*fdtab) * nbfdtab, "fd", "fdtab");
1244
1245
0
  if ((_polled_mask = calloc(global.maxsock, nbfdtab * sizeof(*polled_mask))) == NULL) {
1246
0
    ha_alert("Not enough memory to allocate %d entries for polled_mask!\n", global.maxsock);
1247
0
    goto fail_polledmask;
1248
0
  }
1249
1250
  /*
1251
   * As for fdtab, get on polled_mask per thread group, if needed.
1252
   */
1253
0
  for (p = 0; p < global.nbtgroups; p++) {
1254
0
    if (global.tune.options & GTUNE_NO_TG_FD_SHARING)
1255
0
      ha_tgroup_ctx[p].polled_mask = _polled_mask + p * global.maxsock;
1256
0
    else
1257
0
      ha_tgroup_ctx[p].polled_mask = _polled_mask;
1258
0
  }
1259
  /* immediately set it for the boot thread as well */
1260
0
  polled_mask = ha_tgroup_ctx[0].polled_mask;
1261
0
  vma_set_name(_polled_mask, global.maxsock * sizeof(*polled_mask) * nbfdtab, "fd", "polled_mask");
1262
1263
0
  for (p = 0; p < MAX_TGROUPS; p++)
1264
0
    update_list[p].first = update_list[p].last = -1;
1265
1266
0
  for (p = 0; p < global.maxsock; p++) {
1267
    /* Mark the fd as out of the fd cache */
1268
0
    fdtab[p].update.next = -3;
1269
0
  }
1270
1271
0
  do {
1272
0
    bp = NULL;
1273
0
redo:
1274
0
    for (p = 0; p < nbpollers; p++)
1275
0
      if (!bp || (pollers[p].pref > bp->pref))
1276
0
        bp = &pollers[p];
1277
1278
0
    if (!bp || bp->pref == 0)
1279
0
      break;
1280
1281
0
    if ((global.tune.options & GTUNE_NO_TG_FD_SHARING) &&
1282
0
        !(bp->flags & HAP_POLL_F_NO_FD_SHARING)) {
1283
0
      ha_alert("Can't use poller '%s' with per-thread-group FD tables\n", bp->name);
1284
0
      bp->pref = 0;
1285
0
      goto redo;
1286
0
    }
1287
1288
0
    if (bp->init(bp)) {
1289
0
      memcpy(&cur_poller, bp, sizeof(*bp));
1290
0
      return 1;
1291
0
    }
1292
0
  } while (!bp || bp->pref == 0);
1293
1294
0
 fail_info:
1295
0
  free(_polled_mask);
1296
0
  _polled_mask = polled_mask = NULL;
1297
0
 fail_polledmask:
1298
0
  ha_aligned_free(_fdtab);
1299
0
  _fdtab = fdtab = NULL;
1300
0
 fail_tab:
1301
0
  return 0;
1302
0
}
1303
1304
/*
1305
 * Deinitialize the pollers.
1306
 */
1307
0
void deinit_pollers() {
1308
1309
0
  struct poller *bp;
1310
0
  int p;
1311
1312
0
  for (p = 0; p < nbpollers; p++) {
1313
0
    bp = &pollers[p];
1314
1315
0
    if (bp && bp->pref)
1316
0
      bp->term(bp);
1317
0
  }
1318
1319
  /* free the backing areas, not the (possibly per-group) thread-local
1320
   * pointers which may not point to the start of the allocations.
1321
   */
1322
0
  ha_aligned_free(_fdtab);
1323
0
  _fdtab = fdtab = NULL;
1324
0
  ha_free(&_polled_mask);
1325
0
  polled_mask = NULL;
1326
0
}
1327
1328
/*
1329
 * Lists the known pollers on <out>.
1330
 * Should be performed only before initialization.
1331
 */
1332
int list_pollers(FILE *out)
1333
0
{
1334
0
  int p;
1335
0
  int last, next;
1336
0
  int usable;
1337
0
  struct poller *bp;
1338
1339
0
  fprintf(out, "Available polling systems :\n");
1340
1341
0
  usable = 0;
1342
0
  bp = NULL;
1343
0
  last = next = -1;
1344
0
  while (1) {
1345
0
    for (p = 0; p < nbpollers; p++) {
1346
0
      if ((next < 0 || pollers[p].pref > next)
1347
0
          && (last < 0 || pollers[p].pref < last)) {
1348
0
        next = pollers[p].pref;
1349
0
        if (!bp || (pollers[p].pref > bp->pref))
1350
0
          bp = &pollers[p];
1351
0
      }
1352
0
    }
1353
1354
0
    if (next == -1)
1355
0
      break;
1356
1357
0
    for (p = 0; p < nbpollers; p++) {
1358
0
      if (pollers[p].pref == next) {
1359
0
        fprintf(out, " %10s : ", pollers[p].name);
1360
0
        if (pollers[p].pref == 0)
1361
0
          fprintf(out, "disabled, ");
1362
0
        else
1363
0
          fprintf(out, "pref=%3d, ", pollers[p].pref);
1364
0
        if (pollers[p].test(&pollers[p])) {
1365
0
          fprintf(out, " test result OK");
1366
0
          if (next > 0)
1367
0
            usable++;
1368
0
        } else {
1369
0
          fprintf(out, " test result FAILED");
1370
0
          if (bp == &pollers[p])
1371
0
            bp = NULL;
1372
0
        }
1373
0
        fprintf(out, "\n");
1374
0
      }
1375
0
    }
1376
0
    last = next;
1377
0
    next = -1;
1378
0
  };
1379
0
  fprintf(out, "Total: %d (%d usable), will use %s.\n", nbpollers, usable, bp ? bp->name : "none");
1380
0
  return 0;
1381
0
}
1382
1383
/*
1384
 * Some pollers may lose their connection after a fork(). It may be necessary
1385
 * to create initialize part of them again. Returns 0 in case of failure,
1386
 * otherwise 1. The fork() function may be NULL if unused. In case of error,
1387
 * the the current poller is destroyed and the caller is responsible for trying
1388
 * another one by calling init_pollers() again.
1389
 */
1390
int fork_poller()
1391
0
{
1392
0
  int fd;
1393
0
  for (fd = 0; fd < fd_highest; fd++) {
1394
0
    if (fdtab[fd].owner) {
1395
0
      HA_ATOMIC_OR(&fdtab[fd].state, FD_CLONED);
1396
0
    }
1397
0
  }
1398
1399
0
  if (cur_poller.fork) {
1400
0
    if (cur_poller.fork(&cur_poller))
1401
0
      return 1;
1402
0
    cur_poller.term(&cur_poller);
1403
0
    return 0;
1404
0
  }
1405
0
  return 1;
1406
0
}
1407
1408
/* config parser for global "tune.fd.edge-triggered", accepts "on" or "off" */
1409
static int cfg_parse_tune_fd_edge_triggered(char **args, int section_type, struct proxy *curpx,
1410
                                      const struct proxy *defpx, const char *file, int line,
1411
                                      char **err)
1412
0
{
1413
0
  if (too_many_args(1, args, err, NULL))
1414
0
    return -1;
1415
1416
0
  if (strcmp(args[1], "on") == 0)
1417
0
    global.tune.options |= GTUNE_FD_ET;
1418
0
  else if (strcmp(args[1], "off") == 0)
1419
0
    global.tune.options &= ~GTUNE_FD_ET;
1420
0
  else {
1421
0
    memprintf(err, "'%s' expects either 'on' or 'off' but got '%s'.", args[0], args[1]);
1422
0
    return -1;
1423
0
  }
1424
0
  return 0;
1425
0
}
1426
1427
/* config keyword parsers */
1428
static struct cfg_kw_list cfg_kws = {ILH, {
1429
  { CFG_GLOBAL, "tune.fd.edge-triggered", cfg_parse_tune_fd_edge_triggered, KWF_EXPERIMENTAL },
1430
  { 0, NULL, NULL }
1431
}};
1432
1433
INITCALL1(STG_REGISTER, cfg_register_keywords, &cfg_kws);
1434
1435
REGISTER_PER_THREAD_ALLOC(alloc_pollers_per_thread);
1436
REGISTER_PER_THREAD_INIT(init_pollers_per_thread);
1437
REGISTER_PER_THREAD_DEINIT(deinit_pollers_per_thread);
1438
REGISTER_PER_THREAD_FREE(free_pollers_per_thread);
1439
1440
/*
1441
 * Local variables:
1442
 *  c-indent-level: 8
1443
 *  c-basic-offset: 8
1444
 * End:
1445
 */