rtc::Thread介绍
rtc::Thread类不仅仅实现了线程这个执行器(比如posix底层调用pthread相关接口创建线程,管理线程等),还包括消息队列(message_queue)的实现,rtc::Thread启动后就作为一个永不停止的event loop,没有任务待执行就阻塞等待,添加任务
后就唤醒event loop,去执行任务,周而复始,直到调用stop退出event loop,退出线程(线程join)。
在WebRTC内部,可以将消息队列等同于event loop,消息队列为空,就进行阻塞等待。
class RTC_LOCKABLE Thread : public MessageQueue {
Thread关键接口
public:// Starts the execution of the thread.bool Start(Runnable* runnable = nullptr);// Tells the thread to stop and waits until it is joined.// Never call Stop on the current thread. Instead use the inherited Quit// function which will exit the base MessageQueue without terminating the// underlying OS thread.virtual void Stop();virtual void Send(const Location& posted_from,MessageHandler* phandler,uint32_t id = 0,MessageData* pdata = nullptr);// Convenience method to invoke a functor on another thread. Caller must// provide the |ReturnT| template argument, which cannot (easily) be deduced.// Uses Send() internally, which blocks the current thread until execution// is complete.// Ex: bool result = thread.Invoke<bool>(RTC_FROM_HERE,// &MyFunctionReturningBool);// NOTE: This function can only be called when synchronous calls are allowed.// See ScopedDisallowBlockingCalls for details.template <class ReturnT, class FunctorT>ReturnT Invoke(const Location& posted_from, FunctorT&& functor) {FunctorMessageHandler<ReturnT, FunctorT> handler(std::forward<FunctorT>(functor));InvokeInternal(posted_from, &handler);return handler.MoveResult();}// ProcessMessages will process I/O and dispatch messages until:// 1) cms milliseconds have elapsed (returns true)// 2) Stop() is called (returns false)bool ProcessMessages(int cms);protected:// Blocks the calling thread until this thread has terminated.void Join();
MessageQueue关键接口
public:
virtual void Quit();// 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 void Post(const Location& posted_from,MessageHandler* phandler,uint32_t id = 0,MessageData* pdata = nullptr,bool time_sensitive = false);
virtual void PostDelayed(const Location& posted_from,int cmsDelay,MessageHandler* phandler,uint32_t id = 0,MessageData* pdata = nullptr);
virtual void PostAt(const Location& posted_from,int64_t tstamp,MessageHandler* phandler,uint32_t id = 0,MessageData* pdata = nullptr);virtual void Dispatch(Message* pmsg);
virtual void ReceiveSends();protected:
void WakeUpSocketServer();MessageList msgq_ RTC_GUARDED_BY(crit_);
PriorityQueue dmsgq_ RTC_GUARDED_BY(crit_);
线程启动Start
调用Start接口启动底层线程,同时进入一个永不停止的event loop(除非调用Stop接口)
流程如下:
Start->pthread_create->PreRun->Run
void Thread::Run() {ProcessMessages(kForever);
}
最终通过Get接口获取消息去执行(Dispatch),Get获取不到消息就是进入阻塞状态(wait),等待有消息后被唤醒。
线程消息队列处理消息的流程ProcessMessage
- 1、处理从其他线程发送的要在本线程去执行的消息,即同步调用
接收者线程处理流程:
发送者线程流程:
-
2、处理延迟消息(存储在优先级队列)
延迟消息是通过PostDelayed和PostAt接口调用然后push到优先级队列中(dmsgq_,小根堆)
-
3、异步消息(存储在普通队列里)
延迟消息是通过Pos接口调用然后push到普通队列中(msgq_)
任务提交方式(Invoke/Post)
webrtc内部消息其实是对待执行任务的封装,消息和任务可以认为是一个意思
消息要继承MessageHandler,实现OnMessage
class MessageHandler {public:virtual ~MessageHandler();virtual void OnMessage(Message* msg) = 0;protected:MessageHandler() {}private:RTC_DISALLOW_COPY_AND_ASSIGN(MessageHandler);
};
因为执行消息,实际上就是执行OnMessage(详见Dispatch接口实现)
上一章节其实已经把三种任务提交方式介绍过了
1、同步阻塞调用(Send,Invoke)
Invoke其实最终也是调用Send,Invoke是个函数模版,可以非常方便在目标执行线程执行函数然后获得返回值,Invoke实现如下:
// Convenience method to invoke a functor on another thread. Caller must// provide the |ReturnT| template argument, which cannot (easily) be deduced.// Uses Send() internally, which blocks the current thread until execution// is complete.// Ex: bool result = thread.Invoke<bool>(RTC_FROM_HERE,// &MyFunctionReturningBool);// NOTE: This function can only be called when synchronous calls are allowed.// See ScopedDisallowBlockingCalls for details.template <class ReturnT, class FunctorT>ReturnT Invoke(const Location& posted_from, FunctorT&& functor) {FunctorMessageHandler<ReturnT, FunctorT> handler(std::forward<FunctorT>(functor));InvokeInternal(posted_from, &handler);return handler.MoveResult();}void Thread::InvokeInternal(const Location& posted_from,MessageHandler* handler) {TRACE_EVENT2("webrtc", "Thread::Invoke", "src_file_and_line",posted_from.file_and_line(), "src_func",posted_from.function_name());Send(posted_from, handler);
}
调用方式举例:
bool result = thread.Invoke<bool>(RTC_FROM_HERE, &MyFunctionReturningBool);
2、异步非阻塞延迟调用
PostDelayed和PostAt
3、异步非阻塞调用
Post
线程退出Stop
void Thread::Stop() {MessageQueue::Quit();Join();
}void MessageQueue::Quit() {AtomicOps::ReleaseStore(&stop_, 1);WakeUpSocketServer();
}void Thread::Join() {if (!IsRunning())return;RTC_DCHECK(!IsCurrent());if (Current() && !Current()->blocking_calls_allowed_) {RTC_LOG(LS_WARNING) << "Waiting for the thread to join, "<< "but blocking calls have been disallowed";}#if defined(WEBRTC_WIN)RTC_DCHECK(thread_ != nullptr);WaitForSingleObject(thread_, INFINITE);CloseHandle(thread_);thread_ = nullptr;thread_id_ = 0;
#elif defined(WEBRTC_POSIX)pthread_join(thread_, nullptr);thread_ = 0;
#endif
}