Skip to content

Commit f1b6765

Browse files
committed
src: improve thread safety of TaskQueue
1 parent 5077ea4 commit f1b6765

File tree

2 files changed

+132
-103
lines changed

2 files changed

+132
-103
lines changed

src/node_platform.cc

Lines changed: 111 additions & 95 deletions
Original file line numberDiff line numberDiff line change
@@ -40,9 +40,10 @@ static void PlatformWorkerThread(void* data) {
4040
worker_data->platform_workers_ready->Signal(lock);
4141
}
4242

43-
while (std::unique_ptr<Task> task = pending_worker_tasks->BlockingPop()) {
43+
while (std::unique_ptr<Task> task =
44+
pending_worker_tasks->Lock().BlockingPop()) {
4445
task->Run();
45-
pending_worker_tasks->NotifyOfCompletion();
46+
pending_worker_tasks->Lock().NotifyOfCompletion();
4647
}
4748
}
4849

@@ -73,13 +74,15 @@ class WorkerThreadsTaskRunner::DelayedTaskScheduler {
7374
}
7475

7576
void PostDelayedTask(std::unique_ptr<Task> task, double delay_in_seconds) {
76-
tasks_.Push(std::make_unique<ScheduleTask>(this, std::move(task),
77-
delay_in_seconds));
77+
auto locked = tasks_.Lock();
78+
locked.Push(std::make_unique<ScheduleTask>(
79+
this, std::move(task), delay_in_seconds));
7880
uv_async_send(&flush_tasks_);
7981
}
8082

8183
void Stop() {
82-
tasks_.Push(std::make_unique<StopTask>(this));
84+
auto locked = tasks_.Lock();
85+
locked.Push(std::make_unique<StopTask>(this));
8386
uv_async_send(&flush_tasks_);
8487
}
8588

@@ -100,8 +103,19 @@ class WorkerThreadsTaskRunner::DelayedTaskScheduler {
100103
static void FlushTasks(uv_async_t* flush_tasks) {
101104
DelayedTaskScheduler* scheduler =
102105
ContainerOf(&DelayedTaskScheduler::loop_, flush_tasks->loop);
103-
while (std::unique_ptr<Task> task = scheduler->tasks_.Pop())
106+
107+
std::vector<std::unique_ptr<Task>> tasks_to_run;
108+
{
109+
auto locked = scheduler->tasks_.Lock();
110+
std::unique_ptr<Task> task;
111+
while ((task = locked.Pop())) {
112+
tasks_to_run.push_back(std::move(task));
113+
}
114+
}
115+
116+
for (auto& task : tasks_to_run) {
104117
task->Run();
118+
}
105119
}
106120

107121
class StopTask : public Task {
@@ -149,7 +163,8 @@ class WorkerThreadsTaskRunner::DelayedTaskScheduler {
149163
static void RunTask(uv_timer_t* timer) {
150164
DelayedTaskScheduler* scheduler =
151165
ContainerOf(&DelayedTaskScheduler::loop_, timer->loop);
152-
scheduler->pending_worker_tasks_->Push(scheduler->TakeTimerTask(timer));
166+
scheduler->pending_worker_tasks_->Lock().Push(
167+
scheduler->TakeTimerTask(timer));
153168
}
154169

155170
std::unique_ptr<Task> TakeTimerTask(uv_timer_t* timer) {
@@ -203,7 +218,7 @@ WorkerThreadsTaskRunner::WorkerThreadsTaskRunner(int thread_pool_size) {
203218
}
204219

205220
void WorkerThreadsTaskRunner::PostTask(std::unique_ptr<Task> task) {
206-
pending_worker_tasks_.Push(std::move(task));
221+
pending_worker_tasks_.Lock().Push(std::move(task));
207222
}
208223

209224
void WorkerThreadsTaskRunner::PostDelayedTask(std::unique_ptr<Task> task,
@@ -212,11 +227,11 @@ void WorkerThreadsTaskRunner::PostDelayedTask(std::unique_ptr<Task> task,
212227
}
213228

214229
void WorkerThreadsTaskRunner::BlockingDrain() {
215-
pending_worker_tasks_.BlockingDrain();
230+
pending_worker_tasks_.Lock().BlockingDrain();
216231
}
217232

218233
void WorkerThreadsTaskRunner::Shutdown() {
219-
pending_worker_tasks_.Stop();
234+
pending_worker_tasks_.Lock().Stop();
220235
delayed_task_scheduler_->Stop();
221236
for (size_t i = 0; i < threads_.size(); i++) {
222237
CHECK_EQ(0, uv_thread_join(threads_[i].get()));
@@ -253,29 +268,23 @@ void PerIsolatePlatformData::PostIdleTaskImpl(
253268

254269
void PerIsolatePlatformData::PostTaskImpl(std::unique_ptr<Task> task,
255270
const v8::SourceLocation& location) {
256-
if (flush_tasks_ == nullptr) {
257-
// V8 may post tasks during Isolate disposal. In that case, the only
258-
// sensible path forward is to discard the task.
259-
return;
260-
}
261-
foreground_tasks_.Push(std::move(task));
271+
auto locked = foreground_tasks_.Lock();
272+
if (flush_tasks_ == nullptr) return;
273+
locked.Push(std::move(task));
262274
uv_async_send(flush_tasks_);
263275
}
264276

265277
void PerIsolatePlatformData::PostDelayedTaskImpl(
266278
std::unique_ptr<Task> task,
267279
double delay_in_seconds,
268280
const v8::SourceLocation& location) {
269-
if (flush_tasks_ == nullptr) {
270-
// V8 may post tasks during Isolate disposal. In that case, the only
271-
// sensible path forward is to discard the task.
272-
return;
273-
}
281+
auto locked = foreground_delayed_tasks_.Lock();
282+
if (flush_tasks_ == nullptr) return;
274283
std::unique_ptr<DelayedTask> delayed(new DelayedTask());
275284
delayed->task = std::move(task);
276285
delayed->platform_data = shared_from_this();
277286
delayed->timeout = delay_in_seconds;
278-
foreground_delayed_tasks_.Push(std::move(delayed));
287+
locked.Push(std::move(delayed));
279288
uv_async_send(flush_tasks_);
280289
}
281290

@@ -301,32 +310,30 @@ void PerIsolatePlatformData::AddShutdownCallback(void (*callback)(void*),
301310
}
302311

303312
void PerIsolatePlatformData::Shutdown() {
304-
if (flush_tasks_ == nullptr)
305-
return;
313+
auto foreground_tasks_locked = foreground_tasks_.Lock();
314+
auto foreground_delayed_tasks_locked = foreground_delayed_tasks_.Lock();
306315

307-
// While there should be no V8 tasks in the queues at this point, it is
308-
// possible that Node.js-internal tasks from e.g. the inspector are still
309-
// lying around. We clear these queues and ignore the return value,
310-
// effectively deleting the tasks instead of running them.
311-
foreground_delayed_tasks_.PopAll();
312-
foreground_tasks_.PopAll();
316+
foreground_delayed_tasks_locked.PopAll();
317+
foreground_tasks_locked.PopAll();
313318
scheduled_delayed_tasks_.clear();
314319

315-
// Both destroying the scheduled_delayed_tasks_ lists and closing
316-
// flush_tasks_ handle add tasks to the event loop. We keep a count of all
317-
// non-closed handles, and when that reaches zero, we inform any shutdown
318-
// callbacks that the platform is done as far as this Isolate is concerned.
319-
self_reference_ = shared_from_this();
320-
uv_close(reinterpret_cast<uv_handle_t*>(flush_tasks_),
321-
[](uv_handle_t* handle) {
322-
std::unique_ptr<uv_async_t> flush_tasks {
323-
reinterpret_cast<uv_async_t*>(handle) };
324-
PerIsolatePlatformData* platform_data =
325-
static_cast<PerIsolatePlatformData*>(flush_tasks->data);
326-
platform_data->DecreaseHandleCount();
327-
platform_data->self_reference_.reset();
328-
});
329-
flush_tasks_ = nullptr;
320+
if (flush_tasks_ != nullptr) {
321+
// Both destroying the scheduled_delayed_tasks_ lists and closing
322+
// flush_tasks_ handle add tasks to the event loop. We keep a count of all
323+
// non-closed handles, and when that reaches zero, we inform any shutdown
324+
// callbacks that the platform is done as far as this Isolate is concerned.
325+
self_reference_ = shared_from_this();
326+
uv_close(reinterpret_cast<uv_handle_t*>(flush_tasks_),
327+
[](uv_handle_t* handle) {
328+
std::unique_ptr<uv_async_t> flush_tasks{
329+
reinterpret_cast<uv_async_t*>(handle)};
330+
PerIsolatePlatformData* platform_data =
331+
static_cast<PerIsolatePlatformData*>(flush_tasks->data);
332+
platform_data->DecreaseHandleCount();
333+
platform_data->self_reference_.reset();
334+
});
335+
flush_tasks_ = nullptr;
336+
}
330337
}
331338

332339
void PerIsolatePlatformData::DecreaseHandleCount() {
@@ -472,39 +479,51 @@ void NodePlatform::DrainTasks(Isolate* isolate) {
472479
bool PerIsolatePlatformData::FlushForegroundTasksInternal() {
473480
bool did_work = false;
474481

475-
while (std::unique_ptr<DelayedTask> delayed =
476-
foreground_delayed_tasks_.Pop()) {
477-
did_work = true;
482+
std::vector<std::unique_ptr<DelayedTask>> delayed_tasks_to_schedule;
483+
{
484+
auto locked_tasks = foreground_delayed_tasks_.Lock();
485+
std::unique_ptr<DelayedTask> delayed;
486+
while ((delayed = locked_tasks.Pop())) {
487+
did_work = true;
488+
delayed_tasks_to_schedule.push_back(std::move(delayed));
489+
}
490+
}
491+
492+
for (auto& delayed : delayed_tasks_to_schedule) {
478493
uint64_t delay_millis = llround(delayed->timeout * 1000);
479494

480495
delayed->timer.data = static_cast<void*>(delayed.get());
481496
uv_timer_init(loop_, &delayed->timer);
482-
// Timers may not guarantee queue ordering of events with the same delay if
483-
// the delay is non-zero. This should not be a problem in practice.
497+
// Timers may not guarantee queue ordering of events with the same delay
498+
// if the delay is non-zero. This should not be a problem in practice.
484499
uv_timer_start(&delayed->timer, RunForegroundTask, delay_millis, 0);
485500
uv_unref(reinterpret_cast<uv_handle_t*>(&delayed->timer));
486501
uv_handle_count_++;
487502

488-
scheduled_delayed_tasks_.emplace_back(delayed.release(),
489-
[](DelayedTask* delayed) {
490-
uv_close(reinterpret_cast<uv_handle_t*>(&delayed->timer),
491-
[](uv_handle_t* handle) {
492-
std::unique_ptr<DelayedTask> task {
493-
static_cast<DelayedTask*>(handle->data) };
494-
task->platform_data->DecreaseHandleCount();
495-
});
496-
});
503+
scheduled_delayed_tasks_.emplace_back(
504+
delayed.release(), [](DelayedTask* delayed) {
505+
uv_close(reinterpret_cast<uv_handle_t*>(&delayed->timer),
506+
[](uv_handle_t* handle) {
507+
std::unique_ptr<DelayedTask> task{
508+
static_cast<DelayedTask*>(handle->data)};
509+
task->platform_data->DecreaseHandleCount();
510+
});
511+
});
512+
}
513+
514+
std::queue<std::unique_ptr<Task>> tasks;
515+
{
516+
auto locked = foreground_tasks_.Lock();
517+
tasks = locked.PopAll();
497518
}
498-
// Move all foreground tasks into a separate queue and flush that queue.
499-
// This way tasks that are posted while flushing the queue will be run on the
500-
// next call of FlushForegroundTasksInternal.
501-
std::queue<std::unique_ptr<Task>> tasks = foreground_tasks_.PopAll();
519+
502520
while (!tasks.empty()) {
503521
std::unique_ptr<Task> task = std::move(tasks.front());
504522
tasks.pop();
505523
did_work = true;
506524
RunForegroundTask(std::move(task));
507525
}
526+
508527
return did_work;
509528
}
510529

@@ -594,66 +613,63 @@ TaskQueue<T>::TaskQueue()
594613
outstanding_tasks_(0), stopped_(false), task_queue_() { }
595614

596615
template <class T>
597-
void TaskQueue<T>::Push(std::unique_ptr<T> task) {
598-
Mutex::ScopedLock scoped_lock(lock_);
599-
outstanding_tasks_++;
600-
task_queue_.push(std::move(task));
601-
tasks_available_.Signal(scoped_lock);
616+
TaskQueue<T>::Locked::Locked(TaskQueue* queue)
617+
: queue_(queue), lock_(queue->lock_) {}
618+
619+
template <class T>
620+
void TaskQueue<T>::Locked::Push(std::unique_ptr<T> task) {
621+
queue_->outstanding_tasks_++;
622+
queue_->task_queue_.push(std::move(task));
623+
queue_->tasks_available_.Signal(lock_);
602624
}
603625

604626
template <class T>
605-
std::unique_ptr<T> TaskQueue<T>::Pop() {
606-
Mutex::ScopedLock scoped_lock(lock_);
607-
if (task_queue_.empty()) {
627+
std::unique_ptr<T> TaskQueue<T>::Locked::Pop() {
628+
if (queue_->task_queue_.empty()) {
608629
return std::unique_ptr<T>(nullptr);
609630
}
610-
std::unique_ptr<T> result = std::move(task_queue_.front());
611-
task_queue_.pop();
631+
std::unique_ptr<T> result = std::move(queue_->task_queue_.front());
632+
queue_->task_queue_.pop();
612633
return result;
613634
}
614635

615636
template <class T>
616-
std::unique_ptr<T> TaskQueue<T>::BlockingPop() {
617-
Mutex::ScopedLock scoped_lock(lock_);
618-
while (task_queue_.empty() && !stopped_) {
619-
tasks_available_.Wait(scoped_lock);
637+
std::unique_ptr<T> TaskQueue<T>::Locked::BlockingPop() {
638+
while (queue_->task_queue_.empty() && !queue_->stopped_) {
639+
queue_->tasks_available_.Wait(lock_);
620640
}
621-
if (stopped_) {
641+
if (queue_->stopped_) {
622642
return std::unique_ptr<T>(nullptr);
623643
}
624-
std::unique_ptr<T> result = std::move(task_queue_.front());
625-
task_queue_.pop();
644+
std::unique_ptr<T> result = std::move(queue_->task_queue_.front());
645+
queue_->task_queue_.pop();
626646
return result;
627647
}
628648

629649
template <class T>
630-
void TaskQueue<T>::NotifyOfCompletion() {
631-
Mutex::ScopedLock scoped_lock(lock_);
632-
if (--outstanding_tasks_ == 0) {
633-
tasks_drained_.Broadcast(scoped_lock);
650+
void TaskQueue<T>::Locked::NotifyOfCompletion() {
651+
if (--queue_->outstanding_tasks_ == 0) {
652+
queue_->tasks_drained_.Broadcast(lock_);
634653
}
635654
}
636655

637656
template <class T>
638-
void TaskQueue<T>::BlockingDrain() {
639-
Mutex::ScopedLock scoped_lock(lock_);
640-
while (outstanding_tasks_ > 0) {
641-
tasks_drained_.Wait(scoped_lock);
657+
void TaskQueue<T>::Locked::BlockingDrain() {
658+
while (queue_->outstanding_tasks_ > 0) {
659+
queue_->tasks_drained_.Wait(lock_);
642660
}
643661
}
644662

645663
template <class T>
646-
void TaskQueue<T>::Stop() {
647-
Mutex::ScopedLock scoped_lock(lock_);
648-
stopped_ = true;
649-
tasks_available_.Broadcast(scoped_lock);
664+
void TaskQueue<T>::Locked::Stop() {
665+
queue_->stopped_ = true;
666+
queue_->tasks_available_.Broadcast(lock_);
650667
}
651668

652669
template <class T>
653-
std::queue<std::unique_ptr<T>> TaskQueue<T>::PopAll() {
654-
Mutex::ScopedLock scoped_lock(lock_);
670+
std::queue<std::unique_ptr<T>> TaskQueue<T>::Locked::PopAll() {
655671
std::queue<std::unique_ptr<T>> result;
656-
result.swap(task_queue_);
672+
result.swap(queue_->task_queue_);
657673
return result;
658674
}
659675

src/node_platform.h

Lines changed: 21 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -22,16 +22,28 @@ class PerIsolatePlatformData;
2222
template <class T>
2323
class TaskQueue {
2424
public:
25+
class Locked {
26+
public:
27+
void Push(std::unique_ptr<T> task);
28+
std::unique_ptr<T> Pop();
29+
std::unique_ptr<T> BlockingPop();
30+
void NotifyOfCompletion();
31+
void BlockingDrain();
32+
void Stop();
33+
std::queue<std::unique_ptr<T>> PopAll();
34+
35+
private:
36+
friend class TaskQueue;
37+
explicit Locked(TaskQueue* queue);
38+
39+
TaskQueue* queue_;
40+
Mutex::ScopedLock lock_;
41+
};
42+
2543
TaskQueue();
2644
~TaskQueue() = default;
2745

28-
void Push(std::unique_ptr<T> task);
29-
std::unique_ptr<T> Pop();
30-
std::unique_ptr<T> BlockingPop();
31-
std::queue<std::unique_ptr<T>> PopAll();
32-
void NotifyOfCompletion();
33-
void BlockingDrain();
34-
void Stop();
46+
Locked Lock() { return Locked(this); }
3547

3648
private:
3749
Mutex lock_;
@@ -98,6 +110,8 @@ class PerIsolatePlatformData
98110
void RunForegroundTask(std::unique_ptr<v8::Task> task);
99111
static void RunForegroundTask(uv_timer_t* timer);
100112

113+
uv_async_t* flush_tasks_ = nullptr;
114+
101115
struct ShutdownCallback {
102116
void (*cb)(void*);
103117
void* data;
@@ -110,7 +124,6 @@ class PerIsolatePlatformData
110124

111125
v8::Isolate* const isolate_;
112126
uv_loop_t* const loop_;
113-
uv_async_t* flush_tasks_ = nullptr;
114127
TaskQueue<v8::Task> foreground_tasks_;
115128
TaskQueue<DelayedTask> foreground_delayed_tasks_;
116129

0 commit comments

Comments
 (0)