GCC Code Coverage Report
Directory: ../ Exec Total Coverage
File: /home/iojs/build/workspace/node-test-commit-linux-coverage-daily/nodes/benchmark/out/../src/node_platform.cc Lines: 330 338 97.6 %
Date: 2019-10-05 22:32:21 Branches: 76 104 73.1 %

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

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

9782
    TRACE_EVENT_METADATA1("__metadata", "thread_name", "name",
83
                          "WorkerThreadsTaskRunner::DelayedTaskScheduler");
84
4891
    loop_.data = this;
85
4891
    CHECK_EQ(0, uv_loop_init(&loop_));
86
4891
    flush_tasks_.data = this;
87
4891
    CHECK_EQ(0, uv_async_init(&loop_, &flush_tasks_, FlushTasks));
88
4891
    uv_sem_post(&ready_);
89
90
4891
    uv_run(&loop_, UV_RUN_DEFAULT);
91
4887
    CheckedUvLoopClose(&loop_);
92
4887
  }
93
94
4888
  static void FlushTasks(uv_async_t* flush_tasks) {
95
    DelayedTaskScheduler* scheduler =
96
4888
        ContainerOf(&DelayedTaskScheduler::loop_, flush_tasks->loop);
97
9776
    while (std::unique_ptr<Task> task = scheduler->tasks_.Pop())
98
4888
      task->Run();
99
4888
  }
100
101
9774
  class StopTask : public Task {
102
   public:
103
4887
    explicit StopTask(DelayedTaskScheduler* scheduler): scheduler_(scheduler) {}
104
105
4887
    void Run() override {
106
4887
      std::vector<uv_timer_t*> timers;
107
4887
      for (uv_timer_t* timer : scheduler_->timers_)
108
        timers.push_back(timer);
109
4887
      for (uv_timer_t* timer : timers)
110
        scheduler_->TakeTimerTask(timer);
111
      uv_close(reinterpret_cast<uv_handle_t*>(&scheduler_->flush_tasks_),
112
19548
               [](uv_handle_t* handle) {});
113
4887
    }
114
115
   private:
116
     DelayedTaskScheduler* scheduler_;
117
  };
118
119
2
  class ScheduleTask : public Task {
120
   public:
121
1
    ScheduleTask(DelayedTaskScheduler* scheduler,
122
                 std::unique_ptr<Task> task,
123
                 double delay_in_seconds)
124
      : scheduler_(scheduler),
125
1
        task_(std::move(task)),
126
2
        delay_in_seconds_(delay_in_seconds) {}
127
128
1
    void Run() override {
129
1
      uint64_t delay_millis = llround(delay_in_seconds_ * 1000);
130
1
      std::unique_ptr<uv_timer_t> timer(new uv_timer_t());
131
1
      CHECK_EQ(0, uv_timer_init(&scheduler_->loop_, timer.get()));
132
1
      timer->data = task_.release();
133
1
      CHECK_EQ(0, uv_timer_start(timer.get(), RunTask, delay_millis, 0));
134
1
      scheduler_->timers_.insert(timer.release());
135
1
    }
136
137
   private:
138
    DelayedTaskScheduler* scheduler_;
139
    std::unique_ptr<Task> task_;
140
    double delay_in_seconds_;
141
  };
142
143
1
  static void RunTask(uv_timer_t* timer) {
144
    DelayedTaskScheduler* scheduler =
145
1
        ContainerOf(&DelayedTaskScheduler::loop_, timer->loop);
146
1
    scheduler->pending_worker_tasks_->Push(scheduler->TakeTimerTask(timer));
147
1
  }
148
149
1
  std::unique_ptr<Task> TakeTimerTask(uv_timer_t* timer) {
150
1
    std::unique_ptr<Task> task(static_cast<Task*>(timer->data));
151
1
    uv_timer_stop(timer);
152
3
    uv_close(reinterpret_cast<uv_handle_t*>(timer), [](uv_handle_t* handle) {
153
1
      delete reinterpret_cast<uv_timer_t*>(handle);
154
4
    });
155
1
    timers_.erase(timer);
156
1
    return task;
157
  }
158
159
  uv_sem_t ready_;
160
  TaskQueue<Task>* pending_worker_tasks_;
161
162
  TaskQueue<Task> tasks_;
163
  uv_loop_t loop_;
164
  uv_async_t flush_tasks_;
165
  std::unordered_set<uv_timer_t*> timers_;
166
};
167
168
4891
WorkerThreadsTaskRunner::WorkerThreadsTaskRunner(int thread_pool_size) {
169
4891
  Mutex platform_workers_mutex;
170
9782
  ConditionVariable platform_workers_ready;
171
172
9782
  Mutex::ScopedLock lock(platform_workers_mutex);
173
4891
  int pending_platform_workers = thread_pool_size;
174
175
9782
  delayed_task_scheduler_ = std::make_unique<DelayedTaskScheduler>(
176
4891
      &pending_worker_tasks_);
177
4891
  threads_.push_back(delayed_task_scheduler_->Start());
178
179
24461
  for (int i = 0; i < thread_pool_size; i++) {
180
    PlatformWorkerData* worker_data = new PlatformWorkerData{
181
      &pending_worker_tasks_, &platform_workers_mutex,
182
      &platform_workers_ready, &pending_platform_workers, i
183
19570
    };
184
19570
    std::unique_ptr<uv_thread_t> t { new uv_thread_t() };
185
19570
    if (uv_thread_create(t.get(), PlatformWorkerThread,
186
19570
                         worker_data) != 0) {
187
      break;
188
    }
189
19570
    threads_.push_back(std::move(t));
190
19570
  }
191
192
  // Wait for platform workers to initialize before continuing with the
193
  // bootstrap.
194
26163
  while (pending_platform_workers > 0) {
195
16381
    platform_workers_ready.Wait(lock);
196
4891
  }
197
4891
}
198
199
361667
void WorkerThreadsTaskRunner::PostTask(std::unique_ptr<Task> task) {
200
361667
  pending_worker_tasks_.Push(std::move(task));
201
361667
}
202
203
1
void WorkerThreadsTaskRunner::PostDelayedTask(std::unique_ptr<Task> task,
204
                                              double delay_in_seconds) {
205
1
  delayed_task_scheduler_->PostDelayedTask(std::move(task), delay_in_seconds);
206
1
}
207
208
11794
void WorkerThreadsTaskRunner::BlockingDrain() {
209
11794
  pending_worker_tasks_.BlockingDrain();
210
11794
}
211
212
4887
void WorkerThreadsTaskRunner::Shutdown() {
213
4887
  pending_worker_tasks_.Stop();
214
4887
  delayed_task_scheduler_->Stop();
215
29328
  for (size_t i = 0; i < threads_.size(); i++) {
216
24441
    CHECK_EQ(0, uv_thread_join(threads_[i].get()));
217
  }
218
4887
}
219
220
5491
int WorkerThreadsTaskRunner::NumberOfWorkerThreads() const {
221
5491
  return threads_.size();
222
}
223
224
5104
PerIsolatePlatformData::PerIsolatePlatformData(
225
    Isolate* isolate, uv_loop_t* loop)
226
5104
  : loop_(loop) {
227
5104
  flush_tasks_ = new uv_async_t();
228
5104
  CHECK_EQ(0, uv_async_init(loop, flush_tasks_, FlushTasks));
229
5104
  flush_tasks_->data = static_cast<void*>(this);
230
5104
  uv_unref(reinterpret_cast<uv_handle_t*>(flush_tasks_));
231
5104
}
232
233
6261
void PerIsolatePlatformData::FlushTasks(uv_async_t* handle) {
234
6261
  auto platform_data = static_cast<PerIsolatePlatformData*>(handle->data);
235
6261
  platform_data->FlushForegroundTasksInternal();
236
6260
}
237
238
void PerIsolatePlatformData::PostIdleTask(std::unique_ptr<v8::IdleTask> task) {
239
  UNREACHABLE();
240
}
241
242
7986
void PerIsolatePlatformData::PostTask(std::unique_ptr<Task> task) {
243
7986
  CHECK_NOT_NULL(flush_tasks_);
244
7986
  foreground_tasks_.Push(std::move(task));
245
7986
  uv_async_send(flush_tasks_);
246
7986
}
247
248
5227
void PerIsolatePlatformData::PostDelayedTask(
249
    std::unique_ptr<Task> task, double delay_in_seconds) {
250
5227
  CHECK_NOT_NULL(flush_tasks_);
251
5227
  std::unique_ptr<DelayedTask> delayed(new DelayedTask());
252
5227
  delayed->task = std::move(task);
253
5227
  delayed->platform_data = shared_from_this();
254
5227
  delayed->timeout = delay_in_seconds;
255
5227
  foreground_delayed_tasks_.Push(std::move(delayed));
256
5227
  uv_async_send(flush_tasks_);
257
5227
}
258
259
534
void PerIsolatePlatformData::PostNonNestableTask(std::unique_ptr<Task> task) {
260
534
  PostTask(std::move(task));
261
534
}
262
263
241
void PerIsolatePlatformData::PostNonNestableDelayedTask(
264
    std::unique_ptr<Task> task,
265
    double delay_in_seconds) {
266
241
  PostDelayedTask(std::move(task), delay_in_seconds);
267
241
}
268
269
586
PerIsolatePlatformData::~PerIsolatePlatformData() {
270
293
  Shutdown();
271
293
}
272
273
498
void PerIsolatePlatformData::AddShutdownCallback(void (*callback)(void*),
274
                                                 void* data) {
275
498
  shutdown_callbacks_.emplace_back(ShutdownCallback { callback, data });
276
498
}
277
278
4956
void PerIsolatePlatformData::Shutdown() {
279
4956
  if (flush_tasks_ == nullptr)
280
5249
    return;
281
282
  // While there should be no V8 tasks in the queues at this point, it is
283
  // possible that Node.js-internal tasks from e.g. the inspector are still
284
  // lying around. We clear these queues and ignore the return value,
285
  // effectively deleting the tasks instead of running them.
286
4663
  foreground_delayed_tasks_.PopAll();
287
4663
  foreground_tasks_.PopAll();
288
289
4663
  CancelPendingDelayedTasks();
290
291
4663
  ShutdownCbList* copy = new ShutdownCbList(std::move(shutdown_callbacks_));
292
4663
  flush_tasks_->data = copy;
293
  uv_close(reinterpret_cast<uv_handle_t*>(flush_tasks_),
294
5089
           [](uv_handle_t* handle) {
295
    std::unique_ptr<ShutdownCbList> callbacks(
296
213
        static_cast<ShutdownCbList*>(handle->data));
297
629
    for (const auto& callback : *callbacks)
298
416
      callback.cb(callback.data);
299
213
    delete reinterpret_cast<uv_async_t*>(handle);
300
9752
  });
301
4663
  flush_tasks_ = nullptr;
302
}
303
304
4891
NodePlatform::NodePlatform(int thread_pool_size,
305
4891
                           TracingController* tracing_controller) {
306
4891
  if (tracing_controller) {
307
4891
    tracing_controller_ = tracing_controller;
308
  } else {
309
    tracing_controller_ = new TracingController();
310
  }
311
9782
  worker_thread_task_runner_ =
312
4891
      std::make_shared<WorkerThreadsTaskRunner>(thread_pool_size);
313
4891
}
314
315
5104
void NodePlatform::RegisterIsolate(Isolate* isolate, uv_loop_t* loop) {
316
5104
  Mutex::ScopedLock lock(per_isolate_mutex_);
317
10208
  std::shared_ptr<PerIsolatePlatformData> existing = per_isolate_[isolate];
318
5104
  CHECK(!existing);
319
10208
  per_isolate_[isolate] =
320
10208
      std::make_shared<PerIsolatePlatformData>(isolate, loop);
321
5104
}
322
323
4663
void NodePlatform::UnregisterIsolate(Isolate* isolate) {
324
4663
  Mutex::ScopedLock lock(per_isolate_mutex_);
325
9326
  std::shared_ptr<PerIsolatePlatformData> existing = per_isolate_[isolate];
326
4663
  CHECK(existing);
327
4663
  existing->Shutdown();
328
9326
  per_isolate_.erase(isolate);
329
4663
}
330
331
498
void NodePlatform::AddIsolateFinishedCallback(Isolate* isolate,
332
                                              void (*cb)(void*), void* data) {
333
498
  Mutex::ScopedLock lock(per_isolate_mutex_);
334
498
  auto it = per_isolate_.find(isolate);
335
498
  if (it == per_isolate_.end()) {
336
    CHECK(it->second);
337
    cb(data);
338
498
    return;
339
  }
340
498
  it->second->AddShutdownCallback(cb, data);
341
}
342
343
4887
void NodePlatform::Shutdown() {
344
4887
  worker_thread_task_runner_->Shutdown();
345
346
  {
347
4887
    Mutex::ScopedLock lock(per_isolate_mutex_);
348
4887
    per_isolate_.clear();
349
  }
350
4887
}
351
352
5491
int NodePlatform::NumberOfWorkerThreads() {
353
5491
  return worker_thread_task_runner_->NumberOfWorkerThreads();
354
}
355
356
7954
void PerIsolatePlatformData::RunForegroundTask(std::unique_ptr<Task> task) {
357
7954
  Isolate* isolate = Isolate::GetCurrent();
358
7954
  DebugSealHandleScope scope(isolate);
359
7954
  Environment* env = Environment::GetCurrent(isolate);
360
7954
  if (env != nullptr) {
361
    InternalCallbackScope cb_scope(env, Local<Object>(), { 0, 0 },
362
15906
                                   InternalCallbackScope::kAllowEmptyResource);
363
7953
    task->Run();
364
  } else {
365
1
    task->Run();
366
  }
367
7953
}
368
369
267
void PerIsolatePlatformData::DeleteFromScheduledTasks(DelayedTask* task) {
370
  auto it = std::find_if(scheduled_delayed_tasks_.begin(),
371
                         scheduled_delayed_tasks_.end(),
372
479
                         [task](const DelayedTaskPointer& delayed) -> bool {
373
479
          return delayed.get() == task;
374
267
      });
375
267
  CHECK_NE(it, scheduled_delayed_tasks_.end());
376
267
  scheduled_delayed_tasks_.erase(it);
377
267
}
378
379
267
void PerIsolatePlatformData::RunForegroundTask(uv_timer_t* handle) {
380
267
  DelayedTask* delayed = static_cast<DelayedTask*>(handle->data);
381
267
  RunForegroundTask(std::move(delayed->task));
382
267
  delayed->platform_data->DeleteFromScheduledTasks(delayed);
383
267
}
384
385
9325
void PerIsolatePlatformData::CancelPendingDelayedTasks() {
386
9325
  scheduled_delayed_tasks_.clear();
387
9325
}
388
389
9270
void NodePlatform::DrainTasks(Isolate* isolate) {
390
9270
  std::shared_ptr<PerIsolatePlatformData> per_isolate = ForIsolate(isolate);
391
392
11794
  do {
393
    // Worker tasks aren't associated with an Isolate.
394
11794
    worker_thread_task_runner_->BlockingDrain();
395
21064
  } while (per_isolate->FlushForegroundTasksInternal());
396
9270
}
397
398
18365
bool PerIsolatePlatformData::FlushForegroundTasksInternal() {
399
18365
  bool did_work = false;
400
401
  while (std::unique_ptr<DelayedTask> delayed =
402
23380
      foreground_delayed_tasks_.Pop()) {
403
5015
    did_work = true;
404
5015
    uint64_t delay_millis = llround(delayed->timeout * 1000);
405
406
5015
    delayed->timer.data = static_cast<void*>(delayed.get());
407
5015
    uv_timer_init(loop_, &delayed->timer);
408
    // Timers may not guarantee queue ordering of events with the same delay if
409
    // the delay is non-zero. This should not be a problem in practice.
410
5015
    uv_timer_start(&delayed->timer, RunForegroundTask, delay_millis, 0);
411
5015
    uv_unref(reinterpret_cast<uv_handle_t*>(&delayed->timer));
412
413
5015
    scheduled_delayed_tasks_.emplace_back(delayed.release(),
414
4853
                                          [](DelayedTask* delayed) {
415
      uv_close(reinterpret_cast<uv_handle_t*>(&delayed->timer),
416
5793
               [](uv_handle_t* handle) {
417
470
        delete static_cast<DelayedTask*>(handle->data);
418
10646
      });
419
9868
    });
420
5015
  }
421
  // Move all foreground tasks into a separate queue and flush that queue.
422
  // This way tasks that are posted while flushing the queue will be run on the
423
  // next call of FlushForegroundTasksInternal.
424
18365
  std::queue<std::unique_ptr<Task>> tasks = foreground_tasks_.PopAll();
425
44416
  while (!tasks.empty()) {
426
7687
    std::unique_ptr<Task> task = std::move(tasks.front());
427
7687
    tasks.pop();
428
7687
    did_work = true;
429
7687
    RunForegroundTask(std::move(task));
430
7686
  }
431
23379
  return did_work;
432
}
433
434
361667
void NodePlatform::CallOnWorkerThread(std::unique_ptr<Task> task) {
435
361667
  worker_thread_task_runner_->PostTask(std::move(task));
436
361667
}
437
438
1
void NodePlatform::CallDelayedOnWorkerThread(std::unique_ptr<Task> task,
439
                                             double delay_in_seconds) {
440
1
  worker_thread_task_runner_->PostDelayedTask(std::move(task),
441
1
                                              delay_in_seconds);
442
1
}
443
444
445
std::shared_ptr<PerIsolatePlatformData>
446
37777
NodePlatform::ForIsolate(Isolate* isolate) {
447
37777
  Mutex::ScopedLock lock(per_isolate_mutex_);
448
37779
  std::shared_ptr<PerIsolatePlatformData> data = per_isolate_[isolate];
449
37779
  CHECK(data);
450
37779
  return data;
451
}
452
453
310
bool NodePlatform::FlushForegroundTasks(Isolate* isolate) {
454
310
  return ForIsolate(isolate)->FlushForegroundTasksInternal();
455
}
456
457
4662
void NodePlatform::CancelPendingDelayedTasks(Isolate* isolate) {
458
4662
  ForIsolate(isolate)->CancelPendingDelayedTasks();
459
4662
}
460
461
53570
bool NodePlatform::IdleTasksEnabled(Isolate* isolate) { return false; }
462
463
std::shared_ptr<v8::TaskRunner>
464
23535
NodePlatform::GetForegroundTaskRunner(Isolate* isolate) {
465
23535
  return ForIsolate(isolate);
466
}
467
468
3266996
double NodePlatform::MonotonicallyIncreasingTime() {
469
  // Convert nanos to seconds.
470
3266996
  return uv_hrtime() / 1e9;
471
}
472
473
179464284
double NodePlatform::CurrentClockTimeMillis() {
474
179464284
  return SystemClockTimeMillis();
475
}
476
477
338850
TracingController* NodePlatform::GetTracingController() {
478
338850
  CHECK_NOT_NULL(tracing_controller_);
479
338850
  return tracing_controller_;
480
}
481
482
template <class T>
483
19990
TaskQueue<T>::TaskQueue()
484
    : lock_(), tasks_available_(), tasks_drained_(),
485
19990
      outstanding_tasks_(0), stopped_(false), task_queue_() { }
486
487
template <class T>
488
379768
void TaskQueue<T>::Push(std::unique_ptr<T> task) {
489
379768
  Mutex::ScopedLock scoped_lock(lock_);
490
379769
  outstanding_tasks_++;
491
379769
  task_queue_.push(std::move(task));
492
379769
  tasks_available_.Signal(scoped_lock);
493
379769
}
494
495
template <class T>
496
33156
std::unique_ptr<T> TaskQueue<T>::Pop() {
497
33156
  Mutex::ScopedLock scoped_lock(lock_);
498

33156
  if (task_queue_.empty()) {
499
23253
    return std::unique_ptr<T>(nullptr);
500
  }
501
19806
  std::unique_ptr<T> result = std::move(task_queue_.front());
502
9903
  task_queue_.pop();
503
43059
  return result;
504
}
505
506
template <class T>
507
381168
std::unique_ptr<T> TaskQueue<T>::BlockingPop() {
508
381168
  Mutex::ScopedLock scoped_lock(lock_);
509

1092497
  while (task_queue_.empty() && !stopped_) {
510
330041
    tasks_available_.Wait(scoped_lock);
511
  }
512
381220
  if (stopped_) {
513
19554
    return std::unique_ptr<T>(nullptr);
514
  }
515
723332
  std::unique_ptr<T> result = std::move(task_queue_.front());
516
361666
  task_queue_.pop();
517
742886
  return result;
518
}
519
520
template <class T>
521
361203
void TaskQueue<T>::NotifyOfCompletion() {
522
361203
  Mutex::ScopedLock scoped_lock(lock_);
523
361666
  if (--outstanding_tasks_ == 0) {
524
204692
    tasks_drained_.Broadcast(scoped_lock);
525
361666
  }
526
361529
}
527
528
template <class T>
529
11794
void TaskQueue<T>::BlockingDrain() {
530
11794
  Mutex::ScopedLock scoped_lock(lock_);
531
24415
  while (outstanding_tasks_ > 0) {
532
827
    tasks_drained_.Wait(scoped_lock);
533
11794
  }
534
11794
}
535
536
template <class T>
537
4887
void TaskQueue<T>::Stop() {
538
4887
  Mutex::ScopedLock scoped_lock(lock_);
539
4887
  stopped_ = true;
540
4887
  tasks_available_.Broadcast(scoped_lock);
541
4887
}
542
543
template <class T>
544
27691
std::queue<std::unique_ptr<T>> TaskQueue<T>::PopAll() {
545
27691
  Mutex::ScopedLock scoped_lock(lock_);
546
27691
  std::queue<std::unique_ptr<T>> result;
547
27691
  result.swap(task_queue_);
548
27691
  return result;
549
}
550
551
}  // namespace node