flow.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/socket.h>
#include <cx/net/filter.h>
#include <cx/net/pool.h>
#include <cx/net/net_shared.h>
 
CX_C_BEGIN
 
typedef struct NetFlow NetFlow;
typedef struct NetFlow_WeakRef NetFlow_WeakRef;
typedef struct NetFlow NetFlow;
typedef struct NetFlow_WeakRef NetFlow_WeakRef;
saDeclarePtr(NetFlow);
saDeclarePtr(NetFlow_WeakRef);
#define _sti_NetFlow _sti_object
#define SType_NetFlow NetFlow*
#define STStorageType_NetFlow NetFlow*
#define STypeArg_NetFlow(type, val) stgeneric(object, (ObjInst*)objInstCheckClass(NetFlow, val))
#define STypeArgPtr_NetFlow(type, val) (stgeneric*)objInstCheckClassPtr(NetFlow, val)
#define STypeCheckedArg_NetFlow(type, val)    stType(type), stArg(type, val)
#define STypeCheckedPtrArg_NetFlow(type, val) stType(type), stArgPtr(type, val)
 
 
 
typedef struct NetFlow_ClassIf {
    ObjIface* _implements;
    ObjIface* _parent;
    size_t _size;
 
    // needed to init lastActive
    bool (*close)(_In_ void* self);
    bool (*send)(_In_ void* self, _In_ const uint8* data, size_t len, flags_t flags);
} NetFlow_ClassIf;
extern NetFlow_ClassIf NetFlow_ClassIf_tmpl;
 
typedef struct NetFlow {
    union {
        NetFlow_ClassIf* _;
        void* _is_NetFlow;
        void* _is_ObjInst;
    };
    ObjClassInfo* _clsinfo;
    atomic(uintptr) _ref;
    atomic(ptr) _weakref;
 
    Weak(NetSocket)* socket;
    NetPool* pool;
    NetAddr peer;        
    uint64 key;
    atomic(uint32) claimed;        
    atomic(uint32) queued;        
    atomic(uint32) dying;        
    atomic(ptr) inbox;
    NetMessage* ready;        
    NetMessage* readytail;        
    atomic(uint32) lastActive;
    sa_stvar user;
    NetHandlers* handlers;        
    void* handlerCtx;        
    Weak(ObjInst)* handlerWeak;
    RWLock handlerLock;        
    sa_uint64 timers;        
    sa_NetFlowFilter filters;
    BufRing* encIn;
    NetMsgQueue encInMsgs;
    Mutex filterLock;
    uint8 closeReason;        
} NetFlow;
extern ObjClassInfo NetFlow_clsinfo;
#define NetFlow(inst) objInstCheckClass(NetFlow, inst)
#define NetFlowNone ((NetFlow*)NULL)
 
typedef struct NetFlow_WeakRef {
    union {
        ObjInst* _inst;
        void* _is_NetFlow_WeakRef;
        void* _is_ObjInst_WeakRef;
    };
    atomic(uintptr) _ref;
    RWLock _lock;
} NetFlow_WeakRef;
#define NetFlow_WeakRef(inst) objWeakRefCheckClass(NetFlow, inst)
 
_objfactory_guaranteed NetFlow* NetFlow_create(NetSocket* socket, const NetAddr* peer);
// NetFlow* netflowCreate(NetSocket* socket, const NetAddr* peer);
#define netflowCreate(socket, peer) NetFlow_create(NetSocket(socket), peer)
 
void NetFlow_setHandlers(_In_ NetFlow* self, _In_opt_ const NetHandlers* handlers, _In_opt_ void* ctx);
#define netflowSetHandlers(self, handlers, ctx) NetFlow_setHandlers(NetFlow(self), handlers, ctx)
 
void NetFlow_setHandlersObj(_In_ NetFlow* self, _In_opt_ const NetHandlers* handlers, _In_opt_ ObjInst* ctx);
#define netflowSetHandlersObj(self, handlers, ctx) NetFlow_setHandlersObj(NetFlow(self), handlers, ObjInst(ctx))
 
NetTimerId NetFlow_addTimer(_In_ NetFlow* self, int64 delay, flags_t flags);
#define netflowAddTimer(self, delay, flags) NetFlow_addTimer(NetFlow(self), delay, flags)
 
bool NetFlow_cancelTimer(_In_ NetFlow* self, NetTimerId id);
#define netflowCancelTimer(self, id) NetFlow_cancelTimer(NetFlow(self), id)
 
bool NetFlow_rearmTimer(_In_ NetFlow* self, NetTimerId id, int64 delay);
#define netflowRearmTimer(self, id, delay) NetFlow_rearmTimer(NetFlow(self), id, delay)
 
_Check_return_ bool NetFlow__push(_In_ NetFlow* self, _Inout_ NetMessage* msg);
// bool netflow_push(NetFlow* self, NetMessage* msg);
//
// -----------------------------------------------------------------------------------------
// PRIVATE IMPLEMENTATION DETAILS
//     // Everything below this point is internal plumbing between the net module's own translation
// units. It is not a stable API and carries no compatibility promise -- signatures and
// semantics change whenever the implementation needs them to. Nothing outside cx/net/ and the
// platform backends should call any of it.
// -----------------------------------------------------------------------------------------
// Inbox (flow.c)
// Push a message onto the flow's inbox. Returns true if the caller is responsible for putting
// the flow on the runqueue, which is the case only when it was not already there.
#define netflow_push(self, msg) NetFlow__push(NetFlow(self), msg)
 
_Ret_maybenull_ NetMessage* NetFlow__pop(_In_ NetFlow* self);
// NetMessage* netflow_pop(NetFlow* self);
//
// Pop the next message in FIFO order, refilling the private ready list from the inbox when it
// runs dry. May ONLY be called by the worker holding the flow's claim.
#define netflow_pop(self) NetFlow__pop(NetFlow(self))
 
void NetFlow__unpop(_In_ NetFlow* self, _Inout_ NetMessage* msg);
// void netflow_unpop(NetFlow* self, NetMessage* msg);
//
// Put a popped message back at the head of the ready list, so the next _pop() returns it
// again. For a worker that has to stop draining partway through -- the message has to keep its
// place, and everything behind it keeps its place too. Same rule as _pop(): the claim holder
// only.
#define netflow_unpop(self, msg) NetFlow__unpop(NetFlow(self), msg)
 
bool NetFlow__close(_In_ NetFlow* self, NetCloseReason reason);
// bool netflow_close(NetFlow* self, NetCloseReason reason);
//
// Mark the flow dying and queue its terminal NET_FlowClosed event. Returns false if the flow
// was already closing. The public close() is this with NCR_AppClosed.
#define netflow_close(self, reason) NetFlow__close(NetFlow(self), reason)
 
_Ret_maybenull_ NetQueue* NetFlow__queue(_In_ NetFlow* self);
// NetQueue* netflow_queue(NetFlow* self);
//
// Resolve the flow's queue through socket->queue. Returns a strong reference the caller must
// release, or NULL if either weak arm has already been broken.
#define netflow_queue(self) NetFlow__queue(NetFlow(self))
 
void NetFlow__snapshotFlows(_In_ NetSocket* sock, _Out_ sa_NetFlow* out);
// void netflow_snapshotFlows(NetSocket* sock, sa_NetFlow* out);
//
// Flow table (flow.c)
// Snapshot a socket's live flows into `out`, each with a reference held, and hand the array to
// the caller to destroy. Initializes `out`; it must not already hold an array. Taken under the
// socket's flow lock and meant to be used outside it, because everything worth doing to a flow
// -- installing or dropping a filter chain, closing it -- allocates, primes, or reaches the
// wire, none of which may run with the flow table locked. Declared here rather than on
// NetSocket because sa_NetFlow is only visible from this side of the socket/flow include
// cycle, and because the rest of the flow table lives here too.
#define netflow_snapshotFlows(sock, out) NetFlow__snapshotFlows(NetSocket(sock), out)
 
void NetFlow__addFilter(_In_ NetFlow* self, _Inout_ NetSocket* sock, _In_ NetFilter* filter);
// void netflow_addFilter(NetFlow* self, NetSocket* sock, NetFilter* filter);
//
// Filter chain construction (filter.c)
//     // The socket owns a list of NetFilter factories; every flow it owns carries the parallel list
// of NetFlowFilter stages they produce. The chain is always built before a flow is reachable
// by anything else, so there is no window in which data could move past a flow that should
// have been filtered. Everything here is serialized by the flow's filterLock.
// Create one socket filter's stage for this flow and append it to the chain, allocating the
// staging buffer if this is the flow's first stage. A factory that returns NULL contributes
// nothing.
#define netflow_addFilter(self, sock, filter) NetFlow__addFilter(NetFlow(self), NetSocket(sock), NetFilter(filter))
 
void NetFlow__buildFilters(_In_ NetFlow* self, _Inout_ NetSocket* sock);
// void netflow_buildFilters(NetFlow* self, NetSocket* sock);
//
// Build the whole chain by walking the socket's filter list in order. A no-op for a socket
// with no filters, which is what keeps an unfiltered flow free of any filter cost at all.
#define netflow_buildFilters(self, sock) NetFlow__buildFilters(NetFlow(self), NetSocket(sock))
 
void NetFlow__clearFilters(_In_ NetFlow* self);
// void netflow_clearFilters(NetFlow* self);
//
// Drop the flow's chain, its staging storage, and anything the stages had buffered.
#define netflow_clearFilters(self) NetFlow__clearFilters(NetFlow(self))
 
void NetFlow__filterShutdown(_In_ NetFlow* self);
// void netflow_filterShutdown(NetFlow* self);
//
// Begin an orderly close on every stage (app -> wire). The caller runs one more encode pass
// afterwards to put whatever they produced (a TLS close_notify) on the wire.
#define netflow_filterShutdown(self) NetFlow__filterShutdown(NetFlow(self))
 
void NetFlow__filterNotify(_In_ NetFlow* self, _In_opt_ NetQueue* q, _In_opt_ NetSocket* sock, bool onWorker);
// void netflow_filterNotify(NetFlow* self, NetQueue* q, NetSocket* sock, bool onWorker);
//
// Deliver every notification the chain has raised, clearing each as it goes. `onWorker`
// delivers inline -- correct only from inside the flow's dispatch batch, where it also orders
// the notification ahead of data produced by the same pass; otherwise each is queued on the
// flow so it still lands on a worker in order. Must NOT be called with the flow's filterLock
// held: a handler can call straight back into the send path.
#define netflow_filterNotify(self, q, sock, onWorker) NetFlow__filterNotify(NetFlow(self), NetQueue(q), NetSocket(sock), onWorker)
 
bool NetFlow__filterStreamSend(_In_ NetFlow* self, _In_opt_ NetQueue* q, _Inout_ NetSocket* sock, _In_reads_bytes_opt_(len) const uint8* data, size_t len);
// bool netflow_filterStreamSend(NetFlow* self, NetQueue* q, NetSocket* sock, const uint8* data, size_t len);
//
// Filter data plane (dataplane.c)
//     // The four functions where a filtered flow's chain actually runs, all of them on the shared
// send/recv path rather than in any backend. A stage never touches the wire itself: these
// drivers own the socket's buffers and syscalls on its behalf, which is what stops a stage
// from bypassing one above it in the chain. Each takes the flow's filterLock for the duration
// of the chain walk and drops it before anything can call back into application code.
// Stream send: stage `data` (NULL/0 for a priming pass) into the flow's encIn, run the encode
// chain repeatedly until no stage has more to produce -- each stage reading the previous
// stage's encOut -- and flush what the wire-end stage produced onto the socket's send chain.
// Returns false on a fatal transform error, having closed the flow.
#define netflow_filterStreamSend(self, q, sock, data, len) NetFlow__filterStreamSend(NetFlow(self), NetQueue(q), NetSocket(sock), data, len)
 
void NetFlow__filterStreamRecv(_In_ NetFlow* self, _In_opt_ NetQueue* q, _Inout_ NetSocket* sock);
// void netflow_filterStreamRecv(NetFlow* self, NetQueue* q, NetSocket* sock);
//
// Stream recv: run the decode chain repeatedly, until no stage has more to produce, over the
// raw bytes already in the socket's receive ring -- each stage reading the decOut of the
// stage one hop toward the wire -- push out any wire-bound output the pass produced, deliver
// the stages' notifications, and deliver NET_DataReceived if the head stage produced
// application bytes. Runs on a worker (drainFlow).
#define netflow_filterStreamRecv(self, q, sock) NetFlow__filterStreamRecv(NetFlow(self), NetQueue(q), NetSocket(sock))
 
bool NetFlow__filterDatagramEncode(_In_ NetFlow* self, _In_opt_ NetQueue* q, _In_reads_bytes_opt_(len) const uint8* data, size_t len, _Inout_ NetMsgQueue* out, _Out_opt_ bool* fatalp);
// bool netflow_filterDatagramEncode(NetFlow* self, NetQueue* q, const uint8* data, size_t len, NetMsgQueue* out, bool* fatalp);
//
// Encode half of the datagram send: queue `data` (NULL/0 for a priming pass) on the flow's
// staging queue, run the encode chain repeatedly until no stage has more to produce, and
// collect what the wire-end stage produced into `out` rather than sending it. Split from the
// send below so the chain can be driven without a socket. `fatalp` distinguishes a stage
// failing (the flow is finished) from a payload merely being refused (the caller backs off);
// it may be NULL.
#define netflow_filterDatagramEncode(self, q, data, len, out, fatalp) NetFlow__filterDatagramEncode(NetFlow(self), NetQueue(q), data, len, out, fatalp)
 
bool NetFlow__filterDatagramSend(_In_ NetFlow* self, _In_opt_ NetQueue* q, _Inout_ NetSocket* sock, _In_reads_bytes_opt_(len) const uint8* data, size_t len);
// bool netflow_filterDatagramSend(NetFlow* self, NetQueue* q, NetSocket* sock, const uint8* data, size_t len);
//
// Datagram send: queue `data` (NULL/0 for a priming pass) on the flow's staging queue, run
// the encode chain repeatedly until no stage has more to produce, and send every message the
// wire-end stage produced to the flow's peer. Returns false if the payload could not be
// accepted (staging queue full, pool exhausted) or a stage failed fatally.
#define netflow_filterDatagramSend(self, q, sock, data, len) NetFlow__filterDatagramSend(NetFlow(self), NetQueue(q), NetSocket(sock), data, len)
 
bool NetFlow__filterDatagramRecv(_In_ NetFlow* self, _In_opt_ NetQueue* q, _Inout_ NetSocket* sock, _Inout_ NetMessage* msg, _Inout_ NetMsgQueue* out);
// bool netflow_filterDatagramRecv(NetFlow* self, NetQueue* q, NetSocket* sock, NetMessage* msg, NetMsgQueue* out);
//
// Datagram recv: run one received message through the decode chain (consuming it), send any
// wire-bound output the pass produced, and append whatever came out the application end to
// `out` for the caller to deliver. Runs on a worker (drainFlow). Returns false on a fatal
// transform error.
#define netflow_filterDatagramRecv(self, q, sock, msg, out) NetFlow__filterDatagramRecv(NetFlow(self), NetQueue(q), NetSocket(sock), msg, out)
 
void NetFlow__primeFilters(_In_ NetFlow* self, _In_opt_ NetQueue* q, _Inout_ NetSocket* sock);
// void netflow_primeFilters(NetFlow* self, NetQueue* q, NetSocket* sock);
//
// Run the encode side once with nothing staged, so a filter that initiates a negotiation gets
// to emit its first flight. Called when a chain is built (datagram), when the transport comes
// up (stream), and after shutdown() to flush a close record. A no-op for an unfiltered flow,
// or a stream flow whose socket is not connected yet.
#define netflow_primeFilters(self, q, sock) NetFlow__primeFilters(NetFlow(self), NetQueue(q), NetSocket(sock))
 
//
// needed to init lastActive
#define netflowClose(self) (self)->_->close(NetFlow(self))
#define netflowSend(self, data, len, flags) (self)->_->send(NetFlow(self), data, len, flags)
 
CX_C_END