X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=4f9d74274943fdc433ed31c6ab60b6b05e4d5258;hp=47780e2660c59a22ccf3e32c88cf3dd7a98ef12d;hb=4d513a5063879aa41c5ce27d245a7c56ecaa6e79;hpb=f921c78f991c899c741252949662cc2f589d8741 diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index 47780e266..4f9d74274 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -254,7 +254,7 @@ int notify_thread_pipe(int wpipe) do { ret = write(wpipe, "!", 1); } while (ret < 0 && errno == EINTR); - if (ret < 0) { + if (ret < 0 || ret != 1) { PERROR("write poll pipe"); } @@ -669,7 +669,7 @@ void *relay_thread_dispatcher(void *data) sizeof(struct relay_command)); } while (ret < 0 && errno == EINTR); free(relay_cmd); - if (ret < 0) { + if (ret < 0 || ret != sizeof(struct relay_command)) { PERROR("write cmd pipe"); goto error; } @@ -685,53 +685,6 @@ error: return NULL; } -/* - * Return the realpath(3) of the path even if the last directory token does not - * exist. For example, with /tmp/test1/test2, if test2/ does not exist but the - * /tmp/test1 does, the real path is returned. In normal time, realpath(3) - * fails if the end point directory does not exist. - */ -static -char *expand_full_path(const char *path) -{ - const char *end_path = path; - char *next, *cut_path, *expanded_path, *respath; - - /* Find last token delimited by '/' */ - while ((next = strpbrk(end_path + 1, "/"))) { - end_path = next; - } - - /* Cut last token from original path */ - cut_path = strndup(path, end_path - path); - - expanded_path = malloc(PATH_MAX); - if (expanded_path == NULL) { - respath = NULL; - goto end; - } - - respath = realpath(cut_path, expanded_path); - if (respath == NULL) { - switch (errno) { - case ENOENT: - ERR("%s: No such file or directory", cut_path); - break; - default: - PERROR("realpath"); - break; - } - free(expanded_path); - } else { - /* Add end part to expanded path */ - strcat(respath, end_path); - } -end: - free(cut_path); - return respath; -} - - /* * config_get_default_path * @@ -743,68 +696,6 @@ char *config_get_default_path(void) return getenv("HOME"); } -/* - * Create recursively directory using the FULL path. - */ -static -int mkdir_recursive(char *path, mode_t mode) -{ - char *p, tmp[PATH_MAX]; - struct stat statbuf; - size_t len; - int ret; - - ret = snprintf(tmp, sizeof(tmp), "%s", path); - if (ret < 0) { - PERROR("snprintf mkdir"); - goto error; - } - - len = ret; - if (tmp[len - 1] == '/') { - tmp[len - 1] = 0; - } - - for (p = tmp + 1; *p; p++) { - if (*p == '/') { - *p = 0; - if (tmp[strlen(tmp) - 1] == '.' && - tmp[strlen(tmp) - 2] == '.' && - tmp[strlen(tmp) - 3] == '/') { - ERR("Using '/../' is not permitted in the trace path (%s)", - tmp); - ret = -1; - goto error; - } - ret = stat(tmp, &statbuf); - if (ret < 0) { - ret = mkdir(tmp, mode); - if (ret < 0) { - if (errno != EEXIST) { - PERROR("mkdir recursive"); - ret = -errno; - goto error; - } - } - } - *p = '/'; - } - } - - ret = mkdir(tmp, mode); - if (ret < 0) { - if (errno != EEXIST) { - PERROR("mkdir recursive last piece"); - ret = -errno; - } else { - ret = 0; - } - } - -error: - return ret; -} - static char *create_output_path_auto(char *path_name) { @@ -842,7 +733,7 @@ char *create_output_path_noauto(char *path_name) char *traces_path = NULL; char *full_path; - full_path = expand_full_path(opt_output_path); + full_path = utils_expand_path(opt_output_path); ret = asprintf(&traces_path, "%s/%s", full_path, path_name); if (ret < 0) { PERROR("asprintf trace dir name"); @@ -866,6 +757,34 @@ char *create_output_path(char *path_name) } } +/* + * 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, + struct lttng_ht *streams_ht) +{ + struct lttng_ht_node_ulong *node; + struct lttng_ht_iter iter; + struct relay_stream *ret; + + 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 = NULL; + goto end; + } + + ret = caa_container_of(node, struct relay_stream, stream_n); + +end: + return ret; +} + static void deferred_free_stream(struct rcu_head *head) { @@ -958,6 +877,7 @@ error: send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0); if (send_ret < 0) { ERR("Relayd sending session id"); + ret = send_ret; } return ret; @@ -983,11 +903,15 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, goto end_no_session; } - /* FIXME : use data_size for something ? */ ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info, sizeof(struct lttcomm_relayd_add_stream), 0); if (ret < sizeof(struct lttcomm_relayd_add_stream)) { - ERR("Relay didn't receive valid add_stream struct size : %d", ret); + if (ret == 0) { + /* Orderly shutdown. Not necessary to print an error. */ + DBG("Socket %d did an orderly shutdown", cmd->sock->fd); + } else { + ERR("Relay didn't receive valid add_stream struct size : %d", ret); + } ret = -1; goto end_no_session; } @@ -1008,7 +932,7 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, ret = -1; goto end; } - ret = mkdir_recursive(root_path, S_IRWXU | S_IRWXG); + ret = utils_mkdir_recursive(root_path, S_IRWXU | S_IRWXG); if (ret < 0) { ERR("relay creating output directory"); goto end; @@ -1050,6 +974,7 @@ end: sizeof(struct lttcomm_relayd_status_stream), 0); if (send_ret < 0) { ERR("Relay sending stream id"); + ret = send_ret; } rcu_read_unlock(); @@ -1069,7 +994,6 @@ int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr, 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"); @@ -1083,23 +1007,19 @@ int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr, ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info, sizeof(struct lttcomm_relayd_close_stream), 0); if (ret < sizeof(struct lttcomm_relayd_close_stream)) { - ERR("Relay didn't receive valid add_stream struct size : %d", ret); + if (ret == 0) { + /* Orderly shutdown. Not necessary to print an error. */ + DBG("Socket %d did an orderly shutdown", cmd->sock->fd); + } else { + 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); + stream = relay_stream_from_stream_id(be64toh(stream_info.stream_id), + streams_ht); if (!stream) { ret = -1; goto end_unlock; @@ -1115,6 +1035,7 @@ int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr, if (delret < 0) { PERROR("close stream"); } + iter.iter.node = &stream->stream_n.node; delret = lttng_ht_del(streams_ht, &iter); assert(!delret); call_rcu(&stream->rcu_node, @@ -1134,6 +1055,7 @@ end_unlock: sizeof(struct lttcomm_relayd_generic_reply), 0); if (send_ret < 0) { ERR("Relay sending stream id"); + ret = send_ret; } end_no_session: @@ -1184,34 +1106,6 @@ int relay_start(struct lttcomm_relayd_hdr *recv_hdr, return 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, - struct lttng_ht *streams_ht) -{ - struct lttng_ht_node_ulong *node; - struct lttng_ht_iter iter; - struct relay_stream *ret; - - 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 = NULL; - goto end; - } - - ret = caa_container_of(node, struct relay_stream, stream_n); - -end: - return ret; -} - /* * Append padding to the file pointed by the file descriptor fd. */ @@ -1234,7 +1128,7 @@ static int write_padding_to_file(int fd, uint32_t size) do { ret = write(fd, zeros, size); } while (ret < 0 && errno == EINTR); - if (ret < 0) { + if (ret < 0 || ret != size) { PERROR("write padding to file"); } @@ -1289,8 +1183,13 @@ int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr, DBG2("Relay receiving metadata, waiting for %" PRIu64 " bytes", data_size); ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, 0); if (ret < 0 || ret != data_size) { + if (ret == 0) { + /* Orderly shutdown. Not necessary to print an error. */ + DBG("Socket %d did an orderly shutdown", cmd->sock->fd); + } else { + ERR("Relay didn't receive the whole metadata"); + } ret = -1; - ERR("Relay didn't receive the whole metadata"); goto end; } metadata_struct = (struct lttcomm_relayd_metadata_payload *) data_buffer; @@ -1307,7 +1206,7 @@ int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr, ret = write(metadata_stream->fd, metadata_struct->payload, payload_size); } while (ret < 0 && errno == EINTR); - if (ret < payload_size) { + if (ret < 0 || ret != payload_size) { ERR("Relay error writing metadata on file"); ret = -1; goto end_unlock; @@ -1344,8 +1243,13 @@ int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr, /* Get version from the other side. */ ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0); if (ret < 0 || ret != sizeof(msg)) { + if (ret == 0) { + /* Orderly shutdown. Not necessary to print an error. */ + DBG("Socket %d did an orderly shutdown", cmd->sock->fd); + } else { + ERR("Relay failed to receive the version values."); + } ret = -1; - ERR("Relay failed to receive the version values."); goto end; } @@ -1388,8 +1292,6 @@ int relay_data_pending(struct lttcomm_relayd_hdr *recv_hdr, 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"); @@ -1402,7 +1304,13 @@ int relay_data_pending(struct lttcomm_relayd_hdr *recv_hdr, ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0); if (ret < sizeof(msg)) { - ERR("Relay didn't receive valid data_pending struct size : %d", ret); + if (ret == 0) { + /* Orderly shutdown. Not necessary to print an error. */ + DBG("Socket %d did an orderly shutdown", cmd->sock->fd); + } else { + ERR("Relay didn't receive valid data_pending struct size : %d", + ret); + } ret = -1; goto end_no_session; } @@ -1411,23 +1319,18 @@ int relay_data_pending(struct lttcomm_relayd_hdr *recv_hdr, 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); + stream = relay_stream_from_stream_id(stream_id, streams_ht); + if (stream == NULL) { 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) { + if (((int64_t) (stream->prev_seq - last_net_seq_num)) >= 0) { /* Data has in fact been written and is NOT pending */ ret = 0; } else { @@ -1479,8 +1382,13 @@ int relay_quiescent_control(struct lttcomm_relayd_hdr *recv_hdr, ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0); if (ret < sizeof(msg)) { - ERR("Relay didn't receive valid begin data_pending struct size: %d", - ret); + if (ret == 0) { + /* Orderly shutdown. Not necessary to print an error. */ + DBG("Socket %d did an orderly shutdown", cmd->sock->fd); + } else { + ERR("Relay didn't receive valid begin data_pending struct size: %d", + ret); + } ret = -1; goto end_no_session; } @@ -1540,8 +1448,13 @@ int relay_begin_data_pending(struct lttcomm_relayd_hdr *recv_hdr, ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0); if (ret < sizeof(msg)) { - ERR("Relay didn't receive valid begin data_pending struct size: %d", - ret); + if (ret == 0) { + /* Orderly shutdown. Not necessary to print an error. */ + DBG("Socket %d did an orderly shutdown", cmd->sock->fd); + } else { + ERR("Relay didn't receive valid begin data_pending struct size: %d", + ret); + } ret = -1; goto end_no_session; } @@ -1610,8 +1523,13 @@ int relay_end_data_pending(struct lttcomm_relayd_hdr *recv_hdr, ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0); if (ret < sizeof(msg)) { - ERR("Relay didn't receive valid end data_pending struct size: %d", - ret); + if (ret == 0) { + /* Orderly shutdown. Not necessary to print an error. */ + DBG("Socket %d did an orderly shutdown", cmd->sock->fd); + } else { + ERR("Relay didn't receive valid end data_pending struct size: %d", + ret); + } ret = -1; goto end_no_session; } @@ -1711,7 +1629,12 @@ int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht) ret = cmd->sock->ops->recvmsg(cmd->sock, &data_hdr, sizeof(struct lttcomm_relayd_data_hdr), 0); if (ret <= 0) { - ERR("Connections seems to be closed"); + if (ret == 0) { + /* Orderly shutdown. Not necessary to print an error. */ + DBG("Socket %d did an orderly shutdown", cmd->sock->fd); + } else { + ERR("Unable to receive data header on sock %d", cmd->sock->fd); + } ret = -1; goto end; } @@ -1747,6 +1670,10 @@ int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht) data_size, stream_id, net_seq_num); ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, 0); if (ret <= 0) { + if (ret == 0) { + /* Orderly shutdown. Not necessary to print an error. */ + DBG("Socket %d did an orderly shutdown", cmd->sock->fd); + } ret = -1; goto end_unlock; } @@ -1754,7 +1681,7 @@ int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht) do { ret = write(stream->fd, data_buffer, data_size); } while (ret < 0 && errno == EINTR); - if (ret < data_size) { + if (ret < 0 || ret != data_size) { ERR("Relay error writing data to file"); ret = -1; goto end_unlock; @@ -2180,6 +2107,15 @@ int main(int argc, char **argv) goto exit; } + /* Try to create directory if -o, --output is specified. */ + if (opt_output_path) { + ret = utils_mkdir_recursive(opt_output_path, S_IRWXU | S_IRWXG); + if (ret < 0) { + ERR("Unable to create %s", opt_output_path); + goto exit; + } + } + /* Daemonize */ if (opt_daemon) { ret = daemon(0, 0);