X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=d70a5d1d406b758e1b02cc61f22b440ee006a7b1;hp=26209279ee65a49ec8c06ba88baed8dd275da90e;hb=fd424d99bfddf3581bc55fdf14799a50661beddd;hpb=f8528c7a982be6612eb1d81f17e1badd8910dfd4 diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c index 26209279e..d70a5d1d4 100644 --- a/src/common/consumer/consumer.c +++ b/src/common/consumer/consumer.c @@ -1,20 +1,10 @@ /* - * Copyright (C) 2011 - Julien Desfossez - * Mathieu Desnoyers - * 2012 - David Goulet + * Copyright (C) 2011 Julien Desfossez + * Copyright (C) 2011 Mathieu Desnoyers + * Copyright (C) 2012 David Goulet * - * This program is free software; you can redistribute it and/or modify - * it under the terms of the GNU General Public License, version 2 only, - * as published by the Free Software Foundation. + * SPDX-License-Identifier: GPL-2.0-only * - * This program is distributed in the hope that it will be useful, but WITHOUT - * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or - * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for - * more details. - * - * You should have received a copy of the GNU General Public License along - * with this program; if not, write to the Free Software Foundation, Inc., - * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ #define _LGPL_SOURCE @@ -1680,18 +1670,19 @@ end: */ ssize_t lttng_consumer_on_read_subbuffer_mmap( struct lttng_consumer_local_data *ctx, - struct lttng_consumer_stream *stream, unsigned long len, + struct lttng_consumer_stream *stream, + const struct lttng_buffer_view *buffer, unsigned long padding, struct ctf_packet_index *index) { - unsigned long mmap_offset; - void *mmap_base; ssize_t ret = 0; off_t orig_offset = stream->out_fd_offset; /* Default is on the disk */ int outfd = stream->out_fd; struct consumer_relayd_sock_pair *relayd = NULL; unsigned int relayd_hang_up = 0; + const size_t subbuf_content_size = buffer->size - padding; + size_t write_len; /* RCU lock for the relayd pointer */ rcu_read_lock(); @@ -1707,39 +1698,9 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap( } } - /* get the offset inside the fd to mmap */ - switch (consumer_data.type) { - case LTTNG_CONSUMER_KERNEL: - mmap_base = stream->mmap_base; - ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset); - if (ret < 0) { - PERROR("tracer ctl get_mmap_read_offset"); - goto end; - } - break; - case LTTNG_CONSUMER32_UST: - case LTTNG_CONSUMER64_UST: - mmap_base = lttng_ustctl_get_mmap_base(stream); - if (!mmap_base) { - ERR("read mmap get mmap base for stream %s", stream->name); - ret = -EPERM; - goto end; - } - ret = lttng_ustctl_get_mmap_read_offset(stream, &mmap_offset); - if (ret != 0) { - PERROR("tracer ctl get_mmap_read_offset"); - ret = -EINVAL; - goto end; - } - break; - default: - ERR("Unknown consumer_data type"); - assert(0); - } - /* Handle stream on the relayd if the output is on the network */ if (relayd) { - unsigned long netlen = len; + unsigned long netlen = subbuf_content_size; /* * Lock the control socket for the complete duration of the function @@ -1777,10 +1738,10 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap( goto write_error; } } - } else { - /* No streaming, we have to set the len with the full padding */ - len += padding; + write_len = subbuf_content_size; + } else { + /* No streaming; we have to write the full padding. */ if (stream->metadata_flag && stream->reset_metadata_flag) { ret = utils_truncate_stream_file(stream->out_fd, 0); if (ret < 0) { @@ -1794,7 +1755,7 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap( * Check if we need to change the tracefile before writing the packet. */ if (stream->chan->tracefile_size > 0 && - (stream->tracefile_size_current + len) > + (stream->tracefile_size_current + buffer->size) > stream->chan->tracefile_size) { ret = consumer_stream_rotate_output_files(stream); if (ret) { @@ -1803,19 +1764,21 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap( outfd = stream->out_fd; orig_offset = 0; } - stream->tracefile_size_current += len; + stream->tracefile_size_current += buffer->size; if (index) { index->offset = htobe64(stream->out_fd_offset); } + + write_len = buffer->size; } /* * This call guarantee that len or less is returned. It's impossible to * receive a ret value that is bigger than len. */ - ret = lttng_write(outfd, mmap_base + mmap_offset, len); - DBG("Consumer mmap write() ret %zd (len %lu)", ret, len); - if (ret < 0 || ((size_t) ret != len)) { + ret = lttng_write(outfd, buffer->data, write_len); + DBG("Consumer mmap write() ret %zd (len %lu)", ret, write_len); + if (ret < 0 || ((size_t) ret != write_len)) { /* * Report error to caller if nothing was written else at least send the * amount written. @@ -1836,7 +1799,8 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap( DBG("Consumer mmap write detected relayd hang up"); } else { /* Unhandled error, print it and stop function right now. */ - PERROR("Error in write mmap (ret %zd != len %lu)", ret, len); + PERROR("Error in write mmap (ret %zd != write_len %zu)", ret, + write_len); } goto write_error; } @@ -1845,9 +1809,9 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap( /* 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, len, + lttng_sync_file_range(outfd, stream->out_fd_offset, write_len, SYNC_FILE_RANGE_WRITE); - stream->out_fd_offset += len; + stream->out_fd_offset += write_len; lttng_consumer_sync_trace_file(stream, orig_offset); } @@ -3963,8 +3927,18 @@ int consumer_flush_buffer(struct lttng_consumer_stream *stream, int producer_act } else { ret = kernctl_buffer_flush_empty(stream->wait_fd); if (ret < 0) { - ERR("Failed to flush kernel stream"); - goto end; + /* + * Doing a buffer flush which does not take into + * account empty packets. This is not perfect, + * but required as a fall-back when + * "flush_empty" is not implemented by + * lttng-modules. + */ + ret = kernctl_buffer_flush(stream->wait_fd); + if (ret < 0) { + ERR("Failed to flush kernel stream"); + goto end; + } } } break; @@ -4203,7 +4177,7 @@ int consumer_clear_buffer(struct lttng_consumer_stream *stream) case LTTNG_CONSUMER_KERNEL: ret = kernctl_buffer_clear(stream->wait_fd); if (ret < 0) { - ERR("Failed to flush kernel stream"); + ERR("Failed to clear kernel stream (ret = %d)", ret); goto end; } break; @@ -4282,11 +4256,6 @@ error_unlock: pthread_mutex_unlock(&stream->lock); pthread_mutex_unlock(&channel->lock); rcu_read_unlock(); - if (ret) { - goto error; - } - ret = LTTCOMM_CONSUMERD_SUCCESS; -error: return ret; } @@ -4603,7 +4572,7 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk( * and LTTNG_CONSUMER_DESTROY_TRACE_CHUNK commands. */ created_chunk = lttng_trace_chunk_create(chunk_id, - chunk_creation_timestamp); + chunk_creation_timestamp, NULL); if (!created_chunk) { ERR("Failed to create trace chunk"); ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;