Group utils¶

Modules > utils

Modules¶

Classes¶

Public Functions¶

Macros¶

Public Functions Documentation¶

function dp_ctx_delete_stream¶

Delete the work-queue stream backing ctx's subject.

int dp_ctx_delete_stream (
    dp_pub_t * ctx
) 

Administrative, and deliberately never automatic: a work queue is shared infrastructure, and outliving any one producer is the feature, so closing a context must not end it. Call this only when the caller owns the queue's lifetime a test that made the subject up, or a tool tearing down what it provisioned.

Every frame still in the queue is destroyed with it, acked or not.

Parameters:

• ctx A context whose endpoint names the queue.

Returns:

DP_OK, or DP_ERR_INVALID (see dp_ctx_last_error()).

function dp_ctx_last_error¶

The backend's own account of the last failure on ctx .

const char * dp_ctx_last_error (
    const dp_pub_t * ctx
) 

dp_strerror() names the CLASS of error doppler returned; this names what the transport said, which is the part that distinguishes a slow broker from an absent one. Every non-OK publish status collapses into DP_ERR_SEND, so without this a caller sees "Send error" and has nothing to act on.

Parameters:

• ctx Any send-capable context.

Returns:

A NUL-terminated detail string, or "" when the last call succeeded or the backend offered nothing. Owned by ctx and valid until the next send on it; copy to keep.

function dp_ctx_set_timestamp_ns¶

Override the timestamp_ns the NEXT send onctx will stamp, instead of a freshdp_get_timestamp_ns() read.

void dp_ctx_set_timestamp_ns (
    dp_pub_t * ctx,
    uint64_t timestamp_ns
) 

One-shot: consumed (and cleared) by the very next send call on this context, whether or not it was actually used. Lets a hop that already knows a more precise or truer origin time (e.g. a value derived from dp_sample_clock_stamp_at() over an upstream message's own header, or a passthrough of that upstream header's own timestamp_ns) propagate it downstream instead of every hop silently re-stamping "now" and losing the connection to when the samples actually occurred. ctx accepts any socket role (dp_pub_t / dp_push_t / dp_req_t / dp_rep_t are the same underlying context type).

Parameters:

• ctx Allocated send-capable context (any role).
• timestamp_ns Nanoseconds since the UNIX epoch to stamp on the next send.

function dp_element_size¶

Bytes per payload element for a frame of this kind.

size_t dp_element_size (
    dp_frame_kind_t kind,
    dp_sample_type_t format
) 

For DP_KIND_IQ that is dp_sample_size() of format; for DP_KIND_TLM it is 16, one packed record, and format is not consulted because a record stream has no BLUE code.

Parameters:

• kind What the payload is (dp_frame_kind_t).
• format Sample format, for an I/Q frame.

Returns:

Bytes per element, or 0 when the pair is not something this build can send or decode.

function dp_get_timestamp_ns¶

Return the current wall-clock time as nanoseconds since the UNIX epoch.

uint64_t dp_get_timestamp_ns (
    void
) 

Uses CLOCK_REALTIME. Useful for timestamping samples before calling a send function, or for measuring round-trip latency.

Returns:

Nanoseconds since epoch.

function dp_host_rep¶

This machine's byte-order tag: DP_REP_LE orDP_REP_BE .

const char * dp_host_rep (
    void
) 

Four characters, not NUL-terminated. Derived at run time rather than compiled in, so a big-endian build tags its frames honestly instead of inheriting a constant nobody revisited.

function dp_sample_size¶

Return the byte size of one complex sample for type .

size_t dp_sample_size (
    dp_sample_type_t type
) 

Parameters:

• type Sample type enum value.

Returns:

Byte count (e.g. 2 for CI8, 4 for CI16, 8 for CI32/CF32, 16 for CF64).

function dp_sample_type_is_valid¶

True when type is a sample type this build knows.

int dp_sample_type_is_valid (
    dp_sample_type_t type
) 

Derived from dp_sample_size(), so there is one table: a type with no size is not a type. Ask this rather than range-testing the enum the values are append-only and a RETIRED one (2, the former CF128) sits inside the range while being invalid, so type <= CF32 accepts a value nothing can send or decode.

Parameters:

• type Sample type enum value.

Returns:

Non-zero when the type is known, 0 otherwise.

function dp_sample_type_str¶

Return a short string name for type ("CI8", "CI16", "CI32", "CF32", "CF64").

const char * dp_sample_type_str (
    dp_sample_type_t type
) 

Parameters:

• type Sample type enum value.

Returns:

Statically allocated, null-terminated string.

function dp_strerror¶

Return a human-readable description of an error code.

const char * dp_strerror (
    int err
) 

Parameters:

• err Negative error code returned by any dp_* function.

Returns:

Statically allocated, null-terminated string.

function dp_sub_reasm_stats¶

Read ctx's chunk-reassembly losses intoout .

int dp_sub_reasm_stats (
    const dp_sub_t * ctx,
    dp_reasm_stats_t * out
) 

Only a SUB reassembles (PUB is the only role that chunks), so only a SUB's counts move; any other receiving context reads all zeros, as does a SUB that never received a chunked frame. A frame that IS returned is complete and exact. Chunks of two frames are not merged, because a frame is identified by its whole header (sequence, timestamp, rate, centre frequency, format) and its chunk geometry, not by its sequence alone unless two publishers' headers are bit-identical, which the same explicit timestamp_ns can make them (#2017).

Read it from the receiving thread: a receive in progress on another thread updates the counts without a lock.

Parameters:

• ctx Any receiving context; only a SUB's counts move.
• out Filled on success.

Returns:

DP_OK, or DP_ERR_INVALID for a NULL argument.

// header-code: no-run=needs a NATS broker on 127.0.0.1:4222
dp_sub_t *sub = dp_sub_create ("nats://127.0.0.1:4222/iq");
dp_sub_set_timeout (sub, 1000);
dp_msg_t   *msg = NULL;
dp_header_t hdr;
if (sub && dp_sub_recv (sub, &msg, &hdr) == DP_OK)
  dp_msg_free (msg);
dp_reasm_stats_t st;
if (dp_sub_reasm_stats (sub, &st) == DP_OK && st.abandoned > 0)
  printf ("%llu partial frames given up\n",
          (unsigned long long)st.abandoned);
dp_sub_destroy (sub);

Macro Definition Documentation¶

define DP_WORK_QUEUE_MAX_ACK_PENDING¶

Unacked frames a Pull consumer may hold before the server stops.

#define DP_WORK_QUEUE_MAX_ACK_PENDING `1000`

A Pull is an explicit-ack consumer: past this many frames delivered and not yet acked (dp_msg_ack()), the server hands out no NEW frames, only redeliveries of the pending ones, each an AckWait after the last. That is the backpressure a work queue applies to a slow consumer, and the reason a consumer that never acks stalls at exactly this count.

Python reads it as doppler.stream.WORK_QUEUE_MAX_ACK_PENDING, so a consumer that reasons about redeliveries names this value rather than restating it.

define DP_WORK_QUEUE_MAX_AGE_NS¶

Retention bound doppler gives a work queue it creates itself.

#define DP_WORK_QUEUE_MAX_AGE_NS `(3600LL * 1000000000LL)`

A work queue drops a frame when a consumer ACKS it, so a frame nobody consumes is kept forever and the stream is file-backed. Created with no limits, a producer with no consumer is an unbounded disk sink (doppler#1136: 40 GB of residue from repeated test runs). An AGE bound is the one limit that cannot silently drop a frame a live consumer was about to take, unlike MaxBytes or MaxMsgs.

One hour, in nanoseconds. To choose differently, PRE-PROVISION the stream: doppler adopts an existing one as-is rather than reconfiguring it, which is how a Helm-created R=3 stream already works.