default-job.h 5.02 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
// Copyright 2020 the V8 project authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#ifndef V8_LIBPLATFORM_DEFAULT_JOB_H_
#define V8_LIBPLATFORM_DEFAULT_JOB_H_

#include <atomic>
#include <memory>

#include "include/libplatform/libplatform-export.h"
#include "include/v8-platform.h"
#include "src/base/platform/condition-variable.h"
#include "src/base/platform/mutex.h"

namespace v8 {
namespace platform {

class V8_PLATFORM_EXPORT DefaultJobState
20
    : public std::enable_shared_from_this<DefaultJobState> {
21
 public:
22 23
  class JobDelegate : public v8::JobDelegate {
   public:
24 25
    explicit JobDelegate(DefaultJobState* outer, bool is_joining_thread = false)
        : outer_(outer), is_joining_thread_(is_joining_thread) {}
26 27 28 29 30 31 32 33 34 35
    ~JobDelegate();

    void NotifyConcurrencyIncrease() override {
      outer_->NotifyConcurrencyIncrease();
    }
    bool ShouldYield() override {
      // Thread-safe but may return an outdated result.
      return outer_->is_canceled_.load(std::memory_order_relaxed);
    }
    uint8_t GetTaskId() override;
36
    bool IsJoiningThread() const override { return is_joining_thread_; }
37 38 39 40 41 42 43

   private:
    static constexpr uint8_t kInvalidTaskId =
        std::numeric_limits<uint8_t>::max();

    DefaultJobState* outer_;
    uint8_t task_id_ = kInvalidTaskId;
44
    bool is_joining_thread_;
45 46
  };

47
  DefaultJobState(Platform* platform, std::unique_ptr<JobTask> job_task,
48
                  TaskPriority priority, size_t num_worker_threads);
49 50
  virtual ~DefaultJobState();

51 52 53
  void NotifyConcurrencyIncrease();
  uint8_t AcquireTaskId();
  void ReleaseTaskId(uint8_t task_id);
54 55 56

  void Join();
  void CancelAndWait();
57
  void CancelAndDetach();
58
  bool IsActive();
59 60 61 62 63 64 65 66 67

  // Must be called before running |job_task_| for the first time. If it returns
  // true, then the worker thread must contribute and must call DidRunTask(), or
  // false if it should return.
  bool CanRunFirstTask();
  // Must be called after running |job_task_|. Returns true if the worker thread
  // must contribute again, or false if it should return.
  bool DidRunTask();

68 69
  void UpdatePriority(TaskPriority);

70 71 72 73 74 75 76 77 78 79
 private:
  // Called from the joining thread. Waits for the worker count to be below or
  // equal to max concurrency (will happen when a worker calls
  // DidRunTask()). Returns true if the joining thread should run a task, or
  // false if joining was completed and all other workers returned because
  // there's no work remaining.
  bool WaitForParticipationOpportunityLockRequired();

  // Returns GetMaxConcurrency() capped by the number of threads used by this
  // job.
80
  size_t CappedMaxConcurrency(size_t worker_count) const;
81

82
  void CallOnWorkerThread(TaskPriority priority, std::unique_ptr<Task> task);
83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99

  Platform* const platform_;
  std::unique_ptr<JobTask> job_task_;

  // All members below are protected by |mutex_|.
  base::Mutex mutex_;
  TaskPriority priority_;
  // Number of workers running this job.
  size_t active_workers_ = 0;
  // Number of posted tasks that aren't running this job yet.
  size_t pending_tasks_ = 0;
  // Indicates if the job is canceled.
  std::atomic_bool is_canceled_{false};
  // Number of worker threads available to schedule the worker task.
  size_t num_worker_threads_;
  // Signaled when a worker returns.
  base::ConditionVariable worker_released_condition_;
100 101

  std::atomic<uint32_t> assigned_task_ids_{0};
102 103 104 105 106 107 108
};

class V8_PLATFORM_EXPORT DefaultJobHandle : public JobHandle {
 public:
  explicit DefaultJobHandle(std::shared_ptr<DefaultJobState> state);
  ~DefaultJobHandle() override;

109 110 111
  DefaultJobHandle(const DefaultJobHandle&) = delete;
  DefaultJobHandle& operator=(const DefaultJobHandle&) = delete;

112 113 114 115 116 117
  void NotifyConcurrencyIncrease() override {
    state_->NotifyConcurrencyIncrease();
  }

  void Join() override;
  void Cancel() override;
118
  void CancelAndDetach() override;
119
  bool IsActive() override;
120
  bool IsValid() override { return state_ != nullptr; }
121

122 123 124 125
  bool UpdatePriorityEnabled() const override { return true; }

  void UpdatePriority(TaskPriority) override;

126 127 128 129 130 131 132 133 134 135
 private:
  std::shared_ptr<DefaultJobState> state_;
};

class DefaultJobWorker : public Task {
 public:
  DefaultJobWorker(std::weak_ptr<DefaultJobState> state, JobTask* job_task)
      : state_(std::move(state)), job_task_(job_task) {}
  ~DefaultJobWorker() override = default;

136 137 138
  DefaultJobWorker(const DefaultJobWorker&) = delete;
  DefaultJobWorker& operator=(const DefaultJobWorker&) = delete;

139 140 141 142 143
  void Run() override {
    auto shared_state = state_.lock();
    if (!shared_state) return;
    if (!shared_state->CanRunFirstTask()) return;
    do {
144 145 146
      // Scope of |delegate| must not outlive DidRunTask() so that associated
      // state is freed before the worker becomes inactive.
      DefaultJobState::JobDelegate delegate(shared_state.get());
147
      job_task_->Run(&delegate);
148 149 150 151 152 153 154 155 156 157 158 159 160 161
    } while (shared_state->DidRunTask());
  }

 private:
  friend class DefaultJob;

  std::weak_ptr<DefaultJobState> state_;
  JobTask* job_task_;
};

}  // namespace platform
}  // namespace v8

#endif  // V8_LIBPLATFORM_DEFAULT_JOB_H_