Two Worker Thread Communication

[mbaoqi]

Hi,
There is a question need to help. The program will create two worker thread after push a UI button. For explicitly, call this two worker thread as t1 and t2. I want the two thread's action like that: t1 send a message to t2, t2 get the message and process it, t2 generate a new message, and send back to t1, t1 get the last message and will show it on UI. The demo code is below, but it can not work as I wanted. thanks!

Code: Select all

#include <wx/wxprec.h>

#ifndef WX_PRECOMP
#include <wx/wx.h>
#endif

#include <wx/msgqueue.h>

enum
{
	THREAD_START = wxID_HIGHEST + 1,
	ID_WORKER1,
	ID_WORKER2
};

wxMessageQueue<unsigned int> gMsgQueueInt;
wxMessageQueue<wxString> gMsgQueueStr;

class App : public wxApp
{
public:
	virtual bool OnInit() wxOVERRIDE;
};

IMPLEMENT_APP(App);

class Worker1 : public wxThread
{
public:
	Worker1(wxEvtHandler* parent);
	~Worker1();

	wxThread::ExitCode Entry() wxOVERRIDE;

private:
	wxEvtHandler* m_parent;
};

class Worker2 : public wxThread
{
public:
	Worker2(wxEvtHandler* parent);
	~Worker2();

	wxThread::ExitCode Entry() wxOVERRIDE;

private:
	wxEvtHandler* m_parent;
};

class AppFrame : public wxFrame
{
public:
	AppFrame();
	~AppFrame();

protected:
	void OnThreadStart(wxCommandEvent& event);
	void OnUpdateStatusBar(wxThreadEvent& event);
};

////////////////////////////////////////////
bool App::OnInit()
{
	AppFrame* frame = new AppFrame;
	frame->Centre();
	frame->Show();

	return true;
}

Worker1::Worker1(wxEvtHandler* parent)
	: m_parent(parent)
{

}

Worker1::~Worker1()
{

}

wxThread::ExitCode Worker1::Entry()
{
	while (true) {
		gMsgQueueStr.Post(wxString("**"));
		wxThread::Sleep(300);

		unsigned int val;
		gMsgQueueInt.Receive(val);

		wxThreadEvent evt(wxEVT_THREAD, ID_WORKER1);
		evt.SetInt(val);
		wxQueueEvent(m_parent, evt.Clone());
	}

	return (wxThread::ExitCode)0;
}

Worker2::Worker2(wxEvtHandler* parent)
	: m_parent(parent)
{

}

Worker2::~Worker2()
{

}

wxThread::ExitCode Worker2::Entry()
{
	int val = 0;
	while (true) {
		wxString str;
		gMsgQueueStr.Receive(str);

		wxThreadEvent evt(wxEVT_THREAD, ID_WORKER2);
		evt.SetString(str);
		wxQueueEvent(m_parent, evt.Clone());

		static unsigned int val = 0;
		gMsgQueueInt.Post(++val);
	}
	return (wxThread::ExitCode)0;
}

AppFrame::AppFrame()
	: wxFrame(nullptr, wxID_ANY, "Thread Communication", wxDefaultPosition, wxSize(500, 300))
{
	wxBoxSizer* sizer = new wxBoxSizer(wxVERTICAL);

	wxButton* btnThreadStart = new wxButton(this, THREAD_START, wxT("Thread Start"), wxDefaultPosition, wxSize(100, 50));
	sizer->Add(btnThreadStart, 0, wxALL, 10);

	SetSizer(sizer);

	CreateStatusBar();

	SetFocus();

	Bind(wxEVT_COMMAND_BUTTON_CLICKED, &AppFrame::OnThreadStart, this, THREAD_START);
	Bind(wxEVT_THREAD, &AppFrame::OnUpdateStatusBar, this, ID_WORKER2);
	Bind(wxEVT_THREAD, &AppFrame::OnUpdateStatusBar, this, ID_WORKER1);
}

AppFrame::~AppFrame()
{

}

void AppFrame::OnThreadStart(wxCommandEvent& WXUNUSED(event))
{
	wxButton* button = (wxButton*)FindWindowById(THREAD_START);
	button->Enable(false);

	Worker1* w1 = new Worker1(this);
	w1->Run();

	Worker2* w2 = new Worker2(this);
	w2->Run();
}

void AppFrame::OnUpdateStatusBar(wxThreadEvent& event)
{
	int val = event.GetInt();
	wxString str = event.GetString();
	SetStatusText(wxString::Format("value: %d, %s", val, str));
}

[doublemax]

Thread-wise the code works correctly. It's just a timing issue regarding when the events arrive.

T1 will send ** to the queue and then wait.
T2 will receive the **, send an event and put an int into the queue
main thread will display "**" and 0 in status bar

When T1 is finished sleeping, it will receive INT from the queue, send the event to the main thread, and - and this is the problematic part - also put a new "**" into the queue.

This means that immediately after the main thread receives the event with the integer, it will receive the event with the "**" and 0 integer value and override the previous change to the statusbar.

You can see this easily if you add wxLogDebug calls at the relevant places and increase the sleep time (That's what i did).

[PB]

And what output do you expect to be displayed? The code seems odd to me but it works as expected

Worker1 does not set the event string, so its output is

Code: Select all

"value: 1,"
"value: 2,"
...

Worker2 does not set the event int, so its output will always be

Code: Select all

"value: 0, **"

Worker 2 thread event arrives after Worker 1 thread event so it immediately overwrites Worker 1 output, making it seem as Worker 1 does not send anything.

I recommend using sprinkling wxLogDebug() throughout the code and you will see what happens there, e.g. with

Code: Select all

void AppFrame::OnUpdateStatusBar(wxThreadEvent& event)
{
	int val = event.GetInt();
	wxString str = event.GetString();
	const wxString output = wxString::Format("value: %d, %s", val, str);

	wxLogDebug("(Thread %d) %s", event.GetId(), output);
	SetStatusText(output);
}

you will see

Code: Select all

(Thread 6002) value: 0, **
(Thread 6001) value: 1, 
(Thread 6002) value: 0, **
(Thread 6001) value: 2, 
(Thread 6002) value: 0, **
(Thread 6001) value: 3, 
(Thread 6002) value: 0, **
(Thread 6001) value: 4, 
(Thread 6002) value: 0, **
(Thread 6001) value: 5, 

EDIT:
Too slow. I forgot to write that your code does not match the description, where it is written that only Worker 1 updates the UI, but in the code both threads do.

[mbaoqi]

Thanks @doublemax and @PB! Thanks for your reply!

At the beginning, I want to make t1 thread and t2 thread communicate with each other. t2 get information from t1, t2 process the information, t2 send event to UI, and t2 send process result back to t1, and t1 send an event contains the process result to UI. Now, to avoid UI display override, t2 only process data and do not send event. Example like this:

Code: Select all

#include <wx/wxprec.h>

#ifndef WX_PRECOMP
#include <wx/wx.h>
#endif

#include <wx/msgqueue.h>
#include <stdlib.h>


enum
{
	THREAD_START = wxID_HIGHEST + 1,
	ID_WORKER1,
	ID_WORKER2
};

wxMessageQueue<unsigned int> gMsgQueueInt;
wxMessageQueue<wxString> gMsgQueueStr;

unsigned int gVal = 1;

class App : public wxApp
{
public:
	virtual bool OnInit() wxOVERRIDE;
};

IMPLEMENT_APP(App);

class Worker1 : public wxThread
{
public:
	Worker1(wxEvtHandler* parent);
	~Worker1();

	wxThread::ExitCode Entry() wxOVERRIDE;

private:
	wxEvtHandler* m_parent;
};

class Worker2 : public wxThread
{
public:
	Worker2(wxEvtHandler* parent);
	~Worker2();

	wxThread::ExitCode Entry() wxOVERRIDE;

private:
	wxEvtHandler* m_parent;
};

class AppFrame : public wxFrame
{
public:
	AppFrame();
	~AppFrame();

protected:
	void OnThreadStart(wxCommandEvent& event);
	void OnUpdateStatusBar(wxThreadEvent& event);
};

////////////////////////////////////////////
bool App::OnInit()
{
	AppFrame* frame = new AppFrame;
	frame->Centre();
	frame->Show();

	return true;
}

Worker1::Worker1(wxEvtHandler* parent)
	: m_parent(parent)
{

}

Worker1::~Worker1()
{

}

wxThread::ExitCode Worker1::Entry()
{
	while (true) {
		// simulate produce a result after long time process. send the result to the another worker thread.
		srand(time(NULL));
		int tmp = rand() % 100;
		wxThread::Sleep(1000);
		gMsgQueueInt.Post(tmp);

		// get the back result from the another worker thread.
		wxString str;
		gMsgQueueStr.Receive(str);

		// push both result tmp and str into event and send to UI.
		wxThreadEvent evt(wxEVT_THREAD, ID_WORKER1);
		evt.SetInt(tmp);
		evt.SetString(str);
		wxQueueEvent(m_parent, evt.Clone());
	}

	return (wxThread::ExitCode)0;
}

Worker2::Worker2(wxEvtHandler* parent)
	: m_parent(parent)
{

}

Worker2::~Worker2()
{

}

wxThread::ExitCode Worker2::Entry()
{
	unsigned int val = 0;
	while (true) {
		gMsgQueueInt.Receive(val);

		// simulate process Worker1 post int value and send the string value back to Worker1 thread.
		wxString str;
		if (val % 2 == 0) {
			str = wxString("even, 36.5");
			gMsgQueueStr.Post(str);
		}
		else {
			str = wxString("odd, 37.2");
			gMsgQueueStr.Post(str);
		}

		// Note: do not send the int result using event in this thread to avoid UI display error.
		// wxThreadEvent evt(wxEVT_THREAD, ID_WORKER2);
		// evt.SetInt(val);
		// evt.SetString(str);
		// wxQueueEvent(m_parent, evt.Clone());

	}
	return (wxThread::ExitCode)0;
}

AppFrame::AppFrame()
	: wxFrame(nullptr, wxID_ANY, "Thread Communication", wxDefaultPosition, wxSize(500, 300))
{
	wxBoxSizer* sizer = new wxBoxSizer(wxVERTICAL);

	wxButton* btnThreadStart = new wxButton(this, THREAD_START, wxT("Thread Start"), wxDefaultPosition, wxSize(100, 50));
	sizer->Add(btnThreadStart, 0, wxALL, 10);

	SetSizer(sizer);

	CreateStatusBar();

	SetFocus();

	Bind(wxEVT_COMMAND_BUTTON_CLICKED, &AppFrame::OnThreadStart, this, THREAD_START);
	Bind(wxEVT_THREAD, &AppFrame::OnUpdateStatusBar, this, ID_WORKER2);
	Bind(wxEVT_THREAD, &AppFrame::OnUpdateStatusBar, this, ID_WORKER1);
}

AppFrame::~AppFrame()
{

}

void AppFrame::OnThreadStart(wxCommandEvent& WXUNUSED(event))
{
	wxButton* button = (wxButton*)FindWindowById(THREAD_START);
	button->Enable(false);

	Worker1* w1 = new Worker1(this);
	w1->Run();

	Worker2* w2 = new Worker2(this);
	w2->Run();
}

void AppFrame::OnUpdateStatusBar(wxThreadEvent& event)
{
	int val = event.GetInt();
	wxString str = event.GetString();
	SetStatusText(wxString::Format("(thread id: %d) value: %d, %s", event.GetId(), val, str));
}

[doublemax]

So, is this issue resolved or do you still have questions regarding your new code?

[mbaoqi]

I have achieved my point. thanks!