queue_select.h

#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/queue.h>
 
CX_C_BEGIN
 
typedef struct NetQueueSelect NetQueueSelect;
typedef struct NetQueueSelect_WeakRef NetQueueSelect_WeakRef;
saDeclarePtr(NetQueueSelect);
saDeclarePtr(NetQueueSelect_WeakRef);
#define _sti_NetQueueSelect _sti_object
#define SType_NetQueueSelect NetQueueSelect*
#define STStorageType_NetQueueSelect NetQueueSelect*
#define STypeArg_NetQueueSelect(type, val) stgeneric(object, (ObjInst*)objInstCheckClass(NetQueueSelect, val))
#define STypeArgPtr_NetQueueSelect(type, val) (stgeneric*)objInstCheckClassPtr(NetQueueSelect, val)
#define STypeCheckedArg_NetQueueSelect(type, val)    stType(type), stArg(type, val)
#define STypeCheckedPtrArg_NetQueueSelect(type, val) stType(type), stArgPtr(type, val)
 
typedef struct NetQueueSelect_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);
} NetQueueSelect_ClassIf;
extern NetQueueSelect_ClassIf NetQueueSelect_ClassIf_tmpl;
 
typedef struct NetQueueSelect {
    union {
        NetQueueSelect_ClassIf* _;
        void* _is_NetQueueSelect;
        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;        
    // NetSelectSet* (opaque), created and destroyed by hand
    //     // Held as void* so the private NetSelectSet declaration does not have to leak into this
    // generated header. Cast back in the implementation.
    void* selset;
    // Handle -> socket for the current tick, each entry holding a strong reference
    //     // Rebuilt every tick under the socket lock. Holding a reference means a socket removed by
    // another thread mid-tick stays alive until the map is cleared at the top of the next tick,
    // so ingest never touches freed memory. The object value type manages the references.
    hashtable fdmap;
    // Dedicated select-loop thread in threaded mode (NULL in polled mode)
    //     // select() cannot usefully be shared across threads -- one fd_set, one wait -- so threaded
    // mode runs exactly one ingest thread here that fills the runqueue, plus the base class's N
    // dispatch workers that drain it. In polled mode this stays NULL and tick() runs the loop on
    // the caller's thread. Joined (via nselWake + thrWait) before it is released, in shutdown or as
    // a destroy-time safety net.
    Thread* ingest;
} NetQueueSelect;
extern ObjClassInfo NetQueueSelect_clsinfo;
#define NetQueueSelect(inst) objInstCheckClass(NetQueueSelect, inst)
#define NetQueueSelectNone ((NetQueueSelect*)NULL)
 
typedef struct NetQueueSelect_WeakRef {
    union {
        ObjInst* _inst;
        void* _is_NetQueueSelect_WeakRef;
        void* _is_NetQueue_WeakRef;
        void* _is_ObjInst_WeakRef;
    };
    atomic(uintptr) _ref;
    RWLock _lock;
} NetQueueSelect_WeakRef;
#define NetQueueSelect_WeakRef(inst) objWeakRefCheckClass(NetQueueSelect, inst)
 
_objfactory_guaranteed NetQueueSelect* NetQueueSelect_create(NetQueueConfig* conf);
// NetQueueSelect* netqueueselectCreate(NetQueueConfig* conf);
#define netqueueselectCreate(conf) NetQueueSelect_create(conf)
 
// NetSocket* netqueueselectConnect(NetQueueSelect* self, strref host, uint16 port, const NetHandlers* handlers, void* ctx);
#define netqueueselectConnect(self, host, port, handlers, ctx) NetQueue_connect(NetQueue(self), host, port, handlers, ctx)
 
// NetSocket* netqueueselectConnectPrep(NetQueueSelect* self, strref host, uint16 port, const NetHandlers* handlers, void* ctx, NetConnectPrepCB prep, void* prepctx);
#define netqueueselectConnectPrep(self, host, port, handlers, ctx, prep, prepctx) NetQueue_connectPrep(NetQueue(self), host, port, handlers, ctx, prep, prepctx)
 
// NetSocket* netqueueselectListen(NetQueueSelect* self, const NetAddr* addr, int backlog, const NetHandlers* handlers, void* ctx);
#define netqueueselectListen(self, addr, backlog, handlers, ctx) NetQueue_listen(NetQueue(self), addr, backlog, handlers, ctx)
 
// void netqueueselectSetHandlers(NetQueueSelect* self, const NetHandlers* handlers, void* ctx);
#define netqueueselectSetHandlers(self, handlers, ctx) NetQueue_setHandlers(NetQueue(self), handlers, ctx)
 
// void netqueueselectSetHandlersObj(NetQueueSelect* self, const NetHandlers* handlers, ObjInst* ctx);
#define netqueueselectSetHandlersObj(self, handlers, ctx) NetQueue_setHandlersObj(NetQueue(self), handlers, ObjInst(ctx))
 
// NetFlow* netqueueselectPromoteFlow(NetQueueSelect* self, NetSocket* sock, const NetAddr* peer);
#define netqueueselectPromoteFlow(self, sock, peer) NetQueue_promoteFlow(NetQueue(self), sock, peer)
 
// uint32 netqueueselectDroppedNoBuf(NetQueueSelect* self);
#define netqueueselectDroppedNoBuf(self) NetQueue_droppedNoBuf(NetQueue(self))
 
// void netqueueselect_applyConfig(NetQueueSelect* 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 netqueueselect_applyConfig(self, conf) NetQueue__applyConfig(NetQueue(self), conf)
 
// void netqueueselect_startWorkers(NetQueueSelect* 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 netqueueselect_startWorkers(self, n) NetQueue__startWorkers(NetQueue(self), n)
 
// void netqueueselect_stopWorkers(NetQueueSelect* 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 netqueueselect_stopWorkers(self, timeout) NetQueue__stopWorkers(NetQueue(self), timeout)
 
// bool netqueueselect_ingestDatagram(NetQueueSelect* 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 netqueueselect_ingestDatagram(self, sock, peer, info, buf) NetQueue__ingestDatagram(NetQueue(self), sock, peer, info, buf)
 
// void netqueueselect_submit(NetQueueSelect* 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 netqueueselect_submit(self, flow, msg) NetQueue__submit(NetQueue(self), flow, msg)
 
// bool netqueueselect_dispatch(NetQueueSelect* 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 netqueueselect_dispatch(self) NetQueue__dispatch(NetQueue(self))
 
// void netqueueselect_deliver(NetQueueSelect* 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 netqueueselect_deliver(self, sock, flow, ev) NetQueue__deliver(NetQueue(self), sock, flow, ev)
 
// void netqueueselect_maint(NetQueueSelect* 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 netqueueselect_maint(self) NetQueue__maint(NetQueue(self))
 
// NetTimerId netqueueselect_addTimer(NetQueueSelect* 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 netqueueselect_addTimer(self, flow, delay, flags, fn, ctx) NetQueue__addTimer(NetQueue(self), flow, delay, flags, fn, ctx)
 
// bool netqueueselect_cancelTimer(NetQueueSelect* 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 netqueueselect_cancelTimer(self, id) NetQueue__cancelTimer(NetQueue(self), id)
 
// bool netqueueselect_rearmTimer(NetQueueSelect* 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 netqueueselect_rearmTimer(self, id, delay) NetQueue__rearmTimer(NetQueue(self), id, delay)
 
// void netqueueselect_cancelFlowTimers(NetQueueSelect* 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 netqueueselect_cancelFlowTimers(self, flow) NetQueue__cancelFlowTimers(NetQueue(self), flow)
 
// void netqueueselect_timerSweep(NetQueueSelect* 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 netqueueselect_timerSweep(self) NetQueue__timerSweep(NetQueue(self))
 
// int64 netqueueselect_nextDeadline(NetQueueSelect* 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 netqueueselect_nextDeadline(self) NetQueue__nextDeadline(NetQueue(self))
 
// NetFlow* netqueueselect_findFlow(NetQueueSelect* 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 netqueueselect_findFlow(self, sock, peer, create) NetQueue__findFlow(NetQueue(self), sock, peer, create)
 
// NetFlow* netqueueselect_admitFlow(NetQueueSelect* 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 netqueueselect_admitFlow(self, sock, peer) NetQueue__admitFlow(NetQueue(self), sock, peer)
 
// NetFlow* netqueueselect_admitFlowObj(NetQueueSelect* 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 netqueueselect_admitFlowObj(self, sock, flow) NetQueue__admitFlowObj(NetQueue(self), sock, flow)
 
// uint32 netqueueselect_reclaimFlows(NetQueueSelect* 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 netqueueselect_reclaimFlows(self, sock) NetQueue__reclaimFlows(NetQueue(self), sock)
 
// bool netqueueselectAddSocket(NetQueueSelect* self, NetSocket* socket);
#define netqueueselectAddSocket(self, socket) (self)->_->addSocket(NetQueueSelect(self), socket)
// bool netqueueselectRemoveSocket(NetQueueSelect* self, NetSocket* socket);
#define netqueueselectRemoveSocket(self, socket) (self)->_->removeSocket(NetQueueSelect(self), socket)
// NetSocket* netqueueselectSocket(NetQueueSelect* self, NetSocketType type);
#define netqueueselectSocket(self, type) (self)->_->socket(NetQueueSelect(self), type)
// bool netqueueselectConnectBegin(NetQueueSelect* 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 netqueueselectConnectBegin(self, sock, addr) (self)->_->connectBegin(NetQueueSelect(self), sock, addr)
// void netqueueselectConnectArm(NetQueueSelect* 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 netqueueselectConnectArm(self, sock) (self)->_->connectArm(NetQueueSelect(self), sock)
// void netqueueselectAcceptArm(NetQueueSelect* 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 netqueueselectAcceptArm(self, sock) (self)->_->acceptArm(NetQueueSelect(self), sock)
// bool netqueueselectShutdown(NetQueueSelect* self, int64 timeout);
#define netqueueselectShutdown(self, timeout) (self)->_->shutdown(NetQueueSelect(self), timeout)
// bool netqueueselectTick(NetQueueSelect* self, int64 wait);
#define netqueueselectTick(self, wait) (self)->_->tick(NetQueueSelect(self), wait)
 
CX_C_END