dtourolle
6a4f45f111
RouterNode and FilterNode were the last nodes on a data path still using the
throwing push() and swallowing the result:
try { out_ch_->push(val); } catch (const ChannelOverflowError&) {}
6595e6e made node outputs lossless, 28e0667 stopped them parking a worker,
a8cfe73 did the same for FanoutNode. These two were in none of them.
For ordinary values that is the familiar silent-loss problem: a dropped item
does not degrade a downstream result, it silently changes one, and the
consumer cannot tell it happened.
For a sentinel it is a hang. EOF is what tells every downstream node to shut
down and there is nothing after it to retry, so a filter that passes EOF by
predicate but drops it by backpressure produces a pipeline that never
terminates. scene-actor-extraction's decimator is exactly that shape —
`if (f.eof) return true;` in the predicate, feeding a chain whose slowest
node is an ONNX embedder, so the output is reliably full when EOF arrives.
Everything downstream then waits forever for a token that was discarded, and
the run has to be killed.
Both now route sentinels out-of-band via push_sentinel, which consumes no
ring capacity and cannot overflow, and retry ordinary values until taken.
Like FanoutNode and unlike a pool node, these own a private thread, so
waiting costs no scheduler worker and needs no space-callback park;
stop_flag_ is rechecked every pass so teardown cannot hang on a full output.
Time spent parked is charged to blocked rather than exec, so a held-up node
does not report as busy. An out-of-range router selector still drops by
design — that item was routed nowhere, which is not the same as lost.
is_sentinel_value moves from pool_node.hpp to traits.hpp. Every node type
that forwards a value needs it; these two not having it is the bug.
Verified in both directions. On c73edff the new case delivers 6 of 40 values
and never sets saw_eof; here it delivers 40 and terminates. EOF is emitted
exactly once, as a real source does — a test source that re-offered it would
mask the bug, since a later attempt could find the channel drained.
Note for downstream: the decimator is now a backpressure point rather than a
relief valve, so the source throttles to the face branch instead of quietly
thinning it. That is the intended behaviour, but it changes the shape of a
loaded run and is worth a benchmark comparison on a known clip.
138 lines
5.3 KiB
C++
138 lines
5.3 KiB
C++
#pragma once
|
|
#include <tuple>
|
|
#include <type_traits>
|
|
|
|
namespace kpn {
|
|
|
|
// ── Primary template — not defined; only specialisations match ────────────────
|
|
template<typename F>
|
|
struct function_traits;
|
|
|
|
// Free function
|
|
template<typename R, typename... Args>
|
|
struct function_traits<R(Args...)> {
|
|
using return_t = R;
|
|
using args = std::tuple<Args...>;
|
|
static constexpr std::size_t arity = sizeof...(Args);
|
|
};
|
|
|
|
// Function pointer
|
|
template<typename R, typename... Args>
|
|
struct function_traits<R(*)(Args...)> : function_traits<R(Args...)> {};
|
|
|
|
// Member function pointer (const)
|
|
template<typename C, typename R, typename... Args>
|
|
struct function_traits<R(C::*)(Args...) const> : function_traits<R(Args...)> {};
|
|
|
|
// Member function pointer (non-const)
|
|
template<typename C, typename R, typename... Args>
|
|
struct function_traits<R(C::*)(Args...)> : function_traits<R(Args...)> {};
|
|
|
|
// Callable (lambda / std::function) — delegate to operator()
|
|
template<typename F>
|
|
struct function_traits : function_traits<decltype(&F::operator())> {};
|
|
|
|
// ── Helpers ───────────────────────────────────────────────────────────────────
|
|
|
|
template<typename F>
|
|
using return_t = typename function_traits<std::remove_cvref_t<F>>::return_t;
|
|
|
|
template<typename F>
|
|
using args_t = typename function_traits<std::remove_cvref_t<F>>::args;
|
|
|
|
template<typename F>
|
|
inline constexpr std::size_t arity_v = function_traits<std::remove_cvref_t<F>>::arity;
|
|
|
|
// ── Tuple detection ───────────────────────────────────────────────────────────
|
|
|
|
template<typename T>
|
|
struct is_tuple : std::false_type {};
|
|
|
|
template<typename... Ts>
|
|
struct is_tuple<std::tuple<Ts...>> : std::true_type {};
|
|
|
|
template<typename T>
|
|
inline constexpr bool is_tuple_v = is_tuple<T>::value;
|
|
|
|
// ── Normalise return type to always be a tuple ────────────────────────────────
|
|
// void → std::tuple<>
|
|
// T (non-tup) → std::tuple<T>
|
|
// tuple<...> → tuple<...> (unchanged)
|
|
|
|
template<typename T>
|
|
struct normalise_return {
|
|
using type = std::tuple<T>;
|
|
};
|
|
|
|
template<>
|
|
struct normalise_return<void> {
|
|
using type = std::tuple<>;
|
|
};
|
|
|
|
template<typename... Ts>
|
|
struct normalise_return<std::tuple<Ts...>> {
|
|
using type = std::tuple<Ts...>;
|
|
};
|
|
|
|
template<typename T>
|
|
using normalised_return_t = typename normalise_return<T>::type;
|
|
|
|
// ── Output count from a function type ────────────────────────────────────────
|
|
|
|
template<typename F>
|
|
inline constexpr std::size_t output_count_v =
|
|
std::tuple_size_v<normalised_return_t<return_t<F>>>;
|
|
|
|
// ── repeat_tuple: std::tuple<T, T, ..., T> with N repetitions ────────────────
|
|
|
|
template<typename T, std::size_t N, typename Seq = std::make_index_sequence<N>>
|
|
struct repeat_tuple;
|
|
|
|
template<typename T, std::size_t N, std::size_t... Is>
|
|
struct repeat_tuple<T, N, std::index_sequence<Is...>> {
|
|
template<std::size_t> using always_T = T;
|
|
using type = std::tuple<always_T<Is>...>;
|
|
};
|
|
|
|
template<typename T, std::size_t N>
|
|
using repeat_tuple_t = typename repeat_tuple<T, N>::type;
|
|
|
|
// ── Sentinel detection ────────────────────────────────────────────────────────
|
|
// A value is a "sentinel" (must-deliver control token, e.g. EOF) if its type
|
|
// carries a bool-convertible eof flag — either directly (`v.eof`, as on a raw
|
|
// source Frame) or nested one level under a `.source` member (`v.source.eof`,
|
|
// as on message types that wrap the originating Frame). Sentinels are delivered
|
|
// losslessly and non-blockingly via Channel::push_sentinel() instead of the
|
|
// throwing push(), so backpressure can never drop the token that unblocks
|
|
// downstream teardown.
|
|
//
|
|
// Types with neither shape are never treated as sentinels — both traits are
|
|
// SFINAE-safe and the runtime check compiles away to `false` for them, so this
|
|
// stays a no-op for pipelines that don't use an eof convention.
|
|
//
|
|
// Lives here rather than in pool_node.hpp because every node type that forwards
|
|
// values needs it, not just the pool-scheduled ones. FilterNode and RouterNode
|
|
// not having it is what let an EOF token be dropped on a full output.
|
|
|
|
template<typename T, typename = void>
|
|
struct has_eof_field : std::false_type {};
|
|
template<typename T>
|
|
struct has_eof_field<T, std::void_t<decltype(static_cast<bool>(std::declval<const T&>().eof))>>
|
|
: std::true_type {};
|
|
|
|
template<typename T, typename = void>
|
|
struct has_source_eof_field : std::false_type {};
|
|
template<typename T>
|
|
struct has_source_eof_field<T,
|
|
std::void_t<decltype(static_cast<bool>(std::declval<const T&>().source.eof))>>
|
|
: std::true_type {};
|
|
|
|
template<typename T>
|
|
constexpr bool is_sentinel_value(const T& v) {
|
|
if constexpr (has_eof_field<T>::value) return static_cast<bool>(v.eof);
|
|
else if constexpr (has_source_eof_field<T>::value) return static_cast<bool>(v.source.eof);
|
|
else return false;
|
|
}
|
|
|
|
} // namespace kpn
|