#include <sys/stat.h>
#include <sys/types.h>
#include <sys/wait.h>
+#include <inttypes.h>
#include <urcu/futex.h>
#include <urcu/uatomic.h>
#include <unistd.h>
/* command line options */
static int opt_daemon;
static char *opt_output_path;
-static struct lttng_uri *control_uri = NULL;
-static struct lttng_uri *data_uri = NULL;
+static struct lttng_uri *control_uri;
+static struct lttng_uri *data_uri;
const char *progname;
static int is_root; /* Set to 1 if the daemon is running as root */
static pthread_t dispatcher_thread;
static pthread_t worker_thread;
-static uint64_t last_relay_stream_id = 0;
-static uint64_t last_relay_session_id = 0;
+static uint64_t last_relay_stream_id;
+static uint64_t last_relay_session_id;
/*
* Relay command queue.
static struct relay_cmd_queue relay_cmd_queue;
/* buffer allocated at startup, used to store the trace data */
-static char *data_buffer = NULL;
-static unsigned int data_buffer_size = 0;
+static char *data_buffer;
+static unsigned int data_buffer_size;
/*
* usage function on stderr
{ "help", 0, 0, 'h', },
{ "output", 1, 0, 'o', },
{ "verbose", 0, 0, 'v', },
- { NULL, 0, 0, 0 },
+ { NULL, 0, 0, 0, },
};
while (1) {
{
DBG("Cleaning up");
+ /* free the dynamically allocated opt_output_path */
+ free(opt_output_path);
+
/* Close thread quit pipes */
utils_close_pipe(thread_quit_pipe);
return NULL;
}
+/*
+ * Return nonzero if stream needs to be closed.
+ */
+static
+int close_stream_check(struct relay_stream *stream)
+{
+
+ if (stream->close_flag && stream->prev_seq == stream->last_net_seq_num) {
+ return 1;
+ }
+ return 0;
+}
+
/*
* This thread manages the listening for new connections on the network
*/
static
void *relay_thread_listener(void *data)
{
- int i, ret, pollfd;
+ int i, ret, pollfd, err = -1;
int val = 1;
uint32_t revents, nb_fd;
struct lttng_poll_event events;
struct lttcomm_sock *control_sock, *data_sock;
- /*
- * Get allocated in this thread, enqueued to a global queue, dequeued and
- * freed in the worker thread.
- */
- struct relay_command *relay_cmd = NULL;
-
DBG("[thread] Relay listener started");
control_sock = relay_init_sock(control_uri);
if (!control_sock) {
- goto error_sock;
+ goto error_sock_control;
}
data_sock = relay_init_sock(data_uri);
if (!data_sock) {
- goto error_sock;
+ goto error_sock_relay;
}
/*
/* Thread quit pipe has been closed. Killing thread. */
ret = check_thread_quit_pipe(pollfd, revents);
if (ret) {
- goto error;
+ err = 0;
+ goto exit;
}
if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
ERR("socket poll error");
goto error;
} else if (revents & LPOLLIN) {
- struct lttcomm_sock *newsock = NULL;
+ /*
+ * Get allocated in this thread,
+ * enqueued to a global queue, dequeued
+ * and freed in the worker thread.
+ */
+ struct relay_command *relay_cmd;
+ struct lttcomm_sock *newsock;
relay_cmd = zmalloc(sizeof(struct relay_command));
if (relay_cmd == NULL) {
if (pollfd == data_sock->fd) {
newsock = data_sock->ops->accept(data_sock);
- if (newsock < 0) {
+ if (!newsock) {
PERROR("accepting data sock");
+ free(relay_cmd);
goto error;
}
relay_cmd->type = RELAY_DATA;
DBG("Relay data connection accepted, socket %d", newsock->fd);
- } else if (pollfd == control_sock->fd) {
+ } else {
+ assert(pollfd == control_sock->fd);
newsock = control_sock->ops->accept(control_sock);
- if (newsock < 0) {
+ if (!newsock) {
PERROR("accepting control sock");
+ free(relay_cmd);
goto error;
}
relay_cmd->type = RELAY_CONTROL;
&val, sizeof(int));
if (ret < 0) {
PERROR("setsockopt inet");
+ lttcomm_destroy_sock(newsock);
+ free(relay_cmd);
goto error;
}
relay_cmd->sock = newsock;
}
}
+exit:
error:
error_poll_add:
lttng_poll_clean(&events);
error_create_poll:
- if (control_sock->fd >= 0) {
- ret = control_sock->ops->close(control_sock);
+ if (data_sock->fd >= 0) {
+ ret = data_sock->ops->close(data_sock);
if (ret) {
PERROR("close");
}
- lttcomm_destroy_sock(control_sock);
}
- if (data_sock->fd >= 0) {
- ret = data_sock->ops->close(data_sock);
+ lttcomm_destroy_sock(data_sock);
+error_sock_relay:
+ if (control_sock->fd >= 0) {
+ ret = control_sock->ops->close(control_sock);
if (ret) {
PERROR("close");
}
- lttcomm_destroy_sock(data_sock);
}
-
+ lttcomm_destroy_sock(control_sock);
+error_sock_control:
+ if (err) {
+ DBG("Thread exited with error");
+ }
DBG("Relay listener thread cleanup complete");
stop_threads();
-error_sock:
return NULL;
}
char *expand_full_path(const char *path)
{
const char *end_path = path;
- char *next, *cut_path, *expanded_path;
+ char *next, *cut_path, *expanded_path, *respath;
/* Find last token delimited by '/' */
while ((next = strpbrk(end_path + 1, "/"))) {
expanded_path = malloc(PATH_MAX);
if (expanded_path == NULL) {
- goto error;
+ respath = NULL;
+ goto end;
}
- expanded_path = realpath((char *)cut_path, expanded_path);
- if (expanded_path == NULL) {
+ respath = realpath(cut_path, expanded_path);
+ if (respath == NULL) {
switch (errno) {
case ENOENT:
ERR("%s: No such file or directory", cut_path);
PERROR("realpath");
break;
}
- goto error;
+ free(expanded_path);
+ } else {
+ /* Add end part to expanded path */
+ strcat(respath, end_path);
}
-
- /* Add end part to expanded path */
- strcat(expanded_path, end_path);
-
- free(cut_path);
- return expanded_path;
-
-error:
+end:
free(cut_path);
- return NULL;
+ return respath;
}
if (ret < 0) {
ret = mkdir(tmp, mode);
if (ret < 0) {
- if (!(errno == EEXIST)) {
+ if (errno != EEXIST) {
PERROR("mkdir recursive");
ret = -errno;
goto error;
ret = mkdir(tmp, mode);
if (ret < 0) {
- if (!(errno == EEXIST)) {
+ if (errno != EEXIST) {
PERROR("mkdir recursive last piece");
ret = -errno;
} else {
return ret;
}
+static
+char *create_output_path_auto(char *path_name)
+{
+ int ret;
+ char *traces_path = NULL;
+ char *alloc_path = NULL;
+ char *default_path;
+
+ default_path = config_get_default_path();
+ if (default_path == NULL) {
+ ERR("Home path not found.\n \
+ Please specify an output path using -o, --output PATH");
+ goto exit;
+ }
+ alloc_path = strdup(default_path);
+ if (alloc_path == NULL) {
+ PERROR("Path allocation");
+ goto exit;
+ }
+ ret = asprintf(&traces_path, "%s/" DEFAULT_TRACE_DIR_NAME
+ "/%s", alloc_path, path_name);
+ if (ret < 0) {
+ PERROR("asprintf trace dir name");
+ goto exit;
+ }
+exit:
+ free(alloc_path);
+ return traces_path;
+}
+
+static
+char *create_output_path_noauto(char *path_name)
+{
+ int ret;
+ char *traces_path = NULL;
+ char *full_path;
+
+ full_path = expand_full_path(opt_output_path);
+ ret = asprintf(&traces_path, "%s/%s", full_path, path_name);
+ if (ret < 0) {
+ PERROR("asprintf trace dir name");
+ goto exit;
+ }
+exit:
+ free(full_path);
+ return traces_path;
+}
+
/*
* create_output_path: create the output trace directory
*/
static
char *create_output_path(char *path_name)
{
- int ret = 0;
- char *alloc_path = NULL;
- char *traces_path = NULL;
- char *full_path = NULL;
-
- /* Auto output path */
if (opt_output_path == NULL) {
- alloc_path = strdup(config_get_default_path());
- if (alloc_path == NULL) {
- ERR("Home path not found.\n \
- Please specify an output path using -o, --output PATH");
- ret = -1;
- goto exit;
- }
-
- ret = asprintf(&traces_path, "%s/" DEFAULT_TRACE_DIR_NAME
- "/%s", alloc_path, path_name);
- if (ret < 0) {
- PERROR("asprintf trace dir name");
- goto exit;
- }
+ return create_output_path_auto(path_name);
} else {
- full_path = expand_full_path(opt_output_path);
- ret = asprintf(&traces_path, "%s/%s", full_path, path_name);
- if (ret < 0) {
- PERROR("asprintf trace dir name");
- goto exit;
- }
+ return create_output_path_noauto(path_name);
}
- free(alloc_path);
- free(full_path);
+}
-exit:
- return traces_path;
+static
+void deferred_free_stream(struct rcu_head *head)
+{
+ struct relay_stream *stream =
+ caa_container_of(head, struct relay_stream, rcu_node);
+ free(stream);
}
/*
struct relay_stream *stream;
int ret;
- if (!cmd->session)
+ if (!cmd->session) {
return;
+ }
- DBG("Relay deleting session %lu", cmd->session->id);
+ DBG("Relay deleting session %" PRIu64, cmd->session->id);
free(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);
if (node) {
close(stream->fd);
ret = lttng_ht_del(streams_ht, &iter);
assert(!ret);
- free(stream);
+ call_rcu(&stream->rcu_node,
+ deferred_free_stream);
}
}
}
+ rcu_read_unlock();
}
/*
goto end_no_session;
}
+ rcu_read_lock();
stream->stream_handle = ++last_relay_stream_id;
- stream->seq = 0;
+ stream->prev_seq = -1ULL;
stream->session = session;
root_path = create_output_path(stream_info.pathname);
}
ret = mkdir_recursive(root_path, S_IRWXU | S_IRWXG);
if (ret < 0) {
- free(root_path);
ERR("relay creating output directory");
goto end;
}
free(root_path);
/* send the session id to the client or a negative return code on error */
if (ret < 0) {
- reply.ret_code = htobe32(LTTCOMM_ERR);
+ reply.ret_code = htobe32(LTTNG_ERR_UNK);
} else {
- reply.ret_code = htobe32(LTTCOMM_OK);
+ reply.ret_code = htobe32(LTTNG_OK);
}
reply.handle = htobe64(stream->stream_handle);
send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
if (send_ret < 0) {
ERR("Relay sending stream id");
}
+ rcu_read_unlock();
+
+end_no_session:
+ return ret;
+}
+
+/*
+ * relay_close_stream: close a specific stream
+ */
+static
+int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_command *cmd, struct lttng_ht *streams_ht)
+{
+ struct relay_session *session = cmd->session;
+ struct lttcomm_relayd_close_stream stream_info;
+ struct lttcomm_relayd_generic_reply reply;
+ struct relay_stream *stream;
+ int ret, send_ret;
+ struct lttng_ht_node_ulong *node;
+ struct lttng_ht_iter iter;
+
+ DBG("Close stream received");
+
+ if (!session || session->version_check_done == 0) {
+ ERR("Trying to close a stream before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info,
+ sizeof(struct lttcomm_relayd_close_stream), MSG_WAITALL);
+ if (ret < sizeof(struct lttcomm_relayd_close_stream)) {
+ ERR("Relay didn't receive valid add_stream struct size : %d", ret);
+ ret = -1;
+ goto end_no_session;
+ }
+
+ rcu_read_lock();
+ lttng_ht_lookup(streams_ht,
+ (void *)((unsigned long) be64toh(stream_info.stream_id)),
+ &iter);
+ node = lttng_ht_iter_get_node_ulong(&iter);
+ if (node == NULL) {
+ DBG("Relay stream %" PRIu64 " not found", be64toh(stream_info.stream_id));
+ ret = -1;
+ goto end_unlock;
+ }
+
+ stream = caa_container_of(node, struct relay_stream, stream_n);
+ if (!stream) {
+ ret = -1;
+ goto end_unlock;
+ }
+
+ stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num);
+ stream->close_flag = 1;
+
+ if (close_stream_check(stream)) {
+ int delret;
+
+ close(stream->fd);
+ delret = lttng_ht_del(streams_ht, &iter);
+ assert(!delret);
+ call_rcu(&stream->rcu_node,
+ deferred_free_stream);
+ DBG("Closed tracefile %d from close stream", stream->fd);
+ }
+
+end_unlock:
+ rcu_read_unlock();
+
+ if (ret < 0) {
+ reply.ret_code = htobe32(LTTNG_ERR_UNK);
+ } else {
+ reply.ret_code = htobe32(LTTNG_OK);
+ }
+ send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
+ sizeof(struct lttcomm_relayd_generic_reply), 0);
+ if (send_ret < 0) {
+ ERR("Relay sending stream id");
+ }
end_no_session:
return ret;
struct lttcomm_relayd_generic_reply reply;
int ret;
- reply.ret_code = htobe32(LTTCOMM_ERR);
+ reply.ret_code = htobe32(LTTNG_ERR_UNK);
ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
sizeof(struct lttcomm_relayd_generic_reply), 0);
if (ret < 0) {
int relay_start(struct lttcomm_relayd_hdr *recv_hdr,
struct relay_command *cmd)
{
- int ret = htobe32(LTTCOMM_OK);
+ int ret = htobe32(LTTNG_OK);
struct lttcomm_relayd_generic_reply reply;
struct relay_session *session = cmd->session;
if (!session) {
DBG("Trying to start the streaming without a session established");
- ret = htobe32(LTTCOMM_ERR);
+ ret = htobe32(LTTNG_ERR_UNK);
}
reply.ret_code = ret;
/*
* Get stream from stream id.
+ * Need to be called with RCU read-side lock held.
*/
static
struct relay_stream *relay_stream_from_stream_id(uint64_t stream_id,
&iter);
node = lttng_ht_iter_get_node_ulong(&iter);
if (node == NULL) {
- DBG("Relay stream %lu not found", stream_id);
+ DBG("Relay stream %" PRIu64 " not found", stream_id);
ret = NULL;
goto end;
}
return ret;
}
+/*
+ * Append padding to the file pointed by the file descriptor fd.
+ */
+static int write_padding_to_file(int fd, uint32_t size)
+{
+ int ret = 0;
+ char *zeros;
+
+ if (size == 0) {
+ goto end;
+ }
+
+ zeros = zmalloc(size);
+ if (zeros == NULL) {
+ PERROR("zmalloc zeros for padding");
+ ret = -1;
+ goto end;
+ }
+
+ do {
+ ret = write(fd, zeros, size);
+ } while (ret < 0 && errno == EINTR);
+ if (ret < 0) {
+ PERROR("write padding to file");
+ }
+
+end:
+ return ret;
+}
+
/*
* relay_recv_metadata: receive the metada for the session.
*/
int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr,
struct relay_command *cmd, struct lttng_ht *streams_ht)
{
- int ret = htobe32(LTTCOMM_OK);
+ int ret = htobe32(LTTNG_OK);
struct relay_session *session = cmd->session;
struct lttcomm_relayd_metadata_payload *metadata_struct;
struct relay_stream *metadata_stream;
goto end;
}
- data_size = be64toh(recv_hdr->data_size);
- payload_size = data_size - sizeof(uint64_t);
+ data_size = payload_size = be64toh(recv_hdr->data_size);
+ if (data_size < sizeof(struct lttcomm_relayd_metadata_payload)) {
+ ERR("Incorrect data size");
+ ret = -1;
+ goto end;
+ }
+ payload_size -= sizeof(struct lttcomm_relayd_metadata_payload);
+
if (data_buffer_size < data_size) {
data_buffer = realloc(data_buffer, data_size);
if (!data_buffer) {
data_buffer_size = data_size;
}
memset(data_buffer, 0, data_size);
- DBG2("Relay receiving metadata, waiting for %lu bytes", data_size);
- ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, MSG_WAITALL);
+ DBG2("Relay receiving metadata, waiting for %" PRIu64 " bytes", data_size);
+ ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size,
+ MSG_WAITALL);
if (ret < 0 || ret != data_size) {
ret = -1;
ERR("Relay didn't receive the whole metadata");
goto end;
}
metadata_struct = (struct lttcomm_relayd_metadata_payload *) data_buffer;
+
+ rcu_read_lock();
metadata_stream = relay_stream_from_stream_id(
be64toh(metadata_struct->stream_id), streams_ht);
if (!metadata_stream) {
ret = -1;
- goto end;
+ goto end_unlock;
}
do {
ret = write(metadata_stream->fd, metadata_struct->payload,
payload_size);
} while (ret < 0 && errno == EINTR);
- if (ret < (payload_size)) {
+ if (ret < payload_size) {
ERR("Relay error writing metadata on file");
ret = -1;
- goto end;
+ goto end_unlock;
}
+
+ ret = write_padding_to_file(metadata_stream->fd,
+ be32toh(metadata_struct->padding_size));
+ if (ret < 0) {
+ goto end_unlock;
+ }
+
DBG2("Relay metadata written");
+end_unlock:
+ rcu_read_unlock();
end:
return ret;
}
int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr,
struct relay_command *cmd)
{
- int ret = htobe32(LTTCOMM_OK);
+ int ret;
struct lttcomm_relayd_version reply;
- struct relay_session *session = NULL;
+ struct relay_session *session;
if (cmd->session == NULL) {
session = zmalloc(sizeof(struct relay_session));
goto end;
}
session->id = ++last_relay_session_id;
- DBG("Created session %lu", session->id);
+ DBG("Created session %" PRIu64, session->id);
cmd->session = session;
+ } else {
+ session = cmd->session;
}
session->version_check_done = 1;
- sscanf(VERSION, "%u.%u", &reply.major, &reply.minor);
+ ret = sscanf(VERSION, "%u.%u", &reply.major, &reply.minor);
+ if (ret < 2) {
+ ERR("Error in scanning version");
+ ret = -1;
+ goto end;
+ }
reply.major = htobe32(reply.major);
reply.minor = htobe32(reply.minor);
ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
if (ret < 0) {
ERR("Relay sending version");
}
- DBG("Version check done");
+ DBG("Version check done (%u.%u)", be32toh(reply.major),
+ be32toh(reply.minor));
end:
return ret;
}
+/*
+ * Check for data availability for a given stream id from the session daemon.
+ */
+static
+int relay_data_available(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_available 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 available 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_available 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 available for stream id %" PRIu64 " prev_seq %" PRIu64
+ " and last_seq %" PRIu64, stream_id, stream->prev_seq,
+ last_net_seq_num);
+
+ if (stream->prev_seq == -1UL || stream->prev_seq <= last_net_seq_num) {
+ /* Data has in fact been written and is available */
+ ret = 1;
+ } else {
+ /* Data still being streamed. */
+ ret = 0;
+ }
+
+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 available 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 available ret code failed");
+ }
+
+ return ret;
+}
+
/*
* relay_process_control: Process the commands received on the control socket
*/
case RELAYD_VERSION:
ret = relay_send_version(recv_hdr, cmd);
break;
+ case RELAYD_CLOSE_STREAM:
+ ret = relay_close_stream(recv_hdr, cmd, streams_ht);
+ break;
+ case RELAYD_DATA_AVAILABLE:
+ ret = relay_data_available(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));
struct relay_stream *stream;
struct lttcomm_relayd_data_hdr data_hdr;
uint64_t stream_id;
+ uint64_t net_seq_num;
uint32_t data_size;
ret = cmd->sock->ops->recvmsg(cmd->sock, &data_hdr,
}
stream_id = be64toh(data_hdr.stream_id);
+
+ rcu_read_lock();
stream = relay_stream_from_stream_id(stream_id, streams_ht);
if (!stream) {
ret = -1;
- goto end;
+ goto end_unlock;
}
data_size = be32toh(data_hdr.data_size);
if (!data_buffer) {
ERR("Allocating data buffer");
ret = -1;
- goto end;
+ goto end_unlock;
}
data_buffer_size = data_size;
}
memset(data_buffer, 0, data_size);
+ net_seq_num = be64toh(data_hdr.net_seq_num);
+
+ DBG3("Receiving data of size %u for stream id %" PRIu64 " seqnum %" PRIu64,
+ data_size, stream_id, net_seq_num);
ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, MSG_WAITALL);
if (ret <= 0) {
ret = -1;
- goto end;
+ goto end_unlock;
}
do {
if (ret < data_size) {
ERR("Relay error writing data to file");
ret = -1;
- goto end;
+ 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;
+ }
+
+ stream->prev_seq = net_seq_num;
+
+ /* Check if we need to close the FD */
+ if (close_stream_check(stream)) {
+ struct lttng_ht_iter iter;
+
+ close(stream->fd);
+ iter.iter.node = &stream->stream_n.node;
+ ret = lttng_ht_del(streams_ht, &iter);
+ assert(!ret);
+ call_rcu(&stream->rcu_node,
+ deferred_free_stream);
+ DBG("Closed tracefile %d after recv data", stream->fd);
}
- DBG2("Relay wrote %d bytes to tracefile for stream id %lu", ret, stream->stream_handle);
+end_unlock:
+ rcu_read_unlock();
end:
return ret;
}
static
-void relay_cleanup_connection(struct lttng_ht *relay_connections_ht, struct lttng_poll_event *events,
- struct lttng_ht *streams_ht, int pollfd, struct lttng_ht_iter *iter)
+void relay_cleanup_poll_connection(struct lttng_poll_event *events, int pollfd)
{
int ret;
- ret = lttng_ht_del(relay_connections_ht, iter);
- assert(!ret);
lttng_poll_del(events, pollfd);
ret = close(pollfd);
int relay_add_connection(int fd, struct lttng_poll_event *events,
struct lttng_ht *relay_connections_ht)
{
- int ret;
struct relay_command *relay_connection;
+ int ret;
relay_connection = zmalloc(sizeof(struct relay_command));
if (relay_connection == NULL) {
PERROR("Relay command zmalloc");
- ret = -1;
- goto end;
+ goto error;
}
ret = read(fd, relay_connection, sizeof(struct relay_command));
- if (ret < 0 || ret < sizeof(relay_connection)) {
+ if (ret < 0 || ret < sizeof(struct relay_command)) {
PERROR("read relay cmd pipe");
- ret = -1;
- goto end;
+ goto error_read;
}
lttng_ht_node_init_ulong(&relay_connection->sock_n,
(unsigned long) relay_connection->sock->fd);
+ rcu_read_lock();
lttng_ht_add_unique_ulong(relay_connections_ht,
&relay_connection->sock_n);
- ret = lttng_poll_add(events,
+ rcu_read_unlock();
+ return lttng_poll_add(events,
relay_connection->sock->fd,
LPOLLIN | LPOLLRDHUP);
-end:
- return ret;
+error_read:
+ free(relay_connection);
+error:
+ return -1;
+}
+
+static
+void deferred_free_connection(struct rcu_head *head)
+{
+ struct relay_command *relay_connection =
+ caa_container_of(head, struct relay_command, rcu_node);
+ free(relay_connection);
+}
+
+static
+void relay_del_connection(struct lttng_ht *relay_connections_ht,
+ struct lttng_ht *streams_ht, struct lttng_ht_iter *iter,
+ struct relay_command *relay_connection)
+{
+ int ret;
+
+ ret = lttng_ht_del(relay_connections_ht, iter);
+ assert(!ret);
+ if (relay_connection->type == RELAY_CONTROL) {
+ relay_delete_session(relay_connection, streams_ht);
+ }
+ call_rcu(&relay_connection->rcu_node,
+ deferred_free_connection);
}
/*
static
void *relay_thread_worker(void *data)
{
- int i, ret, pollfd;
+ int i, ret, pollfd, err = -1;
uint32_t revents, nb_fd;
struct relay_command *relay_connection;
struct lttng_poll_event events;
DBG("[thread] Relay worker started");
+ rcu_register_thread();
+
/* table of connections indexed on socket */
relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
+ if (!relay_connections_ht) {
+ goto relay_connections_ht_error;
+ }
/* tables of streams indexed by stream ID */
streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
+ if (!streams_ht) {
+ goto streams_ht_error;
+ }
ret = create_thread_poll_set(&events, 2);
if (ret < 0) {
/* Infinite blocking call, waiting for transmission */
restart:
+ DBG3("Relayd worker thread polling...");
ret = lttng_poll_wait(&events, -1);
if (ret < 0) {
/*
/* Thread quit pipe has been closed. Killing thread. */
ret = check_thread_quit_pipe(pollfd, revents);
if (ret) {
- goto error;
+ err = 0;
+ goto exit;
}
/* Inspect the relay cmd pipe for new connection */
}
}
} else if (revents > 0) {
+ rcu_read_lock();
lttng_ht_lookup(relay_connections_ht,
(void *)((unsigned long) pollfd),
&iter);
node = lttng_ht_iter_get_node_ulong(&iter);
if (node == NULL) {
DBG2("Relay sock %d not found", pollfd);
+ rcu_read_unlock();
goto error;
}
relay_connection = caa_container_of(node,
if (revents & (LPOLLERR)) {
ERR("POLL ERROR");
- relay_cleanup_connection(relay_connections_ht,
- &events, streams_ht, pollfd, &iter);
- free(relay_connection);
+ relay_cleanup_poll_connection(&events, pollfd);
+ relay_del_connection(relay_connections_ht,
+ streams_ht, &iter,
+ relay_connection);
} else if (revents & (LPOLLHUP | LPOLLRDHUP)) {
DBG("Socket %d hung up", pollfd);
- relay_cleanup_connection(relay_connections_ht,
- &events, streams_ht, pollfd, &iter);
- if (relay_connection->type == RELAY_CONTROL) {
- relay_delete_session(relay_connection, streams_ht);
- }
- free(relay_connection);
+ relay_cleanup_poll_connection(&events, pollfd);
+ relay_del_connection(relay_connections_ht,
+ streams_ht, &iter,
+ relay_connection);
} else if (revents & LPOLLIN) {
/* control socket */
if (relay_connection->type == RELAY_CONTROL) {
sizeof(struct lttcomm_relayd_hdr), MSG_WAITALL);
/* connection closed */
if (ret <= 0) {
- relay_cleanup_connection(relay_connections_ht,
- &events, streams_ht, pollfd, &iter);
- relay_delete_session(relay_connection, streams_ht);
- free(relay_connection);
+ relay_cleanup_poll_connection(&events, pollfd);
+ relay_del_connection(relay_connections_ht,
+ streams_ht, &iter,
+ relay_connection);
DBG("Control connection closed with %d", pollfd);
} else {
if (relay_connection->session) {
- DBG2("Relay worker receiving data for session : %lu",
+ DBG2("Relay worker receiving data for session : %" PRIu64,
relay_connection->session->id);
}
ret = relay_process_control(&recv_hdr,
* command: clear the session
* */
if (ret < 0) {
- relay_cleanup_connection(relay_connections_ht,
- &events, streams_ht, pollfd, &iter);
- free(relay_connection);
+ relay_cleanup_poll_connection(&events, pollfd);
+ relay_del_connection(relay_connections_ht,
+ streams_ht, &iter,
+ relay_connection);
DBG("Connection closed with %d", pollfd);
}
}
ret = relay_process_data(relay_connection, streams_ht);
/* connection closed */
if (ret < 0) {
- relay_cleanup_connection(relay_connections_ht,
- &events, streams_ht, pollfd, &iter);
- relay_delete_session(relay_connection, streams_ht);
+ relay_cleanup_poll_connection(&events, pollfd);
+ relay_del_connection(relay_connections_ht,
+ streams_ht, &iter,
+ relay_connection);
DBG("Data connection closed with %d", pollfd);
}
}
}
+ rcu_read_unlock();
}
}
}
+exit:
error:
lttng_poll_clean(&events);
/* empty the hash table and free the memory */
+ rcu_read_lock();
cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, node, node) {
node = lttng_ht_iter_get_node_ulong(&iter);
if (node) {
relay_connection = caa_container_of(node,
struct relay_command, sock_n);
- free(relay_connection);
+ relay_del_connection(relay_connections_ht,
+ streams_ht, &iter,
+ relay_connection);
}
- ret = lttng_ht_del(relay_connections_ht, &iter);
- assert(!ret);
}
+ rcu_read_unlock();
error_poll_create:
- free(data_buffer);
+ lttng_ht_destroy(streams_ht);
+streams_ht_error:
lttng_ht_destroy(relay_connections_ht);
+relay_connections_ht_error:
+ if (err) {
+ DBG("Thread exited with error");
+ }
DBG("Worker thread cleanup complete");
+ free(data_buffer);
stop_threads();
+ rcu_unregister_thread();
return NULL;
}
projects / lttng-tools.git / blobdiff