X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Frelayd%2Frelayd.c;h=3de19c28064d352ef9e1d507f14bac3cd318a823;hp=bb20a64a0cc5d6cb8ffc04f2439c0c581f71803b;hb=53efb85a242809ed5ed21e9ab40effa696ecbc6f;hpb=1c20f0e29cbf8627bfb1ff444572d52d6655c4e2 diff --git a/src/common/relayd/relayd.c b/src/common/relayd/relayd.c index bb20a64a0..3de19c280 100644 --- a/src/common/relayd/relayd.c +++ b/src/common/relayd/relayd.c @@ -26,7 +26,7 @@ #include #include #include -#include +#include #include "relayd.h" @@ -57,6 +57,7 @@ static int send_command(struct lttcomm_relayd_sock *rsock, goto alloc_error; } + memset(&header, 0, sizeof(header)); header.cmd = htobe32(cmd); header.data_size = htobe64(size); @@ -116,6 +117,50 @@ error: return ret; } +/* + * Starting at 2.4, RELAYD_CREATE_SESSION takes additional parameters to + * support the live reading capability. + */ +static int relayd_create_session_2_4(struct lttcomm_relayd_sock *rsock, + uint64_t *session_id, char *session_name, char *hostname, + int session_live_timer, unsigned int snapshot) +{ + int ret; + struct lttcomm_relayd_create_session_2_4 msg; + + strncpy(msg.session_name, session_name, sizeof(msg.session_name)); + strncpy(msg.hostname, hostname, sizeof(msg.hostname)); + msg.live_timer = htobe32(session_live_timer); + msg.snapshot = htobe32(snapshot); + + /* Send command */ + ret = send_command(rsock, RELAYD_CREATE_SESSION, &msg, sizeof(msg), 0); + if (ret < 0) { + goto error; + } + +error: + return ret; +} + +/* + * RELAYD_CREATE_SESSION from 2.1 to 2.3. + */ +static int relayd_create_session_2_1(struct lttcomm_relayd_sock *rsock, + uint64_t *session_id) +{ + int ret; + + /* Send command */ + ret = send_command(rsock, RELAYD_CREATE_SESSION, NULL, 0, 0); + if (ret < 0) { + goto error; + } + +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. @@ -123,7 +168,9 @@ error: * On success, return 0 else a negative value which is either an errno error or * a lttng error code from the relayd. */ -int relayd_create_session(struct lttcomm_relayd_sock *rsock, uint64_t *session_id) +int relayd_create_session(struct lttcomm_relayd_sock *rsock, uint64_t *session_id, + char *session_name, char *hostname, int session_live_timer, + unsigned int snapshot) { int ret; struct lttcomm_relayd_status_session reply; @@ -133,8 +180,19 @@ int relayd_create_session(struct lttcomm_relayd_sock *rsock, uint64_t *session_i DBG("Relayd create session"); - /* Send command */ - ret = send_command(rsock, RELAYD_CREATE_SESSION, NULL, 0, 0); + switch(rsock->minor) { + case 1: + case 2: + case 3: + ret = relayd_create_session_2_1(rsock, session_id); + break; + case 4: + default: + ret = relayd_create_session_2_4(rsock, session_id, session_name, + hostname, session_live_timer, snapshot); + break; + } + if (ret < 0) { goto error; } @@ -187,6 +245,7 @@ int relayd_add_stream(struct lttcomm_relayd_sock *rsock, const char *channel_nam /* Compat with relayd 2.1 */ if (rsock->minor == 1) { + memset(&msg, 0, sizeof(msg)); strncpy(msg.channel_name, channel_name, sizeof(msg.channel_name)); strncpy(msg.pathname, pathname, sizeof(msg.pathname)); @@ -196,6 +255,7 @@ int relayd_add_stream(struct lttcomm_relayd_sock *rsock, const char *channel_nam goto error; } } else { + memset(&msg_2_2, 0, sizeof(msg_2_2)); /* Compat with relayd 2.2+ */ strncpy(msg_2_2.channel_name, channel_name, sizeof(msg_2_2.channel_name)); strncpy(msg_2_2.pathname, pathname, sizeof(msg_2_2.pathname)); @@ -236,6 +296,59 @@ error: return ret; } +/* + * Inform the relay that all the streams for the current channel has been sent. + * + * On success return 0 else return ret_code negative value. + */ +int relayd_streams_sent(struct lttcomm_relayd_sock *rsock) +{ + int ret; + struct lttcomm_relayd_generic_reply reply; + + /* Code flow error. Safety net. */ + assert(rsock); + + DBG("Relayd sending streams sent."); + + /* This feature was introduced in 2.4, ignore it for earlier versions. */ + if (rsock->minor < 4) { + ret = 0; + goto end; + } + + /* Send command */ + ret = send_command(rsock, RELAYD_STREAMS_SENT, NULL, 0, 0); + if (ret < 0) { + goto error; + } + + /* Waiting for reply */ + ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); + if (ret < 0) { + goto error; + } + + /* Back to host bytes order. */ + reply.ret_code = be32toh(reply.ret_code); + + /* Return session id or negative ret code. */ + if (reply.ret_code != LTTNG_OK) { + ret = -1; + ERR("Relayd streams sent replied error %d", reply.ret_code); + goto error; + } else { + /* Success */ + ret = 0; + } + + DBG("Relayd streams sent success"); + +error: +end: + return ret; +} + /* * Check version numbers on the relayd. * If major versions are compatible, we assign minor_to_use to the @@ -254,6 +367,7 @@ int relayd_version_check(struct lttcomm_relayd_sock *rsock) DBG("Relayd version check for major.minor %u.%u", rsock->major, rsock->minor); + memset(&msg, 0, sizeof(msg)); /* Prepare network byte order before transmission. */ msg.major = htobe32(rsock->major); msg.minor = htobe32(rsock->minor); @@ -455,6 +569,7 @@ int relayd_send_close_stream(struct lttcomm_relayd_sock *rsock, uint64_t stream_ DBG("Relayd closing stream id %" PRIu64, stream_id); + memset(&msg, 0, sizeof(msg)); msg.stream_id = htobe64(stream_id); msg.last_net_seq_num = htobe64(last_net_seq_num); @@ -504,6 +619,7 @@ int relayd_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t stream_id, DBG("Relayd data pending for stream id %" PRIu64, stream_id); + memset(&msg, 0, sizeof(msg)); msg.stream_id = htobe64(stream_id); msg.last_net_seq_num = htobe64(last_net_seq_num); @@ -552,6 +668,7 @@ int relayd_quiescent_control(struct lttcomm_relayd_sock *rsock, DBG("Relayd checking quiescent control state"); + memset(&msg, 0, sizeof(msg)); msg.stream_id = htobe64(metadata_stream_id); /* Send command */ @@ -596,6 +713,7 @@ int relayd_begin_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id) DBG("Relayd begin data pending"); + memset(&msg, 0, sizeof(msg)); msg.session_id = htobe64(id); /* Send command */ @@ -634,7 +752,7 @@ error: int relayd_end_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id, unsigned int *is_data_inflight) { - int ret; + int ret, recv_ret; struct lttcomm_relayd_end_data_pending msg; struct lttcomm_relayd_generic_reply reply; @@ -643,6 +761,7 @@ int relayd_end_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id, DBG("Relayd end data pending"); + memset(&msg, 0, sizeof(msg)); msg.session_id = htobe64(id); /* Send command */ @@ -657,15 +776,15 @@ int relayd_end_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id, goto error; } - reply.ret_code = be32toh(reply.ret_code); - if (reply.ret_code < 0) { - ret = reply.ret_code; + recv_ret = be32toh(reply.ret_code); + if (recv_ret < 0) { + ret = recv_ret; goto error; } - *is_data_inflight = reply.ret_code; + *is_data_inflight = recv_ret; - DBG("Relayd end data pending is data inflight: %d", reply.ret_code); + DBG("Relayd end data pending is data inflight: %d", recv_ret); return 0; @@ -677,7 +796,7 @@ error: * Send index to the relayd. */ int relayd_send_index(struct lttcomm_relayd_sock *rsock, - struct lttng_packet_index *index, uint64_t relay_stream_id, + struct ctf_packet_index *index, uint64_t relay_stream_id, uint64_t net_seq_num) { int ret; @@ -695,6 +814,7 @@ int relayd_send_index(struct lttcomm_relayd_sock *rsock, DBG("Relayd sending index for stream ID %" PRIu64, relay_stream_id); + memset(&msg, 0, sizeof(msg)); msg.relay_stream_id = htobe64(relay_stream_id); msg.net_seq_num = htobe64(net_seq_num);