rdma_op.c

/* SPDX-License-Identifier: Apache-2.0
 * Copyright 2026 Zhaochen Zhang and Alibaba Cloud Computing Co., Ltd.
 */

#include "rdma_op.h"

#include <rte_cycles.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include "rdma_rc.h"
#include "state_machine.h"

// Global array of per-QP RDMA operation info
static struct rdma_op_qp_info *g_qp_infos = NULL;
static uint32_t g_qp_count = 0;
static uint32_t g_round_robin_next = 0;
static int g_initialized = 0;

// Initialize QP RDMA operation management (including depth tracking and state machines)
int init_qp_rdma_ops(uint32_t depth_per_qp)
{
	if (g_initialized) {
		printf("QP RDMA ops already initialized\n");
		return 0;
	}

	g_qp_count = rdma_ctx.local_qp_cnt;
	if (g_qp_count == 0) {
		printf("No QPs available for RDMA ops\n");
		return -1;
	}

	g_qp_infos = calloc(g_qp_count, sizeof(struct rdma_op_qp_info));
	if (!g_qp_infos) {
		printf("Failed to allocate memory for QP RDMA ops\n");
		return -1;
	}

	// Initialize the info for each QP
	for (uint32_t i = 0; i < g_qp_count; i++) {
		g_qp_infos[i].pending_wr = 0;
		uint32_t configured_depth = 0;
		if (g_qp_state_machine_info.qp_max_depth && i < (uint32_t)g_qp_state_machine_info.qp_count) {
			configured_depth = g_qp_state_machine_info.qp_max_depth[i];
			// printf("DEBUG QP %u max depth: %u\n", i, configured_depth);
		}
		g_qp_infos[i].max_depth = (configured_depth > 0) ? configured_depth : depth_per_qp;
		g_qp_infos[i].total_posted = 0;
		g_qp_infos[i].total_completed = 0;
		g_qp_infos[i].full_count = 0;
		g_qp_infos[i].state_machine = NULL;
		g_qp_infos[i].state_machine_enabled = false;
	}

	g_round_robin_next = 0;
	g_initialized = 1;

	printf("Initialized QP RDMA ops for %u QPs with max depth %u\n", g_qp_count, depth_per_qp);
	return 0;
}

// Round-robin selection of an available QP
static int select_qp_round_robin(void)
{
	if (!g_initialized || !g_qp_infos) {
		return -1;
	}

	uint32_t start_qp = g_round_robin_next;
	for (uint32_t i = 0; i < g_qp_count / qp_mod; i++) {
		uint32_t qp_idx = (start_qp + i * qp_mod) % g_qp_count;
		if (g_qp_infos[qp_idx].pending_wr < g_qp_infos[qp_idx].max_depth) {
			g_round_robin_next = (qp_idx + 1 * qp_mod) % g_qp_count;
			return qp_idx;
		}
	}
	return -1;  // No available QP
}

// Free WR bundle resources
static void clean_wr_bundle(struct wr_bundle *bundle)
{
	if (bundle) {
		free(bundle->sge);
		free(bundle->wr);
		free(bundle);
	}
}

// Post a write on a single QP (non-blocking)
int post_write_wqe(uint8_t *buf, uint32_t len, int qp_index, uint64_t remote_addr)
{
	if (!g_initialized || !g_qp_infos) {
		printf("QP RDMA ops not initialized\n");
		return -1;
	}

	if (qp_index < 0 || qp_index >= (int)g_qp_count) {
		printf("Invalid QP index: %d\n", qp_index);
		return -1;
	}

	if (g_qp_infos[qp_index].pending_wr >= g_qp_infos[qp_index].max_depth) {
		g_qp_infos[qp_index].full_count++;
		// printf("QP %d is full\n", qp_index);
		return -2;  // QP is full
	}

	// Allocate the WR bundle
	struct wr_bundle *bundle = malloc(sizeof(struct wr_bundle));
	if (!bundle) {
		printf("Failed to allocate WR bundle\n");
		return -1;
	}

	bundle->sge = malloc(sizeof(struct ibv_sge));
	bundle->wr = malloc(sizeof(struct ibv_send_wr));
	if (!bundle->sge || !bundle->wr) {
		printf("Failed to allocate SGE or WR\n");
		clean_wr_bundle(bundle);
		return -1;
	}

	// Configure the SGE
	bundle->sge->addr = (uint64_t)buf;
	bundle->sge->length = len;
	bundle->sge->lkey = rdma_ctx.mr->lkey;

	// Configure the WR
	bundle->wr->opcode = IBV_WR_RDMA_WRITE;
	bundle->wr->wr.rdma.rkey = rdma_ctx.remote_conn_qp_data->rkey;
	bundle->wr->wr.rdma.remote_addr = (remote_addr == 0) ? rdma_ctx.remote_conn_qp_data->buf_addr : remote_addr;
	bundle->wr->num_sge = 1;
	bundle->wr->sg_list = bundle->sge;
	bundle->wr->send_flags = IBV_SEND_SIGNALED;
	bundle->wr->next = NULL;
	bundle->wr->wr_id = ENCODE_WR_ID(qp_index, bundle);

	// Record the post timestamp for completion latency calculation
	bundle->post_tsc = rte_get_tsc_cycles();

	// Post send
	struct ibv_send_wr *bad_wr;
	int ret = ibv_post_send(rdma_ctx.qp[qp_index], bundle->wr, &bad_wr);

	// Immediately record the Post Send timestamp
	record_post_send_timestamp(qp_index);

	if (ret != 0) {
		printf("ibv_post_send failed for QP %d: %d\n", qp_index, ret);
		clean_wr_bundle(bundle);
		return -1;
	}

	// Update statistics
	g_qp_infos[qp_index].pending_wr++;
	g_qp_infos[qp_index].total_posted++;

	return 0;
}

// Post a write with automatic QP selection
int post_write_auto(uint8_t *buf, uint64_t len, int blocking)
{
	if (!g_initialized) {
		printf("QP depth tracking not initialized\n");
		return -1;
	}

	int qp_index;

	if (blocking) {
		// Blocking mode: wait until a QP becomes available
		while (!force_quit) {
			qp_index = select_qp_round_robin();
			if (qp_index >= 0) {
				break;
			}
			// Try to poll some completions to free up space
			managed_poll_cq(8, 1);
			if (qp_index < 0) {
				rte_delay_us(100);  // Brief wait
			}
		}

		if (force_quit) {
			return -1;
		}
	} else {
		// Non-blocking mode: return immediately
		qp_index = select_qp_round_robin();
		if (qp_index < 0) {
			return -2;  // No available QP
		}
	}

	return post_write_wqe(buf, len, qp_index, 0);
}

// Managed CQ polling
int managed_poll_cq(int max_completions, int timeout_ms)
{
	if (!g_initialized) {
		printf("QP RDMA ops not initialized\n");
		return -1;
	}

	if (max_completions <= 0) {
		max_completions = 16;  // Default value
	}

	struct ibv_wc *wc = malloc(sizeof(struct ibv_wc) * max_completions);
	if (!wc) {
		printf("Failed to allocate WC array\n");
		return -1;
	}

	int completed = 0;
	uint64_t start_time = 0;
	uint64_t timeout_cycles = 0;

	if (timeout_ms > 0) {
		start_time = rte_get_tsc_cycles();
		timeout_cycles = timeout_ms * rte_get_tsc_hz() / 1000;
	}

	while (completed < max_completions && !force_quit) {
		int ret = ibv_poll_cq(rdma_ctx.cq, max_completions - completed, &wc[completed]);

		if (ret > 0) {
			// Process the completed WCs
			for (int i = 0; i < ret; i++) {
				uint32_t qp_idx = DECODE_QP_INDEX(wc[completed + i].wr_id);
				struct wr_bundle *bundle = (struct wr_bundle *)DECODE_BUNDLE_PTR(wc[completed + i].wr_id);

				// Compute the completion latency (post -> completion)
				uint64_t completion_tsc = rte_get_tsc_cycles();
				if (bundle && bundle->post_tsc > 0) {
					uint64_t latency_tsc = completion_tsc - bundle->post_tsc;
					record_completion_latency(qp_idx, latency_tsc);
				}

				// Update the QP depth count
				if (qp_idx < g_qp_count) {
					if (g_qp_infos[qp_idx].pending_wr > 0) {
						g_qp_infos[qp_idx].pending_wr--;
					}
					g_qp_infos[qp_idx].total_completed++;
				}

				// Free resources
				clean_wr_bundle(bundle);

				// Error handling
				if (wc[completed + i].status != IBV_WC_SUCCESS) {
					printf("WC error for QP %u: %s (wr_id: 0x%lx)\n", qp_idx,
					       ibv_wc_status_str(wc[completed + i].status), wc[completed + i].wr_id);
				}
			}
			completed += ret;
		} else if (ret < 0) {
			printf("ibv_poll_cq failed: %d\n", ret);
			break;
		}

		// If there are no more completions and waiting is not required, exit
		if (ret == 0 && timeout_ms == 0) {
			break;
		}

		// Timeout check
		if (timeout_ms > 0) {
			uint64_t current_time = rte_get_tsc_cycles();
			if (current_time - start_time > timeout_cycles) {
				break;
			}
		}

		// Brief delay to avoid busy-waiting
		if (ret == 0) {
			rte_delay_us(10);
		}
	}

	free(wc);
	return completed;
}

// Get QP status info
int get_qp_depth_status(int qp_index, uint32_t *pending, uint32_t *max_depth)
{
	if (!g_initialized || !g_qp_infos) {
		return -1;
	}

	if (qp_index < 0 || qp_index >= (int)g_qp_count) {
		return -1;
	}

	if (pending) {
		*pending = g_qp_infos[qp_index].pending_wr;
	}
	if (max_depth) {
		*max_depth = g_qp_infos[qp_index].max_depth;
	}

	return 0;
}

// Get the number of available QPs
int get_empty_qp_count(void)
{
	if (!g_initialized || !g_qp_infos) {
		return -1;
	}

	int empty = 0;
	for (uint32_t i = 0; i < g_qp_count; i++) {
		if (g_qp_infos[i].pending_wr == 0) {
			empty++;
		}
	}

	return empty;
}

// Print the depth status of all QPs
void print_qp_depth_status(void)
{
	if (!g_initialized || !g_qp_infos) {
		printf("QP RDMA ops not initialized\n");
		return;
	}

	printf("QP Depth Status:\n");
	printf("QP Index | Pending | Max | Posted | Completed |   Full  | Utilization | StateMachine\n");
	printf("---------|---------|-----|--------|-----------|--------|-------------|-------------\n");

	for (uint32_t i = 0; i < g_qp_count; i++) {
		double utilization =
		    g_qp_infos[i].max_depth > 0 ? (double)g_qp_infos[i].pending_wr / g_qp_infos[i].max_depth * 100.0 : 0.0;

		const char *sm_status = g_qp_infos[i].state_machine_enabled ? "Enabled" : "Disabled";

		printf("   %2u    |   %2u    | %2u  | %6u | %9u | %6u | %8.1f%% | %s\n", i, g_qp_infos[i].pending_wr,
		       g_qp_infos[i].max_depth, g_qp_infos[i].total_posted, g_qp_infos[i].total_completed,
		       g_qp_infos[i].full_count, utilization, sm_status);
	}
	printf("\n");
}

// Free resources
void cleanup_qp_rdma_ops(void)
{
	if (g_qp_infos) {
		// Destroy each QP's state machine
		for (uint32_t i = 0; i < g_qp_count; i++) {
			if (g_qp_infos[i].state_machine) {
				destroy_state_machine(g_qp_infos[i].state_machine);
				g_qp_infos[i].state_machine = NULL;
			}
		}

		free(g_qp_infos);
		g_qp_infos = NULL;
	}

	g_qp_count = 0;
	g_round_robin_next = 0;
	g_initialized = 0;

	printf("QP RDMA ops cleaned up\n");
}

/**
 * @brief Context structure for a per-QP RDMA send task
 */
typedef struct {
	int qp_index;          // QP index
	uint8_t *buf;          // Send buffer
	uint32_t len;          // Send length (WQE size)
	uint64_t remote_addr;  // Remote address
} qp_rdma_task_context_t;

/**
 * @brief Per-QP RDMA send task function
 */
static void qp_rdma_send_task(void *context)
{
	qp_rdma_task_context_t *ctx = (qp_rdma_task_context_t *)context;

	// Use the modified managed_post_write_single; remote_addr=0 means use the default address
	int ret = post_write_wqe(ctx->buf, ctx->len, ctx->qp_index, 0);
	if (ret == 0) {
		SM_DEBUG_LOG("QP[%d] state machine RDMA send succeeded (len=%lu)\n", ctx->qp_index, ctx->len);
	} else if (ret == -2) {
		SM_DEBUG_LOG("QP[%d] state machine send failed: QP is full\n", ctx->qp_index);
	} else {
		SM_DEBUG_LOG("QP[%d] state machine send failed: %d\n", ctx->qp_index, ret);
	}
}

/**
 * @brief Create a state machine for a single QP from the global config
 */
static int bind_qp_state_machine_tasks(int qp_index)
{
	if (!g_initialized || !g_qp_infos) {
		printf("QP RDMA ops not initialized\n");
		return -1;
	}

	if (qp_index < 0 || qp_index >= (int)g_qp_count) {
		printf("Invalid QP index: %d\n", qp_index);
		return -1;
	}

	// The state machines and state counts built during parsing must exist
	if (!g_qp_state_machine_info.state_machines || !g_qp_state_machine_info.state_counts) {
		printf("No built state machines available for QP %d\n", qp_index);
		return -1;
	}
	int state_count =
	    (g_qp_state_machine_info.state_counts[qp_index] > 0) ? g_qp_state_machine_info.state_counts[qp_index] : 0;
	if (state_count == 0 || g_qp_state_machine_info.state_machines[qp_index] == NULL) {
		printf("QP %d has no built state machine\n", qp_index);
		return -1;
	}

	struct rdma_op_qp_info *qp_info = &g_qp_infos[qp_index];

	// If a state machine already exists, destroy it first
	if (qp_info->state_machine) {
		destroy_state_machine(qp_info->state_machine);
	}

	// Point to the state machine built during parsing
	qp_info->state_machine = g_qp_state_machine_info.state_machines[qp_index];
	int ret = 0;

	// Create a task for each state and add it to the state machine
	for (int i = 0; i < state_count; i++) {
		// Create the RDMA send task context
		qp_rdma_task_context_t *task_ctx = malloc(sizeof(qp_rdma_task_context_t));
		if (!task_ctx) {
			printf("Failed to allocate task context for QP %d state %d\n", qp_index, i);
			return -1;
		}

		task_ctx->qp_index = qp_index;
		task_ctx->buf = rdma_ctx.buf;  // Use the global buffer
		uint32_t rdma_len = 0;
		if (g_qp_state_machine_info.rdma_len && g_qp_state_machine_info.rdma_len[qp_index]) {
			rdma_len = g_qp_state_machine_info.rdma_len[qp_index][i];
		}
		task_ctx->len = (rdma_len > 0) ? rdma_len : RDMA_BUF_SIZE;
		task_ctx->remote_addr = 0;

		// Create the task
		task_t rdma_task = create_task(qp_rdma_send_task, task_ctx, sizeof(qp_rdma_task_context_t));

		// Bind the task to an existing state at runtime
		ret = set_state_task(qp_info->state_machine, (uint32_t)i, &rdma_task);
		if (ret < 0) {
			printf("Failed to set task for state %d on QP %d\n", i, qp_index);
			free(task_ctx);
			return -1;
		}

		printf("QP[%d] State[%d]: len=%u\n", qp_index, i, task_ctx->len);
	}

	qp_info->state_machine_enabled = false;  // Disabled by default; must be started manually

	printf("QP[%d] state machine task binding complete (%d states)\n", qp_index, state_count);
	return 0;
}

/**
 * @brief Initialize the state machines of all QPs from the global config
 */
int init_qp_state_machines(void)
{
	if (!g_initialized || !g_qp_infos) {
		printf("QP RDMA ops not initialized\n");
		return -1;
	}

	if (!g_qp_state_machine_info.state_machines || g_qp_state_machine_info.qp_count == 0) {
		printf("No built QP state machines found\n");
		return -1;
	}

	if (g_qp_state_machine_info.qp_count != (int)g_qp_count) {
		printf("Config count mismatch: %d configs vs %u QPs\n", g_qp_state_machine_info.qp_count, g_qp_count);
		return -1;
	}

	printf("Constructing and wiring up QP tasks from the built state machines...\n");

	for (uint32_t i = 0; i < g_qp_count; i++) {
		if (g_qp_state_machine_info.state_machines[i] && g_qp_state_machine_info.state_counts[i] > 0) {
			int ret = bind_qp_state_machine_tasks(i);
			if (ret < 0) {
				printf("Failed to setup state machine for QP %u\n", i);
				return -1;
			}
		} else {
			// For an unconfigured QP, state_machine stays NULL and will be skipped
			printf("QP[%u] has no configured state machine and will be skipped\n", i);
			g_qp_infos[i].state_machine = NULL;
			g_qp_infos[i].state_machine_enabled = false;
		}
	}

	printf("QP state machine task binding complete\n");

	// Free the build-artifact helper arrays (the state machines themselves are owned by the rdma_op lifecycle)
	cleanup_qp_state_machine_build_artifacts();

	return 0;
}

/**
 * @brief Start the state machines of all QPs
 */
static int start_all_qp_state_machines(void)
{
	if (!g_initialized || !g_qp_infos) {
		SM_DEBUG_LOG("QP RDMA ops not initialized\n");
		return -1;
	}

	SM_DEBUG_LOG("Starting all QP state machines...\n");

	int started_count = 0;
	for (uint32_t i = 0; i < g_qp_count; i++) {
		// Skip unconfigured QPs
		if (!g_qp_infos[i].state_machine) {
			continue;
		}

		int ret = start_state_machine(g_qp_infos[i].state_machine);
		if (ret < 0) {
			printf("Failed to start state machine for QP %u\n", i);
			return -1;
		}
		g_qp_infos[i].state_machine_enabled = true;
		started_count++;
	}

	SM_DEBUG_LOG("All QP state machines started\n");
	return 0;
}

/**
 * @brief Stop the state machines of all QPs
 */
static void stop_all_qp_state_machines(void)
{
	if (!g_initialized || !g_qp_infos) {
		return;
	}

	printf("Stopping all QP state machines...\n");

	int stopped_count = 0;
	for (uint32_t i = 0; i < g_qp_count; i++) {
		// Skip unconfigured QPs
		if (!g_qp_infos[i].state_machine) {
			continue;
		}

		if (g_qp_infos[i].state_machine_enabled) {
			stop_state_machine(g_qp_infos[i].state_machine);
			g_qp_infos[i].state_machine_enabled = false;
			stopped_count++;
		}
	}

	printf("Stopped %d QP state machines\n", stopped_count);
}

/**
 * @brief Main loop for state-machine-controlled RDMA writes
 */
int state_machine_controlled_rdma_write(void)
{
	int ret;
	uint64_t cur_tsc = rte_get_tsc_cycles();
	const uint64_t output_interval = rte_get_tsc_hz();  // Output status once per second

	// 1. Initialize QP RDMA operation management
	SM_DEBUG_LOG("Initializing QP RDMA operation management...\n");
	ret = init_qp_rdma_ops(DEFAULT_QP_MAX_DEPTH);
	if (ret < 0) {
		printf("Failed to initialize QP RDMA ops\n");
		return ret;
	}

	// 2. Initialize the RDMA send timestamp statistics
	SM_DEBUG_LOG("Initializing RDMA send timestamp stats...\n");
	ret = init_rdma_op_stats();
	if (ret < 0) {
		printf("Failed to initialize RDMA send stats\n");
		cleanup_qp_rdma_ops();
		return ret;
	}

	// 2b. Initialize the RDMA completion latency statistics
	ret = init_rdma_completion_stats();
	if (ret < 0) {
		printf("Failed to initialize RDMA completion latency stats\n");
		cleanup_rdma_send_stats();
		cleanup_qp_rdma_ops();
		return ret;
	} else {
		printf("DEBUG: RDMA completion latency stats initialized\n");
	}

	// 3. Start the RDMA send timestamp statistics lcore
	ret = start_rdma_send_stats_lcore();
	if (ret < 0) {
		printf("Failed to start RDMA send stats lcore\n");
		cleanup_rdma_completion_stats();
		cleanup_rdma_send_stats();
		cleanup_qp_rdma_ops();
		return ret;
	}

	// 3b. Start the RDMA completion latency statistics lcore
	ret = start_rdma_completion_stats_lcore();
	if (ret < 0) {
		printf("Failed to start RDMA completion latency stats lcore\n");
		cleanup_rdma_completion_stats();
		cleanup_rdma_send_stats();
		cleanup_qp_rdma_ops();
		return ret;
	}

	// 4. Initialize the state machines of all QPs from the global config
	ret = init_qp_state_machines();
	if (ret < 0) {
		printf("Failed to initialize QP state machines from config\n");
		cleanup_rdma_completion_stats();
		cleanup_rdma_send_stats();
		cleanup_qp_rdma_ops();
		return ret;
	}

	// 5. Start all state machines
	ret = start_all_qp_state_machines();
	if (ret < 0) {
		printf("Failed to start QP state machines\n");
		cleanup_rdma_completion_stats();
		cleanup_rdma_send_stats();
		cleanup_qp_rdma_ops();
		return ret;
	}

	// Print the initial status
	print_qp_depth_status();

	// 6. Main loop: update state machines and poll the CQ
	printf("Starting state-machine-controlled RDMA write operations...\n");
	uint32_t loop_count = 0;
	uint32_t total_state_transitions = 0;

	while ((send_times == 0 || loop_count < send_times) && !force_quit) {
		bool any_active = false;

		// Update the state machines of all QPs (skip unconfigured QPs)
		for (uint32_t i = 0; i < g_qp_count; i++) {
			// Skip QPs without a state machine (unconfigured)
			if (!g_qp_infos[i].state_machine) {
				continue;
			}

			if (g_qp_infos[i].state_machine_enabled) {
				int state_changed = update_state_machine(g_qp_infos[i].state_machine, i, 0);
				if (state_changed == 1) {
					total_state_transitions++;
				}
				any_active = true;
			}
		}

		// If there are no active state machines, exit the loop
		if (!any_active) {
			printf("All state machines have completed; exiting the main loop\n");
			break;
		}

		// Periodically poll the CQ to process completion events
		int completed = managed_poll_cq(32, 0);  // Non-blocking poll

		loop_count++;

		// Print the status once per second
		if (rte_get_tsc_cycles() - cur_tsc > output_interval) {
			cur_tsc = rte_get_tsc_cycles();
			printf("Loop progress: %u, state transitions: %u, completion events: %d\n", loop_count,
			       total_state_transitions, completed);
			print_qp_depth_status();
		}
	}

	// 7. Stop all state machines
	stop_all_qp_state_machines();

	// 8. Wait for all operations to complete
	printf("Waiting for all RDMA operations to complete...\n");
	int max_wait_cycles = 1000;
	int wait_cycles = 0;

	while (wait_cycles < max_wait_cycles && !force_quit) {
		int completed = managed_poll_cq(64, 10);  // Wait up to 10ms
		if (completed == 0) {
			// Check whether there are still pending WRs
			int empty_qps = get_empty_qp_count();
			if (empty_qps == (int)g_qp_count) {
				break;  // All QPs are empty, meaning there are no more pending WRs
			}
		}
		wait_cycles++;
		if (wait_cycles % 100 == 0) {
			printf("Still waiting... cycle %d\n", wait_cycles);
			print_qp_depth_status();
		}
	}

	// 9. Print the final statistics
	printf("\nFinal Statistics:\n");
	printf("Loop count: %u\n", loop_count);
	printf("State transitions: %u\n", total_state_transitions);
	print_qp_depth_status();
	print_rdma_completion_stats_summary();
	print_rdma_send_stats_summary();

	// 10. Free resources
	cleanup_rdma_completion_stats();
	cleanup_rdma_send_stats();
	cleanup_qp_rdma_ops();

	return 0;
}

/**
 * Example function that uses the new managed RDMA operations.
 * Can be called from the sender_test function in exp_sender.c.
 */
int managed_rdma_write(void)
{
	int ret;

	// 1. Initialize QP RDMA operation management (after the RDMA connection is established)
	printf("Initializing QP RDMA ops...\n");
	ret = init_qp_rdma_ops(DEFAULT_QP_MAX_DEPTH);
	if (ret < 0) {
		printf("Failed to initialize QP RDMA ops\n");
		return ret;
	}

	// Initialize the RDMA send timestamp statistics
	printf("Initializing RDMA send timestamp stats...\n");
	ret = init_rdma_op_stats();
	if (ret < 0) {
		printf("Failed to initialize RDMA send stats\n");
		cleanup_qp_rdma_ops();
		return ret;
	}

	// Initialize the RDMA completion latency statistics
	ret = init_rdma_completion_stats();
	if (ret < 0) {
		printf("Failed to initialize RDMA completion latency stats\n");
		cleanup_rdma_send_stats();
		cleanup_qp_rdma_ops();
		return ret;
	}

	// Start the RDMA send timestamp statistics lcore
	ret = start_rdma_send_stats_lcore();
	if (ret < 0) {
		printf("Failed to start RDMA send stats lcore\n");
		cleanup_rdma_completion_stats();
		cleanup_rdma_send_stats();
		cleanup_qp_rdma_ops();
		return ret;
	}

	// Start the RDMA completion latency statistics lcore
	ret = start_rdma_completion_stats_lcore();
	if (ret < 0) {
		printf("Failed to start RDMA completion latency stats lcore\n");
		cleanup_rdma_completion_stats();
		cleanup_rdma_send_stats();
		cleanup_qp_rdma_ops();
		return ret;
	}

	// Print the initial status
	print_qp_depth_status();

	// 2. Main loop for sending data
	printf("Starting managed RDMA write operations...\n");
	uint32_t total_posts = 0;

	while ((send_times == 0 || (total_posts < send_times)) && !force_quit) {
		// Automatically select a QP, non-blocking mode
		ret = post_write_auto(rdma_ctx.buf, RDMA_BUF_SIZE, 0);
		if (ret == 0) {
			total_posts++;
			if (total_posts % 100 == 0) {
				printf("Progress: %u/%u posts\n", total_posts, send_times);
				print_qp_depth_status();
			}
		}

		// Periodically poll the CQ to free completed WRs
		int completed = managed_poll_cq(32, 0);  // Non-blocking poll
	}

	// 3. Wait for all operations to complete
	printf("Waiting for all operations to complete...\n");
	int max_wait_cycles = 1000;  // Wait at most 1000 cycles
	int wait_cycles = 0;

	while (wait_cycles < max_wait_cycles && !force_quit) {
		int completed = managed_poll_cq(64, 10);  // Wait up to 10ms
		if (completed == 0) {
			// Check whether there are still pending WRs
			int empty_qps = get_empty_qp_count();
			if (empty_qps == (int)g_qp_count) {
				break;  // All QPs are empty, meaning there are no more pending WRs
			}
		}
		wait_cycles++;
		if (wait_cycles % 100 == 0) {
			printf("Still waiting... cycle %d\n", wait_cycles);
			print_qp_depth_status();
		}
	}

	// 4. Print the final statistics
	printf("\nFinal Statistics:\n");
	printf("Successful posts: %u\n", total_posts);
	print_qp_depth_status();
	print_rdma_completion_stats_summary();
	print_rdma_send_stats_summary();

	// 5. Free resources
	cleanup_rdma_completion_stats();
	cleanup_rdma_send_stats();
	cleanup_qp_rdma_ops();

	return 0;
}