src/common/compat/Makefile
src/common/relayd/Makefile
src/common/testpoint/Makefile
+ src/common/index/Makefile
src/lib/Makefile
src/lib/lttng-ctl/Makefile
src/lib/lttng-ctl/filter/Makefile
$(top_builddir)/src/common/libconsumer.la \
$(top_builddir)/src/common/sessiond-comm/libsessiond-comm.la \
$(top_builddir)/src/common/libcommon.la \
+ $(top_builddir)/src/common/index/libindex.la \
-lrt
if HAVE_LIBLTTNG_UST_CTL
struct relay_session *session;
struct rcu_head rcu_node;
int fd;
+ int index_fd;
char *path_name;
char *channel_name;
* uses its own credentials for the stream files.
*/
ret = utils_create_stream_file(stream->path_name, stream->channel_name,
- stream->tracefile_size, 0, -1, -1);
+ stream->tracefile_size, 0, -1, -1, NULL);
if (ret < 0) {
ERR("Create output file");
goto end;
ret = utils_rotate_stream_file(stream->path_name,
stream->channel_name, stream->tracefile_size,
stream->tracefile_count, -1, -1,
- stream->fd, &(stream->tracefile_count_current));
+ stream->fd, &(stream->tracefile_count_current),
+ &stream->fd);
if (ret < 0) {
ERR("Rotating output file");
goto end;
AM_CPPFLAGS = -I$(top_srcdir)/include -I$(top_srcdir)/src
SUBDIRS = compat hashtable kernel-ctl sessiond-comm relayd \
- kernel-consumer ust-consumer testpoint
+ kernel-consumer ust-consumer testpoint index
AM_CFLAGS = -fno-strict-aliasing
stream->out_fd = -1;
}
+ if (stream->index_fd >= 0) {
+ ret = close(stream->index_fd);
+ if (ret) {
+ PERROR("close stream index_fd");
+ }
+ stream->index_fd = -1;
+ }
+
/* Check and cleanup relayd if needed. */
rcu_read_lock();
relayd = consumer_find_relayd(stream->net_seq_idx);
#include <common/common.h>
#include <common/utils.h>
#include <common/compat/poll.h>
+#include <common/index/index.h>
#include <common/kernel-ctl/kernel-ctl.h>
#include <common/sessiond-comm/relayd.h>
#include <common/sessiond-comm/sessiond-comm.h>
stream->session_id = session_id;
stream->monitor = monitor;
stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
+ stream->index_fd = -1;
pthread_mutex_init(&stream->lock, NULL);
/* If channel is the metadata, flag this stream as metadata. */
ssize_t lttng_consumer_on_read_subbuffer_mmap(
struct lttng_consumer_local_data *ctx,
struct lttng_consumer_stream *stream, unsigned long len,
- unsigned long padding)
+ unsigned long padding,
+ struct lttng_packet_index *index)
{
unsigned long mmap_offset;
void *mmap_base;
ret = utils_rotate_stream_file(stream->chan->pathname,
stream->name, stream->chan->tracefile_size,
stream->chan->tracefile_count, stream->uid, stream->gid,
- stream->out_fd, &(stream->tracefile_count_current));
+ stream->out_fd, &(stream->tracefile_count_current),
+ &stream->out_fd);
if (ret < 0) {
ERR("Rotating output file");
goto end;
}
- outfd = stream->out_fd = ret;
+ outfd = stream->out_fd;
+
+ if (stream->index_fd >= 0) {
+ ret = index_create_file(stream->chan->pathname,
+ stream->name, stream->uid, stream->gid,
+ stream->chan->tracefile_size,
+ stream->tracefile_count_current);
+ if (ret < 0) {
+ goto end;
+ }
+ stream->index_fd = ret;
+ }
+
/* Reset current size because we just perform a rotation. */
stream->tracefile_size_current = 0;
stream->out_fd_offset = 0;
orig_offset = 0;
}
stream->tracefile_size_current += len;
+ if (index) {
+ index->offset = htobe64(stream->out_fd_offset);
+ }
}
while (len > 0) {
ssize_t lttng_consumer_on_read_subbuffer_splice(
struct lttng_consumer_local_data *ctx,
struct lttng_consumer_stream *stream, unsigned long len,
- unsigned long padding)
+ unsigned long padding,
+ struct lttng_packet_index *index)
{
ssize_t ret = 0, written = 0, ret_splice = 0;
loff_t offset = 0;
ret = utils_rotate_stream_file(stream->chan->pathname,
stream->name, stream->chan->tracefile_size,
stream->chan->tracefile_count, stream->uid, stream->gid,
- stream->out_fd, &(stream->tracefile_count_current));
+ stream->out_fd, &(stream->tracefile_count_current),
+ &stream->out_fd);
if (ret < 0) {
ERR("Rotating output file");
goto end;
}
- outfd = stream->out_fd = ret;
+ outfd = stream->out_fd;
+
+ if (stream->index_fd >= 0) {
+ ret = index_create_file(stream->chan->pathname,
+ stream->name, stream->uid, stream->gid,
+ stream->chan->tracefile_size,
+ stream->tracefile_count_current);
+ if (ret < 0) {
+ goto end;
+ }
+ stream->index_fd = ret;
+ }
+
/* Reset current size because we just perform a rotation. */
stream->tracefile_size_current = 0;
stream->out_fd_offset = 0;
orig_offset = 0;
}
stream->tracefile_size_current += len;
+ index->offset = htobe64(stream->out_fd_offset);
}
while (len > 0) {
#include <common/compat/uuid.h>
#include <common/sessiond-comm/sessiond-comm.h>
#include <common/pipe.h>
+#include <common/index/lttng-index.h>
/* Commands for consumer */
enum lttng_consumer_command {
* to the channel.
*/
uint64_t ust_metadata_pushed;
+ /*
+ * FD of the index file for this stream.
+ */
+ int index_fd;
};
/*
ssize_t lttng_consumer_on_read_subbuffer_mmap(
struct lttng_consumer_local_data *ctx,
struct lttng_consumer_stream *stream, unsigned long len,
- unsigned long padding);
+ unsigned long padding,
+ struct lttng_packet_index *index);
ssize_t lttng_consumer_on_read_subbuffer_splice(
struct lttng_consumer_local_data *ctx,
struct lttng_consumer_stream *stream, unsigned long len,
- unsigned long padding);
+ unsigned long padding,
+ struct lttng_packet_index *index);
int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream);
int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
unsigned long *pos);
void consumer_del_stream_for_data(struct lttng_consumer_stream *stream);
int consumer_add_metadata_stream(struct lttng_consumer_stream *stream);
void consumer_del_stream_for_metadata(struct lttng_consumer_stream *stream);
+int consumer_create_index_file(struct lttng_consumer_stream *stream);
#endif /* LIB_CONSUMER_H */
#define DEFAULT_SNAPSHOT_NAME "snapshot"
#define DEFAULT_SNAPSHOT_MAX_SIZE 0 /* Unlimited. */
+/* Suffix of an index file. */
+#define DEFAULT_INDEX_FILE_SUFFIX ".idx"
+
extern size_t default_channel_subbuf_size;
extern size_t default_metadata_subbuf_size;
extern size_t default_ust_pid_channel_subbuf_size;
--- /dev/null
+AM_CPPFLAGS = -I$(top_srcdir)/include -I$(top_srcdir)/src
+
+noinst_LTLIBRARIES = libindex.la
+
+libindex_la_SOURCES = index.c index.h lttng-index.h
--- /dev/null
+/*
+ * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
+ * David Goulet <dgoulet@efficios.com>
+ *
+ * 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.
+ *
+ * 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 _GNU_SOURCE
+#include <assert.h>
+
+#include <common/common.h>
+#include <common/defaults.h>
+#include <common/utils.h>
+
+#include "index.h"
+
+/*
+ * Create the index file associated with a trace file.
+ *
+ * Return fd on success, a negative value on error.
+ */
+int index_create_file(char *path_name, char *stream_name, int uid, int gid,
+ uint64_t size, uint64_t count)
+{
+ int ret, fd = -1;
+ struct lttng_packet_index_file_hdr hdr;
+
+ ret = utils_create_stream_file(path_name, stream_name, size, count, uid,
+ gid, DEFAULT_INDEX_FILE_SUFFIX);
+ if (ret < 0) {
+ goto error;
+ }
+ fd = ret;
+
+ memcpy(hdr.magic, INDEX_MAGIC, sizeof(hdr.magic));
+ hdr.index_major = htobe32(INDEX_MAJOR);
+ hdr.index_minor = htobe32(INDEX_MINOR);
+
+ do {
+ ret = write(fd, &hdr, sizeof(hdr));
+ } while (ret < 0 && errno == EINTR);
+ if (ret < 0) {
+ PERROR("write index header");
+ goto error;
+ }
+
+ return fd;
+
+error:
+ if (fd >= 0) {
+ int close_ret;
+
+ close_ret = close(fd);
+ if (close_ret < 0) {
+ PERROR("close index fd");
+ }
+ }
+ return ret;
+}
+
+/*
+ * Write index values to the given fd of size len.
+ *
+ * Return 0 on success or else a negative value on error.
+ */
+int index_write(int fd, struct lttng_packet_index *index, size_t len)
+{
+ int ret;
+
+ assert(fd >= 0);
+ assert(index);
+
+ do {
+ ret = write(fd, index, len);
+ } while (ret < 0 && errno == EINTR);
+ if (ret < 0) {
+ PERROR("writing index file");
+ }
+
+ return ret;
+}
--- /dev/null
+/*
+ * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
+ * David Goulet <dgoulet@efficios.com>
+ *
+ * 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.
+ *
+ * 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.
+ */
+
+#ifndef _INDEX_H
+#define _INDEX_H
+
+#include <inttypes.h>
+
+#include "lttng-index.h"
+
+int index_create_file(char *path_name, char *stream_name, int uid, int gid,
+ uint64_t size, uint64_t count);
+int index_write(int fd, struct lttng_packet_index *index, size_t len);
+
+#endif /* _INDEX_H */
--- /dev/null
+/*
+ * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
+ * David Goulet <dgoulet@efficios.com>
+ *
+ * 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.
+ *
+ * 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.
+ */
+
+#ifndef LTTNG_INDEX_H
+#define LTTNG_INDEX_H
+
+#include <limits.h>
+
+#define INDEX_MAGIC "CTFIDX"
+#define INDEX_MAJOR 1
+#define INDEX_MINOR 0
+
+/*
+ * Header at the beginning of each index file.
+ * All integer fields are stored in big endian.
+ */
+struct lttng_packet_index_file_hdr {
+ char magic[6];
+ uint32_t index_major;
+ uint32_t index_minor;
+} __attribute__((__packed__));
+
+/*
+ * Packet index generated for each trace packet store in a trace file.
+ * All integer fields are stored in big endian.
+ */
+struct lttng_packet_index {
+ uint64_t offset; /* offset of the packet in the file, in bytes */
+ uint64_t packet_size; /* packet size, in bits */
+ uint64_t content_size; /* content size, in bits */
+ uint64_t timestamp_begin;
+ uint64_t timestamp_end;
+ uint64_t events_discarded;
+ uint64_t stream_id;
+} __attribute__((__packed__));
+
+#endif /* LTTNG_INDEX_H */
#include <common/relayd/relayd.h>
#include <common/utils.h>
#include <common/consumer-stream.h>
+#include <common/index/index.h>
#include "kernel-consumer.h"
ret = utils_create_stream_file(path, stream->name,
stream->chan->tracefile_size,
stream->tracefile_count_current,
- stream->uid, stream->gid);
+ stream->uid, stream->gid, NULL);
if (ret < 0) {
ERR("utils_create_stream_file");
goto end_unlock;
}
read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len,
- padded_len - len);
+ padded_len - len, NULL);
/*
* We write the padded len in local tracefiles but the data len
* when using a relay. Display the error but continue processing
ret = utils_create_stream_file(path, metadata_stream->name,
metadata_stream->chan->tracefile_size,
metadata_stream->tracefile_count_current,
- metadata_stream->uid, metadata_stream->gid);
+ metadata_stream->uid, metadata_stream->gid, NULL);
if (ret < 0) {
goto error;
}
return -1;
}
+/*
+ * Populate index values of a kernel stream. Values are set in big endian order.
+ *
+ * Return 0 on success or else a negative value.
+ */
+static int get_index_values(struct lttng_packet_index *index, int infd)
+{
+ int ret;
+
+ ret = kernctl_get_timestamp_begin(infd, &index->timestamp_begin);
+ if (ret < 0) {
+ PERROR("kernctl_get_timestamp_begin");
+ goto error;
+ }
+ index->timestamp_begin = htobe64(index->timestamp_begin);
+
+ ret = kernctl_get_timestamp_end(infd, &index->timestamp_end);
+ if (ret < 0) {
+ PERROR("kernctl_get_timestamp_end");
+ goto error;
+ }
+ index->timestamp_end = htobe64(index->timestamp_end);
+
+ ret = kernctl_get_events_discarded(infd, &index->events_discarded);
+ if (ret < 0) {
+ PERROR("kernctl_get_events_discarded");
+ goto error;
+ }
+ index->events_discarded = htobe64(index->events_discarded);
+
+ ret = kernctl_get_content_size(infd, &index->content_size);
+ if (ret < 0) {
+ PERROR("kernctl_get_content_size");
+ goto error;
+ }
+ index->content_size = htobe64(index->content_size);
+
+ ret = kernctl_get_packet_size(infd, &index->packet_size);
+ if (ret < 0) {
+ PERROR("kernctl_get_packet_size");
+ goto error;
+ }
+ index->packet_size = htobe64(index->packet_size);
+
+ ret = kernctl_get_stream_id(infd, &index->stream_id);
+ if (ret < 0) {
+ PERROR("kernctl_get_stream_id");
+ goto error;
+ }
+ index->stream_id = htobe64(index->stream_id);
+
+error:
+ return ret;
+}
+
/*
* Consume data on a file descriptor and write it on a trace file.
*/
struct lttng_consumer_local_data *ctx)
{
unsigned long len, subbuf_size, padding;
- int err;
+ int err, write_index = 0;
ssize_t ret = 0;
int infd = stream->wait_fd;
+ struct lttng_packet_index index;
DBG("In read_subbuffer (infd : %d)", infd);
+
+ /* Indicate that for this stream we have to write the index. */
+ if (stream->index_fd >= 0) {
+ write_index = 1;
+ }
+
/* Get the next subbuffer */
err = kernctl_get_next_subbuf(infd);
if (err != 0) {
goto end;
}
+ if (!stream->metadata_flag && write_index) {
+ ret = get_index_values(&index, infd);
+ if (ret < 0) {
+ goto end;
+ }
+ }
+
switch (stream->chan->output) {
case CONSUMER_CHANNEL_SPLICE:
/*
/* splice the subbuffer to the tracefile */
ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, subbuf_size,
- padding);
+ padding, &index);
/*
* XXX: Splice does not support network streaming so the return value
* is simply checked against subbuf_size and not like the mmap() op.
*/
ERR("Error splicing to tracefile (ret: %zd != len: %lu)",
ret, subbuf_size);
+ write_index = 0;
}
break;
case CONSUMER_CHANNEL_MMAP:
/* write the subbuffer to the tracefile */
ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size,
- padding);
+ padding, &index);
/*
* The mmap operation should write subbuf_size amount of data when
* network streaming or the full padding (len) size when we are _not_
ERR("Error writing to tracefile "
"(ret: %zd != len: %lu != subbuf_size: %lu)",
ret, len, subbuf_size);
+ write_index = 0;
}
break;
default:
goto end;
}
+ /* Write index if needed. */
+ if (write_index) {
+ err = index_write(stream->index_fd, &index, sizeof(index));
+ if (err < 0) {
+ ret = -1;
+ goto end;
+ }
+ }
+
end:
return ret;
}
if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) {
ret = utils_create_stream_file(stream->chan->pathname, stream->name,
stream->chan->tracefile_size, stream->tracefile_count_current,
- stream->uid, stream->gid);
+ stream->uid, stream->gid, NULL);
if (ret < 0) {
goto error;
}
stream->out_fd = ret;
stream->tracefile_size_current = 0;
+
+ if (!stream->metadata_flag) {
+ ret = index_create_file(stream->chan->pathname,
+ stream->name, stream->uid, stream->gid,
+ stream->chan->tracefile_size,
+ stream->tracefile_count_current);
+ if (ret < 0) {
+ goto error;
+ }
+ stream->index_fd = ret;
+ }
}
if (stream->output == LTTNG_EVENT_MMAP) {
{
return ioctl(fd, RING_BUFFER_PUT_SUBBUF);
}
+
+/* Returns the timestamp begin of the current sub-buffer. */
+int kernctl_get_timestamp_begin(int fd, uint64_t *timestamp_begin)
+{
+ return ioctl(fd, LTTNG_RING_BUFFER_GET_TIMESTAMP_BEGIN, timestamp_begin);
+}
+
+/* Returns the timestamp end of the current sub-buffer. */
+int kernctl_get_timestamp_end(int fd, uint64_t *timestamp_end)
+{
+ return ioctl(fd, LTTNG_RING_BUFFER_GET_TIMESTAMP_END, timestamp_end);
+}
+
+/* Returns the number of discarded events in the current sub-buffer. */
+int kernctl_get_events_discarded(int fd, uint64_t *events_discarded)
+{
+ return ioctl(fd, LTTNG_RING_BUFFER_GET_EVENTS_DISCARDED, events_discarded);
+}
+
+/* Returns the content size in the current sub-buffer. */
+int kernctl_get_content_size(int fd, uint64_t *content_size)
+{
+ return ioctl(fd, LTTNG_RING_BUFFER_GET_CONTENT_SIZE, content_size);
+}
+
+/* Returns the packet size in the current sub-buffer. */
+int kernctl_get_packet_size(int fd, uint64_t *packet_size)
+{
+ return ioctl(fd, LTTNG_RING_BUFFER_GET_PACKET_SIZE, packet_size);
+}
+
+/* Returns the stream id of the current sub-buffer. */
+int kernctl_get_stream_id(int fd, uint64_t *stream_id)
+{
+ return ioctl(fd, LTTNG_RING_BUFFER_GET_STREAM_ID, stream_id);
+}
int kernctl_buffer_flush(int fd);
+/* index */
+int kernctl_get_timestamp_begin(int fd, uint64_t *timestamp_begin);
+int kernctl_get_timestamp_end(int fd, uint64_t *timestamp_end);
+int kernctl_get_events_discarded(int fd, uint64_t *events_discarded);
+int kernctl_get_content_size(int fd, uint64_t *content_size);
+int kernctl_get_packet_size(int fd, uint64_t *packet_size);
+int kernctl_get_stream_id(int fd, uint64_t *stream_id);
+
#endif /* _LTTNG_KERNEL_CTL_H */
/* flush the current sub-buffer */
#define RING_BUFFER_FLUSH _IO(0xF6, 0x0C)
+/* returns the timestamp begin of the current sub-buffer */
+#define LTTNG_RING_BUFFER_GET_TIMESTAMP_BEGIN _IOR(0xF6, 0x20, uint64_t)
+/* returns the timestamp end of the current sub-buffer */
+#define LTTNG_RING_BUFFER_GET_TIMESTAMP_END _IOR(0xF6, 0x21, uint64_t)
+/* returns the number of events discarded */
+#define LTTNG_RING_BUFFER_GET_EVENTS_DISCARDED _IOR(0xF6, 0x22, uint64_t)
+/* returns the packet payload size */
+#define LTTNG_RING_BUFFER_GET_CONTENT_SIZE _IOR(0xF6, 0x23, uint64_t)
+/* returns the actual packet size */
+#define LTTNG_RING_BUFFER_GET_PACKET_SIZE _IOR(0xF6, 0x24, uint64_t)
+/* returns the stream id */
+#define LTTNG_RING_BUFFER_GET_STREAM_ID _IOR(0xF6, 0x25, uint64_t)
+
/* Old ABI (without support for 32/64 bits compat) */
/* LTTng file descriptor ioctl */
#define LTTNG_KERNEL_OLD_SESSION _IO(0xF6, 0x40)
#include <common/consumer-stream.h>
#include <common/consumer-timer.h>
#include <common/utils.h>
+#include <common/index/index.h>
#include "ust-consumer.h"
ret = utils_create_stream_file(path, metadata_stream->name,
metadata_stream->chan->tracefile_size,
metadata_stream->tracefile_count_current,
- metadata_stream->uid, metadata_stream->gid);
+ metadata_stream->uid, metadata_stream->gid, NULL);
if (ret < 0) {
goto error_stream;
}
ret = utils_create_stream_file(path, stream->name,
stream->chan->tracefile_size,
stream->tracefile_count_current,
- stream->uid, stream->gid);
+ stream->uid, stream->gid, NULL);
if (ret < 0) {
goto error_unlock;
}
}
read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len,
- padded_len - len);
+ padded_len - len, NULL);
if (use_relayd) {
if (read_len != len) {
ret = -EPERM;
ustctl_destroy_stream(stream->ustream);
}
+/*
+ * Populate index values of a UST stream. Values are set in big endian order.
+ *
+ * Return 0 on success or else a negative value.
+ */
+static int get_index_values(struct lttng_packet_index *index,
+ struct ustctl_consumer_stream *ustream)
+{
+ int ret;
+
+ ret = ustctl_get_timestamp_begin(ustream, &index->timestamp_begin);
+ if (ret < 0) {
+ PERROR("ustctl_get_timestamp_begin");
+ goto error;
+ }
+ index->timestamp_begin = htobe64(index->timestamp_begin);
+
+ ret = ustctl_get_timestamp_end(ustream, &index->timestamp_end);
+ if (ret < 0) {
+ PERROR("ustctl_get_timestamp_end");
+ goto error;
+ }
+ index->timestamp_end = htobe64(index->timestamp_end);
+
+ ret = ustctl_get_events_discarded(ustream, &index->events_discarded);
+ if (ret < 0) {
+ PERROR("ustctl_get_events_discarded");
+ goto error;
+ }
+ index->events_discarded = htobe64(index->events_discarded);
+
+ ret = ustctl_get_content_size(ustream, &index->content_size);
+ if (ret < 0) {
+ PERROR("ustctl_get_content_size");
+ goto error;
+ }
+ index->content_size = htobe64(index->content_size);
+
+ ret = ustctl_get_packet_size(ustream, &index->packet_size);
+ if (ret < 0) {
+ PERROR("ustctl_get_packet_size");
+ goto error;
+ }
+ index->packet_size = htobe64(index->packet_size);
+
+ ret = ustctl_get_stream_id(ustream, &index->stream_id);
+ if (ret < 0) {
+ PERROR("ustctl_get_stream_id");
+ goto error;
+ }
+ index->stream_id = htobe64(index->stream_id);
+
+error:
+ return ret;
+}
+
+
int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
struct lttng_consumer_local_data *ctx)
{
unsigned long len, subbuf_size, padding;
- int err;
+ int err, write_index = 0;
long ret = 0;
char dummy;
struct ustctl_consumer_stream *ustream;
+ struct lttng_packet_index index;
assert(stream);
assert(stream->ustream);
/* Ease our life for what's next. */
ustream = stream->ustream;
+ /* Indicate that for this stream we have to write the index. */
+ if (stream->index_fd >= 0) {
+ write_index = 1;
+ }
+
/* We can consume the 1 byte written into the wait_fd by UST */
if (stream->monitor && !stream->hangup_flush_done) {
ssize_t readlen;
goto end;
}
assert(stream->chan->output == CONSUMER_CHANNEL_MMAP);
+
+ if (!stream->metadata_flag && write_index) {
+ index.offset = htobe64(stream->out_fd_offset);
+ ret = get_index_values(&index, ustream);
+ if (ret < 0) {
+ goto end;
+ }
+ }
+
/* Get the full padded subbuffer size */
err = ustctl_get_padded_subbuf_size(ustream, &len);
assert(err == 0);
padding = len - subbuf_size;
/* write the subbuffer to the tracefile */
- ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, padding);
+ ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, padding, &index);
/*
* The mmap operation should write subbuf_size amount of data when network
* streaming or the full padding (len) size when we are _not_ streaming.
DBG("Error writing to tracefile "
"(ret: %ld != len: %lu != subbuf_size: %lu)",
ret, len, subbuf_size);
+ write_index = 0;
}
err = ustctl_put_next_subbuf(ustream);
assert(err == 0);
+ /* Write index if needed. */
+ if (write_index) {
+ err = index_write(stream->index_fd, &index, sizeof(index));
+ if (err < 0) {
+ ret = -1;
+ goto end;
+ }
+ }
+
end:
return ret;
}
if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) {
ret = utils_create_stream_file(stream->chan->pathname, stream->name,
stream->chan->tracefile_size, stream->tracefile_count_current,
- stream->uid, stream->gid);
+ stream->uid, stream->gid, NULL);
if (ret < 0) {
goto error;
}
stream->out_fd = ret;
stream->tracefile_size_current = 0;
+
+ if (!stream->metadata_flag) {
+ ret = index_create_file(stream->chan->pathname,
+ stream->name, stream->uid, stream->gid,
+ stream->chan->tracefile_size,
+ stream->tracefile_count_current);
+ if (ret < 0) {
+ goto error;
+ }
+ stream->index_fd = ret;
+ }
}
ret = 0;
*/
LTTNG_HIDDEN
int utils_create_stream_file(const char *path_name, char *file_name, uint64_t size,
- uint64_t count, int uid, int gid)
+ uint64_t count, int uid, int gid, char *suffix)
{
int ret, out_fd, flags, mode;
- char full_path[PATH_MAX], *path_name_id = NULL, *path;
+ char full_path[PATH_MAX], *path_name_suffix = NULL, *path;
+ char *extra = NULL;
assert(path_name);
assert(file_name);
goto error;
}
+ /* Setup extra string if suffix or/and a count is needed. */
+ if (size > 0 && suffix) {
+ ret = asprintf(&extra, "_%" PRIu64 "%s", count, suffix);
+ } else if (size > 0) {
+ ret = asprintf(&extra, "_%" PRIu64, count);
+ } else if (suffix) {
+ ret = asprintf(&extra, "%s", suffix);
+ }
+ if (ret < 0) {
+ PERROR("Allocating extra string to name");
+ goto error;
+ }
+
/*
* If we split the trace in multiple files, we have to add the count at the
* end of the tracefile name
*/
- if (size > 0) {
- ret = asprintf(&path_name_id, "%s_%" PRIu64, full_path, count);
+ if (extra) {
+ ret = asprintf(&path_name_suffix, "%s%s", full_path, extra);
if (ret < 0) {
- PERROR("Allocating path name ID");
- goto error;
+ PERROR("Allocating path name with extra string");
+ goto error_free_suffix;
}
- path = path_name_id;
+ path = path_name_suffix;
} else {
path = full_path;
}
ret = out_fd;
error_open:
- free(path_name_id);
+ free(path_name_suffix);
+error_free_suffix:
+ free(extra);
error:
return ret;
}
*/
LTTNG_HIDDEN
int utils_rotate_stream_file(char *path_name, char *file_name, uint64_t size,
- uint64_t count, int uid, int gid, int out_fd, uint64_t *new_count)
+ uint64_t count, int uid, int gid, int out_fd, uint64_t *new_count,
+ int *stream_fd)
{
int ret;
+ assert(new_count);
+ assert(stream_fd);
+
ret = close(out_fd);
if (ret < 0) {
PERROR("Closing tracefile");
(*new_count)++;
}
- return utils_create_stream_file(path_name, file_name, size, *new_count,
- uid, gid);
+ ret = utils_create_stream_file(path_name, file_name, size, *new_count,
+ uid, gid, 0);
+ if (ret < 0) {
+ goto error;
+ }
+ *stream_fd = ret;
+
+ /* Success. */
+ ret = 0;
+
error:
return ret;
}
int utils_create_pid_file(pid_t pid, const char *filepath);
int utils_mkdir_recursive(const char *path, mode_t mode);
int utils_create_stream_file(const char *path_name, char *file_name, uint64_t size,
- uint64_t count, int uid, int gid);
+ uint64_t count, int uid, int gid, char *suffix);
int utils_rotate_stream_file(char *path_name, char *file_name, uint64_t size,
- uint64_t count, int uid, int gid, int out_fd, uint64_t *new_count);
+ uint64_t count, int uid, int gid, int out_fd, uint64_t *new_count,
+ int *stream_fd);
int utils_parse_size_suffix(char *str, uint64_t *size);
int utils_get_count_order_u32(uint32_t x);
char *utils_get_home_dir(void);
projects / lttng-tools.git / commitdiff
