DBG("Consumer delete channel key %" PRIu64, channel->key);
pthread_mutex_lock(&consumer_data.lock);
+ pthread_mutex_lock(&channel->lock);
+
+ /* Delete streams that might have been left in the stream list. */
+ cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
+ send_node) {
+ cds_list_del(&stream->send_node);
+ /*
+ * Once a stream is added to this list, the buffers were created so
+ * we have a guarantee that this call will succeed.
+ */
+ consumer_stream_destroy(stream, NULL);
+ }
switch (consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
break;
case LTTNG_CONSUMER32_UST:
case LTTNG_CONSUMER64_UST:
- /* Delete streams that might have been left in the stream list. */
- cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
- send_node) {
- cds_list_del(&stream->send_node);
- /*
- * Once a stream is added to this list, the buffers were created so
- * we have a guarantee that this call will succeed.
- */
- consumer_stream_destroy(stream, NULL);
- }
lttng_ustconsumer_del_channel(channel);
break;
default:
call_rcu(&channel->node.head, free_channel_rcu);
end:
+ pthread_mutex_unlock(&channel->lock);
pthread_mutex_unlock(&consumer_data.lock);
}
consumer_stream_destroy(stream, ht);
}
+/*
+ * XXX naming of del vs destroy is all mixed up.
+ */
+void consumer_del_stream_for_data(struct lttng_consumer_stream *stream)
+{
+ consumer_stream_destroy(stream, data_ht);
+}
+
+void consumer_del_stream_for_metadata(struct lttng_consumer_stream *stream)
+{
+ consumer_stream_destroy(stream, metadata_ht);
+}
+
struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
uint64_t stream_key,
enum lttng_consumer_stream_state state,
uint64_t session_id,
int cpu,
int *alloc_ret,
- enum consumer_channel_type type)
+ enum consumer_channel_type type,
+ unsigned int monitor)
{
int ret;
struct lttng_consumer_stream *stream;
stream->gid = gid;
stream->net_seq_idx = relayd_id;
stream->session_id = session_id;
+ stream->monitor = monitor;
+ stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
pthread_mutex_init(&stream->lock, NULL);
/* If channel is the metadata, flag this stream as metadata. */
/*
* Add a stream to the global list protected by a mutex.
*/
-static int add_stream(struct lttng_consumer_stream *stream,
- struct lttng_ht *ht)
+int consumer_add_data_stream(struct lttng_consumer_stream *stream)
{
+ struct lttng_ht *ht = data_ht;
int ret = 0;
- struct consumer_relayd_sock_pair *relayd;
assert(stream);
assert(ht);
DBG3("Adding consumer stream %" PRIu64, stream->key);
pthread_mutex_lock(&consumer_data.lock);
+ pthread_mutex_lock(&stream->chan->lock);
+ pthread_mutex_lock(&stream->chan->timer_lock);
pthread_mutex_lock(&stream->lock);
rcu_read_lock();
*/
lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
- /* Check and cleanup relayd */
- relayd = consumer_find_relayd(stream->net_seq_idx);
- if (relayd != NULL) {
- uatomic_inc(&relayd->refcount);
- }
-
/*
* When nb_init_stream_left reaches 0, we don't need to trigger any action
* in terms of destroying the associated channel, because the action that
rcu_read_unlock();
pthread_mutex_unlock(&stream->lock);
+ pthread_mutex_unlock(&stream->chan->timer_lock);
+ pthread_mutex_unlock(&stream->chan->lock);
pthread_mutex_unlock(&consumer_data.lock);
return ret;
}
+void consumer_del_data_stream(struct lttng_consumer_stream *stream)
+{
+ consumer_del_stream(stream, data_ht);
+}
+
/*
* Add relayd socket to global consumer data hashtable. RCU read side lock MUST
* be acquired before calling this.
goto end;
}
uatomic_inc(&relayd->refcount);
+ stream->sent_to_relayd = 1;
} else {
ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't send it.",
stream->key, stream->net_seq_idx);
channel->tracefile_size = tracefile_size;
channel->tracefile_count = tracefile_count;
channel->monitor = monitor;
+ pthread_mutex_init(&channel->lock, NULL);
+ pthread_mutex_init(&channel->timer_lock, NULL);
/*
* In monitor mode, the streams associated with the channel will be put in
struct lttng_ht_iter iter;
pthread_mutex_lock(&consumer_data.lock);
+ pthread_mutex_lock(&channel->lock);
+ pthread_mutex_lock(&channel->timer_lock);
rcu_read_lock();
lttng_ht_lookup(consumer_data.channel_ht, &channel->key, &iter);
end:
rcu_read_unlock();
+ pthread_mutex_unlock(&channel->timer_lock);
+ pthread_mutex_unlock(&channel->lock);
pthread_mutex_unlock(&consumer_data.lock);
if (!ret && channel->wait_fd != -1 &&
struct lttng_consumer_local_data *ctx),
int (*recv_channel)(struct lttng_consumer_channel *channel),
int (*recv_stream)(struct lttng_consumer_stream *stream),
- int (*update_stream)(int stream_key, uint32_t state))
+ int (*update_stream)(uint64_t stream_key, uint32_t state))
{
int ret;
struct lttng_consumer_local_data *ctx;
ctx->consumer_error_socket = -1;
ctx->consumer_metadata_socket = -1;
+ pthread_mutex_init(&ctx->metadata_socket_lock, NULL);
/* assign the callbacks */
ctx->on_buffer_ready = buffer_ready;
ctx->on_recv_channel = recv_channel;
}
pthread_mutex_lock(&consumer_data.lock);
+ pthread_mutex_lock(&stream->chan->lock);
pthread_mutex_lock(&stream->lock);
switch (consumer_data.type) {
break;
case LTTNG_CONSUMER32_UST:
case LTTNG_CONSUMER64_UST:
+ if (stream->monitor) {
+ /* close the write-side in close_metadata */
+ ret = close(stream->ust_metadata_poll_pipe[0]);
+ if (ret < 0) {
+ PERROR("Close UST metadata read-side poll pipe");
+ }
+ }
lttng_ustconsumer_del_stream(stream);
break;
default:
end:
/*
* Nullify the stream reference so it is not used after deletion. The
- * consumer data lock MUST be acquired before being able to check for a
- * NULL pointer value.
+ * channel lock MUST be acquired before being able to check for
+ * a NULL pointer value.
*/
stream->chan->metadata_stream = NULL;
pthread_mutex_unlock(&stream->lock);
+ pthread_mutex_unlock(&stream->chan->lock);
pthread_mutex_unlock(&consumer_data.lock);
if (free_chan) {
* Action done with the metadata stream when adding it to the consumer internal
* data structures to handle it.
*/
-static int add_metadata_stream(struct lttng_consumer_stream *stream,
- struct lttng_ht *ht)
+int consumer_add_metadata_stream(struct lttng_consumer_stream *stream)
{
+ struct lttng_ht *ht = metadata_ht;
int ret = 0;
- struct consumer_relayd_sock_pair *relayd;
struct lttng_ht_iter iter;
struct lttng_ht_node_u64 *node;
DBG3("Adding metadata stream %" PRIu64 " to hash table", stream->key);
pthread_mutex_lock(&consumer_data.lock);
+ pthread_mutex_lock(&stream->chan->lock);
+ pthread_mutex_lock(&stream->chan->timer_lock);
pthread_mutex_lock(&stream->lock);
/*
node = lttng_ht_iter_get_node_u64(&iter);
assert(!node);
- /* Find relayd and, if one is found, increment refcount. */
- relayd = consumer_find_relayd(stream->net_seq_idx);
- if (relayd != NULL) {
- uatomic_inc(&relayd->refcount);
- }
-
/*
* When nb_init_stream_left reaches 0, we don't need to trigger any action
* in terms of destroying the associated channel, because the action that
rcu_read_unlock();
pthread_mutex_unlock(&stream->lock);
+ pthread_mutex_unlock(&stream->chan->lock);
+ pthread_mutex_unlock(&stream->chan->timer_lock);
pthread_mutex_unlock(&consumer_data.lock);
return ret;
}
pipe_len = lttng_pipe_read(ctx->consumer_metadata_pipe,
&stream, sizeof(stream));
if (pipe_len < 0) {
- ERR("read metadata stream, ret: %ld", pipe_len);
+ ERR("read metadata stream, ret: %zd", pipe_len);
/*
* Continue here to handle the rest of the streams.
*/
DBG("Adding metadata stream %d to poll set",
stream->wait_fd);
- ret = add_metadata_stream(stream, metadata_ht);
- if (ret) {
- ERR("Unable to add metadata stream");
- /* Stream was not setup properly. Continuing. */
- consumer_del_metadata_stream(stream, NULL);
- continue;
- }
-
/* Add metadata stream to the global poll events list */
lttng_poll_add(&events, stream->wait_fd,
LPOLLIN | LPOLLPRI);
DBG("Metadata available on fd %d", pollfd);
assert(stream->wait_fd == pollfd);
- len = ctx->on_buffer_ready(stream, ctx);
+ do {
+ len = ctx->on_buffer_ready(stream, ctx);
+ /*
+ * We don't check the return value here since if we get
+ * a negative len, it means an error occured thus we
+ * simply remove it from the poll set and free the
+ * stream.
+ */
+ } while (len > 0);
+
/* It's ok to have an unavailable sub-buffer */
if (len < 0 && len != -EAGAIN && len != -ENODATA) {
/* Clean up stream from consumer and free it. */
lttng_poll_del(&events, stream->wait_fd);
consumer_del_metadata_stream(stream, metadata_ht);
- } else if (len > 0) {
- stream->data_read = 1;
}
}
goto end;
}
- local_stream = zmalloc(sizeof(struct lttng_consumer_stream));
+ local_stream = zmalloc(sizeof(struct lttng_consumer_stream *));
+ if (local_stream == NULL) {
+ PERROR("local_stream malloc");
+ goto end;
+ }
while (1) {
high_prio = 0;
pipe_readlen = lttng_pipe_read(ctx->consumer_data_pipe,
&new_stream, sizeof(new_stream));
if (pipe_readlen < 0) {
- ERR("Consumer data pipe ret %ld", pipe_readlen);
+ ERR("Consumer data pipe ret %zd", pipe_readlen);
/* Continue so we can at least handle the current stream(s). */
continue;
}
continue;
}
- ret = add_stream(new_stream, data_ht);
- if (ret) {
- ERR("Consumer add stream %" PRIu64 " failed. Continuing",
- new_stream->key);
- /*
- * At this point, if the add_stream fails, it is not in the
- * hash table thus passing the NULL value here.
- */
- consumer_del_stream(new_stream, NULL);
- }
-
/* Continue to update the local streams and handle prio ones */
continue;
}
int 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, unsigned int sessiond_id)
+ struct lttcomm_relayd_sock *relayd_sock, uint64_t sessiond_id)
{
int fd = -1, ret = -1, relayd_created = 0;
enum lttng_error_code ret_code = LTTNG_OK;
ret_code = LTTCOMM_CONSUMERD_ENOMEM;
goto error;
} else {
- relayd->sessiond_session_id = (uint64_t) sessiond_id;
+ relayd->sessiond_session_id = sessiond_id;
relayd_created = 1;
}
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)
+{
+ 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;
+ }
+
+ return consumed_pos;
+}
projects / lttng-tools.git / blobdiff