GCC Code Coverage Report
Directory: ./ Exec Total Coverage
File: node_platform.cc Lines: 357 372 96.0 %
Date: 2022-10-14 05:16:24 Branches: 94 122 77.0 %

Line Branch Exec Source
1
#include "node_platform.h"
2
#include "node_internals.h"
3
4
#include "env-inl.h"
5
#include "debug_utils-inl.h"
6
#include <algorithm>  // find_if(), find(), move()
7
#include <cmath>  // llround()
8
#include <memory>  // unique_ptr(), shared_ptr(), make_shared()
9
10
namespace node {
11
12
using v8::Isolate;
13
using v8::Object;
14
using v8::Platform;
15
using v8::Task;
16
17
namespace {
18
19
struct PlatformWorkerData {
20
  TaskQueue<Task>* task_queue;
21
  Mutex* platform_workers_mutex;
22
  ConditionVariable* platform_workers_ready;
23
  int* pending_platform_workers;
24
  int id;
25
};
26
27
22270
static void PlatformWorkerThread(void* data) {
28
  std::unique_ptr<PlatformWorkerData>
29
44504
      worker_data(static_cast<PlatformWorkerData*>(data));
30
31
22270
  TaskQueue<Task>* pending_worker_tasks = worker_data->task_queue;
32

44540
  TRACE_EVENT_METADATA1("__metadata", "thread_name", "name",
33
                        "PlatformWorkerThread");
34
35
  // Notify the main thread that the platform worker is ready.
36
  {
37
44540
    Mutex::ScopedLock lock(*worker_data->platform_workers_mutex);
38
22270
    (*worker_data->pending_platform_workers)--;
39
22270
    worker_data->platform_workers_ready->Signal(lock);
40
  }
41
42
213166
  while (std::unique_ptr<Task> task = pending_worker_tasks->BlockingPop()) {
43
95448
    task->Run();
44
95448
    pending_worker_tasks->NotifyOfCompletion();
45
95448
  }
46
22234
}
47
48
}  // namespace
49
50
5557
class WorkerThreadsTaskRunner::DelayedTaskScheduler {
51
 public:
52
5566
  explicit DelayedTaskScheduler(TaskQueue<Task>* tasks)
53
5566
    : pending_worker_tasks_(tasks) {}
54
55
5566
  std::unique_ptr<uv_thread_t> Start() {
56
16698
    auto start_thread = [](void* data) {
57
5566
      static_cast<DelayedTaskScheduler*>(data)->Run();
58
16689
    };
59
5566
    std::unique_ptr<uv_thread_t> t { new uv_thread_t() };
60
5566
    uv_sem_init(&ready_, 0);
61
5566
    CHECK_EQ(0, uv_thread_create(t.get(), start_thread, this));
62
5566
    uv_sem_wait(&ready_);
63
5566
    uv_sem_destroy(&ready_);
64
5566
    return t;
65
  }
66
67
526
  void PostDelayedTask(std::unique_ptr<Task> task, double delay_in_seconds) {
68
526
    tasks_.Push(std::make_unique<ScheduleTask>(this, std::move(task),
69
                                               delay_in_seconds));
70
526
    uv_async_send(&flush_tasks_);
71
526
  }
72
73
5557
  void Stop() {
74
5557
    tasks_.Push(std::make_unique<StopTask>(this));
75
5557
    uv_async_send(&flush_tasks_);
76
5557
  }
77
78
 private:
79
5566
  void Run() {
80

11132
    TRACE_EVENT_METADATA1("__metadata", "thread_name", "name",
81
                          "WorkerThreadsTaskRunner::DelayedTaskScheduler");
82
5566
    loop_.data = this;
83
5566
    CHECK_EQ(0, uv_loop_init(&loop_));
84
5566
    flush_tasks_.data = this;
85
5566
    CHECK_EQ(0, uv_async_init(&loop_, &flush_tasks_, FlushTasks));
86
5566
    uv_sem_post(&ready_);
87
88
5566
    uv_run(&loop_, UV_RUN_DEFAULT);
89
5557
    CheckedUvLoopClose(&loop_);
90
5557
  }
91
92
6083
  static void FlushTasks(uv_async_t* flush_tasks) {
93
    DelayedTaskScheduler* scheduler =
94
6083
        ContainerOf(&DelayedTaskScheduler::loop_, flush_tasks->loop);
95
18249
    while (std::unique_ptr<Task> task = scheduler->tasks_.Pop())
96
12166
      task->Run();
97
6083
  }
98
99
11114
  class StopTask : public Task {
100
   public:
101
5557
    explicit StopTask(DelayedTaskScheduler* scheduler): scheduler_(scheduler) {}
102
103
5557
    void Run() override {
104
11114
      std::vector<uv_timer_t*> timers;
105
5827
      for (uv_timer_t* timer : scheduler_->timers_)
106
270
        timers.push_back(timer);
107
5827
      for (uv_timer_t* timer : timers)
108
270
        scheduler_->TakeTimerTask(timer);
109
5557
      uv_close(reinterpret_cast<uv_handle_t*>(&scheduler_->flush_tasks_),
110
16671
               [](uv_handle_t* handle) {});
111
5557
    }
112
113
   private:
114
     DelayedTaskScheduler* scheduler_;
115
  };
116
117
1052
  class ScheduleTask : public Task {
118
   public:
119
526
    ScheduleTask(DelayedTaskScheduler* scheduler,
120
                 std::unique_ptr<Task> task,
121
                 double delay_in_seconds)
122
526
      : scheduler_(scheduler),
123
526
        task_(std::move(task)),
124
1052
        delay_in_seconds_(delay_in_seconds) {}
125
126
526
    void Run() override {
127
526
      uint64_t delay_millis = llround(delay_in_seconds_ * 1000);
128
1052
      std::unique_ptr<uv_timer_t> timer(new uv_timer_t());
129
526
      CHECK_EQ(0, uv_timer_init(&scheduler_->loop_, timer.get()));
130
526
      timer->data = task_.release();
131
526
      CHECK_EQ(0, uv_timer_start(timer.get(), RunTask, delay_millis, 0));
132
526
      scheduler_->timers_.insert(timer.release());
133
526
    }
134
135
   private:
136
    DelayedTaskScheduler* scheduler_;
137
    std::unique_ptr<Task> task_;
138
    double delay_in_seconds_;
139
  };
140
141
256
  static void RunTask(uv_timer_t* timer) {
142
    DelayedTaskScheduler* scheduler =
143
256
        ContainerOf(&DelayedTaskScheduler::loop_, timer->loop);
144
256
    scheduler->pending_worker_tasks_->Push(scheduler->TakeTimerTask(timer));
145
256
  }
146
147
526
  std::unique_ptr<Task> TakeTimerTask(uv_timer_t* timer) {
148
526
    std::unique_ptr<Task> task(static_cast<Task*>(timer->data));
149
526
    uv_timer_stop(timer);
150
2104
    uv_close(reinterpret_cast<uv_handle_t*>(timer), [](uv_handle_t* handle) {
151
526
      delete reinterpret_cast<uv_timer_t*>(handle);
152
1578
    });
153
526
    timers_.erase(timer);
154
526
    return task;
155
  }
156
157
  uv_sem_t ready_;
158
  TaskQueue<Task>* pending_worker_tasks_;
159
160
  TaskQueue<Task> tasks_;
161
  uv_loop_t loop_;
162
  uv_async_t flush_tasks_;
163
  std::unordered_set<uv_timer_t*> timers_;
164
};
165
166
5566
WorkerThreadsTaskRunner::WorkerThreadsTaskRunner(int thread_pool_size) {
167
11132
  Mutex platform_workers_mutex;
168
11132
  ConditionVariable platform_workers_ready;
169
170
11132
  Mutex::ScopedLock lock(platform_workers_mutex);
171
5566
  int pending_platform_workers = thread_pool_size;
172
173
11132
  delayed_task_scheduler_ = std::make_unique<DelayedTaskScheduler>(
174
16698
      &pending_worker_tasks_);
175
5566
  threads_.push_back(delayed_task_scheduler_->Start());
176
177
27836
  for (int i = 0; i < thread_pool_size; i++) {
178
    PlatformWorkerData* worker_data = new PlatformWorkerData{
179
22270
      &pending_worker_tasks_, &platform_workers_mutex,
180
      &platform_workers_ready, &pending_platform_workers, i
181
44540
    };
182
44540
    std::unique_ptr<uv_thread_t> t { new uv_thread_t() };
183
44540
    if (uv_thread_create(t.get(), PlatformWorkerThread,
184
22270
                         worker_data) != 0) {
185
      break;
186
    }
187
22270
    threads_.push_back(std::move(t));
188
  }
189
190
  // Wait for platform workers to initialize before continuing with the
191
  // bootstrap.
192
43204
  while (pending_platform_workers > 0) {
193
18819
    platform_workers_ready.Wait(lock);
194
  }
195
5566
}
196
197
95194
void WorkerThreadsTaskRunner::PostTask(std::unique_ptr<Task> task) {
198
95194
  pending_worker_tasks_.Push(std::move(task));
199
95194
}
200
201
526
void WorkerThreadsTaskRunner::PostDelayedTask(std::unique_ptr<Task> task,
202
                                              double delay_in_seconds) {
203
526
  delayed_task_scheduler_->PostDelayedTask(std::move(task), delay_in_seconds);
204
526
}
205
206
19305
void WorkerThreadsTaskRunner::BlockingDrain() {
207
19305
  pending_worker_tasks_.BlockingDrain();
208
19305
}
209
210
5557
void WorkerThreadsTaskRunner::Shutdown() {
211
5557
  pending_worker_tasks_.Stop();
212
5557
  delayed_task_scheduler_->Stop();
213
33348
  for (size_t i = 0; i < threads_.size(); i++) {
214
27791
    CHECK_EQ(0, uv_thread_join(threads_[i].get()));
215
  }
216
5557
}
217
218
47762
int WorkerThreadsTaskRunner::NumberOfWorkerThreads() const {
219
47762
  return threads_.size();
220
}
221
222
6360
PerIsolatePlatformData::PerIsolatePlatformData(
223
6360
    Isolate* isolate, uv_loop_t* loop)
224
6360
  : isolate_(isolate), loop_(loop) {
225
6360
  flush_tasks_ = new uv_async_t();
226
6360
  CHECK_EQ(0, uv_async_init(loop, flush_tasks_, FlushTasks));
227
6360
  flush_tasks_->data = static_cast<void*>(this);
228
6360
  uv_unref(reinterpret_cast<uv_handle_t*>(flush_tasks_));
229
6360
}
230
231
std::shared_ptr<v8::TaskRunner>
232
34407
PerIsolatePlatformData::GetForegroundTaskRunner() {
233
34407
  return shared_from_this();
234
}
235
236
9395
void PerIsolatePlatformData::FlushTasks(uv_async_t* handle) {
237
9395
  auto platform_data = static_cast<PerIsolatePlatformData*>(handle->data);
238
9395
  platform_data->FlushForegroundTasksInternal();
239
9395
}
240
241
void PerIsolatePlatformData::PostIdleTask(std::unique_ptr<v8::IdleTask> task) {
242
  UNREACHABLE();
243
}
244
245
15266
void PerIsolatePlatformData::PostTask(std::unique_ptr<Task> task) {
246
15266
  if (flush_tasks_ == nullptr) {
247
    // V8 may post tasks during Isolate disposal. In that case, the only
248
    // sensible path forward is to discard the task.
249
    return;
250
  }
251
15266
  foreground_tasks_.Push(std::move(task));
252
15266
  uv_async_send(flush_tasks_);
253
}
254
255
6318
void PerIsolatePlatformData::PostDelayedTask(
256
    std::unique_ptr<Task> task, double delay_in_seconds) {
257
6318
  if (flush_tasks_ == nullptr) {
258
    // V8 may post tasks during Isolate disposal. In that case, the only
259
    // sensible path forward is to discard the task.
260
    return;
261
  }
262
12636
  std::unique_ptr<DelayedTask> delayed(new DelayedTask());
263
6318
  delayed->task = std::move(task);
264
6318
  delayed->platform_data = shared_from_this();
265
6318
  delayed->timeout = delay_in_seconds;
266
6318
  foreground_delayed_tasks_.Push(std::move(delayed));
267
6318
  uv_async_send(flush_tasks_);
268
}
269
270
14379
void PerIsolatePlatformData::PostNonNestableTask(std::unique_ptr<Task> task) {
271
14379
  PostTask(std::move(task));
272
14379
}
273
274
void PerIsolatePlatformData::PostNonNestableDelayedTask(
275
    std::unique_ptr<Task> task,
276
    double delay_in_seconds) {
277
  PostDelayedTask(std::move(task), delay_in_seconds);
278
}
279
280
1608
PerIsolatePlatformData::~PerIsolatePlatformData() {
281
804
  CHECK(!flush_tasks_);
282
804
}
283
284
741
void PerIsolatePlatformData::AddShutdownCallback(void (*callback)(void*),
285
                                                 void* data) {
286
741
  shutdown_callbacks_.emplace_back(ShutdownCallback { callback, data });
287
741
}
288
289
5687
void PerIsolatePlatformData::Shutdown() {
290
5687
  if (flush_tasks_ == nullptr)
291
    return;
292
293
  // While there should be no V8 tasks in the queues at this point, it is
294
  // possible that Node.js-internal tasks from e.g. the inspector are still
295
  // lying around. We clear these queues and ignore the return value,
296
  // effectively deleting the tasks instead of running them.
297
5687
  foreground_delayed_tasks_.PopAll();
298
5687
  foreground_tasks_.PopAll();
299
5687
  scheduled_delayed_tasks_.clear();
300
301
  // Both destroying the scheduled_delayed_tasks_ lists and closing
302
  // flush_tasks_ handle add tasks to the event loop. We keep a count of all
303
  // non-closed handles, and when that reaches zero, we inform any shutdown
304
  // callbacks that the platform is done as far as this Isolate is concerned.
305
5687
  self_reference_ = shared_from_this();
306
5687
  uv_close(reinterpret_cast<uv_handle_t*>(flush_tasks_),
307
7295
           [](uv_handle_t* handle) {
308
    std::unique_ptr<uv_async_t> flush_tasks {
309
1608
        reinterpret_cast<uv_async_t*>(handle) };
310
    PerIsolatePlatformData* platform_data =
311
804
        static_cast<PerIsolatePlatformData*>(flush_tasks->data);
312
804
    platform_data->DecreaseHandleCount();
313
804
    platform_data->self_reference_.reset();
314
7295
  });
315
5687
  flush_tasks_ = nullptr;
316
}
317
318
1594
void PerIsolatePlatformData::DecreaseHandleCount() {
319
1594
  CHECK_GE(uv_handle_count_, 1);
320
1594
  if (--uv_handle_count_ == 0) {
321
1545
    for (const auto& callback : shutdown_callbacks_)
322
741
      callback.cb(callback.data);
323
  }
324
1594
}
325
326
5566
NodePlatform::NodePlatform(int thread_pool_size,
327
                           v8::TracingController* tracing_controller,
328
5566
                           v8::PageAllocator* page_allocator) {
329
5566
  if (tracing_controller != nullptr) {
330
5559
    tracing_controller_ = tracing_controller;
331
  } else {
332
7
    tracing_controller_ = new v8::TracingController();
333
  }
334
335
  // V8 will default to its built in allocator if none is provided.
336
5566
  page_allocator_ = page_allocator;
337
338
  // TODO(addaleax): It's a bit icky that we use global state here, but we can't
339
  // really do anything about it unless V8 starts exposing a way to access the
340
  // current v8::Platform instance.
341
5566
  SetTracingController(tracing_controller_);
342
  DCHECK_EQ(GetTracingController(), tracing_controller_);
343
5566
  worker_thread_task_runner_ =
344
11132
      std::make_shared<WorkerThreadsTaskRunner>(thread_pool_size);
345
5566
}
346
347
16671
NodePlatform::~NodePlatform() {
348
5557
  Shutdown();
349
11114
}
350
351
6359
void NodePlatform::RegisterIsolate(Isolate* isolate, uv_loop_t* loop) {
352
12718
  Mutex::ScopedLock lock(per_isolate_mutex_);
353
12718
  auto delegate = std::make_shared<PerIsolatePlatformData>(isolate, loop);
354
6359
  IsolatePlatformDelegate* ptr = delegate.get();
355
  auto insertion = per_isolate_.emplace(
356
    isolate,
357
6359
    std::make_pair(ptr, std::move(delegate)));
358
6359
  CHECK(insertion.second);
359
6359
}
360
361
1
void NodePlatform::RegisterIsolate(Isolate* isolate,
362
                                   IsolatePlatformDelegate* delegate) {
363
2
  Mutex::ScopedLock lock(per_isolate_mutex_);
364
  auto insertion = per_isolate_.emplace(
365
    isolate,
366
1
    std::make_pair(delegate, std::shared_ptr<PerIsolatePlatformData>{}));
367
1
  CHECK(insertion.second);
368
1
}
369
370
5687
void NodePlatform::UnregisterIsolate(Isolate* isolate) {
371
11374
  Mutex::ScopedLock lock(per_isolate_mutex_);
372
5687
  auto existing_it = per_isolate_.find(isolate);
373
5687
  CHECK_NE(existing_it, per_isolate_.end());
374
5687
  auto& existing = existing_it->second;
375
5687
  if (existing.second) {
376
5686
    existing.second->Shutdown();
377
  }
378
5687
  per_isolate_.erase(existing_it);
379
5687
}
380
381
741
void NodePlatform::AddIsolateFinishedCallback(Isolate* isolate,
382
                                              void (*cb)(void*), void* data) {
383
1482
  Mutex::ScopedLock lock(per_isolate_mutex_);
384
741
  auto it = per_isolate_.find(isolate);
385
741
  if (it == per_isolate_.end()) {
386
    cb(data);
387
    return;
388
  }
389
741
  CHECK(it->second.second);
390
741
  it->second.second->AddShutdownCallback(cb, data);
391
}
392
393
11109
void NodePlatform::Shutdown() {
394
11109
  if (has_shut_down_) return;
395
5557
  has_shut_down_ = true;
396
5557
  worker_thread_task_runner_->Shutdown();
397
398
  {
399
11114
    Mutex::ScopedLock lock(per_isolate_mutex_);
400
5557
    per_isolate_.clear();
401
  }
402
}
403
404
47762
int NodePlatform::NumberOfWorkerThreads() {
405
47762
  return worker_thread_task_runner_->NumberOfWorkerThreads();
406
}
407
408
14627
void PerIsolatePlatformData::RunForegroundTask(std::unique_ptr<Task> task) {
409
14627
  if (isolate_->IsExecutionTerminating()) return task->Run();
410
14625
  DebugSealHandleScope scope(isolate_);
411
14625
  Environment* env = Environment::GetCurrent(isolate_);
412
14625
  if (env != nullptr) {
413
18642
    v8::HandleScope scope(isolate_);
414
9325
    InternalCallbackScope cb_scope(env, Object::New(isolate_), { 0, 0 },
415
27975
                                   InternalCallbackScope::kNoFlags);
416
9325
    task->Run();
417
  } else {
418
    // The task is moved out of InternalCallbackScope if env is not available.
419
    // This is a required else block, and should not be removed.
420
    // See comment: https://github.com/nodejs/node/pull/34688#pullrequestreview-463867489
421
5300
    task->Run();
422
  }
423
}
424
425
21
void PerIsolatePlatformData::DeleteFromScheduledTasks(DelayedTask* task) {
426
  auto it = std::find_if(scheduled_delayed_tasks_.begin(),
427
                         scheduled_delayed_tasks_.end(),
428
42
                         [task](const DelayedTaskPointer& delayed) -> bool {
429
42
          return delayed.get() == task;
430
21
      });
431
21
  CHECK_NE(it, scheduled_delayed_tasks_.end());
432
21
  scheduled_delayed_tasks_.erase(it);
433
21
}
434
435
21
void PerIsolatePlatformData::RunForegroundTask(uv_timer_t* handle) {
436
21
  DelayedTask* delayed = ContainerOf(&DelayedTask::timer, handle);
437
21
  delayed->platform_data->RunForegroundTask(std::move(delayed->task));
438
21
  delayed->platform_data->DeleteFromScheduledTasks(delayed);
439
21
}
440
441
11112
void NodePlatform::DrainTasks(Isolate* isolate) {
442
22215
  std::shared_ptr<PerIsolatePlatformData> per_isolate = ForNodeIsolate(isolate);
443
11112
  if (!per_isolate) return;
444
445
8194
  do {
446
    // Worker tasks aren't associated with an Isolate.
447
19305
    worker_thread_task_runner_->BlockingDrain();
448
19305
  } while (per_isolate->FlushForegroundTasksInternal());
449
}
450
451
28704
bool PerIsolatePlatformData::FlushForegroundTasksInternal() {
452
28704
  bool did_work = false;
453
454
  while (std::unique_ptr<DelayedTask> delayed =
455
40688
      foreground_delayed_tasks_.Pop()) {
456
5992
    did_work = true;
457
5992
    uint64_t delay_millis = llround(delayed->timeout * 1000);
458
459
5992
    delayed->timer.data = static_cast<void*>(delayed.get());
460
5992
    uv_timer_init(loop_, &delayed->timer);
461
    // Timers may not guarantee queue ordering of events with the same delay if
462
    // the delay is non-zero. This should not be a problem in practice.
463
5992
    uv_timer_start(&delayed->timer, RunForegroundTask, delay_millis, 0);
464
5992
    uv_unref(reinterpret_cast<uv_handle_t*>(&delayed->timer));
465
5992
    uv_handle_count_++;
466
467
5992
    scheduled_delayed_tasks_.emplace_back(delayed.release(),
468
17304
                                          [](DelayedTask* delayed) {
469
5656
      uv_close(reinterpret_cast<uv_handle_t*>(&delayed->timer),
470
7236
               [](uv_handle_t* handle) {
471
        std::unique_ptr<DelayedTask> task {
472
1580
            static_cast<DelayedTask*>(handle->data) };
473
790
        task->platform_data->DecreaseHandleCount();
474
7236
      });
475
29288
    });
476
5992
  }
477
  // Move all foreground tasks into a separate queue and flush that queue.
478
  // This way tasks that are posted while flushing the queue will be run on the
479
  // next call of FlushForegroundTasksInternal.
480
57400
  std::queue<std::unique_ptr<Task>> tasks = foreground_tasks_.PopAll();
481
57900
  while (!tasks.empty()) {
482
29204
    std::unique_ptr<Task> task = std::move(tasks.front());
483
14606
    tasks.pop();
484
14606
    did_work = true;
485
14606
    RunForegroundTask(std::move(task));
486
  }
487
57392
  return did_work;
488
}
489
490
95194
void NodePlatform::CallOnWorkerThread(std::unique_ptr<Task> task) {
491
95194
  worker_thread_task_runner_->PostTask(std::move(task));
492
95194
}
493
494
526
void NodePlatform::CallDelayedOnWorkerThread(std::unique_ptr<Task> task,
495
                                             double delay_in_seconds) {
496
526
  worker_thread_task_runner_->PostDelayedTask(std::move(task),
497
                                              delay_in_seconds);
498
526
}
499
500
501
34407
IsolatePlatformDelegate* NodePlatform::ForIsolate(Isolate* isolate) {
502
68814
  Mutex::ScopedLock lock(per_isolate_mutex_);
503
68814
  auto data = per_isolate_[isolate];
504
34407
  CHECK_NOT_NULL(data.first);
505
68814
  return data.first;
506
}
507
508
std::shared_ptr<PerIsolatePlatformData>
509
11116
NodePlatform::ForNodeIsolate(Isolate* isolate) {
510
22232
  Mutex::ScopedLock lock(per_isolate_mutex_);
511
22232
  auto data = per_isolate_[isolate];
512
11116
  CHECK_NOT_NULL(data.first);
513
22232
  return data.second;
514
}
515
516
4
bool NodePlatform::FlushForegroundTasks(Isolate* isolate) {
517
8
  std::shared_ptr<PerIsolatePlatformData> per_isolate = ForNodeIsolate(isolate);
518
4
  if (!per_isolate) return false;
519
4
  return per_isolate->FlushForegroundTasksInternal();
520
}
521
522
34158
std::unique_ptr<v8::JobHandle> NodePlatform::CreateJob(
523
    v8::TaskPriority priority, std::unique_ptr<v8::JobTask> job_task) {
524
  return v8::platform::NewDefaultJobHandle(
525
34158
      this, priority, std::move(job_task), NumberOfWorkerThreads());
526
}
527
528
bool NodePlatform::IdleTasksEnabled(Isolate* isolate) {
529
  return ForIsolate(isolate)->IdleTasksEnabled();
530
}
531
532
std::shared_ptr<v8::TaskRunner>
533
34407
NodePlatform::GetForegroundTaskRunner(Isolate* isolate) {
534
34407
  return ForIsolate(isolate)->GetForegroundTaskRunner();
535
}
536
537
250320
double NodePlatform::MonotonicallyIncreasingTime() {
538
  // Convert nanos to seconds.
539
250320
  return uv_hrtime() / 1e9;
540
}
541
542
21982365
double NodePlatform::CurrentClockTimeMillis() {
543
21982365
  return SystemClockTimeMillis();
544
}
545
546
372892
v8::TracingController* NodePlatform::GetTracingController() {
547
372892
  CHECK_NOT_NULL(tracing_controller_);
548
372892
  return tracing_controller_;
549
}
550
551
5566
Platform::StackTracePrinter NodePlatform::GetStackTracePrinter() {
552
5566
  return []() {
553
    fprintf(stderr, "\n");
554
    DumpBacktrace(stderr);
555
    fflush(stderr);
556
11132
  };
557
}
558
559
5567
v8::PageAllocator* NodePlatform::GetPageAllocator() {
560
5567
  return page_allocator_;
561
}
562
563
template <class T>
564
23852
TaskQueue<T>::TaskQueue()
565
    : lock_(), tasks_available_(), tasks_drained_(),
566
23852
      outstanding_tasks_(0), stopped_(false), task_queue_() { }
567
568
template <class T>
569
123117
void TaskQueue<T>::Push(std::unique_ptr<T> task) {
570
246234
  Mutex::ScopedLock scoped_lock(lock_);
571
123117
  outstanding_tasks_++;
572
123117
  task_queue_.push(std::move(task));
573
123117
  tasks_available_.Signal(scoped_lock);
574
123117
}
575
576
template <class T>
577
46862
std::unique_ptr<T> TaskQueue<T>::Pop() {
578
93724
  Mutex::ScopedLock scoped_lock(lock_);
579

46862
  if (task_queue_.empty()) {
580
34787
    return std::unique_ptr<T>(nullptr);
581
  }
582
24150
  std::unique_ptr<T> result = std::move(task_queue_.front());
583
12075
  task_queue_.pop();
584
12075
  return result;
585
}
586
587
template <class T>
588
117718
std::unique_ptr<T> TaskQueue<T>::BlockingPop() {
589
235400
  Mutex::ScopedLock scoped_lock(lock_);
590

336408
  while (task_queue_.empty() && !stopped_) {
591
109381
    tasks_available_.Wait(scoped_lock);
592
  }
593
117682
  if (stopped_) {
594
22234
    return std::unique_ptr<T>(nullptr);
595
  }
596
190896
  std::unique_ptr<T> result = std::move(task_queue_.front());
597
95448
  task_queue_.pop();
598
95448
  return result;
599
}
600
601
template <class T>
602
95448
void TaskQueue<T>::NotifyOfCompletion() {
603
190896
  Mutex::ScopedLock scoped_lock(lock_);
604
95448
  if (--outstanding_tasks_ == 0) {
605
49952
    tasks_drained_.Broadcast(scoped_lock);
606
  }
607
95448
}
608
609
template <class T>
610
19305
void TaskQueue<T>::BlockingDrain() {
611
38610
  Mutex::ScopedLock scoped_lock(lock_);
612
25991
  while (outstanding_tasks_ > 0) {
613
3343
    tasks_drained_.Wait(scoped_lock);
614
  }
615
19305
}
616
617
template <class T>
618
5557
void TaskQueue<T>::Stop() {
619
11114
  Mutex::ScopedLock scoped_lock(lock_);
620
5557
  stopped_ = true;
621
5557
  tasks_available_.Broadcast(scoped_lock);
622
5557
}
623
624
template <class T>
625
40078
std::queue<std::unique_ptr<T>> TaskQueue<T>::PopAll() {
626
80156
  Mutex::ScopedLock scoped_lock(lock_);
627
40078
  std::queue<std::unique_ptr<T>> result;
628
40078
  result.swap(task_queue_);
629
80156
  return result;
630
}
631
632

16869
}  // namespace node