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

#include "cc/resources/worker_pool.h"

#include <algorithm>
#include <queue>

#include "base/bind.h"
#include "base/containers/hash_tables.h"
#include "base/debug/trace_event.h"
#include "base/strings/stringprintf.h"
#include "base/synchronization/condition_variable.h"
#include "base/threading/simple_thread.h"
#include "base/threading/thread_restrictions.h"
#include "cc/base/scoped_ptr_deque.h"

namespace cc {

namespace internal {

WorkerPoolTask::WorkerPoolTask()
    : did_schedule_(false),
      did_run_(false),
      did_complete_(false) {
}

WorkerPoolTask::~WorkerPoolTask() {
  DCHECK_EQ(did_schedule_, did_complete_);
  DCHECK(!did_run_ || did_schedule_);
  DCHECK(!did_run_ || did_complete_);
}

void WorkerPoolTask::DidSchedule() {
  DCHECK(!did_complete_);
  did_schedule_ = true;
}

void WorkerPoolTask::WillRun() {
  DCHECK(did_schedule_);
  DCHECK(!did_complete_);
  DCHECK(!did_run_);
}

void WorkerPoolTask::DidRun() {
  did_run_ = true;
}

void WorkerPoolTask::WillComplete() {
  DCHECK(!did_complete_);
}

void WorkerPoolTask::DidComplete() {
  DCHECK(did_schedule_);
  DCHECK(!did_complete_);
  did_complete_ = true;
}

bool WorkerPoolTask::HasFinishedRunning() const {
  return did_run_;
}

bool WorkerPoolTask::HasCompleted() const {
  return did_complete_;
}

GraphNode::GraphNode(internal::WorkerPoolTask* task, unsigned priority)
    : task_(task),
      priority_(priority),
      num_dependencies_(0) {
}

GraphNode::~GraphNode() {
}

}  // namespace internal

// Internal to the worker pool. Any data or logic that needs to be
// shared between threads lives in this class. All members are guarded
// by |lock_|.
class WorkerPool::Inner : public base::DelegateSimpleThread::Delegate {
 public:
  Inner(size_t num_threads, const std::string& thread_name_prefix);
  virtual ~Inner();

  void Shutdown();

  // Schedule running of tasks in |graph|. Tasks previously scheduled but
  // no longer needed will be canceled unless already running. Canceled
  // tasks are moved to |completed_tasks_| without being run. The result
  // is that once scheduled, a task is guaranteed to end up in the
  // |completed_tasks_| queue even if they later get canceled by another
  // call to SetTaskGraph().
  void SetTaskGraph(TaskGraph* graph);

  // Collect all completed tasks in |completed_tasks|.
  void CollectCompletedTasks(TaskVector* completed_tasks);

 private:
  class PriorityComparator {
   public:
    bool operator()(const internal::GraphNode* a,
                    const internal::GraphNode* b) {
      // In this system, numerically lower priority is run first.
      if (a->priority() != b->priority())
        return a->priority() > b->priority();

      // Run task with most dependents first when priority is the same.
      return a->dependents().size() < b->dependents().size();
    }
  };

  // Overridden from base::DelegateSimpleThread:
  virtual void Run() OVERRIDE;

  // This lock protects all members of this class except
  // |worker_pool_on_origin_thread_|. Do not read or modify anything
  // without holding this lock. Do not block while holding this lock.
  mutable base::Lock lock_;

  // Condition variable that is waited on by worker threads until new
  // tasks are ready to run or shutdown starts.
  base::ConditionVariable has_ready_to_run_tasks_cv_;

  // Provides each running thread loop with a unique index. First thread
  // loop index is 0.
  unsigned next_thread_index_;

  // Set during shutdown. Tells workers to exit when no more tasks
  // are pending.
  bool shutdown_;

  // This set contains all pending tasks.
  GraphNodeMap pending_tasks_;

  // Ordered set of tasks that are ready to run.
  typedef std::priority_queue<internal::GraphNode*,
                              std::vector<internal::GraphNode*>,
                              PriorityComparator> TaskQueue;
  TaskQueue ready_to_run_tasks_;

  // This set contains all currently running tasks.
  GraphNodeMap running_tasks_;

  // Completed tasks not yet collected by origin thread.
  TaskVector completed_tasks_;

  ScopedPtrDeque<base::DelegateSimpleThread> workers_;

  DISALLOW_COPY_AND_ASSIGN(Inner);
};

WorkerPool::Inner::Inner(
    size_t num_threads, const std::string& thread_name_prefix)
    : lock_(),
      has_ready_to_run_tasks_cv_(&lock_),
      next_thread_index_(0),
      shutdown_(false) {
  base::AutoLock lock(lock_);

  while (workers_.size() < num_threads) {
    scoped_ptr<base::DelegateSimpleThread> worker = make_scoped_ptr(
        new base::DelegateSimpleThread(
            this,
            thread_name_prefix +
            base::StringPrintf(
                "Worker%u",
                static_cast<unsigned>(workers_.size() + 1)).c_str()));
    worker->Start();
#if defined(OS_ANDROID) || defined(OS_LINUX)
    worker->SetThreadPriority(base::kThreadPriority_Background);
#endif
    workers_.push_back(worker.Pass());
  }
}

WorkerPool::Inner::~Inner() {
  base::AutoLock lock(lock_);

  DCHECK(shutdown_);

  DCHECK_EQ(0u, pending_tasks_.size());
  DCHECK_EQ(0u, ready_to_run_tasks_.size());
  DCHECK_EQ(0u, running_tasks_.size());
  DCHECK_EQ(0u, completed_tasks_.size());
}

void WorkerPool::Inner::Shutdown() {
  {
    base::AutoLock lock(lock_);

    DCHECK(!shutdown_);
    shutdown_ = true;

    // Wake up a worker so it knows it should exit. This will cause all workers
    // to exit as each will wake up another worker before exiting.
    has_ready_to_run_tasks_cv_.Signal();
  }

  while (workers_.size()) {
    scoped_ptr<base::DelegateSimpleThread> worker = workers_.take_front();
    // http://crbug.com/240453 - Join() is considered IO and will block this
    // thread. See also http://crbug.com/239423 for further ideas.
    base::ThreadRestrictions::ScopedAllowIO allow_io;
    worker->Join();
  }
}

void WorkerPool::Inner::SetTaskGraph(TaskGraph* graph) {
  // It is OK to call SetTaskGraph() after shutdown if |graph| is empty.
  DCHECK(graph->empty() || !shutdown_);

  GraphNodeMap new_pending_tasks;
  GraphNodeMap new_running_tasks;
  TaskQueue new_ready_to_run_tasks;

  new_pending_tasks.swap(*graph);

  {
    base::AutoLock lock(lock_);

    // First remove all completed tasks from |new_pending_tasks| and
    // adjust number of dependencies.
    for (TaskVector::iterator it = completed_tasks_.begin();
         it != completed_tasks_.end(); ++it) {
      internal::WorkerPoolTask* task = it->get();

      scoped_ptr<internal::GraphNode> node = new_pending_tasks.take_and_erase(
          task);
      if (node) {
        for (internal::GraphNode::Vector::const_iterator it =
                 node->dependents().begin();
             it != node->dependents().end(); ++it) {
          internal::GraphNode* dependent_node = *it;
          dependent_node->remove_dependency();
        }
      }
    }

    // Build new running task set.
    for (GraphNodeMap::iterator it = running_tasks_.begin();
         it != running_tasks_.end(); ++it) {
      internal::WorkerPoolTask* task = it->first;
      // Transfer scheduled task value from |new_pending_tasks| to
      // |new_running_tasks| if currently running. Value must be set to
      // NULL if |new_pending_tasks| doesn't contain task. This does
      // the right in both cases.
      new_running_tasks.set(task, new_pending_tasks.take_and_erase(task));
    }

    // Build new "ready to run" tasks queue.
    // TODO(reveman): Create this queue when building the task graph instead.
    for (GraphNodeMap::iterator it = new_pending_tasks.begin();
         it != new_pending_tasks.end(); ++it) {
      internal::WorkerPoolTask* task = it->first;
      DCHECK(task);
      internal::GraphNode* node = it->second;

      // Completed tasks should not exist in |new_pending_tasks|.
      DCHECK(!task->HasFinishedRunning());

      // Call DidSchedule() to indicate that this task has been scheduled.
      // Note: This is only for debugging purposes.
      task->DidSchedule();

      if (!node->num_dependencies())
        new_ready_to_run_tasks.push(node);

      // Erase the task from old pending tasks.
      pending_tasks_.erase(task);
    }

    completed_tasks_.reserve(completed_tasks_.size() + pending_tasks_.size());

    // The items left in |pending_tasks_| need to be canceled.
    for (GraphNodeMap::const_iterator it = pending_tasks_.begin();
         it != pending_tasks_.end();
         ++it) {
      completed_tasks_.push_back(it->first);
    }

    // Swap task sets.
    // Note: old tasks are intentionally destroyed after releasing |lock_|.
    pending_tasks_.swap(new_pending_tasks);
    running_tasks_.swap(new_running_tasks);
    std::swap(ready_to_run_tasks_, new_ready_to_run_tasks);

    // If |ready_to_run_tasks_| is empty, it means we either have
    // running tasks, or we have no pending tasks.
    DCHECK(!ready_to_run_tasks_.empty() ||
           (pending_tasks_.empty() || !running_tasks_.empty()));

    // If there is more work available, wake up worker thread.
    if (!ready_to_run_tasks_.empty())
      has_ready_to_run_tasks_cv_.Signal();
  }
}

void WorkerPool::Inner::CollectCompletedTasks(TaskVector* completed_tasks) {
  base::AutoLock lock(lock_);

  DCHECK_EQ(0u, completed_tasks->size());
  completed_tasks->swap(completed_tasks_);
}

void WorkerPool::Inner::Run() {
  base::AutoLock lock(lock_);

  // Get a unique thread index.
  int thread_index = next_thread_index_++;

  while (true) {
    if (ready_to_run_tasks_.empty()) {
      // Exit when shutdown is set and no more tasks are pending.
      if (shutdown_ && pending_tasks_.empty())
        break;

      // Wait for more tasks.
      has_ready_to_run_tasks_cv_.Wait();
      continue;
    }

    // Take top priority task from |ready_to_run_tasks_|.
    scoped_refptr<internal::WorkerPoolTask> task(
        ready_to_run_tasks_.top()->task());
    ready_to_run_tasks_.pop();

    // Move task from |pending_tasks_| to |running_tasks_|.
    DCHECK(pending_tasks_.contains(task.get()));
    DCHECK(!running_tasks_.contains(task.get()));
    running_tasks_.set(task.get(), pending_tasks_.take_and_erase(task.get()));

    // There may be more work available, so wake up another worker thread.
    has_ready_to_run_tasks_cv_.Signal();

    // Call WillRun() before releasing |lock_| and running task.
    task->WillRun();

    {
      base::AutoUnlock unlock(lock_);

      task->RunOnWorkerThread(thread_index);
    }

    // This will mark task as finished running.
    task->DidRun();

    // Now iterate over all dependents to remove dependency and check
    // if they are ready to run.
    scoped_ptr<internal::GraphNode> node = running_tasks_.take_and_erase(
        task.get());
    if (node) {
      for (internal::GraphNode::Vector::const_iterator it =
               node->dependents().begin();
           it != node->dependents().end(); ++it) {
        internal::GraphNode* dependent_node = *it;

        dependent_node->remove_dependency();
        // Task is ready if it has no dependencies. Add it to
        // |ready_to_run_tasks_|.
        if (!dependent_node->num_dependencies())
          ready_to_run_tasks_.push(dependent_node);
      }
    }

    // Finally add task to |completed_tasks_|.
    completed_tasks_.push_back(task);
  }

  // We noticed we should exit. Wake up the next worker so it knows it should
  // exit as well (because the Shutdown() code only signals once).
  has_ready_to_run_tasks_cv_.Signal();
}

WorkerPool::WorkerPool(size_t num_threads,
                       const std::string& thread_name_prefix)
    : in_dispatch_completion_callbacks_(false),
      inner_(make_scoped_ptr(new Inner(num_threads, thread_name_prefix))) {
}

WorkerPool::~WorkerPool() {
}

void WorkerPool::Shutdown() {
  TRACE_EVENT0("cc", "WorkerPool::Shutdown");

  DCHECK(!in_dispatch_completion_callbacks_);

  inner_->Shutdown();
}

void WorkerPool::CheckForCompletedTasks() {
  TRACE_EVENT0("cc", "WorkerPool::CheckForCompletedTasks");

  DCHECK(!in_dispatch_completion_callbacks_);

  TaskVector completed_tasks;
  inner_->CollectCompletedTasks(&completed_tasks);
  ProcessCompletedTasks(completed_tasks);
}

void WorkerPool::ProcessCompletedTasks(
    const TaskVector& completed_tasks) {
  TRACE_EVENT1("cc", "WorkerPool::ProcessCompletedTasks",
               "completed_task_count", completed_tasks.size());

  // Worker pool instance is not reentrant while processing completed tasks.
  in_dispatch_completion_callbacks_ = true;

  for (TaskVector::const_iterator it = completed_tasks.begin();
       it != completed_tasks.end();
       ++it) {
    internal::WorkerPoolTask* task = it->get();

    task->WillComplete();
    task->CompleteOnOriginThread();
    task->DidComplete();
  }

  in_dispatch_completion_callbacks_ = false;
}

void WorkerPool::SetTaskGraph(TaskGraph* graph) {
  TRACE_EVENT1("cc", "WorkerPool::SetTaskGraph",
               "num_tasks", graph->size());

  DCHECK(!in_dispatch_completion_callbacks_);

  inner_->SetTaskGraph(graph);
}

}  // namespace cc