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;
/*
* 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)
{
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) {
/*
* 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)
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) {
/*
* 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)
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)
{
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));
/*
* 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,
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) {
DBG2("Channel monitoring pipe successfully sent");
error:
+ pthread_mutex_unlock(consumer_sock->lock);
return ret;
}
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;
} 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;
}
health_code_update();
+ pthread_mutex_lock(socket->lock);
ret = consumer_send_msg(socket, &msg);
+ pthread_mutex_unlock(socket->lock);
if (ret < 0) {
goto error;
}