- DBG("Evaluating rotate pending for chunk id %" PRIu64, chunk_id);
-
- /*
- * Iterate over all the streams in the session and check if they are
- * still waiting for data to perform their rotation.
- */
- rcu_read_lock();
- cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
- node.node) {
- if (!stream_get(stream)) {
- continue;
- }
- if (stream->trace->session != session) {
- stream_put(stream);
- continue;
- }
- pthread_mutex_lock(&stream->lock);
- if (stream->rotate_at_seq_num != -1ULL) {
- /* We have not yet performed the rotation. */
- rotate_pending = true;
- DBG("Stream %" PRIu64 " is still rotating",
- stream->stream_handle);
- } else if (stream->chunk_id < chunk_id) {
- /*
- * Stream closed on the consumer but still active on the
- * relay.
- */
- rotate_pending = true;
- DBG("Stream %" PRIu64 " did not exist on the consumer "
- "when the last rotation started, but is"
- "still waiting for data before getting"
- "closed",
- stream->stream_handle);
- }
- pthread_mutex_unlock(&stream->lock);
- stream_put(stream);
- if (rotate_pending) {
- goto send_reply;
- }
- }
-
-send_reply:
- rcu_read_unlock();
- memset(&reply, 0, sizeof(reply));
- reply.generic.ret_code = htobe32((uint32_t) LTTNG_OK);
- reply.is_pending = (uint8_t) !!rotate_pending;
- send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
- sizeof(reply), 0);
+end:
+ reply.ret_code = htobe32((uint32_t) reply_code);
+ send_ret = conn->sock->ops->sendmsg(conn->sock,
+ &reply,
+ sizeof(struct lttcomm_relayd_generic_reply),
+ 0);