add share queue test

Signed-off-by: chaoqin-li1123 <[email protected]>

Author: chaoqin-li1123

src/shared_queue.cc

@@ -136,18 +136,19 @@ WasmResult SharedQueue::enqueue(uint32_t token, std::string_view value) {
     call_on_thread = target_queue->call_on_thread;
     target_queue->queue.push_back(std::string(value));
   }
-
-  call_on_thread([vm_key, context_id, token] {
-    // This code may or may not execute in another thread.
-    // Make sure that the lock is no longer held here.
-    auto wasm = getThreadLocalWasm(vm_key);
-    if (wasm) {
-      auto context = wasm->wasm()->getContext(context_id);
-      if (context) {
-        context->onQueueReady(token);
+  if (call_on_thread != nullptr) {
+    call_on_thread([vm_key, context_id, token] {
+      // This code may or may not execute in another thread.
+      // Make sure that the lock is no longer held here.
+      auto wasm = getThreadLocalWasm(vm_key);
+      if (wasm) {
+        auto context = wasm->wasm()->getContext(context_id);
+        if (context) {
+          context->onQueueReady(token);
+        }
       }
-    }
-  });
+    });
+  }
   return WasmResult::Ok;
 }
 

test/test_data/abi_export.rs

@@ -23,6 +23,9 @@ pub extern "C" fn proxy_on_vm_start(_: u32, _: usize) -> bool {
 #[no_mangle]
 pub extern "C" fn proxy_on_context_create(_: u32, _: u32) {}
 
+#[no_mangle]
+pub extern "C" fn proxy_on_queue_ready(_: u32, _: u32) {}
+
 #[no_mangle]
 pub extern "C" fn proxy_on_memory_allocate(size: usize) -> *mut u8 {
     let mut vec: Vec<u8> = Vec::with_capacity(size);

test/wasm_test.cc

@@ -16,6 +16,7 @@
 
 #include "gtest/gtest.h"
 #include <memory>
+#include <string>
 
 #include "test/utility.h"
 
@@ -147,4 +148,47 @@ TEST_P(WasmTest, DifferentRootContextsFromDifferentPluginKeys) {
   EXPECT_TRUE(root_context1->root_id() == root_context2->root_id());
 }
 
+TEST_P(WasmTest, SharedQueueProducerConsumer) {
+  const std::string plugin_name = "plugin_name";
+  const std::string root_id = "root_id";
+  const std::string plugin_config = "plugin_config";
+  const bool fail_open = false;
+  const std::shared_ptr<PluginBase> plugin1 = std::make_shared<PluginBase>(
+      plugin_name, root_id, vm_id_, runtime_, plugin_config, fail_open, "plugin1_key");
+
+  const std::string vm_key = "vm_key";
+  // Create base Wasm via createWasm.
+  std::shared_ptr<WasmHandleBase> base_wasm_handle =
+      createWasm(vm_key, source_, plugin1, wasm_handle_factory_, wasm_handle_clone_factory_, false);
+  base_wasm_handle->wasm()->start(plugin1);
+  ContextBase *root_context1 = base_wasm_handle->wasm()->getRootContext(plugin1, false);
+  EXPECT_TRUE(root_context1 != nullptr);
+
+  // Create a new plugin with different key.
+  const std::shared_ptr<PluginBase> plugin2 = std::make_shared<PluginBase>(
+      plugin_name, root_id, vm_id_, runtime_, plugin_config, fail_open, "plugin2_key");
+  EXPECT_TRUE(base_wasm_handle->wasm()->getRootContext(plugin2, false) == nullptr);
+
+  // Create context from a plugin2.
+  base_wasm_handle->wasm()->start(plugin2);
+  ContextBase *root_context2 = base_wasm_handle->wasm()->getRootContext(plugin2, false);
+  EXPECT_TRUE(root_context2 != nullptr);
+
+  // Verify that the 2 contexts can communicate through a shared queue.
+  std::string queue_name{"queue"};
+  SharedQueueDequeueToken token1;
+  EXPECT_EQ(root_context1->registerSharedQueue(queue_name, &token1), WasmResult::Ok);
+  SharedQueueDequeueToken token2;
+  EXPECT_EQ(root_context2->lookupSharedQueue(vm_id_, queue_name, &token2), WasmResult::Ok);
+  EXPECT_EQ(token1, token2);
+  for (int i = 0; i < 5; i++) {
+    root_context1->enqueueSharedQueue(token1, std::to_string(i));
+  }
+  std::string data;
+  for (int i = 0; i < 5; i++) {
+    root_context1->dequeueSharedQueue(token1, &data);
+    EXPECT_EQ(data, std::to_string(i));
+  }
+}
+
 } // namespace proxy_wasm