Apologies if this is a dumb question but I have some code that needs to do some intensive processing of real-time data on several thousand Order books - where each book can have several hundred/thousands of orders. The code is taking too long to perform the required processing and calculations - so I am trying to multithread the processing function. I have not written any serious C++ for over a decade and certainly not done any multithreading for even longer...
The problem is when the function is run via a thread, the orderbook data I think I am passing in, is not what the thread function receives.
int threadCnt = 5;
void processOrderBooks()
{
std::vector<std::thread> threadVector;
for (OrderBooks::iterator Ih = orderBooks.begin(); Ih != orderBooks.end();)
{
for (int threadIdx = 0; (threadIdx < threadCnt); threadIdx++)
{
OrderBook<OrderBookEntry>* orderCache = Ih->second;
threadVector.emplace_back([&]() {processOrderEntry(std::ref(*orderCache)); });
std::cout << "Thread started " << orderCache->OrderBookID() << std::endl;
if (++Ih == orderBooks.end())
break;
}
for (auto& t : threadVector)
{
t.join();
}
threadVector.clear();
}
}
void processOrderEntry(OrderBook<OrderBookEntry> orderCache)
{
std::cout << "Calc : " << orderCache.OrderBookID() << std::endl;
.... do some work....
}
However, the output I get from the above is:
Thread started 1301.T
Thread started 1332.T
Thread started 1333.T
Thread started 1375.T
Calc : 1333.T
Calc : 1375.T
Calc : 1375.T
Thread started 1376.T
Calc : 1376.T
Calc : 1376.T
Thread started 1377.T
Thread started 1379.T
Thread started 1380.T
Calc : 1380.T
Thread started 1381.T
Calc : 1381.T
....
Thread started 2268.T
Thread started 2198.T
Thread started 2224.T
Thread started 2300.T
Thread started 2229.T
Calc : 2300.T
Calc : 2198.T
Calc : 2229.T
Calc : 2229.T
Calc : 2229.T
As you can see, Calc is being executed multiple times for the same orderbook and not at all for others. Can someone please help me understand what I am doing wrong? Thanks in advance!
you need to capture the pointer by value, otherwise the thread is using a reference to a pointer on the stack that doesn't exist anymore when the thread starts.
threadVector.emplace_back([orderCache]() {processOrderEntry(std::ref(*orderCache)); });
^^^^^^^^^^
threads that outlive the current scope shouldn't capture stuff by reference, unless you can be certain that the reference will be valid for the duration of the thread. and in your case it is not.