fml::MessageLoopTaskQueues Class Reference

#include <message_loop_task_queues.h>

Public Member Functions
TaskQueueId	CreateTaskQueue ()
void	Dispose (TaskQueueId queue_id)
void	DisposeTasks (TaskQueueId queue_id)
void	RegisterTask (TaskQueueId queue_id, const fml::closure &task, fml::TimePoint target_time, fml::TaskSourceGrade task_source_grade=fml::TaskSourceGrade::kUnspecified)
bool	HasPendingTasks (TaskQueueId queue_id) const
fml::closure	GetNextTaskToRun (TaskQueueId queue_id, fml::TimePoint from_time)
size_t	GetNumPendingTasks (TaskQueueId queue_id) const
void	AddTaskObserver (TaskQueueId queue_id, intptr_t key, const fml::closure &callback)
void	RemoveTaskObserver (TaskQueueId queue_id, intptr_t key)
std::vector< fml::closure >	GetObserversToNotify (TaskQueueId queue_id) const
void	SetWakeable (TaskQueueId queue_id, fml::Wakeable *wakeable)
bool	Merge (TaskQueueId owner, TaskQueueId subsumed)
bool	Unmerge (TaskQueueId owner, TaskQueueId subsumed)
bool	Owns (TaskQueueId owner, TaskQueueId subsumed) const
	Returns true if owner owns the subsumed task queue.
std::set< TaskQueueId >	GetSubsumedTaskQueueId (TaskQueueId owner) const
void	PauseSecondarySource (TaskQueueId queue_id)
void	ResumeSecondarySource (TaskQueueId queue_id)

Static Public Member Functions
static MessageLoopTaskQueues *	GetInstance ()
static TaskSourceGrade	GetCurrentTaskSourceGrade ()

Detailed Description

A singleton container for all tasks and observers associated with all fml::MessageLoops.

This also wakes up the loop at the required times.

See also

fml::MessageLoop

fml::Wakeable

Definition at line 65 of file message_loop_task_queues.h.

Member Function Documentation

AddTaskObserver()

void fml::MessageLoopTaskQueues::AddTaskObserver	(	TaskQueueId	queue_id,
		intptr_t	key,
		const fml::closure &	callback
	)

Definition at line 167 of file message_loop_task_queues.cc.

                                                                        {
  std::lock_guard guard(queue_mutex_);
  FML_DCHECK(callback != nullptr) << "Observer callback must be non-null.";
  queue_entries_.at(queue_id)->task_observers[key] = callback;
}

References callback, FML_DCHECK, and key.

Referenced by fml::MessageLoopImpl::AddTaskObserver(), flutter::testing::ShellTest::CreateSettingsForFixture(), FlutterDartProject::defaultBundleIdentifier, and main().

CreateTaskQueue()

TaskQueueId fml::MessageLoopTaskQueues::CreateTaskQueue

(

)

Definition at line 50 of file message_loop_task_queues.cc.

                                                   {
  std::lock_guard guard(queue_mutex_);
  TaskQueueId loop_id = TaskQueueId(task_queue_id_counter_);
  ++task_queue_id_counter_;
  queue_entries_[loop_id] = std::make_unique<TaskQueueEntry>(loop_id);
  return loop_id;
}

Dispose()

void fml::MessageLoopTaskQueues::Dispose

(

TaskQueueId

queue_id

)

Definition at line 65 of file message_loop_task_queues.cc.

                                                        {
  std::lock_guard guard(queue_mutex_);
  const auto& queue_entry = queue_entries_.at(queue_id);
  FML_DCHECK(queue_entry->subsumed_by == kUnmerged);
  auto& subsumed_set = queue_entry->owner_of;
  for (auto& subsumed : subsumed_set) {
    queue_entries_.erase(subsumed);
  }
  // Erase owner queue_id at last to avoid &subsumed_set from being invalid
  queue_entries_.erase(queue_id);
}

References FML_DCHECK, and fml::kUnmerged.

Referenced by fml::MessageLoopImpl::~MessageLoopImpl().

DisposeTasks()

void fml::MessageLoopTaskQueues::DisposeTasks

(

TaskQueueId

queue_id

)

Definition at line 77 of file message_loop_task_queues.cc.

                                                             {
  std::lock_guard guard(queue_mutex_);
  const auto& queue_entry = queue_entries_.at(queue_id);
  FML_DCHECK(queue_entry->subsumed_by == kUnmerged);
  auto& subsumed_set = queue_entry->owner_of;
  queue_entry->task_source->ShutDown();
  for (auto& subsumed : subsumed_set) {
    queue_entries_.at(subsumed)->task_source->ShutDown();
  }
}

References FML_DCHECK, and fml::kUnmerged.

Referenced by fml::MessageLoopImpl::DoRun().

GetCurrentTaskSourceGrade()

TaskSourceGrade fml::MessageLoopTaskQueues::GetCurrentTaskSourceGrade ( )

static

Definition at line 88 of file message_loop_task_queues.cc.

                                                                 {
  return tls_task_source_grade.get()->task_source_grade;
}

References fml::tls_task_source_grade.

Referenced by flutter::testing::TEST_F().

GetInstance()

MessageLoopTaskQueues * fml::MessageLoopTaskQueues::GetInstance ( )

static

Definition at line 45 of file message_loop_task_queues.cc.

                                                          {
  static MessageLoopTaskQueues* instance = new MessageLoopTaskQueues;
  return instance;
}

References instance.

Referenced by fml::benchmarking::BM_RegisterAndGetTasks(), flutter::testing::ShellTest::CreateSettingsForFixture(), FlutterDartProject::defaultBundleIdentifier, main(), flutter::Engine::Run(), fml::TaskRunnerChecker::RunsOnTheSameThread(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), fml::testing::TEST(), and fml::testing::TestNotifyObservers().

GetNextTaskToRun()

fml::closure fml::MessageLoopTaskQueues::GetNextTaskToRun	(	TaskQueueId	queue_id,
		fml::TimePoint	from_time
	)

Definition at line 118 of file message_loop_task_queues.cc.

                                                                             {
  std::lock_guard guard(queue_mutex_);
  if (!HasPendingTasksUnlocked(queue_id)) {
    return nullptr;
  }
  TaskSource::TopTask top = PeekNextTaskUnlocked(queue_id);
 
  if (!HasPendingTasksUnlocked(queue_id)) {
    WakeUpUnlocked(queue_id, fml::TimePoint::Max());
  } else {
    WakeUpUnlocked(queue_id, GetNextWakeTimeUnlocked(queue_id));
  }
 
  if (top.task.GetTargetTime() > from_time) {
    return nullptr;
  }
  fml::closure invocation = top.task.GetTask();
  const auto task_source_grade = top.task.GetTaskSourceGrade();
  queue_entries_.at(top.task_queue_id)->task_source->PopTask(task_source_grade);
  tls_task_source_grade.reset(new TaskSourceGradeHolder{task_source_grade});
  return invocation;
}

References fml::DelayedTask::GetTargetTime(), fml::DelayedTask::GetTask(), fml::DelayedTask::GetTaskSourceGrade(), fml::TimePoint::Max(), fml::TaskSource::TopTask::task, fml::TaskSource::TopTask::task_queue_id, and fml::tls_task_source_grade.

Referenced by fml::testing::CountRemainingTasks().

GetNumPendingTasks()

size_t fml::MessageLoopTaskQueues::GetNumPendingTasks

(

TaskQueueId

queue_id

)

const

Definition at line 149 of file message_loop_task_queues.cc.

                                                                           {
  std::lock_guard guard(queue_mutex_);
  const auto& queue_entry = queue_entries_.at(queue_id);
  if (queue_entry->subsumed_by != kUnmerged) {
    return 0;
  }
 
  size_t total_tasks = 0;
  total_tasks += queue_entry->task_source->GetNumPendingTasks();
 
  auto& subsumed_set = queue_entry->owner_of;
  for (auto& subsumed : subsumed_set) {
    const auto& subsumed_entry = queue_entries_.at(subsumed);
    total_tasks += subsumed_entry->task_source->GetNumPendingTasks();
  }
  return total_tasks;
}

References fml::kUnmerged.

GetObserversToNotify()

std::vector< fml::closure > fml::MessageLoopTaskQueues::GetObserversToNotify

(

TaskQueueId

queue_id

)

const

Definition at line 181 of file message_loop_task_queues.cc.

                                {
  std::lock_guard guard(queue_mutex_);
  std::vector<fml::closure> observers;
 
  if (queue_entries_.at(queue_id)->subsumed_by != kUnmerged) {
    return observers;
  }
 
  for (const auto& observer : queue_entries_.at(queue_id)->task_observers) {
    observers.push_back(observer.second);
  }
 
  auto& subsumed_set = queue_entries_.at(queue_id)->owner_of;
  for (auto& subsumed : subsumed_set) {
    for (const auto& observer : queue_entries_.at(subsumed)->task_observers) {
      observers.push_back(observer.second);
    }
  }
 
  return observers;
}

References fml::kUnmerged.

GetSubsumedTaskQueueId()

std::set< TaskQueueId > fml::MessageLoopTaskQueues::GetSubsumedTaskQueueId

(

TaskQueueId

owner

)

const

Definition at line 316 of file message_loop_task_queues.cc.

                             {
  std::lock_guard guard(queue_mutex_);
  return queue_entries_.at(owner)->owner_of;
}

HasPendingTasks()

bool fml::MessageLoopTaskQueues::HasPendingTasks

(

TaskQueueId

queue_id

)

const

Definition at line 113 of file message_loop_task_queues.cc.

                                                                      {
  std::lock_guard guard(queue_mutex_);
  return HasPendingTasksUnlocked(queue_id);
}

Merge()

bool fml::MessageLoopTaskQueues::Merge	(	TaskQueueId	owner,
		TaskQueueId	subsumed
	)

Definition at line 212 of file message_loop_task_queues.cc.

                                                                         {
  if (owner == subsumed) {
    return true;
  }
  std::lock_guard guard(queue_mutex_);
  auto& owner_entry = queue_entries_.at(owner);
  auto& subsumed_entry = queue_entries_.at(subsumed);
  auto& subsumed_set = owner_entry->owner_of;
  if (subsumed_set.find(subsumed) != subsumed_set.end()) {
    return true;
  }
 
  // Won't check owner_entry->owner_of, because it may contains items when
  // merged with other different queues.
 
  // Ensure owner_entry->subsumed_by being kUnmerged
  if (owner_entry->subsumed_by != kUnmerged) {
    FML_LOG(WARNING) << "Thread merging failed: owner_entry was already "
                        "subsumed by others, owner="
                     << owner << ", subsumed=" << subsumed
                     << ", owner->subsumed_by=" << owner_entry->subsumed_by;
    return false;
  }
  // Ensure subsumed_entry->owner_of being empty
  if (!subsumed_entry->owner_of.empty()) {
    FML_LOG(WARNING)
        << "Thread merging failed: subsumed_entry already owns others, owner="
        << owner << ", subsumed=" << subsumed
        << ", subsumed->owner_of.size()=" << subsumed_entry->owner_of.size();
    return false;
  }
  // Ensure subsumed_entry->subsumed_by being kUnmerged
  if (subsumed_entry->subsumed_by != kUnmerged) {
    FML_LOG(WARNING) << "Thread merging failed: subsumed_entry was already "
                        "subsumed by others, owner="
                     << owner << ", subsumed=" << subsumed
                     << ", subsumed->subsumed_by="
                     << subsumed_entry->subsumed_by;
    return false;
  }
  // All checking is OK, set merged state.
  owner_entry->owner_of.insert(subsumed);
  subsumed_entry->subsumed_by = owner;
 
  if (HasPendingTasksUnlocked(owner)) {
    WakeUpUnlocked(owner, GetNextWakeTimeUnlocked(owner));
  }
 
  return true;
}

References FML_LOG, and fml::kUnmerged.

Referenced by fml::SharedThreadMerger::MergeWithLease(), flutter::Engine::Run(), and fml::testing::TEST().

Owns()

bool fml::MessageLoopTaskQueues::Owns	(	TaskQueueId	owner,
		TaskQueueId	subsumed
	)		const

Returns true if owner owns the subsumed task queue.

Definition at line 306 of file message_loop_task_queues.cc.

                                                             {
  std::lock_guard guard(queue_mutex_);
  if (owner == kUnmerged || subsumed == kUnmerged) {
    return false;
  }
  auto& subsumed_set = queue_entries_.at(owner)->owner_of;
  return subsumed_set.find(subsumed) != subsumed_set.end();
}

References fml::kUnmerged.

PauseSecondarySource()

void fml::MessageLoopTaskQueues::PauseSecondarySource

(

TaskQueueId

queue_id

)

Definition at line 322 of file message_loop_task_queues.cc.

                                                                     {
  std::lock_guard guard(queue_mutex_);
  queue_entries_.at(queue_id)->task_source->PauseSecondary();
}

RegisterTask()

void fml::MessageLoopTaskQueues::RegisterTask	(	TaskQueueId	queue_id,
		const fml::closure &	task,
		fml::TimePoint	target_time,
		fml::TaskSourceGrade	task_source_grade = fml::TaskSourceGrade::kUnspecified
	)

Definition at line 92 of file message_loop_task_queues.cc.

                                          {
  std::lock_guard guard(queue_mutex_);
  size_t order = order_++;
  const auto& queue_entry = queue_entries_.at(queue_id);
  queue_entry->task_source->RegisterTask(
      {order, task, target_time, task_source_grade});
  TaskQueueId loop_to_wake = queue_id;
  if (queue_entry->subsumed_by != kUnmerged) {
    loop_to_wake = queue_entry->subsumed_by;
  }
 
  // This can happen when the secondary tasks are paused.
  if (HasPendingTasksUnlocked(loop_to_wake)) {
    WakeUpUnlocked(loop_to_wake, GetNextWakeTimeUnlocked(loop_to_wake));
  }
}

References fml::kUnmerged.

Referenced by fml::MessageLoopImpl::PostTask().

RemoveTaskObserver()

void fml::MessageLoopTaskQueues::RemoveTaskObserver	(	TaskQueueId	queue_id,
		intptr_t	key
	)

Definition at line 175 of file message_loop_task_queues.cc.

                                                             {
  std::lock_guard guard(queue_mutex_);
  queue_entries_.at(queue_id)->task_observers.erase(key);
}

References key.

Referenced by flutter::testing::ShellTest::CreateSettingsForFixture(), FlutterDartProject::defaultBundleIdentifier, main(), and fml::MessageLoopImpl::RemoveTaskObserver().

ResumeSecondarySource()

void fml::MessageLoopTaskQueues::ResumeSecondarySource

(

TaskQueueId

queue_id

)

Definition at line 327 of file message_loop_task_queues.cc.

                                                                      {
  std::lock_guard guard(queue_mutex_);
  queue_entries_.at(queue_id)->task_source->ResumeSecondary();
  // Schedule a wake as needed.
  if (HasPendingTasksUnlocked(queue_id)) {
    WakeUpUnlocked(queue_id, GetNextWakeTimeUnlocked(queue_id));
  }
}

SetWakeable()

void fml::MessageLoopTaskQueues::SetWakeable	(	TaskQueueId	queue_id,
		fml::Wakeable *	wakeable
	)

Definition at line 204 of file message_loop_task_queues.cc.

                                                               {
  std::lock_guard guard(queue_mutex_);
  FML_CHECK(!queue_entries_.at(queue_id)->wakeable)
      << "Wakeable can only be set once.";
  queue_entries_.at(queue_id)->wakeable = wakeable;
}

References FML_CHECK.

Referenced by fml::MessageLoopImpl::MessageLoopImpl().

Unmerge()

bool fml::MessageLoopTaskQueues::Unmerge	(	TaskQueueId	owner,
		TaskQueueId	subsumed
	)

Definition at line 263 of file message_loop_task_queues.cc.

                                                                           {
  std::lock_guard guard(queue_mutex_);
  const auto& owner_entry = queue_entries_.at(owner);
  if (owner_entry->owner_of.empty()) {
    FML_LOG(WARNING)
        << "Thread unmerging failed: owner_entry doesn't own anyone, owner="
        << owner << ", subsumed=" << subsumed;
    return false;
  }
  if (owner_entry->subsumed_by != kUnmerged) {
    FML_LOG(WARNING)
        << "Thread unmerging failed: owner_entry was subsumed by others, owner="
        << owner << ", subsumed=" << subsumed
        << ", owner_entry->subsumed_by=" << owner_entry->subsumed_by;
    return false;
  }
  if (queue_entries_.at(subsumed)->subsumed_by == kUnmerged) {
    FML_LOG(WARNING) << "Thread unmerging failed: subsumed_entry wasn't "
                        "subsumed by others, owner="
                     << owner << ", subsumed=" << subsumed;
    return false;
  }
  if (owner_entry->owner_of.find(subsumed) == owner_entry->owner_of.end()) {
    FML_LOG(WARNING) << "Thread unmerging failed: owner_entry didn't own the "
                        "given subsumed queue id, owner="
                     << owner << ", subsumed=" << subsumed;
    return false;
  }
 
  queue_entries_.at(subsumed)->subsumed_by = kUnmerged;
  owner_entry->owner_of.erase(subsumed);
 
  if (HasPendingTasksUnlocked(owner)) {
    WakeUpUnlocked(owner, GetNextWakeTimeUnlocked(owner));
  }
 
  if (HasPendingTasksUnlocked(subsumed)) {
    WakeUpUnlocked(subsumed, GetNextWakeTimeUnlocked(subsumed));
  }
 
  return true;
}

References FML_LOG, and fml::kUnmerged.

Referenced by fml::testing::TEST().

---

The documentation for this class was generated from the following files:

• fml/message_loop_task_queues.h
• fml/message_loop_task_queues.cc