simple mpmsc example

Author: taegyunkim

ddcommon-ffi/src/array_queue.rs

@@ -97,14 +97,18 @@ pub unsafe extern "C" fn ddog_array_queue_drop(queue_ptr: *mut ArrayQueue) {
 #[allow(unused)]
 #[repr(C)]
 pub enum ArrayQueuePushResult {
-    Ok(bool),
+    Ok,
+    Full(*mut c_void),
     Err(Error),
 }
 
-impl From<Result<(), anyhow::Error>> for ArrayQueuePushResult {
-    fn from(result: Result<(), anyhow::Error>) -> Self {
+impl From<Result<Result<(), *mut c_void>, anyhow::Error>> for ArrayQueuePushResult {
+    fn from(result: Result<Result<(), *mut c_void>, anyhow::Error>) -> Self {
         match result {
-            Ok(_) => ArrayQueuePushResult::Ok(true),
+            Ok(value) => match value {
+                Ok(()) => ArrayQueuePushResult::Ok,
+                Err(value) => ArrayQueuePushResult::Full(value),
+            },
             Err(err) => ArrayQueuePushResult::Err(err.into()),
         }
     }
@@ -121,9 +125,7 @@ pub unsafe extern "C" fn ddog_array_queue_push(
 ) -> ArrayQueuePushResult {
     (|| {
         let queue = ddog_array_queue_ptr_to_inner(queue_ptr)?;
-        queue
-            .push(value)
-            .map_err(|_| anyhow::anyhow!("array_queue full"))
+        anyhow::Ok(queue.push(value))
     })()
     .context("array_queue_push failed")
     .into()
@@ -133,13 +135,17 @@ pub unsafe extern "C" fn ddog_array_queue_push(
 #[repr(C)]
 pub enum ArrayQueuePopResult {
     Ok(*mut c_void),
+    Empty,
     Err(Error),
 }
 
-impl From<anyhow::Result<*mut c_void>> for ArrayQueuePopResult {
-    fn from(result: anyhow::Result<*mut c_void>) -> Self {
+impl From<anyhow::Result<Option<*mut c_void>>> for ArrayQueuePopResult {
+    fn from(result: anyhow::Result<Option<*mut c_void>>) -> Self {
         match result {
-            Ok(value) => ArrayQueuePopResult::Ok(value),
+            Ok(value) => match value {
+                Some(value) => ArrayQueuePopResult::Ok(value),
+                None => ArrayQueuePopResult::Empty,
+            },
             Err(err) => ArrayQueuePopResult::Err(err.into()),
         }
     }
@@ -152,26 +158,24 @@ impl From<anyhow::Result<*mut c_void>> for ArrayQueuePopResult {
 pub unsafe extern "C" fn ddog_array_queue_pop(queue_ptr: *mut ArrayQueue) -> ArrayQueuePopResult {
     (|| {
         let queue = ddog_array_queue_ptr_to_inner(queue_ptr)?;
-        queue
-            .pop()
-            .ok_or_else(|| anyhow::anyhow!("array_queue empty"))
+        anyhow::Ok(queue.pop())
     })()
     .context("array_queue_pop failed")
     .into()
 }
 
 #[allow(unused)]
 #[repr(C)]
-pub enum ArrayQueueIsEmptyResult {
+pub enum ArrayQueueBoolResult {
     Ok(bool),
     Err(Error),
 }
 
-impl From<anyhow::Result<bool>> for ArrayQueueIsEmptyResult {
+impl From<anyhow::Result<bool>> for ArrayQueueBoolResult {
     fn from(result: anyhow::Result<bool>) -> Self {
         match result {
-            Ok(value) => ArrayQueueIsEmptyResult::Ok(value),
-            Err(err) => ArrayQueueIsEmptyResult::Err(err.into()),
+            Ok(value) => ArrayQueueBoolResult::Ok(value),
+            Err(err) => ArrayQueueBoolResult::Err(err.into()),
         }
     }
 }
@@ -182,7 +186,7 @@ impl From<anyhow::Result<bool>> for ArrayQueueIsEmptyResult {
 #[no_mangle]
 pub unsafe extern "C" fn ddog_array_queue_is_empty(
     queue_ptr: *mut ArrayQueue,
-) -> ArrayQueueIsEmptyResult {
+) -> ArrayQueueBoolResult {
     (|| {
         let queue = ddog_array_queue_ptr_to_inner(queue_ptr)?;
         anyhow::Ok(queue.is_empty())
@@ -193,16 +197,16 @@ pub unsafe extern "C" fn ddog_array_queue_is_empty(
 
 #[allow(unused)]
 #[repr(C)]
-pub enum ArrayQueueLenResult {
+pub enum ArrayQueueUsizeResult {
     Ok(usize),
     Err(Error),
 }
 
-impl From<anyhow::Result<usize>> for ArrayQueueLenResult {
+impl From<anyhow::Result<usize>> for ArrayQueueUsizeResult {
     fn from(result: anyhow::Result<usize>) -> Self {
         match result {
-            Ok(value) => ArrayQueueLenResult::Ok(value),
-            Err(err) => ArrayQueueLenResult::Err(err.into()),
+            Ok(value) => ArrayQueueUsizeResult::Ok(value),
+            Err(err) => ArrayQueueUsizeResult::Err(err.into()),
         }
     }
 }
@@ -211,7 +215,7 @@ impl From<anyhow::Result<usize>> for ArrayQueueLenResult {
 /// # Safety
 /// The pointer is null or points to a valid memory location allocated by array_queue_new.
 #[no_mangle]
-pub unsafe extern "C" fn ddog_array_queue_len(queue_ptr: *mut ArrayQueue) -> ArrayQueueLenResult {
+pub unsafe extern "C" fn ddog_array_queue_len(queue_ptr: *mut ArrayQueue) -> ArrayQueueUsizeResult {
     (|| {
         let queue = ddog_array_queue_ptr_to_inner(queue_ptr)?;
         anyhow::Ok(queue.len())

examples/ffi/.gitignore

@@ -1,10 +1,14 @@
 CMakeFiles/
-array_queue
 CMakeCache.txt
+cmake_install.cmake
+Makefile
+compile_commands.json
+
+array_queue
 crashinfo
 crashtracking
 exporter
-Makefile
 profiles
 telemetry
 telemetry_metrics
+trace_exporter

examples/ffi/CMakeLists.txt

@@ -4,6 +4,7 @@ project(datadog_profiling_ffi_examples LANGUAGES C CXX)
 find_package(Datadog REQUIRED)
 
 set(VCRUNTIME_LINK_TYPE DLL CACHE STRING "Specify the Runtime Library to use when compiling with MSVC")
+set(CMAKE_EXPORT_COMPILE_COMMANDS on)
 
 function(set_vcruntime_link_type binary link_type)
   if (CMAKE_CXX_COMPILER_ID MATCHES "MSVC")

examples/ffi/array_queue.cpp

@@ -4,8 +4,12 @@
 extern "C" {
 #include <datadog/common.h>
 }
+#include <atomic>
+#include <cassert>
 #include <cstdio>
 #include <memory>
+#include <thread>
+#include <vector>
 
 struct Sample {
   int x;
@@ -24,11 +28,72 @@ void print_error(const char *s, const ddog_Error &err) {
 }
 
 int main(void) {
-  ddog_ArrayQueue_NewResult array_queue_new_result = ddog_array_queue_new(10, delete_fn);
+  ddog_ArrayQueue_NewResult array_queue_new_result = ddog_array_queue_new(5, delete_fn);
   if (array_queue_new_result.tag != DDOG_ARRAY_QUEUE_NEW_RESULT_OK) {
     print_error("Failed to create array queue", array_queue_new_result.err);
     ddog_Error_drop(&array_queue_new_result.err);
     return 1;
   }
   std::unique_ptr<ddog_ArrayQueue, Deleter> array_queue(&array_queue_new_result.ok);
+
+  size_t num_threads = 4;
+  size_t num_elements = 50;
+  std::vector<std::atomic<size_t>> counts(num_elements);
+  for (size_t i = 0; i < num_elements; ++i) {
+    counts[i].store(0);
+  }
+
+  auto consumer = [&array_queue, &counts, num_elements]() {
+    for (size_t i = 0; i < num_elements; ++i) {
+      do {
+        ddog_ArrayQueue_PopResult pop_result = ddog_array_queue_pop(array_queue.get());
+        if (pop_result.tag == DDOG_ARRAY_QUEUE_POP_RESULT_OK) {
+          Sample *sample = (Sample *)pop_result.ok;
+          counts[sample->x].fetch_add(1, std::memory_order_seq_cst);
+          delete sample;
+          break;
+        } else if (pop_result.tag == DDOG_ARRAY_QUEUE_POP_RESULT_EMPTY) {
+          std::this_thread::yield();
+        } else {
+          print_error("Failed to pop from array queue", pop_result.err);
+          ddog_Error_drop(&pop_result.err);
+          return;
+        }
+      } while (true);
+    }
+  };
+
+  auto producer = [&array_queue, num_elements]() {
+    for (size_t i = 0; i < num_elements; ++i) {
+      Sample *sample = new Sample();
+      sample->x = i;
+      sample->y = i;
+      do {
+        ddog_ArrayQueue_PushResult push_result = ddog_array_queue_push(array_queue.get(), sample);
+        if (push_result.tag == DDOG_ARRAY_QUEUE_PUSH_RESULT_OK) {
+          break;
+        } else if (push_result.tag == DDOG_ARRAY_QUEUE_PUSH_RESULT_FULL) {
+          std::this_thread::yield();
+        } else {
+          print_error("Failed to push to array queue", push_result.err);
+          ddog_Error_drop(&push_result.err);
+          return;
+        }
+      } while (true);
+    }
+  };
+
+  std::vector<std::thread> threads;
+  for (size_t i = 0; i < num_threads; ++i) {
+    threads.emplace_back(consumer);
+    threads.emplace_back(producer);
+  }
+
+  for (auto &t : threads) {
+    t.join();
+  }
+
+  for (const auto &c : counts) {
+    assert(c.load(std::memory_order_seq_cst) == num_threads);
+  }
 }