struct lttng_ht_node_u64 node;
/* Size of subbuffers in this channel. */
size_t subbuf_size;
+ /* Number of subbuffers per stream. */
+ size_t num_subbuf;
union {
/* Original object data that MUST be copied over. */
struct lttng_ust_object_data *ust;
*/
static int record_kernel_snapshot(struct ltt_kernel_session *ksess,
struct snapshot_output *output, struct ltt_session *session,
- int wait, int nb_streams)
+ int wait, uint64_t nb_packets_per_stream)
{
int ret;
goto error_snapshot;
}
- ret = kernel_snapshot_record(ksess, output, wait, nb_streams);
+ ret = kernel_snapshot_record(ksess, output, wait, nb_packets_per_stream);
if (ret != LTTNG_OK) {
goto error_snapshot;
}
*/
static int record_ust_snapshot(struct ltt_ust_session *usess,
struct snapshot_output *output, struct ltt_session *session,
- int wait, int nb_streams)
+ int wait, uint64_t nb_packets_per_stream)
{
int ret;
goto error_snapshot;
}
- ret = ust_app_snapshot_record(usess, output, wait, nb_streams);
+ ret = ust_app_snapshot_record(usess, output, wait, nb_packets_per_stream);
if (ret < 0) {
switch (-ret) {
case EINVAL:
return ret;
}
-/*
- * Returns the total number of streams for a session or a negative value
- * on error.
- */
-static unsigned int get_total_nb_stream(struct ltt_session *session)
+static
+uint64_t get_session_size_one_more_packet_per_stream(struct ltt_session *session,
+ uint64_t cur_nr_packets)
{
- unsigned int total_streams = 0;
+ uint64_t tot_size = 0;
if (session->kernel_session) {
+ struct ltt_kernel_channel *chan;
struct ltt_kernel_session *ksess = session->kernel_session;
- total_streams += ksess->stream_count_global;
+ cds_list_for_each_entry(chan, &ksess->channel_list.head, list) {
+ if (cur_nr_packets >= chan->channel->attr.num_subbuf) {
+ /*
+ * Don't take channel into account if we
+ * already grab all its packets.
+ */
+ continue;
+ }
+ tot_size += chan->channel->attr.subbuf_size
+ * chan->stream_count;
+ }
}
if (session->ust_session) {
struct ltt_ust_session *usess = session->ust_session;
- total_streams += ust_app_get_nb_stream(usess);
+ tot_size += ust_app_get_size_one_more_packet_per_stream(usess,
+ cur_nr_packets);
}
- return total_streams;
+ return tot_size;
+}
+
+/*
+ * Calculate the number of packets we can grab from each stream that
+ * fits within the overall snapshot max size.
+ *
+ * Returns -1 on error, 0 means infinite number of packets, else > 0 is
+ * the number of packets per stream.
+ *
+ * TODO: this approach is not perfect: we consider the worse case
+ * (packet filling the sub-buffers) as an upper bound, but we could do
+ * better if we do this calculation while we actually grab the packet
+ * content: we would know how much padding we don't actually store into
+ * the file.
+ *
+ * This algorithm is currently bounded by the number of packets per
+ * stream.
+ *
+ * Since we call this algorithm before actually grabbing the data, it's
+ * an approximation: for instance, applications could appear/disappear
+ * in between this call and actually grabbing data.
+ */
+static
+int64_t get_session_nb_packets_per_stream(struct ltt_session *session, uint64_t max_size)
+{
+ int64_t size_left;
+ uint64_t cur_nb_packets = 0;
+
+ if (!max_size) {
+ return 0; /* Infinite */
+ }
+
+ size_left = max_size;
+ for (;;) {
+ uint64_t one_more_packet_tot_size;
+
+ one_more_packet_tot_size = get_session_size_one_more_packet_per_stream(session,
+ cur_nb_packets);
+ if (!one_more_packet_tot_size) {
+ /* We are already grabbing all packets. */
+ break;
+ }
+ size_left -= one_more_packet_tot_size;
+ if (size_left < 0) {
+ break;
+ }
+ cur_nb_packets++;
+ }
+ if (!cur_nb_packets) {
+ /* Not enough room to grab one packet of each stream, error. */
+ return -1;
+ }
+ return cur_nb_packets;
}
/*
int ret = LTTNG_OK;
unsigned int use_tmp_output = 0;
struct snapshot_output tmp_output;
- unsigned int nb_streams, snapshot_success = 0;
+ unsigned int snapshot_success = 0;
assert(session);
use_tmp_output = 1;
}
- /*
- * Get the total number of stream of that session which is used by the
- * maximum size of the snapshot feature.
- */
- nb_streams = get_total_nb_stream(session);
-
if (session->kernel_session) {
struct ltt_kernel_session *ksess = session->kernel_session;
if (use_tmp_output) {
+ int64_t nb_packets_per_stream;
+
+ nb_packets_per_stream = get_session_nb_packets_per_stream(session,
+ tmp_output.max_size);
+ if (nb_packets_per_stream < 0) {
+ ret = LTTNG_ERR_INVALID;
+ goto error;
+ }
ret = record_kernel_snapshot(ksess, &tmp_output, session,
- wait, nb_streams);
+ wait, nb_packets_per_stream);
if (ret != LTTNG_OK) {
goto error;
}
rcu_read_lock();
cds_lfht_for_each_entry(session->snapshot.output_ht->ht,
&iter.iter, sout, node.node) {
+ int64_t nb_packets_per_stream;
+
/*
* Make a local copy of the output and assign the possible
* temporary value given by the caller.
memset(&tmp_output, 0, sizeof(tmp_output));
memcpy(&tmp_output, sout, sizeof(tmp_output));
- /* Use temporary max size. */
if (output->max_size != (uint64_t) -1ULL) {
tmp_output.max_size = output->max_size;
}
+ nb_packets_per_stream = get_session_nb_packets_per_stream(session,
+ tmp_output.max_size);
+ if (nb_packets_per_stream < 0) {
+ ret = LTTNG_ERR_INVALID;
+ goto error;
+ }
+
/* Use temporary name. */
if (*output->name != '\0') {
strncpy(tmp_output.name, output->name,
tmp_output.nb_snapshot = session->snapshot.nb_snapshot;
ret = record_kernel_snapshot(ksess, &tmp_output,
- session, wait, nb_streams);
+ session, wait, nb_packets_per_stream);
if (ret != LTTNG_OK) {
rcu_read_unlock();
goto error;
struct ltt_ust_session *usess = session->ust_session;
if (use_tmp_output) {
+ int64_t nb_packets_per_stream;
+
+ nb_packets_per_stream = get_session_nb_packets_per_stream(session,
+ tmp_output.max_size);
+ if (nb_packets_per_stream < 0) {
+ ret = LTTNG_ERR_INVALID;
+ goto error;
+ }
ret = record_ust_snapshot(usess, &tmp_output, session,
- wait, nb_streams);
+ wait, nb_packets_per_stream);
if (ret != LTTNG_OK) {
goto error;
}
rcu_read_lock();
cds_lfht_for_each_entry(session->snapshot.output_ht->ht,
&iter.iter, sout, node.node) {
+ int64_t nb_packets_per_stream;
+
/*
* Make a local copy of the output and assign the possible
* temporary value given by the caller.
memset(&tmp_output, 0, sizeof(tmp_output));
memcpy(&tmp_output, sout, sizeof(tmp_output));
- /* Use temporary max size. */
if (output->max_size != (uint64_t) -1ULL) {
tmp_output.max_size = output->max_size;
}
+ nb_packets_per_stream = get_session_nb_packets_per_stream(session,
+ tmp_output.max_size);
+ if (nb_packets_per_stream < 0) {
+ ret = LTTNG_ERR_INVALID;
+ rcu_read_unlock();
+ goto error;
+ }
+
/* Use temporary name. */
if (*output->name != '\0') {
strncpy(tmp_output.name, output->name,
tmp_output.nb_snapshot = session->snapshot.nb_snapshot;
ret = record_ust_snapshot(usess, &tmp_output, session,
- wait, nb_streams);
+ wait, nb_packets_per_stream);
if (ret != LTTNG_OK) {
rcu_read_unlock();
goto error;
*/
int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key,
struct snapshot_output *output, int metadata, uid_t uid, gid_t gid,
- const char *session_path, int wait, int max_stream_size)
+ const char *session_path, int wait, uint64_t nb_packets_per_stream)
{
int ret;
struct lttcomm_consumer_msg msg;
memset(&msg, 0, sizeof(msg));
msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL;
msg.u.snapshot_channel.key = key;
- msg.u.snapshot_channel.max_stream_size = max_stream_size;
+ msg.u.snapshot_channel.nb_packets_per_stream = nb_packets_per_stream;
msg.u.snapshot_channel.metadata = metadata;
if (output->consumer->type == CONSUMER_DST_NET) {
/* Snapshot command. */
int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key,
struct snapshot_output *output, int metadata, uid_t uid, gid_t gid,
- const char *session_path, int wait, int max_size_per_stream);
+ const char *session_path, int wait, uint64_t nb_packets_per_stream);
#endif /* _CONSUMER_H */
* Return 0 on success or else return a LTTNG_ERR code.
*/
int kernel_snapshot_record(struct ltt_kernel_session *ksess,
- struct snapshot_output *output, int wait, unsigned int nb_streams)
+ struct snapshot_output *output, int wait,
+ uint64_t nb_packets_per_stream)
{
int err, ret, saved_metadata_fd;
struct consumer_socket *socket;
goto error_open_stream;
}
- if (output->max_size > 0 && nb_streams > 0) {
- max_size_per_stream = output->max_size / nb_streams;
- }
-
/* Send metadata to consumer and snapshot everything. */
cds_lfht_for_each_entry(ksess->consumer->socks->ht, &iter.iter,
socket, node.node) {
/* For each channel, ask the consumer to snapshot it. */
cds_list_for_each_entry(chan, &ksess->channel_list.head, list) {
- if (max_size_per_stream &&
- chan->channel->attr.subbuf_size > max_size_per_stream) {
- ret = LTTNG_ERR_INVALID;
- DBG3("Kernel snapshot record maximum stream size %" PRIu64
- " is smaller than subbuffer size of %" PRIu64,
- max_size_per_stream, chan->channel->attr.subbuf_size);
- (void) kernel_consumer_destroy_metadata(socket,
- ksess->metadata);
- goto error_consumer;
- }
-
pthread_mutex_lock(socket->lock);
ret = consumer_snapshot_channel(socket, chan->fd, output, 0,
ksess->uid, ksess->gid,
DEFAULT_KERNEL_TRACE_DIR, wait,
- max_size_per_stream);
+ nb_packets_per_stream);
pthread_mutex_unlock(socket->lock);
if (ret < 0) {
ret = LTTNG_ERR_KERN_CONSUMER_FAIL;
pthread_mutex_lock(socket->lock);
ret = consumer_snapshot_channel(socket, ksess->metadata->fd, output,
1, ksess->uid, ksess->gid,
- DEFAULT_KERNEL_TRACE_DIR, wait, max_size_per_stream);
+ DEFAULT_KERNEL_TRACE_DIR, wait, 0);
pthread_mutex_unlock(socket->lock);
if (ret < 0) {
ret = LTTNG_ERR_KERN_CONSUMER_FAIL;
void kernel_destroy_session(struct ltt_kernel_session *ksess);
void kernel_destroy_channel(struct ltt_kernel_channel *kchan);
int kernel_snapshot_record(struct ltt_kernel_session *ksess,
- struct snapshot_output *output, int wait, unsigned int nb_streams);
-
+ struct snapshot_output *output, int wait,
+ uint64_t nb_packets_per_stream);
int init_kernel_workarounds(void);
#endif /* _LTT_KERNEL_CTL_H */
{
int ret = 0, i;
- assert(output);
-
memset(output, 0, sizeof(struct snapshot_output));
+ /*
+ * max_size of -1ULL means unset. Set to default (unlimited).
+ */
if (max_size == (uint64_t) -1ULL) {
max_size = 0;
}
assert(reg_chan);
reg_chan->consumer_key = ua_chan->key;
reg_chan->subbuf_size = ua_chan->attr.subbuf_size;
+ reg_chan->num_subbuf = ua_chan->attr.num_subbuf;
/* Create and add a channel registry to session. */
ret = ust_registry_channel_add(reg_sess->reg.ust,
* Return 0 on success or else a negative value.
*/
int ust_app_snapshot_record(struct ltt_ust_session *usess,
- struct snapshot_output *output, int wait, unsigned int nb_streams)
+ struct snapshot_output *output, int wait,
+ uint64_t nb_packets_per_stream)
{
int ret = 0;
unsigned int snapshot_done = 0;
rcu_read_lock();
- /*
- * Compute the maximum size of a single stream if a max size is asked by
- * the caller.
- */
- if (output->max_size > 0 && nb_streams > 0) {
- max_stream_size = output->max_size / nb_streams;
- }
-
switch (usess->buffer_type) {
case LTTNG_BUFFER_PER_UID:
{
/* Add the UST default trace dir to path. */
cds_lfht_for_each_entry(reg->registry->channels->ht, &iter.iter,
reg_chan, node.node) {
-
- /*
- * Make sure the maximum stream size is not lower than the
- * subbuffer size or else it's an error since we won't be able to
- * snapshot anything.
- */
- if (max_stream_size &&
- reg_chan->subbuf_size > max_stream_size) {
- ret = -EINVAL;
- DBG3("UST app snapshot record maximum stream size %" PRIu64
- " is smaller than subbuffer size of %zu",
- max_stream_size, reg_chan->subbuf_size);
- goto error;
- }
- ret = consumer_snapshot_channel(socket, reg_chan->consumer_key, output, 0,
- usess->uid, usess->gid, pathname, wait,
- max_stream_size);
+ ret = consumer_snapshot_channel(socket, reg_chan->consumer_key,
+ output, 0, usess->uid, usess->gid, pathname, wait,
+ nb_packets_per_stream);
if (ret < 0) {
goto error;
}
}
- ret = consumer_snapshot_channel(socket, reg->registry->reg.ust->metadata_key, output,
- 1, usess->uid, usess->gid, pathname, wait,
- max_stream_size);
+ ret = consumer_snapshot_channel(socket,
+ reg->registry->reg.ust->metadata_key, output, 1,
+ usess->uid, usess->gid, pathname, wait, 0);
if (ret < 0) {
goto error;
}
cds_lfht_for_each_entry(ua_sess->channels->ht, &chan_iter.iter,
ua_chan, node.node) {
- /*
- * Make sure the maximum stream size is not lower than the
- * subbuffer size or else it's an error since we won't be able to
- * snapshot anything.
- */
- if (max_stream_size &&
- ua_chan->attr.subbuf_size > max_stream_size) {
- ret = -EINVAL;
- DBG3("UST app snapshot record maximum stream size %" PRIu64
- " is smaller than subbuffer size of %" PRIu64,
- max_stream_size, ua_chan->attr.subbuf_size);
- goto error;
- }
-
- ret = consumer_snapshot_channel(socket, ua_chan->key, output, 0,
- ua_sess->euid, ua_sess->egid, pathname, wait,
- max_stream_size);
+ ret = consumer_snapshot_channel(socket, ua_chan->key, output,
+ 0, ua_sess->euid, ua_sess->egid, pathname, wait,
+ nb_packets_per_stream);
if (ret < 0) {
goto error;
}
registry = get_session_registry(ua_sess);
assert(registry);
ret = consumer_snapshot_channel(socket, registry->metadata_key, output,
- 1, ua_sess->euid, ua_sess->egid, pathname, wait,
- max_stream_size);
+ 1, ua_sess->euid, ua_sess->egid, pathname, wait, 0);
if (ret < 0) {
goto error;
}
}
/*
- * Return the number of streams for a UST session.
+ * Return the size taken by one more packet per stream.
*/
-unsigned int ust_app_get_nb_stream(struct ltt_ust_session *usess)
+uint64_t ust_app_get_size_one_more_packet_per_stream(struct ltt_ust_session *usess,
+ uint64_t cur_nr_packets)
{
- unsigned int ret = 0;
+ uint64_t tot_size = 0;
struct ust_app *app;
struct lttng_ht_iter iter;
rcu_read_lock();
cds_lfht_for_each_entry(reg->registry->channels->ht, &iter.iter,
reg_chan, node.node) {
- ret += reg_chan->stream_count;
+ if (cur_nr_packets >= reg_chan->num_subbuf) {
+ /*
+ * Don't take channel into account if we
+ * already grab all its packets.
+ */
+ continue;
+ }
+ tot_size += reg_chan->subbuf_size * reg_chan->stream_count;
}
rcu_read_unlock();
}
cds_lfht_for_each_entry(ua_sess->channels->ht, &chan_iter.iter,
ua_chan, node.node) {
- ret += ua_chan->streams.count;
+ if (cur_nr_packets >= ua_chan->attr.num_subbuf) {
+ /*
+ * Don't take channel into account if we
+ * already grab all its packets.
+ */
+ continue;
+ }
+ tot_size += ua_chan->attr.subbuf_size * ua_chan->streams.count;
}
}
rcu_read_unlock();
break;
}
- return ret;
+ return tot_size;
}
struct consumer_socket *socket, int send_zero_data);
void ust_app_destroy(struct ust_app *app);
int ust_app_snapshot_record(struct ltt_ust_session *usess,
- struct snapshot_output *output, int wait, unsigned int nb_streams);
-unsigned int ust_app_get_nb_stream(struct ltt_ust_session *usess);
+ struct snapshot_output *output, int wait,
+ uint64_t nb_packets_per_stream);
+uint64_t ust_app_get_size_one_more_packet_per_stream(struct ltt_ust_session *usess,
+ uint64_t cur_nr_packets);
struct ust_app *ust_app_find_by_sock(int sock);
static inline
--- /dev/null
+#ifndef _LTTNG_ALIGN_H
+#define _LTTNG_ALIGN_H
+
+/*
+ * align.h
+ *
+ * (C) Copyright 2010-2011 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ */
+
+#include "bug.h"
+#include <unistd.h>
+#include <limits.h>
+
+#ifndef PAGE_SIZE /* Cygwin limits.h defines its own PAGE_SIZE */
+#define PAGE_SIZE sysconf(_SC_PAGE_SIZE)
+#endif
+
+#define PAGE_MASK (~(PAGE_SIZE - 1))
+#define __ALIGN_MASK(v, mask) (((v) + (mask)) & ~(mask))
+#define ALIGN(v, align) __ALIGN_MASK(v, (__typeof__(v)) (align) - 1)
+#define PAGE_ALIGN(addr) ALIGN(addr, PAGE_SIZE)
+
+/**
+ * offset_align - Calculate the offset needed to align an object on its natural
+ * alignment towards higher addresses.
+ * @align_drift: object offset from an "alignment"-aligned address.
+ * @alignment: natural object alignment. Must be non-zero, power of 2.
+ *
+ * Returns the offset that must be added to align towards higher
+ * addresses.
+ */
+#define offset_align(align_drift, alignment) \
+ ({ \
+ LTTNG_BUILD_RUNTIME_BUG_ON((alignment) == 0 \
+ || ((alignment) & ((alignment) - 1))); \
+ (((alignment) - (align_drift)) & ((alignment) - 1)); \
+ })
+
+/**
+ * offset_align_floor - Calculate the offset needed to align an object
+ * on its natural alignment towards lower addresses.
+ * @align_drift: object offset from an "alignment"-aligned address.
+ * @alignment: natural object alignment. Must be non-zero, power of 2.
+ *
+ * Returns the offset that must be substracted to align towards lower addresses.
+ */
+#define offset_align_floor(align_drift, alignment) \
+ ({ \
+ LTTNG_BUILD_RUNTIME_BUG_ON((alignment) == 0 \
+ || ((alignment) & ((alignment) - 1))); \
+ (((align_drift) - (alignment)) & ((alignment) - 1)); \
+ })
+
+#endif /* _LTTNG_ALIGN_H */
--- /dev/null
+#ifndef _LTTNG_BUG_H
+#define _LTTNG_BUG_H
+
+/*
+ * lttng/bug.h
+ *
+ * (C) Copyright 2010-2011 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ */
+
+#include <urcu/compiler.h>
+#include <stdio.h>
+#include <stdlib.h>
+
+#define LTTNG_BUG_ON(condition) \
+ do { \
+ if (caa_unlikely(condition)) { \
+ fprintf(stderr, \
+ "LTTng BUG in file %s, line %d.\n", \
+ __FILE__, __LINE__); \
+ exit(EXIT_FAILURE); \
+ } \
+ } while (0)
+
+#define LTTNG_BUILD_BUG_ON(condition) \
+ ((void) sizeof(char[-!!(condition)]))
+
+/**
+ * LTTNG_BUILD_RUNTIME_BUG_ON - check condition at build (if constant) or runtime
+ * @condition: the condition which should be false.
+ *
+ * If the condition is a constant and true, the compiler will generate a build
+ * error. If the condition is not constant, a BUG will be triggered at runtime
+ * if the condition is ever true. If the condition is constant and false, no
+ * code is emitted.
+ */
+#define LTTNG_BUILD_RUNTIME_BUG_ON(condition) \
+ do { \
+ if (__builtin_constant_p(condition)) \
+ LTTNG_BUILD_BUG_ON(condition); \
+ else \
+ LTTNG_BUG_ON(condition); \
+ } while (0)
+
+#endif
#include "consumer.h"
#include "consumer-stream.h"
#include "consumer-testpoint.h"
+#include "align.h"
struct lttng_consumer_global_data consumer_data = {
.stream_count = 0,
return lttcomm_send_unix_sock(sock, &msg, sizeof(msg));
}
-/*
- * Using a maximum stream size with the produced and consumed position of a
- * stream, computes the new consumed position to be as close as possible to the
- * maximum possible stream size.
- *
- * If maximum stream size is lower than the possible buffer size (produced -
- * consumed), the consumed_pos given is returned untouched else the new value
- * is returned.
- */
-unsigned long consumer_get_consumed_maxsize(unsigned long consumed_pos,
- unsigned long produced_pos, uint64_t max_stream_size)
+unsigned long consumer_get_consume_start_pos(unsigned long consumed_pos,
+ unsigned long produced_pos, uint64_t nb_packets_per_stream,
+ uint64_t max_sb_size)
{
- if (max_stream_size && max_stream_size < (produced_pos - consumed_pos)) {
- /* Offset from the produced position to get the latest buffers. */
- return produced_pos - max_stream_size;
- }
+ unsigned long start_pos;
- return consumed_pos;
+ if (!nb_packets_per_stream) {
+ return consumed_pos; /* Grab everything */
+ }
+ start_pos = produced_pos - offset_align_floor(produced_pos, max_sb_size);
+ start_pos -= max_sb_size * nb_packets_per_stream;
+ if ((long) (start_pos - consumed_pos) < 0) {
+ return consumed_pos; /* Grab everything */
+ }
+ return start_pos;
}
void notify_thread_del_channel(struct lttng_consumer_local_data *ctx,
uint64_t key);
void consumer_destroy_relayd(struct consumer_relayd_sock_pair *relayd);
-unsigned long consumer_get_consumed_maxsize(unsigned long consumed_pos,
- unsigned long produced_pos, uint64_t max_stream_size);
+unsigned long consumer_get_consume_start_pos(unsigned long consumed_pos,
+ unsigned long produced_pos, uint64_t nb_packets_per_stream,
+ uint64_t max_sb_size);
int consumer_add_data_stream(struct lttng_consumer_stream *stream);
void consumer_del_stream_for_data(struct lttng_consumer_stream *stream);
int consumer_add_metadata_stream(struct lttng_consumer_stream *stream);
* Returns 0 on success, < 0 on error
*/
int lttng_kconsumer_snapshot_channel(uint64_t key, char *path,
- uint64_t relayd_id, uint64_t max_stream_size,
+ uint64_t relayd_id, uint64_t nb_packets_per_stream,
struct lttng_consumer_local_data *ctx)
{
int ret;
}
}
- /*
- * The original value is sent back if max stream size is larger than
- * the possible size of the snapshot. Also, we asume that the session
- * daemon should never send a maximum stream size that is lower than
- * subbuffer size.
- */
- consumed_pos = consumer_get_consumed_maxsize(consumed_pos,
- produced_pos, max_stream_size);
+ consumed_pos = consumer_get_consume_start_pos(consumed_pos,
+ produced_pos, nb_packets_per_stream,
+ stream->max_sb_size);
while (consumed_pos < produced_pos) {
ssize_t read_len;
ret = lttng_kconsumer_snapshot_channel(msg.u.snapshot_channel.key,
msg.u.snapshot_channel.pathname,
msg.u.snapshot_channel.relayd_id,
- msg.u.snapshot_channel.max_stream_size,
+ msg.u.snapshot_channel.nb_packets_per_stream,
ctx);
if (ret < 0) {
ERR("Snapshot channel failed");
uint32_t metadata; /* This a metadata snapshot. */
uint64_t relayd_id; /* Relayd id if apply. */
uint64_t key;
- uint64_t max_stream_size;
+ uint64_t nb_packets_per_stream;
} LTTNG_PACKED snapshot_channel;
struct {
uint64_t channel_key;
* Returns 0 on success, < 0 on error
*/
static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
- uint64_t max_stream_size, struct lttng_consumer_local_data *ctx)
+ uint64_t nb_packets_per_stream, struct lttng_consumer_local_data *ctx)
{
int ret;
unsigned use_relayd = 0;
/*
* The original value is sent back if max stream size is larger than
- * the possible size of the snapshot. Also, we asume that the session
+ * the possible size of the snapshot. Also, we assume that the session
* daemon should never send a maximum stream size that is lower than
* subbuffer size.
*/
- consumed_pos = consumer_get_consumed_maxsize(consumed_pos,
- produced_pos, max_stream_size);
+ consumed_pos = consumer_get_consume_start_pos(consumed_pos,
+ produced_pos, nb_packets_per_stream,
+ stream->max_sb_size);
while (consumed_pos < produced_pos) {
ssize_t read_len;
ret = snapshot_channel(msg.u.snapshot_channel.key,
msg.u.snapshot_channel.pathname,
msg.u.snapshot_channel.relayd_id,
- msg.u.snapshot_channel.max_stream_size,
+ msg.u.snapshot_channel.nb_packets_per_stream,
ctx);
if (ret < 0) {
ERR("Snapshot channel failed");
+++ /dev/null
-#ifndef _LTTNG_ALIGN_H
-#define _LTTNG_ALIGN_H
-
-/*
- * align.h
- *
- * (C) Copyright 2010-2011 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
- *
- * Permission is hereby granted, free of charge, to any person obtaining a copy
- * of this software and associated documentation files (the "Software"), to deal
- * in the Software without restriction, including without limitation the rights
- * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
- * copies of the Software, and to permit persons to whom the Software is
- * furnished to do so, subject to the following conditions:
- *
- * The above copyright notice and this permission notice shall be included in
- * all copies or substantial portions of the Software.
- */
-
-#include "bug.h"
-#include <unistd.h>
-#include <limits.h>
-
-#ifndef PAGE_SIZE /* Cygwin limits.h defines its own PAGE_SIZE */
-#define PAGE_SIZE sysconf(_SC_PAGE_SIZE)
-#endif
-
-#define PAGE_MASK (~(PAGE_SIZE - 1))
-#define __ALIGN_MASK(v, mask) (((v) + (mask)) & ~(mask))
-#define ALIGN(v, align) __ALIGN_MASK(v, (__typeof__(v)) (align) - 1)
-#define PAGE_ALIGN(addr) ALIGN(addr, PAGE_SIZE)
-
-/**
- * offset_align - Calculate the offset needed to align an object on its natural
- * alignment towards higher addresses.
- * @align_drift: object offset from an "alignment"-aligned address.
- * @alignment: natural object alignment. Must be non-zero, power of 2.
- *
- * Returns the offset that must be added to align towards higher
- * addresses.
- */
-#define offset_align(align_drift, alignment) \
- ({ \
- LTTNG_BUILD_RUNTIME_BUG_ON((alignment) == 0 \
- || ((alignment) & ((alignment) - 1))); \
- (((alignment) - (align_drift)) & ((alignment) - 1)); \
- })
-
-/**
- * offset_align_floor - Calculate the offset needed to align an object
- * on its natural alignment towards lower addresses.
- * @align_drift: object offset from an "alignment"-aligned address.
- * @alignment: natural object alignment. Must be non-zero, power of 2.
- *
- * Returns the offset that must be substracted to align towards lower addresses.
- */
-#define offset_align_floor(align_drift, alignment) \
- ({ \
- LTTNG_BUILD_RUNTIME_BUG_ON((alignment) == 0 \
- || ((alignment) & ((alignment) - 1))); \
- (((align_drift) - (alignment)) & ((alignment) - 1); \
- })
-
-#endif /* _LTTNG_ALIGN_H */
+++ /dev/null
-#ifndef _LTTNG_BUG_H
-#define _LTTNG_BUG_H
-
-/*
- * lttng/bug.h
- *
- * (C) Copyright 2010-2011 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
- *
- * Permission is hereby granted, free of charge, to any person obtaining a copy
- * of this software and associated documentation files (the "Software"), to deal
- * in the Software without restriction, including without limitation the rights
- * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
- * copies of the Software, and to permit persons to whom the Software is
- * furnished to do so, subject to the following conditions:
- *
- * The above copyright notice and this permission notice shall be included in
- * all copies or substantial portions of the Software.
- */
-
-#include <urcu/compiler.h>
-#include <stdio.h>
-#include <stdlib.h>
-
-#define LTTNG_BUG_ON(condition) \
- do { \
- if (caa_unlikely(condition)) { \
- fprintf(stderr, \
- "LTTng BUG in file %s, line %d.\n", \
- __FILE__, __LINE__); \
- exit(EXIT_FAILURE); \
- } \
- } while (0)
-
-#define LTTNG_BUILD_BUG_ON(condition) \
- ((void) sizeof(char[-!!(condition)]))
-
-/**
- * LTTNG_BUILD_RUNTIME_BUG_ON - check condition at build (if constant) or runtime
- * @condition: the condition which should be false.
- *
- * If the condition is a constant and true, the compiler will generate a build
- * error. If the condition is not constant, a BUG will be triggered at runtime
- * if the condition is ever true. If the condition is constant and false, no
- * code is emitted.
- */
-#define LTTNG_BUILD_RUNTIME_BUG_ON(condition) \
- do { \
- if (__builtin_constant_p(condition)) \
- LTTNG_BUILD_BUG_ON(condition); \
- else \
- LTTNG_BUG_ON(condition); \
- } while (0)
-
-#endif
#include <stdlib.h>
#include <string.h>
#include <errno.h>
-#include "align.h"
+#include "common/align.h"
#include "filter-bytecode.h"
#include "filter-ir.h"
#include "filter-ast.h"
projects / lttng-tools.git / commitdiff
