X-Git-Url: http://git.lttng.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fconsumer.c;h=5f551e31dd5a05d6283e0aca2cbe01ca660096b4;hb=61ba6b6dc790245afa7fbec89a9811f6c7603b00;hp=6ee3975792bb061b46035d6d7b5923af1ab6fd0f;hpb=0dd01979d6f26886199ef746377640b57260421c;p=lttng-tools.git diff --git a/src/bin/lttng-sessiond/consumer.c b/src/bin/lttng-sessiond/consumer.c index 6ee397579..5f551e31d 100644 --- a/src/bin/lttng-sessiond/consumer.c +++ b/src/bin/lttng-sessiond/consumer.c @@ -561,7 +561,7 @@ struct consumer_output *consumer_copy_output(struct consumer_output *obj) } output->enabled = obj->enabled; output->net_seq_index = obj->net_seq_index; - memcpy(output->subdir, obj->subdir, PATH_MAX); + memcpy(output->subdir, obj->subdir, sizeof(output->subdir)); output->snapshot = obj->snapshot; output->relay_major_version = obj->relay_major_version; output->relay_minor_version = obj->relay_minor_version; @@ -715,11 +715,12 @@ int consumer_set_network_uri(struct consumer_output *obj, goto error; } - if (lttng_strncpy(obj->subdir, tmp_path, sizeof(obj->subdir))) { + if (lttng_strncpy(obj->dst.net.base_dir, tmp_path, + sizeof(obj->dst.net.base_dir))) { ret = -LTTNG_ERR_INVALID; goto error; } - DBG3("Consumer set network uri subdir path %s", tmp_path); + DBG3("Consumer set network uri base_dir path %s", tmp_path); } return 0; @@ -731,6 +732,8 @@ error: /* * Send file descriptor to consumer via sock. + * + * The consumer socket lock must be held by the caller. */ int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd) { @@ -739,6 +742,7 @@ int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd) assert(fds); assert(sock); assert(nb_fd > 0); + assert(pthread_mutex_trylock(sock->lock) == EBUSY); ret = lttcomm_send_fds_unix_sock(*sock->fd_ptr, fds, nb_fd); if (ret < 0) { @@ -748,13 +752,14 @@ int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd) } ret = consumer_recv_status_reply(sock); - error: return ret; } /* * Consumer send communication message structure to consumer. + * + * The consumer socket lock must be held by the caller. */ int consumer_send_msg(struct consumer_socket *sock, struct lttcomm_consumer_msg *msg) @@ -763,6 +768,7 @@ int consumer_send_msg(struct consumer_socket *sock, assert(msg); assert(sock); + assert(pthread_mutex_trylock(sock->lock) == EBUSY); ret = consumer_socket_send(sock, msg, sizeof(struct lttcomm_consumer_msg)); if (ret < 0) { @@ -777,6 +783,8 @@ error: /* * Consumer send channel communication message structure to consumer. + * + * The consumer socket lock must be held by the caller. */ int consumer_send_channel(struct consumer_socket *sock, struct lttcomm_consumer_msg *msg) @@ -806,6 +814,7 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, unsigned int switch_timer_interval, unsigned int read_timer_interval, unsigned int live_timer_interval, + unsigned int monitor_timer_interval, int output, int type, uint64_t session_id, @@ -822,6 +831,7 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, uint64_t session_id_per_pid, unsigned int monitor, uint32_t ust_app_uid, + int64_t blocking_timeout, const char *root_shm_path, const char *shm_path) { @@ -837,6 +847,7 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, msg->u.ask_channel.switch_timer_interval = switch_timer_interval; msg->u.ask_channel.read_timer_interval = read_timer_interval; msg->u.ask_channel.live_timer_interval = live_timer_interval; + msg->u.ask_channel.monitor_timer_interval = monitor_timer_interval; msg->u.ask_channel.output = output; msg->u.ask_channel.type = type; msg->u.ask_channel.session_id = session_id; @@ -850,6 +861,7 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, msg->u.ask_channel.tracefile_count = tracefile_count; msg->u.ask_channel.monitor = monitor; msg->u.ask_channel.ust_app_uid = ust_app_uid; + msg->u.ask_channel.blocking_timeout = blocking_timeout; memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid)); @@ -892,7 +904,8 @@ void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg, uint64_t tracefile_size, uint64_t tracefile_count, unsigned int monitor, - unsigned int live_timer_interval) + unsigned int live_timer_interval, + unsigned int monitor_timer_interval) { assert(msg); @@ -913,6 +926,7 @@ void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg, msg->u.channel.tracefile_count = tracefile_count; msg->u.channel.monitor = monitor; msg->u.channel.live_timer_interval = live_timer_interval; + msg->u.channel.monitor_timer_interval = monitor_timer_interval; strncpy(msg->u.channel.pathname, pathname, sizeof(msg->u.channel.pathname)); @@ -985,6 +999,8 @@ error: /* * Send relayd socket to consumer associated with a session name. * + * The consumer socket lock must be held by the caller. + * * On success return positive value. On error, negative value. */ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock, @@ -1048,6 +1064,37 @@ error: return ret; } +int consumer_send_channel_monitor_pipe(struct consumer_socket *consumer_sock, + int pipe) +{ + int ret; + struct lttcomm_consumer_msg msg; + + /* Code flow error. Safety net. */ + + memset(&msg, 0, sizeof(msg)); + msg.cmd_type = LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE; + + pthread_mutex_lock(consumer_sock->lock); + DBG3("Sending set_channel_monitor_pipe command to consumer"); + ret = consumer_send_msg(consumer_sock, &msg); + if (ret < 0) { + goto error; + } + + DBG3("Sending channel monitoring pipe %d to consumer on socket %d", + pipe, *consumer_sock->fd_ptr); + ret = consumer_send_fds(consumer_sock, &pipe, 1); + if (ret < 0) { + goto error; + } + + DBG2("Channel monitoring pipe successfully sent"); +error: + pthread_mutex_unlock(consumer_sock->lock); + return ret; +} + /* * Set consumer subdirectory using the session name and a generated datetime if * needed. This is appended to the current subdirectory. @@ -1371,8 +1418,11 @@ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key, msg.u.snapshot_channel.use_relayd = 1; ret = snprintf(msg.u.snapshot_channel.pathname, sizeof(msg.u.snapshot_channel.pathname), - "%s/%s-%s-%" PRIu64 "%s", output->consumer->subdir, - output->name, output->datetime, output->nb_snapshot, + "%s/%s/%s-%s-%" PRIu64 "%s", + output->consumer->dst.net.base_dir, + output->consumer->subdir, + output->name, output->datetime, + output->nb_snapshot, session_path); if (ret < 0) { ret = -LTTNG_ERR_NOMEM; @@ -1381,8 +1431,10 @@ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key, } else { ret = snprintf(msg.u.snapshot_channel.pathname, sizeof(msg.u.snapshot_channel.pathname), - "%s/%s-%s-%" PRIu64 "%s", output->consumer->dst.trace_path, - output->name, output->datetime, output->nb_snapshot, + "%s/%s-%s-%" PRIu64 "%s", + output->consumer->dst.session_root_path, + output->name, output->datetime, + output->nb_snapshot, session_path); if (ret < 0) { ret = -LTTNG_ERR_NOMEM; @@ -1402,7 +1454,9 @@ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key, } health_code_update(); + pthread_mutex_lock(socket->lock); ret = consumer_send_msg(socket, &msg); + pthread_mutex_unlock(socket->lock); if (ret < 0) { goto error; }