#include <sys/mman.h>
#include <sys/socket.h>
#include <sys/types.h>
+#include <inttypes.h>
#include <unistd.h>
#include <sys/stat.h>
extern int consumer_poll_timeout;
extern volatile int consumer_quit;
-/*
- * Mmap the ring buffer, read it and write the data to the tracefile.
- *
- * Returns the number of bytes written
- */
-ssize_t lttng_kconsumer_on_read_subbuffer_mmap(
- struct lttng_consumer_local_data *ctx,
- struct lttng_consumer_stream *stream, unsigned long len)
-{
- unsigned long mmap_offset;
- ssize_t ret = 0, written = 0;
- off_t orig_offset = stream->out_fd_offset;
- int fd = stream->wait_fd;
- /* Default is on the disk */
- 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) {
- goto end;
- }
- }
-
- /* get the offset inside the fd to mmap */
- ret = kernctl_get_mmap_read_offset(fd, &mmap_offset);
- if (ret != 0) {
- errno = -ret;
- perror("kernctl_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;
-
- /*
- * Lock the control socket for the complete duration of the function
- * since from this point on we will use the socket.
- */
- if (stream->metadata_flag) {
- /* Metadata requires 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) {
- /* Use the returned socket. */
- 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);
- /*
- * We do this so the return value can match the len passed as
- * argument to this function.
- */
- written -= sizeof(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("Error in file write");
- written += ret;
- goto end;
- } else {
- len -= ret;
- mmap_offset += ret;
- }
-
- /* 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:
- /* Unlock only if ctrl socket used */
- 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.
- */
-ssize_t lttng_kconsumer_on_read_subbuffer_splice(
- struct lttng_consumer_local_data *ctx,
- struct lttng_consumer_stream *stream, unsigned long len)
-{
- ssize_t ret = 0, written = 0, ret_splice = 0;
- loff_t offset = 0;
- off_t orig_offset = stream->out_fd_offset;
- int fd = stream->wait_fd;
- /* Default is on the disk */
- 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) {
- goto end;
- }
- }
-
- /* Write metadata stream id before payload */
- if (stream->metadata_flag && relayd) {
- /*
- * Lock the control socket for the complete duration of the function
- * since from this point on we will use the socket.
- */
- pthread_mutex_lock(&relayd->ctrl_sock_mutex);
-
- metadata_id = htobe64(stream->relayd_stream_id);
- do {
- ret = write(ctx->consumer_thread_pipe[1],
- (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);
- }
-
- while (len > 0) {
- DBG("splice chan to pipe offset %lu of len %lu (fd : %d)",
- (unsigned long)offset, len, fd);
- ret_splice = splice(fd, &offset, ctx->consumer_thread_pipe[1], NULL, len,
- SPLICE_F_MOVE | SPLICE_F_MORE);
- DBG("splice chan to pipe, ret %zd", ret_splice);
- if (ret_splice < 0) {
- perror("Error in relay splice");
- if (written == 0) {
- written = ret_splice;
- }
- ret = errno;
- goto splice_error;
- }
-
- /* Handle stream on the relayd if the output is on the network */
- if (relayd) {
- if (stream->metadata_flag) {
- /* Update counter to fit the spliced data */
- ret_splice += sizeof(stream->relayd_stream_id);
- len += sizeof(stream->relayd_stream_id);
- /*
- * We do this so the return value can match the len passed as
- * argument to this function.
- */
- written -= sizeof(stream->relayd_stream_id);
- }
-
- ret = consumer_handle_stream_before_relayd(stream, ret_splice);
- if (ret >= 0) {
- /* Use the returned socket. */
- outfd = ret;
- } else {
- if (outfd == -1) {
- ERR("Remote relayd disconnected. Stopping");
- goto end;
- }
- }
- }
-
- DBG3("Kernel consumer splice data in %d to out %d",
- ctx->consumer_thread_pipe[0], outfd);
- ret_splice = splice(ctx->consumer_thread_pipe[0], NULL, outfd, NULL,
- ret_splice, SPLICE_F_MOVE | SPLICE_F_MORE);
- DBG("splice pipe to file, ret %zd", ret_splice);
- if (ret_splice < 0) {
- perror("Error in file splice");
- if (written == 0) {
- written = ret_splice;
- }
- ret = errno;
- goto splice_error;
- }
- if (ret_splice > len) {
- errno = EINVAL;
- PERROR("Wrote more data than requested %zd (len: %lu)",
- ret_splice, len);
- written += ret_splice;
- ret = errno;
- goto splice_error;
- }
- len -= ret_splice;
-
- /* 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_splice,
- SYNC_FILE_RANGE_WRITE);
- stream->out_fd_offset += ret_splice;
- }
- written += ret_splice;
- }
- lttng_consumer_sync_trace_file(stream, orig_offset);
-
- ret = ret_splice;
-
- goto end;
-
-splice_error:
- /* send the appropriate error description to sessiond */
- switch (ret) {
- case EBADF:
- lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_EBADF);
- break;
- case EINVAL:
- lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_EINVAL);
- break;
- case ENOMEM:
- lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_ENOMEM);
- break;
- case ESPIPE:
- lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_ESPIPE);
- break;
- }
-
-end:
- if (relayd && stream->metadata_flag) {
- pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
- }
-
- rcu_read_unlock();
-
- return written;
-}
-
/*
* Take a snapshot for a specific fd
*
switch (msg.cmd_type) {
case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
{
- int fd;
- struct consumer_relayd_sock_pair *relayd;
-
- DBG("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, &fd, 1);
if (ret != sizeof(fd)) {
lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
+ rcu_read_unlock();
return ret;
}
msg.u.stream.metadata_flag);
if (new_stream == NULL) {
lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
- goto end;
+ 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);
}
case LTTNG_CONSUMER_UPDATE_STREAM:
{
- if (ctx->on_update_stream != NULL) {
- ret = ctx->on_update_stream(msg.u.stream.stream_key, msg.u.stream.state);
- if (ret == 0) {
- consumer_change_stream_state(msg.u.stream.stream_key, msg.u.stream.state);
- } else if (ret < 0) {
- goto end;
- }
- } else {
- consumer_change_stream_state(msg.u.stream.stream_key,
- msg.u.stream.state);
+ rcu_read_unlock();
+ return -ENOSYS;
+ }
+ case LTTNG_CONSUMER_DESTROY_RELAYD:
+ {
+ struct consumer_relayd_sock_pair *relayd;
+
+ DBG("Kernel consumer destroying relayd %" PRIu64,
+ msg.u.destroy_relayd.net_seq_idx);
+
+ /* Get relayd reference if exists. */
+ relayd = consumer_find_relayd(msg.u.destroy_relayd.net_seq_idx);
+ if (relayd == NULL) {
+ ERR("Unable to find relayd %" PRIu64,
+ msg.u.destroy_relayd.net_seq_idx);
+ goto end_nosignal;
}
- break;
+
+ /* Set destroy flag for this object */
+ uatomic_set(&relayd->destroy_flag, 1);
+
+ /* Destroy the relayd if refcount is 0 else set the destroy flag. */
+ if (uatomic_read(&relayd->refcount) == 0) {
+ consumer_destroy_relayd(relayd);
+ }
+ goto end_nosignal;
}
default:
- break;
+ goto end_nosignal;
}
-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).
* display the error but continue processing to try
* to release the subbuffer
*/
- ERR("Error splicing to tracefile (ret: %ld != len: %ld)",
+ ERR("Error splicing to tracefile (ret: %zd != len: %lu)",
ret, len);
}
projects / lttng-tools.git / blobdiff