diff --git a/src/threading/MsgThread.cc b/src/threading/MsgThread.cc index ddcd3df1dd..0b91f8790a 100644 --- a/src/threading/MsgThread.cc +++ b/src/threading/MsgThread.cc @@ -10,13 +10,21 @@ namespace threading { ////// Messages. -// Signals child thread to terminate. This is actually a no-op; its only -// purpose is unblock the current read operation so that the child's Run() -// methods can check the termination status. -class TerminateMessage : public InputMessage +// Signals child thread to shutdown operation. +class FinishMessage : public InputMessage { public: - TerminateMessage(MsgThread* thread) : InputMessage("Terminate", thread) { } + FinishMessage(MsgThread* thread) : InputMessage("Finish", thread) { } + + virtual bool Process() { return Object()->DoFinish(); } +}; + +// A dummy message that's only purpose is unblock the current read operation +// so that the child's Run() methods can check the termination status. +class UnblockMessage : public InputMessage +{ +public: + UnblockMessage(MsgThread* thread) : InputMessage("Unblock", thread) { } virtual bool Process() { return true; } }; @@ -130,13 +138,30 @@ bool ReporterMessage::Process() MsgThread::MsgThread() : BasicThread() { cnt_sent_in = cnt_sent_out = 0; + finished = false; thread_mgr->AddMsgThread(this); } void MsgThread::OnStop() { - // This is to unblock the current queue read operation. - SendIn(new TerminateMessage(this), true); + // Signal thread to terminate and wait until it has acknowledged. + SendIn(new FinishMessage(this), true); + + int cnt = 0; + while ( ! finished ) + { + if ( ++cnt > 1000 ) // Insurance against broken threads ... + { + abort(); + reporter->Warning("thread %s didn't finish in time", Name().c_str()); + break; + } + + usleep(1000); + } + + // One more message to make sure the current queue read operation unblocks. + SendIn(new UnblockMessage(this), true); } void MsgThread::Heartbeat() @@ -157,6 +182,14 @@ bool MsgThread::DoHeartbeat(double network_time, double current_time) return true; } +bool MsgThread::DoFinish() + { + // This is thread-safe "enough", we're the only one ever writing + // there. + finished = true; + return true; + } + void MsgThread::Info(const char* msg) { SendOut(new ReporterMessage(ReporterMessage::INFO, this, msg)); diff --git a/src/threading/MsgThread.h b/src/threading/MsgThread.h index 5ac1c0f780..16e6a92772 100644 --- a/src/threading/MsgThread.h +++ b/src/threading/MsgThread.h @@ -171,6 +171,8 @@ public: protected: friend class Manager; friend class HeartbeatMessage; + friend class FinishMessage; + friend class FinishedMessage; /** * Pops a message sent by the child from the child-to-main queue. @@ -215,6 +217,12 @@ protected: */ virtual bool DoHeartbeat(double network_time, double current_time); + /** Triggered for execution in the child thread just before shutting threads down. + * The child thread shoudl finish its operations and then *must* + * call this class' implementation. + */ + virtual bool DoFinish(); + private: /** * Pops a message sent by the main thread from the main-to-chold @@ -270,6 +278,8 @@ private: uint64_t cnt_sent_in; // Counts message sent to child. uint64_t cnt_sent_out; // Counts message sent by child. + + bool finished; // Set to true by Finished message. }; /**