blob: 628ce4433b824396c952400836d39bd8df023a6c [file] [log] [blame]
* Copyright 2004 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 <string.h>
#include <algorithm>
#include <list>
#include <memory>
#include <queue>
#include <vector>
#include "webrtc/base/basictypes.h"
#include "webrtc/base/constructormagic.h"
#include "webrtc/base/criticalsection.h"
#include "webrtc/base/location.h"
#include "webrtc/base/messagehandler.h"
#include "webrtc/base/scoped_ref_ptr.h"
#include "webrtc/base/sharedexclusivelock.h"
#include "webrtc/base/sigslot.h"
#include "webrtc/base/socketserver.h"
#include "webrtc/base/timeutils.h"
#include "webrtc/base/thread_annotations.h"
namespace rtc {
struct Message;
class MessageQueue;
// MessageQueueManager does cleanup of of message queues
class MessageQueueManager {
static void Add(MessageQueue *message_queue);
static void Remove(MessageQueue *message_queue);
static void Clear(MessageHandler *handler);
// For testing purposes, we expose whether or not the MessageQueueManager
// instance has been initialized. It has no other use relative to the rest of
// the functions of this class, which auto-initialize the underlying
// MessageQueueManager instance when necessary.
static bool IsInitialized();
// Mainly for testing purposes, for use with a simulated clock.
// Ensures that all message queues have processed delayed messages
// up until the current point in time.
static void ProcessAllMessageQueues();
static MessageQueueManager* Instance();
void AddInternal(MessageQueue *message_queue);
void RemoveInternal(MessageQueue *message_queue);
void ClearInternal(MessageHandler *handler);
void ProcessAllMessageQueuesInternal();
static MessageQueueManager* instance_;
// This list contains all live MessageQueues.
std::vector<MessageQueue*> message_queues_ GUARDED_BY(crit_);
// Acquire this with DebugNonReentrantCritScope.
CriticalSection crit_;
bool locked_ GUARDED_BY(crit_);
// Derive from this for specialized data
// App manages lifetime, except when messages are purged
class MessageData {
MessageData() {}
virtual ~MessageData() {}
template <class T>
class TypedMessageData : public MessageData {
explicit TypedMessageData(const T& data) : data_(data) { }
const T& data() const { return data_; }
T& data() { return data_; }
T data_;
// Like TypedMessageData, but for pointers that require a delete.
template <class T>
class ScopedMessageData : public MessageData {
explicit ScopedMessageData(T* data) : data_(data) { }
const std::unique_ptr<T>& data() const { return data_; }
std::unique_ptr<T>& data() { return data_; }
std::unique_ptr<T> data_;
// Like ScopedMessageData, but for reference counted pointers.
template <class T>
class ScopedRefMessageData : public MessageData {
explicit ScopedRefMessageData(T* data) : data_(data) { }
const scoped_refptr<T>& data() const { return data_; }
scoped_refptr<T>& data() { return data_; }
scoped_refptr<T> data_;
template<class T>
inline MessageData* WrapMessageData(const T& data) {
return new TypedMessageData<T>(data);
template<class T>
inline const T& UseMessageData(MessageData* data) {
return static_cast< TypedMessageData<T>* >(data)->data();
template<class T>
class DisposeData : public MessageData {
explicit DisposeData(T* data) : data_(data) { }
virtual ~DisposeData() { delete data_; }
T* data_;
const uint32_t MQID_ANY = static_cast<uint32_t>(-1);
const uint32_t MQID_DISPOSE = static_cast<uint32_t>(-2);
// No destructor
struct Message {
: phandler(nullptr), message_id(0), pdata(nullptr), ts_sensitive(0) {}
inline bool Match(MessageHandler* handler, uint32_t id) const {
return (handler == NULL || handler == phandler)
&& (id == MQID_ANY || id == message_id);
Location posted_from;
MessageHandler *phandler;
uint32_t message_id;
MessageData *pdata;
int64_t ts_sensitive;
typedef std::list<Message> MessageList;
// DelayedMessage goes into a priority queue, sorted by trigger time. Messages
// with the same trigger time are processed in num_ (FIFO) order.
class DelayedMessage {
DelayedMessage(int64_t delay,
int64_t trigger,
uint32_t num,
const Message& msg)
: cmsDelay_(delay), msTrigger_(trigger), num_(num), msg_(msg) {}
bool operator< (const DelayedMessage& dmsg) const {
return (dmsg.msTrigger_ < msTrigger_)
|| ((dmsg.msTrigger_ == msTrigger_) && (dmsg.num_ < num_));
int64_t cmsDelay_; // for debugging
int64_t msTrigger_;
uint32_t num_;
Message msg_;
class MessageQueue {
static const int kForever = -1;
// Create a new MessageQueue and optionally assign it to the passed
// SocketServer. Subclasses that override Clear should pass false for
// init_queue and call DoInit() from their constructor to prevent races
// with the MessageQueueManager using the object while the vtable is still
// being created.
MessageQueue(SocketServer* ss, bool init_queue);
MessageQueue(std::unique_ptr<SocketServer> ss, bool init_queue);
// DoDestroy() IN THEIR DESTRUCTORS! This is required to avoid a data race
// between the destructor modifying the vtable, and the MessageQueueManager
// calling Clear on the object from a different thread.
virtual ~MessageQueue();
SocketServer* socketserver();
void set_socketserver(SocketServer* ss);
// Note: The behavior of MessageQueue has changed. When a MQ is stopped,
// futher Posts and Sends will fail. However, any pending Sends and *ready*
// Posts (as opposed to unexpired delayed Posts) will be delivered before
// Get (or Peek) returns false. By guaranteeing delivery of those messages,
// we eliminate the race condition when an MessageHandler and MessageQueue
// may be destroyed independently of each other.
virtual void Quit();
virtual bool IsQuitting();
virtual void Restart();
// Get() will process I/O until:
// 1) A message is available (returns true)
// 2) cmsWait seconds have elapsed (returns false)
// 3) Stop() is called (returns false)
virtual bool Get(Message *pmsg, int cmsWait = kForever,
bool process_io = true);
virtual bool Peek(Message *pmsg, int cmsWait = 0);
virtual void Post(const Location& posted_from,
MessageHandler* phandler,
uint32_t id = 0,
MessageData* pdata = NULL,
bool time_sensitive = false);
virtual void PostDelayed(const Location& posted_from,
int cmsDelay,
MessageHandler* phandler,
uint32_t id = 0,
MessageData* pdata = NULL);
virtual void PostAt(const Location& posted_from,
int64_t tstamp,
MessageHandler* phandler,
uint32_t id = 0,
MessageData* pdata = NULL);
// TODO(honghaiz): Remove this when all the dependencies are removed.
virtual void PostAt(const Location& posted_from,
uint32_t tstamp,
MessageHandler* phandler,
uint32_t id = 0,
MessageData* pdata = NULL);
virtual void Clear(MessageHandler* phandler,
uint32_t id = MQID_ANY,
MessageList* removed = NULL);
virtual void Dispatch(Message *pmsg);
virtual void ReceiveSends();
// Amount of time until the next message can be retrieved
virtual int GetDelay();
bool empty() const { return size() == 0u; }
size_t size() const {
CritScope cs(&crit_); // msgq_.size() is not thread safe.
return msgq_.size() + dmsgq_.size() + (fPeekKeep_ ? 1u : 0u);
// Internally posts a message which causes the doomed object to be deleted
template<class T> void Dispose(T* doomed) {
if (doomed) {
Post(RTC_FROM_HERE, NULL, MQID_DISPOSE, new DisposeData<T>(doomed));
// When this signal is sent out, any references to this queue should
// no longer be used.
sigslot::signal0<> SignalQueueDestroyed;
class PriorityQueue : public std::priority_queue<DelayedMessage> {
container_type& container() { return c; }
void reheap() { make_heap(c.begin(), c.end(), comp); }
void DoDelayPost(const Location& posted_from,
int64_t cmsDelay,
int64_t tstamp,
MessageHandler* phandler,
uint32_t id,
MessageData* pdata);
// Perform initialization, subclasses must call this from their constructor
// if false was passed as init_queue to the MessageQueue constructor.
void DoInit();
// Perform cleanup, subclasses that override Clear must call this from the
// destructor.
void DoDestroy();
void WakeUpSocketServer();
bool fPeekKeep_;
Message msgPeek_;
MessageList msgq_ GUARDED_BY(crit_);
PriorityQueue dmsgq_ GUARDED_BY(crit_);
uint32_t dmsgq_next_num_ GUARDED_BY(crit_);
CriticalSection crit_;
bool fInitialized_;
bool fDestroyed_;
volatile int stop_;
// The SocketServer might not be owned by MessageQueue.
SocketServer* ss_ GUARDED_BY(ss_lock_);
// Used if SocketServer ownership lies with |this|.
std::unique_ptr<SocketServer> own_ss_;
SharedExclusiveLock ss_lock_;
} // namespace rtc