1
    2
    3
    4
    5
    6
    7
    8
    9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24
   25
   26
   27
   28
   29
   30
   31
   32
   33
   34
   35
   36
   37
   38
   39
   40
   41
   42
   43
   44
   45
   46
   47
   48
   49
   50
   51
   52
   53
   54
   55
   56
   57
   58
   59
   60
   61
   62
   63
   64
   65
   66
   67
   68
   69
   70
   71
   72
   73
   74
   75
   76
   77
   78
   79
   80
   81
   82
   83
   84
   85
   86
   87
   88
   89
   90
   91
   92
   93
   94
   95
   96
   97
   98
   99
  100
  101
  102
  103
  104
  105
  106
  107
  108
  109
  110
  111
  112
  113
  114
  115
  116
  117
  118
  119
  120
  121
  122
  123
  124
  125
  126
  127
  128
  129
  130
  131
  132
  133
  134
  135
  136
  137
  138
  139
  140
  141
  142
  143
  144
  145
  146
  147
  148
  149
  150
  151
  152
  153
  154
  155
  156
  157
  158
  159
  160
  161
  162
  163
  164
  165
  166
  167
  168
  169
  170
  171
  172
  173
  174
  175
  176
  177
  178
  179
  180
  181
  182
  183
  184
  185
  186
  187
  188
  189

base / task / thread_pool / delayed_priority_queue.cc [blame]

// Copyright 2022 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "base/task/thread_pool/delayed_priority_queue.h"

#include <utility>

#include "base/check_op.h"
#include "base/memory/ptr_util.h"
#include "base/stl_util.h"

namespace base::internal {

// A class combining a TaskSource and the delayed_sort_key that determines its
// position in a DelayedPriorityQueue. Instances are only mutable via
// take_task_source() which can only be called once and renders its instance
// invalid after the call.
class DelayedPriorityQueue::TaskSourceAndDelayedSortKey {
 public:
  TaskSourceAndDelayedSortKey() = default;
  TaskSourceAndDelayedSortKey(scoped_refptr<TaskSource> task_source,
                              const TimeTicks& delayed_sort_key)
      : task_source_(std::move(task_source)),
        delayed_sort_key_(delayed_sort_key) {
    DCHECK(task_source_);
  }
  TaskSourceAndDelayedSortKey(const TaskSourceAndDelayedSortKey&) = delete;
  TaskSourceAndDelayedSortKey& operator=(const TaskSourceAndDelayedSortKey&) =
      delete;

  // Note: while |task_source_| should always be non-null post-move (i.e. we
  // shouldn't be moving an invalid TaskSourceAndDelayedSortKey around), there
  // can't be a DCHECK(task_source_) on moves as IntrusiveHeap moves elements on
  // pop instead of overwriting them: resulting in the move of a
  // TaskSourceAndDelayedSortKey with a null |task_source_| in
  // Transaction::Pop()'s implementation.
  TaskSourceAndDelayedSortKey(TaskSourceAndDelayedSortKey&& other) = default;
  TaskSourceAndDelayedSortKey& operator=(TaskSourceAndDelayedSortKey&& other) =
      default;

  // Extracts |task_source_| from this object. This object is invalid after this
  // call.
  scoped_refptr<TaskSource> take_task_source() {
    DCHECK(task_source_);
    task_source_->ClearDelayedHeapHandle();
    return std::move(task_source_);
  }

  // Compares this TaskSourceAndDelayedSortKey to |other| based on their
  // respective |delayed_sort_key_|. Used for a max-heap.
  bool operator<(const TaskSourceAndDelayedSortKey& other) const {
    return delayed_sort_key_ < other.delayed_sort_key_;
  }

  // Required by IntrusiveHeap.
  void SetHeapHandle(const HeapHandle& handle) {
    DCHECK(task_source_);
    task_source_->SetDelayedHeapHandle(handle);
  }

  // Required by IntrusiveHeap.
  void ClearHeapHandle() {
    // Ensure |task_source_| is not nullptr, which may be the case if
    // take_task_source() was called before this.
    if (task_source_)
      task_source_->ClearDelayedHeapHandle();
  }

  // Required by IntrusiveHeap.
  HeapHandle GetHeapHandle() const {
    if (task_source_)
      return task_source_->GetDelayedHeapHandle();
    return HeapHandle::Invalid();
  }

  scoped_refptr<TaskSource> task_source() const { return task_source_; }

  TimeTicks delayed_sort_key() const { return delayed_sort_key_; }

 private:
  scoped_refptr<TaskSource> task_source_;
  TimeTicks delayed_sort_key_;
};

DelayedPriorityQueue::DelayedPriorityQueue() = default;

DelayedPriorityQueue::~DelayedPriorityQueue() {
  if (!is_flush_task_sources_on_destroy_enabled_)
    return;

  while (!container_.empty()) {
    auto task_source = PopTaskSource();
    task_source->ClearForTesting();  // IN-TEST
  }
}

DelayedPriorityQueue& DelayedPriorityQueue::operator=(
    DelayedPriorityQueue&& other) = default;

void DelayedPriorityQueue::Push(scoped_refptr<TaskSource> task_source,
                                TimeTicks task_source_delayed_sort_key) {
  container_.insert(TaskSourceAndDelayedSortKey(std::move(task_source),
                                                task_source_delayed_sort_key));
}

const TimeTicks DelayedPriorityQueue::PeekDelayedSortKey() const {
  DCHECK(!IsEmpty());
  return container_.top().delayed_sort_key();
}

scoped_refptr<TaskSource> DelayedPriorityQueue::PeekTaskSource() const {
  DCHECK(!IsEmpty());

  auto& task_source_and_delayed_sort_key = container_.top();
  return task_source_and_delayed_sort_key.task_source();
}

scoped_refptr<TaskSource> DelayedPriorityQueue::PopTaskSource() {
  DCHECK(!IsEmpty());

  auto task_source_and_delayed_sort_key = container_.take_top();
  scoped_refptr<TaskSource> task_source =
      task_source_and_delayed_sort_key.take_task_source();
  return task_source;
}

scoped_refptr<TaskSource> DelayedPriorityQueue::RemoveTaskSource(
    scoped_refptr<TaskSource> task_source) {
  if (IsEmpty())
    return nullptr;

  const HeapHandle heap_handle = task_source->delayed_heap_handle();
  if (!heap_handle.IsValid())
    return nullptr;

  TaskSourceAndDelayedSortKey& task_source_and_delayed_sort_key =
      const_cast<DelayedPriorityQueue::TaskSourceAndDelayedSortKey&>(
          container_.at(heap_handle));
  DCHECK_EQ(task_source_and_delayed_sort_key.task_source(), task_source);
  task_source = task_source_and_delayed_sort_key.take_task_source();

  container_.erase(heap_handle);
  return task_source;
}

void DelayedPriorityQueue::UpdateDelayedSortKey(
    scoped_refptr<TaskSource> task_source) {
  if (IsEmpty())
    return;

  const HeapHandle heap_handle = task_source->delayed_heap_handle();
  if (!heap_handle.IsValid())
    return;

  DCHECK_EQ(container_.at(heap_handle).task_source(), task_source);

  task_source = const_cast<DelayedPriorityQueue::TaskSourceAndDelayedSortKey&>(
                    container_.at(heap_handle))
                    .take_task_source();
  auto delayed_sort_key = task_source->GetDelayedSortKey();
  container_.Replace(
      heap_handle,
      TaskSourceAndDelayedSortKey(std::move(task_source), delayed_sort_key));
}

bool DelayedPriorityQueue::IsEmpty() const {
  return container_.empty();
}

size_t DelayedPriorityQueue::Size() const {
  return container_.size();
}

void DelayedPriorityQueue::EnableFlushTaskSourcesOnDestroyForTesting() {
  DCHECK(!is_flush_task_sources_on_destroy_enabled_);
  is_flush_task_sources_on_destroy_enabled_ = true;
}

// Delayed tasks are ordered by latest_delayed_run_time(). The top task may
// not be the first task eligible to run, but tasks will always become ripe
// before their latest_delayed_run_time().
bool DelayedPriorityQueue::DelayedPriorityQueueComparisonOperator::operator()(
    const TaskSourceAndDelayedSortKey& lhs,
    const TaskSourceAndDelayedSortKey& rhs) const {
  return lhs.delayed_sort_key() > rhs.delayed_sort_key();
}

}  // namespace base::internal