Adding Com Channel fast data path capabilities

Mellanox · Dec 25, 2023 · ff45a73 · ff45a73
1 parent 2ccd99a
commit ff45a73
Show file tree

Hide file tree

Showing 7 changed files with 515 additions and 24 deletions.
diff --git a/src/client.cpp b/src/client.cpp
@@ -888,8 +888,7 @@ int Client<IoType, SwitchCycleDuration, PongModeCare>::initBeforeLoop() {
 #if defined(USING_DOCA_COMM_CHANNEL_API)
             if (s_user_params.doca_comm_channel) {
                 // Waiting for connection
-                struct cc_ctx_client *ctx_client = (struct cc_ctx_client *)data->doca_cc_ctx;
-                while (ctx_client->state != CC_CONNECTED) {
+                while (data->doca_cc_ctx->state != CC_CONNECTED) {
                     doca_pe_progress(s_user_params.pe);
                 }
                 log_dbg("[fd=%d] Client connected successfully", ifd);

diff --git a/src/common.h b/src/common.h
@@ -115,6 +115,7 @@ static inline int msg_sendto(int fd, uint8_t *buf, int nbytes,
 #if defined(USING_DOCA_COMM_CHANNEL_API)
         if (s_user_params.doca_comm_channel) {
             doca_error_t doca_error;
+            struct doca_cc_producer_send_task *producer_task;
             struct doca_cc_send_task *task;
             struct doca_task *task_obj;
             struct timespec ts = {
@@ -123,14 +124,26 @@ static inline int msg_sendto(int fd, uint8_t *buf, int nbytes,
             };
             struct cc_ctx *ctx = g_fds_array[fd]->doca_cc_ctx;
             do {
-                if (s_user_params.mode == MODE_SERVER) {
-                    struct cc_ctx_server *ctx_server = (struct cc_ctx_server*)ctx;
-                    doca_error = doca_cc_server_send_task_alloc_init(ctx_server->server, ctx_server->ctx.connection, buf,
-                                                                    nbytes, &task);
-                } else { // MODE_CLIENT
-                    struct cc_ctx_client *ctx_client = (struct cc_ctx_client *)ctx;
-                    doca_error = doca_cc_client_send_task_alloc_init(ctx_client->client, ctx_client->ctx.connection, buf,
-                                                                    nbytes, &task);
+                if (s_user_params.doca_fast_path) {
+                    struct doca_buf *doca_buf;
+                    //doca_buf_get_data(doca_buf,)
+                    doca_error = doca_buf_inventory_buf_get_by_data(ctx->ctx_fifo.producer_mem.buf_inv, ctx->ctx_fifo.producer_mem.mmap,
+                                ctx->ctx_fifo.consumer_mem.mem, nbytes, &doca_buf);
+                    if (doca_error != DOCA_SUCCESS) {
+		                log_err("Failed to get doca buf from producer mmap with error = %s", doca_error_get_name(doca_error));
+	                }
+                    doca_error = doca_cc_producer_send_task_alloc_init(ctx->ctx_fifo.producer, doca_buf,
+                                                    ctx->ctx_fifo.remote_consumer_id, &producer_task);
+                } else { // Not doca fast path
+                    if (s_user_params.mode == MODE_SERVER) {
+                        struct cc_ctx_server *ctx_server = (struct cc_ctx_server*)ctx;
+                        doca_error = doca_cc_server_send_task_alloc_init(ctx_server->server, ctx_server->ctx.connection, buf,
+                                                                        nbytes, &task);
+                    } else { // MODE_CLIENT
+                        struct cc_ctx_client *ctx_client = (struct cc_ctx_client *)ctx;
+                        doca_error = doca_cc_client_send_task_alloc_init(ctx_client->client, ctx_client->ctx.connection, buf,
+                                                                        nbytes, &task);
+                    }
                 }
                 if (doca_error == DOCA_ERROR_NO_MEMORY) {
                     // Queue is full of tasks, need to free tasks with completion callback
@@ -147,7 +160,11 @@ static inline int msg_sendto(int fd, uint8_t *buf, int nbytes,
                     ret = RET_SOCKET_SHUTDOWN;
                 }
             } else { // task_alloc_init succeeded
-                task_obj = doca_cc_send_task_as_task(task);
+                if (s_user_params.doca_fast_path) {
+                    task_obj = doca_cc_producer_send_task_as_task(producer_task);
+                } else {
+                    task_obj = doca_cc_send_task_as_task(task);
+                }
                 do {
                     doca_error = doca_task_submit(task_obj);
                     if (doca_error == DOCA_ERROR_AGAIN) {

diff --git a/src/defs.h b/src/defs.h
@@ -301,7 +301,8 @@ enum {
 #if defined(USING_DOCA_COMM_CHANNEL_API)
     OPT_DOCA,
     OPT_PCI,
-    OPT_PCI_REP
+    OPT_PCI_REP,
+    OPT_DOCA_FAST_PATH
 #endif /* USING_DOCA_COMM_CHANNEL_API */
 
 };
@@ -820,6 +821,7 @@ struct user_params_t {
 #endif /* DEFINED_TLS */
 #if defined(USING_DOCA_COMM_CHANNEL_API)
     bool doca_comm_channel = false;                         /* Flag to indicate using Com Channel*/
+    bool doca_fast_path = false;                            /* Flag to indicate using fast path*/
     char cc_dev_pci_addr[PCI_ADDR_LEN];                     /* Comm Channel DOCA device PCI address */
     char cc_dev_rep_pci_addr[PCI_ADDR_LEN];                 /* Comm Channel DOCA device representor PCI address */
     struct doca_pe *pe = nullptr;                           /* Progress engine for doca, one per thread*/