Thread safe event loop

[jonathan]

Hello all,
Just wondering if someone can help me with something I haven't done
in C++ before. I'd like to add a simple event loop to a music player
I've written. I'd like to know if I can do this in portable C++, or
will I need to insert some assembly language?

For context, the basic idea is that I have playback in one thread,
and UI in another. Currently I'm using the Qt libraries to handle
this. I don't mind this so much, but if I can remove the dependency
I'd like to.

I only need to be able to post messages to the playback thread like
"play", "stop", "pause", "new song", etc. Nothing extremely time
critical. I just don't want the (unlikely) event of losing messages.

My first thought was to create a "message" and add it to an event
queue 'as quick as possible'. Say,

last_queued_message->next = &my_newly_created_message;

I recognize that if I have two threads doing this at the same time
they could theoretically both write to "next" and one of them will be
lost. Can I get around this in C++? Do I need to use cmpxchg (I'm on
x86)?

Any ideas are much appreciated
--Jonathan

 

[red floyd]

Hello all,
  Just wondering if someone can help me with something I haven't done
in C++ before. I'd like to add a simple event loop to a music player
I've written. I'd like to know if I can do this in portable C++, or
will I need to insert some assembly language?

  For context, the basic idea is that I have playback in one thread,
and UI in another. Currently I'm using the Qt libraries to handle
this. I don't mind this so much, but if I can remove the dependency
I'd like to.

I'm sure that Qt provides synchronization primitives. You'd most
likely get
better answers on a forum dedicated to Qt.

 

[Marcel Müller]

Hi,

My first thought was to create a "message" and add it to an event
queue 'as quick as possible'. Say,

last_queued_message->next = &my_newly_created_message;

I recognize that if I have two threads doing this at the same time
they could theoretically both write to "next" and one of them will be
lost. Can I get around this in C++?

currently not. You need some synchronization library. If you have qt,
you probably also have pthreads.

But you are not portable anyway, because C++ won't let you start a
thread and qt not C++ standard. So the discussion about 'standard' is
academic.

Do I need to use cmpxchg (I'm on x86)?

If you want to write your synchronization construct by your own, feel
free to do so. But keep in mind that you will also need a signal to
notify the player thread about the new message. And you have to protect
the root pointer of the queue.

In any other way look for a thread safe queue class or simply use a mutex.


To give some more hints:

If you want your player to be really responsive in almost all cases you
need at last 3 or 4 threads.
- Thread 1 GUI, no I/O must be done here.
- Thread 2 the decoder, also feeds the sound and/or video device.
- Thread 3 Controller (queue consumer), controls the decoder state, also
when the decoder is currently busy or paused.
- Thread 4 Retrieves data asynchronously to prevent the multimedia
devices from running out of data while Thread 2 is blocking at an I/O
function, although enough data is in a buffer.

Furthermore the GUI might like to be informed about executed or failed
commands. So you need a command callback.

If your platform supports asynchronous I/O you will not need the 4th thread.

If you deal with large playlists, you might also want to start a worker
thread, that loads and stores information without passing the object to
the controller thread. In case the playlists can be nested, more than
one worker is recommended.


Marcel

 

[jonathan]

Hi Red,

I'm sure that Qt provides synchronization primitives.  You'd most
likely get better answers on a forum dedicated to Qt.

This is actually my current solution but I wanted to move away from
this. I figured Qt is written in C++ so there must be some underlying
code which does what I want without all that Qt-ness.

Thanks for your reply.

 

[jonathan]

Sam,
I can go the POSIX route, and probably will. I guess I was mostly
thinking: pthreads implements a mutex, so how do _they_ do it? If I
could just do that myself then why not? If only for the learning
experience.

Marcel,
It's true that the portability issue is moot because I'm using
threads in the first place. My thought was that I could design the
threads themselves to be "unaware" that they were threads. For
example, if the Qt libs are unavailable my player drops to the command
line and just starts playing song after song in one thread. No UI per
se; just a radio. So I can write everything in a "standard" way, and
all that naughty thread stuff will go in a source file with the UI,
which will not be especially portable anyway.

Thank you for your detailed response, too.
--Jonathan

 

[red floyd]

Hi Red,


This is actually my current solution but I wanted to move away from
this. I figured Qt is written in C++ so there must be some underlying
code which does what I want without all that Qt-ness.

Yeah, but it uses OS specific APIs. If you stick with Qt, you'll get
portability across all Qt platforms.

That said, C++03 does not include any multithreading features. C++0x,
when it's released, will have them, at which point you could use them.

 

[jonathan]

For anyone interested, this is what I ran off. Note that it uses an
__asm__ block...
Any comments welcome. This shouldn't be considered "robust".

--Jonathan

// header-ish stuff

enum event_type { // Types of messages that can be sent. Examples
only.
	IGNORE, PLAY, STOP, DIE };

struct Message { // Structure containing the message
	// Data
	event_type event;  // what kind of event is this?
	Message * next;    // Next event in the queue

	// Constructor for convenience
	Message(event_type e = IGNORE) : event(e), next(0) {};

};

class MessageQueue { // Event queue with some primitive "mutex" built
in
	public:
		MessageQueue() : mutex(0), root(0) {};
		void Post(event_type);
		bool Get(Message&); // Could implement as operator= prolly
	private:
		void lock() { // mutex lock
			unsigned long i = 0;
			do {
				__asm__ ( // NOTE: X86_64. Non-portable
					"lock bts $0, %1; sbbq %%rax, %%rax;"
					: "=a"(i) : "m"(mutex));
			} while (i != 0 && waitsome());
		}
		void unlock() { mutex = 0; }
		bool waitsome() { // Dummy sleep function. Fill it in.
			return true; // must return true
		};

		unsigned long mutex; // 0 is unlocked, anything else is locked
		Message * root; // Message queue
};

// Add a message to the queue of type "event".
void MessageQueue::Post(event_type event) {
	Message *m = new Message(event); // Create the message

	lock();
	if (root) { // Add the message to the end of the queue
		Message *q, *p = root;
		while (q = p->next) // find the end
			p = q;
		p->next = m;
	} else root = m; // There is no queue, add our message to the
beginning
	unlock();
}

// Check for a message in the queue. Non-blocking. Returns true if
there is
// (a message) and copies it to m. Returns false otherwise (and leaves
m alone)
bool MessageQueue::Get(Message& m) {
	bool was_an_event = false;

	lock();
	if (root) { // There's a message in the queue
		Message *n = root->next;

		m = *root;     // Copy the message to the user
		delete root;   // remove the message from the queue
		root = n;
		was_an_event = true;
	}
	unlock();
	return was_an_event;
}

// main.cpp

#include <cstdio>

MessageQueue m;
Message what_to_do;

int main(void) {
	m.Post(DIE);

	while (m.Get(what_to_do)) { // Enter the "event loop"
		switch(what_to_do.event) {
			case PLAY: printf("I can has play music?\n"); break;
			case STOP: printf("Boo. No music.\n"); break;
			default:   printf("WTF\n");
		}

	}

	return 0;
}

 

[James Kanze]

For anyone interested, this is what I ran off. Note that it uses an
__asm__ block...
Any comments welcome. This shouldn't be considered "robust".

The only real comment is: why aren't you using the system level
primitives. A CriticalSection (under Windows) or a
pthread_mutex_t (under Unix and look-alikes) shouldn't cost any
more if there is no contention, and will handle the waiting a
lot more elegantly (and probably with less total overhead) if
there is. A queue, of course, will require a conditional (I
don't know how this is done under Windows) if you actually want
to wait on it.

The usual solution is to wrap the locking itself in a class,
with the destructor releasing the lock, see boost::mutex, for
example. And boost::conditional.

There do exist non-locking algorithms as well, but they require
careful ordering and atomic access to specific variables in the
data structure itself (thus, asm access). The ones I'm familiar
with also only work on fixed length structures (which may be
perfectly acceptable in your case).

If you don't need to support wait, and use boost::mutex and
std::deque, the entire code should only be about 10 lines, and
will probably be considerably faster than your solution.
(Requiring dynamic allocation for an enum is NOT a very good
solution.)

 

[James Kanze]

Sam,
I can go the POSIX route, and probably will. I guess I was
mostly thinking: pthreads implements a mutex, so how do _they_
do it? If I could just do that myself then why not? If only
for the learning experience.

For starters, the implementation probably uses priviledged
instructions, which you can't use outside of kernel mode. (The
usual implementation will do a quick test and set outside of
kernel mode, but if that fails---there is contention---then it
will drop into kernel mode for the more complicated handling
necessary to suspend the thread.)

Marcel,
It's true that the portability issue is moot because I'm using
threads in the first place. My thought was that I could design
the threads themselves to be "unaware" that they were threads.
For example, if the Qt libs are unavailable my player drops to
the command line and just starts playing song after song in
one thread. No UI per se; just a radio. So I can write
everything in a "standard" way, and all that naughty thread
stuff will go in a source file with the UI, which will not be
especially portable anyway.

I'm not too sure how this would work under Windows, but the
classical Unix solution would be to use two processes (not
threads) for this. The basic player only supports command line
input, probably polling (e.g. function select or poll under
Unix, with a timeout of 0). The GUI is a separate process,
which starts the basic player, using a pipe to communicate with
it.

 

[Chris M. Thomasson]

The only real comment is: why aren't you using the system level
primitives. A CriticalSection (under Windows) or a
pthread_mutex_t (under Unix and look-alikes) shouldn't cost any
more if there is no contention, and will handle the waiting a
lot more elegantly (and probably with less total overhead) if
there is. A queue, of course, will require a conditional (I
don't know how this is done under Windows) if you actually want
to wait on it.

<pseudo-code>
_________________________________________________________________
void push(T* state) {
1) lock;
2) if (collection is empty) open gate;
3) push state into collection;
4) unlock;
}


T* pop() {
for (; {
1) lock;
2) if (collection is empty) {
2a) unlock;
2b) wait on gate;
2c) continue;
}
3) T* state = remove state from collection;
4) if (collection is empty) close gate;
5) unlock;
6) return state;
}
}
_________________________________________________________________




the gate is a manual-reset event. The lock is a CRITICAL_SECTION.

 

[jonathan]

Sam,

>This is one of those wheels that are not just worth reinventing.

Thing is I don't need the whole wheel. I just need mutual exclusion
for a few lines to add and remove from the queue. It's probable that
I'll end up using the existing libraries like pthreads, or boost. But
I think that's a bit of overkill for my specific task with this
specific application.

James Kanze,

why aren't you using the system level primitives.

I'm just looking to see if I can do it myself. If not, I'll use
existing libraries.

and will handle the waiting a lot more elegantly

Yeah.. that code above is pretty ugly. In my application, however,
waiting should basically be a non-issue. I'm handling user generated
messages, not system generated. Probably I'll only get events every
couple of minutes. At worst a few times a second if somebody is
hammering on the "next" button (it's a music player). Either way I
never have a lock on for very long.

Requiring dynamic allocation for an enum is NOT a very good solution.

Hahah.. no of course not. It was just an (nearly) empty structure
meaning to suggest the idea of an "event" along with associated data.
The exact mechanism for that part of it is beyond the scope of what I
was trying to demonstrate.

Chris M. Thomasson,
Thanks for your example. That's mostly what I had in mind... without
the "gate".. I'll look at that.