queue.h

Go to the documentation of this file.

 
 
#pragma once
// This header file is auto-generated!
// Do not make changes to this file or they will be overwritten.
// clang-format off
#include <cx/obj.h>
#include <cx/struct.h>
#include <cx/net/net_shared.h>
#include <cx/thread.h>
#include <cx/net/pool.h>
 
CX_C_BEGIN
 
typedef struct NetQueue NetQueue;
typedef struct NetQueue_WeakRef NetQueue_WeakRef;
saDeclarePtr(NetQueue);
saDeclarePtr(NetQueue_WeakRef);
#define _sti_NetQueue _sti_object
#define SType_NetQueue NetQueue*
#define STStorageType_NetQueue NetQueue*
#define STypeArg_NetQueue(type, val) stgeneric(object, (ObjInst*)objInstCheckClass(NetQueue, val))
#define STypeArgPtr_NetQueue(type, val) (stgeneric*)objInstCheckClassPtr(NetQueue, val)
#define STypeCheckedArg_NetQueue(type, val)    stType(type), stArg(type, val)
#define STypeCheckedPtrArg_NetQueue(type, val) stType(type), stArgPtr(type, val)
 
 
 
typedef struct NetQueue_ClassIf {
    ObjIface* _implements;
    ObjIface* _parent;
    size_t _size;
 
    bool (*addSocket)(_In_ void* self, NetSocket* socket);
    bool (*removeSocket)(_In_ void* self, NetSocket* socket);
    NetSocket* (*socket)(_In_ void* self, NetSocketType type);
    // Begin one outbound connect attempt against a single resolved address
    //     // The one backend-specific piece of connect. By the time this is called, the portable state
    // machine has already picked the address and armed the deadline; this just gets a fresh OS
    // handle of the address's family and starts the attempt. A readiness backend resets the
    // handle, issues a non-blocking connect(), and watches it for writability (and the except
    // set, where Windows signals a failed connect). A completion backend resets and binds the
    // handle and posts an overlapped ConnectEx.
    //     // The outcome is reported back through netsocket_connectResult() -- immediately for a
    // synchronous result, or later from readiness, a completion, or the timeout sweep.
    //     // @param sock Socket to connect (a stream socket in NS_Connecting)
    // @param addr Address to attempt
    // @return true if the attempt was initiated or resolved, false if it could not be started
    bool (*connectBegin)(_In_ void* self, NetSocket* sock, const NetAddr* addr);
    // React to a socket having just become connected
    //     // Called once, after a successful connect has transitioned the socket to NS_Connected and
    // its NET_Connection event has been queued. The backend starts servicing the socket for
    // receive here -- select rebuilds its watch set to read-watch it, IOCP posts the first
    // WSARecv -- the same thing addSocket() does for a socket that was already connected when it
    // joined the queue.
    //     // @param sock Socket that just connected
    void (*connectArm)(_In_ void* self, NetSocket* sock);
    // Begin (or resume) accepting connections on a now-listening socket
    //     // The accept-side counterpart of connectArm. Called once the socket enters NS_Listening --
    // from the platform listen(), or from addSocket() if the socket was already listening when
    // it joined the queue. A readiness backend wakes its loop so the next rebuild read-watches
    // the listener; a completion backend posts its initial batch of overlapped AcceptEx
    // operations. Each accepted connection is wrapped and handed to netsocket_accepted().
    //     // @param sock Listening socket (a stream socket in NS_Listening)
    void (*acceptArm)(_In_ void* self, NetSocket* sock);
    bool (*shutdown)(_In_ void* self, int64 timeout);
    bool (*tick)(_In_ void* self, int64 wait);
} NetQueue_ClassIf;
extern NetQueue_ClassIf NetQueue_ClassIf_tmpl;
 
typedef struct NetQueue {
    union {
        NetQueue_ClassIf* _;
        void* _is_NetQueue;
        void* _is_ObjInst;
    };
    ObjClassInfo* _clsinfo;
    atomic(uintptr) _ref;
    atomic(ptr) _weakref;
 
    NetQueueConfig conf;        
    NetHandlers handlers;
    void* handlerCtx;        
    Weak(ObjInst)* handlerWeak;
    RWLock handlerLock;        
    hashtable sockets;        
    RWLock lock;        
    PrQueue runq;
    NetPool* pool;
    atomic(uint32) gcLastLo;
    atomic(uint32) droppedNoBuf;
    atomic(uint32) nflows;        
    NetTimerEntry* timers;
    uint32 ntimers;        
    uint32 timersCap;        
    uint64 timerSerial;        
    hashtable timerIdx;        
    Mutex timerLock;
    atomic(uint32) timersLive;        
    sa_Thread workers;
    Semaphore runqSema;        
    atomic(uint32) shutdownReq;        
    NetSendPumpFn sendPump;
    void* sendCtx;        
    NetWakeFn wake;
    void* wakeCtx;        
} NetQueue;
extern ObjClassInfo NetQueue_clsinfo;
#define NetQueue(inst) objInstCheckClass(NetQueue, inst)
#define NetQueueNone ((NetQueue*)NULL)
 
typedef struct NetQueue_WeakRef {
    union {
        ObjInst* _inst;
        void* _is_NetQueue_WeakRef;
        void* _is_ObjInst_WeakRef;
    };
    atomic(uintptr) _ref;
    RWLock _lock;
} NetQueue_WeakRef;
#define NetQueue_WeakRef(inst) objWeakRefCheckClass(NetQueue, inst)
 
void NetQueue_presetClient(_Out_ NetQueueConfig* conf);
#define netqueuePresetClient(conf) NetQueue_presetClient(conf)
 
void NetQueue_presetServer(_Out_ NetQueueConfig* conf);
#define netqueuePresetServer(conf) NetQueue_presetServer(conf)
 
_Ret_maybenull_ NetSocket* NetQueue_connect(_In_ NetQueue* self, _In_opt_ strref host, uint16 port, _In_opt_ const NetHandlers* handlers, _In_opt_ void* ctx);
#define netqueueConnect(self, host, port, handlers, ctx) NetQueue_connect(NetQueue(self), host, port, handlers, ctx)
 
_Ret_maybenull_ NetSocket* NetQueue_connectPrep(_In_ NetQueue* self, _In_opt_ strref host, uint16 port, _In_opt_ const NetHandlers* handlers, _In_opt_ void* ctx, NetConnectPrepCB prep, _In_opt_ void* prepctx);
#define netqueueConnectPrep(self, host, port, handlers, ctx, prep, prepctx) NetQueue_connectPrep(NetQueue(self), host, port, handlers, ctx, prep, prepctx)
 
_Ret_maybenull_ NetSocket* NetQueue_listen(_In_ NetQueue* self, _In_ const NetAddr* addr, int backlog, _In_opt_ const NetHandlers* handlers, _In_opt_ void* ctx);
#define netqueueListen(self, addr, backlog, handlers, ctx) NetQueue_listen(NetQueue(self), addr, backlog, handlers, ctx)
 
void NetQueue_setHandlers(_In_ NetQueue* self, _In_opt_ const NetHandlers* handlers, _In_opt_ void* ctx);
#define netqueueSetHandlers(self, handlers, ctx) NetQueue_setHandlers(NetQueue(self), handlers, ctx)
 
void NetQueue_setHandlersObj(_In_ NetQueue* self, _In_opt_ const NetHandlers* handlers, _In_opt_ ObjInst* ctx);
#define netqueueSetHandlersObj(self, handlers, ctx) NetQueue_setHandlersObj(NetQueue(self), handlers, ObjInst(ctx))
 
_Ret_maybenull_ NetFlow* NetQueue_promoteFlow(_In_ NetQueue* self, _Inout_ NetSocket* sock, _In_ const NetAddr* peer);
#define netqueuePromoteFlow(self, sock, peer) NetQueue_promoteFlow(NetQueue(self), sock, peer)
 
uint32 NetQueue_droppedNoBuf(_In_ NetQueue* self);
#define netqueueDroppedNoBuf(self) NetQueue_droppedNoBuf(NetQueue(self))
 
void NetQueue__applyConfig(_In_ NetQueue* self, _In_opt_ const NetQueueConfig* conf);
// void netqueue_applyConfig(NetQueue* self, const NetQueueConfig* conf);
//
// -----------------------------------------------------------------------------------------
// PRIVATE IMPLEMENTATION DETAILS
//     // Everything below this point is internal plumbing between the net module's own translation
// units and the platform backends. It is not a stable API and carries no compatibility
// promise -- signatures and semantics change whenever the implementation needs them to.
// -----------------------------------------------------------------------------------------
// Construction (queue.c)
// Apply a configuration to a queue under construction. Called by a backend's factory before
// objInstInit(), so that the sizes everything else is built from are already in place when the
// base class initializes. `conf` is NULL for the client preset.
#define netqueue_applyConfig(self, conf) NetQueue__applyConfig(NetQueue(self), conf)
 
void NetQueue__startWorkers(_In_ NetQueue* self, int32 n);
// void netqueue_startWorkers(NetQueue* self, int32 n);
//
// Worker pool (queue.c)
//     // The generic dispatch pool the base class owns: N threads that drain the runqueue and block
// on runqSema when it is empty. A backend's factory starts them in threaded mode; polled
// queues never call these and run everything on the caller's tick() thread.
// Start n dispatch worker threads. Fewer may end up running if thread creation fails; the
// actual count lands in self->workers.
#define netqueue_startWorkers(self, n) NetQueue__startWorkers(NetQueue(self), n)
 
void NetQueue__stopWorkers(_In_ NetQueue* self, int64 timeout);
// void netqueue_stopWorkers(NetQueue* self, int64 timeout);
//
// Ask every dispatch worker to exit and join it. Each worker drains the runqueue one last time
// on its way out, so terminal events queued during shutdown are still delivered. `timeout`
// <= 0 waits indefinitely. Idempotent -- safe to call when no workers are running.
#define netqueue_stopWorkers(self, timeout) NetQueue__stopWorkers(NetQueue(self), timeout)
 
bool NetQueue__ingestDatagram(_In_ NetQueue* self, _Inout_ NetSocket* sock, _In_ NetAddr* peer, _In_opt_ const NetPktInfo* info, _Inout_ Buffer* buf);
// bool netqueue_ingestDatagram(NetQueue* self, NetSocket* sock, NetAddr* peer, const NetPktInfo* info, Buffer* buf);
//
// Ingest (queue.c)
//     // Where a packet enters the core. _ingestDatagram is the datagram front door and belongs to the
// queue; _submit is the runqueue's producer half and lives beside its consumer in dispatch.c.
// Hand a received datagram to the core for demultiplexing and dispatch. Takes ownership of
// `buf`, which must have come from the queue's own buffer pool. Resolves the flow for the
// source address (creating one, reclaiming under cap pressure, or falling back to the
// flowRefused handler), pushes the packet onto that flow's inbox, and enqueues the flow if it
// was idle. Returns true if the packet was queued to a flow, false if it was dropped or handed
// to a NET_FlowRefused handler instead.
#define netqueue_ingestDatagram(self, sock, peer, info, buf) NetQueue__ingestDatagram(NetQueue(self), sock, peer, info, buf)
 
void NetQueue__submit(_In_ NetQueue* self, _Inout_ NetFlow* flow, _Inout_ NetMessage* msg);
// void netqueue_submit(NetQueue* self, NetFlow* flow, NetMessage* msg);
//
// Hand a message to a flow and make sure something will come along to run it. Takes ownership
// of the message. `self` may be NULL, in which case the message is simply released.
#define netqueue_submit(self, flow, msg) NetQueue__submit(NetQueue(self), flow, msg)
 
bool NetQueue__dispatch(_In_ NetQueue* self);
// bool netqueue_dispatch(NetQueue* self);
//
// Dispatch (dispatch.c)
//     // The runqueue and the event delivery it feeds. _submit above is the producer half of the same
// claim protocol _dispatch consumes.
// Claim one flow from the runqueue and drain it. This is the body of a worker thread, and it
// is also what tick() runs inline in polled mode -- the same code either way, just without the
// threads. Returns false if the runqueue was empty.
#define netqueue_dispatch(self) NetQueue__dispatch(NetQueue(self))
 
void NetQueue__deliver(_In_ NetQueue* self, _In_opt_ NetSocket* sock, _In_opt_ NetFlow* flow, _Inout_ NetEvent* ev);
// void netqueue_deliver(NetQueue* self, NetSocket* sock, NetFlow* flow, NetEvent* ev);
//
// Resolve and invoke the handler for an event, filling in the event's queue/socket/flow/ctx
// fields; a no-op when no level supplies a handler. In dev builds the callback is timed, and
// one that holds its worker past the warning threshold gets logged with the socket and event
// type -- handlers are expected to return quickly, and this is how we notice when one doesn't.
// Every callback invocation should go through here so timing stays consistent. `self` may be
// NULL, in which case only the socket and flow levels are consulted.
#define netqueue_deliver(self, sock, flow, ev) NetQueue__deliver(NetQueue(self), sock, flow, ev)
 
void NetQueue__maint(_In_ NetQueue* self);
// void netqueue_maint(NetQueue* self);
//
// Maintenance (queue.c)
// Opportunistic GC of the queue's dynamic PrQueues (runqueue and receive pool). Call it from a
// thread that has just drained the runqueue and is about to idle -- the moment the PrQueue
// docs recommend for collection. Time-gated and single-runner, so a whole worker pool or a
// tight poll loop costs one GC attempt per interval; safe to call as often as convenient.
#define netqueue_maint(self) NetQueue__maint(NetQueue(self))
 
NetTimerId NetQueue__addTimer(_In_ NetQueue* self, _Inout_ NetFlow* flow, int64 delay, flags_t flags, NetTimerFn fn, _In_opt_ void* ctx);
// NetTimerId netqueue_addTimer(NetQueue* self, NetFlow* flow, int64 delay, flags_t flags, NetTimerFn fn, void* ctx);
//
// Timers (timer.c)
//     // The queue-wide deadline heap behind netflowAddTimer(). Backends drive it with two calls per
// wait: _nextDeadline() to bound the sleep, and _timerSweep() to fire whatever came due.
// Arm a timer on a flow, `delay` microseconds from now. `fn` is NULL for an application timer
// (delivered as NET_Timer through the flow's inbox) or a framework hook fired inline on the
// sweeping thread. Returns the new id, or 0 if the flow is already dying or has no queue.
#define netqueue_addTimer(self, flow, delay, flags, fn, ctx) NetQueue__addTimer(NetQueue(self), flow, delay, flags, fn, ctx)
 
bool NetQueue__cancelTimer(_In_ NetQueue* self, NetTimerId id);
// bool netqueue_cancelTimer(NetQueue* self, NetTimerId id);
//
// Cancel an armed timer. Returns true only if this call removed it from the heap -- a timer
// already popped for delivery answers false. A successful cancel means the caller is the one
// that gets to act on it. Cancelling an unknown or already-fired id is a harmless false.
#define netqueue_cancelTimer(self, id) NetQueue__cancelTimer(NetQueue(self), id)
 
bool NetQueue__rearmTimer(_In_ NetQueue* self, NetTimerId id, int64 delay);
// bool netqueue_rearmTimer(NetQueue* self, NetTimerId id, int64 delay);
//
// Move an armed timer's deadline to `delay` microseconds from now. Returns false if the timer
// is not armed (cancelled, or already popped for delivery).
#define netqueue_rearmTimer(self, id, delay) NetQueue__rearmTimer(NetQueue(self), id, delay)
 
void NetQueue__cancelFlowTimers(_In_ NetQueue* self, _Inout_ NetFlow* flow);
// void netqueue_cancelFlowTimers(NetQueue* self, NetFlow* flow);
//
// Cancel every timer armed on a flow. Called on the flow's terminal path so a long deadline on
// a dead connection cannot keep the flow object alive until it expires.
#define netqueue_cancelFlowTimers(self, flow) NetQueue__cancelFlowTimers(NetQueue(self), flow)
 
void NetQueue__timerSweep(_In_ NetQueue* self);
// void netqueue_timerSweep(NetQueue* self);
//
// Fire every timer whose deadline has passed. A no-op on one relaxed load while nothing is
// armed. Entries are popped under timerLock and fired outside it, so a callback is free to arm,
// cancel, or send. Called by every backend right after its wait returns, beside _maint().
#define netqueue_timerSweep(self) NetQueue__timerSweep(NetQueue(self))
 
int64 NetQueue__nextDeadline(_In_ NetQueue* self);
// int64 netqueue_nextDeadline(NetQueue* self);
//
// The nearest armed deadline in absolute clockTimer() microseconds, or 0 if nothing is armed.
// Backends cap their wait to this so a timer fires close to when it is due rather than on the
// next unrelated wakeup.
#define netqueue_nextDeadline(self) NetQueue__nextDeadline(NetQueue(self))
 
_Ret_maybenull_ NetFlow* NetQueue__findFlow(_In_ NetQueue* self, _Inout_ NetSocket* sock, _In_ NetAddr* peer, bool create);
// NetFlow* netqueue_findFlow(NetQueue* self, NetSocket* sock, NetAddr* peer, bool create);
//
// Flow table and lifecycle (flow.c)
// Find the flow for a peer address, creating one if there is room under the cap. Returns a
// strong reference the caller must release, or NULL if the queue is at its cap and nothing
// could be reclaimed -- in which case the caller should fall back to the flowRefused handler.
// A stream or QUIC socket has no address-keyed table, so `peer` is ignored there and the
// socket's single control flow is returned.
#define netqueue_findFlow(self, sock, peer, create) NetQueue__findFlow(NetQueue(self), sock, peer, create)
 
_Ret_maybenull_ NetFlow* NetQueue__admitFlow(_In_ NetQueue* self, _Inout_ NetSocket* sock, _In_ NetAddr* peer);
// NetFlow* netqueue_admitFlow(NetQueue* self, NetSocket* sock, NetAddr* peer);
//
// Create a flow for a peer and register it with the socket, bypassing the cap check. Returns a
// strong reference the caller must release. A thin wrapper over _admitFlowObj below.
#define netqueue_admitFlow(self, sock, peer) NetQueue__admitFlow(NetQueue(self), sock, peer)
 
_Ret_maybenull_ NetFlow* NetQueue__admitFlowObj(_In_ NetQueue* self, _Inout_ NetSocket* sock, _Inout_ NetFlow* flow);
// NetFlow* netqueue_admitFlowObj(NetQueue* self, NetSocket* sock, NetFlow* flow);
//
// Register an already-constructed flow with its socket: build its filter chain, publish it in
// the socket's table (keyed on `peer` for a datagram socket, `key` for a QUIC one), count it
// against the queue, queue its NET_FlowOpen ahead of anything else that could land in its
// inbox, and prime the chain. Split out of _admitFlow so cxquic can register a NetFlow
// subclass of its own for a QUIC stream.
//     // Consumes the caller's reference and returns the one to use, which is normally the same flow
// -- but is the winner's flow when another thread inserted the same key first, and NULL if the
// table lost it. Do not use `flow` after the call; use the return value.
#define netqueue_admitFlowObj(self, sock, flow) NetQueue__admitFlowObj(NetQueue(self), sock, flow)
 
uint32 NetQueue__reclaimFlows(_In_ NetQueue* self, _Inout_ NetSocket* sock);
// uint32 netqueue_reclaimFlows(NetQueue* self, NetSocket* sock);
//
// Mark up to reclaimBatch least-recently-active flows on this socket for teardown, skipping
// flows active more recently than reclaimMinIdle allows. Runs inline on the ingest path at the
// moment of a cap hit; there is no timer thread.
#define netqueue_reclaimFlows(self, sock) NetQueue__reclaimFlows(NetQueue(self), sock)
 
#define netqueueAddSocket(self, socket) (self)->_->addSocket(NetQueue(self), socket)
#define netqueueRemoveSocket(self, socket) (self)->_->removeSocket(NetQueue(self), socket)
#define netqueueSocket(self, type) (self)->_->socket(NetQueue(self), type)
// bool netqueueConnectBegin(NetQueue* self, NetSocket* sock, const NetAddr* addr);
//
// Begin one outbound connect attempt against a single resolved address
//     // The one backend-specific piece of connect. By the time this is called, the portable state
// machine has already picked the address and armed the deadline; this just gets a fresh OS
// handle of the address's family and starts the attempt. A readiness backend resets the
// handle, issues a non-blocking connect(), and watches it for writability (and the except
// set, where Windows signals a failed connect). A completion backend resets and binds the
// handle and posts an overlapped ConnectEx.
//     // The outcome is reported back through netsocket_connectResult() -- immediately for a
// synchronous result, or later from readiness, a completion, or the timeout sweep.
//     // @param sock Socket to connect (a stream socket in NS_Connecting)
// @param addr Address to attempt
// @return true if the attempt was initiated or resolved, false if it could not be started
#define netqueueConnectBegin(self, sock, addr) (self)->_->connectBegin(NetQueue(self), sock, addr)
// void netqueueConnectArm(NetQueue* self, NetSocket* sock);
//
// React to a socket having just become connected
//     // Called once, after a successful connect has transitioned the socket to NS_Connected and
// its NET_Connection event has been queued. The backend starts servicing the socket for
// receive here -- select rebuilds its watch set to read-watch it, IOCP posts the first
// WSARecv -- the same thing addSocket() does for a socket that was already connected when it
// joined the queue.
//     // @param sock Socket that just connected
#define netqueueConnectArm(self, sock) (self)->_->connectArm(NetQueue(self), sock)
// void netqueueAcceptArm(NetQueue* self, NetSocket* sock);
//
// Begin (or resume) accepting connections on a now-listening socket
//     // The accept-side counterpart of connectArm. Called once the socket enters NS_Listening --
// from the platform listen(), or from addSocket() if the socket was already listening when
// it joined the queue. A readiness backend wakes its loop so the next rebuild read-watches
// the listener; a completion backend posts its initial batch of overlapped AcceptEx
// operations. Each accepted connection is wrapped and handed to netsocket_accepted().
//     // @param sock Listening socket (a stream socket in NS_Listening)
#define netqueueAcceptArm(self, sock) (self)->_->acceptArm(NetQueue(self), sock)
#define netqueueShutdown(self, timeout) (self)->_->shutdown(NetQueue(self), timeout)
#define netqueueTick(self, wait) (self)->_->tick(NetQueue(self), wait)
 
CX_C_END