X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=18e60d1efbbe40fb45c678adc149b4e3c366468e;hp=dc6d2564a043abb44bb567d85a2a215d3317363f;hb=6d805429e9cb049eb0c9205fcf742a53e3166caf;hpb=7f51dcba1efdc03c85412a75ab71894a1b5af798 diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index dc6d2564a..18e60d1ef 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -241,6 +241,9 @@ void cleanup(void) /* Close relay cmd pipes */ utils_close_pipe(relay_cmd_pipe); + + uri_free(control_uri); + uri_free(data_uri); } /* @@ -884,7 +887,8 @@ void relay_delete_session(struct relay_command *cmd, struct lttng_ht *streams_ht } DBG("Relay deleting session %" PRIu64, cmd->session->id); - free(cmd->session->sock); + + lttcomm_destroy_sock(cmd->session->sock); rcu_read_lock(); cds_lfht_for_each_entry(streams_ht->ht, &iter.iter, node, node) { @@ -902,6 +906,8 @@ void relay_delete_session(struct relay_command *cmd, struct lttng_ht *streams_ht } } rcu_read_unlock(); + + free(cmd->session); } /* @@ -1046,6 +1052,7 @@ int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr, goto end_unlock; } + stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num); stream->close_flag = 1; if (close_stream_check(stream)) { @@ -1175,6 +1182,8 @@ static int write_padding_to_file(int fd, uint32_t size) PERROR("write padding to file"); } + free(zeros); + end: return ret; } @@ -1207,9 +1216,12 @@ int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr, payload_size -= sizeof(struct lttcomm_relayd_metadata_payload); if (data_buffer_size < data_size) { + /* In case the realloc fails, we can free the memory */ + char *tmp_data_ptr = data_buffer; data_buffer = realloc(data_buffer, data_size); if (!data_buffer) { ERR("Allocating data buffer"); + free(tmp_data_ptr); ret = -1; goto end; } @@ -1304,6 +1316,103 @@ end: return ret; } +/* + * Check for data pending for a given stream id from the session daemon. + */ +static +int relay_data_pending(struct lttcomm_relayd_hdr *recv_hdr, + struct relay_command *cmd, struct lttng_ht *streams_ht) +{ + struct relay_session *session = cmd->session; + struct lttcomm_relayd_data_pending msg; + struct lttcomm_relayd_generic_reply reply; + struct relay_stream *stream; + int ret; + struct lttng_ht_node_ulong *node; + struct lttng_ht_iter iter; + uint64_t last_net_seq_num, stream_id; + + DBG("Data pending command received"); + + if (!session || session->version_check_done == 0) { + ERR("Trying to check for data before version check"); + ret = -1; + goto end_no_session; + } + + ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), MSG_WAITALL); + if (ret < sizeof(msg)) { + ERR("Relay didn't receive valid data_pending struct size : %d", ret); + ret = -1; + goto end_no_session; + } + + stream_id = be64toh(msg.stream_id); + last_net_seq_num = be64toh(msg.last_net_seq_num); + + rcu_read_lock(); + lttng_ht_lookup(streams_ht, (void *)((unsigned long) stream_id), &iter); + node = lttng_ht_iter_get_node_ulong(&iter); + if (node == NULL) { + DBG("Relay stream %" PRIu64 " not found", stream_id); + ret = -1; + goto end_unlock; + } + + stream = caa_container_of(node, struct relay_stream, stream_n); + assert(stream); + + DBG("Data pending for stream id %" PRIu64 " prev_seq %" PRIu64 + " and last_seq %" PRIu64, stream_id, stream->prev_seq, + last_net_seq_num); + + /* Avoid wrapping issue */ + if (((int64_t) (stream->prev_seq - last_net_seq_num)) <= 0) { + /* Data has in fact been written and is NOT pending */ + ret = 0; + } else { + /* Data still being streamed thus pending */ + ret = 1; + } + +end_unlock: + rcu_read_unlock(); + + reply.ret_code = htobe32(ret); + ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0); + if (ret < 0) { + ERR("Relay data pending ret code failed"); + } + +end_no_session: + return ret; +} + +/* + * Wait for the control socket to reach a quiescent state. + * + * Note that for now, when receiving this command from the session daemon, this + * means that every subsequent commands or data received on the control socket + * has been handled. So, this is why we simply return OK here. + */ +static +int relay_quiescent_control(struct lttcomm_relayd_hdr *recv_hdr, + struct relay_command *cmd) +{ + int ret; + struct lttcomm_relayd_generic_reply reply; + + DBG("Checking quiescent state on control socket"); + + reply.ret_code = htobe32(LTTNG_OK); + ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0); + if (ret < 0) { + ERR("Relay data quiescent control ret code failed"); + } + + return ret; +} + /* * relay_process_control: Process the commands received on the control socket */ @@ -1334,6 +1443,12 @@ int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr, case RELAYD_CLOSE_STREAM: ret = relay_close_stream(recv_hdr, cmd, streams_ht); break; + case RELAYD_DATA_PENDING: + ret = relay_data_pending(recv_hdr, cmd, streams_ht); + break; + case RELAYD_QUIESCENT_CONTROL: + ret = relay_quiescent_control(recv_hdr, cmd); + break; case RELAYD_UPDATE_SYNC_INFO: default: ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd)); @@ -1378,9 +1493,11 @@ int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht) data_size = be32toh(data_hdr.data_size); if (data_buffer_size < data_size) { + char *tmp_data_ptr = data_buffer; data_buffer = realloc(data_buffer, data_size); if (!data_buffer) { ERR("Allocating data buffer"); + free(tmp_data_ptr); ret = -1; goto end_unlock; } @@ -1407,14 +1524,14 @@ int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht) goto end_unlock; } + DBG2("Relay wrote %d bytes to tracefile for stream id %" PRIu64, + ret, stream->stream_handle); + ret = write_padding_to_file(stream->fd, be32toh(data_hdr.padding_size)); if (ret < 0) { goto end_unlock; } - DBG2("Relay wrote %d bytes to tracefile for stream id %" PRIu64, - ret, stream->stream_handle); - stream->prev_seq = net_seq_num; /* Check if we need to close the FD */ @@ -1488,6 +1605,8 @@ void deferred_free_connection(struct rcu_head *head) { struct relay_command *relay_connection = caa_container_of(head, struct relay_command, rcu_node); + + lttcomm_destroy_sock(relay_connection->sock); free(relay_connection); } @@ -1503,6 +1622,7 @@ void relay_del_connection(struct lttng_ht *relay_connections_ht, if (relay_connection->type == RELAY_CONTROL) { relay_delete_session(relay_connection, streams_ht); } + call_rcu(&relay_connection->rcu_node, deferred_free_connection); }