Skip to content

Commit b38ca0d

Browse files
boazsegevboazsegev
committed
improved signalling logic + IO reactor shutdown signal (un)blocking
1 parent 8f7a48a commit b38ca0d

File tree

9 files changed

+182
-72
lines changed

9 files changed

+182
-72
lines changed

‎fio-stl.h‎

Lines changed: 88 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -7517,7 +7517,8 @@ Signal Monitoring API
75177517
*/
75187518
SFUNC int fio_signal_monitor(int sig,
75197519
void (*callback)(int sig, void *),
7520-
void *udata);
7520+
void *udata,
7521+
bool propagate);
75217522

75227523
/** Reviews all signals, calling any relevant callbacks. */
75237524
SFUNC int fio_signal_review(void);
@@ -7549,7 +7550,8 @@ POSIX implementation
75497550

75507551
static struct {
75517552
int32_t sig;
7552-
volatile unsigned flag;
7553+
uint16_t propagate;
7554+
volatile uint16_t flag;
75537555
void (*callback)(int sig, void *);
75547556
void *udata;
75557557
struct sigaction old;
@@ -7564,7 +7566,8 @@ FIO_SFUNC void fio___signal_catcher(int sig) {
75647566
/* mark flag */
75657567
fio___signal_watchers[i].flag = 1;
75667568
/* pass-through if exists */
7567-
if (fio___signal_watchers[i].old.sa_handler != SIG_IGN &&
7569+
if (fio___signal_watchers[i].propagate &&
7570+
fio___signal_watchers[i].old.sa_handler != SIG_IGN &&
75687571
fio___signal_watchers[i].old.sa_handler != SIG_DFL)
75697572
fio___signal_watchers[i].old.sa_handler(sig);
75707573
return;
@@ -7576,14 +7579,16 @@ FIO_SFUNC void fio___signal_catcher(int sig) {
75767579
*/
75777580
SFUNC int fio_signal_monitor(int sig,
75787581
void (*callback)(int sig, void *),
7579-
void *udata) {
7582+
void *udata,
7583+
bool propagate) {
75807584
if (!sig)
75817585
return -1;
75827586
for (size_t i = 0; i < FIO_SIGNAL_MONITOR_MAX; ++i) {
75837587
/* updating an existing monitor */
75847588
if (fio___signal_watchers[i].sig == sig) {
75857589
fio___signal_watchers[i].callback = callback;
75867590
fio___signal_watchers[i].udata = udata;
7591+
fio___signal_watchers[i].propagate = propagate;
75877592
return 0;
75887593
}
75897594
/* slot busy */
@@ -7596,6 +7601,7 @@ SFUNC int fio_signal_monitor(int sig,
75967601
fio___signal_watchers[i].sig = sig;
75977602
fio___signal_watchers[i].callback = callback;
75987603
fio___signal_watchers[i].udata = udata;
7604+
fio___signal_watchers[i].propagate = propagate;
75997605
act.sa_handler = fio___signal_catcher;
76007606
sigemptyset(&act.sa_mask);
76017607
act.sa_flags = SA_RESTART | SA_NOCLDSTOP;
@@ -7604,6 +7610,7 @@ SFUNC int fio_signal_monitor(int sig,
76047610
fio___signal_watchers[i].callback = NULL;
76057611
fio___signal_watchers[i].udata = (void *)1;
76067612
fio___signal_watchers[i].sig = 0;
7613+
fio___signal_watchers[i].propagate = 0;
76077614
return -1;
76087615
}
76097616
return 0;
@@ -7615,22 +7622,26 @@ SFUNC int fio_signal_monitor(int sig,
76157622
SFUNC int fio_signal_forget(int sig) {
76167623
if (!sig)
76177624
return -1;
7625+
struct sigaction act = {0};
7626+
act.sa_handler = SIG_DFL;
76187627
for (size_t i = 0; i < FIO_SIGNAL_MONITOR_MAX; ++i) {
76197628
if (!fio___signal_watchers[i].sig && !fio___signal_watchers[i].udata)
7620-
return -1; /* initialized list is finishe */
7629+
break; /* initialized list is finished */
76217630
if (fio___signal_watchers[i].sig != sig)
76227631
continue;
76237632
fio___signal_watchers[i].callback = NULL;
76247633
fio___signal_watchers[i].udata = (void *)1;
76257634
fio___signal_watchers[i].sig = 0;
7626-
struct sigaction act;
7627-
memset(&act, 0, sizeof(act));
7628-
if (sigaction(sig, &fio___signal_watchers[i].old, &act)) {
7635+
fio___signal_watchers[i].propagate = 0;
7636+
struct sigaction old = fio___signal_watchers[i].old;
7637+
old = act;
7638+
if (sigaction(sig, &old, &act)) {
76297639
FIO_LOG_ERROR("couldn't unset signal handler: %s", strerror(errno));
76307640
return -1;
76317641
}
76327642
return 0;
76337643
}
7644+
sigaction(sig, &act, NULL);
76347645
return -1;
76357646
}
76367647

@@ -7656,7 +7667,7 @@ FIO_SFUNC void fio___signal_catcher(int sig) {
76567667
/* mark flag */
76577668
fio___signal_watchers[i].flag = 1;
76587669
/* pass-through if exists */
7659-
if (fio___signal_watchers[i].old &&
7670+
if (fio___signal_watchers[i].propagate && fio___signal_watchers[i].old &&
76607671
(intptr_t)fio___signal_watchers[i].old != (intptr_t)SIG_IGN &&
76617672
(intptr_t)fio___signal_watchers[i].old != (intptr_t)SIG_DFL) {
76627673
fio___signal_watchers[i].old(sig);
@@ -7673,7 +7684,8 @@ FIO_SFUNC void fio___signal_catcher(int sig) {
76737684
*/
76747685
SFUNC int fio_signal_monitor(int sig,
76757686
void (*callback)(int sig, void *),
7676-
void *udata) {
7687+
void *udata,
7688+
bool propagate) {
76777689
if (!sig)
76787690
return -1;
76797691
for (size_t i = 0; i < FIO_SIGNAL_MONITOR_MAX; ++i) {
@@ -7690,12 +7702,14 @@ SFUNC int fio_signal_monitor(int sig,
76907702
fio___signal_watchers[i].sig = sig;
76917703
fio___signal_watchers[i].callback = callback;
76927704
fio___signal_watchers[i].udata = udata;
7705+
fio___signal_watchers[i].propagate = propagate;
76937706
fio___signal_watchers[i].old = signal(sig, fio___signal_catcher);
76947707
if ((intptr_t)SIG_ERR == (intptr_t)fio___signal_watchers[i].old) {
76957708
fio___signal_watchers[i].sig = 0;
76967709
fio___signal_watchers[i].callback = NULL;
76977710
fio___signal_watchers[i].udata = (void *)1;
76987711
fio___signal_watchers[i].old = NULL;
7712+
fio___signal_watchers[i].propagate = 0;
76997713
FIO_LOG_ERROR("couldn't set signal handler: %s", strerror(errno));
77007714
return -1;
77017715
}
@@ -7708,26 +7722,32 @@ SFUNC int fio_signal_monitor(int sig,
77087722
SFUNC int fio_signal_forget(int sig) {
77097723
if (!sig)
77107724
return -1;
7711-
for (size_t i = 0; i < FIO_SIGNAL_MONITOR_MAX; ++i) {
7725+
size_t i = 0;
7726+
for (; i < FIO_SIGNAL_MONITOR_MAX; ++i) {
77127727
if (!fio___signal_watchers[i].sig && !fio___signal_watchers[i].udata)
77137728
return -1; /* initialized list is finished */
77147729
if (fio___signal_watchers[i].sig != sig)
77157730
continue;
77167731
fio___signal_watchers[i].callback = NULL;
77177732
fio___signal_watchers[i].udata = (void *)1;
77187733
fio___signal_watchers[i].sig = 0;
7719-
if (fio___signal_watchers[i].old) {
7734+
if (fio___signal_watchers[i].old.sa_handler &&
7735+
fio___signal_watchers[i].old.sa_handler != SIG_DFL) {
77207736
if ((intptr_t)signal(sig, fio___signal_watchers[i].old) ==
77217737
(intptr_t)SIG_ERR)
77227738
goto sig_error;
77237739
} else {
77247740
if ((intptr_t)signal(sig, SIG_DFL) == (intptr_t)SIG_ERR)
77257741
goto sig_error;
77267742
}
7743+
fio___signal_watchers[i].old.sa_handler = SIG_DFL;
77277744
return 0;
77287745
}
7746+
signal(sig, SIG_DFL);
77297747
return -1;
77307748
sig_error:
7749+
fio___signal_watchers[i].old.sa_handler = SIG_DFL;
7750+
signal(sig, SIG_DFL);
77317751
FIO_LOG_ERROR("couldn't unset signal handler: %s", strerror(errno));
77327752
return -1;
77337753
}
@@ -7789,7 +7809,7 @@ FIO_SFUNC void FIO_NAME_TEST(stl, signal)(void) {
77897809
size_t e = 0;
77907810
fprintf(stderr, "* testing signal monitoring (setup / cleanup only).\n");
77917811
for (size_t i = 0; i < sizeof(t) / sizeof(t[0]); ++i) {
7792-
if (fio_signal_monitor(t[i].sig, NULL, NULL)) {
7812+
if (fio_signal_monitor(t[i].sig, NULL, NULL, 1)) {
77937813
FIO_LOG_ERROR("couldn't set signal monitoring for %s (%d)",
77947814
t[i].name,
77957815
t[i].sig);
@@ -34494,6 +34514,12 @@ SFUNC void fio_io_restart(int workers);
3449434514
/** Sets a signal to listen to for a hot restart (see `fio_io_restart`). */
3449534515
SFUNC void fio_io_restart_on_signal(int signal);
3449634516

34517+
/** Returns the shutdown timeout for the reactor. */
34518+
SFUNC size_t fio_io_shutdown_timsout(void);
34519+
34520+
/** Sets the shutdown timeout for the reactor, returning the new value. */
34521+
SFUNC size_t fio_io_shutdown_timsout_set(size_t milliseconds);
34522+
3449734523
/* *****************************************************************************
3449834524
The IO Reactor's State
3449934525
***************************************************************************** */
@@ -35525,11 +35551,13 @@ static struct FIO___IO_S {
3552535551
uint32_t to_spawn;
3552635552
fio_io_s *wakeup;
3552735553
FIO___LOCK_TYPE lock;
35554+
size_t shutdown_timeout;
3552835555
} FIO___IO = {
3552935556
.tick = 0,
3553035557
.wakeup_fd = -1,
3553135558
.stop = 1,
3553235559
.lock = FIO___LOCK_INIT,
35560+
.shutdown_timeout = FIO_IO_SHUTDOWN_TIMEOUT,
3553335561
};
3553435562

3553535563
/** Stopping the IO reactor. */
@@ -35584,6 +35612,14 @@ SFUNC void fio_io_run_every FIO_NOOP(fio_timer_schedule_args_s args) {
3558435612
/** Returns a pointer for the IO reactor's queue. */
3558535613
SFUNC fio_queue_s *fio_io_queue(void) { return &FIO___IO.queue; }
3558635614

35615+
/** Returns the shutdown timeout for the reactor. */
35616+
SFUNC size_t fio_io_shutdown_timsout(void) { return FIO___IO.shutdown_timeout; }
35617+
35618+
/** Sets the shutdown timeout for the reactor, returning the new value. */
35619+
SFUNC size_t fio_io_shutdown_timsout_set(size_t milliseconds) {
35620+
return (FIO___IO.shutdown_timeout = milliseconds);
35621+
}
35622+
3558735623
/* *****************************************************************************
3558835624
IO Type
3558935625
***************************************************************************** */
@@ -36839,18 +36875,19 @@ FIO_SFUNC void fio___io_after_fork(void *ignr_) {
3683936875
}
3684036876
fio_queue_perform_all(&FIO___IO.queue);
3684136877
fio_queue_destroy(&FIO___IO.queue);
36878+
FIO___IO.pids = FIO_LIST_INIT(FIO___IO.pids);
3684236879
}
3684336880

3684436881
FIO_SFUNC void fio___io_cleanup_at_exit(void *ignr_) {
36882+
#ifdef SIGKILL
36883+
FIO_LIST_EACH(fio___io_pid_s, node, &FIO___IO.pids, w) {
36884+
fio_thread_kill(w->pid, SIGKILL);
36885+
}
36886+
#endif /* SIGKILL */
36887+
FIO___LOCK_DESTROY(FIO___IO.lock);
3684536888
fio___io_after_fork(ignr_);
3684636889
fio_poll_destroy(&FIO___IO.poll);
3684736890
fio___io_env_safe_destroy(&FIO___IO.env);
36848-
#if FIO_VALIDITY_MAP_USE
36849-
fio_validity_map_destroy(&FIO___IO.valid);
36850-
#if FIO_VALIDATE_IO_MUTEX
36851-
fio_thread_mutex_destroy(&FIO___IO.valid_lock);
36852-
#endif
36853-
#endif /* FIO_VALIDATE_IO_MUTEX / FIO_VALIDITY_MAP_USE */
3685436891
FIO___IO.tick = FIO___IO_GET_TIME_MILLI();
3685536892
fio_queue_perform_all(&FIO___IO.queue);
3685636893
fio_timer_destroy(&FIO___IO.timer);
@@ -36866,6 +36903,7 @@ FIO_CONSTRUCTOR(fio___io) {
3686636903
FIO___IO.tick = FIO___IO_GET_TIME_MILLI();
3686736904
FIO___IO.root_pid = FIO___IO.pid = fio_thread_getpid();
3686836905
FIO___IO.async = FIO_LIST_INIT(FIO___IO.async);
36906+
FIO___IO.pids = FIO_LIST_INIT(FIO___IO.pids);
3686936907
fio___io_init_protocol(&FIO___IO_MOCK_PROTOCOL, 0);
3687036908
fio_poll_init(&FIO___IO.poll,
3687136909
.on_data = fio___io_poll_on_data_schd,
@@ -36901,9 +36939,24 @@ Copyright and License: see header file (000 copyright.h) or top of file
3690136939
The IO Reactor Cycle (the actual work)
3690236940
***************************************************************************** */
3690336941

36942+
static void fio___io_signal_crash(int sig, void *flg) {
36943+
FIO_LOG_FATAL("(%d) additional stop signal(!) - should crash.", FIO___IO.pid);
36944+
fio_signal_forget(sig);
36945+
/* cannot lock, signal may be received during critical section */
36946+
FIO_LIST_EACH(fio___io_pid_s, node, &FIO___IO.pids, pos) {
36947+
if (!pos->done)
36948+
fio_thread_kill(pos->pid, SIGKILL);
36949+
}
36950+
fio_thread_kill(FIO___IO.root_pid, SIGKILL);
36951+
fio_thread_kill(FIO___IO.pid, SIGKILL);
36952+
exit(-1);
36953+
(void)sig, (void)flg;
36954+
}
36955+
3690436956
static void fio___io_signal_stop(int sig, void *flg) {
3690536957
FIO_LOG_INFO("(%d) stop signal detected.", FIO___IO.pid);
3690636958
fio_io_stop();
36959+
fio_signal_monitor(sig, fio___io_signal_crash, flg, 0);
3690736960
(void)sig, (void)flg;
3690836961
}
3690936962

@@ -36954,9 +37007,9 @@ FIO_SFUNC void fio___io_run_async_as_sync(void *ignr_1, void *ignr_2) {
3695437007
}
3695537008

3695637009
FIO_SFUNC void fio___io_shutdown_task(void *shutdown_start_, void *a2) {
36957-
intptr_t shutdown_start = (intptr_t)shutdown_start_;
36958-
if (shutdown_start + FIO_IO_SHUTDOWN_TIMEOUT < FIO___IO.tick ||
36959-
FIO_LIST_IS_EMPTY(&FIO___IO.protocols))
37010+
intptr_t shutdown_start =
37011+
(intptr_t)shutdown_start_ + FIO___IO.shutdown_timeout;
37012+
if (shutdown_start < FIO___IO.tick || FIO_LIST_IS_EMPTY(&FIO___IO.protocols))
3696037013
return;
3696137014
fio___io_tick(fio_queue_count(&FIO___IO.queue) ? 0 : 100);
3696237015
fio_queue_push(&FIO___IO.queue, fio___io_run_async_as_sync);
@@ -37042,11 +37095,10 @@ FIO_SFUNC void fio___io_work(int is_worker) {
3704237095
}
3704337096
/* signal all child workers to terminate, parent is going away. */
3704437097
FIO___LOCK_LOCK(FIO___IO.lock);
37045-
if (FIO___IO.pids.next)
37046-
FIO_LIST_EACH(fio___io_pid_s, node, &FIO___IO.pids, pos) {
37047-
if (!pos->done)
37048-
fio_thread_kill(pos->pid, SIGTERM);
37049-
}
37098+
FIO_LIST_EACH(fio___io_pid_s, node, &FIO___IO.pids, pos) {
37099+
if (!pos->done)
37100+
fio_thread_kill(pos->pid, SIGTERM);
37101+
}
3705037102
FIO___LOCK_UNLOCK(FIO___IO.lock);
3705137103

3705237104
fio_queue_perform_all(&FIO___IO.queue);
@@ -37077,8 +37129,6 @@ static void *fio___io_worker_sentinel(void *pid_data) {
3707737129
(void *)thr);
3707837130

3707937131
FIO___LOCK_LOCK(FIO___IO.lock);
37080-
if (!FIO___IO.pids.next)
37081-
FIO___IO.pids = FIO_LIST_INIT(FIO___IO.pids);
3708237132
FIO_LIST_PUSH(&FIO___IO.pids, &sentinal.node);
3708337133
FIO___LOCK_UNLOCK(FIO___IO.lock);
3708437134

@@ -37240,13 +37290,16 @@ SFUNC void fio_io_start(int workers) {
3724037290

3724137291
fio_state_callback_force(FIO_CALL_PRE_START);
3724237292
fio_queue_perform_all(&FIO___IO.queue);
37243-
fio_signal_monitor(SIGINT, fio___io_signal_stop, NULL);
37244-
fio_signal_monitor(SIGTERM, fio___io_signal_stop, NULL);
37293+
fio_signal_monitor(SIGINT, fio___io_signal_stop, NULL, 0);
37294+
fio_signal_monitor(SIGTERM, fio___io_signal_stop, NULL, 0);
3724537295
if (FIO___IO.restart_signal)
37246-
fio_signal_monitor(FIO___IO.restart_signal, fio___io_signal_restart, NULL);
37296+
fio_signal_monitor(FIO___IO.restart_signal,
37297+
fio___io_signal_restart,
37298+
NULL,
37299+
0);
3724737300

3724837301
#ifdef SIGPIPE
37249-
fio_signal_monitor(SIGPIPE, NULL, NULL);
37302+
fio_signal_monitor(SIGPIPE, NULL, NULL, 0);
3725037303
#endif
3725137304
FIO___IO.tick = FIO___IO_GET_TIME_MILLI();
3725237305
if (workers) {
@@ -38241,7 +38294,7 @@ FIO_CONSTRUCTOR(fio___openssl_setup_default) {
3824138294
FIO___OPENSSL_IO_FUNCS = fio_openssl_io_functions();
3824238295
fio_io_tls_default_functions(&FIO___OPENSSL_IO_FUNCS);
3824338296
#ifdef SIGPIPE
38244-
fio_signal_monitor(SIGPIPE, NULL, NULL); /* avoid OpenSSL issue... */
38297+
fio_signal_monitor(SIGPIPE, NULL, NULL, 0); /* avoid OpenSSL issue... */
3824538298
#endif
3824638299
}
3824738300

@@ -46946,7 +46999,7 @@ FIO_SFUNC void fio___http_controller_http1_on_finish_task(void *c_,
4694646999
if (c->h || !fio_io_is_open(c->io))
4694747000
goto something_is_wrong;
4694847001
c->h = (fio_http_s *)upgraded;
46949-
{
47002+
{ /* TODO! test if safe to move to user thread for callback execution? */
4695047003
const size_t pr_i = fio_http_is_websocket(c->h) ? FIO___HTTP_PROTOCOL_WS
4695147004
: FIO___HTTP_PROTOCOL_SSE;
4695247005
fio_http_controller_set(

‎fio-stl.md‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3087,13 +3087,15 @@ When defining `FIO_SIGNAL`, the following function are defined.
30873087
#### `fio_signal_monitor`
30883088

30893089
```c
3090-
int fio_signal_monitor(int sig, void (*callback)(int sig, void *), void *udata);
3090+
int fio_signal_monitor(int sig, void (*callback)(int sig, void *), void *udata, bool propagate);
30913091
```
30923092

30933093
Starts to monitor for the specified signal, setting an optional callback.
30943094

30953095
If the signal is already being monitored, the callback and `udata` pointers are updated.
30963096

3097+
If `propagate` is true and a previous signal handler was set, it will be called.
3098+
30973099
**Note**: `udata` stands for "user data", it is an opaque pointer that is simply passed along to the callback.
30983100

30993101
#### `fio_signal_review`

0 commit comments

Comments
 (0)