engine: begin the liboakengine C ABI facade with the IPC subsystem
- oakengine/export.h establishes the OAKENGINE_API visibility macros; include/oakengine/ipc.h is the first pure-C surface (41 functions: shm, frame slot pool, and the worker IPC messages as POD<->JSON build/parse), implemented in engine/src/capi/ - the IPC implementations move to engine/src/oliveimpl (namespace olive::engine::internal::ipc); engine/render/ipc/*.h are rebuilt as same-name/same-API wrapper classes forwarding across the C boundary - FrameSlotMeta is shared with the C header verbatim so the app/worker wire format (v1) is bit-identical; static_asserts pin sizeof and field offsets - spscringbuffer.h moves to include/oakengine/ as an inline-only header (no symbols, not ABI) - new pure-C test oakengine_ipc_test (make_oakengine_test, no GL) covers shm, frame pool, message round-trips and the layout asserts; full gtest suite stays green (1986 tests)
This commit is contained in:
@@ -14,14 +14,26 @@
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
# The IPC subsystem is migrating behind the liboakengine C ABI facade:
|
||||
# - include/oakengine/*.h public C API (+ the inline SPSC header)
|
||||
# - src/oliveimpl/render/ipc/*.h internal implementation headers
|
||||
# - src/capi/ipc.cpp C ABI implementation
|
||||
# - render/ipc/*.{h,cpp} implementation sources and the
|
||||
# consumer-side C++ wrapper headers
|
||||
set(OLIVE_SOURCES
|
||||
${OLIVE_SOURCES}
|
||||
include/oakengine/export.h
|
||||
include/oakengine/ipc.h
|
||||
include/oakengine/spscringbuffer.h
|
||||
src/capi/ipc.cpp
|
||||
src/oliveimpl/render/ipc/frameslotpool.h
|
||||
src/oliveimpl/render/ipc/ipcmessage.h
|
||||
src/oliveimpl/render/ipc/sharedmemoryregion.h
|
||||
render/ipc/frameslotpool.cpp
|
||||
render/ipc/frameslotpool.h
|
||||
render/ipc/ipcmessage.cpp
|
||||
render/ipc/ipcmessage.h
|
||||
render/ipc/sharedmemoryregion.cpp
|
||||
render/ipc/sharedmemoryregion.h
|
||||
render/ipc/spscringbuffer.h
|
||||
PARENT_SCOPE
|
||||
)
|
||||
)
|
||||
|
||||
@@ -18,12 +18,16 @@
|
||||
|
||||
***/
|
||||
|
||||
#include "frameslotpool.h"
|
||||
#include "oliveimpl/render/ipc/frameslotpool.h"
|
||||
|
||||
#include <cstring>
|
||||
|
||||
namespace olive
|
||||
{
|
||||
namespace engine
|
||||
{
|
||||
namespace internal
|
||||
{
|
||||
namespace ipc
|
||||
{
|
||||
|
||||
@@ -45,9 +49,9 @@ size_t FrameSlotPool::bytes_needed(uint32_t slot_count, size_t slot_data_bytes)
|
||||
const uint32_t ring_cap = ring_capacity(slot_count);
|
||||
size_t total = align_up(sizeof(Header), k_align);
|
||||
total +=
|
||||
align_up(SpscRingBuffer::bytes_needed(ring_cap), k_align); // free ring
|
||||
align_up(olive::ipc::SpscRingBuffer::bytes_needed(ring_cap), k_align); // free ring
|
||||
total +=
|
||||
align_up(SpscRingBuffer::bytes_needed(ring_cap), k_align); // ready ring
|
||||
align_up(olive::ipc::SpscRingBuffer::bytes_needed(ring_cap), k_align); // ready ring
|
||||
total +=
|
||||
align_up(sizeof(FrameSlotMeta) * slot_count, k_align); // metadata array
|
||||
total += align_up(slot_data_bytes, k_align) * slot_count; // pixel data blocks
|
||||
@@ -67,10 +71,10 @@ FrameSlotPool FrameSlotPool::create(void *mem, uint32_t slot_count,
|
||||
offset += align_up(sizeof(Header), k_align);
|
||||
|
||||
const size_t free_off = offset;
|
||||
offset += align_up(SpscRingBuffer::bytes_needed(ring_cap), k_align);
|
||||
offset += align_up(olive::ipc::SpscRingBuffer::bytes_needed(ring_cap), k_align);
|
||||
|
||||
const size_t ready_off = offset;
|
||||
offset += align_up(SpscRingBuffer::bytes_needed(ring_cap), k_align);
|
||||
offset += align_up(olive::ipc::SpscRingBuffer::bytes_needed(ring_cap), k_align);
|
||||
|
||||
const size_t meta_off = offset;
|
||||
offset += align_up(sizeof(FrameSlotMeta) * slot_count, k_align);
|
||||
@@ -86,8 +90,8 @@ FrameSlotPool FrameSlotPool::create(void *mem, uint32_t slot_count,
|
||||
pool.header_->meta_offset = meta_off;
|
||||
pool.header_->data_offset = data_off;
|
||||
|
||||
pool.free_ring_ = SpscRingBuffer::create(pool.base_ + free_off, ring_cap);
|
||||
pool.ready_ring_ = SpscRingBuffer::create(pool.base_ + ready_off, ring_cap);
|
||||
pool.free_ring_ = olive::ipc::SpscRingBuffer::create(pool.base_ + free_off, ring_cap);
|
||||
pool.ready_ring_ = olive::ipc::SpscRingBuffer::create(pool.base_ + ready_off, ring_cap);
|
||||
pool.meta_ = reinterpret_cast<FrameSlotMeta *>(pool.base_ + meta_off);
|
||||
pool.data_ = pool.base_ + data_off;
|
||||
|
||||
@@ -115,9 +119,9 @@ FrameSlotPool FrameSlotPool::attach(void *mem)
|
||||
}
|
||||
|
||||
pool.free_ring_ =
|
||||
SpscRingBuffer::attach(pool.base_ + pool.header_->free_ring_offset);
|
||||
olive::ipc::SpscRingBuffer::attach(pool.base_ + pool.header_->free_ring_offset);
|
||||
pool.ready_ring_ =
|
||||
SpscRingBuffer::attach(pool.base_ + pool.header_->ready_ring_offset);
|
||||
olive::ipc::SpscRingBuffer::attach(pool.base_ + pool.header_->ready_ring_offset);
|
||||
pool.meta_ = reinterpret_cast<FrameSlotMeta *>(pool.base_ +
|
||||
pool.header_->meta_offset);
|
||||
pool.data_ = pool.base_ + pool.header_->data_offset;
|
||||
@@ -176,4 +180,6 @@ bool FrameSlotPool::release(uint32_t index)
|
||||
}
|
||||
|
||||
} // namespace ipc
|
||||
} // namespace internal
|
||||
} // namespace engine
|
||||
} // namespace olive
|
||||
@@ -24,7 +24,7 @@
|
||||
#include <cstddef>
|
||||
#include <cstdint>
|
||||
|
||||
#include "spscringbuffer.h"
|
||||
#include "oakengine/ipc.h"
|
||||
|
||||
namespace olive
|
||||
{
|
||||
@@ -34,150 +34,196 @@ namespace ipc
|
||||
/**
|
||||
* @brief Per-slot metadata describing the frame currently occupying a slot.
|
||||
*
|
||||
* Trivially-copyable POD that lives in shared memory alongside the pixel data. Carries everything
|
||||
* the consumer needs to reconstruct an olive::Frame without any out-of-band information. We store
|
||||
* the Rational timestamp as an explicit numerator/denominator pair to stay POD (olive::Rational is
|
||||
* not guaranteed shared-memory-safe).
|
||||
* Trivially-copyable POD that lives in shared memory alongside the pixel data, part of the
|
||||
* version-1 wire protocol with the render worker. This is the C ABI oak_frame_slot_meta struct,
|
||||
* aliased so the shared-memory layout is defined exactly once, in oakengine/ipc.h.
|
||||
*/
|
||||
struct FrameSlotMeta {
|
||||
int64_t id; ///< Caller-defined tag (e.g. ticket id, or footage stream hash).
|
||||
int64_t time_num; ///< Frame timestamp numerator.
|
||||
int64_t time_den; ///< Frame timestamp denominator.
|
||||
int32_t width;
|
||||
int32_t height;
|
||||
int32_t format; ///< olive::PixelFormat::Format value.
|
||||
int32_t channel_count;
|
||||
int32_t linesize; ///< Bytes per scanline (stride).
|
||||
int32_t data_size; ///< Valid bytes written into the slot's data block.
|
||||
char colorspace[128]; ///< Input colorspace name for color-managed footage.
|
||||
};
|
||||
typedef oak_frame_slot_meta FrameSlotMeta;
|
||||
|
||||
/**
|
||||
* @brief A fixed-size pool of equal-sized frame slots in shared memory, with lock-free hand-off.
|
||||
*
|
||||
* Consumer-side wrapper over the liboakengine C ABI: the object only holds an opaque
|
||||
* OakFrameSlotPool handle and forwards every call across the C boundary. The public API is
|
||||
* unchanged from the original implementation; see oakengine/ipc.h for the protocol description.
|
||||
*
|
||||
* One pool models a single direction of frame flow (e.g. worker -> main for rendered output, or
|
||||
* main -> worker for decoded input). Ownership of a slot is transferred via two SPSC ring buffers
|
||||
* of slot indices, so no mutex is ever taken:
|
||||
*
|
||||
* - free_ring: indices of slots available to the FILLER. The drainer returns slots here.
|
||||
* - ready_ring: indices of slots holding a published frame, produced by the FILLER for the
|
||||
* DRAINER to consume.
|
||||
*
|
||||
* Lifecycle (filler = producer of frames, drainer = consumer of frames):
|
||||
* filler: Acquire() -> pop a free index -> write meta + pixels -> Publish() -> push to ready
|
||||
* drainer: Consume() -> pop a ready index -> read meta + pixels -> Release() -> push to free
|
||||
*
|
||||
* Because each ring has exactly one producer and one consumer (the filler owns free.Pop +
|
||||
* ready.Push, the drainer owns ready.Pop + free.Push), the SPSC invariant holds and the whole
|
||||
* exchange is lock-free.
|
||||
*
|
||||
* All slots are sized to `slot_data_bytes`, computed for the maximum supported frame (e.g. 8K RGBA
|
||||
* half-float). Frames smaller than that simply use a prefix of the slot.
|
||||
*
|
||||
* The pool does NOT own the memory; it is constructed over a SharedMemoryRegion mapping. Use
|
||||
* BytesNeeded() to size that region.
|
||||
* main -> worker for decoded input). The pool does NOT own the memory; it is constructed over a
|
||||
* SharedMemoryRegion mapping. Use bytes_needed() to size that region.
|
||||
*/
|
||||
class FrameSlotPool {
|
||||
public:
|
||||
FrameSlotPool() = default;
|
||||
|
||||
FrameSlotPool(const FrameSlotPool &rhs)
|
||||
: handle_(oakengine_ipc_framepool_copy(rhs.handle_))
|
||||
{
|
||||
}
|
||||
|
||||
FrameSlotPool(FrameSlotPool &&rhs) noexcept
|
||||
: handle_(rhs.handle_)
|
||||
{
|
||||
rhs.handle_ = nullptr;
|
||||
}
|
||||
|
||||
~FrameSlotPool()
|
||||
{
|
||||
oakengine_ipc_framepool_free(handle_);
|
||||
}
|
||||
|
||||
FrameSlotPool &operator=(const FrameSlotPool &rhs)
|
||||
{
|
||||
if (this != &rhs) {
|
||||
oakengine_ipc_framepool_free(handle_);
|
||||
handle_ = oakengine_ipc_framepool_copy(rhs.handle_);
|
||||
}
|
||||
return *this;
|
||||
}
|
||||
|
||||
FrameSlotPool &operator=(FrameSlotPool &&rhs) noexcept
|
||||
{
|
||||
if (this != &rhs) {
|
||||
oakengine_ipc_framepool_free(handle_);
|
||||
handle_ = rhs.handle_;
|
||||
rhs.handle_ = nullptr;
|
||||
}
|
||||
return *this;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Total bytes a region must provide to back a pool of `slot_count` x `slot_data_bytes`.
|
||||
*/
|
||||
static size_t bytes_needed(uint32_t slot_count, size_t slot_data_bytes);
|
||||
static size_t bytes_needed(uint32_t slot_count, size_t slot_data_bytes)
|
||||
{
|
||||
return oakengine_ipc_framepool_bytes_needed(slot_count, slot_data_bytes);
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Lay out and initialize a brand-new pool over `mem` (owner side, once).
|
||||
*
|
||||
* Initializes both rings, seeds the free ring with every slot index, and zeroes metadata.
|
||||
* `mem` must provide at least BytesNeeded(slot_count, slot_data_bytes) bytes.
|
||||
* `mem` must provide at least bytes_needed(slot_count, slot_data_bytes) bytes.
|
||||
*/
|
||||
static FrameSlotPool create(void *mem, uint32_t slot_count,
|
||||
size_t slot_data_bytes);
|
||||
size_t slot_data_bytes)
|
||||
{
|
||||
return from_handle(oakengine_ipc_framepool_create(mem, slot_count,
|
||||
slot_data_bytes));
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Map an existing, already-initialized pool (peer side).
|
||||
*
|
||||
* Reads slot_count/slot_data_bytes from the in-memory header written by Create().
|
||||
* Reads slot_count/slot_data_bytes from the in-memory header written by create().
|
||||
*/
|
||||
static FrameSlotPool attach(void *mem);
|
||||
static FrameSlotPool attach(void *mem)
|
||||
{
|
||||
return from_handle(oakengine_ipc_framepool_attach(mem));
|
||||
}
|
||||
|
||||
bool is_valid() const
|
||||
{
|
||||
return header_ != nullptr;
|
||||
return oakengine_ipc_framepool_is_valid(handle_) != 0;
|
||||
}
|
||||
|
||||
uint32_t slot_count() const;
|
||||
size_t slot_data_bytes() const;
|
||||
uint32_t slot_count() const
|
||||
{
|
||||
return oakengine_ipc_framepool_slot_count(handle_);
|
||||
}
|
||||
|
||||
size_t slot_data_bytes() const
|
||||
{
|
||||
return oakengine_ipc_framepool_slot_data_bytes(handle_);
|
||||
}
|
||||
|
||||
// ---- Filler side ----
|
||||
|
||||
/**
|
||||
* @brief Take ownership of a free slot. Returns false (and leaves *index untouched) if none free.
|
||||
*/
|
||||
bool acquire(uint32_t *index);
|
||||
bool acquire(uint32_t *index)
|
||||
{
|
||||
return oakengine_ipc_framepool_acquire(handle_, index) != 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Pointer to a slot's pixel data block (slot_data_bytes available).
|
||||
*/
|
||||
void *slot_data(uint32_t index);
|
||||
void *slot_data(uint32_t index)
|
||||
{
|
||||
return oakengine_ipc_framepool_slot_data(handle_, index);
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Mutable metadata for a slot. Filler writes this before Publish().
|
||||
* @brief Mutable metadata for a slot. Filler writes this before publish().
|
||||
*/
|
||||
FrameSlotMeta *meta(uint32_t index);
|
||||
FrameSlotMeta *meta(uint32_t index)
|
||||
{
|
||||
return oakengine_ipc_framepool_meta(handle_, index);
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Publish a filled slot to the drainer. Must follow a successful Acquire() of `index`.
|
||||
* @brief Publish a filled slot to the drainer. Must follow a successful acquire() of `index`.
|
||||
*/
|
||||
bool publish(uint32_t index);
|
||||
bool publish(uint32_t index)
|
||||
{
|
||||
return oakengine_ipc_framepool_publish(handle_, index) != 0;
|
||||
}
|
||||
|
||||
// ---- Drainer side ----
|
||||
|
||||
/**
|
||||
* @brief Take the next published slot. Returns false if nothing is ready.
|
||||
*/
|
||||
bool consume(uint32_t *index);
|
||||
|
||||
/**
|
||||
* @brief Return a consumed slot to the free pool for reuse. Must follow Consume() of `index`.
|
||||
*/
|
||||
bool release(uint32_t index);
|
||||
|
||||
const FrameSlotMeta *meta(uint32_t index) const;
|
||||
const void *slot_data(uint32_t index) const;
|
||||
|
||||
public:
|
||||
FrameSlotPool() = default;
|
||||
|
||||
private:
|
||||
struct Header {
|
||||
uint32_t magic;
|
||||
uint32_t slot_count;
|
||||
uint64_t slot_data_bytes;
|
||||
// Byte offsets from the start of the segment to each sub-region.
|
||||
uint64_t free_ring_offset;
|
||||
uint64_t ready_ring_offset;
|
||||
uint64_t meta_offset;
|
||||
uint64_t data_offset;
|
||||
};
|
||||
|
||||
static constexpr uint32_t k_magic = 0x4F4B5350; // 'OKSP'
|
||||
|
||||
// Ring capacity must exceed slot_count by one because a ring can hold at most capacity-1 entries
|
||||
// and we need to be able to enqueue every slot at once.
|
||||
static uint32_t ring_capacity(uint32_t slot_count)
|
||||
bool consume(uint32_t *index)
|
||||
{
|
||||
return slot_count + 1;
|
||||
return oakengine_ipc_framepool_consume(handle_, index) != 0;
|
||||
}
|
||||
|
||||
uint8_t *base_ = nullptr;
|
||||
Header *header_ = nullptr;
|
||||
SpscRingBuffer *free_ring_ = nullptr;
|
||||
SpscRingBuffer *ready_ring_ = nullptr;
|
||||
FrameSlotMeta *meta_ = nullptr;
|
||||
uint8_t *data_ = nullptr;
|
||||
/**
|
||||
* @brief Return a consumed slot to the free pool for reuse. Must follow consume() of `index`.
|
||||
*/
|
||||
bool release(uint32_t index)
|
||||
{
|
||||
return oakengine_ipc_framepool_release(handle_, index) != 0;
|
||||
}
|
||||
|
||||
const FrameSlotMeta *meta(uint32_t index) const
|
||||
{
|
||||
return oakengine_ipc_framepool_meta_const(handle_, index);
|
||||
}
|
||||
|
||||
const void *slot_data(uint32_t index) const
|
||||
{
|
||||
return oakengine_ipc_framepool_slot_data_const(handle_, index);
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief The wrapped C handle, for cross-type wrappers and direct C API use
|
||||
*/
|
||||
OakFrameSlotPool *handle() const
|
||||
{
|
||||
return handle_;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Wraps an owned C handle (takes ownership)
|
||||
*/
|
||||
static FrameSlotPool from_handle(OakFrameSlotPool *handle)
|
||||
{
|
||||
return FrameSlotPool(handle);
|
||||
}
|
||||
|
||||
private:
|
||||
explicit FrameSlotPool(OakFrameSlotPool *handle)
|
||||
: handle_(handle)
|
||||
{
|
||||
}
|
||||
|
||||
OakFrameSlotPool *handle_ = nullptr;
|
||||
};
|
||||
|
||||
} // namespace ipc
|
||||
} // namespace olive
|
||||
|
||||
#endif // OAK_IPC_FRAMESLOTPOOL_H
|
||||
#endif // OAK_IPC_FRAMESLOTPOOL_H
|
||||
|
||||
@@ -18,7 +18,7 @@
|
||||
|
||||
***/
|
||||
|
||||
#include "ipcmessage.h"
|
||||
#include "oliveimpl/render/ipc/ipcmessage.h"
|
||||
|
||||
#include <QJsonArray>
|
||||
#include <QJsonDocument>
|
||||
@@ -26,6 +26,10 @@
|
||||
|
||||
namespace olive
|
||||
{
|
||||
namespace engine
|
||||
{
|
||||
namespace internal
|
||||
{
|
||||
namespace ipc
|
||||
{
|
||||
|
||||
@@ -227,4 +231,6 @@ bool LoadGraphMsg::from_json(const QJsonObject &o, LoadGraphMsg *out)
|
||||
}
|
||||
|
||||
} // namespace ipc
|
||||
} // namespace internal
|
||||
} // namespace engine
|
||||
} // namespace olive
|
||||
|
||||
+238
-23
@@ -21,12 +21,17 @@
|
||||
#ifndef OAK_IPC_IPCMESSAGE_H
|
||||
#define OAK_IPC_IPCMESSAGE_H
|
||||
|
||||
#include <algorithm>
|
||||
#include <cstdint>
|
||||
#include <cstring>
|
||||
#include <QByteArray>
|
||||
#include <QJsonDocument>
|
||||
#include <QJsonObject>
|
||||
#include <QString>
|
||||
#include <QVector>
|
||||
|
||||
#include "oakengine/ipc.h"
|
||||
|
||||
class QIODevice;
|
||||
|
||||
namespace olive
|
||||
@@ -43,6 +48,10 @@ namespace ipc
|
||||
* bulk pixel data travels through the shared-memory FrameSlotPool, and the (potentially large)
|
||||
* serialized node graph travels via a temporary file referenced by path.
|
||||
*
|
||||
* Consumer-side wrapper over the liboakengine C ABI: the typed builders/parsers below convert
|
||||
* through the oakengine_ipc_*_to_json/parse functions, and the QIODevice framing stays inline
|
||||
* here. The public API is unchanged from the original implementation.
|
||||
*
|
||||
* Every message object has a "type" string field. Directionality (M = main, W = worker):
|
||||
* "handshake" M<->W Negotiate protocol version and announce shared-memory key/geometry.
|
||||
* "load_graph" M ->W Path to a temporary file holding the serialized node graph.
|
||||
@@ -55,23 +64,58 @@ namespace ipc
|
||||
*/
|
||||
namespace msgtype
|
||||
{
|
||||
constexpr const char *k_handshake = "handshake";
|
||||
constexpr const char *k_load_graph = "load_graph";
|
||||
constexpr const char *k_render_frame = "render_frame";
|
||||
constexpr const char *k_frame_ready = "frame_ready";
|
||||
constexpr const char *k_cancel = "cancel";
|
||||
constexpr const char *k_graph_update = "graph_update";
|
||||
constexpr const char *k_shutdown = "shutdown";
|
||||
constexpr const char *k_error = "error";
|
||||
constexpr const char *k_handshake = OAKENGINE_IPC_MSGTYPE_HANDSHAKE;
|
||||
constexpr const char *k_load_graph = OAKENGINE_IPC_MSGTYPE_LOAD_GRAPH;
|
||||
constexpr const char *k_render_frame = OAKENGINE_IPC_MSGTYPE_RENDER_FRAME;
|
||||
constexpr const char *k_frame_ready = OAKENGINE_IPC_MSGTYPE_FRAME_READY;
|
||||
constexpr const char *k_cancel = OAKENGINE_IPC_MSGTYPE_CANCEL;
|
||||
constexpr const char *k_graph_update = OAKENGINE_IPC_MSGTYPE_GRAPH_UPDATE;
|
||||
constexpr const char *k_shutdown = OAKENGINE_IPC_MSGTYPE_SHUTDOWN;
|
||||
constexpr const char *k_error = OAKENGINE_IPC_MSGTYPE_ERROR;
|
||||
} // namespace msgtype
|
||||
|
||||
namespace detail
|
||||
{
|
||||
|
||||
inline void copy_str(const QString &s, char *dst, size_t cap)
|
||||
{
|
||||
const QByteArray utf = s.toUtf8();
|
||||
const size_t n = std::min(size_t(utf.size()), cap - 1);
|
||||
memcpy(dst, utf.constData(), n);
|
||||
dst[n] = '\0';
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Run a C to_json function (buf/size convention) and reparse the compact JSON text.
|
||||
*/
|
||||
template <typename F> QJsonObject via_c_json(F &&to_json)
|
||||
{
|
||||
const int size = to_json(nullptr, 0);
|
||||
QByteArray buf(size + 1, '\0');
|
||||
to_json(buf.data(), size + 1);
|
||||
buf.resize(size);
|
||||
return QJsonDocument::fromJson(buf).object();
|
||||
}
|
||||
|
||||
inline QByteArray compact_json(const QJsonObject &o)
|
||||
{
|
||||
return QJsonDocument(o).toJson(QJsonDocument::Compact);
|
||||
}
|
||||
|
||||
} // namespace detail
|
||||
|
||||
/**
|
||||
* @brief Write one NDJSON message line to `device`.
|
||||
*
|
||||
* Serializes `obj` to compact JSON, appends '\n', and writes the whole line in one call. Returns
|
||||
* true only if the full line was written.
|
||||
*/
|
||||
bool write_message(QIODevice *device, const QJsonObject &obj);
|
||||
inline bool write_message(QIODevice *device, const QJsonObject &obj)
|
||||
{
|
||||
QByteArray line = detail::compact_json(obj);
|
||||
line.append('\n');
|
||||
return device->write(line) == line.size();
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Pull one complete NDJSON line out of `buffer` and parse it.
|
||||
@@ -82,13 +126,46 @@ bool write_message(QIODevice *device, const QJsonObject &obj);
|
||||
* and continue rather than wedge. Supports the typical "append bytes as they arrive, then drain
|
||||
* complete lines" reader loop on a pipe.
|
||||
*/
|
||||
bool read_message(QByteArray *buffer, QJsonObject *out, bool *ok = nullptr);
|
||||
inline bool read_message(QByteArray *buffer, QJsonObject *out, bool *ok = nullptr)
|
||||
{
|
||||
while (true) {
|
||||
const int newline = buffer->indexOf('\n');
|
||||
if (newline < 0) {
|
||||
// No complete line buffered yet.
|
||||
return false;
|
||||
}
|
||||
|
||||
const QByteArray line = buffer->left(newline);
|
||||
buffer->remove(0, newline + 1);
|
||||
|
||||
// Skip blank lines silently (e.g. a stray newline) without flagging an error.
|
||||
if (line.trimmed().isEmpty()) {
|
||||
continue;
|
||||
}
|
||||
|
||||
QJsonParseError err;
|
||||
const QJsonDocument doc = QJsonDocument::fromJson(line, &err);
|
||||
if (err.error != QJsonParseError::NoError || !doc.isObject()) {
|
||||
if (ok) {
|
||||
*ok = false;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
*out = doc.object();
|
||||
if (ok) {
|
||||
*ok = true;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
// ---- Typed message builders / parsers -------------------------------------------------------
|
||||
//
|
||||
// Thin helpers that construct or read the QJsonObject for each message type, keeping field names in
|
||||
// one place so main and worker agree. Fields use plain JSON numbers/strings; 64-bit ids are stored
|
||||
// as JSON numbers (doubles exactly represent integers up to 2^53, ample for our counters).
|
||||
// Thin wrappers that convert each struct to/from the C ABI POD form and let the library build or
|
||||
// read the JSON, keeping field names in one place so main and worker agree. Fields use plain JSON
|
||||
// numbers/strings; 64-bit ids are stored as JSON numbers (doubles exactly represent integers up
|
||||
// to 2^53, ample for our counters).
|
||||
|
||||
struct HandshakeMsg {
|
||||
int protocol_version = 0;
|
||||
@@ -100,8 +177,38 @@ struct HandshakeMsg {
|
||||
qint64 slot_data_bytes = 0; ///< Per-output-slot pixel block size.
|
||||
qint64 input_slot_data_bytes = 0; ///< Per-input-slot pixel block size.
|
||||
|
||||
QJsonObject to_json() const;
|
||||
static bool from_json(const QJsonObject &o, HandshakeMsg *out);
|
||||
QJsonObject to_json() const
|
||||
{
|
||||
oak_ipc_handshake c;
|
||||
c.protocol_version = protocol_version;
|
||||
detail::copy_str(shm_key, c.shm_key, sizeof(c.shm_key));
|
||||
detail::copy_str(input_shm_key, c.input_shm_key,
|
||||
sizeof(c.input_shm_key));
|
||||
c.input_slots = input_slots;
|
||||
c.output_slots = output_slots;
|
||||
c.slot_data_bytes = slot_data_bytes;
|
||||
c.input_slot_data_bytes = input_slot_data_bytes;
|
||||
return detail::via_c_json([&](char *buf, int size) {
|
||||
return oakengine_ipc_handshake_to_json(&c, buf, size);
|
||||
});
|
||||
}
|
||||
|
||||
static bool from_json(const QJsonObject &o, HandshakeMsg *out)
|
||||
{
|
||||
oak_ipc_handshake c;
|
||||
if (!oakengine_ipc_handshake_parse(detail::compact_json(o).constData(),
|
||||
&c)) {
|
||||
return false;
|
||||
}
|
||||
out->protocol_version = c.protocol_version;
|
||||
out->shm_key = QString::fromUtf8(c.shm_key);
|
||||
out->input_shm_key = QString::fromUtf8(c.input_shm_key);
|
||||
out->input_slots = c.input_slots;
|
||||
out->output_slots = c.output_slots;
|
||||
out->slot_data_bytes = c.slot_data_bytes;
|
||||
out->input_slot_data_bytes = c.input_slot_data_bytes;
|
||||
return true;
|
||||
}
|
||||
};
|
||||
|
||||
struct RenderFrameMsg {
|
||||
@@ -129,30 +236,138 @@ struct RenderFrameMsg {
|
||||
QString color_view;
|
||||
QString color_look;
|
||||
|
||||
QJsonObject to_json() const;
|
||||
static bool from_json(const QJsonObject &o, RenderFrameMsg *out);
|
||||
QJsonObject to_json() const
|
||||
{
|
||||
oak_ipc_render_frame c;
|
||||
c.ticket_id = ticket_id;
|
||||
detail::copy_str(node_uuid, c.node_uuid, sizeof(c.node_uuid));
|
||||
c.time_num = time_num;
|
||||
c.time_den = time_den;
|
||||
c.width = width;
|
||||
c.height = height;
|
||||
c.format = format;
|
||||
c.channel_count = channel_count;
|
||||
c.mode = mode;
|
||||
c.input_slot = input_slot;
|
||||
c.input_slot_count = std::min(int(input_slots.size()),
|
||||
OAK_IPC_INPUT_SLOTS_CAP);
|
||||
for (int i = 0; i < c.input_slot_count; i++) {
|
||||
c.input_slots[i] = input_slots.at(i);
|
||||
}
|
||||
c.has_color_transform = has_color_transform ? 1 : 0;
|
||||
c.color_is_display = color_is_display ? 1 : 0;
|
||||
detail::copy_str(color_output, c.color_output,
|
||||
sizeof(c.color_output));
|
||||
detail::copy_str(color_view, c.color_view, sizeof(c.color_view));
|
||||
detail::copy_str(color_look, c.color_look, sizeof(c.color_look));
|
||||
return detail::via_c_json([&](char *buf, int size) {
|
||||
return oakengine_ipc_render_frame_to_json(&c, buf, size);
|
||||
});
|
||||
}
|
||||
|
||||
static bool from_json(const QJsonObject &o, RenderFrameMsg *out)
|
||||
{
|
||||
oak_ipc_render_frame c;
|
||||
if (!oakengine_ipc_render_frame_parse(
|
||||
detail::compact_json(o).constData(), &c)) {
|
||||
return false;
|
||||
}
|
||||
out->ticket_id = c.ticket_id;
|
||||
out->node_uuid = QString::fromUtf8(c.node_uuid);
|
||||
out->time_num = c.time_num;
|
||||
out->time_den = c.time_den;
|
||||
out->width = c.width;
|
||||
out->height = c.height;
|
||||
out->format = c.format;
|
||||
out->channel_count = c.channel_count;
|
||||
out->mode = c.mode;
|
||||
out->input_slot = c.input_slot;
|
||||
out->input_slots.clear();
|
||||
for (int i = 0; i < c.input_slot_count; i++) {
|
||||
out->input_slots.append(c.input_slots[i]);
|
||||
}
|
||||
out->has_color_transform = c.has_color_transform != 0;
|
||||
out->color_is_display = c.color_is_display != 0;
|
||||
out->color_output = QString::fromUtf8(c.color_output);
|
||||
out->color_view = QString::fromUtf8(c.color_view);
|
||||
out->color_look = QString::fromUtf8(c.color_look);
|
||||
return true;
|
||||
}
|
||||
};
|
||||
|
||||
struct FrameReadyMsg {
|
||||
qint64 ticket_id = 0;
|
||||
int output_slot = 0; ///< Index into the worker->main output FrameSlotPool.
|
||||
|
||||
QJsonObject to_json() const;
|
||||
static bool from_json(const QJsonObject &o, FrameReadyMsg *out);
|
||||
QJsonObject to_json() const
|
||||
{
|
||||
oak_ipc_frame_ready c;
|
||||
c.ticket_id = ticket_id;
|
||||
c.output_slot = output_slot;
|
||||
return detail::via_c_json([&](char *buf, int size) {
|
||||
return oakengine_ipc_frame_ready_to_json(&c, buf, size);
|
||||
});
|
||||
}
|
||||
|
||||
static bool from_json(const QJsonObject &o, FrameReadyMsg *out)
|
||||
{
|
||||
oak_ipc_frame_ready c;
|
||||
if (!oakengine_ipc_frame_ready_parse(
|
||||
detail::compact_json(o).constData(), &c)) {
|
||||
return false;
|
||||
}
|
||||
out->ticket_id = c.ticket_id;
|
||||
out->output_slot = c.output_slot;
|
||||
return true;
|
||||
}
|
||||
};
|
||||
|
||||
struct CancelMsg {
|
||||
qint64 ticket_id = 0;
|
||||
|
||||
QJsonObject to_json() const;
|
||||
static bool from_json(const QJsonObject &o, CancelMsg *out);
|
||||
QJsonObject to_json() const
|
||||
{
|
||||
oak_ipc_cancel c;
|
||||
c.ticket_id = ticket_id;
|
||||
return detail::via_c_json([&](char *buf, int size) {
|
||||
return oakengine_ipc_cancel_to_json(&c, buf, size);
|
||||
});
|
||||
}
|
||||
|
||||
static bool from_json(const QJsonObject &o, CancelMsg *out)
|
||||
{
|
||||
oak_ipc_cancel c;
|
||||
if (!oakengine_ipc_cancel_parse(detail::compact_json(o).constData(),
|
||||
&c)) {
|
||||
return false;
|
||||
}
|
||||
out->ticket_id = c.ticket_id;
|
||||
return true;
|
||||
}
|
||||
};
|
||||
|
||||
struct LoadGraphMsg {
|
||||
QString path; ///< Temporary file holding the serialized node graph.
|
||||
|
||||
QJsonObject to_json() const;
|
||||
static bool from_json(const QJsonObject &o, LoadGraphMsg *out);
|
||||
QJsonObject to_json() const
|
||||
{
|
||||
oak_ipc_load_graph c;
|
||||
detail::copy_str(path, c.path, sizeof(c.path));
|
||||
return detail::via_c_json([&](char *buf, int size) {
|
||||
return oakengine_ipc_load_graph_to_json(&c, buf, size);
|
||||
});
|
||||
}
|
||||
|
||||
static bool from_json(const QJsonObject &o, LoadGraphMsg *out)
|
||||
{
|
||||
oak_ipc_load_graph c;
|
||||
if (!oakengine_ipc_load_graph_parse(detail::compact_json(o).constData(),
|
||||
&c)) {
|
||||
return false;
|
||||
}
|
||||
out->path = QString::fromUtf8(c.path);
|
||||
return true;
|
||||
}
|
||||
};
|
||||
|
||||
} // namespace ipc
|
||||
|
||||
@@ -18,7 +18,7 @@
|
||||
|
||||
***/
|
||||
|
||||
#include "sharedmemoryregion.h"
|
||||
#include "oliveimpl/render/ipc/sharedmemoryregion.h"
|
||||
|
||||
#include <QtGlobal>
|
||||
|
||||
@@ -35,6 +35,10 @@
|
||||
|
||||
namespace olive
|
||||
{
|
||||
namespace engine
|
||||
{
|
||||
namespace internal
|
||||
{
|
||||
namespace ipc
|
||||
{
|
||||
|
||||
@@ -228,4 +232,6 @@ void SharedMemoryRegion::close()
|
||||
#endif
|
||||
|
||||
} // namespace ipc
|
||||
} // namespace internal
|
||||
} // namespace engine
|
||||
} // namespace olive
|
||||
@@ -24,6 +24,8 @@
|
||||
#include <cstddef>
|
||||
#include <QString>
|
||||
|
||||
#include "oakengine/ipc.h"
|
||||
|
||||
namespace olive
|
||||
{
|
||||
namespace ipc
|
||||
@@ -32,27 +34,34 @@ namespace ipc
|
||||
/**
|
||||
* @brief A named, fixed-size shared memory segment mapped into the process address space.
|
||||
*
|
||||
* One process Create()s the segment (owner); the peer process Attach()es to it by the same key.
|
||||
* The mapping is a raw contiguous byte range accessible via data() — the IPC ring buffers and frame
|
||||
* slot pools are laid out inside it. Nothing here is locked; synchronization is entirely the
|
||||
* caller's responsibility via the lock-free structures placed in the mapping.
|
||||
* Consumer-side wrapper over the liboakengine C ABI: the object only holds an opaque
|
||||
* OakSharedMemoryRegion handle and forwards every call across the C boundary. The public API is
|
||||
* unchanged from the original implementation.
|
||||
*
|
||||
* We deliberately use the raw OS primitives (POSIX shm_open + mmap, Windows CreateFileMapping +
|
||||
* MapViewOfFile) rather than QSharedMemory: QSharedMemory carries an implicit semaphore and a 1-byte
|
||||
* header convention, attaches/detaches with reference counting we don't want, and historically has
|
||||
* cross-platform lifetime quirks. For a render pipeline pushing large frames we want a plain mmap.
|
||||
* One process open()s the segment with k_create (owner); the peer process open()s it by the same
|
||||
* key with k_attach. The mapping is a raw contiguous byte range accessible via data() — the IPC
|
||||
* ring buffers and frame slot pools are laid out inside it. Nothing here is locked;
|
||||
* synchronization is entirely the caller's responsibility via the lock-free structures placed in
|
||||
* the mapping.
|
||||
*/
|
||||
class SharedMemoryRegion {
|
||||
public:
|
||||
enum Mode {
|
||||
/// Create (and own) the segment. Fails if it already exists; unlinks on destruction.
|
||||
k_create,
|
||||
k_create = OAK_IPC_SHM_MODE_CREATE,
|
||||
/// Attach to a segment created by the peer. Does not unlink on destruction.
|
||||
k_attach
|
||||
k_attach = OAK_IPC_SHM_MODE_ATTACH
|
||||
};
|
||||
|
||||
SharedMemoryRegion();
|
||||
~SharedMemoryRegion();
|
||||
SharedMemoryRegion()
|
||||
: handle_(oakengine_ipc_shm_create())
|
||||
{
|
||||
}
|
||||
|
||||
~SharedMemoryRegion()
|
||||
{
|
||||
oakengine_ipc_shm_free(handle_);
|
||||
}
|
||||
|
||||
SharedMemoryRegion(const SharedMemoryRegion &) = delete;
|
||||
SharedMemoryRegion &operator=(const SharedMemoryRegion &) = delete;
|
||||
@@ -63,26 +72,36 @@ public:
|
||||
* `key` is a short identifier (no leading slash needed; the platform prefix is added internally).
|
||||
* Returns true on success. On failure, error() carries a human-readable reason.
|
||||
*/
|
||||
bool open(const QString &key, size_t size, Mode mode);
|
||||
bool open(const QString &key, size_t size, Mode mode)
|
||||
{
|
||||
const bool ok = oakengine_ipc_shm_open(
|
||||
handle_, key.toUtf8().constData(), size,
|
||||
static_cast<oak_ipc_shm_mode>(mode)) != 0;
|
||||
refresh_caches();
|
||||
return ok;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Unmap and (if owner) unlink the segment. Called automatically by the destructor.
|
||||
*/
|
||||
void close();
|
||||
void close()
|
||||
{
|
||||
oakengine_ipc_shm_close(handle_);
|
||||
}
|
||||
|
||||
bool is_valid() const
|
||||
{
|
||||
return data_ != nullptr;
|
||||
return oakengine_ipc_shm_is_valid(handle_) != 0;
|
||||
}
|
||||
|
||||
void *data() const
|
||||
{
|
||||
return data_;
|
||||
return oakengine_ipc_shm_data(handle_);
|
||||
}
|
||||
|
||||
size_t size() const
|
||||
{
|
||||
return size_;
|
||||
return oakengine_ipc_shm_size(handle_);
|
||||
}
|
||||
|
||||
const QString &key() const
|
||||
@@ -100,24 +119,50 @@ public:
|
||||
*
|
||||
* Centralized so the owner and the spawned worker agree on the same name.
|
||||
*/
|
||||
static QString make_key(qint64 owner_pid, int worker_index);
|
||||
static QString make_key(qint64 owner_pid, int worker_index)
|
||||
{
|
||||
const int size = oakengine_ipc_shm_make_key(owner_pid, worker_index,
|
||||
nullptr, 0);
|
||||
QByteArray buf(size + 1, '\0');
|
||||
oakengine_ipc_shm_make_key(owner_pid, worker_index, buf.data(),
|
||||
size + 1);
|
||||
return QString::fromUtf8(buf.constData());
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief The wrapped C handle, for cross-type wrappers and direct C API use
|
||||
*/
|
||||
OakSharedMemoryRegion *handle() const
|
||||
{
|
||||
return handle_;
|
||||
}
|
||||
|
||||
private:
|
||||
QString key_;
|
||||
size_t size_;
|
||||
void *data_;
|
||||
Mode mode_;
|
||||
QString error_;
|
||||
static QString query_string(int (*query)(const OakSharedMemoryRegion *,
|
||||
char *, int),
|
||||
const OakSharedMemoryRegion *handle)
|
||||
{
|
||||
const int size = query(handle, nullptr, 0);
|
||||
if (size <= 0) {
|
||||
return QString();
|
||||
}
|
||||
QByteArray buf(size + 1, '\0');
|
||||
query(handle, buf.data(), size + 1);
|
||||
return QString::fromUtf8(buf.constData());
|
||||
}
|
||||
|
||||
#if defined(Q_OS_WIN)
|
||||
void *handle_; // HANDLE from CreateFileMapping/OpenFileMapping
|
||||
#else
|
||||
int fd_; // file descriptor from shm_open
|
||||
QString shm_name_; // the platform-prefixed name actually passed to shm_open
|
||||
#endif
|
||||
void refresh_caches()
|
||||
{
|
||||
key_ = query_string(oakengine_ipc_shm_key, handle_);
|
||||
error_ = query_string(oakengine_ipc_shm_error, handle_);
|
||||
}
|
||||
|
||||
OakSharedMemoryRegion *handle_;
|
||||
QString key_;
|
||||
QString error_;
|
||||
};
|
||||
|
||||
} // namespace ipc
|
||||
} // namespace olive
|
||||
|
||||
#endif // OAK_IPC_SHAREDMEMORYREGION_H
|
||||
#endif // OAK_IPC_SHAREDMEMORYREGION_H
|
||||
|
||||
@@ -1,185 +0,0 @@
|
||||
/***
|
||||
|
||||
Oak - Non-Linear Video Editor
|
||||
Copyright (C) 2026 Oak Team
|
||||
|
||||
This program is free software: you can redistribute it and/or modify
|
||||
it under the terms of the GNU General Public License as published by
|
||||
the Free Software Foundation, either version 3 of the License, or
|
||||
(at your option) any later version.
|
||||
|
||||
This program is distributed in the hope that it will be useful,
|
||||
but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
GNU General Public License for more details.
|
||||
|
||||
You should have received a copy of the GNU General Public License
|
||||
along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
***/
|
||||
|
||||
#ifndef OAK_IPC_SPSCRINGBUFFER_H
|
||||
#define OAK_IPC_SPSCRINGBUFFER_H
|
||||
|
||||
#include <atomic>
|
||||
#include <cstddef>
|
||||
#include <cstdint>
|
||||
|
||||
namespace olive
|
||||
{
|
||||
namespace ipc
|
||||
{
|
||||
|
||||
/**
|
||||
* @brief A lock-free single-producer / single-consumer ring buffer of uint32 indices.
|
||||
*
|
||||
* This is the core synchronization primitive for cross-process communication. It is designed to
|
||||
* live in a shared memory segment: the control block (head/tail cursors) and the slot array are a
|
||||
* single trivially-copyable, self-contained POD. Both the producer process and the consumer
|
||||
* process map the same memory and operate on it concurrently.
|
||||
*
|
||||
* Correctness relies on the classic SPSC invariant:
|
||||
* - Exactly ONE thread/process calls Push() (the producer).
|
||||
* - Exactly ONE thread/process calls Pop() (the consumer).
|
||||
* Under that invariant, no mutex is required. The producer only ever writes `head_`, the consumer
|
||||
* only ever writes `tail_`, and the acquire/release ordering on those two atomics publishes the
|
||||
* slot writes safely across the process boundary.
|
||||
*
|
||||
* Capacity note: one slot is always left empty to disambiguate the full and empty states, so a
|
||||
* buffer constructed with kCapacity slots can hold at most (kCapacity - 1) live entries.
|
||||
*
|
||||
* The payload stored is a single uint32_t per entry, intended to be an index into a separately
|
||||
* managed slot pool (see FrameSlotPool). We never put pointers in shared memory.
|
||||
*/
|
||||
class SpscRingBuffer {
|
||||
public:
|
||||
/**
|
||||
* @brief In-place construct a ring buffer header at `mem` with `capacity` index slots.
|
||||
*
|
||||
* `mem` must point to at least BytesNeeded(capacity) bytes of zero-initializable memory. This is
|
||||
* intended to be placement-style initialization performed exactly once by whichever process owns
|
||||
* the segment's creation; the peer process uses Attach() instead.
|
||||
*/
|
||||
static SpscRingBuffer *create(void *mem, uint32_t capacity)
|
||||
{
|
||||
auto *self = reinterpret_cast<SpscRingBuffer *>(mem);
|
||||
self->capacity_ = capacity;
|
||||
self->head_.store(0, std::memory_order_relaxed);
|
||||
self->tail_.store(0, std::memory_order_relaxed);
|
||||
for (uint32_t i = 0; i < capacity; i++) {
|
||||
self->slot_array()[i] = 0;
|
||||
}
|
||||
return self;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Re-interpret already-initialized memory as a ring buffer (peer process side).
|
||||
*
|
||||
* No writes are performed; the cursors and capacity are assumed already set by Create().
|
||||
*/
|
||||
static SpscRingBuffer *attach(void *mem)
|
||||
{
|
||||
return reinterpret_cast<SpscRingBuffer *>(mem);
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Total bytes required to hold the header plus `capacity` index slots.
|
||||
*/
|
||||
static size_t bytes_needed(uint32_t capacity)
|
||||
{
|
||||
return sizeof(SpscRingBuffer) + size_t(capacity) * sizeof(uint32_t);
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Producer side: enqueue an index. Returns false if the buffer is full.
|
||||
*/
|
||||
bool push(uint32_t value)
|
||||
{
|
||||
const uint32_t head = head_.load(std::memory_order_relaxed);
|
||||
const uint32_t next = increment(head);
|
||||
|
||||
// Buffer is full if advancing head would collide with the consumer's tail.
|
||||
if (next == tail_.load(std::memory_order_acquire)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
slot_array()[head] = value;
|
||||
head_.store(next, std::memory_order_release);
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Consumer side: dequeue an index into `out`. Returns false if the buffer is empty.
|
||||
*/
|
||||
bool pop(uint32_t *out)
|
||||
{
|
||||
const uint32_t tail = tail_.load(std::memory_order_relaxed);
|
||||
|
||||
// Buffer is empty if the consumer has caught up to the producer.
|
||||
if (tail == head_.load(std::memory_order_acquire)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
*out = slot_array()[tail];
|
||||
tail_.store(increment(tail), std::memory_order_release);
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Approximate number of entries currently queued.
|
||||
*
|
||||
* Safe to call from either side, but the value may be stale the instant it returns. Intended for
|
||||
* metrics/backpressure heuristics, not for correctness decisions.
|
||||
*/
|
||||
uint32_t size_approx() const
|
||||
{
|
||||
const uint32_t head = head_.load(std::memory_order_acquire);
|
||||
const uint32_t tail = tail_.load(std::memory_order_acquire);
|
||||
return (head + capacity_ - tail) % capacity_;
|
||||
}
|
||||
|
||||
bool is_empty_approx() const
|
||||
{
|
||||
return head_.load(std::memory_order_acquire) ==
|
||||
tail_.load(std::memory_order_acquire);
|
||||
}
|
||||
|
||||
uint32_t capacity() const
|
||||
{
|
||||
return capacity_;
|
||||
}
|
||||
|
||||
private:
|
||||
uint32_t increment(uint32_t index) const
|
||||
{
|
||||
// capacity_ is small and this avoids requiring a power-of-two capacity.
|
||||
return (index + 1) % capacity_;
|
||||
}
|
||||
|
||||
// The index slot array is allocated immediately after this struct in the same contiguous block.
|
||||
// (Named slot_array() rather than slots() to avoid Qt's `slots` keyword macro.)
|
||||
uint32_t *slot_array()
|
||||
{
|
||||
return reinterpret_cast<uint32_t *>(this + 1);
|
||||
}
|
||||
const uint32_t *slot_array() const
|
||||
{
|
||||
return reinterpret_cast<const uint32_t *>(this + 1);
|
||||
}
|
||||
|
||||
// Producer writes head_, consumer writes tail_. Kept on separate cache lines would be ideal, but
|
||||
// since these live in shared memory with a trailing flexible array we keep the header compact and
|
||||
// rely on acquire/release ordering for correctness.
|
||||
std::atomic<uint32_t> head_;
|
||||
std::atomic<uint32_t> tail_;
|
||||
uint32_t capacity_;
|
||||
|
||||
static_assert(
|
||||
sizeof(std::atomic<uint32_t>) == sizeof(uint32_t),
|
||||
"atomic<uint32_t> must be lock-free POD-sized for shared memory use");
|
||||
};
|
||||
|
||||
} // namespace ipc
|
||||
} // namespace olive
|
||||
|
||||
#endif // OAK_IPC_SPSCRINGBUFFER_H
|
||||
Reference in New Issue
Block a user