pmem::obj::experimental::mpsc_queue Class Reference

Persistent memory aware implementation of multi producer single consumer queue. More...

#include <libpmemobj++/experimental/mpsc_queue.hpp>

Classes
class	batch_type
	Type representing the range of the mpsc_queue elements. More...
class	pmem_log_type
	Type representing persistent data, which may be managed by mpsc_queue. More...
class	worker
	mpsc_queue producer worker class. More...

Public Member Functions
	mpsc_queue (pmem_log_type &pmem, size_t max_workers=1)
	mpsc_queue constructor. More...
worker	register_worker ()
	Registers the producer worker. More...
template<typename Function >
bool	try_consume_batch (Function &&f)
	Evaluates callback function f() for the data, which is ready to be consumed. More...

Detailed Description

Persistent memory aware implementation of multi producer single consumer queue.

In case of crash or shutdown, reading and writing may be continued by new process, from the last position without loss of any, already produced data.

Note

try_consume_batch() MUST be called after creation of mpsc_queue object if pmem_log_type objcect was already used by instance of mpsc_queue - e.g. in previous run of application. If try_consume_batch() is not called, produce may fail, even if the queue is empty.

 
struct root {
    pmem::obj::persistent_ptr<
        pmem::obj::experimental::mpsc_queue::pmem_log_type>
        log;
};
 
void
single_threaded(pmem::obj::pool<root> pop)
{
 
    std::vector<std::string> values_to_produce = {"xxx", "aaaaaaa", "bbbbb",
                              "cccc", "ddddddddddd"};
    pmem::obj::persistent_ptr<root> proot = pop.root();
 
    /* Create mpsc_queue, which uses pmem_log_type object to store
     * data. */
    auto queue = pmem::obj::experimental::mpsc_queue(*proot->log, 1);
 
    /* Consume data, which was stored in the queue in the previous run of
     * the application. */
    queue.try_consume_batch(
        [&](pmem::obj::experimental::mpsc_queue::batch_type rd_acc) {
            for (pmem::obj::string_view str : rd_acc) {
                std::cout << std::string(str.data(), str.size())
                      << std::endl;
            }
        });
    /* Produce and consume data. */
    pmem::obj::experimental::mpsc_queue::worker worker =
        queue.register_worker();
    for (std::string &value : values_to_produce) {
        /* Produce data. */
        worker.try_produce(value);
        /* Consume produced data. */
        queue.try_consume_batch(
            [&](pmem::obj::experimental::mpsc_queue::batch_type
                    rd_acc) {
                for (pmem::obj::string_view str : rd_acc) {
                    std::cout << std::string(str.data(),
                                 str.size())
                          << std::endl;
                }
            });
    }
    /* Produce data to be consumed in next run of the application. */
    worker.try_produce("Left for next run");
}
 

Constructor & Destructor Documentation

mpsc_queue()

pmem::obj::experimental::mpsc_queue::mpsc_queue	(	pmem_log_type &	pmem,
		size_t	max_workers = 1
	)

mpsc_queue constructor.

Parameters

[in]	pmem	reference to already allocated pmem_log_type object
[in]	max_workers	maximum number of workers which may be added to mpsc_queue at the same time.

Member Function Documentation

register_worker()

mpsc_queue::worker pmem::obj::experimental::mpsc_queue::register_worker ( )

inline

Registers the producer worker.

Number of workers have to be less or equal to max_workers specified in the mpsc_queue constructor.

Returns

producer worker object.

    pmem::obj::experimental::mpsc_queue::worker worker =
        queue.register_worker();

try_consume_batch()

template<typename Function >

bool pmem::obj::experimental::mpsc_queue::try_consume_batch ( Function &&  f )

inline

Evaluates callback function f() for the data, which is ready to be consumed.

try_consume_batch() accesses data, and evaluates callback inside a transaction. If an exception is thrown within callback, it gets propagated to the caller and causes a transaction abort. In such case, next try_consume_batch() call would consume the same data.

Returns

true if consumed any data, false otherwise.

Exceptions

transaction_scope_error

Note

try_consume_batch() MUST be called after creation of mpsc_queue object if pmem_log_type objcect was already used by any instance of mpsc_queue. Otherwise produce might fail even if the queue is empty)

See also

mpsc_queue::worker::try_produce()

    queue.try_consume_batch(
        [&](pmem::obj::experimental::mpsc_queue::batch_type rd_acc) {
            for (pmem::obj::string_view str : rd_acc) {
                std::cout << std::string(str.data(), str.size())
                      << std::endl;
            }
        });

---

The documentation for this class was generated from the following file:

• libpmemobj++/experimental/mpsc_queue.hpp