From f945a7040e0a93c32b8c382dbd5b23cf675de2fa Mon Sep 17 00:00:00 2001 From: Timo Dritschler Date: Mon, 1 Dec 2014 19:21:35 +0100 Subject: KIRO Client now has a communication event handler for receives --- src/kiro-client.c | 103 +++++++++++++++++++++++++++++++++++++++--------------- 1 file changed, 74 insertions(+), 29 deletions(-) (limited to 'src/kiro-client.c') diff --git a/src/kiro-client.c b/src/kiro-client.c index 714a003..2b042c0 100644 --- a/src/kiro-client.c +++ b/src/kiro-client.c @@ -56,7 +56,8 @@ struct _KiroClientPrivate { gboolean close_signal; // Flag used to signal event listening to stop for connection tear-down GMainLoop *main_loop; // Main loop of the server for event polling and handling - GIOChannel *g_io_ec; // GLib IO Channel encapsulation for the connection manager event channel + GIOChannel *conn_ec; // GLib IO Channel encapsulation for the connection manager event channel + GIOChannel *rdma_ec; // GLib IO Channel encapsulation for the connection manager event channel GThread *main_thread; // Main KIRO client thread }; @@ -147,6 +148,59 @@ process_cm_event (GIOChannel *source, GIOCondition condition, gpointer data) } + +static gboolean +process_rdma_event (GIOChannel *source, GIOCondition condition, gpointer data) +{ + // Right now, we don't need 'source' and 'condition' + // Tell the compiler to ignore them by (void)-ing them + (void) source; + (void) condition; + + KiroClientPrivate *priv = (KiroClientPrivate *)data; + struct ibv_wc wc; + + if (rdma_get_recv_comp (priv->conn, &wc) < 0) { + g_critical ("Failure waiting for POST from server: %s", strerror (errno)); + return FALSE; + } + + struct kiro_connection_context *ctx = (struct kiro_connection_context *)priv->conn->context; + guint type = ((struct kiro_ctrl_msg *)ctx->cf_mr_recv->mem)->msg_type; + g_debug ("Received a message from the Server of type: %u", type); + + if (type == KIRO_ACK_RDMA) { + g_debug ("Got RDMI Access information from Server"); + ctx->peer_mr = (((struct kiro_ctrl_msg *) (ctx->cf_mr_recv->mem))->peer_mri); + g_debug ("Expected Memory Size is: %zu", ctx->peer_mr.length); + ctx->rdma_mr = kiro_create_rdma_memory (priv->conn->pd, ctx->peer_mr.length, IBV_ACCESS_LOCAL_WRITE); + + if (!ctx->rdma_mr) { + //TODO: Connection teardown in an event handler routine? Not a good + //idea... + g_critical ("Failed to allocate memory for receive buffer (Out of memory?)"); + rdma_disconnect (priv->conn); + kiro_destroy_connection_context (&ctx); + rdma_destroy_ep (priv->conn); + return FALSE; + } + } + + //Post a generic receive in order to stay responsive to any messages from + //the server + if (rdma_post_recv (priv->conn, priv->conn, ctx->cf_mr_recv->mem, ctx->cf_mr_recv->size, ctx->cf_mr_recv->mr)) { + //TODO: Connection teardown in an event handler routine? Not a good + //idea... + g_critical ("Posting generic receive for connection failed: %s", strerror (errno)); + kiro_destroy_connection_context (&ctx); + rdma_destroy_ep (priv->conn); + return FALSE; + } + + return TRUE; +} + + gpointer start_client_main_loop (gpointer data) { @@ -195,6 +249,11 @@ kiro_client_connect (KiroClient *self, const char *address, const char *port) g_critical ("Endpoint creation failed: %s", strerror (errno)); return -1; } + // make sure the receive queue pushes an event onto its completion channel + // This is needed in order to see any events on the recv_cq on its + // respective completion_channel file descriptor. We will use this + // mechanismn in our main loop to poll for those events. + ibv_req_notify_cq (priv->conn->recv_cq, 0); g_debug ("Route to server resolved"); struct kiro_connection_context *ctx = (struct kiro_connection_context *)g_try_malloc (sizeof (struct kiro_connection_context)); @@ -218,6 +277,7 @@ kiro_client_connect (KiroClient *self, const char *address, const char *port) ctx->cf_mr_recv->size = ctx->cf_mr_send->size = sizeof (struct kiro_ctrl_msg); priv->conn->context = ctx; + //Post an preemtive receive for the servers welcome message if (rdma_post_recv (priv->conn, priv->conn, ctx->cf_mr_recv->mem, ctx->cf_mr_recv->size, ctx->cf_mr_recv->mr)) { g_critical ("Posting preemtive receive for connection failed: %s", strerror (errno)); kiro_destroy_connection_context (&ctx); @@ -232,40 +292,21 @@ kiro_client_connect (KiroClient *self, const char *address, const char *port) return -1; } - g_message ("Connection to server established"); priv->ec = priv->conn->channel; //For easy access - struct ibv_wc wc; - - if (rdma_get_recv_comp (priv->conn, &wc) < 0) { - g_critical ("Failure waiting for POST from server: %s", strerror (errno)); - rdma_disconnect (priv->conn); - kiro_destroy_connection_context (&ctx); - rdma_destroy_ep (priv->conn); - return -1; - } - - g_debug ("Got RDMI Access information from Server"); - ctx->peer_mr = (((struct kiro_ctrl_msg *) (ctx->cf_mr_recv->mem))->peer_mri); - g_debug ("Expected Memory Size is: %zu", ctx->peer_mr.length); - ctx->rdma_mr = kiro_create_rdma_memory (priv->conn->pd, ctx->peer_mr.length, IBV_ACCESS_LOCAL_WRITE); - - if (!ctx->rdma_mr) { - g_critical ("Failed to allocate memory for receive buffer (Out of memory?)"); - rdma_disconnect (priv->conn); - kiro_destroy_connection_context (&ctx); - rdma_destroy_ep (priv->conn); - return -1; - } priv->main_loop = g_main_loop_new (NULL, FALSE); - priv->g_io_ec = g_io_channel_unix_new (priv->ec->fd); - g_io_add_watch (priv->g_io_ec, G_IO_IN | G_IO_PRI | G_IO_ERR | G_IO_HUP, process_cm_event, (gpointer)priv); + priv->conn_ec = g_io_channel_unix_new (priv->ec->fd); + priv->rdma_ec = g_io_channel_unix_new (priv->conn->recv_cq_channel->fd); + g_io_add_watch (priv->conn_ec, G_IO_IN | G_IO_PRI | G_IO_ERR | G_IO_HUP, process_cm_event, (gpointer)priv); + g_io_add_watch (priv->rdma_ec, G_IO_IN | G_IO_PRI | G_IO_ERR | G_IO_HUP, process_rdma_event, (gpointer)priv); priv->main_thread = g_thread_new ("KIRO Client main loop", start_client_main_loop, priv->main_loop); // We gave control to the main_loop (with add_watch) and don't need our ref // any longer - g_io_channel_unref (priv->g_io_ec); + g_io_channel_unref (priv->conn_ec); + g_io_channel_unref (priv->rdma_ec); + g_message ("Connection to server established"); g_message ("Connected to %s:%s", address, port); return 0; } @@ -375,8 +416,12 @@ kiro_client_disconnect (KiroClient *self) // We don't need the connection management IO channel container any more. // Unref and thus free it. - g_io_channel_unref (priv->g_io_ec); - priv->g_io_ec = NULL; + g_io_channel_unref (priv->conn_ec); + priv->conn_ec = NULL; + + // The same goes for the cp channels + g_io_channel_unref (priv->rdma_ec); + priv->rdma_ec = NULL; priv->close_signal = FALSE; -- cgit v1.2.3 From 8579e596df0bebee274dcadf766ae425bad9b1e8 Mon Sep 17 00:00:00 2001 From: Timo Dritschler Date: Wed, 3 Dec 2014 14:52:24 +0100 Subject: Fixed a race condition in kiro_client_connect --- src/kiro-client.c | 34 +++++++++++++++++++--------------- 1 file changed, 19 insertions(+), 15 deletions(-) (limited to 'src/kiro-client.c') diff --git a/src/kiro-client.c b/src/kiro-client.c index 2b042c0..0bb95fa 100644 --- a/src/kiro-client.c +++ b/src/kiro-client.c @@ -57,7 +57,7 @@ struct _KiroClientPrivate { gboolean close_signal; // Flag used to signal event listening to stop for connection tear-down GMainLoop *main_loop; // Main loop of the server for event polling and handling GIOChannel *conn_ec; // GLib IO Channel encapsulation for the connection manager event channel - GIOChannel *rdma_ec; // GLib IO Channel encapsulation for the connection manager event channel + GIOChannel *rdma_ec; // GLib IO Channel encapsulation for the communication event channel GThread *main_thread; // Main KIRO client thread }; @@ -90,7 +90,6 @@ kiro_client_init (KiroClient *self) { KiroClientPrivate *priv = KIRO_CLIENT_GET_PRIVATE (self); memset (priv, 0, sizeof (&priv)); - //Hack to make the 'unused function' from the kiro-rdma include go away... kiro_attach_qp (NULL); } @@ -164,7 +163,7 @@ process_rdma_event (GIOChannel *source, GIOCondition condition, gpointer data) g_critical ("Failure waiting for POST from server: %s", strerror (errno)); return FALSE; } - + struct kiro_connection_context *ctx = (struct kiro_connection_context *)priv->conn->context; guint type = ((struct kiro_ctrl_msg *)ctx->cf_mr_recv->mem)->msg_type; g_debug ("Received a message from the Server of type: %u", type); @@ -269,9 +268,7 @@ kiro_client_connect (KiroClient *self, const char *address, const char *port) if (!ctx->cf_mr_recv || !ctx->cf_mr_send) { g_critical ("Failed to register control message memory (Out of memory?)"); - kiro_destroy_connection_context (&ctx); - rdma_destroy_ep (priv->conn); - return -1; + goto fail; } ctx->cf_mr_recv->size = ctx->cf_mr_send->size = sizeof (struct kiro_ctrl_msg); @@ -280,20 +277,23 @@ kiro_client_connect (KiroClient *self, const char *address, const char *port) //Post an preemtive receive for the servers welcome message if (rdma_post_recv (priv->conn, priv->conn, ctx->cf_mr_recv->mem, ctx->cf_mr_recv->size, ctx->cf_mr_recv->mr)) { g_critical ("Posting preemtive receive for connection failed: %s", strerror (errno)); - kiro_destroy_connection_context (&ctx); - rdma_destroy_ep (priv->conn); - return -1; + goto fail; } if (rdma_connect (priv->conn, NULL)) { g_critical ("Failed to establish connection to the server: %s", strerror (errno)); - kiro_destroy_connection_context (&ctx); - rdma_destroy_ep (priv->conn); - return -1; + goto fail; } - priv->ec = priv->conn->channel; //For easy access + g_message ("Connection to server established. Waiting for response."); + if (!process_rdma_event (NULL, 0, (gpointer)priv)) { + g_critical ("No RDMA access information received from the server. Failed to connect."); + goto fail; + } + + g_message ("Connected to %s:%s", address, port); + priv->ec = priv->conn->channel; //For easy access priv->main_loop = g_main_loop_new (NULL, FALSE); priv->conn_ec = g_io_channel_unix_new (priv->ec->fd); priv->rdma_ec = g_io_channel_unix_new (priv->conn->recv_cq_channel->fd); @@ -306,9 +306,13 @@ kiro_client_connect (KiroClient *self, const char *address, const char *port) g_io_channel_unref (priv->conn_ec); g_io_channel_unref (priv->rdma_ec); - g_message ("Connection to server established"); - g_message ("Connected to %s:%s", address, port); return 0; + +fail: + kiro_destroy_connection_context (&ctx); + rdma_destroy_ep (priv->conn); + priv->conn = NULL; + return -1; } -- cgit v1.2.3 From cc2059982024af79136b9420eaec6fcfedabf3fb Mon Sep 17 00:00:00 2001 From: Timo Dritschler Date: Wed, 3 Dec 2014 19:29:03 +0100 Subject: KIRO Server now has a message event handler for receives --- src/kiro-client.c | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) (limited to 'src/kiro-client.c') diff --git a/src/kiro-client.c b/src/kiro-client.c index 0bb95fa..bb2645c 100644 --- a/src/kiro-client.c +++ b/src/kiro-client.c @@ -154,7 +154,8 @@ process_rdma_event (GIOChannel *source, GIOCondition condition, gpointer data) // Right now, we don't need 'source' and 'condition' // Tell the compiler to ignore them by (void)-ing them (void) source; - (void) condition; + //(void) condition; + g_debug ("Message condidition: %i", condition); KiroClientPrivate *priv = (KiroClientPrivate *)data; struct ibv_wc wc; @@ -297,8 +298,8 @@ kiro_client_connect (KiroClient *self, const char *address, const char *port) priv->main_loop = g_main_loop_new (NULL, FALSE); priv->conn_ec = g_io_channel_unix_new (priv->ec->fd); priv->rdma_ec = g_io_channel_unix_new (priv->conn->recv_cq_channel->fd); - g_io_add_watch (priv->conn_ec, G_IO_IN | G_IO_PRI | G_IO_ERR | G_IO_HUP, process_cm_event, (gpointer)priv); - g_io_add_watch (priv->rdma_ec, G_IO_IN | G_IO_PRI | G_IO_ERR | G_IO_HUP, process_rdma_event, (gpointer)priv); + g_io_add_watch (priv->conn_ec, G_IO_IN | G_IO_PRI, process_cm_event, (gpointer)priv); + g_io_add_watch (priv->rdma_ec, G_IO_IN | G_IO_PRI, process_rdma_event, (gpointer)priv); priv->main_thread = g_thread_new ("KIRO Client main loop", start_client_main_loop, priv->main_loop); // We gave control to the main_loop (with add_watch) and don't need our ref -- cgit v1.2.3 From b71e3e75e13a40838c1b386fdeff3afd2f453e9f Mon Sep 17 00:00:00 2001 From: Timo Dritschler Date: Thu, 4 Dec 2014 18:33:54 +0100 Subject: Fixed a race condition concerning message handling in kiro client --- src/kiro-client.c | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) (limited to 'src/kiro-client.c') diff --git a/src/kiro-client.c b/src/kiro-client.c index bb2645c..084e4b8 100644 --- a/src/kiro-client.c +++ b/src/kiro-client.c @@ -197,6 +197,11 @@ process_rdma_event (GIOChannel *source, GIOCondition condition, gpointer data) return FALSE; } + // make sure the next incoming work completion causes an event on the + // receive completion channel. We will poll() the channels file descriptor + // for this in the kiro client main loop. + ibv_req_notify_cq (priv->conn->recv_cq, 0); + return TRUE; } @@ -249,11 +254,6 @@ kiro_client_connect (KiroClient *self, const char *address, const char *port) g_critical ("Endpoint creation failed: %s", strerror (errno)); return -1; } - // make sure the receive queue pushes an event onto its completion channel - // This is needed in order to see any events on the recv_cq on its - // respective completion_channel file descriptor. We will use this - // mechanismn in our main loop to poll for those events. - ibv_req_notify_cq (priv->conn->recv_cq, 0); g_debug ("Route to server resolved"); struct kiro_connection_context *ctx = (struct kiro_connection_context *)g_try_malloc (sizeof (struct kiro_connection_context)); -- cgit v1.2.3 From 50a297a290c78c7feb0a4918efba82edd019590b Mon Sep 17 00:00:00 2001 From: Timo Dritschler Date: Wed, 10 Dec 2014 15:27:32 +0100 Subject: Fixed KIRO client getting stuck in the RDMA event handler Fix #8: KIRO Server and Client now have routines to handle async communication Fix #6: Added kiro_client_ping_server and respective event handling to server Changed kiro-test-latency to use the new kiro_client_ping_server instead --- src/kiro-client.c | 152 +++++++++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 145 insertions(+), 7 deletions(-) (limited to 'src/kiro-client.c') diff --git a/src/kiro-client.c b/src/kiro-client.c index 084e4b8..6e140b5 100644 --- a/src/kiro-client.c +++ b/src/kiro-client.c @@ -65,6 +65,11 @@ struct _KiroClientPrivate { G_DEFINE_TYPE (KiroClient, kiro_client, G_TYPE_OBJECT); +// Temporary storage and lock for PING timing +G_LOCK_DEFINE (ping_time); +volatile struct timeval ping_time; + + KiroClient * kiro_client_new (void) { @@ -92,6 +97,8 @@ kiro_client_init (KiroClient *self) memset (priv, 0, sizeof (&priv)); //Hack to make the 'unused function' from the kiro-rdma include go away... kiro_attach_qp (NULL); + ping_time.tv_sec = -1; + ping_time.tv_usec = -1; } @@ -160,10 +167,15 @@ process_rdma_event (GIOChannel *source, GIOCondition condition, gpointer data) KiroClientPrivate *priv = (KiroClientPrivate *)data; struct ibv_wc wc; - if (rdma_get_recv_comp (priv->conn, &wc) < 0) { - g_critical ("Failure waiting for POST from server: %s", strerror (errno)); - return FALSE; + if (ibv_poll_cq (priv->conn->recv_cq, 1, &wc) < 0) { + g_critical ("Failure getting receive completion event from the queue: %s", strerror (errno)); + return FALSE; } + void *cq_ctx; + struct ibv_cq *cq; + int err = ibv_get_cq_event (priv->conn->recv_cq_channel, &cq, &cq_ctx); + if (!err) + ibv_ack_cq_events (cq, 1); struct kiro_connection_context *ctx = (struct kiro_connection_context *)priv->conn->context; guint type = ((struct kiro_ctrl_msg *)ctx->cf_mr_recv->mem)->msg_type; @@ -185,6 +197,22 @@ process_rdma_event (GIOChannel *source, GIOCondition condition, gpointer data) return FALSE; } } + if (type == KIRO_PONG) { + G_LOCK (ping_time); + struct timeval local_time; + gettimeofday (&local_time, NULL); + + if (ping_time.tv_sec == 0 && ping_time.tv_usec == 0) { + g_debug ("Received PONG message from server"); + ping_time.tv_sec = local_time.tv_sec; + ping_time.tv_usec = local_time.tv_usec; + } + else { + g_debug ("Received unexpected PONG message from server"); + } + + G_UNLOCK (ping_time); + } //Post a generic receive in order to stay responsive to any messages from //the server @@ -202,6 +230,7 @@ process_rdma_event (GIOChannel *source, GIOCondition condition, gpointer data) // for this in the kiro client main loop. ibv_req_notify_cq (priv->conn->recv_cq, 0); + g_debug ("Finished RDMA event handling"); return TRUE; } @@ -287,6 +316,7 @@ kiro_client_connect (KiroClient *self, const char *address, const char *port) } g_message ("Connection to server established. Waiting for response."); + ibv_req_notify_cq (priv->conn->recv_cq, 0); // Make the respective Queue push events onto the channel if (!process_rdma_event (NULL, 0, (gpointer)priv)) { g_critical ("No RDMA access information received from the server. Failed to connect."); goto fail; @@ -355,11 +385,119 @@ kiro_client_sync (KiroClient *self) } fail: - kiro_destroy_connection (&(priv->conn)); + kiro_destroy_connection (&(priv->conn)); return -1; } +gboolean +ping_timeout (gpointer data) { + + //Not needed. Void it to prevent 'unused variable' warning + (void) data; + + G_LOCK (ping_time); + + // Maybe the server did answer while dispatching the timeout? + if (ping_time.tv_sec != 0 || ping_time.tv_usec != 0) { + goto done; + } + + ping_time.tv_usec = -1; + ping_time.tv_sec = -1; + + +done: + G_UNLOCK (ping_time); + + // Return FALSE to automtically stop the timeout from reoccuring + return FALSE; +} + + +gint +kiro_client_ping_server (KiroClient *self) +{ + // Will be returned. -1 for error. + gint t_usec = 0; + + KiroClientPrivate *priv = KIRO_CLIENT_GET_PRIVATE (self); + if (!priv->conn) { + g_warning ("Client not connected"); + return -1; + } + + struct kiro_connection_context *ctx = (struct kiro_connection_context *)priv->conn->context; + + struct kiro_ctrl_msg *msg = (struct kiro_ctrl_msg *)(ctx->cf_mr_send->mem); + msg->msg_type = KIRO_PING; + + G_LOCK (ping_time); + ping_time.tv_sec = 0; + ping_time.tv_usec = 0; + struct timeval local_time; + gettimeofday (&local_time, NULL); + + if (rdma_post_send (priv->conn, priv->conn, ctx->cf_mr_send->mem, ctx->cf_mr_send->size, ctx->cf_mr_send->mr, IBV_SEND_SIGNALED)) { + g_warning ("Failure while trying to post SEND for PING: %s", strerror (errno)); + t_usec = -1; + goto end; + } + G_UNLOCK (ping_time); + + struct ibv_wc wc; + if (rdma_get_send_comp (priv->conn, &wc) < 0) { + g_warning ("Failure during PING send: %s", strerror (errno)); + t_usec = -1; + goto end; + } + + // Set a two-second timeout for the ping + guint timeout = g_timeout_add_seconds (2, ping_timeout, NULL); + + //Wait for ping response + while (ping_time.tv_sec == 0 && ping_time.tv_usec == 0) {}; + + + G_LOCK (ping_time); + // No response from the server. Timeout kicked in + // (Note: The timeout callback has already deregistered itself. We don't + // need to do that here again) + if (ping_time.tv_sec == -1 && ping_time.tv_usec == -1) { + g_message ("PING timed out."); + G_UNLOCK (ping_time); + t_usec = -1; + goto end; + } + + // Remove the timeout + GSource *timeout_source = g_main_context_find_source_by_id (NULL, timeout); + if (timeout_source) { + g_source_destroy (timeout_source); + } + + gint secs = ping_time.tv_sec - local_time.tv_sec; + + // tv_usecs wraps back to 0 at 1000000us (1s). + // This might cause our calculation to produce negative numbers when time > 1s. + for (int i = 0; i < secs; i++) { + ping_time.tv_usec += 1000 * 1000; + } + t_usec = ping_time.tv_usec - local_time.tv_usec; + gint millis = (gint)(t_usec/1000.); + G_UNLOCK (ping_time); + + g_debug ("Server responded to PING in: %is, %ims, %ius", secs, millis, t_usec); + +end: + G_LOCK (ping_time); + ping_time.tv_sec = -1; + ping_time.tv_usec = -1; + G_UNLOCK (ping_time); + return t_usec; +} + + void * kiro_client_get_memory (KiroClient *self) { @@ -377,7 +515,7 @@ kiro_client_get_memory (KiroClient *self) } -size_t +size_t kiro_client_get_memory_size (KiroClient *self) { KiroClientPrivate *priv = KIRO_CLIENT_GET_PRIVATE (self); @@ -431,13 +569,13 @@ kiro_client_disconnect (KiroClient *self) priv->close_signal = FALSE; //kiro_destroy_connection does not free RDMA memory. Therefore, we need to - //cache the memory pointer and free the memory afterwards manually + //cache the memory pointer and free the memory afterwards manually struct kiro_connection_context *ctx = (struct kiro_connection_context *) (priv->conn->context); void *rdma_mem = ctx->rdma_mr->mem; kiro_destroy_connection (&(priv->conn)); free (rdma_mem); - // priv->ec is just an easy-access pointer. Don't free it. Just NULL it + // priv->ec is just an easy-access pointer. Don't free it. Just NULL it priv->ec = NULL; g_message ("Client disconnected from server"); } -- cgit v1.2.3