Действительно ли Boost_lockFree_SPSC_queue имеет переменный размер и потери данных?

Question

Привет, я планировал использовать дополнительную SPSC-очередь. У меня должен быть слушатель, который всегда прослушивает входящие сообщения, как только он его получит, поместит сообщение в очередь и вернется, чтобы снова ждать новых сообщений. И есть потребитель, который выдает сообщения из очереди и обрабатывает его.

Поскольку я не хочу, чтобы мой слушатель подождал, чтобы заставить замок записать в очередь, я предпочел boost_lockfree_spsc_queue. И я полагал, что размер очереди меняется, если я устанавливаю fixed_sized<false> при инициализации очереди. Но, похоже, это не так.

Ниже мой код:

boost::lockfree::spsc_queue<int, boost::lockfree::fixed_sized<false>> lockFreeQ{10}; 

void sharedQueue::lockFreeProduce() 
{ 

    for(int i = 0; i < 100; i++) 
    { 
     lockFreeQ.push(i);   
     cout<<"--"<<i<<endl; 
    } 

} 
void sharedQueue::lockFreeConsume(){ 

for(int i = 0; i <100; i++){ 

    /* Implement a blocking pop to the queue in order to waste cpu time by busy waiting*/ 

     while(!(lockFreeQ.read_available() > 0)) 
      { 
       //std::this_thread::yield(); 
       std::this_thread::sleep_for(std::chrono::nanoseconds(10)); 
      } 

     cout<<" ***"<<lockFreeQ.front()<<endl; 
     lockFreeQ.pop(); 

    } 

} 

void sharedQueue:: TestLockFreeQueue() 
{ 

    std::thread t1([this]() { this->lockFreeProduce(); }); 
    std::thread t2([this]() { this->lockFreeConsume(); }); 
    t1.join(); 
    t2.join(); 
} 

Выходные данные показывают, что потеря данных. Ниже приводится частичный результат. но он пропускает номер 87, 92, например.

**85 
***86 
***88 
***90 
***91 
***93 
***95 
***96 
***98 

Если это переменный размер, он должен расширять размер очереди и не должен потерять данные. Кажется, что он перезаписывает данные, как описано A consequence of the circular buffer is that when it is full and a subsequent write is performed, then it starts overwriting the oldest data.. В таком случае, как можно справиться с этим без потери данных?

Thanks

Answer 1

Перевести комментарии для ответа.

Путь политика работает spsc_queue, как показано ниже:

typedef typename mpl::if_c<runtime_sized,         
          runtime_sized_ringbuffer<T, allocator>, 
          compile_time_sized_ringbuffer<T, capacity> 
          >::type ringbuffer_type; 

В вашем случае он выбирает runtime_sized_ringbuffer буфер вместо compile_time_sized_ringbuffer.

Различие между обоими в том, что, в случае runtime_sized_ringbuffer массив размера (передается в качестве аргумента в конструктор) выделяется с использованием allocator передается в качестве параметра шаблона и в случае compile_time_sized_ringbuffer массив выделяется на стек.

Итак, если размер вашей очереди большой, вам нужно использовать runtime_sized_ringbuffer вместо compile_time_sized_ringbuffer.

Это единственное отличие, базовая структура данных - это circular_buffer в обоих случаях. то есть даже в случае runtime_sized_ringbuffer, самый старый элемент будет перезаписан после того, как вы закончите запись ко всем местоположениям/индексу в буфере. Это довольно прямо вперед из кода ниже

static size_t next_index(size_t arg, size_t max_size) 
{ 
    size_t ret = arg + 1; 
    while (unlikely(ret >= max_size)) 
    ret -= max_size; 
    return ret; 
} 

Если вам действительно нужно стопорное структуру данных бесплатно список, вы должны смотреть на знаменитый замок свободной структуры данных списка по Dmitry Vyukov