Multi-Threaded Memory Preallocation

Question

I need to figure out the best way to deal with memory pre-allocation

Below is pseudo-code for what I am doing now, and it seems to be working fine.

I am sure there is a better way to do this, I would like to see if anyone has any good idea.

In Thread A I need to allocate 300 MB of memory and zero it out:

Thread A:

char* myMemory = new char[10*30*1024*1024];

In Thread B I incrementally get 10 sets of data 30 MB each. As I get this data it must be written to memory

Thread B:

int index1 = 0;
char* newData = getData(...); // get a pointer to 30 MB of data
memcpy(&myMemory[index1*30*1024*1024], &newData[0], 30*1024*1024];
SetEvent(...) // tell Thread C that I have written new data

// use index i as circular Buffer
if (index1 < 10)
 index1++;
else
 index1 = 0;

In Thread C, when new data is written, I need to get it and process it

Thread C:

int index2 = 0;
char* dataToProcess[] = new char[30*1024*1024];
if (event Fires) // if event in Thread B is set
 memcpy(&dataToProcess[0], &myMemory[index2*30*1024*1024], 30*1024*1024];
 processData(dataToProcess)

// use index i as circular Buffer
if (index2 < 10)
 index2++;
else
 index2 = 0;

Answer 1

Another approach I can think of could be this. Get rid of myMemory altogether. Once you receive data in thread B, allocate memory and copy that data (as you're doing now). Send thread C a message that data is received. Once thread C copies the data into its internal buffer, thread C can send message back to thread B that the data has been copied. Thread B then deallocate the buffer.

Your design seems fine to me. Yet another approach could be avoiding the data copy in thread C.

int index2 = 0;
char* dataToProcess;
if (event Fires) // if event in Thread B is set
 dataToProcess = &myMemory[index2*30*1024*1024];
 processData(dataToProcess)
// Rest of the code

Answer 2

I'm not sure how your threads work, but it looks like you need some more synchronization.

If B and C are a pair, I don't see indexing getting out of control, but have you thought about out-of-order execution and reordering of memory writes? B's CPU might execute code firing the event before completing the write to myMemory, resulting in C reading partial data. Granted, it may be highly unlikely or downright impossible in your case, but it's something to think about. You may need a memory barrier or a higher-level synchronization concept.

In fact, why aren't you using a message queue? Or something simpler, like a circular buffer? (Incidentally, boost::circular_buffer's thread-safe bounded buffer example looks very similar to your question.)

Or if you could attach the 30 MB data to the events, it would be a very simple solution.

Answer 3

you could use an approach like this (pseudocode);

template < typename T, size_t BlockSize_ >
class t_lockable_array {
public:
    static const size_t BlockSize = BlockSize_;
    typedef T t_object;

/* ...stuff... */

    void copyElements(std::vector& dest, const t_range& range) {
        t_lock_scope lock(this->mutex());
        /* copy @a range to @a dest here... */
    }

/* ...stuff... */

private:
    t_mutex d_mutex; /* assumption in this example: lock/unlock is a mutation */
    t_object d_object[BlockSize];
};

template < typename T, size_t BlockSize_, size_t BlockCount_ >
class t_partitioned_array {

public:
    static const size_t BlockSize = BlockSize_;
    static const size_t BlockCount = BlockCount_;
    static const size_t ObjectCount = BlockCount * BlockSize;
    typedef T t_object;

/* ...stuff... */

    t_lockable_array<t_object, BlockSize>& at(const size_t& idx) {
    /*
    direct access: each partition handles its locking/interface.
    alternatively, you could rewrap the lockable array's
    interface if you prefer encapsulation here.
    */
        return this->d_array[idx];
    }

/* ...stuff... */

private:
    t_lockable_array<t_object, BlockSize> d_array[BlockCount];
};

then just create a t_partitioned_array<char,10,ThirtyMB> on thread A

on thread B:

t_partitioned_array<char,10,ThirtyMB>& array(this->getSharedArray());
std::vector<char>& localBuffer(this->getLocalBuffer());

// the accessed partition will lock itself during read/write
array.at(this->partitionToRead()).copyElements(localBuffer, this->rangeOfMemoryToRead());

this->processData(localBuffer);