Mpsc queue

[karczex]

This PR is very far from being ready.
Anyhow, interesting part for now is mpsc_queue.hpp api, concurrent test for it (I know it is not real test) and integration of mpsc_queue into multitier example (cherry-picked from #1066)

TODO:

• ringbuf/utils.h should be removed, and std::atomic should be used instead of custom implementation
• recovery - current implementation just doesn't work
• Add more/better tests
• move mpsc_queue to experimental directory
• Fix license

---

This change is 

[igchor]

Reviewable status: 0 of 8 files reviewed, 4 unresolved discussions (waiting on @karczex)

---

examples/radix_tree/multi_tier_map.cpp, line 79 at r3 (raw file):

		ld.value = v;

		auto worker = queue.register_worker();

I guess the worker should be only registered once - in the map ctor

---

include/libpmemobj++/container/mpsc_queue.hpp, line 126 at r3 (raw file):

	ringbuf_t *ring_buffer;
	static pmem::detail::id_manager id_manager;

not needed since you have get_id_manager function?

---

include/libpmemobj++/container/mpsc_queue.hpp, line 135 at r3 (raw file):

	mpsc_queue(pmem::obj::persistent_ptr<char[]> *log, size_t buff_size,size_t max_workers=1)

why persistent_ptr* and not just persistent_ptr?

---

include/libpmemobj++/container/mpsc_queue.hpp, line 143 at r3 (raw file):

		ringbuf_get_sizes(max_workers, &ring_buffer_size, NULL);
		ring_buffer = (ringbuf_t*) malloc(ring_buffer_size);

I think we shouln't use malloc in C++. It would be better to either allocate char array (new char[ring_buffer_size]) or use approach suggested here: https://stackoverflow.com/questions/184537/in-what-cases-do-i-use-malloc-and-or-new

void *p = operator new(size);
   ...
operator delete(p);

[igchor]

In the commit in which you are importing the ringbuffer you could add a origin repo URL and a commit id

Reviewable status: 0 of 8 files reviewed, 4 unresolved discussions (waiting on @karczex)

[KFilipek]

Reviewed 1 of 6 files at r1, 1 of 2 files at r2, 1 of 3 files at r3.
Reviewable status: 3 of 8 files reviewed, 6 unresolved discussions (waiting on @karczex)

---

examples/radix_tree/multi_tier_map.cpp, line 24 at r3 (raw file):

	pmem::obj::p<size_t> data_1;
	pmem::obj::p<size_t> data_2;
	pmem::obj::p<size_t> data_3;

Why not

> pmem::obj::p<size_t> data_[3];

---

examples/radix_tree/multi_tier_map.cpp, line 31 at r3 (raw file):

		return data_1 == rhs.data_1 && data_2 == rhs.data_2 &&
			data_3 == rhs.data_3;

then:

for (int i = 0; i < 3; i++)
    if (data_[i] != rhs.data_[i]) return false;
return true;

[igchor]

Reviewable status: 3 of 8 files reviewed, 7 unresolved discussions (waiting on @karczex)

---

include/libpmemobj++/container/detail/ringbuf/utils.h, line 106 at r3 (raw file):

#define SPINLOCK_BACKOFF_HOOK
#endif
#define	SPINLOCK_BACKOFF(count)					\

you can replace SPINLOCK_BACKOFF with detail/atomic_backoff.hpp

[pbalcer]

Reviewable status: 3 of 8 files reviewed, 9 unresolved discussions (waiting on @karczex)

---

include/libpmemobj++/container/detail/ringbuf/ringbuf.h, line 5 at r3 (raw file):

 * Use is subject to license terms, as specified in the LICENSE file.
 */

Can't find this file.

---

include/libpmemobj++/container/detail/ringbuf/utils.h, line 34 at r3 (raw file):

 *	@(#)cdefs.h	8.8 (Berkeley) 1/9/95
 */

I think the "complete" version of this patch shouldn't have this file and just use the C++11 idiomatic equivalents.

[pbalcer]

Reviewable status: 3 of 8 files reviewed, 13 unresolved discussions (waiting on @karczex)

---

examples/radix_tree/multi_tier_map.cpp, line 39 at r3 (raw file):

struct root {
	pmem::obj::persistent_ptr<kv_type> kv;
	pmem::obj::persistent_ptr<char[]> log;

wouldn't it be better to have a specialized type for the queue/ringbuffer for persistent state?

---

examples/radix_tree/multi_tier_map.cpp, line 81 at r3 (raw file):

		auto worker = queue.register_worker();
		auto acc = worker.produce(req_size);
		acc.add((char *)&ld, req_size);

maybe commit/publish instead of add?

---

examples/radix_tree/multi_tier_map.cpp, line 130 at r3 (raw file):

	bg_work()
	{
		while (!stopped.load()) {

is this a busy loop? if so, please add a way to wait for new data to consume.

---

examples/radix_tree/multi_tier_map.cpp, line 131 at r3 (raw file):

	{
		while (!stopped.load()) {
			auto acc = queue.consume();

shouldn't consume also be a two step operation?

[igchor]

Reviewable status: 3 of 8 files reviewed, 14 unresolved discussions (waiting on @karczex)

---

include/libpmemobj++/container/mpsc_queue.hpp, line 44 at r3 (raw file):

			w = worker;
			queue = q;
			auto offset = ringbuf_acquire(queue->ring_buffer, w, len);

you should handle ringbuf_acquire failure somehow (it can return -1 if there is no space)

[igchor]

Reviewable status: 3 of 8 files reviewed, 14 unresolved discussions (waiting on @karczex and @pbalcer)

---

examples/radix_tree/multi_tier_map.cpp, line 39 at r3 (raw file):

Previously, pbalcer (Piotr Balcer) wrote…

wouldn't it be better to have a specialized type for the queue/ringbuffer for persistent state?

Yeah, you're right. I think it would be best to just use pmem::obj::vector and create alias for it (using mpsc_queue::persistent_state = pmem::obj::vector)

---

examples/radix_tree/multi_tier_map.cpp, line 81 at r3 (raw file):

Previously, pbalcer (Piotr Balcer) wrote…

maybe commit/publish instead of add?

Hm, maybe add should be renamed to produce and produce to acquire like in original C implementation? Although produce is also not perfect because the data becomes visible to consumers not after add but only after acc dtor is called.

---

examples/radix_tree/multi_tier_map.cpp, line 130 at r3 (raw file):

Previously, pbalcer (Piotr Balcer) wrote…

is this a busy loop? if so, please add a way to wait for new data to consume.

Right, this is just for simplicity for now. Eventually, we want to add some conditional variable here.

---

examples/radix_tree/multi_tier_map.cpp, line 131 at r3 (raw file):

Previously, pbalcer (Piotr Balcer) wrote…

shouldn't consume also be a two step operation?

Actually, it is, right now the second step is executed in acc destructor (similar to std::unique_lock etc.)

---

include/libpmemobj++/container/detail/ringbuf/utils.h, line 34 at r3 (raw file):

Previously, pbalcer (Piotr Balcer) wrote…

I think the "complete" version of this patch shouldn't have this file and just use the C++11 idiomatic equivalents.

Yes, that's the plan

[pbalcer]

Reviewable status: 3 of 8 files reviewed, 13 unresolved discussions (waiting on @igchor, @karczex, and @pbalcer)

---

examples/radix_tree/multi_tier_map.cpp, line 81 at r3 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

Hm, maybe add should be renamed to produce and produce to acquire like in original C implementation? Although produce is also not perfect because the data becomes visible to consumers not after add but only after acc dtor is called.

I think we should be consistent in the way we reserve and "commit" resources.

maybe instead, we should do something like
worker.produce(size, |data: some internal buffer of size "size"| { ...; });
Same for consume. This way, it matches how we do the same thing in normal transactions.

Here's something similar:
https://ferrous-systems.com/blog/lock-free-ring-buffer/
https://www.codeproject.com/Articles/3479/The-Bip-Buffer-The-Circular-Buffer-with-a-Twist

[pbalcer]

Reviewable status: 3 of 8 files reviewed, 11 unresolved discussions (waiting on @igchor and @karczex)

---

examples/radix_tree/multi_tier_map.cpp, line 130 at r3 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

Right, this is just for simplicity for now. Eventually, we want to add some conditional variable here.

the consume() function should by default wait until there's something to consume, and try_wait() should fail if the buffer is empty.

This should be integrated into the library, not as something left for the user to do (it's not as simple as just adding a cond variable).

[igchor]

Reviewable status: 3 of 8 files reviewed, 11 unresolved discussions (waiting on @karczex and @pbalcer)

---

examples/radix_tree/multi_tier_map.cpp, line 81 at r3 (raw file):

Previously, pbalcer (Piotr Balcer) wrote…

I think we should be consistent in the way we reserve and "commit" resources.

maybe instead, we should do something like
worker.produce(size, |data: some internal buffer of size "size"| { ...; });
Same for consume. This way, it matches how we do the same thing in normal transactions.

Here's something similar:
https://ferrous-systems.com/blog/lock-free-ring-buffer/
https://www.codeproject.com/Articles/3479/The-Bip-Buffer-The-Circular-Buffer-with-a-Twist

You mean to use lambda? I think that's a good idea for produce - we would not need accessor at all and it would be consistent with tx API as you said.

Regarding this internal buffer, our idea was to have a DRAM buffer to which we copy the data and then (on acc dtor or after this lambda execution) we memcpy data from buffer to pmem using nontemporal stores. This is useful if you want to write several small values (< 64B) to avoid writing partial cache lines to pmem. But this is not very practical for very big values. Do you think having some threshold would make sense? (if value is big enough it goes directly to pmem)

---

examples/radix_tree/multi_tier_map.cpp, line 130 at r3 (raw file):

Previously, pbalcer (Piotr Balcer) wrote…

the consume() function should by default wait until there's something to consume, and try_wait() should fail if the buffer is empty.

This should be integrated into the library, not as something left for the user to do (it's not as simple as just adding a cond variable).

Ok, I agree we should expose blocking variants - for now I would suggest to rename consume to e.g. try_consume (and produce to try_produce)

[pbalcer]

Reviewable status: 3 of 8 files reviewed, 11 unresolved discussions (waiting on @igchor and @karczex)

---

examples/radix_tree/multi_tier_map.cpp, line 81 at r3 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

You mean to use lambda? I think that's a good idea for produce - we would not need accessor at all and it would be consistent with tx API as you said.

Regarding this internal buffer, our idea was to have a DRAM buffer to which we copy the data and then (on acc dtor or after this lambda execution) we memcpy data from buffer to pmem using nontemporal stores. This is useful if you want to write several small values (< 64B) to avoid writing partial cache lines to pmem. But this is not very practical for very big values. Do you think having some threshold would make sense? (if value is big enough it goes directly to pmem)

yes, using lambdas. I think using the same approach for consume also makes sense, because you want to make sure that the whole "transaction" (the content of the lambda) is executed before the value is actually consumed.

---

examples/radix_tree/multi_tier_map.cpp, line 130 at r3 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

Ok, I agree we should expose blocking variants - for now I would suggest to rename consume to e.g. try_consume (and produce to try_produce)

sounds good (and yes, I meant try_consume/try_produce)

[lukaszstolarczuk]

Reviewed 2 of 6 files at r1, 2 of 7 files at r4, 1 of 4 files at r5, 1 of 2 files at r6.
Reviewable status: 5 of 10 files reviewed, 14 unresolved discussions (waiting on @igchor, @karczex, and @KFilipek)

a discussion (no related file):
pls update LICENSE file in the repo root directory (with the exception)

---

---

include/libpmemobj++/container/mpsc_queue.hpp, line 1 at r6 (raw file):

#ifndef LIBPMEMOBJ_MPSC_QUEUE_HPP

aren't we missing here some license?

---

tests/mpsc_queue/basic.cpp, line 6 at r6 (raw file):

#include "unittest.hpp"
#include <cstring>
#include <iostream>

these includes are redundant (imported in unittest.hpp)

[igchor]

Reviewable status: 5 of 10 files reviewed, 16 unresolved discussions (waiting on @igchor, @karczex, and @KFilipek)

---

CMakeLists.txt, line 263 at r6 (raw file):

install(DIRECTORY include/ DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}
	FILES_MATCHING PATTERN "*.hpp")

you don't install the .h file

---

include/libpmemobj++/container/mpsc_queue.hpp, line 166 at r6 (raw file):

	};

	class worker {

think about what ctors/assignment operator should this have (probably only move ctor/assignment)

[pbalcer]

Include the LICENSE for this file!!

[igchor]

Reviewable status: 4 of 10 files reviewed, 18 unresolved discussions (waiting on @igchor, @karczex, @KFilipek, and @lukaszstolarczuk)

---

include/libpmemobj++/container/mpsc_queue.hpp, line 219 at r8 (raw file):

	consume(Function &&f)
	{
		// XXX:Change this try-catch with something resonable

TBH, I would just move all the logic from read_accessor to this function:

auto len = ringbuf_consume(...);
if (len == 0)
   return false;
...
f(accessor(queue->data + offset, len));
...

ringbuf_release(...);

[karczex]

---

include/libpmemobj++/container/mpsc_queue.hpp, line 222 at r8 (raw file):

		try {
			auto acc = read_accessor(this);
			f(acc);

In current implementation, if lambda passed to consume gets read_accesor object (copy) instead of reference to it, the destructor of read_accessor is called twice, and whole queue become unusable mess.

[igchor]

Reviewed 1 of 7 files at r4, 8 of 9 files at r9.
Reviewable status: 13 of 14 files reviewed, 26 unresolved discussions (waiting on @karczex)

---

tests/mpsc_queue/basic.cpp, line 34 at r9 (raw file):

	auto worker = queue.register_worker();
	bool consumed = queue.try_consume([&](auto rd_acc) { ASSERT_REACHED; });

Shouldn't you name it ASSERT_UNREACHABLE? ASSERT_REACHED suggest that it should be reached.

---

tests/mpsc_queue/mt.cpp, line 41 at r9 (raw file):

		if (thread_id == 0) {
			/* Read data while writting */
			while (semaphore.load() > 0) {

the name is quite confusing, it's not a semaphore, it's just an atomic counter.

---

tests/mpsc_queue/ringbuf.cpp, line 27 at r9 (raw file):

 */
// SPDX-License-Identifier: BSD-3-Clause
// Copyright 2021, Intel Corporation

I think you should also port t_stree.c test from the original repo. And ultimately, we would like to use this test to test our mpsc_queue, right? and not just the underlying ringbuf.

---

include/libpmemobj++/detail/ringbuf.hpp, line 78 at r9 (raw file):

void ringbuf_release(ringbuf_t *, size_t);

#define RBUF_OFF_MASK (0x00000000ffffffffUL)

I would change defines to constants - defines are not bound to this scope (they pollute the global namespace).

---

include/libpmemobj++/detail/ringbuf.hpp, line 88 at r9 (raw file):

struct ringbuf_worker {
	volatile std::atomic<ringbuf_off_t> seen_off;

remove volatile

---

include/libpmemobj++/detail/ringbuf.hpp, line 101 at r9 (raw file):

	 * the producer can update the 'end' offset.
	 */
	volatile std::atomic<ringbuf_off_t> next;

remove volatile

---

include/libpmemobj++/detail/ringbuf.hpp, line 112 at r9 (raw file):

	{
		written.store(0);
		workers = new ringbuf_worker[max_workers];

Actually, you could probably just use std::vector with fixed size, you won't need custom dtor. Of course the vector cannot be resized.

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 33 at r9 (raw file):

static constexpr size_t CACHELINE_SIZE = 64ULL;
#define ALIGN_UP(size, align) (((size) + (align)-1) & ~((align)-1))

please change this to a function

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 136 at r9 (raw file):

		{
			entry dram_entry;
			size_t aligned_len = ALIGN_UP(size, CACHELINE_SIZE);

probably some assert (aligned_len == CACHELINE_SIZE)?

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 173 at r9 (raw file):

public:
	mpsc_queue(pmem::obj::persistent_ptr<char[]> log, size_t buff_size,

can you change the persistent_ptr<char[]> to pmem::obj::vector (probably create some alias like pmem_log_type = pmem::obj::vector)? You can then get rid of buff_size.

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 188 at r9 (raw file):

	~mpsc_queue()
	{
		delete ring_buffer;

can't you just keep ringbuf as direct member of mpsc_queue (no pointers)?

[karczex]

---

include/libpmemobj++/detail/ringbuf.hpp, line 112 at r9 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

Actually, you could probably just use std::vector with fixed size, you won't need custom dtor. Of course the vector cannot be resized.

std::array probably would be better

[karczex]

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 188 at r9 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

can't you just keep ringbuf as direct member of mpsc_queue (no pointers)?

As (at least currently) all internal methods gets pointer to ringbuf, it's (IMO) less error prone to delete object in dtor, than get pointer to varaible in every call.

[igchor]

Reviewable status: 13 of 14 files reviewed, 26 unresolved discussions (waiting on @karczex)

---

include/libpmemobj++/detail/ringbuf.hpp, line 112 at r9 (raw file):

Previously, karczex (Paweł Karczewski) wrote…

std::array probably would be better

std::array requires compile time knowledge about the size.

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 188 at r9 (raw file):

Previously, karczex (Paweł Karczewski) wrote…

As (at least currently) all internal methods gets pointer to ringbuf, it's (IMO) less error prone to delete object in dtor, than get pointer to varaible in every call.

I don't see any problems with passing &ring_buffer to methods tbh, if you try to pass something else probably you get a compiler error.

[karczex]

Reviewable status: 7 of 14 files reviewed, 24 unresolved discussions (waiting on @igchor, @karczex, @KFilipek, @lukaszstolarczuk, and @pbalcer)

a discussion (no related file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

pls update LICENSE file in the repo root directory (with the exception)

Done.

---

---

CMakeLists.txt, line 263 at r6 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

you don't install the .h file

header changed to hpp

---

examples/radix_tree/multi_tier_map.cpp, line 24 at r3 (raw file):

Previously, KFilipek (Krzysztof Filipek) wrote…

	pmem::obj::p<size_t> data_1;
	pmem::obj::p<size_t> data_2;
	pmem::obj::p<size_t> data_3;

Why not

> pmem::obj::p<size_t> data_[3];

example was removed from this PR.

---

examples/radix_tree/multi_tier_map.cpp, line 31 at r3 (raw file):

Previously, KFilipek (Krzysztof Filipek) wrote…

		return data_1 == rhs.data_1 && data_2 == rhs.data_2 &&
			data_3 == rhs.data_3;

then:

for (int i = 0; i < 3; i++)
    if (data_[i] != rhs.data_[i]) return false;
return true;

example was removed from this PR.

---

examples/radix_tree/multi_tier_map.cpp, line 79 at r3 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

I guess the worker should be only registered once - in the map ctor

example was removed from this PR.

---

examples/radix_tree/multi_tier_map.cpp, line 81 at r3 (raw file):

Previously, pbalcer (Piotr Balcer) wrote…

yes, using lambdas. I think using the same approach for consume also makes sense, because you want to make sure that the whole "transaction" (the content of the lambda) is executed before the value is actually consumed.

Done.

---

examples/radix_tree/multi_tier_map.cpp, line 130 at r3 (raw file):

Previously, pbalcer (Piotr Balcer) wrote…

sounds good (and yes, I meant try_consume/try_produce)

Done.

---

include/libpmemobj++/container/detail/ringbuf/utils.h, line 106 at r3 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

you can replace SPINLOCK_BACKOFF with detail/atomic_backoff.hpp

Done.

---

tests/mpsc_queue/basic.cpp, line 6 at r6 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

these includes are redundant (imported in unittest.hpp)

Done.

---

tests/mpsc_queue/basic.cpp, line 34 at r9 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

Shouldn't you name it ASSERT_UNREACHABLE? ASSERT_REACHED suggest that it should be reached.

Done.

---

tests/mpsc_queue/mt.cpp, line 41 at r9 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

the name is quite confusing, it's not a semaphore, it's just an atomic counter.

Done.

---

include/libpmemobj++/container/mpsc_queue.hpp, line 135 at r3 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

why persistent_ptr* and not just persistent_ptr?

Done.

---

include/libpmemobj++/container/mpsc_queue.hpp, line 143 at r3 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

I think we shouln't use malloc in C++. It would be better to either allocate char array (new char[ring_buffer_size]) or use approach suggested here: https://stackoverflow.com/questions/184537/in-what-cases-do-i-use-malloc-and-or-new

void *p = operator new(size);
   ...
operator delete(p);

Done.

---

include/libpmemobj++/container/mpsc_queue.hpp, line 1 at r6 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

aren't we missing here some license?

No, this is our file.

---

include/libpmemobj++/container/mpsc_queue.hpp, line 219 at r8 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

TBH, I would just move all the logic from read_accessor to this function:

auto len = ringbuf_consume(...);
if (len == 0)
   return false;
...
f(accessor(queue->data + offset, len));
...

ringbuf_release(...);

Done.

---

include/libpmemobj++/container/detail/ringbuf.h, line 5 at r7 (raw file):

Previously, pbalcer (Piotr Balcer) wrote…

Include the LICENSE for this file!!

Done.

---

include/libpmemobj++/container/detail/ringbuf/ringbuf.h, line 5 at r3 (raw file):

Previously, pbalcer (Piotr Balcer) wrote…

 * Use is subject to license terms, as specified in the LICENSE file.
 */

Can't find this file.

I put license in the imported files.

---

include/libpmemobj++/detail/ringbuf.hpp, line 78 at r9 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

I would change defines to constants - defines are not bound to this scope (they pollute the global namespace).

Done.

---

include/libpmemobj++/detail/ringbuf.hpp, line 88 at r9 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

remove volatile

Done.

---

include/libpmemobj++/detail/ringbuf.hpp, line 101 at r9 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

remove volatile

Done.

---

include/libpmemobj++/detail/ringbuf.hpp, line 112 at r9 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

std::array requires compile time knowledge about the size.

I replaced it with shared_ptr, so I don't need to introduce custom dtor, yet do not have to be careful about resizing vector.

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 33 at r9 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

please change this to a function

Done.

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 136 at r9 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

probably some assert (aligned_len == CACHELINE_SIZE)?

Done.

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 173 at r9 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

can you change the persistent_ptr<char[]> to pmem::obj::vector (probably create some alias like pmem_log_type = pmem::obj::vector)? You can then get rid of buff_size.

As discussed offline we would left it as is.

[igchor]

Reviewable status: 5 of 14 files reviewed, 11 unresolved discussions (waiting on @igchor, @karczex, @KFilipek, @lukaszstolarczuk, and @pbalcer)

---

tests/mpsc_queue/mt.cpp, line 72 at r13 (raw file):

	});

	/* Consume rest of the data. Need to call try_consume twice as queue do

"as queue do not overlap" - what does it mean? I don't quire understand

---

tests/mpsc_queue/mt.cpp, line 75 at r13 (raw file):

	 * not overlap. */
	for (int i = 0; i < 2; i++) {
		queue.try_consume(

shoudln't you assert return value?

[igchor]

Reviewable status: 5 of 14 files reviewed, 12 unresolved discussions (waiting on @igchor, @karczex, @KFilipek, @lukaszstolarczuk, and @pbalcer)

---

tests/mpsc_queue/mt.cpp, line 79 at r13 (raw file):

				    rd_acc1) {
				for (auto str : rd_acc1) {
					std::cout << str.size() << " ";

cout? probably remove this

[lukaszstolarczuk]

Reviewed 1 of 7 files at r4, 1 of 9 files at r9, 1 of 6 files at r10, 2 of 8 files at r12.
Reviewable status: 7 of 14 files reviewed, 19 unresolved discussions (waiting on @igchor, @karczex, @KFilipek, @lukaszstolarczuk, and @pbalcer)

a discussion (no related file):
in commit messages:

• "Change ringbufer implenetation to c++" - 2 misspells (ringbuffer, implementation)
• "..manage workers**'** array.." - missing ' apostrohpe

---

---

tests/mpsc_queue/ringbuf.cpp, line 27 at r13 (raw file):

 */
// SPDX-License-Identifier: BSD-3-Clause
// Copyright 2021, Intel Corporation

/**/

---

tests/mpsc_queue/ringbuf.cpp, line 286 at r13 (raw file):

				}
				break;
			case 2: // producer 2

pls make all 3 comments here /**/, as well

---

tests/mpsc_queue/ringbuf.cpp, line 314 at r13 (raw file):

	test_overlap();
	test_random();
	puts("ok");

this puts can be removed, I guess

---

include/libpmemobj++/detail/ringbuf.hpp, line 30 at r10 (raw file):

/* Copyright 2021, Intel Corporation */

#ifndef _RINGBUF_H_

rename this, I guess (HPP)

---

include/libpmemobj++/detail/ringbuf.hpp, line 450 at r13 (raw file):

}
}
#endif

pls add comment what's the endif for

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 52 at r13 (raw file):

			}

			// Invalidates data after increment

make it a /**/ comment

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 238 at r13 (raw file):

	}

	// XXX - Move logic from this function to consume (this requires setting

/**/

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 239 at r13 (raw file):

	// XXX - Move logic from this function to consume (this requires setting
	// reader/writer offsets int ringubf)

int? perhaps you mean in..?

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 256 at r13 (raw file):

}
}
}

pls add comments, which namespaces are these

[igchor]

Reviewed 4 of 8 files at r12.
Reviewable status: 9 of 14 files reviewed, 22 unresolved discussions (waiting on @igchor, @karczex, @KFilipek, @lukaszstolarczuk, and @pbalcer)

---

include/libpmemobj++/detail/ringbuf.hpp, line 51 at r13 (raw file):

#include <libpmemobj++/detail/atomic_backoff.hpp>

#define __predict_false(x) __builtin_expect((x) != 0, 0)

We have no way of knowing right now but it will fail on windows probably, please handle also the windows case. (You can just do nothing).

---

include/libpmemobj++/detail/ringbuf.hpp, line 65 at r13 (raw file):

{

typedef struct ringbuf ringbuf_t;

you can remove those 2 lines, as well as the function declarations below (all ringbuf_*)

---

include/libpmemobj++/detail/ringbuf.hpp, line 137 at r13 (raw file):

 */
ringbuf_worker_t *
ringbuf_register(ringbuf_t *rbuf, unsigned i)

All functions in here should be inline, otherwise if you include this file from two translation units you will get an error: https://softwareengineering.stackexchange.com/questions/339486/when-a-function-should-be-declared-inline-in-c

[igchor]

Reviewed 1 of 6 files at r10, 3 of 8 files at r12.
Reviewable status: 12 of 14 files reviewed, 23 unresolved discussions (waiting on @igchor, @karczex, @KFilipek, and @pbalcer)

---

tests/mpsc_queue/mpsc_queue_mt_0.cmake, line 1 at r13 (raw file):

# SPDX-License-Identifier: BSD-3-Clause

Do you actually need this file? If you don't specify SCRIPT parameter in add_test_generic it will use the default cmake script which is (I think) the same as you have here

[lukaszstolarczuk]

Reviewed 2 of 8 files at r12, 1 of 6 files at r14, 3 of 3 files at r15.
Reviewable status: 11 of 14 files reviewed, 21 unresolved discussions (waiting on @igchor, @karczex, @KFilipek, @lukaszstolarczuk, and @pbalcer)

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 22 at r13 (raw file):

#include <libpmemobj++/detail/ringbuf.hpp>
#include <memory>
#include <stdexcept>

I guess you can order these 'includes' a little

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 35 at r13 (raw file):

static constexpr size_t CACHELINE_SIZE = 64ULL;

class mpsc_queue {

you could add some descriptions, at least for class(es) and public methods

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 194 at r14 (raw file):

	static size_t
	ALIGN_UP(size_t size, size_t align)

why all upper case?

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 223 at r14 (raw file):

	}

	template <typename Function>

keep one naming either F (as below) or Function (e.g. above, in try_produce)

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 239 at r14 (raw file):

	/* XXX - Move logic from this function to consume (this requires setting
	   reader/writer offsets in ringubf) */

ringubf misspell

[karczex]

Reviewable status: 11 of 14 files reviewed, 21 unresolved discussions (waiting on @igchor, @karczex, @KFilipek, @lukaszstolarczuk, and @pbalcer)

a discussion (no related file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

in commit messages:

• "Change ringbufer implenetation to c++" - 2 misspells (ringbuffer, implementation)
• "..manage workers**'** array.." - missing ' apostrohpe

Done.

---

---

tests/mpsc_queue/mpsc_queue_mt_0.cmake, line 1 at r13 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

Do you actually need this file? If you don't specify SCRIPT parameter in add_test_generic it will use the default cmake script which is (I think) the same as you have here

Done.

---

tests/mpsc_queue/mt.cpp, line 72 at r13 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

"as queue do not overlap" - what does it mean? I don't quire understand

Done.

---

tests/mpsc_queue/mt.cpp, line 75 at r13 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

shoudln't you assert return value?

I hope, the comment is enough

---

tests/mpsc_queue/mt.cpp, line 79 at r13 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

cout? probably remove this

Done.

---

tests/mpsc_queue/ringbuf.cpp, line 27 at r13 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

/**/

Done.

---

tests/mpsc_queue/ringbuf.cpp, line 286 at r13 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

pls make all 3 comments here /**/, as well

Done.

---

tests/mpsc_queue/ringbuf.cpp, line 314 at r13 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

this puts can be removed, I guess

Done.

---

include/libpmemobj++/detail/ringbuf.hpp, line 30 at r10 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

rename this, I guess (HPP)

Done.

---

include/libpmemobj++/detail/ringbuf.hpp, line 51 at r13 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

We have no way of knowing right now but it will fail on windows probably, please handle also the windows case. (You can just do nothing).

Done.

---

include/libpmemobj++/detail/ringbuf.hpp, line 65 at r13 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

you can remove those 2 lines, as well as the function declarations below (all ringbuf_*)

Done.

---

include/libpmemobj++/detail/ringbuf.hpp, line 137 at r13 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

All functions in here should be inline, otherwise if you include this file from two translation units you will get an error: https://softwareengineering.stackexchange.com/questions/339486/when-a-function-should-be-declared-inline-in-c

Done.

---

include/libpmemobj++/detail/ringbuf.hpp, line 450 at r13 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

pls add comment what's the endif for

Done.

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 52 at r13 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

make it a /**/ comment

Done.

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 238 at r13 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

/**/

Done.

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 239 at r13 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

int? perhaps you mean in..?

Done.

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 256 at r13 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

pls add comments, which namespaces are these

Done.

[igchor]

Reviewable status: 11 of 14 files reviewed, 16 unresolved discussions (waiting on @igchor, @karczex, @KFilipek, @lukaszstolarczuk, and @pbalcer)

---

tests/mpsc_queue/mt.cpp, line 75 at r13 (raw file):

Previously, karczex (Paweł Karczewski) wrote…

I hope, the comment is enough

Great, thanks! Maybe we should assert the return value based on the queue size? Because it's deterministic, right?

---

include/libpmemobj++/detail/ringbuf.hpp, line 81 at r15 (raw file):

typedef uint64_t ringbuf_off_t;

typedef struct ringbuf_worker {

typedef is not needed at all in c++

[karczex]

---

tests/mpsc_queue/mt.cpp, line 75 at r13 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

Great, thanks! Maybe we should assert the return value based on the queue size? Because it's deterministic, right?

It's not fully deterministic, as we cannot be sure if first spawned thread ended first . We may assert result of 3rd consume.

[karczex]

Reviewable status: 10 of 14 files reviewed, 16 unresolved discussions (waiting on @igchor, @karczex, @KFilipek, @lukaszstolarczuk, and @pbalcer)

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 22 at r13 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

I guess you can order these 'includes' a little

Done.

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 194 at r14 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

why all upper case?

Done.

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 223 at r14 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

keep one naming either F (as below) or Function (e.g. above, in try_produce)

good point. thx.

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 239 at r14 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

ringubf misspell

Done.

[karczex]

Reviewable status: 9 of 14 files reviewed, 16 unresolved discussions (waiting on @igchor, @KFilipek, @lukaszstolarczuk, and @pbalcer)

---

include/libpmemobj++/detail/ringbuf.hpp, line 81 at r15 (raw file):

Previously, igchor (Igor Chorążewicz) wrote…

typedef is not needed at all in c++

Done.

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 35 at r13 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

you could add some descriptions, at least for class(es) and public methods

Documentation would be added in next PR

[lukaszstolarczuk]

besides these minor issues in tests, it looks good! 👍

Reviewed 1 of 8 files at r12, 1 of 6 files at r14, 3 of 4 files at r16.
Reviewable status: 13 of 14 files reviewed, 16 unresolved discussions (waiting on @igchor, @karczex, @KFilipek, and @pbalcer)

---

tests/mpsc_queue/basic.cpp, line 114 at r16 (raw file):

		UT_ASSERT(values_on_pmem == values);

		std::string tmp = "old";

why tmp? and why you're doing this extra try_produce here? pls add a short comment

---

tests/mpsc_queue/basic.cpp, line 129 at r16 (raw file):

			});
		UT_ASSERTeq(values_on_pmem.size(), 1);
		UT_ASSERTeq(values_on_pmem[0].size(), 3);

magic 3 - perhaps tmp.size()

---

tests/mpsc_queue/basic.cpp, line 131 at r16 (raw file):

		UT_ASSERTeq(values_on_pmem[0].size(), 3);
		UT_ASSERT(values_on_pmem[0] == std::string("old"));
		std::cout << values_on_pmem[0] << std::endl;

drop this cout as well, pls

---

tests/mpsc_queue/mt.cpp, line 21 at r16 (raw file):

};

int

pls add a short comment what is this test doing

---

tests/mpsc_queue/mt.cpp, line 67 at r16 (raw file):

						});
				};
			}

you can assert here x>0, I guess

---

tests/mpsc_queue/mt.cpp, line 113 at r16 (raw file):

	constexpr size_t concurrency = 16;
	size_t buffer_size =
		pmem::obj::experimental::CACHELINE_SIZE * concurrency * 3;

why * 3?

---

tests/mpsc_queue/ringbuf.cpp, line 27 at r13 (raw file):

Previously, karczex (Paweł Karczewski) wrote…

Done.

ups, you overdid this ;)

we usually make this like this (actually I don't know why)

// SPDX-License-Identifier: BSD-3-Clause
/* Copyright 2021, Intel Corporation */

---

tests/mpsc_queue/ringbuf.cpp, line 29 at r16 (raw file):

 * Copyright 2021, Intel Corporation
 */

you can state here a short comment about these tests

[KFilipek]

Reviewed 1 of 7 files at r4, 1 of 4 files at r8, 1 of 9 files at r9, 3 of 8 files at r12, 1 of 6 files at r14, 1 of 3 files at r15, 3 of 4 files at r16.
Reviewable status: 13 of 14 files reviewed, 13 unresolved discussions (waiting on @igchor, @karczex, and @pbalcer)

---

tests/mpsc_queue/mt.cpp, line 63 at r16 (raw file):

Quoted 4 lines of code…

							std::copy_n(
								e.begin(),
								e.size(),
								range.begin());

worst formatting I've ever seen (non blocking)

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 129 at r16 (raw file):

		worker(worker &&other)
		{
			*this = std::move(other);

what about release old allocated data?

[igchor]

Reviewable status: 13 of 14 files reviewed, 13 unresolved discussions (waiting on @igchor, @karczex, and @pbalcer)

[karczex]

Reviewable status: 11 of 14 files reviewed, 13 unresolved discussions (waiting on @igchor, @KFilipek, @lukaszstolarczuk, and @pbalcer)

---

tests/mpsc_queue/basic.cpp, line 114 at r16 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

why tmp? and why you're doing this extra try_produce here? pls add a short comment

Done.

---

tests/mpsc_queue/basic.cpp, line 129 at r16 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

magic 3 - perhaps tmp.size()

Done.

---

tests/mpsc_queue/basic.cpp, line 131 at r16 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

drop this cout as well, pls

Done.

---

tests/mpsc_queue/mt.cpp, line 21 at r16 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

pls add a short comment what is this test doing

Done.

---

tests/mpsc_queue/mt.cpp, line 63 at r16 (raw file):

Previously, KFilipek (Krzysztof Filipek) wrote…

							std::copy_n(
								e.begin(),
								e.size(),
								range.begin());

worst formatting I've ever seen (non blocking)

I totally agree. (non bloking :)

---

tests/mpsc_queue/mt.cpp, line 67 at r16 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

you can assert here x>0, I guess

Done.

---

tests/mpsc_queue/mt.cpp, line 113 at r16 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

why * 3?

Done.

---

tests/mpsc_queue/ringbuf.cpp, line 27 at r13 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

ups, you overdid this ;)

we usually make this like this (actually I don't know why)

// SPDX-License-Identifier: BSD-3-Clause
/* Copyright 2021, Intel Corporation */

Done. However I don't get why we do it this way

---

tests/mpsc_queue/ringbuf.cpp, line 29 at r16 (raw file):

Previously, lukaszstolarczuk (Łukasz Stolarczuk) wrote…

you can state here a short comment about these tests

Done.

---

include/libpmemobj++/experimental/mpsc_queue.hpp, line 129 at r16 (raw file):

Previously, KFilipek (Krzysztof Filipek) wrote…

what about release old allocated data?

It's released in move assignment operator (C++ here is tricky as hell).
https://docs.microsoft.com/en-us/cpp/cpp/move-constructors-and-move-assignment-operators-cpp?view=msvc-160

[lukaszstolarczuk]

Reviewed 2 of 3 files at r17.
Reviewable status: 13 of 14 files reviewed, 5 unresolved discussions (waiting on @igchor, @KFilipek, and @pbalcer)

---

tests/mpsc_queue/basic.cpp, line 114 at r16 (raw file):

Previously, karczex (Paweł Karczewski) wrote…

Done.

thx!