Add support for priorities to TaskQueue.

webrtc/base/task_queue_libevent.cc

 /*
  *  Copyright 2016 The WebRTC 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 in the root of the source
  *  tree. An additional intellectual property rights grant can be found
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */
 
@@ skipping 12 matching lines @@
 
 namespace rtc {
 using internal::GetQueuePtrTls;
 using internal::AutoSetCurrentQueuePtr;
 
 namespace {
 static const char kQuit = 1;
 static const char kRunTask = 2;
 static const char kRunReplyTask = 3;
 
+using Priority = TaskQueue::Priority;
+
 // This ignores the SIGPIPE signal on the calling thread.
 // This signal can be fired when trying to write() to a pipe that's being
 // closed or while closing a pipe that's being written to.
 // We can run into that situation (e.g. reply tasks that don't get a chance to
 // run because the task queue is being deleted) so we ignore this signal and
 // continue as normal.
 // As a side note for this implementation, it would be great if we could safely
 // restore the sigmask, but unfortunately the operation of restoring it, can
 // itself actually cause SIGPIPE to be signaled :-| (e.g. on MacOS)
 // The SIGPIPE signal by default causes the process to be terminated, so we
@@ skipping 34 matching lines @@
                  short events,
                  void (*callback)(int, short, void*),
                  void* arg) {
 #if defined(_EVENT2_EVENT_H_)
   RTC_CHECK_EQ(0, event_assign(ev, base, fd, events, callback, arg));
 #else
   event_set(ev, fd, events, callback, arg);
   RTC_CHECK_EQ(0, event_base_set(base, ev));
 #endif
 }
+
+ThreadPriority TaskQueuePriorityToThreadPriority(Priority priority) {
+  switch (priority) {
+    case Priority::HIGH:
+      return kRealtimePriority;
+    case Priority::LOW:
+      return kLowPriority;
+    case Priority::NORMAL:
+      return kNormalPriority;
+    default:
+      RTC_NOTREACHED();
+      break;
+  }
+  return kNormalPriority;
+}
 }  // namespace
 
 struct TaskQueue::QueueContext {
   explicit QueueContext(TaskQueue* q) : queue(q), is_active(true) {}
   TaskQueue* queue;
   bool is_active;
   // Holds a list of events pending timers for cleanup when the loop exits.
   std::list<TimerEvent*> pending_timers_;
 };
 
@@ skipping 97 matching lines @@
         std::move(task_),
         post_time > milliseconds_ ? 0 : milliseconds_ - post_time);
     return true;
   }
 
   std::unique_ptr<QueuedTask> task_;
   const uint32_t milliseconds_;
   const uint32_t posted_;
 };
 
-TaskQueue::TaskQueue(const char* queue_name)
+TaskQueue::TaskQueue(const char* queue_name, Priority priority /*= NORMAL*/)
     : event_base_(event_base_new()),
       wakeup_event_(new event()),
-      thread_(&TaskQueue::ThreadMain, this, queue_name) {
+      thread_(&TaskQueue::ThreadMain,
+              this,
+              queue_name,
+              TaskQueuePriorityToThreadPriority(priority)) {
   RTC_DCHECK(queue_name);
   int fds[2];
   RTC_CHECK(pipe(fds) == 0);
   SetNonBlocking(fds[0]);
   SetNonBlocking(fds[1]);
   wakeup_pipe_out_ = fds[0];
   wakeup_pipe_in_ = fds[1];
 
   EventAssign(wakeup_event_.get(), event_base_, wakeup_pipe_out_,
               EV_READ | EV_PERSIST, OnWakeup, this);
@@ skipping 182 matching lines @@
   delete timer;
 }
 
 void TaskQueue::PrepareReplyTask(scoped_refptr<ReplyTaskOwnerRef> reply_task) {
   RTC_DCHECK(reply_task);
   CritScope lock(&pending_lock_);
   pending_replies_.push_back(std::move(reply_task));
 }
 
 }  // namespace rtc