*
*/
-#include "common/index/ctf-index.h"
-#include <stdint.h>
#define _LGPL_SOURCE
+#include <inttypes.h>
#include <poll.h>
#include <pthread.h>
+#include <signal.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
-#include <inttypes.h>
-#include <signal.h>
-#include <bin/lttng-consumerd/health-consumerd.h>
-#include <common/common.h>
-#include <common/utils.h>
-#include <common/time.h>
-#include <common/compat/poll.h>
-#include <common/compat/endian.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>
-#include <common/kernel-consumer/kernel-consumer.h>
-#include <common/relayd/relayd.h>
-#include <common/ust-consumer/ust-consumer.h>
-#include <common/consumer/consumer-timer.h>
-#include <common/consumer/consumer.h>
-#include <common/consumer/consumer-stream.h>
-#include <common/consumer/consumer-testpoint.h>
-#include <common/align.h>
-#include <common/consumer/consumer-metadata-cache.h>
-#include <common/trace-chunk.h>
-#include <common/trace-chunk-registry.h>
-#include <common/string-utils/format.h>
-#include <common/dynamic-array.h>
+#include <bin/lttng-consumerd/health-consumerd.hpp>
+#include <common/align.hpp>
+#include <common/common.hpp>
+#include <common/compat/endian.hpp>
+#include <common/compat/poll.hpp>
+#include <common/consumer/consumer-metadata-cache.hpp>
+#include <common/consumer/consumer-stream.hpp>
+#include <common/consumer/consumer-testpoint.hpp>
+#include <common/consumer/consumer-timer.hpp>
+#include <common/consumer/consumer.hpp>
+#include <common/dynamic-array.hpp>
+#include <common/index/ctf-index.hpp>
+#include <common/index/index.hpp>
+#include <common/kernel-consumer/kernel-consumer.hpp>
+#include <common/kernel-ctl/kernel-ctl.hpp>
+#include <common/relayd/relayd.hpp>
+#include <common/sessiond-comm/relayd.hpp>
+#include <common/sessiond-comm/sessiond-comm.hpp>
+#include <common/string-utils/format.hpp>
+#include <common/time.hpp>
+#include <common/trace-chunk-registry.hpp>
+#include <common/trace-chunk.hpp>
+#include <common/ust-consumer/ust-consumer.hpp>
+#include <common/utils.hpp>
lttng_consumer_global_data the_consumer_data;
CONSUMER_CHANNEL_QUIT,
};
+namespace {
struct consumer_channel_msg {
enum consumer_channel_action action;
struct lttng_consumer_channel *chan; /* add */
uint64_t key; /* del */
};
+/*
+ * Global hash table containing respectively metadata and data streams. The
+ * stream element in this ht should only be updated by the metadata poll thread
+ * for the metadata and the data poll thread for the data.
+ */
+struct lttng_ht *metadata_ht;
+struct lttng_ht *data_ht;
+} /* namespace */
+
/* Flag used to temporarily pause data consumption from testpoints. */
int data_consumption_paused;
*/
int consumer_quit;
-/*
- * Global hash table containing respectively metadata and data streams. The
- * stream element in this ht should only be updated by the metadata poll thread
- * for the metadata and the data poll thread for the data.
- */
-static struct lttng_ht *metadata_ht;
-static struct lttng_ht *data_ht;
-
static const char *get_consumer_domain(void)
{
switch (the_consumer_data.type) {
goto error;
}
- obj = (consumer_relayd_sock_pair *) zmalloc(sizeof(struct consumer_relayd_sock_pair));
+ obj = zmalloc<consumer_relayd_sock_pair>();
if (obj == NULL) {
PERROR("zmalloc relayd sock");
goto error;
}
}
- channel = (lttng_consumer_channel *) zmalloc(sizeof(*channel));
+ channel = zmalloc<lttng_consumer_channel>();
if (channel == NULL) {
PERROR("malloc struct lttng_consumer_channel");
goto end;
the_consumer_data.type == type);
the_consumer_data.type = type;
- ctx = (lttng_consumer_local_data *) zmalloc(sizeof(struct lttng_consumer_local_data));
+ ctx = zmalloc<lttng_consumer_local_data>();
if (ctx == NULL) {
PERROR("allocating context");
goto error;
health_code_update();
- local_stream = (lttng_consumer_stream **) zmalloc(sizeof(struct lttng_consumer_stream *));
+ local_stream = zmalloc<lttng_consumer_stream *>();
if (local_stream == NULL) {
PERROR("local_stream malloc");
goto end;
local_stream = NULL;
/* Allocate for all fds */
- pollfd = (struct pollfd *) zmalloc((the_consumer_data.stream_count +
- nb_pipes_fd) *
- sizeof(struct pollfd));
+ pollfd = calloc<struct pollfd>(the_consumer_data.stream_count + nb_pipes_fd);
if (pollfd == NULL) {
PERROR("pollfd malloc");
pthread_mutex_unlock(&the_consumer_data.lock);
goto end;
}
- local_stream = (lttng_consumer_stream **) zmalloc((the_consumer_data.stream_count +
- nb_pipes_fd) *
- sizeof(struct lttng_consumer_stream *));
+ local_stream = calloc<lttng_consumer_stream *>(the_consumer_data.stream_count + nb_pipes_fd);
if (local_stream == NULL) {
PERROR("local_stream malloc");
pthread_mutex_unlock(&the_consumer_data.lock);
*/
if (stream->rotate_ready) {
DBG("Rotate stream before consuming data");
- ret = lttng_consumer_rotate_stream(ctx, stream);
+ ret = lttng_consumer_rotate_stream(stream);
if (ret < 0) {
ERR("Stream rotation error before consuming data");
goto end;
*/
rotation_ret = lttng_consumer_stream_is_rotate_ready(stream);
if (rotation_ret == 1) {
- rotation_ret = lttng_consumer_rotate_stream(ctx, stream);
+ rotation_ret = lttng_consumer_rotate_stream(stream);
if (rotation_ret < 0) {
ret = rotation_ret;
ERR("Stream rotation error after consuming data");
* Returns 0 on success, < 0 on error
*/
int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel,
- uint64_t key, uint64_t relayd_id, uint32_t metadata,
- struct lttng_consumer_local_data *ctx)
+ uint64_t key, uint64_t relayd_id)
{
int ret;
struct lttng_consumer_stream *stream;
* Perform the rotation a local stream file.
*/
static
-int rotate_local_stream(struct lttng_consumer_local_data *ctx,
- struct lttng_consumer_stream *stream)
+int rotate_local_stream(struct lttng_consumer_stream *stream)
{
int ret = 0;
*
* Return 0 on success, a negative number of error.
*/
-int lttng_consumer_rotate_stream(struct lttng_consumer_local_data *ctx,
- struct lttng_consumer_stream *stream)
+int lttng_consumer_rotate_stream(struct lttng_consumer_stream *stream)
{
int ret;
}
if (stream->net_seq_idx == (uint64_t) -1ULL) {
- ret = rotate_local_stream(ctx, stream);
+ ret = rotate_local_stream(stream);
if (ret < 0) {
ERR("Failed to rotate stream, ret = %i", ret);
goto error;
* Returns 0 on success, < 0 on error
*/
int lttng_consumer_rotate_ready_streams(struct lttng_consumer_channel *channel,
- uint64_t key, struct lttng_consumer_local_data *ctx)
+ uint64_t key)
{
int ret;
struct lttng_consumer_stream *stream;
}
DBG("Consumer rotate ready stream %" PRIu64, stream->key);
- ret = lttng_consumer_rotate_stream(ctx, stream);
+ ret = lttng_consumer_rotate_stream(stream);
pthread_mutex_unlock(&stream->lock);
pthread_mutex_unlock(&stream->chan->lock);
if (ret) {
enum lttcomm_return_code lttng_consumer_init_command(
struct lttng_consumer_local_data *ctx,
- const lttng_uuid sessiond_uuid)
+ const lttng_uuid& sessiond_uuid)
{
enum lttcomm_return_code ret;
char uuid_str[LTTNG_UUID_STR_LEN];
}
ctx->sessiond_uuid.is_set = true;
- memcpy(ctx->sessiond_uuid.value, sessiond_uuid, sizeof(lttng_uuid));
+ ctx->sessiond_uuid.value = sessiond_uuid;
ret = LTTCOMM_CONSUMERD_SUCCESS;
lttng_uuid_to_str(sessiond_uuid, uuid_str);
DBG("Received session daemon UUID: %s", uuid_str);
projects / lttng-tools.git / blobdiff