From c5b6f4f08fe8d1abff74c7f6ad3630b7dcf0669d Mon Sep 17 00:00:00 2001 From: David Goulet Date: Wed, 12 Dec 2012 17:05:45 -0500 Subject: [PATCH] Add the relayd create session command This is needed in order to fix a specific condition of the data pending where we need to have streams associated with a session and this command will be used for new feature in the future. Acked-by: Mathieu Desnoyers Signed-off-by: David Goulet --- src/bin/lttng-relayd/lttng-relayd.h | 2 +- src/bin/lttng-relayd/main.c | 76 ++++++++++++++++++++--------- src/common/consumer.c | 11 +++++ src/common/consumer.h | 3 ++ src/common/relayd/relayd.c | 48 ++++++++++++++++++ src/common/relayd/relayd.h | 5 +- src/common/sessiond-comm/relayd.h | 8 +++ 7 files changed, 126 insertions(+), 27 deletions(-) diff --git a/src/bin/lttng-relayd/lttng-relayd.h b/src/bin/lttng-relayd/lttng-relayd.h index f231f0f6c..2a442b095 100644 --- a/src/bin/lttng-relayd/lttng-relayd.h +++ b/src/bin/lttng-relayd/lttng-relayd.h @@ -42,7 +42,6 @@ enum connection_type { struct relay_session { uint64_t id; struct lttcomm_sock *sock; - unsigned int version_check_done:1; }; /* @@ -72,6 +71,7 @@ struct relay_command { struct lttng_ht_node_ulong sock_n; struct rcu_head rcu_node; enum connection_type type; + unsigned int version_check_done:1; }; #endif /* LTTNG_RELAYD_H */ diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index 2a7152b89..74e785e2f 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -885,8 +885,6 @@ void relay_delete_session(struct relay_command *cmd, struct lttng_ht *streams_ht DBG("Relay deleting session %" PRIu64, cmd->session->id); - lttcomm_destroy_sock(cmd->session->sock); - rcu_read_lock(); cds_lfht_for_each_entry(streams_ht->ht, &iter.iter, node, node) { node = lttng_ht_iter_get_node_ulong(&iter); @@ -910,6 +908,54 @@ void relay_delete_session(struct relay_command *cmd, struct lttng_ht *streams_ht free(cmd->session); } +/* + * Handle the RELAYD_CREATE_SESSION command. + * + * On success, send back the session id or else return a negative value. + */ +static +int relay_create_session(struct lttcomm_relayd_hdr *recv_hdr, + struct relay_command *cmd) +{ + int ret = 0, send_ret; + struct relay_session *session; + struct lttcomm_relayd_status_session reply; + + assert(recv_hdr); + assert(cmd); + + memset(&reply, 0, sizeof(reply)); + + session = zmalloc(sizeof(struct relay_session)); + if (session == NULL) { + PERROR("relay session zmalloc"); + ret = -1; + goto error; + } + + session->id = ++last_relay_session_id; + session->sock = cmd->sock; + cmd->session = session; + + reply.session_id = htobe64(session->id); + + DBG("Created session %" PRIu64, session->id); + +error: + if (ret < 0) { + reply.ret_code = htobe32(LTTNG_ERR_FATAL); + } else { + reply.ret_code = htobe32(LTTNG_OK); + } + + send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0); + if (send_ret < 0) { + ERR("Relayd sending session id"); + } + + return ret; +} + /* * relay_add_stream: allocate a new stream for a session */ @@ -924,7 +970,7 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, char *path = NULL, *root_path = NULL; int ret, send_ret; - if (!session || session->version_check_done == 0) { + if (!session || cmd->version_check_done == 0) { ERR("Trying to add a stream before version check"); ret = -1; goto end_no_session; @@ -1021,7 +1067,7 @@ int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr, DBG("Close stream received"); - if (!session || session->version_check_done == 0) { + if (!session || cmd->version_check_done == 0) { ERR("Trying to close a stream before version check"); ret = -1; goto end_no_session; @@ -1281,22 +1327,10 @@ int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr, { int ret; struct lttcomm_relayd_version reply, msg; - struct relay_session *session; - if (cmd->session == NULL) { - session = zmalloc(sizeof(struct relay_session)); - if (session == NULL) { - PERROR("relay session zmalloc"); - ret = -1; - goto end; - } - session->id = ++last_relay_session_id; - DBG("Created session %" PRIu64, session->id); - cmd->session = session; - } else { - session = cmd->session; - } - session->version_check_done = 1; + assert(cmd); + + cmd->version_check_done = 1; /* Get version from the other side. */ ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0); @@ -1351,7 +1385,7 @@ int relay_data_pending(struct lttcomm_relayd_hdr *recv_hdr, DBG("Data pending command received"); - if (!session || session->version_check_done == 0) { + if (!session || cmd->version_check_done == 0) { ERR("Trying to check for data before version check"); ret = -1; goto end_no_session; @@ -1440,11 +1474,9 @@ int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr, int ret = 0; switch (be32toh(recv_hdr->cmd)) { - /* case RELAYD_CREATE_SESSION: ret = relay_create_session(recv_hdr, cmd); break; - */ case RELAYD_ADD_STREAM: ret = relay_add_stream(recv_hdr, cmd, streams_ht); break; diff --git a/src/common/consumer.c b/src/common/consumer.c index f57cfffff..f4cfa82ce 100644 --- a/src/common/consumer.c +++ b/src/common/consumer.c @@ -2743,6 +2743,17 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type, /* Assign new file descriptor */ relayd->control_sock.fd = fd; + + /* + * Create a session on the relayd and store the returned id. No need to + * grab the socket lock since the relayd object is not yet visible. + */ + ret = relayd_create_session(&relayd->control_sock, + &relayd->session_id); + if (ret < 0) { + goto error; + } + break; case LTTNG_STREAM_DATA: /* Copy received lttcomm socket */ diff --git a/src/common/consumer.h b/src/common/consumer.h index a476dd5bf..80bb46a83 100644 --- a/src/common/consumer.h +++ b/src/common/consumer.h @@ -186,6 +186,9 @@ struct consumer_relayd_sock_pair { */ struct lttcomm_sock data_sock; struct lttng_ht_node_ulong node; + + /* Session id on the relayd side for the sockets. */ + uint64_t session_id; }; /* diff --git a/src/common/relayd/relayd.c b/src/common/relayd/relayd.c index 9cc682713..56ca98223 100644 --- a/src/common/relayd/relayd.c +++ b/src/common/relayd/relayd.c @@ -99,6 +99,54 @@ error: return ret; } +/* + * Send a RELAYD_CREATE_SESSION command to the relayd with the given socket and + * set session_id of the relayd if we have a successful reply from the relayd. + * + * On success, return 0 else a negative value being a lttng_error_code returned + * from the relayd. + */ +int relayd_create_session(struct lttcomm_sock *sock, uint64_t *session_id) +{ + int ret; + struct lttcomm_relayd_status_session reply; + + assert(sock); + assert(session_id); + + DBG("Relayd create session"); + + /* Send command */ + ret = send_command(sock, RELAYD_CREATE_SESSION, NULL, 0, 0); + if (ret < 0) { + goto error; + } + + /* Recevie response */ + ret = recv_reply(sock, (void *) &reply, sizeof(reply)); + if (ret < 0) { + goto error; + } + + reply.session_id = be64toh(reply.session_id); + reply.ret_code = be32toh(reply.ret_code); + + /* Return session id or negative ret code. */ + if (reply.ret_code != LTTNG_OK) { + ret = -reply.ret_code; + ERR("Relayd create session replied error %d", ret); + goto error; + } else { + ret = 0; + *session_id = reply.session_id; + } + + DBG("Relayd session created with id %" PRIu64, reply.session_id); + +error: + return ret; +} + /* * Add stream on the relayd and assign stream handle to the stream_id argument. * diff --git a/src/common/relayd/relayd.h b/src/common/relayd/relayd.h index 8857bc9fb..9d65c3b8a 100644 --- a/src/common/relayd/relayd.h +++ b/src/common/relayd/relayd.h @@ -25,10 +25,7 @@ int relayd_connect(struct lttcomm_sock *sock); int relayd_close(struct lttcomm_sock *sock); -#if 0 -int relayd_create_session(struct lttcomm_sock *sock, const char *hostname, - const char *session_name); -#endif +int relayd_create_session(struct lttcomm_sock *sock, uint64_t *session_id); int relayd_add_stream(struct lttcomm_sock *sock, const char *channel_name, const char *pathname, uint64_t *stream_id); int relayd_send_close_stream(struct lttcomm_sock *sock, uint64_t stream_id, diff --git a/src/common/sessiond-comm/relayd.h b/src/common/sessiond-comm/relayd.h index 7bc7a1217..6cd9a21f4 100644 --- a/src/common/sessiond-comm/relayd.h +++ b/src/common/sessiond-comm/relayd.h @@ -53,6 +53,14 @@ struct lttcomm_relayd_data_hdr { uint32_t padding_size; /* Size of 0 padding the data */ } __attribute__ ((__packed__)); +/* + * Reply from a create session command. + */ +struct lttcomm_relayd_status_session { + uint64_t session_id; + uint32_t ret_code; +} __attribute__ ((__packed__)); + /* * Used to add a stream on the relay daemon. */ -- 2.34.1