[Feat] add ucx based transport layer

.github/workflows/build-release.yaml

@@ -111,6 +111,8 @@ jobs:
     permissions:
       id-token: write
     name: publish
+    env:
+      CMAKE_POLICY_VERSION_MINIMUM: 3.5
     steps:
       - name: Free Disk Space
         uses: jlumbroso/free-disk-space@main
@@ -131,7 +133,6 @@ jobs:
                               gcc \
                               g++ \
                               make \
-                              cmake \
                               binutils \
                               libibverbs-dev \
                               librdmacm-dev \
@@ -143,6 +144,24 @@ jobs:
                               libhiredis-dev \
                               liburing-dev
 
+          # install cmake
+          cd /tmp && \
+          wget -q https://cmake.org/files/v4.1/cmake-4.1.3-linux-x86_64.sh && \
+          sudo bash cmake-4.1.3-linux-x86_64.sh --skip-license --prefix=/usr && \
+          rm cmake-4.1.3-linux-x86_64.sh && \
+          cd -
+
+          # libevhtp uses TestEndianess.c.in
+          sudo ln -s /usr/share/cmake-4.1/Modules/TestEndianness.c.in /usr/share/cmake-4.1/Modules/TestEndianess.c.in
+
+          # install ucx
+          cd /tmp && \
+          wget -q https://github.com/openucx/ucx/releases/download/v1.19.0/ucx-1.19.0-ubuntu22.04-mofed5-cuda12-x86_64.tar.bz2 && \
+          tar xvf ucx-1.19.0-ubuntu22.04-mofed5-cuda12-x86_64.tar.bz2 && \
+          sudo dpkg -i ucx-1.19.0.deb ucx-cuda-1.19.0.deb ucx-xpmem-1.19.0.deb && \
+          rm /tmp/ucx-1.19.0-ubuntu22.04-mofed5-cuda12-x86_64.tar.bz2 /tmp/ucx-1.19.0.deb /tmp/ucx-cuda-1.19.0.deb /tmp/ucx-xpmem-1.19.0.deb && \
+          cd -
+
           # install python dependencies
           pip install -r requirements.txt
       - name: Build

.github/workflows/build-test.yaml

@@ -44,6 +44,8 @@ jobs:
       matrix:
         os: [ubuntu-22.04]
         python-version: ["3.11", "3.12"]
+    env:
+      CMAKE_POLICY_VERSION_MINIMUM: 3.5
     steps:
       - name: Free Disk Space
         uses: jlumbroso/free-disk-space@main
@@ -71,7 +73,6 @@ jobs:
                               gcc \
                               g++ \
                               make \
-                              cmake \
                               binutils \
                               libibverbs-dev \
                               librdmacm-dev \
@@ -83,6 +84,24 @@ jobs:
                               libhiredis-dev \
                               liburing-dev
 
+          # install cmake
+          cd /tmp && \
+          wget -q https://cmake.org/files/v4.1/cmake-4.1.3-linux-x86_64.sh && \
+          sudo bash cmake-4.1.3-linux-x86_64.sh --skip-license --prefix=/usr && \
+          rm cmake-4.1.3-linux-x86_64.sh && \
+          cd -
+
+          # libevhtp uses TestEndianess.c.in
+          sudo ln -s /usr/share/cmake-4.1/Modules/TestEndianness.c.in /usr/share/cmake-4.1/Modules/TestEndianess.c.in
+
+          # install ucx
+          cd /tmp && \
+          wget -q https://github.com/openucx/ucx/releases/download/v1.19.0/ucx-1.19.0-ubuntu22.04-mofed5-cuda12-x86_64.tar.bz2 && \
+          tar xvf ucx-1.19.0-ubuntu22.04-mofed5-cuda12-x86_64.tar.bz2 && \
+          sudo dpkg -i ucx-1.19.0.deb ucx-cuda-1.19.0.deb ucx-xpmem-1.19.0.deb && \
+          rm /tmp/ucx-1.19.0-ubuntu22.04-mofed5-cuda12-x86_64.tar.bz2 /tmp/ucx-1.19.0.deb /tmp/ucx-cuda-1.19.0.deb /tmp/ucx-xpmem-1.19.0.deb && \
+          cd -
+
           # install clang-format-19
           sudo curl -fsSL --retry 3 --retry-delay 2 https://github.com/muttleyxd/clang-tools-static-binaries/releases/download/master-46b8640/clang-format-19_linux-amd64 -o /usr/bin/clang-format-19 || { echo "ERROR: Failed to download clang-format-19"; exit 1; }
           test -s /usr/bin/clang-format-19 || { echo "ERROR: Downloaded file is empty"; exit 1; }

README.md

@@ -4,9 +4,10 @@
 [![License](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](LICENSE)
 
 PrisKV is specifically designed for modern high-performance computing (HPC) and
-artificial intelligence (AI) computing. It solely supports RDMA. PrisKV also
-supports GDR (GPU Direct RDMA), enabling the value of a key to be directly
-transferred between PrisKV and the GPU.
+artificial intelligence (AI) computing. It supports common transport protocols,
+including RDMA, TCP, and shared memory, to enable efficient communication for
+different scenarios. PrisKV also supports GDR (GPU Direct RDMA), enabling the
+value of a key to be directly transferred between PrisKV and the GPU.
 
 ## How to Build
 

client/Makefile

@@ -33,7 +33,7 @@ VALKEY_LDFLAGS = -I$(VALKEY_INCLUDE_PATH)
 PRISKV_TARGETS_ALL = priskv-client priskv-benchmark priskv-example priskv-test_runtime
 PRISKV_TARGETS = priskv-client priskv-benchmark
 PRISKV_TARGETS_SRCS = $(patsubst priskv-%, %.c, $(PRISKV_TARGETS_ALL))
-SRCS = $(filter-out $(PRISKV_TARGETS_SRCS) $(VALKEY_BENCHMARK_SRC), $(wildcard *.c))
+SRCS = $(filter-out $(PRISKV_TARGETS_SRCS) $(VALKEY_BENCHMARK_SRC), $(wildcard *.c transport/*.c))
 OBJS := $(SRCS:%.c=%.o)
 DEPS := $(OBJS:%.o=%.d)
 
@@ -47,6 +47,9 @@ LIBIBVERBS_VERSION_MAJOR = $(shell echo ${LIBIBVERBS_VERSION} | cut -d "." -f 1)
 LIBIBVERBS_VERSION_MINOR = $(shell echo ${LIBIBVERBS_VERSION} | cut -d "." -f 2)
 RDMA_LDFLAGS = -lrdmacm -libverbs -DLIBIBVERBS_VERSION_MAJOR=$(LIBIBVERBS_VERSION_MAJOR) -DLIBIBVERBS_VERSION_MINOR=$(LIBIBVERBS_VERSION_MINOR)
 
+UCX_CFLAGS = $(shell pkg-config --cflags ucx)
+UCX_LDFLAGS = $(shell pkg-config --libs ucx)
+
 CUDA_HOME ?= /usr/local/cuda
 CUDA_LIB ?= $(CUDA_HOME)/lib64
 CUDA_INC ?= $(CUDA_HOME)/include
@@ -102,13 +105,13 @@ $(THIRDPARTY_SRCS):
 	git submodule update --init --recursive ../thirdparty/linenoise
 
 $(PRISKV_TARGETS): priskv-%: %.c $(STATIC_LIB) $(THIRDPARTY_SRCS)
-	$(CC) $< $(THIRDPARTY_SRCS) -o $@ $(STATIC_LIB) $(CFLAGS) $(RDMA_LDFLAGS) $(THIRDPARTY_INCFLAGS)
+	$(CC) $< $(THIRDPARTY_SRCS) -o $@ $(STATIC_LIB) $(CFLAGS) $(RDMA_LDFLAGS) $(UCX_CFLAGS) $(UCX_LDFLAGS) $(THIRDPARTY_INCFLAGS)
 
 $(VALKEY_STATIC_LIB):
 	cd $(VALKEY_PATH) && USE_RDMA=1 make
 
 $(VALKEY_BENCHMARK_NAME): $(VALKEY_BENCHMARK_SRC) $(STATIC_LIB) $(VALKEY_STATIC_LIB)
-	$(CC) $^ $(CFLAGS) $(RDMA_LDFLAGS) $(CUDA_LDFLAGS) $(VALKEY_LDFLAGS) -o $@
+	$(CC) $^ $(CFLAGS) $(RDMA_LDFLAGS) $(UCX_CFLAGS) $(UCX_LDFLAGS) $(CUDA_LDFLAGS) $(VALKEY_LDFLAGS) -o $@
 
 install: $(PRISKV_TARGETS)
 	install -m 755 -d $(PRISKV_DESTDIR)/$(PRISKV_BINPATH)
@@ -129,3 +132,4 @@ clean:
 
 format:
 	$(FMT) -i *.c *.h
+	$(FMT) -i transport/*.c transport/*.h

client/benchmark.c

@@ -132,6 +132,7 @@ typedef struct {
     int (*get_fd)(void *ctx);
     void (*handler)(int fd, void *opaque, uint32_t events);
     const char *(*status_str)(int status);
+    bool (*is_error)(int status);
     void (*get)(void *ctx, const char *key, void *value, uint32_t value_len,
                 void (*cb)(int, void *), void *cbarg);
     void (*set)(void *ctx, const char *key, void *value, uint32_t value_len,
@@ -1049,6 +1050,12 @@ static void priskv_drv_test(void *ctx, const char *key, void (*cb)(int, void *),
     priskv_test_async(priskv_ctx->client, key, (uint64_t)priskv_req_ctx, priskv_req_cb);
 }
 
+static bool priskv_drv_is_error(int status)
+{
+    return status != PRISKV_STATUS_OK && status != PRISKV_STATUS_NO_SUCH_KEY &&
+           status != PRISKV_STATUS_KEY_UPDATING;
+}
+
 static const kv_driver priskv_drv = {
     .name = "priskv",
     .transfer = false,
@@ -1057,6 +1064,7 @@ static const kv_driver priskv_drv = {
     .get_fd = priskv_drv_get_fd,
     .handler = priskv_drv_handler,
     .status_str = priskv_drv_status_str,
+    .is_error = priskv_drv_is_error,
     .get = priskv_drv_get,
     .set = priskv_drv_set,
     .del = priskv_drv_del,
@@ -1071,6 +1079,7 @@ static const kv_driver priskv_drv_transfer = {
     .get_fd = priskv_drv_get_fd,
     .handler = priskv_drv_handler,
     .status_str = priskv_drv_status_str,
+    .is_error = priskv_drv_is_error,
     .get = priskv_drv_get_transfer,
     .set = priskv_drv_set_transfer,
     .del = priskv_drv_del,
@@ -1161,7 +1170,7 @@ static void job_cb(int status, void *arg)
 {
     job_context *job = arg;
 
-    if (status) {
+    if (job->kv_drv->is_error(status)) {
         job_set_error(job, "resp status[%d]: %s", status, job->kv_drv->status_str(status));
     }
 
@@ -1647,9 +1656,6 @@ static int job_init(job_context *job, int threadid)
     }
 
     job->threadid = threadid;
-    priskv_set_fd_handler(job->epollfd, job_process, NULL, job);
-    priskv_thread_add_event_handler(priskv_threadpool_get_iothread(g_threadpool, job->threadid),
-                                  job->epollfd);
 
     printf("job[%d] finish to init\n", threadid);
     return 0;
@@ -1677,6 +1683,9 @@ static void job_kick(job_context *job)
 {
     uint64_t u = 1;
 
+    priskv_set_fd_handler(job->epollfd, job_process, NULL, job);
+    priskv_thread_add_event_handler(priskv_threadpool_get_iothread(g_threadpool, job->threadid),
+                                    job->epollfd);
     write(job->eventfd, &u, sizeof(u));
 }
 

client/client.c

@@ -37,6 +37,7 @@
 #include "priskv-log.h"
 #include "priskv-logo.h"
 #include "linenoise.h"
+#include "transport/transport.h"
 
 #define VALUE_MAX_LEN (64 * 1024)
 

client/priskv.h

@@ -44,16 +44,17 @@ typedef struct priskv_memory priskv_memory;
  * @lport: local port. Ignore port on NULL @laddr
  * @nqueue: the number of worker threads
  */
-priskv_client *priskv_connect(const char *raddr, int rport, const char *laddr, int lport, int nqueue);
+priskv_client *priskv_connect(const char *raddr, int rport, const char *laddr, int lport,
+                              int nqueue);
 
 /* Close a client context created by @priskv_connect */
 void priskv_close(priskv_client *client);
 
 /* Get memory handler
  * @offset: uint64_t type on valid fd(>= 0); or void * type on invalid fd.
  */
-priskv_memory *priskv_reg_memory(priskv_client *client, uint64_t offset, size_t length, uint64_t iova,
-                             int fd);
+priskv_memory *priskv_reg_memory(priskv_client *client, uint64_t offset, size_t length,
+                                 uint64_t iova, int fd);
 
 /* Put a memory handler
  * @mem: created by @priskv_reg_memory
@@ -122,8 +123,8 @@ typedef enum priskv_status {
     /* RDMA disconnected from the server side */
     PRISKV_STATUS_DISCONNECTED = 0xF00,
 
-    /* local RDMA error occurs */
-    PRISKV_STATUS_RDMA_ERROR,
+    /* local transport error occurs */
+    PRISKV_STATUS_TRANSPORT_ERROR,
 
     /* does inflight requests exceed @max_inflight_command? */
     PRISKV_STATUS_BUSY,
@@ -180,8 +181,8 @@ static inline const char *priskv_status_str(priskv_status status)
     case PRISKV_STATUS_DISCONNECTED:
         return "Disconnected";
 
-    case PRISKV_STATUS_RDMA_ERROR:
-        return "RDMA error";
+    case PRISKV_STATUS_TRANSPORT_ERROR:
+        return "Transport error";
 
     case PRISKV_STATUS_BUSY:
         return "Busy";
@@ -218,30 +219,31 @@ typedef void (*priskv_generic_cb)(uint64_t request_id, priskv_status status, voi
  *                                PRISKV_STATUS_VALUE_TOO_BIG as soon as possible.
  */
 int priskv_get_async(priskv_client *client, const char *key, priskv_sgl *sgl, uint16_t nsgl,
-                   uint64_t request_id, priskv_generic_cb cb);
+                     uint64_t request_id, priskv_generic_cb cb);
 
 /* Set value of a key */
 int priskv_set_async(priskv_client *client, const char *key, priskv_sgl *sgl, uint16_t nsgl,
-                   uint64_t timeout, uint64_t request_id, priskv_generic_cb cb);
+                     uint64_t timeout, uint64_t request_id, priskv_generic_cb cb);
 
 /* Test a key-value exist or not */
-int priskv_test_async(priskv_client *client, const char *key, uint64_t request_id, priskv_generic_cb cb);
+int priskv_test_async(priskv_client *client, const char *key, uint64_t request_id,
+                      priskv_generic_cb cb);
 
 /* Delete a key-value exist or not */
 int priskv_delete_async(priskv_client *client, const char *key, uint64_t request_id,
-                      priskv_generic_cb cb);
+                        priskv_generic_cb cb);
 
 /* Set expire time of a key */
-int priskv_expire_async(priskv_client *client, const char *key, uint64_t timeout, uint64_t request_id,
-                      priskv_generic_cb cb);
+int priskv_expire_async(priskv_client *client, const char *key, uint64_t timeout,
+                        uint64_t request_id, priskv_generic_cb cb);
 
 /* Get the number of keys which match the @regex */
 int priskv_nrkeys_async(priskv_client *client, const char *regex, uint64_t request_id,
-                      priskv_generic_cb cb);
+                        priskv_generic_cb cb);
 
 /* Flush the keys which match the @regex */
 int priskv_flush_async(priskv_client *client, const char *regex, uint64_t request_id,
-                     priskv_generic_cb cb);
+                       priskv_generic_cb cb);
 
 /* for *KEYS* command */
 typedef struct priskv_key {
@@ -259,13 +261,14 @@ void priskv_keyset_free(priskv_keyset *keyset);
 
 /* Get the keys which match the @regex and return the result in priskv_generic_cb */
 int priskv_keys_async(priskv_client *client, const char *regex, uint64_t request_id,
-                    priskv_generic_cb cb);
+                      priskv_generic_cb cb);
 
 /* sync APIs */
 int priskv_get(priskv_client *client, const char *key, priskv_sgl *sgl, uint16_t nsgl,
-             uint32_t *valuelen);
+               uint32_t *valuelen);
 
-int priskv_set(priskv_client *client, const char *key, priskv_sgl *sgl, uint16_t nsgl, uint64_t timeout);
+int priskv_set(priskv_client *client, const char *key, priskv_sgl *sgl, uint16_t nsgl,
+               uint64_t timeout);
 
 int priskv_test(priskv_client *client, const char *key, uint32_t *valuelen);