salmanoff/stimBuffApis/livoxGen1/ioUringAssemblyEngine.h

#ifndef _LIVOX_GEN1_IOURING_ASSEMBLY_ENGINE_H
#define _LIVOX_GEN1_IOURING_ASSEMBLY_ENGINE_H

#include <boostAsioLinkageFix.h>
#include <config.h>
#include <cstdint>
#include <cstddef>
#include <memory>
#include <functional>
#include <vector>
#include <chrono>
#include <atomic>
#include <random>
#include <liburing.h>
#include <boost/asio/io_context.hpp>
#include <boost/asio/deadline_timer.hpp>
#include <boost/asio/posix/stream_descriptor.hpp>
#include <livoxProto1/device.h>
#include <spinscale/cps/asynchronousContinuation.h>
#include <spinscale/asynchronousLoop.h>
#include <spinscale/cps/callback.h>
#include <spinscale/syncCancelerForAsyncWork.h>
#include <user/frameAssemblyDesc.h>
#include <user/stagingBuffer.h>

#define IOURINGASSM_ENGN_FRAME_ASSEM_TIMEOUT_MS		\
	(CONFIG_STIMBUFF_FRAME_PERIOD_MS / 2)

namespace smo {
namespace stim_buff {

class PcloudStimulusProducer;

class IoUringAssemblyEngine
{
public:
	explicit IoUringAssemblyEngine(
		PcloudStimulusProducer& parent, size_t nDgramsPerStagingBufferFrame);
	~IoUringAssemblyEngine() = default;

	bool setup();
	void finalize();

	typedef std::function<void(bool, sscl::AsynchronousLoop)> assembleFrameReqCbFn;
	void assembleFrameReq(sscl::cps::Callback<assembleFrameReqCbFn> cb);

	// Telemetry helpers
	static size_t computePointsPerFrame(int returnMode, size_t nDgramsPerFrame)
		{ return livoxProto1::Device::getNPointsPerDgram(returnMode) * nDgramsPerFrame; }
	static bool compactionIsNeeded(uint32_t nSucceeded, uint32_t nTotal)
		{ return nSucceeded != 0 && nTotal != 0 && nSucceeded != nTotal; }

	// Get assembly execution duration in milliseconds
	std::chrono::milliseconds getAssemblyDuration() const;

private:
	typedef std::function<void(void*, int)> resetAndAssembleFrameCbFn;
	void resetAndAssembleFrame(resetAndAssembleFrameCbFn onCqeReady);
	void assemblyCycleComplete();
	bool stop();

private:
	PcloudStimulusProducer& parent;

	// Cached descriptor for reuse across iterations
	std::shared_ptr<FrameAssemblyDesc> frameAssemblyDesc;

	// io_uring infrastructure
	struct io_uring ring;

	// Eventfd for CQE notifications (used with boost's unified loop)
	int eventfdFd;
	std::unique_ptr<boost::asio::posix::stream_descriptor> eventfdDesc;
	uint64_t eventfd_value;  // Buffer for async_read_some
	// Point cloud data socket descriptor
	std::shared_ptr<boost::asio::posix::stream_descriptor> pcloudDataFdDesc;
	std::unique_ptr<StagingBuffer::IoUringPinner> assemblyBufferIoUringPinner;

	// Stall detection timer
	boost::asio::deadline_timer stallTimer;
	// Callback for CQE ntfns (called with user_data+result from each CQE)
	resetAndAssembleFrameCbFn onCqeReadyCallback;
	sscl::SyncCancelerForAsyncWork ioUringAssemblyEngnCanceler;

	size_t nDgramsPerStagingBufferFrame;

	struct SlotAssemblyDesc
	{
		bool assembled;
		struct msghdr msgHdr;
		struct iovec ioVec;
	};

	// Track which slots have been successfully assembled and maintain persistent iovecs
	std::vector<SlotAssemblyDesc> assembledSlotsTracker;

	// Random number generation for dummy slot creation
	std::random_device randomDevice;
	std::mt19937 randomGenerator;

	// Timestamp tracking for assembly execution
	std::chrono::high_resolution_clock::time_point assemblyStartTime;
	std::chrono::high_resolution_clock::time_point assemblyEndTime;

	void fillUnAssembledSlotsWithDummyDgrams();
	void randomDummySlotFiller(sscl::AsynchronousLoop& loop);
	void onEventfdRead(
		const boost::system::error_code& error, std::size_t bytes_transferred);

	class AssembleFrameReq;
	friend class AssembleFrameReq;

public:
	void printSlotBytes(size_t slotIndex, size_t nBytes);
};

} // namespace stim_buff
} // namespace smo

#endif // _LIVOX_GEN1_IOURING_ASSEMBLY_ENGINE_H