salmanoff/stimBuffApis/livoxGen1/pcloudDataProducer.cpp

#include <config.h>
#include <opts.h>
#include <algorithm>
#include <unistd.h>
#include <iomanip>
#include <user/spMcRingBuffer.h>
#include <componentThread.h>
#include <asynchronousLoop.h>
#include <user/stimulusFrame.h>
#include "pcloudDataProducer.h"
#include "frameAssemblyDesc.h"

namespace smo {
namespace stim_buff {

extern const SmoCallbacks* smoHooksPtr;

// OpenCL kernels are used to collate and produce our StimFrames.
static SpMcRingBuffer::InputEngineConstraints openClInputConstraints(
	static_cast<size_t>(sysconf(_SC_PAGE_SIZE)), sizeof(void *));

PcloudDataProducer::PcloudDataProducer(
	const std::shared_ptr<device::DeviceAttachmentSpec> &deviceAttachmentSpec,
	std::shared_ptr<livoxProto1::Device> &device,
	const PcloudFormatDesc& formatDesc,
	int histbuffMs,
	size_t nDgramsPerStagingBufferFrame)
:	StimulusProducer(
		deviceAttachmentSpec,
		static_cast<size_t>(histbuffMs / CONFIG_STIMBUFF_FRAME_PERIOD_MS),
		openClInputConstraints,
		device->componentThread->getIoService()),
device(device),
formatDesc(formatDesc),
openClCollatingAndMeshingEngine(*this),
assemblyBuffer(
	StagingBuffer::IOEngineConstraints::ioUringConstraints,
	StagingBuffer::IOEngineConstraints::openClInputConstraints,
	nDgramsPerStagingBufferFrame),
ioUringAssemblyEngine(*this, nDgramsPerStagingBufferFrame),
collationBuffer(
	StagingBuffer::IOEngineConstraints::openClInputConstraints,
	StagingBuffer::IOEngineConstraints::openClInputConstraints,
	nDgramsPerStagingBufferFrame)
{
	if (smoHooksPtr->OptionParser_getOptions().verbose)
	{
		std::cout << __func__ << ": assembly buffer : "
			<< assemblyBuffer.stringify()
			<< "\n\tcollation buffer : " << collationBuffer.stringify()
			<< "\n";
	}

	std::cout << __func__ << ": Device's component thread is "
		<< device->componentThread->name << std::endl;

#ifndef CONFIG_WORLD_USE_BODY_THREAD
	if (smoHooksPtr->ComponentThread_getSelf()->id != ComponentThread::WORLD)
#else
	if (smoHooksPtr->ComponentThread_getSelf()->id != ComponentThread::BODY)
#endif
	{
		std::string errMsg = std::string(__func__) +
			": PcloudDataProducer constructor called on non-world/body thread " +
			smoHooksPtr->ComponentThread_getSelf()->name;

		std::cout << errMsg << std::endl;
	//		throw std::runtime_error(errMsg);
	}
}

void PcloudDataProducer::start()
{
	// Call ioUringAssemblyEngine setup() as the first step
	if (!ioUringAssemblyEngine.setup())
	{
		std::cout <<__func__ <<"Failed to setup() "
			<<"IOUringAssemblyEngine.\n";
		return;
	}

	if (!openClCollatingAndMeshingEngine.setup())
	{
		std::cout <<__func__ <<"Failed to setup() "
			<<"OClCollMeshEngine.\n";
		return;
	}

	// Call base class start() as the final step
	StimulusProducer::start();
}

void PcloudDataProducer::stop()
{
	// Call base class stop() as the first step
	StimulusProducer::stop();
	// Call ioUringAssemblyEngine stop() as the final step
	openClCollatingAndMeshingEngine.finalize();
	ioUringAssemblyEngine.finalize();
}

void produceStimFrameAck(void)
{
}

void PcloudDataProducer::stimFrameProductionTimesliceInd()
{
	produceFrameReq({nullptr, nullptr});
}

class PcloudDataProducer::ProduceFrameReq
:	public PostedAsynchronousContinuation<produceFrameReqCbFn>
{
private:
	PcloudDataProducer& stimBuff;
	AsynchronousLoop frameAssemblyResult;
	StimulusFrame& stimulusFrame;

public:
	ProduceFrameReq(
		PcloudDataProducer& buffer,
		const std::shared_ptr<ComponentThread>& caller,
		Callback<produceFrameReqCbFn> cb)
	:	PostedAsynchronousContinuation<produceFrameReqCbFn>(caller, cb),
	stimBuff(buffer),
	frameAssemblyResult(0),
	stimulusFrame(buffer.frames_[0])
	{}

public:
	void callOriginalCallback()
	{
		stimBuff.allowNextStimulusFrame();
		callOriginalCb();
	}

public:
	void produceFrameReq1_doAssemble_posted(
		std::shared_ptr<ProduceFrameReq> context)
	{
		SpinLock::Guard lock(stimBuff.shouldContinueLock);
		if (!stimBuff.shouldContinue)
		{
			callOriginalCallback();
			return;
		}

		stimBuff.ioUringAssemblyEngine.assembleFrameReq(
			{context, std::bind(
				&ProduceFrameReq::produceFrameReq2_assembleDone,
				context.get(), context,
				std::placeholders::_1, std::placeholders::_2)});
	}

	void produceFrameReq2_assembleDone(
		std::shared_ptr<ProduceFrameReq> context,
		bool success, AsynchronousLoop loop)
	{
		SpinLock::Guard lock(stimBuff.shouldContinueLock);
		if (!stimBuff.shouldContinue)
		{
			callOriginalCallback();
			return;
		}

		if (!success)
		{
			std::cerr << __func__ << ": Failed to assemble frame" << std::endl;
			callOriginalCallback();
			return;
		}

		context->frameAssemblyResult = loop;

		stimBuff.openClCollatingAndMeshingEngine.compactCollateAndMeshFrameReq(
			loop, stimulusFrame,
			{context, std::bind(
				&ProduceFrameReq::produceFrameReq3_compactCollateDone,
				context.get(), context,
				std::placeholders::_1, std::placeholders::_2)});
	}

	void produceFrameReq3_compactCollateDone(
		[[maybe_unused]] std::shared_ptr<ProduceFrameReq> context,
		bool success, StimulusFrame& /*stimulusFrame*/)
	{
		SpinLock::Guard lock(stimBuff.shouldContinueLock);
		if (!stimBuff.shouldContinue)
		{
			callOriginalCallback();
			return;
		}

		if (!success) {
			std::cerr << __func__ << ": Failed to compact and collate frame" << std::endl;
		} else {
			std::cout << __func__ << ": Successfully compacted and collated frame" << std::endl;
		}

		// Print kernel execution durations
		auto compactDuration = stimBuff.openClCollatingAndMeshingEngine.getCompactKernelDuration();
		auto collateDuration = stimBuff.openClCollatingAndMeshingEngine.getCollateKernelDuration();
		std::cout << __func__ << ": compactKernelDuration=" << compactDuration.count()
			<< "ms, collateKernelDuration=" << collateDuration.count() << "ms" << std::endl;

		callOriginalCallback();
	}
};

void PcloudDataProducer::produceFrameReq(
	smo::Callback<produceFrameReqCbFn> callback)
{
	/**	EXPLANATION:
	 * We shouldn't acquire the StimulusProducer::shouldContinueLock here because
	 * this function is called from
	 * StimulusProducer::stimFrameProductionTimesliceInd(), which is already
	 * holding the lock.
	 */
	auto caller = smoHooksPtr->ComponentThread_getSelf();
	auto request = std::make_shared<ProduceFrameReq>(
		*this, caller, std::move(callback));

	// Post the doAssemble method to the component thread
	device->componentThread->getIoService().post(
		STC(std::bind(
			&ProduceFrameReq::produceFrameReq1_doAssemble_posted,
			request.get(), request)));
}

} // namespace stim_buff
} // namespace smo