lockfree spsc_queue Sınıfı

Giriş
wait free ya da lockfree şu anlama gelir. Yani okuyan veya yazan thread hiçbir zaman bloke olmaz.

A wait-free implementation of a concurrent data object is one that guarantees that any process can complete any operation in a finite number of steps, regardless of the execution speeds of the other processes.

spc kelimei single producer single consumer anlamına gelir. Bu sınıf kendi içinde bir circular buffer içerir. Dolayısıyla veri kaybına sebep olabilir. Bence bu sınıfı kullanmamak lazım. Hatta bu sınıfın ismi ring_buffer olsa daha iyi olurdu.

So, to conclude, you must choose a sufficiently high value while constructing the queue which can ensure in best case that the data would not be overwritten. Basically you cannot guarantee this behaviour using a circular queue because you never know when the consumer would crash or start processing slowly.

T için kurallar şöyle

T must have a default constructor
T must be copyable

Şu satırı dahil ederiz.

#include <boost/lockfree/spsc_queue.hpp>

Tanımlama
Örnek
Şöyle yaparız. Böylece circular buffer stack'te tanımlanır.

boost::lockfree::spsc_queue<Foo, boost::lockfree::capacity<10>> q;

Örnek
Şöyle yaparız.

typedef boost::lockfree::spsc_queue<
        double, 
        boost::lockfree::capacity<200> 
    > q;

Örnek
Şöyle yaparız. Böylece circular buffer heap'te tanımlanır.

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

Örnek
Şöyle yaparız.

namespace bip = boost::interprocess;

namespace shm
{
  using char_alloc = bip::allocator<char, bip::managed_shared_memory::segment_manager>;
  using shared_string = bip::basic_string<char, std::char_traits<char>, char_alloc>;
  using ring_buffer = boost::lockfree::spsc_queue<shared_string,
    boost::lockfree::capacity<200>>;
}

shared memory yaratmak için şöyle yaparız.

pShm = std::make_unique<bip::managed_shared_memory>(
  bip::open_or_create, "Queuing", rSHMSize);

spsc nesnesini yaratmak için şöyle yaparız.

auto pSharedMemAddr = pShm->construct<
   shm::ring_buffer>("MyQueue")[1](/*aMaxNumMessages*/);

Constructor - capacity + std::allocator
Pointer tanımlamak için .h dosyasında şöyle yaparız.

boost::lockfree::spsc_queue<foo>* myQueue;

.cpp dosyasında şöyle yaparız.

std::allocator<myStruct> alloc;
myQueue = new boost::lockfree::spsc_queue<foo>(512,alloc);

front metodu

Şöyle yaparız. İlk elemanın referansını alırız.

int readValue = q.front();

is_lock_free metodu
Şöyle yaparız

if (!q.is_lock_free()) {...}

pop metodu
Şöyle yaparız. İşlem başarılı ise true, başarısız ise false döner.

q.pop(); 

pop - value
Örnek
Kuyrukta int değerler varsa şöyle yaparız.

int value;
while (q.pop(value)) {
 ...
}

Örnek
Kuyrukta double değerler varsa şöyle yaparız.

double v;
if (q->pop(v)) {...}

pop metodu - array
Şöyle yaparız. Kaç tane eleman silindigini döner.

int popped; //actual amount of popped objects
const int pop_amount = 10000;
Foo foos [pop_amount];

//pop objects from buffer
popped = q.pop (foos, pop_amount);

push metodu
Açıklaması şöyle

bool push(T const & t);

Pushes object t to the ringbuffer.

Note: Thread-safe and wait-free

İşlem başarılı ise true, başarısız ise false döner. Şöyle yaparız. Circular buffer kullandığı için eğer kuyruk doluysa en eski elemanın üzerine yazabilir.

Foo foo = ...
q.push (foo);

Yazma işleminin gerçekleşmesinden emin olmak için bu metodu döngü içinde kullanmak gerekir. Şöyle yaparız.

while (!q.push(value))
;

read_available metodu
Şöyle yaparız.

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