Fix: Unexpected payload size in cmd_recv_stream_2_11
[lttng-tools.git] / src / common / consumer / consumer.cpp
index fb5aeae7bc094873d01bdaefcd967b4055a8bf25..838990ccedefd7e24d693721140c0eb1924e3d22 100644 (file)
@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2011 Julien Desfossez <julien.desfossez@polymtl.ca>
+ * Copyright (C) 2011 EfficiOS Inc.
  * Copyright (C) 2011 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
  * Copyright (C) 2012 David Goulet <dgoulet@efficios.com>
  *
@@ -8,6 +8,7 @@
  */
 
 #include "common/index/ctf-index.h"
+#include <stdint.h>
 #define _LGPL_SOURCE
 #include <poll.h>
 #include <pthread.h>
@@ -246,6 +247,8 @@ struct lttng_consumer_channel *consumer_find_channel(uint64_t key)
        struct lttng_ht_node_u64 *node;
        struct lttng_consumer_channel *channel = NULL;
 
+       ASSERT_RCU_READ_LOCKED();
+
        /* -1ULL keys are lookup failures */
        if (key == (uint64_t) -1ULL) {
                return NULL;
@@ -521,6 +524,7 @@ void lttng_consumer_cleanup_relayd(struct consumer_relayd_sock_pair *relayd)
 void consumer_flag_relayd_for_destroy(struct consumer_relayd_sock_pair *relayd)
 {
        LTTNG_ASSERT(relayd);
+       ASSERT_RCU_READ_LOCKED();
 
        /* Set destroy flag for this object */
        uatomic_set(&relayd->destroy_flag, 1);
@@ -633,6 +637,7 @@ static int add_relayd(struct consumer_relayd_sock_pair *relayd)
        struct lttng_ht_iter iter;
 
        LTTNG_ASSERT(relayd);
+       ASSERT_RCU_READ_LOCKED();
 
        lttng_ht_lookup(the_consumer_data.relayd_ht, &relayd->net_seq_idx,
                        &iter);
@@ -690,6 +695,8 @@ struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key)
        struct lttng_ht_node_u64 *node;
        struct consumer_relayd_sock_pair *relayd = NULL;
 
+       ASSERT_RCU_READ_LOCKED();
+
        /* Negative keys are lookup failures */
        if (key == (uint64_t) -1ULL) {
                goto error;
@@ -3558,18 +3565,24 @@ error:
  * This will create a relayd socket pair and add it to the relayd hash table.
  * The caller MUST acquire a RCU read side lock before calling it.
  */
- void consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
-               struct lttng_consumer_local_data *ctx, int sock,
+void consumer_add_relayd_socket(uint64_t net_seq_idx,
+               int sock_type,
+               struct lttng_consumer_local_data *ctx,
+               int sock,
                struct pollfd *consumer_sockpoll,
-               struct lttcomm_relayd_sock *relayd_sock, uint64_t sessiond_id,
-               uint64_t relayd_session_id)
+               uint64_t sessiond_id,
+               uint64_t relayd_session_id,
+               uint32_t relayd_version_major,
+               uint32_t relayd_version_minor,
+               enum lttcomm_sock_proto relayd_socket_protocol)
 {
        int fd = -1, ret = -1, relayd_created = 0;
        enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
        struct consumer_relayd_sock_pair *relayd = NULL;
 
        LTTNG_ASSERT(ctx);
-       LTTNG_ASSERT(relayd_sock);
+       LTTNG_ASSERT(sock >= 0);
+       ASSERT_RCU_READ_LOCKED();
 
        DBG("Consumer adding relayd socket (idx: %" PRIu64 ")", net_seq_idx);
 
@@ -3638,54 +3651,25 @@ error:
        switch (sock_type) {
        case LTTNG_STREAM_CONTROL:
                /* Copy received lttcomm socket */
-               lttcomm_copy_sock(&relayd->control_sock.sock, &relayd_sock->sock);
-               ret = lttcomm_create_sock(&relayd->control_sock.sock);
-               /* Handle create_sock error. */
-               if (ret < 0) {
-                       ret_code = LTTCOMM_CONSUMERD_ENOMEM;
-                       goto error;
-               }
-               /*
-                * Close the socket created internally by
-                * lttcomm_create_sock, so we can replace it by the one
-                * received from sessiond.
-                */
-               if (close(relayd->control_sock.sock.fd)) {
-                       PERROR("close");
-               }
+               ret = lttcomm_populate_sock_from_open_socket(
+                               &relayd->control_sock.sock, fd,
+                               relayd_socket_protocol);
 
-               /* Assign new file descriptor */
-               relayd->control_sock.sock.fd = fd;
                /* Assign version values. */
-               relayd->control_sock.major = relayd_sock->major;
-               relayd->control_sock.minor = relayd_sock->minor;
+               relayd->control_sock.major = relayd_version_major;
+               relayd->control_sock.minor = relayd_version_minor;
 
                relayd->relayd_session_id = relayd_session_id;
 
                break;
        case LTTNG_STREAM_DATA:
                /* Copy received lttcomm socket */
-               lttcomm_copy_sock(&relayd->data_sock.sock, &relayd_sock->sock);
-               ret = lttcomm_create_sock(&relayd->data_sock.sock);
-               /* Handle create_sock error. */
-               if (ret < 0) {
-                       ret_code = LTTCOMM_CONSUMERD_ENOMEM;
-                       goto error;
-               }
-               /*
-                * Close the socket created internally by
-                * lttcomm_create_sock, so we can replace it by the one
-                * received from sessiond.
-                */
-               if (close(relayd->data_sock.sock.fd)) {
-                       PERROR("close");
-               }
-
-               /* Assign new file descriptor */
-               relayd->data_sock.sock.fd = fd;
+               ret = lttcomm_populate_sock_from_open_socket(
+                               &relayd->data_sock.sock, fd,
+                               relayd_socket_protocol);
                /* Assign version values. */
-               relayd->data_sock.major = relayd_sock->major;
-               relayd->data_sock.minor = relayd_sock->minor;
+               relayd->data_sock.major = relayd_version_major;
+               relayd->data_sock.minor = relayd_version_minor;
                break;
        default:
                ERR("Unknown relayd socket type (%d)", sock_type);
@@ -3693,6 +3677,11 @@ error:
                goto error;
        }
 
+       if (ret < 0) {
+               ret_code = LTTCOMM_CONSUMERD_FATAL;
+               goto error;
+       }
+
        DBG("Consumer %s socket created successfully with net idx %" PRIu64 " (fd: %d)",
                        sock_type == LTTNG_STREAM_CONTROL ? "control" : "data",
                        relayd->net_seq_idx, fd);
@@ -3749,6 +3738,8 @@ static struct consumer_relayd_sock_pair *find_relayd_by_session_id(uint64_t id)
        struct lttng_ht_iter iter;
        struct consumer_relayd_sock_pair *relayd = NULL;
 
+       ASSERT_RCU_READ_LOCKED();
+
        /* Iterate over all relayd since they are indexed by net_seq_idx. */
        cds_lfht_for_each_entry(the_consumer_data.relayd_ht->ht, &iter.iter,
                        relayd, node.node) {
@@ -4013,6 +4004,8 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel,
        struct lttng_dynamic_pointer_array streams_packet_to_open;
        size_t stream_idx;
 
+       ASSERT_RCU_READ_LOCKED();
+
        DBG("Consumer sample rotate position for channel %" PRIu64, key);
 
        lttng_dynamic_array_init(&stream_rotation_positions,
@@ -4672,6 +4665,8 @@ int lttng_consumer_rotate_ready_streams(struct lttng_consumer_channel *channel,
        struct lttng_ht_iter iter;
        struct lttng_ht *ht = the_consumer_data.stream_per_chan_id_ht;
 
+       ASSERT_RCU_READ_LOCKED();
+
        rcu_read_lock();
 
        DBG("Consumer rotate ready streams in channel %" PRIu64, key);
This page took 0.026533 seconds and 4 git commands to generate.