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: 317 328 96.6 %
Date: 2019-03-02 22:23:06 Branches: 76 102 74.5 %

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 "util.h"
7
#include <algorithm>
8
9
namespace node {
10
11
using v8::HandleScope;
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
16822
static void PlatformWorkerThread(void* data) {
30
  std::unique_ptr<PlatformWorkerData>
31
16822
      worker_data(static_cast<PlatformWorkerData*>(data));
32
33
16955
  TaskQueue<Task>* pending_worker_tasks = worker_data->task_queue;
34

33668
  TRACE_EVENT_METADATA1("__metadata", "thread_name", "name",
35
                        "PlatformWorkerThread");
36
37
  // Notify the main thread that the platform worker is ready.
38
  {
39
16905
    Mutex::ScopedLock lock(*worker_data->platform_workers_mutex);
40
16994
    (*worker_data->pending_platform_workers)--;
41
16994
    worker_data->platform_workers_ready->Signal(lock);
42
  }
43
44
149430
  while (std::unique_ptr<Task> task = pending_worker_tasks->BlockingPop()) {
45
132265
    task->Run();
46
132339
    pending_worker_tasks->NotifyOfCompletion();
47
132449
  }
48
16974
}
49
50
}  // namespace
51
52
4242
class WorkerThreadsTaskRunner::DelayedTaskScheduler {
53
 public:
54
4247
  explicit DelayedTaskScheduler(TaskQueue<Task>* tasks)
55
4247
    : pending_worker_tasks_(tasks) {}
56
57
4247
  std::unique_ptr<uv_thread_t> Start() {
58
12741
    auto start_thread = [](void* data) {
59
4247
      static_cast<DelayedTaskScheduler*>(data)->Run();
60
12736
    };
61
4247
    std::unique_ptr<uv_thread_t> t { new uv_thread_t() };
62
4247
    uv_sem_init(&ready_, 0);
63
4247
    CHECK_EQ(0, uv_thread_create(t.get(), start_thread, this));
64
4247
    uv_sem_wait(&ready_);
65
4247
    uv_sem_destroy(&ready_);
66
4247
    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
4242
  void Stop() {
76
4242
    tasks_.Push(std::unique_ptr<Task>(new StopTask(this)));
77
4242
    uv_async_send(&flush_tasks_);
78
4242
  }
79
80
 private:
81
4247
  void Run() {
82

8494
    TRACE_EVENT_METADATA1("__metadata", "thread_name", "name",
83
                          "WorkerThreadsTaskRunner::DelayedTaskScheduler");
84
4247
    loop_.data = this;
85
4247
    CHECK_EQ(0, uv_loop_init(&loop_));
86
4247
    flush_tasks_.data = this;
87
4247
    CHECK_EQ(0, uv_async_init(&loop_, &flush_tasks_, FlushTasks));
88
4247
    uv_sem_post(&ready_);
89
90
4247
    uv_run(&loop_, UV_RUN_DEFAULT);
91
4242
    CheckedUvLoopClose(&loop_);
92
4242
  }
93
94
4243
  static void FlushTasks(uv_async_t* flush_tasks) {
95
    DelayedTaskScheduler* scheduler =
96
4243
        ContainerOf(&DelayedTaskScheduler::loop_, flush_tasks->loop);
97
8486
    while (std::unique_ptr<Task> task = scheduler->tasks_.Pop())
98
4243
      task->Run();
99
4243
  }
100
101
8484
  class StopTask : public Task {
102
   public:
103
4242
    explicit StopTask(DelayedTaskScheduler* scheduler): scheduler_(scheduler) {}
104
105
4242
    void Run() override {
106
4242
      std::vector<uv_timer_t*> timers;
107
4242
      for (uv_timer_t* timer : scheduler_->timers_)
108
        timers.push_back(timer);
109
4242
      for (uv_timer_t* timer : timers)
110
        scheduler_->TakeTimerTask(timer);
111
      uv_close(reinterpret_cast<uv_handle_t*>(&scheduler_->flush_tasks_),
112
16968
               [](uv_handle_t* handle) {});
113
4242
    }
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
      uint64_t delay_millis =
130
1
          static_cast<uint64_t>(delay_in_seconds_ + 0.5) * 1000;
131
1
      std::unique_ptr<uv_timer_t> timer(new uv_timer_t());
132
1
      CHECK_EQ(0, uv_timer_init(&scheduler_->loop_, timer.get()));
133
1
      timer->data = task_.release();
134
1
      CHECK_EQ(0, uv_timer_start(timer.get(), RunTask, delay_millis, 0));
135
1
      scheduler_->timers_.insert(timer.release());
136
1
    }
137
138
   private:
139
    DelayedTaskScheduler* scheduler_;
140
    std::unique_ptr<Task> task_;
141
    double delay_in_seconds_;
142
  };
143
144
1
  static void RunTask(uv_timer_t* timer) {
145
    DelayedTaskScheduler* scheduler =
146
1
        ContainerOf(&DelayedTaskScheduler::loop_, timer->loop);
147
1
    scheduler->pending_worker_tasks_->Push(scheduler->TakeTimerTask(timer));
148
1
  }
149
150
1
  std::unique_ptr<Task> TakeTimerTask(uv_timer_t* timer) {
151
1
    std::unique_ptr<Task> task(static_cast<Task*>(timer->data));
152
1
    uv_timer_stop(timer);
153
3
    uv_close(reinterpret_cast<uv_handle_t*>(timer), [](uv_handle_t* handle) {
154
1
      delete reinterpret_cast<uv_timer_t*>(handle);
155
4
    });
156
1
    timers_.erase(timer);
157
1
    return task;
158
  }
159
160
  uv_sem_t ready_;
161
  TaskQueue<v8::Task>* pending_worker_tasks_;
162
163
  TaskQueue<v8::Task> tasks_;
164
  uv_loop_t loop_;
165
  uv_async_t flush_tasks_;
166
  std::unordered_set<uv_timer_t*> timers_;
167
};
168
169
4247
WorkerThreadsTaskRunner::WorkerThreadsTaskRunner(int thread_pool_size) {
170
4247
  Mutex platform_workers_mutex;
171
8494
  ConditionVariable platform_workers_ready;
172
173
8494
  Mutex::ScopedLock lock(platform_workers_mutex);
174
4247
  int pending_platform_workers = thread_pool_size;
175
176
  delayed_task_scheduler_.reset(
177
4247
      new DelayedTaskScheduler(&pending_worker_tasks_));
178
4247
  threads_.push_back(delayed_task_scheduler_->Start());
179
180
21241
  for (int i = 0; i < thread_pool_size; i++) {
181
    PlatformWorkerData* worker_data = new PlatformWorkerData{
182
      &pending_worker_tasks_, &platform_workers_mutex,
183
      &platform_workers_ready, &pending_platform_workers, i
184
16994
    };
185
16994
    std::unique_ptr<uv_thread_t> t { new uv_thread_t() };
186
16994
    if (uv_thread_create(t.get(), PlatformWorkerThread,
187
16994
                         worker_data) != 0) {
188
      break;
189
    }
190
16994
    threads_.push_back(std::move(t));
191
16994
  }
192
193
  // Wait for platform workers to initialize before continuing with the
194
  // bootstrap.
195
22267
  while (pending_platform_workers > 0) {
196
13773
    platform_workers_ready.Wait(lock);
197
4247
  }
198
4247
}
199
200
132459
void WorkerThreadsTaskRunner::PostTask(std::unique_ptr<Task> task) {
201
132459
  pending_worker_tasks_.Push(std::move(task));
202
132459
}
203
204
1
void WorkerThreadsTaskRunner::PostDelayedTask(std::unique_ptr<v8::Task> task,
205
                                              double delay_in_seconds) {
206
1
  delayed_task_scheduler_->PostDelayedTask(std::move(task), delay_in_seconds);
207
1
}
208
209
14044
void WorkerThreadsTaskRunner::BlockingDrain() {
210
14044
  pending_worker_tasks_.BlockingDrain();
211
14044
}
212
213
4242
void WorkerThreadsTaskRunner::Shutdown() {
214
4242
  pending_worker_tasks_.Stop();
215
4242
  delayed_task_scheduler_->Stop();
216
25458
  for (size_t i = 0; i < threads_.size(); i++) {
217
21216
    CHECK_EQ(0, uv_thread_join(threads_[i].get()));
218
  }
219
4242
}
220
221
4644
int WorkerThreadsTaskRunner::NumberOfWorkerThreads() const {
222
4644
  return threads_.size();
223
}
224
225
4423
PerIsolatePlatformData::PerIsolatePlatformData(
226
    v8::Isolate* isolate, uv_loop_t* loop)
227
4423
  : loop_(loop) {
228
4423
  flush_tasks_ = new uv_async_t();
229
4423
  CHECK_EQ(0, uv_async_init(loop, flush_tasks_, FlushTasks));
230
4423
  flush_tasks_->data = static_cast<void*>(this);
231
4423
  uv_unref(reinterpret_cast<uv_handle_t*>(flush_tasks_));
232
4423
}
233
234
5341
void PerIsolatePlatformData::FlushTasks(uv_async_t* handle) {
235
5341
  auto platform_data = static_cast<PerIsolatePlatformData*>(handle->data);
236
5341
  platform_data->FlushForegroundTasksInternal();
237
5340
}
238
239
void PerIsolatePlatformData::PostIdleTask(std::unique_ptr<v8::IdleTask> task) {
240
  UNREACHABLE();
241
}
242
243
10791
void PerIsolatePlatformData::PostTask(std::unique_ptr<Task> task) {
244
10791
  CHECK_NOT_NULL(flush_tasks_);
245
10791
  foreground_tasks_.Push(std::move(task));
246
10791
  uv_async_send(flush_tasks_);
247
10791
}
248
249
136
void PerIsolatePlatformData::PostDelayedTask(
250
    std::unique_ptr<Task> task, double delay_in_seconds) {
251
136
  CHECK_NOT_NULL(flush_tasks_);
252
136
  std::unique_ptr<DelayedTask> delayed(new DelayedTask());
253
136
  delayed->task = std::move(task);
254
136
  delayed->platform_data = shared_from_this();
255
136
  delayed->timeout = delay_in_seconds;
256
136
  foreground_delayed_tasks_.Push(std::move(delayed));
257
136
  uv_async_send(flush_tasks_);
258
136
}
259
260
7850
PerIsolatePlatformData::~PerIsolatePlatformData() {
261
3925
  Shutdown();
262
3925
}
263
264
7977
void PerIsolatePlatformData::Shutdown() {
265
7977
  if (flush_tasks_ == nullptr)
266
11902
    return;
267
268
4052
  CHECK_NULL(foreground_delayed_tasks_.Pop());
269
4052
  CHECK_NULL(foreground_tasks_.Pop());
270
4052
  CancelPendingDelayedTasks();
271
272
  uv_close(reinterpret_cast<uv_handle_t*>(flush_tasks_),
273
4404
           [](uv_handle_t* handle) {
274
176
    delete reinterpret_cast<uv_async_t*>(handle);
275
8456
  });
276
4052
  flush_tasks_ = nullptr;
277
}
278
279
4420
void PerIsolatePlatformData::ref() {
280
4420
  ref_count_++;
281
4420
}
282
283
8104
int PerIsolatePlatformData::unref() {
284
8104
  return --ref_count_;
285
}
286
287
4247
NodePlatform::NodePlatform(int thread_pool_size,
288
4247
                           TracingController* tracing_controller) {
289
4247
  if (tracing_controller) {
290
4247
    tracing_controller_ = tracing_controller;
291
  } else {
292
    tracing_controller_ = new TracingController();
293
  }
294
8494
  worker_thread_task_runner_ =
295
4247
      std::make_shared<WorkerThreadsTaskRunner>(thread_pool_size);
296
4247
}
297
298
8843
void NodePlatform::RegisterIsolate(Isolate* isolate, uv_loop_t* loop) {
299
8843
  Mutex::ScopedLock lock(per_isolate_mutex_);
300
17686
  std::shared_ptr<PerIsolatePlatformData> existing = per_isolate_[isolate];
301
8843
  if (existing) {
302
4420
    CHECK_EQ(loop, existing->event_loop());
303
4420
    existing->ref();
304
  } else {
305
8846
    per_isolate_[isolate] =
306
4423
        std::make_shared<PerIsolatePlatformData>(isolate, loop);
307
8843
  }
308
8843
}
309
310
8104
void NodePlatform::UnregisterIsolate(Isolate* isolate) {
311
8104
  Mutex::ScopedLock lock(per_isolate_mutex_);
312
16208
  std::shared_ptr<PerIsolatePlatformData> existing = per_isolate_[isolate];
313
8104
  CHECK(existing);
314
8104
  if (existing->unref() == 0) {
315
4052
    existing->Shutdown();
316
4052
    per_isolate_.erase(isolate);
317
8104
  }
318
8104
}
319
320
4242
void NodePlatform::Shutdown() {
321
4242
  worker_thread_task_runner_->Shutdown();
322
323
  {
324
4242
    Mutex::ScopedLock lock(per_isolate_mutex_);
325
4242
    per_isolate_.clear();
326
  }
327
4242
}
328
329
4644
int NodePlatform::NumberOfWorkerThreads() {
330
4644
  return worker_thread_task_runner_->NumberOfWorkerThreads();
331
}
332
333
10518
void PerIsolatePlatformData::RunForegroundTask(std::unique_ptr<Task> task) {
334
10518
  Isolate* isolate = Isolate::GetCurrent();
335
10529
  HandleScope scope(isolate);
336
10535
  Environment* env = Environment::GetCurrent(isolate);
337
  InternalCallbackScope cb_scope(env, Local<Object>(), { 0, 0 },
338
31603
                                 InternalCallbackScope::kAllowEmptyResource);
339
21067
  task->Run();
340
10532
}
341
342
3
void PerIsolatePlatformData::DeleteFromScheduledTasks(DelayedTask* task) {
343
  auto it = std::find_if(scheduled_delayed_tasks_.begin(),
344
                         scheduled_delayed_tasks_.end(),
345
3
                         [task](const DelayedTaskPointer& delayed) -> bool {
346
3
          return delayed.get() == task;
347
3
      });
348
3
  CHECK_NE(it, scheduled_delayed_tasks_.end());
349
3
  scheduled_delayed_tasks_.erase(it);
350
3
}
351
352
3
void PerIsolatePlatformData::RunForegroundTask(uv_timer_t* handle) {
353
3
  DelayedTask* delayed = static_cast<DelayedTask*>(handle->data);
354
3
  RunForegroundTask(std::move(delayed->task));
355
3
  delayed->platform_data->DeleteFromScheduledTasks(delayed);
356
3
}
357
358
8103
void PerIsolatePlatformData::CancelPendingDelayedTasks() {
359
8103
  scheduled_delayed_tasks_.clear();
360
8103
}
361
362
8084
void NodePlatform::DrainTasks(Isolate* isolate) {
363
8084
  std::shared_ptr<PerIsolatePlatformData> per_isolate = ForIsolate(isolate);
364
365
14044
  do {
366
    // Worker tasks aren't associated with an Isolate.
367
14044
    worker_thread_task_runner_->BlockingDrain();
368
22126
  } while (per_isolate->FlushForegroundTasksInternal());
369
8082
}
370
371
172798
bool PerIsolatePlatformData::FlushForegroundTasksInternal() {
372
172798
  bool did_work = false;
373
374
  while (std::unique_ptr<DelayedTask> delayed =
375
172933
      foreground_delayed_tasks_.Pop()) {
376
135
    did_work = true;
377
    uint64_t delay_millis =
378
135
        static_cast<uint64_t>(delayed->timeout + 0.5) * 1000;
379
135
    delayed->timer.data = static_cast<void*>(delayed.get());
380
135
    uv_timer_init(loop_, &delayed->timer);
381
    // Timers may not guarantee queue ordering of events with the same delay if
382
    // the delay is non-zero. This should not be a problem in practice.
383
135
    uv_timer_start(&delayed->timer, RunForegroundTask, delay_millis, 0);
384
135
    uv_unref(reinterpret_cast<uv_handle_t*>(&delayed->timer));
385
386
135
    scheduled_delayed_tasks_.emplace_back(delayed.release(),
387
131
                                          [](DelayedTask* delayed) {
388
      uv_close(reinterpret_cast<uv_handle_t*>(&delayed->timer),
389
139
               [](uv_handle_t* handle) {
390
4
        delete static_cast<DelayedTask*>(handle->data);
391
270
      });
392
266
    });
393
135
  }
394
  // Move all foreground tasks into a separate queue and flush that queue.
395
  // This way tasks that are posted while flushing the queue will be run on the
396
  // next call of FlushForegroundTasksInternal.
397
172798
  std::queue<std::unique_ptr<Task>> tasks = foreground_tasks_.PopAll();
398
356109
  while (!tasks.empty()) {
399
10526
    std::unique_ptr<Task> task = std::move(tasks.front());
400
10531
    tasks.pop();
401
10531
    did_work = true;
402
10531
    RunForegroundTask(std::move(task));
403
10529
  }
404
172929
  return did_work;
405
}
406
407
132459
void NodePlatform::CallOnWorkerThread(std::unique_ptr<v8::Task> task) {
408
132459
  worker_thread_task_runner_->PostTask(std::move(task));
409
132459
}
410
411
1
void NodePlatform::CallDelayedOnWorkerThread(std::unique_ptr<v8::Task> task,
412
                                             double delay_in_seconds) {
413
1
  worker_thread_task_runner_->PostDelayedTask(std::move(task),
414
1
                                              delay_in_seconds);
415
1
}
416
417
418
std::shared_ptr<PerIsolatePlatformData>
419
185212
NodePlatform::ForIsolate(Isolate* isolate) {
420
185212
  Mutex::ScopedLock lock(per_isolate_mutex_);
421
185213
  std::shared_ptr<PerIsolatePlatformData> data = per_isolate_[isolate];
422
185213
  CHECK(data);
423
185213
  return data;
424
}
425
426
void NodePlatform::CallOnForegroundThread(Isolate* isolate, Task* task) {
427
  ForIsolate(isolate)->PostTask(std::unique_ptr<Task>(task));
428
}
429
430
void NodePlatform::CallDelayedOnForegroundThread(Isolate* isolate,
431
                                                 Task* task,
432
                                                 double delay_in_seconds) {
433
  ForIsolate(isolate)->PostDelayedTask(
434
    std::unique_ptr<Task>(task), delay_in_seconds);
435
}
436
437
153413
bool NodePlatform::FlushForegroundTasks(v8::Isolate* isolate) {
438
153413
  return ForIsolate(isolate)->FlushForegroundTasksInternal();
439
}
440
441
4051
void NodePlatform::CancelPendingDelayedTasks(v8::Isolate* isolate) {
442
4051
  ForIsolate(isolate)->CancelPendingDelayedTasks();
443
4051
}
444
445
31089
bool NodePlatform::IdleTasksEnabled(Isolate* isolate) { return false; }
446
447
std::shared_ptr<v8::TaskRunner>
448
19665
NodePlatform::GetForegroundTaskRunner(Isolate* isolate) {
449
19665
  return ForIsolate(isolate);
450
}
451
452
1195819
double NodePlatform::MonotonicallyIncreasingTime() {
453
  // Convert nanos to seconds.
454
1195819
  return uv_hrtime() / 1e9;
455
}
456
457
107943296
double NodePlatform::CurrentClockTimeMillis() {
458
107943296
  return SystemClockTimeMillis();
459
}
460
461
215259
TracingController* NodePlatform::GetTracingController() {
462
215259
  CHECK_NOT_NULL(tracing_controller_);
463
215259
  return tracing_controller_;
464
}
465
466
template <class T>
467
17340
TaskQueue<T>::TaskQueue()
468
    : lock_(), tasks_available_(), tasks_drained_(),
469
17340
      outstanding_tasks_(0), stopped_(false), task_queue_() { }
470
471
template <class T>
472
147630
void TaskQueue<T>::Push(std::unique_ptr<T> task) {
473
147630
  Mutex::ScopedLock scoped_lock(lock_);
474
147630
  outstanding_tasks_++;
475
147630
  task_queue_.push(std::move(task));
476
147630
  tasks_available_.Signal(scoped_lock);
477
147630
}
478
479
template <class T>
480
189523
std::unique_ptr<T> TaskQueue<T>::Pop() {
481
189523
  Mutex::ScopedLock scoped_lock(lock_);
482

189523
  if (task_queue_.empty()) {
483
185145
    return std::unique_ptr<T>(nullptr);
484
  }
485
8756
  std::unique_ptr<T> result = std::move(task_queue_.front());
486
4378
  task_queue_.pop();
487
193901
  return result;
488
}
489
490
template <class T>
491
149405
std::unique_ptr<T> TaskQueue<T>::BlockingPop() {
492
149405
  Mutex::ScopedLock scoped_lock(lock_);
493

434364
  while (task_queue_.empty() && !stopped_) {
494
135476
    tasks_available_.Wait(scoped_lock);
495
  }
496
149434
  if (stopped_) {
497
16974
    return std::unique_ptr<T>(nullptr);
498
  }
499
264920
  std::unique_ptr<T> result = std::move(task_queue_.front());
500
132460
  task_queue_.pop();
501
281894
  return result;
502
}
503
504
template <class T>
505
132331
void TaskQueue<T>::NotifyOfCompletion() {
506
132331
  Mutex::ScopedLock scoped_lock(lock_);
507
132460
  if (--outstanding_tasks_ == 0) {
508
88546
    tasks_drained_.Broadcast(scoped_lock);
509
132460
  }
510
132436
}
511
512
template <class T>
513
14044
void TaskQueue<T>::BlockingDrain() {
514
14044
  Mutex::ScopedLock scoped_lock(lock_);
515
28198
  while (outstanding_tasks_ > 0) {
516
110
    tasks_drained_.Wait(scoped_lock);
517
14044
  }
518
14044
}
519
520
template <class T>
521
4242
void TaskQueue<T>::Stop() {
522
4242
  Mutex::ScopedLock scoped_lock(lock_);
523
4242
  stopped_ = true;
524
4242
  tasks_available_.Broadcast(scoped_lock);
525
4242
}
526
527
template <class T>
528
172798
std::queue<std::unique_ptr<T>> TaskQueue<T>::PopAll() {
529
172798
  Mutex::ScopedLock scoped_lock(lock_);
530
172798
  std::queue<std::unique_ptr<T>> result;
531
172782
  result.swap(task_queue_);
532
172797
  return result;
533
}
534
535
}  // namespace node