diff --git a/Makefile b/Makefile index 34817430b..b048dee28 100644 --- a/Makefile +++ b/Makefile @@ -378,6 +378,7 @@ hrpc_calc_server: prepare hrpc_calc_protoc unittest: prepare libhv + $(CC) -g -Wall -O0 -std=c99 -I. -Ibase -Issl -Ievent -o bin/hloop_timer_fairness_test unittest/hloop_timer_fairness_test.c -Llib -lhv -pthread $(CC) -g -Wall -O0 -std=c99 -I. -Ibase -o bin/rbtree_test unittest/rbtree_test.c base/rbtree.c $(CC) -g -Wall -O0 -std=c99 -I. -Ibase -o bin/hbase_test unittest/hbase_test.c base/hbase.c $(CC) -g -Wall -O0 -std=c99 -I. -Ibase -o bin/mkdir_p unittest/mkdir_test.c base/hbase.c diff --git a/event/hloop.c b/event/hloop.c index b12802435..700b04a3b 100644 --- a/event/hloop.c +++ b/event/hloop.c @@ -155,9 +155,14 @@ int hloop_process_events(hloop_t* loop, int timeout_ms) { int64_t min_timeout = TIMER_ENTRY(loop->realtimers.root)->next_timeout - hloop_now_us(loop); blocktime_us = MIN(blocktime_us, min_timeout); } - if (blocktime_us < 0) goto process_timers; - blocktime_ms = blocktime_us / 1000 + 1; - blocktime_ms = MIN(blocktime_ms, timeout_ms); + if (blocktime_us <= 0) { + // Poll ready I/O even when a timer is already due. + blocktime_ms = 0; + } + else { + blocktime_ms = blocktime_us / 1000 + 1; + blocktime_ms = MIN(blocktime_ms, timeout_ms); + } } if (loop->nios) { @@ -171,7 +176,6 @@ int hloop_process_events(hloop_t* loop, int timeout_ms) { return 0; } -process_timers: if (loop->ntimers) { ntimers = hloop_process_timers(loop); } diff --git a/scripts/unittest.sh b/scripts/unittest.sh index ca1046e8d..90d0e17fd 100755 --- a/scripts/unittest.sh +++ b/scripts/unittest.sh @@ -6,6 +6,8 @@ export DYLD_LIBRARY_PATH=${ROOT_DIR}/lib:${DYLD_LIBRARY_PATH} export LD_LIBRARY_PATH=${ROOT_DIR}/lib:${LD_LIBRARY_PATH} cd ${ROOT_DIR} +bin/hloop_timer_fairness_test || exit $? + bin/rbtree_test bin/hbase_test bin/date diff --git a/unittest/CMakeLists.txt b/unittest/CMakeLists.txt index 6e17408b3..f5b4f1726 100644 --- a/unittest/CMakeLists.txt +++ b/unittest/CMakeLists.txt @@ -171,6 +171,17 @@ target_link_libraries(epoll_init_test ${HV_LIBRARIES}) set(EVENT_UNITTEST_TARGETS epoll_init_test) endif() +# ------event: overdue timer fairness------ +add_executable(hloop_timer_fairness_test hloop_timer_fairness_test.c) +target_include_directories(hloop_timer_fairness_test PRIVATE .. ../base ../ssl ../event) +target_link_libraries(hloop_timer_fairness_test ${HV_LIBRARIES}) +if(CMAKE_SYSTEM_NAME STREQUAL "Linux" AND NOT BUILD_SHARED AND CMAKE_C_COMPILER_ID MATCHES "GNU|Clang") + target_compile_definitions(hloop_timer_fairness_test PRIVATE HLOOP_TEST_WRAP) + set_property(TARGET hloop_timer_fairness_test APPEND_STRING PROPERTY LINK_FLAGS + " -Wl,--wrap=iowatcher_poll_events,--wrap=gethrtime_us,--wrap=usleep") +endif() +list(APPEND EVENT_UNITTEST_TARGETS hloop_timer_fairness_test) + # ------event: async dns------ add_executable(hdns_test hdns_test.c) target_include_directories(hdns_test PRIVATE .. ../base ../ssl ../event) diff --git a/unittest/hloop_timer_fairness_test.c b/unittest/hloop_timer_fairness_test.c new file mode 100644 index 000000000..c6d460b31 --- /dev/null +++ b/unittest/hloop_timer_fairness_test.c @@ -0,0 +1,210 @@ +#include "hevent.h" +#include "hsocket.h" + +/* Keep checks active in Release builds too. */ +static void require(int condition, const char *message) { + if (!condition) { + fprintf(stderr, "hloop_timer_fairness_test: %s\n", message); + exit(EXIT_FAILURE); + } +} + +typedef struct { + int turn; + int timers; + int reads; + int posts; +} progress_t; + +static void on_timer(htimer_t *timer) { + progress_t *progress = (progress_t *)hevent_userdata(timer); + ++progress->timers; +} + +static void on_read(hio_t *io, void *buf, int size) { + progress_t *progress = (progress_t *)hevent_userdata(io); + require(size == 1 && *(char *)buf == 'x', "unexpected socket data"); + require(progress->timers == progress->turn, "timer priority changed relative to I/O"); + ++progress->reads; +} + +static void on_post(hevent_t *event) { + progress_t *progress = (progress_t *)hevent_userdata(event); + require(progress->timers == progress->turn, "timer priority changed relative to posted callbacks"); + ++progress->posts; +} + +static void set_deadline(htimer_t *timer, uint64_t deadline) { + struct heap *timers = timer->event_type == HEVENT_TYPE_TIMEOUT ? &timer->loop->timers : &timer->loop->realtimers; + heap_remove(timers, &timer->node); + timer->next_timeout = deadline; + heap_insert(timers, &timer->node); +} + +/* Socketpair uses loopback TCP on Windows. Establish actual readiness before + * forcing the timer overdue, without making any libhv event pending by hand. */ +static void wait_readable(int fd) { + fd_set readfds; + struct timeval timeout; +#ifndef OS_WIN + require(fd >= 0 && fd < FD_SETSIZE, "descriptor outside select range"); +#endif + FD_ZERO(&readfds); + FD_SET(fd, &readfds); + timeout.tv_sec = 2; + timeout.tv_usec = 0; + require(select(fd + 1, &readfds, NULL, NULL, &timeout) == 1, "descriptor did not become readable"); +} + +static void test_progress(int realtime) { + enum { TURNS = 8 }; + int sockets[2]; + progress_t progress = {0}; + hevent_t event; + hloop_t *loop = hloop_new(HLOOP_FLAG_RUN_ONCE); + require(loop != NULL, "hloop_new failed"); + require(Socketpair(AF_INET, SOCK_STREAM, 0, sockets) == 0, "Socketpair failed"); + hio_t *io = hread(loop, sockets[0], NULL, 0, on_read); + require(io != NULL, "hread failed"); + hevent_set_userdata(io, &progress); + htimer_t *timer = realtime ? htimer_add_period(loop, on_timer, -1, -1, -1, -1, -1, INFINITE) : htimer_add(loop, on_timer, 1, INFINITE); + require(timer != NULL, "timer creation failed"); + hevent_set_userdata(timer, &progress); + /* Period timers normally share the wakeup descriptor's priority. Use an + * explicit priority so discovery order cannot decide the expected order. */ + hevent_set_priority(timer, HEVENT_HIGHEST_PRIORITY); + + for (progress.turn = 1; progress.turn <= TURNS; ++progress.turn) { + require(send(sockets[1], "x", 1, 0) == 1, "socket write failed"); + memset(&event, 0, sizeof(event)); + event.cb = on_post; + event.userdata = &progress; + hloop_post_event(loop, &event); + wait_readable(sockets[0]); + wait_readable(loop->eventfds[0]); + + /* Model a timer that is late on every turn, without timing-sensitive + * sleeps or changing its positive repeat interval. */ + hloop_update_time(loop); + set_deadline(timer, (realtime ? hloop_now_us(loop) : loop->cur_hrtime) - 1); + require(hloop_run(loop) == 0, "hloop_run failed"); + require(progress.timers == progress.turn, "repeating timer did not fire once per turn"); + require(timer->repeat == INFINITE && !timer->destroy, "repeating timer was removed"); + } + + /* Check before removing the timer or allowing a non-overdue turn. The old + * implementation runs all eight timers but never discovers either reader. */ + printf("%s timer progress: timers=%d reads=%d posts=%d\n", realtime ? "realtime" : "monotonic", progress.timers, progress.reads, progress.posts); + require(progress.reads == TURNS, "overdue timers starved ready socket reads"); + require(progress.posts == TURNS, "overdue timers starved posted callbacks"); + htimer_del(timer); + hio_close(io); + closesocket(sockets[1]); + hloop_free(&loop); + puts(realtime ? "realtime timer progress passed" : "monotonic timer progress passed"); +} + +#ifdef HLOOP_TEST_WRAP +/* Static Linux builds also observe the actual backend timeout. A ready socket + * alone would hide an accidental positive wait. Linker wrappers leave the + * real-backend tests above unchanged and require no production test hooks. */ +extern int hloop_process_events(hloop_t *loop, int timeout_ms); +extern int __real_iowatcher_poll_events(hloop_t *loop, int timeout_ms); +extern unsigned long long __real_gethrtime_us(void); +extern int __real_usleep(unsigned int usec); + +static int probing; +static int poll_calls; +static int poll_timeout; +static int sleep_calls; +static unsigned int sleep_timeout; +static int stop_in_poll; +static unsigned long long clock_us; + +unsigned long long __wrap_gethrtime_us(void) { + return probing ? clock_us : __real_gethrtime_us(); +} + +int __wrap_iowatcher_poll_events(hloop_t *loop, int timeout_ms) { + if (!probing) { + return __real_iowatcher_poll_events(loop, timeout_ms); + } + ++poll_calls; + poll_timeout = timeout_ms; + ++clock_us; + if (stop_in_poll) { + loop->status = HLOOP_STATUS_STOP; + } + return 0; +} + +int __wrap_usleep(unsigned int usec) { + if (!probing) { + return __real_usleep(usec); + } + ++sleep_calls; + sleep_timeout = usec; + ++clock_us; + return 0; +} + +static void test_timeout(int offset_us, int caller_timeout, int expected_timeout, int with_io, int stop) { + progress_t progress = {0}; + clock_us = __real_gethrtime_us(); + probing = 1; + poll_calls = sleep_calls = 0; + poll_timeout = -1; + sleep_timeout = (unsigned int)-1; + stop_in_poll = stop; + hloop_t *loop = hloop_new(0); + require(loop != NULL, "probe loop creation failed"); + htimer_t *timer = htimer_add(loop, on_timer, 1, INFINITE); + require(timer != NULL, "probe timer creation failed"); + hevent_set_userdata(timer, &progress); + set_deadline(timer, clock_us + offset_us); + if (with_io) { + /* Register the real wakeup descriptor, but intercept only its poll. */ + hloop_wakeup(loop); + require(loop->nios > 0, "probe has no registered descriptor"); + } + loop->status = HLOOP_STATUS_RUNNING; + hloop_process_events(loop, caller_timeout); + if (with_io) { + require(poll_calls == 1 && sleep_calls == 0, "timer bypassed backend polling"); + require(poll_timeout == expected_timeout, "incorrect backend timeout"); + } + else { + require(poll_calls == 0 && sleep_calls == 1, "incorrect no-I/O wait path"); + require(sleep_timeout == (unsigned int)expected_timeout * 1000, "incorrect no-I/O sleep duration"); + } + require(loop->cur_hrtime == clock_us, "loop time was not refreshed after polling"); + require(progress.timers == (!stop && offset_us <= 1), "unexpected timer callback dispatch"); + hloop_free(&loop); + probing = 0; +} + +static void test_timeouts(void) { + test_timeout(-1, 100, 0, 1, 0); + test_timeout(0, 100, 0, 1, 0); + test_timeout(1, 100, 1, 1, 0); + test_timeout(999, 100, 1, 1, 0); + test_timeout(1000, 100, 2, 1, 0); + test_timeout(1001, 100, 2, 1, 0); + test_timeout(200000, 100, 100, 1, 0); + test_timeout(1000, 0, 0, 1, 0); + test_timeout(-1, 100, 0, 0, 0); + test_timeout(0, 100, 0, 0, 0); + test_timeout(-1, 100, 0, 1, 1); + puts("poll timeout, no-I/O wait, time refresh and STOP checks passed"); +} +#endif + +int main(void) { + test_progress(0); + test_progress(1); +#ifdef HLOOP_TEST_WRAP + test_timeouts(); +#endif + puts("hloop_timer_fairness_test passed"); + return 0; +}