rtc_base/task_queue_libevent.cc - platform/external/webrtc - Gitiles

 /*
  *  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.
  */

 #include "rtc_base/task_queue_libevent.h"

 #include <errno.h>
 #include <fcntl.h>
 #include <pthread.h>
 #include <signal.h>
 #include <stdint.h>
 #include <time.h>
 #include <unistd.h>
 #include <list>
 #include <memory>
 #include <type_traits>
 #include <utility>

 #include "absl/memory/memory.h"
 #include "absl/strings/string_view.h"
 #include "api/task_queue/queued_task.h"
 #include "api/task_queue/task_queue_base.h"
 #include "base/third_party/libevent/event.h"
 #include "rtc_base/checks.h"
 #include "rtc_base/critical_section.h"
 #include "rtc_base/logging.h"
 #include "rtc_base/numerics/safe_conversions.h"
 #include "rtc_base/platform_thread.h"
 #include "rtc_base/platform_thread_types.h"
 #include "rtc_base/thread_annotations.h"
 #include "rtc_base/time_utils.h"

 namespace webrtc {
 namespace {
 constexpr char kQuit = 1;
 constexpr char kRunTask = 2;

 using Priority = TaskQueueFactory::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 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
 // don't want to risk that.
 // An alternative to this approach is to ignore the signal for the whole
 // process:
 //   signal(SIGPIPE, SIG_IGN);
 void IgnoreSigPipeSignalOnCurrentThread() {
   sigset_t sigpipe_mask;
   sigemptyset(&sigpipe_mask);
   sigaddset(&sigpipe_mask, SIGPIPE);
   pthread_sigmask(SIG_BLOCK, &sigpipe_mask, nullptr);
 }

 bool SetNonBlocking(int fd) {
   const int flags = fcntl(fd, F_GETFL);
   RTC_CHECK(flags != -1);
   return (flags & O_NONBLOCK) || fcntl(fd, F_SETFL, flags | O_NONBLOCK) != -1;
 }

 // TODO(tommi): This is a hack to support two versions of libevent that we're
 // compatible with.  The method we really want to call is event_assign(),
 // since event_set() has been marked as deprecated (and doesn't accept
 // passing event_base__ as a parameter).  However, the version of libevent
 // that we have in Chromium, doesn't have event_assign(), so we need to call
 // event_set() there.
 void EventAssign(struct event* ev,
                  struct event_base* base,
                  int fd,
                  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
 }

 rtc::ThreadPriority TaskQueuePriorityToThreadPriority(Priority priority) {
   switch (priority) {
     case Priority::HIGH:
       return rtc::kRealtimePriority;
     case Priority::LOW:
       return rtc::kLowPriority;
     case Priority::NORMAL:
       return rtc::kNormalPriority;
     default:
       RTC_NOTREACHED();
       break;
   }
   return rtc::kNormalPriority;
 }

 class TaskQueueLibevent final : public TaskQueueBase {
  public:
   TaskQueueLibevent(absl::string_view queue_name, rtc::ThreadPriority priority);

   void Delete() override;
   void PostTask(std::unique_ptr<QueuedTask> task) override;
   void PostDelayedTask(std::unique_ptr<QueuedTask> task,
                        uint32_t milliseconds) override;

  private:
   class SetTimerTask;
   struct TimerEvent;

   ~TaskQueueLibevent() override = default;

   static void ThreadMain(void* context);
   static void OnWakeup(int socket, short flags, void* context);  // NOLINT
   static void RunTask(int fd, short flags, void* context);       // NOLINT
   static void RunTimer(int fd, short flags, void* context);      // NOLINT

   bool is_active_ = true;
   int wakeup_pipe_in_ = -1;
   int wakeup_pipe_out_ = -1;
   event_base* event_base_;
   event wakeup_event_;
   rtc::PlatformThread thread_;
   rtc::CriticalSection pending_lock_;
   std::list<std::unique_ptr<QueuedTask>> pending_ RTC_GUARDED_BY(pending_lock_);
   // Holds a list of events pending timers for cleanup when the loop exits.
   std::list<TimerEvent*> pending_timers_;
 };

 struct TaskQueueLibevent::TimerEvent {
   TimerEvent(TaskQueueLibevent* task_queue, std::unique_ptr<QueuedTask> task)
       : task_queue(task_queue), task(std::move(task)) {}
   ~TimerEvent() { event_del(&ev); }

   event ev;
   TaskQueueLibevent* task_queue;
   std::unique_ptr<QueuedTask> task;
 };

 class TaskQueueLibevent::SetTimerTask : public QueuedTask {
  public:
   SetTimerTask(std::unique_ptr<QueuedTask> task, uint32_t milliseconds)
       : task_(std::move(task)),
         milliseconds_(milliseconds),
         posted_(rtc::Time32()) {}

  private:
   bool Run() override {
     // Compensate for the time that has passed since construction
     // and until we got here.
     uint32_t post_time = rtc::Time32() - posted_;
     TaskQueueLibevent::Current()->PostDelayedTask(
         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_;
 };

 TaskQueueLibevent::TaskQueueLibevent(absl::string_view queue_name,
                                      rtc::ThreadPriority priority)
     : event_base_(event_base_new()),
       thread_(&TaskQueueLibevent::ThreadMain, this, queue_name, priority) {
   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_, event_base_, wakeup_pipe_out_,
               EV_READ | EV_PERSIST, OnWakeup, this);
   event_add(&wakeup_event_, 0);
   thread_.Start();
 }

 void TaskQueueLibevent::Delete() {
   RTC_DCHECK(!IsCurrent());
   struct timespec ts;
   char message = kQuit;
   while (write(wakeup_pipe_in_, &message, sizeof(message)) != sizeof(message)) {
     // The queue is full, so we have no choice but to wait and retry.
     RTC_CHECK_EQ(EAGAIN, errno);
     ts.tv_sec = 0;
     ts.tv_nsec = 1000000;
     nanosleep(&ts, nullptr);
   }

   thread_.Stop();

   event_del(&wakeup_event_);

   IgnoreSigPipeSignalOnCurrentThread();

   close(wakeup_pipe_in_);
   close(wakeup_pipe_out_);
   wakeup_pipe_in_ = -1;
   wakeup_pipe_out_ = -1;

   event_base_free(event_base_);
   delete this;
 }

 void TaskQueueLibevent::PostTask(std::unique_ptr<QueuedTask> task) {
   RTC_DCHECK(task.get());
   // libevent isn't thread safe.  This means that we can't use methods such
   // as event_base_once to post tasks to the worker thread from a different
   // thread.  However, we can use it when posting from the worker thread itself.
   if (IsCurrent()) {
     if (event_base_once(event_base_, -1, EV_TIMEOUT,
                         &TaskQueueLibevent::RunTask, task.get(),
                         nullptr) == 0) {
       task.release();
     }
   } else {
     QueuedTask* task_id = task.get();  // Only used for comparison.
     {
       rtc::CritScope lock(&pending_lock_);
       pending_.push_back(std::move(task));
     }
     char message = kRunTask;
     if (write(wakeup_pipe_in_, &message, sizeof(message)) != sizeof(message)) {
       RTC_LOG(WARNING) << "Failed to queue task.";
       rtc::CritScope lock(&pending_lock_);
       pending_.remove_if([task_id](std::unique_ptr<QueuedTask>& t) {
         return t.get() == task_id;
       });
     }
   }
 }

 void TaskQueueLibevent::PostDelayedTask(std::unique_ptr<QueuedTask> task,
                                         uint32_t milliseconds) {
   if (IsCurrent()) {
     TimerEvent* timer = new TimerEvent(this, std::move(task));
     EventAssign(&timer->ev, event_base_, -1, 0, &TaskQueueLibevent::RunTimer,
                 timer);
     pending_timers_.push_back(timer);
     timeval tv = {rtc::dchecked_cast<int>(milliseconds / 1000),
                   rtc::dchecked_cast<int>(milliseconds % 1000) * 1000};
     event_add(&timer->ev, &tv);
   } else {
     PostTask(absl::make_unique<SetTimerTask>(std::move(task), milliseconds));
   }
 }

 // static
 void TaskQueueLibevent::ThreadMain(void* context) {
   TaskQueueLibevent* me = static_cast<TaskQueueLibevent*>(context);

   {
     CurrentTaskQueueSetter set_current(me);
     while (me->is_active_)
       event_base_loop(me->event_base_, 0);
   }

   for (TimerEvent* timer : me->pending_timers_)
     delete timer;
 }

 // static
 void TaskQueueLibevent::OnWakeup(int socket,
                                  short flags,  // NOLINT
                                  void* context) {
   TaskQueueLibevent* me = static_cast<TaskQueueLibevent*>(context);
   RTC_DCHECK(me->wakeup_pipe_out_ == socket);
   char buf;
   RTC_CHECK(sizeof(buf) == read(socket, &buf, sizeof(buf)));
   switch (buf) {
     case kQuit:
       me->is_active_ = false;
       event_base_loopbreak(me->event_base_);
       break;
     case kRunTask: {
       std::unique_ptr<QueuedTask> task;
       {
         rtc::CritScope lock(&me->pending_lock_);
         RTC_DCHECK(!me->pending_.empty());
         task = std::move(me->pending_.front());
         me->pending_.pop_front();
         RTC_DCHECK(task.get());
       }
       if (!task->Run())
         task.release();
       break;
     }
     default:
       RTC_NOTREACHED();
       break;
   }
 }

 // static
 void TaskQueueLibevent::RunTask(int fd, short flags, void* context) {  // NOLINT
   auto* task = static_cast<QueuedTask*>(context);
   if (task->Run())
     delete task;
 }

 // static
 void TaskQueueLibevent::RunTimer(int fd,
                                  short flags,  // NOLINT
                                  void* context) {
   TimerEvent* timer = static_cast<TimerEvent*>(context);
   if (!timer->task->Run())
     timer->task.release();
   timer->task_queue->pending_timers_.remove(timer);
   delete timer;
 }

 class TaskQueueLibeventFactory final : public TaskQueueFactory {
  public:
   std::unique_ptr<TaskQueueBase, TaskQueueDeleter> CreateTaskQueue(
       absl::string_view name,
       Priority priority) const override {
     return std::unique_ptr<TaskQueueBase, TaskQueueDeleter>(
         new TaskQueueLibevent(name,
                               TaskQueuePriorityToThreadPriority(priority)));
   }
 };

 }  // namespace

 std::unique_ptr<TaskQueueFactory> CreateTaskQueueLibeventFactory() {
   return absl::make_unique<TaskQueueLibeventFactory>();
 }

 }  // namespace webrtc
	/*
	* 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.
	*/

	#include "rtc_base/task_queue_libevent.h"

	#include <errno.h>
	#include <fcntl.h>
	#include <pthread.h>
	#include <signal.h>
	#include <stdint.h>
	#include <time.h>
	#include <unistd.h>
	#include <list>
	#include <memory>
	#include <type_traits>
	#include <utility>

	#include "absl/memory/memory.h"
	#include "absl/strings/string_view.h"
	#include "api/task_queue/queued_task.h"
	#include "api/task_queue/task_queue_base.h"
	#include "base/third_party/libevent/event.h"
	#include "rtc_base/checks.h"
	#include "rtc_base/critical_section.h"
	#include "rtc_base/logging.h"
	#include "rtc_base/numerics/safe_conversions.h"
	#include "rtc_base/platform_thread.h"
	#include "rtc_base/platform_thread_types.h"
	#include "rtc_base/thread_annotations.h"
	#include "rtc_base/time_utils.h"

	namespace webrtc {
	namespace {
	constexpr char kQuit = 1;
	constexpr char kRunTask = 2;

	using Priority = TaskQueueFactory::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 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
	// don't want to risk that.
	// An alternative to this approach is to ignore the signal for the whole
	// process:
	// signal(SIGPIPE, SIG_IGN);
	void IgnoreSigPipeSignalOnCurrentThread() {
	sigset_t sigpipe_mask;
	sigemptyset(&sigpipe_mask);
	sigaddset(&sigpipe_mask, SIGPIPE);
	pthread_sigmask(SIG_BLOCK, &sigpipe_mask, nullptr);
	}

	bool SetNonBlocking(int fd) {
	const int flags = fcntl(fd, F_GETFL);
	RTC_CHECK(flags != -1);
	return (flags & O_NONBLOCK) \|\| fcntl(fd, F_SETFL, flags \| O_NONBLOCK) != -1;
	}

	// TODO(tommi): This is a hack to support two versions of libevent that we're
	// compatible with. The method we really want to call is event_assign(),
	// since event_set() has been marked as deprecated (and doesn't accept
	// passing event_base__ as a parameter). However, the version of libevent
	// that we have in Chromium, doesn't have event_assign(), so we need to call
	// event_set() there.
	void EventAssign(struct event* ev,
	struct event_base* base,
	int fd,
	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
	}

	rtc::ThreadPriority TaskQueuePriorityToThreadPriority(Priority priority) {
	switch (priority) {
	case Priority::HIGH:
	return rtc::kRealtimePriority;
	case Priority::LOW:
	return rtc::kLowPriority;
	case Priority::NORMAL:
	return rtc::kNormalPriority;
	default:
	RTC_NOTREACHED();
	break;
	}
	return rtc::kNormalPriority;
	}

	class TaskQueueLibevent final : public TaskQueueBase {
	public:
	TaskQueueLibevent(absl::string_view queue_name, rtc::ThreadPriority priority);

	void Delete() override;
	void PostTask(std::unique_ptr<QueuedTask> task) override;
	void PostDelayedTask(std::unique_ptr<QueuedTask> task,
	uint32_t milliseconds) override;

	private:
	class SetTimerTask;
	struct TimerEvent;

	~TaskQueueLibevent() override = default;

	static void ThreadMain(void* context);
	static void OnWakeup(int socket, short flags, void* context); // NOLINT
	static void RunTask(int fd, short flags, void* context); // NOLINT
	static void RunTimer(int fd, short flags, void* context); // NOLINT

	bool is_active_ = true;
	int wakeup_pipe_in_ = -1;
	int wakeup_pipe_out_ = -1;
	event_base* event_base_;
	event wakeup_event_;
	rtc::PlatformThread thread_;
	rtc::CriticalSection pending_lock_;
	std::list<std::unique_ptr<QueuedTask>> pending_ RTC_GUARDED_BY(pending_lock_);
	// Holds a list of events pending timers for cleanup when the loop exits.
	std::list<TimerEvent*> pending_timers_;
	};

	struct TaskQueueLibevent::TimerEvent {
	TimerEvent(TaskQueueLibevent* task_queue, std::unique_ptr<QueuedTask> task)
	: task_queue(task_queue), task(std::move(task)) {}
	~TimerEvent() { event_del(&ev); }

	event ev;
	TaskQueueLibevent* task_queue;
	std::unique_ptr<QueuedTask> task;
	};

	class TaskQueueLibevent::SetTimerTask : public QueuedTask {
	public:
	SetTimerTask(std::unique_ptr<QueuedTask> task, uint32_t milliseconds)
	: task_(std::move(task)),
	milliseconds_(milliseconds),
	posted_(rtc::Time32()) {}

	private:
	bool Run() override {
	// Compensate for the time that has passed since construction
	// and until we got here.
	uint32_t post_time = rtc::Time32() - posted_;
	TaskQueueLibevent::Current()->PostDelayedTask(
	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_;
	};

	TaskQueueLibevent::TaskQueueLibevent(absl::string_view queue_name,
	rtc::ThreadPriority priority)
	: event_base_(event_base_new()),
	thread_(&TaskQueueLibevent::ThreadMain, this, queue_name, priority) {
	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_, event_base_, wakeup_pipe_out_,
	EV_READ \| EV_PERSIST, OnWakeup, this);
	event_add(&wakeup_event_, 0);
	thread_.Start();
	}

	void TaskQueueLibevent::Delete() {
	RTC_DCHECK(!IsCurrent());
	struct timespec ts;
	char message = kQuit;
	while (write(wakeup_pipe_in_, &message, sizeof(message)) != sizeof(message)) {
	// The queue is full, so we have no choice but to wait and retry.
	RTC_CHECK_EQ(EAGAIN, errno);
	ts.tv_sec = 0;
	ts.tv_nsec = 1000000;
	nanosleep(&ts, nullptr);
	}

	thread_.Stop();

	event_del(&wakeup_event_);

	IgnoreSigPipeSignalOnCurrentThread();

	close(wakeup_pipe_in_);
	close(wakeup_pipe_out_);
	wakeup_pipe_in_ = -1;
	wakeup_pipe_out_ = -1;

	event_base_free(event_base_);
	delete this;
	}

	void TaskQueueLibevent::PostTask(std::unique_ptr<QueuedTask> task) {
	RTC_DCHECK(task.get());
	// libevent isn't thread safe. This means that we can't use methods such
	// as event_base_once to post tasks to the worker thread from a different
	// thread. However, we can use it when posting from the worker thread itself.
	if (IsCurrent()) {
	if (event_base_once(event_base_, -1, EV_TIMEOUT,
	&TaskQueueLibevent::RunTask, task.get(),
	nullptr) == 0) {
	task.release();
	}
	} else {
	QueuedTask* task_id = task.get(); // Only used for comparison.
	{
	rtc::CritScope lock(&pending_lock_);
	pending_.push_back(std::move(task));
	}
	char message = kRunTask;
	if (write(wakeup_pipe_in_, &message, sizeof(message)) != sizeof(message)) {
	RTC_LOG(WARNING) << "Failed to queue task.";
	rtc::CritScope lock(&pending_lock_);
	pending_.remove_if([task_id](std::unique_ptr<QueuedTask>& t) {
	return t.get() == task_id;
	});
	}
	}
	}

	void TaskQueueLibevent::PostDelayedTask(std::unique_ptr<QueuedTask> task,
	uint32_t milliseconds) {
	if (IsCurrent()) {
	TimerEvent* timer = new TimerEvent(this, std::move(task));
	EventAssign(&timer->ev, event_base_, -1, 0, &TaskQueueLibevent::RunTimer,
	timer);
	pending_timers_.push_back(timer);
	timeval tv = {rtc::dchecked_cast<int>(milliseconds / 1000),
	rtc::dchecked_cast<int>(milliseconds % 1000) * 1000};
	event_add(&timer->ev, &tv);
	} else {
	PostTask(absl::make_unique<SetTimerTask>(std::move(task), milliseconds));
	}
	}

	// static
	void TaskQueueLibevent::ThreadMain(void* context) {
	TaskQueueLibevent* me = static_cast<TaskQueueLibevent*>(context);

	{
	CurrentTaskQueueSetter set_current(me);
	while (me->is_active_)
	event_base_loop(me->event_base_, 0);
	}

	for (TimerEvent* timer : me->pending_timers_)
	delete timer;
	}

	// static
	void TaskQueueLibevent::OnWakeup(int socket,
	short flags, // NOLINT
	void* context) {
	TaskQueueLibevent* me = static_cast<TaskQueueLibevent*>(context);
	RTC_DCHECK(me->wakeup_pipe_out_ == socket);
	char buf;
	RTC_CHECK(sizeof(buf) == read(socket, &buf, sizeof(buf)));
	switch (buf) {
	case kQuit:
	me->is_active_ = false;
	event_base_loopbreak(me->event_base_);
	break;
	case kRunTask: {
	std::unique_ptr<QueuedTask> task;
	{
	rtc::CritScope lock(&me->pending_lock_);
	RTC_DCHECK(!me->pending_.empty());
	task = std::move(me->pending_.front());
	me->pending_.pop_front();
	RTC_DCHECK(task.get());
	}
	if (!task->Run())
	task.release();
	break;
	}
	default:
	RTC_NOTREACHED();
	break;
	}
	}

	// static
	void TaskQueueLibevent::RunTask(int fd, short flags, void* context) { // NOLINT
	auto* task = static_cast<QueuedTask*>(context);
	if (task->Run())
	delete task;
	}

	// static
	void TaskQueueLibevent::RunTimer(int fd,
	short flags, // NOLINT
	void* context) {
	TimerEvent* timer = static_cast<TimerEvent*>(context);
	if (!timer->task->Run())
	timer->task.release();
	timer->task_queue->pending_timers_.remove(timer);
	delete timer;
	}

	class TaskQueueLibeventFactory final : public TaskQueueFactory {
	public:
	std::unique_ptr<TaskQueueBase, TaskQueueDeleter> CreateTaskQueue(
	absl::string_view name,
	Priority priority) const override {
	return std::unique_ptr<TaskQueueBase, TaskQueueDeleter>(
	new TaskQueueLibevent(name,
	TaskQueuePriorityToThreadPriority(priority)));
	}
	};

	} // namespace

	std::unique_ptr<TaskQueueFactory> CreateTaskQueueLibeventFactory() {
	return absl::make_unique<TaskQueueLibeventFactory>();
	}

	} // namespace webrtc