#define _GNU_SOURCE
#include <assert.h>
+#include <lttng/ust-ctl.h>
#include <poll.h>
#include <pthread.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/types.h>
+#include <inttypes.h>
#include <unistd.h>
-#include <lttng/ust-ctl.h>
#include <common/common.h>
#include <common/sessiond-comm/sessiond-comm.h>
extern volatile int consumer_quit;
/*
- * Mmap the ring buffer, read it and write the data to the tracefile.
- *
- * Returns the number of bytes written, else negative value on error.
- */
-ssize_t lttng_ustconsumer_on_read_subbuffer_mmap(
- struct lttng_consumer_local_data *ctx,
- struct lttng_consumer_stream *stream, unsigned long len)
-{
- unsigned long mmap_offset;
- long ret = 0, written = 0;
- off_t orig_offset = stream->out_fd_offset;
- int outfd = stream->out_fd;
- uint64_t metadata_id;
- struct consumer_relayd_sock_pair *relayd = NULL;
-
- /* RCU lock for the relayd pointer */
- rcu_read_lock();
-
- /* Flag that the current stream if set for network streaming. */
- if (stream->net_seq_idx != -1) {
- relayd = consumer_find_relayd(stream->net_seq_idx);
- if (relayd == NULL) {
- ERR("UST consumer mmap(), unable to find relay for index %d",
- stream->net_seq_idx);
- goto end;
- }
- }
-
- /* get the offset inside the fd to mmap */
- ret = ustctl_get_mmap_read_offset(stream->chan->handle,
- stream->buf, &mmap_offset);
- if (ret != 0) {
- errno = -ret;
- PERROR("ustctl_get_mmap_read_offset");
- written = ret;
- goto end;
- }
-
- /* Handle stream on the relayd if the output is on the network */
- if (relayd) {
- unsigned long netlen = len;
-
- if (stream->metadata_flag) {
- /* Only lock if metadata since we use the control socket. */
- pthread_mutex_lock(&relayd->ctrl_sock_mutex);
- netlen += sizeof(stream->relayd_stream_id);
- }
-
- ret = consumer_handle_stream_before_relayd(stream, netlen);
- if (ret >= 0) {
- outfd = ret;
-
- /* Write metadata stream id before payload */
- if (stream->metadata_flag) {
- metadata_id = htobe64(stream->relayd_stream_id);
- do {
- ret = write(outfd, (void *) &metadata_id,
- sizeof(stream->relayd_stream_id));
- } while (ret < 0 && errno == EINTR);
- if (ret < 0) {
- PERROR("write metadata stream id");
- written = ret;
- goto end;
- }
- DBG("Metadata stream id %zu written before data",
- stream->relayd_stream_id);
- }
- }
- /* Else, use the default set before which is the filesystem. */
- }
-
- while (len > 0) {
- do {
- ret = write(outfd, stream->mmap_base + mmap_offset, len);
- } while (ret < 0 && errno == EINTR);
- if (ret < 0) {
- PERROR("Error in file write");
- if (written == 0) {
- written = ret;
- }
- goto end;
- } else if (ret > len) {
- PERROR("ret %ld > len %lu", ret, len);
- written += ret;
- goto end;
- } else {
- len -= ret;
- mmap_offset += ret;
- }
- DBG("UST mmap write() ret %ld (len %lu)", ret, len);
-
- /* This call is useless on a socket so better save a syscall. */
- if (!relayd) {
- /* This won't block, but will start writeout asynchronously */
- lttng_sync_file_range(outfd, stream->out_fd_offset, ret,
- SYNC_FILE_RANGE_WRITE);
- stream->out_fd_offset += ret;
- }
- written += ret;
- }
- lttng_consumer_sync_trace_file(stream, orig_offset);
-
-end:
- if (relayd && stream->metadata_flag) {
- pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
- }
- rcu_read_unlock();
- return written;
-}
-
-/*
- * Splice the data from the ring buffer to the tracefile.
- *
- * Returns the number of bytes spliced.
+ * Wrapper over the mmap() read offset from ust-ctl library. Since this can be
+ * compiled out, we isolate it in this library.
*/
-ssize_t lttng_ustconsumer_on_read_subbuffer_splice(
- struct lttng_consumer_local_data *ctx,
- struct lttng_consumer_stream *stream, unsigned long len)
+int lttng_ustctl_get_mmap_read_offset(struct lttng_ust_shm_handle *handle,
+ struct lttng_ust_lib_ring_buffer *buf, unsigned long *off)
{
- return -ENOSYS;
-}
+ return ustctl_get_mmap_read_offset(handle, buf, off);
+};
/*
* Take a snapshot for a specific fd
return ret;
}
+/*
+ * Receive command from session daemon and process it.
+ *
+ * Return 1 on success else a negative value or 0.
+ */
int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
int sock, struct pollfd *consumer_sockpoll)
{
if (ret != sizeof(msg)) {
DBG("Consumer received unexpected message size %zd (expects %zu)",
ret, sizeof(msg));
- lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
+ lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
return ret;
}
if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
return -ENOENT;
}
- /* relayd need RCU read-side lock */
+ /* relayd needs RCU read-side lock */
rcu_read_lock();
switch (msg.cmd_type) {
case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
{
- int fd;
- struct consumer_relayd_sock_pair *relayd;
-
- DBG("UST Consumer adding relayd socket");
-
- /* Get relayd reference if exists. */
- relayd = consumer_find_relayd(msg.u.relayd_sock.net_index);
- if (relayd == NULL) {
- /* Not found. Allocate one. */
- relayd = consumer_allocate_relayd_sock_pair(
- msg.u.relayd_sock.net_index);
- if (relayd == NULL) {
- lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
- goto end_nosignal;
- }
- }
-
- /* Poll on consumer socket. */
- if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
- return -EINTR;
- }
-
- /* Get relayd socket from session daemon */
- ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
- if (ret != sizeof(fd)) {
- lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
- goto end_nosignal;
- }
-
- /* Copy socket information and received FD */
- switch (msg.u.relayd_sock.type) {
- case LTTNG_STREAM_CONTROL:
- /* Copy received lttcomm socket */
- lttcomm_copy_sock(&relayd->control_sock, &msg.u.relayd_sock.sock);
- ret = lttcomm_create_sock(&relayd->control_sock);
- if (ret < 0) {
- goto end_nosignal;
- }
-
- /* Close the created socket fd which is useless */
- close(relayd->control_sock.fd);
-
- /* Assign new file descriptor */
- relayd->control_sock.fd = fd;
- break;
- case LTTNG_STREAM_DATA:
- /* Copy received lttcomm socket */
- lttcomm_copy_sock(&relayd->data_sock, &msg.u.relayd_sock.sock);
- ret = lttcomm_create_sock(&relayd->data_sock);
- if (ret < 0) {
- goto end_nosignal;
- }
-
- /* Close the created socket fd which is useless */
- close(relayd->data_sock.fd);
-
- /* Assign new file descriptor */
- relayd->data_sock.fd = fd;
- break;
- default:
- ERR("Unknown relayd socket type");
- goto end_nosignal;
- }
-
- DBG("Consumer %s socket created successfully with net idx %d (fd: %d)",
- msg.u.relayd_sock.type == LTTNG_STREAM_CONTROL ? "control" : "data",
- relayd->net_seq_idx, fd);
-
- /*
- * Add relayd socket pair to consumer data hashtable. If object already
- * exists or on error, the function gracefully returns.
- */
- consumer_add_relayd(relayd);
-
+ ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
+ msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
+ &msg.u.relayd_sock.sock);
goto end_nosignal;
}
case LTTNG_CONSUMER_ADD_CHANNEL:
/* block */
if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
+ rcu_read_unlock();
return -EINTR;
}
ret = lttcomm_recv_fds_unix_sock(sock, fds, nb_fd);
if (ret != sizeof(fds)) {
- lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
+ lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
+ rcu_read_unlock();
return ret;
}
msg.u.channel.mmap_len,
msg.u.channel.max_sb_size);
if (new_channel == NULL) {
- lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
+ lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
goto end_nosignal;
}
if (ctx->on_recv_channel != NULL) {
/* block */
if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
+ rcu_read_unlock();
return -EINTR;
}
ret = lttcomm_recv_fds_unix_sock(sock, fds, nb_fd);
if (ret != sizeof(fds)) {
- lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
+ lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
+ rcu_read_unlock();
return ret;
}
msg.u.stream.net_index,
msg.u.stream.metadata_flag);
if (new_stream == NULL) {
- lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
- goto end;
+ lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
+ goto end_nosignal;
}
/* The stream is not metadata. Get relayd reference if exists. */
&new_stream->relayd_stream_id);
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
- goto end;
+ goto end_nosignal;
}
} else if (msg.u.stream.net_index != -1) {
ERR("Network sequence index %d unknown. Not adding stream.",
msg.u.stream.net_index);
free(new_stream);
- goto end;
+ goto end_nosignal;
}
if (ctx->on_recv_stream != NULL) {
if (ret == 0) {
consumer_add_stream(new_stream);
} else if (ret < 0) {
- goto end;
+ goto end_nosignal;
}
} else {
consumer_add_stream(new_stream);
}
- DBG("UST consumer_add_stream %s (%d,%d) with relayd id %lu",
+ DBG("UST consumer_add_stream %s (%d,%d) with relayd id %" PRIu64,
msg.u.stream.path_name, fds[0], fds[1],
new_stream->relayd_stream_id);
break;
}
case LTTNG_CONSUMER_DESTROY_RELAYD:
{
+ uint64_t index = msg.u.destroy_relayd.net_seq_idx;
struct consumer_relayd_sock_pair *relayd;
- DBG("UST consumer destroying relayd %zu",
- msg.u.destroy_relayd.net_seq_idx);
+ DBG("UST consumer destroying relayd %" PRIu64, index);
/* Get relayd reference if exists. */
- relayd = consumer_find_relayd(msg.u.destroy_relayd.net_seq_idx);
+ relayd = consumer_find_relayd(index);
if (relayd == NULL) {
- ERR("Unable to find relayd %zu",
- msg.u.destroy_relayd.net_seq_idx);
+ ERR("Unable to find relayd %" PRIu64, index);
+ goto end_nosignal;
}
- /* Set destroy flag for this object */
- uatomic_set(&relayd->destroy_flag, 1);
+ /*
+ * Each relayd socket pair has a refcount of stream attached to it
+ * which tells if the relayd is still active or not depending on the
+ * refcount value.
+ *
+ * This will set the destroy flag of the relayd object and destroy it
+ * if the refcount reaches zero when called.
+ *
+ * The destroy can happen either here or when a stream fd hangs up.
+ */
+ consumer_flag_relayd_for_destroy(relayd);
- /* Destroy the relayd if refcount is 0 else set the destroy flag. */
- if (uatomic_read(&relayd->refcount) == 0) {
- consumer_destroy_relayd(relayd);
- }
- break;
+ goto end_nosignal;
}
case LTTNG_CONSUMER_UPDATE_STREAM:
{
+ rcu_read_unlock();
return -ENOSYS;
#if 0
if (ctx->on_update_stream != NULL) {
consumer_change_stream_state(msg.u.stream.stream_key,
msg.u.stream.state);
}
-#endif
break;
+#endif
}
default:
break;
}
-end:
+
/*
- * Wake-up the other end by writing a null byte in the pipe
- * (non-blocking). Important note: Because writing into the
- * pipe is non-blocking (and therefore we allow dropping wakeup
- * data, as long as there is wakeup data present in the pipe
- * buffer to wake up the other end), the other end should
- * perform the following sequence for waiting:
+ * Wake-up the other end by writing a null byte in the pipe (non-blocking).
+ * Important note: Because writing into the pipe is non-blocking (and
+ * therefore we allow dropping wakeup data, as long as there is wakeup data
+ * present in the pipe buffer to wake up the other end), the other end
+ * should perform the following sequence for waiting:
+ *
* 1) empty the pipe (reads).
* 2) perform update operation.
* 3) wait on the pipe (poll).
} while (ret < 0 && errno == EINTR);
end_nosignal:
rcu_read_unlock();
- return 0;
+
+ /*
+ * Return 1 to indicate success since the 0 value can be a socket
+ * shutdown during the recv() or send() call.
+ */
+ return 1;
}
int lttng_ustconsumer_allocate_channel(struct lttng_consumer_channel *chan)
* display the error but continue processing to try
* to release the subbuffer
*/
- ERR("Error writing to tracefile");
+ ERR("Error writing to tracefile (expected: %ld, got: %ld)", ret, len);
}
err = ustctl_put_next_subbuf(handle, buf);
assert(err == 0);
projects / lttng-tools.git / blobdiff