Use assignment instead of memcpy
A better implementation of Obtain can return a non-const reference which you can assign to:
T& Obtain() {
return storage[curWriteNum & MASK];
}
Used like this:
void Processor::EnqueueFutOrderbook(orders* param)
{
orders& storageItem = futOrdersUpdates.Obtain();
storageItem = *param;
futOrdersUpdates.Commit();
}
That deals with the "use the copy constructor instead of memcpy" suggested by @ratchetfreak.
If you use memcpy instead of assignment then your code is incorrect for all types of T which have a non-trivial (e.g. user-defined) copy constructor.
Copy-construct will be slower than memcpy, isn't it?
The reason for using assignment is if T has a non-trivial copy-constructor (more accurately, an 'assignment operator'): for example if T is a std::auto_ptr<> or if it contains a std::string: if T has a non-trivial user-defined assignment operator, memcpy avoids calling it: which, may be faster but is incorrect behaviour!
In your example though your T is myStruct, which has a default (or non-existent) assignment operator. In that case the assignment is done by the compiler as an intrinsic, which is likely to be at least as fast as the run-time-library's memcpy function.
Which of the following do you think is faster:
void test()
{
T source, target;
// test the speed of `memcpy`
for (int i = 0; i; i < 100000)
memcpy(&target,&source,sizeof(T));
// test the speed of compiler-generated assignment with copy-constructor
for (int i = 0; i; i < 100000)
target = source;
}
An implementation which locks each method could be safe
One ('atomic') Store function would be easier to use than separate Obtain and Commit methods:
void Store(const T& newValue) {
storage[curWriteNum & MASK] = newValue;
++curWriteNum;
if (curWriteNum - curReadNum > length)
{
std::cout <<
"ArrayPool curWriteNum - curReadNum > length! " <<
curWriteNum << " - " << curReadNum << " > " << length << std::endl;
}
}
Instead of trying "lock-free" code, adding locks as follows would make it safe.
#pragma once
#include <stdint.h>
#include <iostream>
#include <mutex>
template<class T> class ArrayPool
{
public:
bool IsEmpty() {
std::lock_guard<std::mutex> lck (mtx);
return curReadNum == curWriteNum;
}
bool TryGet(T& output)
{
std::lock_guard<std::mutex> lck (mtx);
if (curReadNum == curWriteNum)
{
return false;
}
output = storage[curReadNum++ & MASK];
return true;
}
void Store(const T& newValue) {
std::lock_guard<std::mutex> lck (mtx);
if (((curWriteNum+1) & MASK) == (curReadNum & MASK))
{
std::cout <<
"ArrayPool curWriteNum - curReadNum > length! " <<
curWriteNum << " - " << curReadNum << " > " << length << std::endl;
throw std::exception();
}
storage[curWriteNum++ & MASK] = newValue;
}
private:
static const uint32_t length = 1 << 4;
static const uint32_t MASK = length - 1;
T storage[length];
uint32_t curWriteNum;
uint32_t curReadNum;
std::mutex mtx;
};
Modify your client functions to call it like this:
void ReadThread() {
myStruct entry;
while(true) {
if (pool.TryGet(entry)) {
std::cout << entry.value << std::endl;
}
}
}
void WriteThread(int id) {
std::chrono::milliseconds dura(1000 * id);
std::this_thread::sleep_for(dura);
myStruct storage;
storage.value = id;
pool.Store(storage);
std::cout << "Commited value! " << id << std::endl;
}
My primary requirement is latency. That's why I don't want to use lock, only memory barrier if absolutely required. And you don't ansered main question - if reader is guaranteed to see latest value. Should I insert memory barrier somewhere or something?
When I write code my primary requirement is always 'correctness' first. Usually the cost of a lock is:
- Trivial (unnoticeable) compared to whatever other processing the program is doing (for example, in your program, writing to
std::out).
- Worth it for thread-safety.
Your buffer overrun detection is broken
I marked the following with <strike> because it is not relevent if you are not trying to detect a buffer overrun.
This statement won't work after you store 2^32 items:
if (curWriteNum - curReadNum > length) { ... }
Instead use something like this, before you write to storage and increment curWriteNum:
if (((curWriteNum+1) & MASK) == (curReadNum & MASK))
{
std::cout <<
"ArrayPool curWriteNum - curReadNum > length! " <<
curWriteNum << " - " << curReadNum << " > " << length << std::endl;
throw std::exception(); // or, return false;
}
The following is very unsafe, because you return a pointer to memory which (because you've incremented ++curReadNum) you now think it is already safe to overwrite:
T* TryGet()
{
if (curReadNum == curWriteNum)
{
return NULL;
}
T* result = &storage[curReadNum & MASK];
++curReadNum;
return result;
}
Try something like this instead:
bool TryGet(T& output)
{
if (curReadNum == curWriteNum)
{
return false;
}
output = storage[curReadNum++ & MASK];
return true;
}
I not always have object which I can copy so I have to use Obtain.
In that case you can modify TryGet to return a pointer, however in that case you should delay incrementing curReadNum until after you have finished using the pointer:
T* TryPeek()
{
std::lock_guard<std::mutex> lck (mtx);
if (curReadNum == curWriteNum)
{
return NULL; // or 'return 0;'
}
return &storage[curReadNum & MASK];
}
// Call this after TryPeek() to discard most recently peeked
void Pop()
{
std::lock_guard<std::mutex> lck (mtx);
++curReadNum;
}
Which you can call like this:
void ReadThread() {
myStruct* entry;
while(true) {
if (entry = pool.TryPeek()) {
// Use the entry
std::cout << entry->value << std::endl;
// Discard the entry after finished using it
pool.Pop();
}
}
}
Thanks, I think no need to declare CurReadNum as volatile - buffer is never "overflow" so CurReadNum is actually read from one thread only.
That's true if (only if) you don't care about detecting buffer overruns.
So declaring CurReadNum as volatile is absolutely useless ...
If you have code to detect buffer overruns (and you did write some such code in your OP) then it's not "absolutely useless": it may be required, to make that code work correctly. A problem with multi-threaded code is that testing cannot prove that it's correct: testing can only prove that it's incorrect. Correctness needs to built-in, by design and code inspection.
... but will likely add some latency.
I doubt whether you can devise a performance test which, in practice, can detect any difference in latency.
Attempting a lock-free implementation
And main question still not answered - if declaring curWriteNum as volatile is enough on modern Intel Xeon processor? I'm using 2 phisical processors server BTW. Is volatile enough and mandatory? Or memory barrier must/can be better?
If it were my code and I wanted it to be thread-safe, then I would use some kind of lock.
I surely don't want to use lock as they are extremmely slow.
I think it's sufficient to put a memory fence at the start and end of every method, to emulate the memory fences which are implied by the lock_guard statements.
of course I can "guard" every method, but I want to find certain places where memory barrier is REQUIRED but not more? I do not want to use more barriers than required.
I think the worry is that a compiler and CPU are allowed to reorder statements, and could theoretically write an incremented value to curWriteNum before writing to storage.
If you remove the unwanted overrun-detection code from my Store method then, as you say, the only memory that's used by both threads are curWriteNum and storage; so:
#pragma once
#include <stdint.h>
#include <iostream>
template<class T> class ArrayPool
{
public:
bool IsEmpty() {
// 'data dependency' memory barrier here
// or not because we don't mind if this return 'false positive' because we'll check again later
return curReadNum == curWriteNum;
}
bool TryGet(T& output)
{
if (IsEmpty())
{
return false;
}
output = storage[curReadNum++ & MASK];
return true;
}
T* TryPeek() // Unlike TryGet this leaves the element in Storage
{
if (IsEmpty())
{
return 0;
}
return &storage[curReadNum];
}
void Pop() // Call this after a successful TryPeek
{
++curReadNum;
}
void Store(const T& newValue) {
storage[curWriteNum & MASK] = newValue;
// Ensure storage is written before mask is incremented
_MemoryBarrier();
++curWriteNum;
}
private:
static const uint32_t length = 1 << 4;
static const uint32_t MASK = length - 1;
T storage[length];
volatile uint32_t curWriteNum;
uint32_t curReadNum;
};
The reason why I also declared curWriteNum as volatile is that, for some compilers, volatile is is a hint that the variable should not be enregistered.
For example, without volatile, a test like while (curReadNum == curWriteNum) might be translated to:
mov eax,[curReadNum] ; move memory value to a CPU register
mov ebx,[curWriteNum] ; move other memory value to a different CPU register
label_loop_top:
cmp eax,ebc ; compare the two values-in-registers
je label_loop_top ; loop while the in-register values are equal
This code (caused by not declaring that curWriteNum is volatile) will never detect a subsequent write to curWriteNum.
With volatile, the same compiler might translate while (curReadNum == curWriteNum) to:
mov eax,[curReadNum] ; move memory value to a CPU register
label_loop_top:
cmp eax,[curWriteNum] ; compare with the in-memory value
je label_loop_top ; loop while it matches the in-memory value
This code (caused by declaring that curWriteNum is volatile) should eventually detect a subsequent write to curWriteNum.
If you don't trust that volatile is sufficient to do this job, then theoretically you should have a memory barrier in the IsEmpty function above (to ensure that it will read curWriteNum from memory and not from a register), as well as having one in Store (to ensure that storage is written before curWriteNum).
If you don't have this second memory barrier then theoretically a compiler or machine (perhaps a smarter one than the one you're using now) will read curWriteNum from memory on its first loop through IsEmpty(), notice that your ReadThread() code never modifies curWriteNum, and therefore assume it can thereafter reuse its old, cached, enregistered value of curWriteNum instead of reading it from memory again.
