X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;ds=sidebyside;f=src%2Fcommon%2Fconsumer%2Fconsumer.cpp;h=a9de1696e4ce862d52bf58b5c6ab44f73ad76b28;hb=HEAD;hp=27b34a39f249071867e49e9df7fb438517bb6ee8;hpb=56047f5a23df5c2c583a102b8015bbec5a7da9f1;p=lttng-tools.git diff --git a/src/common/consumer/consumer.cpp b/src/common/consumer/consumer.cpp index 27b34a39f..1da243601 100644 --- a/src/common/consumer/consumer.cpp +++ b/src/common/consumer/consumer.cpp @@ -20,6 +20,7 @@ #include #include #include +#include #include #include #include @@ -34,6 +35,7 @@ #include #include +#include #include #include #include @@ -43,6 +45,7 @@ #include #include #include +#include #include lttng_consumer_global_data the_consumer_data; @@ -175,13 +178,6 @@ static void clean_channel_stream_list(struct lttng_consumer_channel *channel) /* Delete streams that might have been left in the stream list. */ cds_list_for_each_entry_safe (stream, stmp, &channel->streams.head, send_node) { - /* - * Once a stream is added to this list, the buffers were created so we - * have a guarantee that this call will succeed. Setting the monitor - * mode to 0 so we don't lock nor try to delete the stream from the - * global hash table. - */ - stream->monitor = 0; consumer_stream_destroy(stream, nullptr); } } @@ -301,7 +297,8 @@ static void free_channel_rcu(struct rcu_head *head) ERR("Unknown consumer_data type"); abort(); } - free(channel); + + delete channel; } /* @@ -460,6 +457,8 @@ static void update_endpoint_status_by_netidx(uint64_t net_seq_idx, cds_lfht_for_each_entry (metadata_ht->ht, &iter.iter, stream, node.node) { if (stream->net_seq_idx == net_seq_idx) { uatomic_set(&stream->endpoint_status, status); + stream->chan->metadata_pushed_wait_queue.wake_all(); + DBG("Delete flag set to metadata stream %d", stream->wait_fd); } } @@ -1015,9 +1014,11 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key, } } - channel = zmalloc(); - if (channel == nullptr) { - PERROR("malloc struct lttng_consumer_channel"); + try { + channel = new lttng_consumer_channel; + } catch (const std::bad_alloc& e) { + ERR("Failed to allocate lttng_consumer_channel: %s", e.what()); + channel = nullptr; goto end; } @@ -1031,8 +1032,8 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key, channel->monitor = monitor; channel->live_timer_interval = live_timer_interval; channel->is_live = is_in_live_session; - pthread_mutex_init(&channel->lock, nullptr); - pthread_mutex_init(&channel->timer_lock, nullptr); + pthread_mutex_init(&channel->lock, NULL); + pthread_mutex_init(&channel->timer_lock, NULL); switch (output) { case LTTNG_EVENT_SPLICE: @@ -1043,7 +1044,7 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key, break; default: abort(); - free(channel); + delete channel; channel = nullptr; goto end; } @@ -1248,11 +1249,17 @@ void lttng_consumer_set_command_sock_path(struct lttng_consumer_local_data *ctx, * Send return code to the session daemon. * If the socket is not defined, we return 0, it is not a fatal error */ -int lttng_consumer_send_error(struct lttng_consumer_local_data *ctx, int cmd) +int lttng_consumer_send_error(struct lttng_consumer_local_data *ctx, + enum lttcomm_return_code error_code) { if (ctx->consumer_error_socket > 0) { + const std::int32_t comm_code = std::int32_t(error_code); + + static_assert( + sizeof(comm_code) >= sizeof(std::underlying_type), + "Fixed-size communication type too small to accomodate lttcomm_return_code"); return lttcomm_send_unix_sock( - ctx->consumer_error_socket, &cmd, sizeof(enum lttcomm_sessiond_command)); + ctx->consumer_error_socket, &comm_code, sizeof(comm_code)); } return 0; @@ -1336,7 +1343,6 @@ void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx) */ static void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream, off_t orig_offset) { - int ret; int outfd = stream->out_fd; /* @@ -1348,31 +1354,8 @@ static void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream, if (orig_offset < stream->max_sb_size) { return; } - lttng_sync_file_range(outfd, - orig_offset - stream->max_sb_size, - stream->max_sb_size, - SYNC_FILE_RANGE_WAIT_BEFORE | SYNC_FILE_RANGE_WRITE | - SYNC_FILE_RANGE_WAIT_AFTER); - /* - * Give hints to the kernel about how we access the file: - * POSIX_FADV_DONTNEED : we won't re-access data in a near future after - * we write it. - * - * We need to call fadvise again after the file grows because the - * kernel does not seem to apply fadvise to non-existing parts of the - * file. - * - * Call fadvise _after_ having waited for the page writeback to - * complete because the dirty page writeback semantic is not well - * defined. So it can be expected to lead to lower throughput in - * streaming. - */ - ret = posix_fadvise( - outfd, orig_offset - stream->max_sb_size, stream->max_sb_size, POSIX_FADV_DONTNEED); - if (ret && ret != -ENOSYS) { - errno = ret; - PERROR("posix_fadvise on fd %i", outfd); - } + lttng::io::hint_flush_range_dont_need_sync( + outfd, orig_offset - stream->max_sb_size, stream->max_sb_size); } /* @@ -1733,8 +1716,7 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(struct lttng_consumer_stream *stre /* This call is useless on a socket so better save a syscall. */ if (!relayd) { /* This won't block, but will start writeout asynchronously */ - lttng_sync_file_range( - outfd, stream->out_fd_offset, write_len, SYNC_FILE_RANGE_WRITE); + lttng::io::hint_flush_range_async(outfd, stream->out_fd_offset, write_len); stream->out_fd_offset += write_len; lttng_consumer_sync_trace_file(stream, orig_offset); } @@ -1933,8 +1915,7 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(struct lttng_consumer_local_data /* This call is useless on a socket so better save a syscall. */ if (!relayd) { /* This won't block, but will start writeout asynchronously */ - lttng_sync_file_range( - outfd, stream->out_fd_offset, ret_splice, SYNC_FILE_RANGE_WRITE); + lttng::io::hint_flush_range_async(outfd, stream->out_fd_offset, ret_splice); stream->out_fd_offset += ret_splice; } stream->output_written += ret_splice; @@ -2154,6 +2135,7 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream, struct l * pointer value. */ channel->metadata_stream = nullptr; + channel->metadata_pushed_wait_queue.wake_all(); if (channel->metadata_cache) { pthread_mutex_unlock(&channel->metadata_cache->lock); @@ -2516,7 +2498,7 @@ error_testpoint: */ void *consumer_thread_data_poll(void *data) { - int num_rdy, num_hup, high_prio, ret, i, err = -1; + int num_rdy, high_prio, ret, i, err = -1; struct pollfd *pollfd = nullptr; /* local view of the streams */ struct lttng_consumer_stream **local_stream = nullptr, *new_stream = nullptr; @@ -2549,7 +2531,6 @@ void *consumer_thread_data_poll(void *data) health_code_update(); high_prio = 0; - num_hup = 0; /* * the fds set has been updated, we need to update our @@ -2765,21 +2746,18 @@ void *consumer_thread_data_poll(void *data) if (!local_stream[i]->has_data_left_to_be_read_before_teardown) { consumer_del_stream(local_stream[i], data_ht); local_stream[i] = nullptr; - num_hup++; } } else if (pollfd[i].revents & POLLERR) { ERR("Error returned in polling fd %d.", pollfd[i].fd); if (!local_stream[i]->has_data_left_to_be_read_before_teardown) { consumer_del_stream(local_stream[i], data_ht); local_stream[i] = nullptr; - num_hup++; } } else if (pollfd[i].revents & POLLNVAL) { ERR("Polling fd %d tells fd is not open.", pollfd[i].fd); if (!local_stream[i]->has_data_left_to_be_read_before_teardown) { consumer_del_stream(local_stream[i], data_ht); local_stream[i] = nullptr; - num_hup++; } } if (local_stream[i] != nullptr) {