relayd: Create output directory in main()
[lttng-tools.git] / src / bin / lttng-relayd / main.c
index 47780e2660c59a22ccf3e32c88cf3dd7a98ef12d..4f9d74274943fdc433ed31c6ab60b6b05e4d5258 100644 (file)
@@ -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);
This page took 0.029249 seconds and 4 git commands to generate.