content: Add task queue manager

base/task/task_queue_manager.h

+// Copyright 2014 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.
+
+#ifndef BASE_TASK_TASK_QUEUE_MANAGER_H_
+#define BASE_TASK_TASK_QUEUE_MANAGER_H_
+
+#include "base/atomic_sequence_num.h"
+#include "base/debug/task_annotator.h"
+#include "base/macros.h"
+#include "base/memory/scoped_vector.h"
+#include "base/memory/weak_ptr.h"
+#include "base/pending_task.h"
+#include "base/single_thread_task_runner.h"
+#include "base/synchronization/lock.h"
+#include "base/threading/thread_checker.h"
+
+namespace base {
+class TaskQueueScheduler;
+
+// The task queue manager provides N task queues and a scheduling interface for
+// choosing which task queue to service next. Each task queue consists of two
+// sub queues:
+//
+// 1. Incoming task queue. Tasks that are posted get immediately appended here.
+//    When a task is appended into an empty incoming queue, the task manager
+//    work function (DoWork) is scheduled to run on the main task runner.
+//
+// 2. Work queue. If a work queue is empty when DoWork() is entered, tasks from
+//    the incoming task queue (if any) are moved here. The work queues are
+//    registered with the scheduler as input to the scheduling decision.
+//
+class BASE_EXPORT TaskQueueManager {
+ public:
+  // Create a task queue manager with |task_queue_count| task queues.
+  // |main_task_runner| identifies the thread on which where the tasks are
+  // eventually run. |scheduler| is used to choose which task queue to service.
+  // It should outlive this class.
+  TaskQueueManager(size_t task_queue_count,
+                   scoped_refptr<SingleThreadTaskRunner> main_task_runner,
+                   TaskQueueScheduler* scheduler);
+  ~TaskQueueManager();
+
+  // Returns the task runner which targets the queue selected by |queue_index|.
+  scoped_refptr<SingleThreadTaskRunner> TaskRunnerForQueue(size_t queue_index);
+
+  // If |auto_pump| is false, tasks posted to the given incoming queue will not
+  // be automatically scheduled for execution or transferred to the work queue.
+  // Instead, the scheduler should call PumpQueue() when necessary to bring in
+  // new tasks for execution.
+  void SetAutoPump(size_t queue_index, bool auto_pump);

[alexclarke, 2014/10/15 09:48:59]

I suspect this method will only ever get called once, should this flag should be
set via the constructor?

[Sami, 2014/10/15 12:05:24]

I'm thinking it might be useful to keep it toggleable, say, for a background
policy where idle work can happen whenever. It's not a huge overhead to keep it
that way.

+
+  // If the work queue is empty, reloads new tasks from the incoming queue.
+  // After this, this function ensures any tasks in the work queue are
+  // scheduled for execution.
+  void PumpQueue(size_t queue_index);

[petrcermak, 2014/10/15 10:05:40]

Shouldn't this method reload tasks whenever the incoming queue is non-empty
(independent of the emptiness of the work queue)? Otherwise, if some idle tasks
didn't finish in the previous idle period, newly scheduled idle tasks will be
ignored in the current one.

[Sami, 2014/10/15 12:05:24]

I think you're right. I had this this way because the reload is usually done
with a swap, but it doesn't really make sense to leak that implementation detail
to the caller.

Fixed and added a test.

+
+  // Returns true if there are any tasks in either the work or incoming task
+  // queue identified by |queue_index|.
+  bool PollQueue(size_t queue_index);

[alexclarke, 2014/10/15 09:48:59]

How about IsQueueEmpty()?

[Sami, 2014/10/15 12:05:24]

I prefer poll since it conveys that this operation has a cost and is not a
simple getter. I think I'll add a note about that.

+
+ private:
+  class TaskRunner : public SingleThreadTaskRunner {
+   public:
+    TaskRunner(TaskQueueManager* task_queue_manager, size_t queue_index);
+
+    // SingleThreadTaskRunner implementation.
+    virtual bool RunsTasksOnCurrentThread() const override;
+    virtual bool PostDelayedTask(const tracked_objects::Location& from_here,
+                                 const Closure& task,
+                                 TimeDelta delay) override;
+    virtual bool PostNonNestableDelayedTask(
+        const tracked_objects::Location& from_here,
+        const Closure& task,
+        TimeDelta delay) override;
+
+   private:
+    virtual ~TaskRunner();
+
+    TaskQueueManager* task_queue_manager_;
+    const size_t queue_index_;
+  };
+  friend class TaskRunner;
+
+  // Adds a task at the end of the incoming task queue for |queue_index| and
+  // schedules a call to DoWork() if the incoming queue was empty and automatic
+  // pumping is enabled. Can be called on an arbitrary thread.
+  void EnqueueTask(size_t queue_index, const PendingTask& pending_task);
+
+  // Post a task to call DoWork() on the main task runner.
+  void ScheduleWork();
+
+  // Use the scheduler to choose a pending task and run it.
+  void DoWork();
+
+  // Reloads the work queue identified by |queue_index| and returns true if
+  // it ended up having tasks. The work queue must be empty when this function
+  // is called.
+  bool ReloadWorkQueue(size_t queue_index);
+
+  // Returns true if any work queue is non-empty.
+  bool ReloadWorkQueues();
+
+  // Runs a single task from the work queue designated by |queue_index|. The
+  // queue must not be empty.
+  void RunTaskFromWorkQueue(size_t queue_index);
+
+  bool RunsTasksOnCurrentThread() const;
+  bool PostDelayedTask(size_t queue_index,
+                       const tracked_objects::Location& from_here,
+                       const Closure& task,
+                       TimeDelta delay);
+  bool PostNonNestableDelayedTask(size_t queue_index,
+                                  const tracked_objects::Location& from_here,
+                                  const Closure& task,
+                                  TimeDelta delay);
+
+  struct InternalTaskQueue {
+    InternalTaskQueue(size_t index);
+    ~InternalTaskQueue();
+
+    const size_t index;

[alexclarke, 2014/10/15 09:48:59]

The only reason for this appears to be because of the C++11 style ranged for
loop.  As much as I like C++11 style loops I think it would be better to get rid
of this :)

[Sami, 2014/10/15 12:05:24]

Heh, well spotted :) Removed.

+    scoped_refptr<TaskRunner> task_runner;
+
+    Lock incoming_queue_lock;
+    TaskQueue incoming_queue;
+
+    bool auto_pump;
+    TaskQueue work_queue;
+  };
+
+  ScopedVector<InternalTaskQueue> queues_;
+  AtomicSequenceNumber task_sequence_num_;
+  debug::TaskAnnotator task_annotator_;
+
+  base::ThreadChecker main_thread_checker_;
+  scoped_refptr<base::SingleThreadTaskRunner> main_task_runner_;
+  TaskQueueScheduler* scheduler_;
+
+  base::WeakPtrFactory<TaskQueueManager> weak_factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(TaskQueueManager);
+};
+
+}  // namespace content
+
+#endif  // BASE_TASK_TASK_QUEUE_MANAGER_H_