Task: calculate average price for each stock, using input data, which consists of stock id (int) and its price (float). It's important to detect jumps in prices, therefore need to keep correct sequence of input data, i.e. prices for the same stock id should be processed in correct sequence (from top to bottom).
Stock id# = [1 - 6], Price - any. Example data (txt format):
2 12.6
4 22.8
3 60.3
4 22.2
1 5.1
2 11.6
3 60.9
4 21.2
Solution: Idea is to use one thread per stock id, i.e. process prices for each stock id in a single thread. Assuming 6 cores are available.
I've tried to adopt producer/consumer idiom. Using std::queue
to store prices from producer and launched separate thread per stock id. Each thread retrieving data from corresponding queue and calculates the mean. (And they could do other calculations as well, like sigma jump detection)
class ThrQueue
provides std::queue
per stock id/thread, with thread safe access, and I used std::map
to map stock id vs ThrQueue
object.
To compile:
g++ -std=c++1y -pthread filename.cpp
The code below seemingly works, but not sure if its the best practice or if there are no hidden problems that I couldn't detect. Also, time delays here and there are just experimental, not sure why..? Any comments/suggestions are welcome! (please let me know in comments if anything needs to be clarified)
#include<chrono>
#include<future>
#include<condition_variable>
#include<map>
#include<unordered_map>
#include<queue>
#include<fstream>
#include<iostream>
using namespace std;
#define NTHR 6
bool go(true);
float results[NTHR] = {0};
template<typename T>
class ThrQueue {
public:
ThrQueue(T&& t) {
m_queue.push(std::forward<T>(t));
}
~ThrQueue() {
if(!m_queue.empty()) cout << " ERROR: queue not empty" << endl;
}
T pop() {
std::unique_lock<std::mutex> lock(m_mutex);
m_cv.wait(lock, [this]{return !m_queue.empty();});
auto it = m_queue.front() ;
m_queue.pop() ;
return it;
}
void push(T const& x) {
{
std::unique_lock<std::mutex> lock(m_mutex);
m_queue.push(x) ;
}
m_cv.notify_one();
}
bool empty() {
std::unique_lock<std::mutex> lock(m_mutex);
return m_queue.empty() ;
}
private:
std::queue<T> m_queue ;
std::mutex m_mutex ;
std::condition_variable m_cv;
};
template<typename F, typename... Ts>
inline auto reallyAsync(F&& f, Ts&&... params){
return std::async(
std::launch::async,
std::forward<F>(f),
std::forward<Ts>(params)...);
}
template<typename T>
void consumer(int n, T it) {
float avg = 0; int count = 0;
while(go) {
while(!it->second.empty()){
avg += it->second.pop() ;
count++;
}
std::this_thread::sleep_for(std::chrono::milliseconds(2)); // ??
}
results[n-1] = avg/count ;
std::this_thread::sleep_for(std::chrono::milliseconds(2)); // ??
}
int main() {
unsigned short n(0), count(0) ; float x(0);
std::future<void> futs[6];
std::unordered_map<int, ThrQueue<float>> queue_map ;
std::ifstream infile("test_sig_in.txt");
if(infile.is_open()){
while(infile >> n >> x) {
//cout << " " << n << " " << x << std::flush;
auto it = queue_map.find(n);
if(it != queue_map.end()) {
it->second.push(x) ;
}
else {
cout << " launch async for id: " << n << endl;
queue_map.emplace(std::piecewise_construct,
std::forward_as_tuple(n),
std::forward_as_tuple(std::move(x)));
futs[count++] = reallyAsync(&consumer<decltype(queue_map.find(n))>, n, queue_map.find(n));
}
}
}
infile.close();
std::this_thread::sleep_for(std::chrono::milliseconds(10)); // ??
go = 0;
for(int i=0; i<NTHR; ++i) {
while(futs[i].wait_for(10ms) != std::future_status::ready) {cout << " wait for " << i << endl;}
cout << " exit consumer: " << i << " avg: " << results[i] << endl;
}
}