Refactor: lttng_ustctl_flush_buffer is a duplicate of lttng_ustconsumer_flush_buffer

[lttng-tools.git] / src / common / consumer / consumer.c

diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c
index b07939113cfe2e5bb1d87ec23a9da745d7a22508..3d725528a3a8638b54db5d0196a92936c6a44ba6 100644 (file)
--- a/src/common/consumer/consumer.c
+++ b/src/common/consumer/consumer.c
@@ -3568,7 +3568,6 @@ error:
 
                /* Assign new file descriptor */
                relayd->control_sock.sock.fd = fd;
-               fd = -1;        /* For error path */
                /* Assign version values. */
                relayd->control_sock.major = relayd_sock->major;
                relayd->control_sock.minor = relayd_sock->minor;
@@ -3596,7 +3595,6 @@ error:
 
                /* Assign new file descriptor */
                relayd->data_sock.sock.fd = fd;
-               fd = -1;        /* for eventual error paths */
                /* Assign version values. */
                relayd->data_sock.major = relayd_sock->major;
                relayd->data_sock.minor = relayd_sock->minor;
@@ -3610,6 +3608,11 @@ 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);
+       /*
+        * We gave the ownership of the fd to the relayd structure. Set the
+        * fd to -1 so we don't call close() on it in the error path below.
+        */
+       fd = -1;
 
        /* We successfully added the socket. Send status back. */
        ret = consumer_send_status_msg(sock, ret_code);
@@ -3880,7 +3883,7 @@ int consumer_flush_buffer(struct lttng_consumer_stream *stream, int producer_act
                break;
        case LTTNG_CONSUMER32_UST:
        case LTTNG_CONSUMER64_UST:
-               lttng_ustctl_flush_buffer(stream, producer_active);
+               lttng_ustconsumer_flush_buffer(stream, producer_active);
                break;
        default:
                ERR("Unknown consumer_data type");
@@ -3896,15 +3899,16 @@ end:
  * is already at the rotate position (produced == consumed), we flag it as
  * ready for rotation. The rotation of ready streams occurs after we have
  * replied to the session daemon that we have finished sampling the positions.
+ * Must be called with RCU read-side lock held to ensure existence of channel.
  *
  * Returns 0 on success, < 0 on error
  */
-int lttng_consumer_rotate_channel(uint64_t key, const char *path,
-               uint64_t relayd_id, uint32_t metadata, uint64_t new_chunk_id,
+int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel,
+               uint64_t key, const char *path, uint64_t relayd_id,
+               uint32_t metadata, uint64_t new_chunk_id,
                struct lttng_consumer_local_data *ctx)
 {
        int ret;
-       struct lttng_consumer_channel *channel;
        struct lttng_consumer_stream *stream;
        struct lttng_ht_iter iter;
        struct lttng_ht *ht = consumer_data.stream_per_chan_id_ht;
@@ -3913,13 +3917,6 @@ int lttng_consumer_rotate_channel(uint64_t key, const char *path,
 
        rcu_read_lock();
 
-       channel = consumer_find_channel(key);
-       if (!channel) {
-               ERR("No channel found for key %" PRIu64, key);
-               ret = -1;
-               goto end;
-       }
-
        pthread_mutex_lock(&channel->lock);
        channel->current_chunk_id = new_chunk_id;
 
@@ -4232,14 +4229,15 @@ error:
  * This is especially important for low throughput streams that have already
  * been consumed, we cannot wait for their next packet to perform the
  * rotation.
+ * Need to be called with RCU read-side lock held to ensure existence of
+ * channel.
  *
  * Returns 0 on success, < 0 on error
  */
-int lttng_consumer_rotate_ready_streams(uint64_t key,
-               struct lttng_consumer_local_data *ctx)
+int lttng_consumer_rotate_ready_streams(struct lttng_consumer_channel *channel,
+               uint64_t key, struct lttng_consumer_local_data *ctx)
 {
        int ret;
-       struct lttng_consumer_channel *channel;
        struct lttng_consumer_stream *stream;
        struct lttng_ht_iter iter;
        struct lttng_ht *ht = consumer_data.stream_per_chan_id_ht;
@@ -4248,13 +4246,6 @@ int lttng_consumer_rotate_ready_streams(uint64_t key,
 
        DBG("Consumer rotate ready streams in channel %" PRIu64, key);
 
-       channel = consumer_find_channel(key);
-       if (!channel) {
-               ERR("No channel found for key %" PRIu64, key);
-               ret = -1;
-               goto end;
-       }
-
        cds_lfht_for_each_entry_duplicate(ht->ht,
                        ht->hash_fct(&channel->key, lttng_ht_seed),
                        ht->match_fct, &channel->key, &iter.iter,