Merge pull request #1087 from igchor/queue_arbitrary_size

mpsc_queue: extend queue to support arbitrary sized values

Author: lukaszstolarczuk

.github/workflows/gha.yml

@@ -91,6 +91,7 @@ jobs:
   windows:
     name: Windows
     runs-on: windows-latest
+
     env:
       platform: x64
       VCPKG_DEFAULT_TRIPLET: x64-windows
@@ -154,7 +155,6 @@ jobs:
                     -DTESTS_TBB=ON
                     -DDEVELOPER_MODE=ON
                     -DCXX_STANDARD="${{ matrix.CXX_STANDARD }}"
-                    -DTEST_MPSC_QUEUE=OFF
 
       - name: Build
         run: msbuild build/ALL_BUILD.vcxproj /property:Configuration=${{ matrix.BUILD_TYPE }} /verbosity:minimal /m

include/libpmemobj++/detail/common.hpp

@@ -150,6 +150,14 @@ namespace experimental
 namespace detail
 {
 
+#if defined(__x86_64) || defined(_M_X64) || defined(__aarch64__)
+static constexpr size_t CACHELINE_SIZE = 64ULL;
+#elif defined(__PPC64__)
+static constexpr size_t CACHELINE_SIZE = 128ULL;
+#else
+#error unable to recognize architecture at compile time
+#endif
+
 /*
  * Conditionally add 'count' objects to a transaction.
  *
@@ -303,6 +311,18 @@ mssb_index64(uint64_t value)
 
 #endif
 
+static constexpr size_t
+align_up(size_t size, size_t align)
+{
+	return ((size) + (align)-1) & ~((align)-1);
+}
+
+static constexpr size_t
+align_down(size_t size, size_t align)
+{
+	return (size) & ~((align)-1);
+}
+
 } /* namespace detail */
 
 } /* namespace pmem */

include/libpmemobj++/detail/ringbuf.hpp

@@ -115,6 +115,22 @@ struct ringbuf_t {
 		space = length;
 		end = RBUF_OFF_MAX;
 		nworkers = max_workers;
+
+		/* Helgrind/Drd does not understand std::atomic */
+#if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED
+		VALGRIND_HG_DISABLE_CHECKING(&next, sizeof(next));
+		VALGRIND_HG_DISABLE_CHECKING(&end, sizeof(end));
+		VALGRIND_HG_DISABLE_CHECKING(&written, sizeof(written));
+
+		for (size_t i = 0; i < max_workers; i++) {
+			VALGRIND_HG_DISABLE_CHECKING(
+				&workers[i].seen_off,
+				sizeof(workers[i].seen_off));
+			VALGRIND_HG_DISABLE_CHECKING(
+				&workers[i].registered,
+				sizeof(workers[i].registered));
+		}
+#endif
 	}
 };
 
@@ -185,7 +201,7 @@ stable_seenoff(ringbuf_worker_t *w)
  * => On success: returns the offset at which the space is available.
  * => On failure: returns -1.
  */
-inline ssize_t
+inline ptrdiff_t
 ringbuf_acquire(ringbuf_t *rbuf, ringbuf_worker_t *w, size_t len)
 {
 	ringbuf_off_t seen, next, target;
@@ -286,7 +302,7 @@ ringbuf_acquire(ringbuf_t *rbuf, ringbuf_worker_t *w, size_t len)
 			std::memory_order_release);
 	}
 	assert((target & RBUF_OFF_MASK) <= rbuf->space);
-	return (ssize_t)next;
+	return (ptrdiff_t)next;
 }
 
 /*