Copying memory between two asynchronous threads in c++

Lu[ifer

I have a program with two asynchronous threads running simultaneously. Basically thread A continually recieves some data, and thread B is continually looking for this data. Let me try and explain this a bit better...

//Thread A
void HereIsData(/*in*/ int length, /*in*/ char* data){}

//Thread B
void GetData(/*out */ int* length, /*out*/ char* data) {}

So I want to copy the data from HereIsData() to GetData().

I tried something like this....

int gLength; //g for global...
char* gData;


//Thread B
void HereIsData(/*in*/ int length, /*in*/ char* data)
{
  gLength = lenght;
  gData = data;
  SetEvent(...);
  WaitForEvent(...);
}

//Thread B
void GetData(/*out */ int* length, /*out*/ char* data) 
{
  WaitForEvent(...)
  *length = gLength;
  memcpy(data, gData, gLength);
  SetEvent(...);
}

The problem here is the amount of blocking going on.
I must block in the HereIsData() function to ensure that the global varaibles are copied in the GetData() function before this function is called again. Obviously I must block in the GetData() funciton to wait till more data is ready. I would prefer implement some form of queue to better improve efficency of this code. Also, if the data becomes sufficenly large, it will be necessary to write to buffer to disk. I would appreciate any suggestions as to how best to do this buffering to disk.

- Darragh

flav0rflav

Originally posted by Lu[ifer
I would prefer implement some form of queue to better improve efficency of this code.

Expand on this. Why do you believe it is inefficient? It is quite simple and doing what is needed. Perhaps some more info would help. Are there other jobs to be done? Is this on single or multiple processors? How in-efficient is it? How much more efficient do you want it to be/think it can be?

You could use a single writer, single reader queue, but you still need some event/signal passing.

Lu[ifer

Specifically the GetData() function runs in a larger context which i don't want to block unnecessarily. So, one improvement i could make would be to simply return with no data if there's no data to be had. If i was to implement a queue, how would i go about? What i don't understand to well is how to synchronize access to it, ie. i doubt i should allow one thread to add an element, while another is removing an element, etc. Is there some sort of i locks i need to use?

flav0rflav

You can use a structure as follows.

A queue of buffers in order ie. an array or linked list.

A control structure for each queue with a an in, out and last pointers/indexes.

One thread will update the in pointer, wrapping when hitting the last, checking it doesn't loop and bump into the out pointer.

The other thread will read from the out pointer and update it as far as in the in pointer, including wrapping.

(With a linked list is a circle you don't need the last pointer. With an array you need the last pointer.)

It looks something like:

|

---

out

---

in

---

last|

With in always moving ahead of out.

You can either send an event from the writer to the reader, or just let the reader poll whenever it feels like it.

Oh and think about what you want it to do if the queue fills up, ie in loops around and catches up with out.

There is probably some better examples/descriptions/code on the net. 'single reader writer queue threads' eg http://groups.google.ie/groups?q=single+reader+writer+queue+thread&hl=en&lr=&ie=UTF-8&selm=Yq_Ea.625949%24Si4.568451%40rwcrnsc51.ops.asp.att.net&rnum=2

Talliesin

A lot of this depends on the synchronisation primatives you have. Neither C++ nor the standard library have any built-in synchronisation features. IIRC POSIX does, so your best bet for compatibility may be to go with them. Myself I write thin wrappers around the Windows primatives, the idea being that I can rewrite the wrappers if I ever have to port, but I haven't actually ported such code so just how well that would work in practice is hard to say.

Now, there are some primatives on some systems that enable an attempt to obtain a lock to fail immediately as you gave for one of your ideas (in the Windows world NT4.0, 2000 and XP provide this, the others don't - if you are using WindowsNT look at the TryEnterCriticalSection function).
There are some primatives that allow fast thread-safe manipulation of an integer value, in cases where you know only two threads will deal with the shared resource this can (with some care) be used to enable ownership checking with less blocking. It can also be used for one thread to inform another about where it is dealing with, this means a writing thread can know that the reading thread is not using the range it wants to and that it can write to it without locking first. This is tricky to get right though.

Some systems also have primatives and/or functions for exactly this kind of shared buffer reading and writing and it may be worth seeing if yours does.

BTW. On the subject of queues, unless you have a good reason not to it is advisable to add thread-safety to something built on an std::queue or std::deque rather than creating a thread-safe queue from scratch. All C++ programmers should know how std::queue and std::deque work so it will be easier for someone else (or yourself after enough time has passed for you to forget what you were doing) to work with the code if you use them. If you can't use them at least try to have a similar interface so it behaves like the STL types.

Lu[ifer

I'm going to have a look at the std::queue container.
Is this along the right track?

queue<MyData> q;
CCritSec csMyLock;  // Critical section is not locked yet.

//Thread A
void HereIsData(/*in*/ int length, /*in*/ char* data)
{
  CAutoLock cObjectLock(&csMyLock);  // Lock the critical section.  
  MyData d = new MyData(length, data);
  q.push(d);
}  // Lock goes out of scope here.

//Thread B
void GetData(/*out */ int* length, /*out*/ char* data) 
{
  CAutoLock cObjectLock(&csMyLock);  // Lock the critical section.
  if (q.empty()) return;
  MyData d = q.front();
  *length = d.length;
  memcpy(data, d.data, d.length);
  q.pop();
}  // Lock goes out of scope here.

Will this make the access to the queue thread-safe? The CCritSec and CAutoLock classes are from the DirectX sdk (which I'm using anyway; the topic here is VoD).

If I want to also buffer MyData to disk, anyone got suggestions for this?
I I overload << and >> can I write something like...

fstream f("Filename");
CCritSec csMyLock;  // Critical section is not locked yet.

//Thread A
void HereIsData(/*in*/ int length, /*in*/ char* data)
{
  CAutoLock cObjectLock(&csMyLock);  // Lock the critical section.  
  MyData d = new MyData(length, data);
  f << d;
}  // Lock goes out of scope here.

//Thread B
void GetData(/*out */ int* length, /*out*/ char* data) 
{
  CAutoLock cObjectLock(&csMyLock);  // Lock the critical section.
  MyData d;
  *length = d.length;
  memcpy(data, d.data, d.length);
  d << f;
}  // Lock goes out of scope here.

Basically I need std::queue functionality, but on disk instead on in memory.

-Darragh

Talliesin

That seems correct at first glance. One possible improvement would be to tie the queue and the critical section together in a class so that the class handles the critical secion management ensuring that there is no way for a piece of code to use it without obtaining the lock (doing this is trickier than it appears at first if any of the new classes members return pointers or references into the queue though). Simply wrapping the code you have there in a class and making HereIsData and GetData public members should do this.

Other possible improvements would be to test whether you could write or read without obtaining a lock (if the read and write areas where suitable distant from each other). This would require a few things, firstly you would need a data structure that wouldn't move its representation in memory if a write caused it to exceed its current buffer. By default std::queue is implemented by an std::deque, and deque's will potentially rewrite it's internal structure, invalidating all iterators. An std::list will not, and can also be used as the implementation of an std::queue:

std::queue<MyData, std::list<MyData> > q;

(note the space between the first and second ">").

There are quite a few issues to attempting this optimisation. Unless you find locking is happening a lot I would avoid it.

A half-way solution is to have locks on sections of a shared buffer. This can lead to improvements, but it can just as easily lead to performance decreases (if calls frequently need more than one lock).

BTW, you don't seem to have any bounds checking to ensure that GetData doesn't overflow the buffer sent to it. I would rewrite it as something like:

size_t GetData(/*in*/ size_t length, /*out*/ char* data) 
{
  CAutoLock cObjectLock(&csMyLock);  // Lock the critical section.
  if (q.empty()) return 0;
  MyData d = q.front();
  size_t retLen = std::max(length, d.length)
  memcpy(data, d.data, retLen);
  q.pop();
  return retLen;
}  // Lock goes out of scope here.

Another alternative would be to return some type which managed the data by storing a pointer and size.

As for file operations your suggested code seems okay at first glance except that in GetData you are using d << f instead of f >> d (streams always go to the left of << or >>, while I can see what you mean by d << f it isn't done that way because there would be some cases where it wasn't as clear), and you are extracting data from the (default) MyData before you read into it, you should make f >> d the very next operation after you create d (in some cases it's worth using a constructor that takes an ostream& either for efficiency reasons, or because there is no suitable default state).
It's also important that you flush and sync the stream at appropriate points (flushing after each write and synching before each read is a conservative approach that should be safe, albiet expensive).

If you are using Windows take a look at the OVERLAPPED structure and the calls that use it. It is designed for managing asychronous access to files and while it requires some lower-level operations it has strong advantages if you don't need to worry about other systems.