Skip to content

Commit 90891b4

Browse files
addaleaxsantigimeno
addaleax
authored and
santigimeno
committed
unix,win: limit concurrent DNS calls to nthreads/2
If `nthreads / 2` (rounded up) DNS calls are outstanding, queue more work of that kind instead of letting it take over more positions in the thread pool, blocking other work such as the (usually much faster) file system I/O or user-scheduled work. Fixes: nodejs/node#8436 PR-URL: #1845 Reviewed-By: Ben Noordhuis <[email protected]> Reviewed-By: Santiago Gimeno <[email protected]>
1 parent 69c43d9 commit 90891b4

8 files changed

Lines changed: 94 additions & 15 deletions

File tree

‎src/threadpool.c‎

Lines changed: 73 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -33,12 +33,18 @@ static uv_once_t once = UV_ONCE_INIT;
3333
static uv_cond_t cond;
3434
static uv_mutex_t mutex;
3535
static unsigned int idle_threads;
36+
static unsigned int slow_io_work_running;
3637
static unsigned int nthreads;
3738
static uv_thread_t* threads;
3839
static uv_thread_t default_threads[4];
3940
static QUEUE exit_message;
4041
static QUEUE wq;
42+
static QUEUE run_slow_work_message;
43+
static QUEUE slow_io_pending_wq;
4144

45+
static unsigned int slow_work_thread_threshold(void) {
46+
return (nthreads + 1) / 2;
47+
}
4248

4349
static void uv__cancelled(struct uv__work* w) {
4450
abort();
@@ -51,38 +57,73 @@ static void uv__cancelled(struct uv__work* w) {
5157
static void worker(void* arg) {
5258
struct uv__work* w;
5359
QUEUE* q;
60+
int is_slow_work;
5461

5562
uv_sem_post((uv_sem_t*) arg);
5663
arg = NULL;
5764

5865
for (;;) {
5966
uv_mutex_lock(&mutex);
6067

61-
while (QUEUE_EMPTY(&wq)) {
68+
wait_for_work:
69+
/* Keep waiting while either no work is present or only slow I/O
70+
and we're at the threshold for that. */
71+
while (QUEUE_EMPTY(&wq) ||
72+
(QUEUE_HEAD(&wq) == &run_slow_work_message &&
73+
QUEUE_NEXT(&run_slow_work_message) == &wq &&
74+
slow_io_work_running >= slow_work_thread_threshold())) {
6275
idle_threads += 1;
6376
uv_cond_wait(&cond, &mutex);
6477
idle_threads -= 1;
6578
}
6679

6780
q = QUEUE_HEAD(&wq);
68-
69-
if (q == &exit_message)
81+
if (q == &exit_message) {
7082
uv_cond_signal(&cond);
71-
else {
83+
uv_mutex_unlock(&mutex);
84+
break;
85+
}
86+
87+
QUEUE_REMOVE(q);
88+
QUEUE_INIT(q); /* Signal uv_cancel() that the work req is executing. */
89+
90+
is_slow_work = 0;
91+
if (q == &run_slow_work_message) {
92+
/* If we're at the slow I/O threshold, re-schedule until after all
93+
other work in the queue is done. */
94+
if (slow_io_work_running >= slow_work_thread_threshold()) {
95+
QUEUE_INSERT_TAIL(&wq, q);
96+
goto wait_for_work;
97+
}
98+
99+
/* If we encountered a request to run slow I/O work but there is none
100+
to run, that means it's cancelled => Start over. */
101+
if (QUEUE_EMPTY(&slow_io_pending_wq))
102+
goto wait_for_work;
103+
104+
is_slow_work = 1;
105+
slow_io_work_running++;
106+
107+
q = QUEUE_HEAD(&slow_io_pending_wq);
72108
QUEUE_REMOVE(q);
73-
QUEUE_INIT(q); /* Signal uv_cancel() that the work req is
74-
executing. */
109+
QUEUE_INIT(q);
110+
111+
/* If there is more slow I/O work, schedule it to be run as well. */
112+
if (!QUEUE_EMPTY(&slow_io_pending_wq)) {
113+
QUEUE_INSERT_TAIL(&wq, &run_slow_work_message);
114+
if (idle_threads > 0)
115+
uv_cond_signal(&cond);
116+
}
75117
}
76118

77119
uv_mutex_unlock(&mutex);
78120

79-
if (q == &exit_message)
80-
break;
81-
82121
w = QUEUE_DATA(q, struct uv__work, wq);
83122
w->work(w);
84123

85124
uv_mutex_lock(&w->loop->wq_mutex);
125+
if (is_slow_work)
126+
slow_io_work_running--;
86127
w->work = NULL; /* Signal uv_cancel() that the work req is done
87128
executing. */
88129
QUEUE_INSERT_TAIL(&w->loop->wq, &w->wq);
@@ -92,8 +133,20 @@ static void worker(void* arg) {
92133
}
93134

94135

95-
static void post(QUEUE* q) {
136+
static void post(QUEUE* q, enum uv__work_kind kind) {
96137
uv_mutex_lock(&mutex);
138+
if (kind == UV__WORK_SLOW_IO) {
139+
/* Insert into a separate queue. */
140+
QUEUE_INSERT_TAIL(&slow_io_pending_wq, q);
141+
if (!QUEUE_EMPTY(&run_slow_work_message)) {
142+
/* Running slow I/O tasks is already scheduled => Nothing to do here.
143+
The worker that runs said other task will schedule this one as well. */
144+
uv_mutex_unlock(&mutex);
145+
return;
146+
}
147+
q = &run_slow_work_message;
148+
}
149+
97150
QUEUE_INSERT_TAIL(&wq, q);
98151
if (idle_threads > 0)
99152
uv_cond_signal(&cond);
@@ -108,7 +161,7 @@ UV_DESTRUCTOR(static void cleanup(void)) {
108161
if (nthreads == 0)
109162
return;
110163

111-
post(&exit_message);
164+
post(&exit_message, UV__WORK_CPU);
112165

113166
for (i = 0; i < nthreads; i++)
114167
if (uv_thread_join(threads + i))
@@ -156,6 +209,8 @@ static void init_threads(void) {
156209
abort();
157210

158211
QUEUE_INIT(&wq);
212+
QUEUE_INIT(&slow_io_pending_wq);
213+
QUEUE_INIT(&run_slow_work_message);
159214

160215
if (uv_sem_init(&sem, 0))
161216
abort();
@@ -194,13 +249,14 @@ static void init_once(void) {
194249

195250
void uv__work_submit(uv_loop_t* loop,
196251
struct uv__work* w,
252+
enum uv__work_kind kind,
197253
void (*work)(struct uv__work* w),
198254
void (*done)(struct uv__work* w, int status)) {
199255
uv_once(&once, init_once);
200256
w->loop = loop;
201257
w->work = work;
202258
w->done = done;
203-
post(&w->wq);
259+
post(&w->wq, kind);
204260
}
205261

206262

@@ -284,7 +340,11 @@ int uv_queue_work(uv_loop_t* loop,
284340
req->loop = loop;
285341
req->work_cb = work_cb;
286342
req->after_work_cb = after_work_cb;
287-
uv__work_submit(loop, &req->work_req, uv__queue_work, uv__queue_done);
343+
uv__work_submit(loop,
344+
&req->work_req,
345+
UV__WORK_CPU,
346+
uv__queue_work,
347+
uv__queue_done);
288348
return 0;
289349
}
290350

‎src/unix/fs.c‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -120,7 +120,11 @@
120120
do { \
121121
if (cb != NULL) { \
122122
uv__req_register(loop, req); \
123-
uv__work_submit(loop, &req->work_req, uv__fs_work, uv__fs_done); \
123+
uv__work_submit(loop, \
124+
&req->work_req, \
125+
UV__WORK_FAST_IO, \
126+
uv__fs_work, \
127+
uv__fs_done); \
124128
return 0; \
125129
} \
126130
else { \

‎src/unix/getaddrinfo.c‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -186,6 +186,7 @@ int uv_getaddrinfo(uv_loop_t* loop,
186186
if (cb) {
187187
uv__work_submit(loop,
188188
&req->work_req,
189+
UV__WORK_SLOW_IO,
189190
uv__getaddrinfo_work,
190191
uv__getaddrinfo_done);
191192
return 0;

‎src/unix/getnameinfo.c‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,7 @@ int uv_getnameinfo(uv_loop_t* loop,
109109
if (getnameinfo_cb) {
110110
uv__work_submit(loop,
111111
&req->work_req,
112+
UV__WORK_SLOW_IO,
112113
uv__getnameinfo_work,
113114
uv__getnameinfo_done);
114115
return 0;

‎src/uv-common.h‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -164,8 +164,15 @@ void uv__fs_poll_close(uv_fs_poll_t* handle);
164164

165165
int uv__getaddrinfo_translate_error(int sys_err); /* EAI_* error. */
166166

167+
enum uv__work_kind {
168+
UV__WORK_CPU,
169+
UV__WORK_FAST_IO,
170+
UV__WORK_SLOW_IO
171+
};
172+
167173
void uv__work_submit(uv_loop_t* loop,
168174
struct uv__work *w,
175+
enum uv__work_kind kind,
169176
void (*work)(struct uv__work *w),
170177
void (*done)(struct uv__work *w, int status));
171178

‎src/win/fs.c‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,11 @@
5555
do { \
5656
if (cb != NULL) { \
5757
uv__req_register(loop, req); \
58-
uv__work_submit(loop, &req->work_req, uv__fs_work, uv__fs_done); \
58+
uv__work_submit(loop, \
59+
&req->work_req, \
60+
UV__WORK_FAST_IO, \
61+
uv__fs_work, \
62+
uv__fs_done); \
5963
return 0; \
6064
} else { \
6165
uv__fs_work(&req->work_req); \

‎src/win/getaddrinfo.c‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -368,6 +368,7 @@ int uv_getaddrinfo(uv_loop_t* loop,
368368
if (getaddrinfo_cb) {
369369
uv__work_submit(loop,
370370
&req->work_req,
371+
UV__WORK_SLOW_IO,
371372
uv__getaddrinfo_work,
372373
uv__getaddrinfo_done);
373374
return 0;

‎src/win/getnameinfo.c‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -145,6 +145,7 @@ int uv_getnameinfo(uv_loop_t* loop,
145145
if (getnameinfo_cb) {
146146
uv__work_submit(loop,
147147
&req->work_req,
148+
UV__WORK_SLOW_IO,
148149
uv__getnameinfo_work,
149150
uv__getnameinfo_done);
150151
return 0;

0 commit comments

Comments
 (0)