Eu tenho dois threads, um é o produtor e o outro é o consumidor. Meu consumidor está sempre atrasado (devido a alguma chamada de função cara, simulada no código abaixo usando sleeps), então usei o buffer de anel porque posso perder alguns eventos.
Estou procurando ver se meu bloqueio parece bom e comentários de revisão geral do c ++.
#include <iostream> #include <thread> #include <chrono> #include <vector> #include <atomic> #include <boost/circular_buffer.hpp> #include <condition_variable> #include <functional> std::atomic<bool> mRunning; std::mutex m_mutex; std::condition_variable m_condVar; class VecBuf { private: std::vector<int8_t> vec; public: VecBuf() = default; VecBuf(std::vector<int8_t> v) { vec = v; } }; std::vector<int8_t> data{ 10, 20, 30 }; class Detacher { public: template<typename Function, typename ... Args> void createTask(Function &&func, Args&& ... args) { m_threads.emplace_back(std::forward<Function>(func), std::forward<Args>(args)...); } Detacher() = default; Detacher(const Detacher&) = delete; Detacher & operator=(const Detacher&) = delete; Detacher(Detacher&&) = default; Detacher& operator=(Detacher&&) = default; ~Detacher() { for (auto& thread : m_threads) { thread.join(); } } private: std::vector<std::thread> m_threads; }; void foo_1(boost::circular_buffer<VecBuf> *cb) { while (mRunning) { std::unique_lock<std::mutex> mlock(m_mutex); m_condVar.wait(mlock, [=]() { return !cb->empty(); }); VecBuf local_data(cb->front()); cb->pop_front(); mlock.unlock(); if (!mRunning) { break; } //simulate time consuming function call and consume local_data here std::this_thread::sleep_for(std::chrono::milliseconds(16)); } while (cb->size()) { VecBuf local_data(cb->front()); cb->pop_front(); if (!mRunning) { break; } } } void foo_2(boost::circular_buffer<VecBuf> *cb) { while (mRunning) { std::unique_lock<std::mutex> mlock(m_mutex); while (cb->full()) { mlock.unlock(); /* can we do better than this? */ std::this_thread::sleep_for(std::chrono::milliseconds(100)); mlock.lock(); } cb->push_back(VecBuf(data)); m_condVar.notify_one(); } } int main() { mRunning = true; boost::circular_buffer<VecBuf> cb(100); Detacher thread_1; thread_1.createTask(foo_1, &cb); Detacher thread_2; thread_2.createTask(foo_2, &cb); std::this_thread::sleep_for(std::chrono::milliseconds(20000)); mRunning = false; } Resposta
/* can we do better than this? */
Em um contexto de buffer circular, a maneira padrão de evitar a espera ocupada é ter dois semáforos. Primeiro, para bloquear um produtor quando um buffer estiver cheio e, segundo, para bloquear um consumidor quando o buffer estiver vazio. Depois que um processo passa por seu semáforo e faz seu trabalho, ele deve sinalizar o par.
O buffer circular é bom quando o consumidor só se atrasa algumas vezes e você não pode permitir a perda de dados. Na sua situação, parece uma solução errada: o produtor fica estrangulado pela taxa de consumo e o consumidor é apresentado com os dados desatualizados.
Uma resposta típica é deixar o produtor funcionar em velocidade total e buffer triplo da produção (pelo menos, garante que o consumidor obterá os dados produzidos mais recentemente). Perdoe a desavergonhada autopromoção .
Comentários
- obrigado pela revisão comentário e o link. Porém, no meu caso meu consumidor só bloqueia algum tempo. Acho que ainda faz sentido usar semáforos em vez da minha solução atual usando variável de condição + bloqueio.
- @nomanpouigt " algum dia " é bem diferente de " sempre atrasado ".
- desculpe, quero dizer o buffer de anel o tamanho é escolhido de forma a poder acomodar o atraso. Querendo saber em c ++ não há semáforos e é implementado usando mutexes e variável de condição, então se faz sentido usar semáforos.
- Você também pode elaborar sua solução de buffer tripple e como ela se aplicaria no meu caso ? Quer dizer que não ' não faça nenhum bloqueio no lado do produtor e deixe-o rodar e o buffer do produtor precisa ser armazenado em buffer triplo para que o consumidor possa obter dados recentes em caso de saturação?