Recycle frame payload buffers from core back to the group audio threads #68

Problem

Each encoded frame and each visualizer cell carries its payload in a std::vector<uint8_t> that the group audio threads allocate for that frame, and core frees when its frame ring drops it. In steady state this is an allocation per frame on the audio threads and a free per frame on core. Reusing the buffers needs a way to hand them from core back to the audio threads, and no channel runs that way today: the group-to-core FIFO goes one way and the pipeline-set slot carries configuration only. Split from #23, which reused the other per-frame buffers.

Where

• The payload:

  sendspin-server-cpp/src/sched/frames.h

  Line 63 in 87821e0

  std::vector<uint8_t> payload;

• A frame per pipeline cell, its payload filled by the encoder:

  sendspin-server-cpp/src/group_audio.cpp

  Lines 912 to 933 in 87821e0

  	bool GroupAudio::emit(Instance& instance, const JobView& view, size_t samples) {
  	sched::Frame frame;
  	frame.epoch = this->epoch_;
  	frame.index = instance.next;
  	frame.ts_us = sched::frame_timestamp(view.anchor, instance.next);
  	// An Opus packet decodes to a whole frame (spec player/v1.md "Codec framing"), a drained
  	// epoch's last one padded with silence
  	frame.samples = static_cast<uint32_t>(
  	instance.format.codec == sched::Codec::OPUS ? instance.frame_samples : samples);
  	frame.duration_us = sched::frame_duration_us(frame.samples, instance.format.sample_rate);
  	// With a lead, a drained epoch's last frames take what pending still holds
  	const size_t bytes = std::min(samples * instance.sample_bytes, instance.pending.size());
  	if (!instance.encoder->encode_quantized(std::span(instance.pending.data(), bytes),
  	frame.payload)) {
  	return false;
  	}
  	this->fifo_.push(FrameReady{instance.id, std::move(frame)});
  	instance.pending.erase(instance.pending.begin(),
  	instance.pending.begin() + static_cast<std::ptrdiff_t>(bytes));
  	instance.last = instance.next;
  	instance.next += instance.cells;
  	return true;

• A frame per visualizer cell, its payload reserved per cell:

  sendspin-server-cpp/src/group_audio.cpp

  Lines 1046 to 1080 in 87821e0

  	sched::Frame frame;
  	frame.epoch = this->epoch_;
  	frame.index = cell;
  	frame.ts_us = cell_start + shift;
  	frame.payload.reserve(analysis.cell_bytes);
  	const auto instant = [&](int64_t at, bool is_periodic, bool onset) {
  	// A drained epoch's last cell has no input past the drain position to analyze
  	const uint64_t position = sched::time_position(anchor, at);
  	if (position > end) {
  	return true;
  	}
  	if (!this->feed_analysis(analysis, view, position)) {
  	return false;
  	}
  	const size_t before = frame.payload.size();
  	analysis.analysis->analyze(at + shift, is_periodic, onset, frame.payload);
  	if (frame.payload.size() != before) {
  	frame.ts_us = at + shift;
  	}
  	return true;
  	};
  	const bool merged = periodic && *periodic == cell_start;
  	if (!instant(cell_start, merged, true) ||
  	(periodic && !merged && !instant(*periodic, true, false)) ||
  	!this->feed_analysis(analysis, view, end)) {
  	// Input below the head is staged, so only a failure, or a read the view refused, stops it
  	// here: the cell starts again with no history
  	analysis.begun = false;
  	analysis.start = cell;
  	return analysis.failed;
  	}
  	this->fifo_.push(FrameReady{analysis.id, std::move(frame)});
  	analysis.last = cell;
  	analysis.next = cell + 1;
  	return true;

• The one-way FIFO and the slot:

  sendspin-server-cpp/src/group_audio.h

  Lines 113 to 230 in 87821e0

  	class GroupToCoreFifo {
  	struct State;
  	public:
  	/// The most events one handler delivers before it posts itself again
  	static constexpr size_t DRAIN_BUDGET = 64;
  	/// The events the FIFO and a handler's batch have room for before they grow: a few jobs' frames
  	/// and epoch events between two drains
  	static constexpr size_t QUEUED_EVENTS = 16;
  	/// @brief The pushing end, which a job holds on its thread and which may outlive the FIFO
  	class Producer {
  	public:
  	/// @brief Any thread. Once the FIFO is destroyed, the event is dropped.
  	void push(GroupEvent event) const;
  	private:
  	friend class GroupToCoreFifo;
  	Producer(EventLoop& loop, std::shared_ptr<State> state);
  	/// Outlives every producer: the group audio threads are joined before the loop goes
  	EventLoop* loop_;
  	std::shared_ptr<State> state_;
  	};
  	/// @param deliver called on core, in push order
  	GroupToCoreFifo(EventLoop& loop, std::function<void(GroupEvent&)> deliver);
  	/// @brief A posted handler still pending delivers nothing, and a later push is dropped. On
  	/// core.
  	~GroupToCoreFifo();
  	GroupToCoreFifo(const GroupToCoreFifo&) = delete;
  	GroupToCoreFifo& operator=(const GroupToCoreFifo&) = delete;
  	Producer producer() const;
  	private:
  	struct State {
  	std::mutex mutex;
  	/// Its producers bound it (above); room for QUEUED_EVENTS at first
  	Ring<GroupEvent> events{QUEUED_EVENTS, Ring<GroupEvent>::UNLIMITED,
  	MemoryLocation::PREFER_EXTERNAL};
  	bool drain_posted{false};
  	/// The FIFO is gone
  	bool closed{false};
  	/// Core only
  	std::function<void(GroupEvent&)> deliver;
  	/// Core only: the batch a handler delivers
  	Ring<GroupEvent> taking{QUEUED_EVENTS, DRAIN_BUDGET, MemoryLocation::PREFER_EXTERNAL};
  	};
  	static void drain(EventLoop& loop, const std::shared_ptr<State>& state);
  	EventLoop& loop_;
  	std::shared_ptr<State> state_;
  	};
  	// ==============================================================================
  	// Pipeline-set slot
  	// ==============================================================================
  	/// @brief One pipeline instance core wants
  	struct PipelineRequest {
  	sched::InstanceId instance;
  	/// The epoch from_index belongs to
  	sched::EpochId epoch;
  	sched::AudioFormat format;
  	/// The first cell a new instance produces, in epoch; one added mid-epoch catches up to it from
  	/// the staging ring's history
  	int64_t from_index{0};
  	bool operator==(const PipelineRequest&) const = default;
  	};
  	/// @brief One visualizer analysis core wants: the messages of one visualizer stream configuration,
  	/// one cell at a time
  	struct AnalysisRequest {
  	sched::InstanceId instance;
  	/// The epoch from_index belongs to
  	sched::EpochId epoch;
  	VisualizerConfig config;
  	/// The first cell a new analysis produces, in epoch
  	int64_t from_index{0};
  	bool operator==(const AnalysisRequest&) const = default;
  	};
  	/// @brief Every instance core wants: the frame instances and the visualizer analyses
  	struct PipelineSet {
  	std::vector<PipelineRequest> frames;
  	std::vector<AnalysisRequest> analyses;
  	bool operator==(const PipelineSet&) const = default;
  	};
  	/// @brief Latest wins: core publishes the whole set, the pipelines read the latest. Shared by the
  	/// group and its pipelines, whose job may outlive the group.
  	class PipelineSetSlot {
  	public:
  	void publish(std::vector<PipelineRequest> frames, std::vector<AnalysisRequest> analyses = {});
  	/// @return the set, when it changed since the version seen, which it updates
  	std::optional<PipelineSet> read_if_newer(uint64_t& seen) const;
  	private:
  	mutable std::mutex mutex_;
  	PipelineSet set_;
  	uint64_t version_{0};
  	};
  	// ==============================================================================
  	// GroupAudio
  	// ==============================================================================
  	/// @brief One group's pipelines: one instance per output format, each the input through the channel
  	/// mix when the channel count differs, the resampler when the rate differs, the quantize stage and
  	/// its PCM, FLAC or Opus encoder. An instance runs each input segment through the stages that
  	/// segment needs, so its output format and its frames' cell grid hold across set_format(). Each

Evidence

Confirmed by reading: GroupAudio::emit() and the visualizer cell each build a fresh sched::Frame and move it into the FIFO; FrameEncoder fills the empty payload with assign or resize, so every frame allocates.

Direction

The owner's call, since it adds a cross-thread channel (docs/conventions.md, docs/internals.md's channel table). Options:

• A bounded return FIFO per group from core to GroupAudio, which core fills with the payloads its rings drop and the audio threads take from before allocating.
• Placed payload storage per ring (a slab of fixed-size slots sized from the format's largest frame), with a slot released by core and reused by the producer.
• Leave it: the allocation is outside the host's write() path and bounded by the frame rate.

Done when

• In steady state, producing and dropping frames allocates nothing on either thread.
• The new channel, if any, is in docs/internals.md and bounded.