Closed zhouqi1727 closed 2 years ago
hosseinmoein
commented
2 years ago You should read the multithreading section of the documentation and see the code samples it points you to.
DataFrame is not multithreaded safe. You need to provide a SpinLock and your own mutex, if you are using the same instance of the DataFrame in multiple threads.
Also, I noticed in the sales thread
if(dataframere.get_index().size()>100)is not inside the mutex protection
hosseinmoein
commented
2 years ago BTW, what is the error?
zhouqi1727
commented
2 years ago the error is Segmentation fault in dataframe.append_column,i have provide a SpenLock in thread ,thanks.
zhouqi1727
commented
2 years ago Thread 6 "main" received signal SIGSEGV, Segmentation fault.
[Switching to Thread 0x7ffff585d700 (LWP 446988)]
0x00005555555a58d7 in gnu_cxx::__enable_if<std::is_scalar
()i update my code like this:
`
mutex mutex1,mutex2;
SpinLock lock;
producer:
while(true){
lock_guard
sales:
while(1){
if(dataframere .get_index().size()>100){
std::lock_guard
but i have got this error:
Thread 7 "main" received signal SIGSEGV, Segmentation fault.
[Switching to Thread 0x7ffff505c700 (LWP 454182)]
0x00005555555af84b in std::detail::_Hash_code_base<hmdf::HeteroVector const, std::pair<hmdf::HeteroVector const const, std::vector<double, std::allocator
hosseinmoein
commented
2 years ago I really don't know the logic of your program, so I am guessing here. If you only have one instance of DataFrame in your entire process and you are protecting it with your own mutex, then you do not need a SpinLock.
Again, I don't know your logic but from what I can tell, you want something like this:
std::mutex mutex1;
// producer:
while (true) {
std::lock_guard<std::mutex> lk (mutex1);
if (dataframe.get_index().size() > 300) {
auto index = dataframe.get_index()[0];
auto endIndex = dataframe.get_index()[99];
dataframe.remove_data_by_idx({ index, endIndex });
continue;
}
dataframe.append_index ((time + 1) * interval);
dataframe.append_column ("agg_price",agg_price);
dataframe.append_column ("agg_vol",agg_vol);
dataframe.append_column ("buy_price",buy_price);
dataframe.append_column ("sell_price",sell_price);
}
// sales:
while (true) {
{
std::lock_guard<std::mutex> lk (mutex1);
if (dataframere.get_index().size() > 100) {
new_dataframere = dataframere ;
vec = get_feats(new_dataframere ,"b");
}
}
usleep(10000);
}I've run the scheme you gave me before, but this scheme is not available to get_feats will not run, that is, it will not enter the sales process and will always be locked in the producer.
hosseinmoein
commented
2 years ago As I said I don’t know the logic of your process. So you have to adjust the code I gave you. Probably you should also sleep in the producer
One thread produces data and one thread processes data. When only the production thread is running, the program can run for a long time, such as more than one day, but when the processing thread is running at the same time, an error will be reported in a few hours. The error is reported in append_ Column, the following is the code of the two functions： producer:
while(true){ lock_guard<std::mutex> lk(mutex1); if(dataframe.get_index().size()>300){ try{ auto index = dataframe.get_index()[0]; auto endIndex = dataframe.get_index()[99]; dataframe.remove_data_by_idx<double>({index,endIndex}); }catch(const std::exception& e){ cout<<"dataframere move data error:"<<e.what()<<endl; } continue; } dataframe.append_index((time+1)*interval); dataframe.append_column("agg_price",agg_price); dataframe.append_column("agg_vol",agg_vol); dataframe.append_column("buy_price",buy_price); dataframe.append_column("sell_price",sell_price);sales:while(1){ if(dataframere .get_index().size()>100){ std::lock_guard<std::mutex> lck2(mutex2); new_dataframere = dataframere ; vec = get_feats(new_dataframere ,"b"); } usleep(10000); }The two functions are executed in different threads.