vdma (alibaba#801)

Add a common interface for DMA-like one-sided data communication

Introduce a unified API for performing direct memory access (DMA)-style operations with support for one-sided read/write semantics. The initial implementation utilizes shared memory as the underlying transport mechanism.

Author: hapulus798

CMakeLists.txt

@@ -202,6 +202,7 @@ file(GLOB PHOTON_SRC RELATIVE "${PROJECT_SOURCE_DIR}"
         net/*.cpp
         net/http/*.cpp
         net/security-context/tls-stream.cpp
+        net/vdma/*.cpp
         rpc/*.cpp
         thread/*.cpp
         )

examples/CMakeLists.txt

@@ -27,7 +27,10 @@ target_link_libraries(rpc-example-server PRIVATE photon_static)
 add_executable(sync-primitive sync-primitive/sync-primitive.cpp)
 target_link_libraries(sync-primitive PRIVATE photon_static)
 
+add_executable(vdma-example vdma-transfer/vdma-example.cpp)
+target_link_libraries(vdma-example PRIVATE photon_static)
+
 if (PHOTON_ENABLE_FSTACK_DPDK)
     add_executable(fstack-dpdk-demo fstack-dpdk/fstack-dpdk-demo.cpp)
     target_link_libraries(fstack-dpdk-demo PRIVATE ${FSTACK_LIBRARIES} photon_static)
-endif ()
+endif ()

examples/vdma-transfer/vdma-example.cpp

@@ -0,0 +1,52 @@
+#include <photon/photon.h>
+#include <photon/common/alog-stdstring.h>
+#include <photon/net/vdma.h>
+
+#include <string>
+#include <iomanip>
+#include <sstream>
+
+
+int main() {
+    photon::init();
+    DEFER(photon::fini());
+
+    // transfer procedure simple simulation
+    std::string shm_name = "foo";
+    size_t shm_size = 65536;    // 64KB
+    size_t unit = 4096;         // 4KB
+    // (1) create a target (want data)
+    // map shm to local addr space, and new a allocator to manage it
+    auto target = photon::new_shm_vdma_target(shm_name.c_str(), shm_size, unit);
+    // (2) create a initiator (has data)
+    auto initiator = photon::new_shm_vdma_initiator(shm_name.c_str(), shm_size);
+    // (3) target alloc a shared memory buffer
+    auto t_buffer = target->alloc(unit);
+    auto id = t_buffer->id();
+    LOG_DEBUG("step3: shared memory buffer real address is ", HEX((uint64_t)t_buffer->address()));
+    size_t want_data_size = 4096;
+    off_t want_data_offset = 0;
+    // (4) target send a request msg to initiator, tell initiator: logical addr + want which data
+    // (5) initiator use the logical addr map shm to its local addr space
+    auto i_buffer = initiator->map(id);
+    // (6) initiator copy corresponding data
+    char thedata[4096];
+    memset(thedata, 0x22, 4096);
+    memcpy((char*)i_buffer->address() + want_data_offset, thedata, want_data_size);
+    // (7) initiator single-side write
+    initiator->write(i_buffer, want_data_size, want_data_offset);
+    // (8) initiator unmap the shm buffer
+    initiator->unmap(i_buffer);
+    // (9) initiator send a response msg to target, tell target: transfer done
+    // (10) target check data
+    uint8_t* ptr = (uint8_t*)((char*)t_buffer->address() + want_data_offset);
+    for (size_t i=0; i<want_data_size; i++) {
+        if (ptr[i] != 0x22) {
+            target->dealloc(t_buffer);
+            LOG_ERROR_RETURN(0, -1, "failed, not same at ", i, "want ", 0x22, ", now ", ptr[i]);
+        }
+    }
+    // (11) target dealloc the shm buffer
+    target->dealloc(t_buffer);
+    LOG_INFO("success");
+}

include/photon/net/vdma.h

@@ -0,0 +1 @@
+../../../net/vdma.h

net/test/CMakeLists.txt

@@ -24,4 +24,8 @@ target_link_libraries(test-client PRIVATE photon_shared)
 
 add_executable(test-ipv6 test-ipv6.cpp)
 target_link_libraries(test-ipv6 PRIVATE photon_shared)
-add_test(NAME test-ipv6 COMMAND $<TARGET_FILE:test-ipv6>)
+add_test(NAME test-ipv6 COMMAND $<TARGET_FILE:test-ipv6>)
+
+add_executable(test-vdma test-vdma.cpp)
+target_link_libraries(test-vdma PRIVATE photon_shared)
+add_test(NAME test-vdma COMMAND $<TARGET_FILE:test-vdma>)

net/test/test-vdma.cpp

@@ -0,0 +1,240 @@
+#include "../../test/gtest.h"
+#include "../vdma/shm.cpp"
+#include <photon/photon.h>
+#include <photon/common/alog.h>
+#include <photon/thread/thread11.h>
+#include <photon/thread/workerpool.h>
+
+#include <vector>
+
+TEST(Buffer, encode_decode) {
+    std::string str;
+    photon::SharedMemoryBuffer::encode_to(str, 111, 222);
+    uint64_t idx = 0;
+    size_t bufsz = 0;
+    photon::SharedMemoryBuffer::decode_from(str, &idx, &bufsz);
+    EXPECT_EQ(111, idx);
+    EXPECT_EQ(222, bufsz);
+}
+
+TEST(Target, basic) {
+    photon::vDMATarget* target = photon::new_shm_vdma_target("foo", 65536, 4096);
+    DEFER(delete target);
+    photon::vDMAInitiator* initiator= photon::new_shm_vdma_initiator("foo", 65536);
+    DEFER(delete initiator);
+
+    std::string id_0_4096, id_1_4096, id_2_4096, id_3_4096;
+    photon::SharedMemoryBuffer::encode_to(id_0_4096, 0, 4096);
+    photon::SharedMemoryBuffer::encode_to(id_1_4096, 1, 4096);
+    photon::SharedMemoryBuffer::encode_to(id_2_4096, 2, 4096);
+    photon::SharedMemoryBuffer::encode_to(id_3_4096, 3, 4096);
+
+    // in this version, alloc size must same as unit
+    photon::vDMABuffer* buffer = target->alloc(512);
+    EXPECT_EQ(buffer, nullptr);
+    
+    // target alloc
+    photon::vDMABuffer* t_buffer0 = target->alloc(4096);
+    EXPECT_EQ(reinterpret_cast<photon::SharedMemoryBuffer*>(t_buffer0)->idx(), 0);
+    EXPECT_EQ(t_buffer0->buf_size(), 4096);
+    EXPECT_EQ(t_buffer0->id().size(), id_0_4096.size());
+    EXPECT_EQ(memcmp(t_buffer0->id().data(), id_0_4096.data(), id_0_4096.size()), 0);
+
+    photon::vDMABuffer* t_buffer1 = target->alloc(4096);
+    EXPECT_EQ(t_buffer1->id().size(), id_1_4096.size());
+    EXPECT_EQ(memcmp(t_buffer1->id().data(), id_1_4096.data(), id_1_4096.size()), 0);
+    EXPECT_EQ(t_buffer1->buf_size(), 4096);
+    EXPECT_EQ(t_buffer1->is_valid(), true);
+    EXPECT_EQ(t_buffer1->is_registered(), true);
+    EXPECT_EQ(t_buffer1->address(), (char*)t_buffer0->address()+4096);
+
+    photon::vDMABuffer* t_buffer2 = target->alloc(4096);
+    EXPECT_EQ(t_buffer2->id().size(), id_2_4096.size());
+    EXPECT_EQ(memcmp(t_buffer2->id().data(), id_2_4096.data(), id_2_4096.size()), 0);
+    EXPECT_EQ(t_buffer2->address(), (char*)t_buffer1->address()+4096);
+
+    // initiator map
+    photon::vDMABuffer* i_buffer0 = initiator->map(t_buffer0->id());
+    EXPECT_EQ(i_buffer0->buf_size(), t_buffer0->buf_size());
+    EXPECT_EQ(i_buffer0->id().size(), t_buffer0->id().size());
+    EXPECT_EQ(memcmp(i_buffer0->id().data(), t_buffer0->id().data(), 16), 0);
+
+    char tmp[4096];
+    memset(tmp, 0x5F, 4096);
+    memcpy(t_buffer0->address(), tmp, 4096);
+    EXPECT_EQ(memcmp(i_buffer0->address(), tmp, 4096), 0);
+    memset(tmp, 0x22, 4096);
+    memcpy(i_buffer0->address(), tmp, 4096);
+    EXPECT_EQ(memcmp(t_buffer0->address(), tmp, 4096), 0);
+
+    initiator->unmap(i_buffer0);
+    photon::vDMABuffer* t_buffer3 = target->alloc(4096);
+    EXPECT_EQ(t_buffer3->id(), id_3_4096);
+
+    i_buffer0 = initiator->map(t_buffer0->id());
+    photon::vDMABuffer* i_buffer1 = initiator->map(t_buffer1->id());
+
+    initiator->unmap(i_buffer0);
+    initiator->unmap(i_buffer1);
+
+    target->dealloc(t_buffer0);
+    target->dealloc(t_buffer1);
+    target->dealloc(t_buffer2);
+    target->dealloc(t_buffer3);
+}
+
+void func1(photon::vDMATarget* target, int id) {
+    auto buf = target->alloc(4096);
+    LOG_INFO("thread ", id, " get buf ", reinterpret_cast<photon::SharedMemoryBuffer*>(buf)->idx());
+    photon::thread_sleep(1);
+    target->dealloc(buf);
+    LOG_INFO("thread ", id, " release buf");
+}
+
+TEST(Target, single_vcpu_multi_thread_alloc_1) {
+    // 16 buffer, 16 thread(alloc - sleep 1s - free)
+    // The result should be that each thread is allocated a different buffer
+    auto target = photon::new_shm_vdma_target("foo", 65536, 4096);
+    std::vector<photon::join_handle*> threads;
+    for (int i=0; i<16; i++) {
+        threads.push_back(photon::thread_enable_join(photon::thread_create11(func1, target, i)));
+    }
+    LOG_INFO("all create success");
+    for (int i=0; i<16; i++) {
+        photon::thread_join(threads[i]);
+    }
+}
+
+void func2(photon::vDMATarget* target, int id) {
+    std::vector<int> stats(16, 0);
+    for (int i=0; i<1000000; i++) {
+        auto buf = target->alloc(4096);
+        if (!buf) {
+            LOG_INFO("some wrong ", id, " not get buffer in the middle");
+        }
+        for (int j=0; j<10; j++) {
+            photon::thread_yield();
+        }
+        stats[reinterpret_cast<photon::SharedMemoryBuffer*>(buf)->idx()] += 1;
+        target->dealloc(buf);
+    }
+    std::string output;
+    for (const auto& t : stats) output.append(std::to_string(t) + " ");
+    LOG_INFO(id, " res ", output.c_str());
+}
+
+TEST(Target, single_vcpu_multi_thread_alloc_2) {
+    // 16 buffer, 16 thread (repeat "alloc-free" 1000000 times)
+    // alloc should be all success, test hold-release lock
+    auto target = photon::new_shm_vdma_target("foo", 65536, 4096);
+    std::vector<photon::join_handle*> threads;
+    for (int i=0; i<16; i++) {
+        threads.push_back(photon::thread_enable_join(photon::thread_create11(func2, target, i)));
+    }
+    LOG_INFO("all create success");
+    for (int i=0; i<16; i++) {
+        photon::thread_join(threads[i]);
+    }
+}
+
+void func3(photon::vDMATarget* target, int id) {
+    auto buf = target->alloc(4096);
+    if (buf) {
+        LOG_INFO("thread ", id, " get buf ", reinterpret_cast<photon::SharedMemoryBuffer*>(buf)->idx());
+        photon::thread_sleep(10);
+        target->dealloc(buf);
+        LOG_INFO("thread ", id, " release buf");
+    }
+    else {
+        LOG_INFO("thread ", id, " not get buf");
+    }
+}
+
+TEST(Target, single_vcpu_multi_thread_alloc_3) {
+    // 16 buffer, 20 thread (alloc - sleep 10s - dealloc)
+    // only 16 thread can get buffer, other 4 thread will retry.
+    // now, sleep time is very long
+    // so, other 4 will get a nullptr in the end.
+    // After a while, 16 thread release their buffer.
+    auto target = photon::new_shm_vdma_target("foo", 65536, 4096);
+    std::vector<photon::join_handle*> threads;
+    for (int i=0; i<20; i++) {
+        threads.push_back(photon::thread_enable_join(photon::thread_create11(func3, target, i)));
+    }
+    LOG_INFO("all create success");
+    for (int i=0; i<20; i++) {
+        photon::thread_join(threads[i]);
+    }
+}
+
+void func4(photon::vDMATarget* target, int id) {
+    auto buf = target->alloc(4096);
+    if (buf) {
+        LOG_INFO("thread ", id, " get buf ", reinterpret_cast<photon::SharedMemoryBuffer*>(buf)->idx());
+        photon::thread_usleep(1);
+        target->dealloc(buf);
+        LOG_INFO("thread ", id, " release buf");
+    }
+    else {
+        LOG_INFO("thread ", id, " not get buf");
+    }
+}
+
+TEST(Target, single_vcpu_multi_thread_alloc_4) {
+    // 16 buffer, 20 thread (alloc - sleep 1s - dealloc)
+    // only 16 thread can get buffer at first, other thread will retry
+    // sleep time is short, other 4 thread may get a buffer
+    auto target = photon::new_shm_vdma_target("foo", 65536, 4096);
+    std::vector<photon::join_handle*> threads;
+    for (int i=0; i<20; i++) {
+        threads.push_back(photon::thread_enable_join(photon::thread_create11(func4, target, i)));
+    }
+    LOG_INFO("all create success");
+    for (int i=0; i<20; i++) {
+        photon::thread_join(threads[i]);
+    }
+}
+
+TEST(Target, multi_vcpu_multi_thread) {
+    // 16 buffer, 4 vcpu, each 16 thread (total 64 thread), 10000 task
+    auto target = photon::new_shm_vdma_target("foo", 65536, 4096);
+    photon::WorkPool wp(4, photon::INIT_EVENT_DEFAULT, photon::INIT_IO_NONE, 32);
+    photon::semaphore sem(0);
+    // rough statstic
+    std::vector<int> stats(16, 0);
+    int n_notget = 0;
+    for (int i=0; i<10000; i++) {
+        wp.async_call(new auto ([&]{
+            auto buf = target->alloc(4096);
+            if (buf) {
+                int buf_idx = reinterpret_cast<photon::SharedMemoryBuffer*>(buf)->idx();
+                LOG_INFO("task ", i, " get buf ", buf_idx);
+                photon::thread_usleep(1);
+                target->dealloc(buf);
+                LOG_INFO("task ", i, " release buf ", buf_idx);
+                stats[buf_idx] += 1;
+            }
+            else {
+                LOG_INFO("task ", i, " not get buf");
+                n_notget++;
+            }
+            sem.signal(1);
+        }));
+    }
+
+    sem.wait(10000);    // wait most async task to complete
+
+    std::string output;
+    for (int i=0; i<16; i++) output.append(std::to_string(stats[i]) + " ");
+    LOG_INFO(output.c_str());
+    LOG_INFO("get failed: ", n_notget);
+}
+
+int main(int argc, char **argv){
+    srand(time(nullptr));
+    ::testing::InitGoogleTest(&argc, argv);
+    if (photon::init(photon::INIT_EVENT_DEFAULT, photon::INIT_IO_NONE))
+        return -1;
+    DEFER(photon::fini());
+    return RUN_ALL_TESTS();
+}