#include <common/hashtable/hashtable.h>
#include <common/sessiond-comm/relayd.h>
#include <common/uri.h>
+#include <common/utils.h>
#include "lttng-relayd.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 int relay_cmd_pipe[2] = { -1, -1 };
+/* Shared between threads */
static int dispatch_thread_exit;
static pthread_t listener_thread;
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
char *default_address;
static struct option long_options[] = {
- { "control-port", 1, 0, 'C' },
- { "data-port", 1, 0, 'D' },
- { "daemonize", 0, 0, 'd' },
- { "help", 0, 0, 'h' },
- { "output", 1, 0, 'o' },
- { "verbose", 0, 0, 'v' },
- { NULL, 0, 0, 0 }
+ { "control-port", 1, 0, 'C', },
+ { "data-port", 1, 0, 'D', },
+ { "daemonize", 0, 0, 'd', },
+ { "help", 0, 0, 'h', },
+ { "output", 1, 0, 'o', },
+ { "verbose", 0, 0, 'v', },
+ { NULL, 0, 0, 0, },
};
while (1) {
static
void cleanup(void)
{
- int i, ret;
-
DBG("Cleaning up");
- for (i = 0; i < 2; i++) {
- if (thread_quit_pipe[i] >= 0) {
- ret = close(thread_quit_pipe[i]);
- if (ret) {
- PERROR("close");
- }
- }
- }
+ /* free the dynamically allocated opt_output_path */
+ free(opt_output_path);
+
+ /* Close thread quit pipes */
+ utils_close_pipe(thread_quit_pipe);
+
+ /* Close relay cmd pipes */
+ utils_close_pipe(relay_cmd_pipe);
}
/*
{
int ret;
- ret = write(wpipe, "!", 1);
+ do {
+ ret = write(wpipe, "!", 1);
+ } while (ret < 0 && errno == EINTR);
if (ret < 0) {
PERROR("write poll pipe");
}
}
/* Dispatch thread */
- dispatch_thread_exit = 1;
+ CMM_STORE_SHARED(dispatch_thread_exit, 1);
futex_nto1_wake(&relay_cmd_queue.futex);
}
static
int init_thread_quit_pipe(void)
{
- int ret, i;
-
- ret = pipe(thread_quit_pipe);
- if (ret < 0) {
- PERROR("thread quit pipe");
- goto error;
- }
+ int ret;
- for (i = 0; i < 2; i++) {
- ret = fcntl(thread_quit_pipe[i], F_SETFD, FD_CLOEXEC);
- if (ret < 0) {
- PERROR("fcntl");
- goto error;
- }
- }
+ ret = utils_create_pipe_cloexec(thread_quit_pipe);
-error:
return ret;
}
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;
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)) {
}
}
+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;
}
DBG("[thread] Relay dispatcher started");
- while (!dispatch_thread_exit) {
+ while (!CMM_LOAD_SHARED(dispatch_thread_exit)) {
/* Atomically prepare the queue futex */
futex_nto1_prepare(&relay_cmd_queue.futex);
* call is blocking so we can be assured that the data will be read
* at some point in time or wait to the end of the world :)
*/
- ret = write(relay_cmd_pipe[1], relay_cmd,
- sizeof(struct relay_command));
+ do {
+ ret = write(relay_cmd_pipe[1], relay_cmd,
+ sizeof(struct relay_command));
+ } while (ret < 0 && errno == EINTR);
free(relay_cmd);
if (ret < 0) {
PERROR("write cmd pipe");
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;
}
-/*
- * create_output_path: create the output trace directory
- */
static
-char *create_output_path(char *path_name)
+char *create_output_path_auto(char *path_name)
{
- int ret = 0;
- char *alloc_path = NULL;
+ int ret;
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;
- }
+ char *alloc_path = NULL;
+ char *default_path;
- ret = asprintf(&traces_path, "%s/" DEFAULT_TRACE_DIR_NAME
- "/%s", alloc_path, path_name);
- if (ret < 0) {
- PERROR("asprintf trace dir name");
- goto exit;
- }
- } 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;
- }
+ 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);
- free(full_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)
+{
+ if (opt_output_path == NULL) {
+ return create_output_path_auto(path_name);
+ } else {
+ return create_output_path_noauto(path_name);
+ }
+}
+
/*
* relay_delete_session: Free all memory associated with a session and
* close all the FDs
struct relay_stream *stream;
int ret;
- if (!cmd->session)
+ if (!cmd->session) {
return;
+ }
DBG("Relay deleting session %lu", cmd->session->id);
free(cmd->session->sock);
}
ret = mkdir_recursive(root_path, S_IRWXU | S_IRWXG);
if (ret < 0) {
- free(root_path);
ERR("relay creating output directory");
goto end;
}
goto end;
}
- ret = write(metadata_stream->fd, metadata_struct->payload,
- payload_size);
+ do {
+ ret = write(metadata_stream->fd, metadata_struct->payload,
+ payload_size);
+ } while (ret < 0 && errno == EINTR);
if (ret < (payload_size)) {
ERR("Relay error writing metadata on file");
ret = -1;
goto end;
}
- ret = write(stream->fd, data_buffer, data_size);
+ do {
+ ret = write(stream->fd, data_buffer, data_size);
+ } while (ret < 0 && errno == EINTR);
if (ret < data_size) {
ERR("Relay error writing data to file");
ret = -1;
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;
/* 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) {
/* 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 */
ERR("POLL ERROR");
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);
} else if (revents & (LPOLLHUP | LPOLLRDHUP)) {
DBG("Socket %d hung up", pollfd);
if (ret < 0) {
relay_cleanup_connection(relay_connections_ht,
&events, streams_ht, pollfd, &iter);
+ relay_delete_session(relay_connection, streams_ht);
free(relay_connection);
DBG("Connection closed with %d", pollfd);
}
relay_cleanup_connection(relay_connections_ht,
&events, streams_ht, pollfd, &iter);
relay_delete_session(relay_connection, streams_ht);
+ free(relay_connection);
DBG("Data connection closed with %d", pollfd);
}
}
}
}
+exit:
error:
lttng_poll_clean(&events);
if (node) {
relay_connection = caa_container_of(node,
struct relay_command, sock_n);
+ if (relay_connection->type == RELAY_CONTROL) {
+ relay_delete_session(relay_connection, streams_ht);
+ }
free(relay_connection);
}
ret = lttng_ht_del(relay_connections_ht, &iter);
assert(!ret);
}
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();
return NULL;
}
*/
static int create_relay_cmd_pipe(void)
{
- int ret, i;
-
- ret = pipe(relay_cmd_pipe);
- if (ret < 0) {
- PERROR("relay cmd pipe");
- goto error;
- }
+ int ret;
- for (i = 0; i < 2; i++) {
- ret = fcntl(relay_cmd_pipe[i], F_SETFD, FD_CLOEXEC);
- if (ret < 0) {
- PERROR("fcntl relay_cmd_pipe");
- goto error;
- }
- }
+ ret = utils_create_pipe_cloexec(relay_cmd_pipe);
-error:
return ret;
}
/* Parse arguments */
progname = argv[0];
if ((ret = parse_args(argc, argv) < 0)) {
- goto error;
+ goto exit;
}
if ((ret = set_signal_handler()) < 0) {
ret = daemon(0, 0);
if (ret < 0) {
PERROR("daemon");
- goto error;
+ goto exit;
}
}
if (control_uri->port < 1024 || data_uri->port < 1024) {
ERR("Need to be root to use ports < 1024");
ret = -1;
- goto error;
+ goto exit;
}
}
if (!ret) {
exit(EXIT_SUCCESS);
}
+
error:
exit(EXIT_FAILURE);
}
projects / lttng-tools.git / blobdiff