hayodea/salmanoff
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

IoUringAssmEngine: Remove dead wood

Browse Source
This commit is contained in:
Hayodea Hekol
2025-11-01 20:18:05 -04:00
parent ba955ef633
commit 972979cc10
2 changed files with 13 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
stimBuffApis/livoxGen1/ioUringAssemblyEngine.cpp
+13 -89
View File
@@ -6,98 +6,22 @@
namespace smo {
namespace stim_buff {
namespace {
struct DummyLivoxEthHeader
{
uint8_t version;
uint8_t slot;
uint8_t id;
uint8_t rsvd;
enum : uint32_t {
INVALID_ERR_CODE = 0xFFFFFFFFu
};
enum : uint8_t {
INVALID_TIMESTAMP_TYPE = 0xFFu,
INVALID_DATA_TYPE = 0xFFu
};
uint8_t version, slot, id, rsvd;
uint32_t err_code;
uint8_t timestamp_type;
uint8_t data_type;
uint8_t timestamp_type, data_type;
uint8_t timestamp[8];
};
constexpr uint32_t kInvalidErrCode = 0xFFFFFFFFu;
constexpr uint8_t kInvalidTimestampType = 0xFFu;
constexpr uint8_t kInvalidDataType = 0xFFu;
}
IoUringAssemblyEngine::IoUringAssemblyEngine()
: udpFd(-1), running(false), paused(false), frameCb(nullptr),
desc(nullptr), frameBase(nullptr), frameStrideBytes(0),
timeoutMs(15), frameIndex(0)
{}
bool IoUringAssemblyEngine::setSocketFromDevice(
const std::shared_ptr<livoxProto1::Device>& device
)
{
if (!device) { return false; }
// Expect device to expose pcloudDataSocketDesc()
udpFd = device->pcloudDataSocketDesc->native_handle();
return (udpFd >= 0);
}
bool IoUringAssemblyEngine::start(
FrameAssemblyDesc& descRef, uint8_t* frameBasePtr,
size_t frameStrideBytesArg, uint32_t timeoutMsArg
)
{
if (udpFd < 0 || !frameBasePtr || descRef.slots.empty())
{ return false; }
desc = &descRef;
frameBase = frameBasePtr;
frameStrideBytes = frameStrideBytesArg;
timeoutMs = timeoutMsArg;
paused.store(false);
running.store(true);
// Placeholder: actual io_uring submission/handling would run in a dedicated thread
// For now, assume synchronous loop stub
postReceives();
handleCqesAndTimeout();
return true;
}
void IoUringAssemblyEngine::pause()
{
paused.store(true);
}
void IoUringAssemblyEngine::resume()
{
if (!running.load())
{ return; }
paused.store(false);
postReceives();
}
void IoUringAssemblyEngine::stop()
{
running.store(false);
}
void IoUringAssemblyEngine::onFrameComplete(FrameCompleteCallback cb)
{
frameCb = std::move(cb);
}
void IoUringAssemblyEngine::postReceives()
{
// Intentionally left minimal; real implementation will submit N RECVMSG SQEs
}
void IoUringAssemblyEngine::handleCqesAndTimeout()
{
// Intentionally left minimal; real implementation will poll CQEs with timeout
cancelIncompleteAndFillDummies();
if (frameCb)
{ frameCb(frameIndex++); }
}
void IoUringAssemblyEngine::cancelIncompleteAndFillDummies()
{
if (!desc)
@@ -108,9 +32,9 @@ void IoUringAssemblyEngine::cancelIncompleteAndFillDummies()
// In the real path, decide from CQE accounting whether slot i completed.
// Here, demonstrate dummy header insertion API.
auto* hdr = reinterpret_cast<DummyLivoxEthHeader*>(desc->slots[i].vaddr);
hdr->err_code = kInvalidErrCode;
hdr->timestamp_type = kInvalidTimestampType;
hdr->data_type = kInvalidDataType;
hdr->err_code = DummyLivoxEthHeader::INVALID_ERR_CODE;
hdr->timestamp_type = DummyLivoxEthHeader::INVALID_TIMESTAMP_TYPE;
hdr->data_type = DummyLivoxEthHeader::INVALID_DATA_TYPE;
}
}
stimBuffApis/livoxGen1/ioUringAssemblyEngine.h
-30
View File
@@ -19,48 +19,18 @@ namespace stim_buff {
class IoUringAssemblyEngine
{
public:
using FrameCompleteCallback = std::function<void(size_t /*frameIndex*/)>;
IoUringAssemblyEngine();
~IoUringAssemblyEngine() = default;
// Configure UDP socket by querying the device's pcloud data socket descriptor
bool setSocketFromDevice(const std::shared_ptr<livoxProto1::Device>& device);
// Start posting N RECVMSG SQEs based on the descriptor; timeoutMs defaults to 15ms
bool start(
FrameAssemblyDesc& desc,
uint8_t* frameBase,
size_t frameStrideBytes,
uint32_t timeoutMs = 15);
void pause();
void resume();
void stop();
void onFrameComplete(FrameCompleteCallback cb);
// Telemetry helpers
static size_t computePointsPerDgram(int returnMode);
static size_t computePointsPerFrame(int returnMode, size_t nDgramsPerFrame)
{ return computePointsPerDgram(returnMode) * nDgramsPerFrame; }
private:
int udpFd;
std::atomic<bool> running;
std::atomic<bool> paused;
FrameCompleteCallback frameCb;
// Cached descriptor for reuse across iterations
FrameAssemblyDesc* desc;
uint8_t* frameBase;
size_t frameStrideBytes;
uint32_t timeoutMs;
size_t frameIndex;
// Internal helpers
void postReceives();
void handleCqesAndTimeout();
void cancelIncompleteAndFillDummies();
};