Pace configuration-packet sends to avoid device UDP buffer overrun

examples/CMakeLists.txt

@@ -8,6 +8,7 @@ set(CBSDK_EXAMPLES
     "central_client:CentralClient/central_client.cpp"
     "ccf_test:CCFTest/ccf_test.cpp"
     "recording_test:RecordingTest/recording_test.cpp"
+    "clear_config_repro:ClearConfigRepro/clear_config_repro.cpp"
 )
 
 # cbdev examples (link against cbdev only - no cbshm, no cbsdk)

examples/ClearConfigRepro/clear_config_repro.cpp

@@ -0,0 +1,134 @@
+///////////////////////////////////////////////////////////////////////////////////////////////////
+/// @file   clear_config_repro.cpp
+/// @brief  Reproduce Orion's "Clear-config sync failed: Response timeout" against a real device.
+///
+/// Faithfully mirrors Orion CereLink.cpp initialConfig()'s clear-config phase:
+///   setSampleGroup(NONE) -> per-channel clear LNC (CHANSETAINP)
+///   -> per-channel clear spike (CHANSETSPK) -> sync(), retried in a loop.
+///
+/// Usage: clear_config_repro [HUB1|HUB2|HUB3|NSP|GEMINI_NSP|NPLAY]   (default HUB1)
+///////////////////////////////////////////////////////////////////////////////////////////////////
+
+#include <cbsdk/sdk_session.h>
+#include <cbproto/cbproto.h>
+
+#include <chrono>
+#include <iostream>
+#include <string>
+#include <thread>
+#include <vector>
+#include <cstdlib>
+
+using cbsdk::SdkSession;
+using cbsdk::SdkConfig;
+using cbsdk::ChannelType;
+
+static cbsdk::DeviceType parseType(const std::string& s) {
+    if (s == "NSP") return cbsdk::DeviceType::LEGACY_NSP;
+    if (s == "GEMINI_NSP") return cbsdk::DeviceType::NSP;
+    if (s == "HUB2") return cbsdk::DeviceType::HUB2;
+    if (s == "HUB3") return cbsdk::DeviceType::HUB3;
+    if (s == "NPLAY") return cbsdk::DeviceType::NPLAY;
+    return cbsdk::DeviceType::HUB1;
+}
+
+int main(int argc, char* argv[]) {
+    const std::string type_str = (argc > 1) ? argv[1] : "HUB1";
+
+    SdkConfig config;
+    config.device_type = parseType(type_str);
+    config.autorun = true;
+    config.non_blocking = false;
+
+    std::cout << "Connecting to " << type_str << " ...\n";
+    auto result = SdkSession::create(config);
+    if (result.isError()) {
+        std::cout << "  create() failed: " << result.error() << "\n";
+        return 1;
+    }
+    auto session = std::move(result.value());
+    std::cout << "  connected. protocol code=" << session.getProtocolVersion()
+              << "  ident=\"" << session.getProcIdent() << "\"";
+    if (const auto* si = session.getSysInfo()) {
+        std::cout << "  sysfreq=" << si->sysfreq;
+    }
+    std::cout << "\n";
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(500));
+
+    // Gather all analog-in channels (mirrors Orion's allChannels).
+    std::vector<uint32_t> allChannels;
+    for (uint32_t ch = 1; ch <= cbNUM_ANALOG_CHANS; ++ch) {
+        if (session.getChanInfo(ch) != nullptr) allChannels.push_back(ch);
+    }
+    std::cout << "  discovered " << allChannels.size() << " analog channels\n";
+    if (allChannels.empty()) return 2;
+
+    // Orion clears these bits (see CereLink.cpp clearLineNoiseCancellation / clearSpikeProcessing).
+    constexpr uint32_t SPIKE_PROCESSING_OPTION_MASK =
+        cbAINPSPK_EXTRACT | cbAINPSPK_REJART | cbAINPSPK_REJCLIP | cbAINPSPK_ALIGNPK |
+        cbAINPSPK_REJAMPL | cbAINPSPK_THRLEVEL | cbAINPSPK_THRENERGY | cbAINPSPK_THRAUTO |
+        cbAINPSPK_ALLSORT;
+
+    auto applyClearConfig = [&]() -> bool {
+        auto grp = session.setSampleGroup(static_cast<uint32_t>(allChannels.size()),
+                                          ChannelType::ANALOG_IN,
+                                          cbsdk::SampleRate::NONE, false, allChannels.data());
+        if (grp.isError()) { std::cout << "    setSampleGroup: " << grp.error() << "\n"; return false; }
+
+        // Optional pacing: env PACE=<n> sleeps 1ms every n setChannelConfig sends.
+        const char* pace_env = std::getenv("PACE");
+        const int pace = pace_env ? std::atoi(pace_env) : 0;
+        int sent = 0;
+        auto paced = [&]() {
+            if (pace > 0 && (++sent % pace) == 0)
+                std::this_thread::sleep_for(std::chrono::milliseconds(1));
+        };
+
+        for (uint32_t ch : allChannels) {          // clear LNC
+            const auto* base = session.getChanInfo(ch);
+            if (!base) return false;
+            cbPKT_CHANINFO ci = *base;
+            ci.chan = ch;
+            ci.cbpkt_header.type = cbPKTTYPE_CHANSETAINP;
+            ci.ainpopts &= ~(cbAINP_LNC_MASK | cbAINP_RAWSTREAM);
+            ci.lncrate = 0;
+            if (auto r = session.setChannelConfig(ci); r.isError()) {
+                std::cout << "    clearLNC ch" << ch << ": " << r.error() << "\n"; return false;
+            }
+            paced();
+        }
+        for (uint32_t ch : allChannels) {          // clear spike processing
+            const auto* base = session.getChanInfo(ch);
+            if (!base) return false;
+            cbPKT_CHANINFO ci = *base;
+            ci.chan = ch;
+            ci.cbpkt_header.type = cbPKTTYPE_CHANSETSPK;
+            ci.spkopts &= ~SPIKE_PROCESSING_OPTION_MASK;
+            if (auto r = session.setChannelConfig(ci); r.isError()) {
+                std::cout << "    clearSpike ch" << ch << ": " << r.error() << "\n"; return false;
+            }
+            paced();
+        }
+
+        auto s = session.sync(3000);
+        if (s.isError()) { std::cout << "    sync: " << s.error() << "\n"; return false; }
+        return true;
+    };
+
+    // Orion's runConfigLoop retries for ~5 s.
+    const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(15);
+    int attempt = 0;
+    while (std::chrono::steady_clock::now() < deadline) {
+        ++attempt;
+        std::cout << "  clear-config attempt " << attempt << ":\n";
+        if (applyClearConfig()) {
+            std::cout << "\nRESULT: clear-config + sync SUCCEEDED (attempt " << attempt << ").\n";
+            return 0;
+        }
+        std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    }
+
+    std::cout << "\nRESULT: clear-config timed out (reproduces Orion failure).\n";
+    return 3;
+}

src/cbdev/src/device_session.cpp

@@ -204,6 +204,16 @@ struct DeviceSession::Impl {
     // Device configuration (from REQCONFIGALL)
     cbproto::DeviceConfig device_config{};
 
+    // Outbound pacing for configuration packets. A device's UDP receive buffer
+    // can be as small as ~8 KB (~8 CHANINFO packets); a caller that sends many
+    // per-channel setChannelConfig() packets back-to-back (e.g. clearing LNC /
+    // spike processing on every channel) can overrun it, dropping packets —
+    // including a following runlevel sync barrier, which then times out. We
+    // enforce a minimum gap between consecutive configuration-channel sends so
+    // bursts drain at a rate the device can sustain. Isolated sends (the gap
+    // has already elapsed) and data packets are unaffected.
+    std::chrono::steady_clock::time_point last_config_send{};
+
     // Timestamp conversion for non-Gemini devices
     // Gemini devices, our primary use case, send timestamps in nanoseconds; default to true.
     // Non-Gemini (i.e. NPlay and Legacy NSP) send sample counts.
@@ -564,6 +574,21 @@ Result<void> DeviceSession::sendPacket(const cbPKT_GENERIC& pkt) {
         return Result<void>::error("Device not connected");
     }
 
+    // Pace bursts of configuration packets so they drain at a rate the device
+    // can sustain (see Impl::last_config_send). Only configuration-channel
+    // packets are throttled; streaming/data sends and isolated config sends
+    // (where the gap has already elapsed) incur no delay.
+    if ((pkt.cbpkt_header.chid & cbPKTCHAN_CONFIGURATION) == cbPKTCHAN_CONFIGURATION) {
+        constexpr auto kConfigSendMinInterval = std::chrono::microseconds(200);
+        if (m_impl->last_config_send.time_since_epoch().count() != 0) {
+            const auto elapsed = std::chrono::steady_clock::now() - m_impl->last_config_send;
+            if (elapsed < kConfigSendMinInterval) {
+                std::this_thread::sleep_for(kConfigSendMinInterval - elapsed);
+            }
+        }
+        m_impl->last_config_send = std::chrono::steady_clock::now();
+    }
+
     // Non-Gemini: convert nanosecond timestamp back to device clock ticks.
     // This is the inverse of the ticks→ns conversion in receivePackets().
     if (!m_impl->timestamps_are_nanoseconds && m_impl->ts_convert_den > 1 &&

tests/unit/test_device_session.cpp

@@ -13,6 +13,7 @@
 #include <gtest/gtest.h>
 #include "cbdev/device_session.h"
 #include "cbdev/device_factory.h"
+#include <cbproto/cbproto.h>
 #include <cstring>
 #include <thread>
 #include <chrono>
@@ -252,3 +253,46 @@ TEST_F(DeviceSessionTest, Error_SendPacketsEmpty) {
     auto send_result = session->sendPackets(empty_pkts);
     EXPECT_TRUE(send_result.isError());
 }
+
+///////////////////////////////////////////////////////////////////////////////////////////////////
+// Configuration-packet send pacing
+//
+// Bursts of per-channel setChannelConfig() sends must be throttled so they don't
+// overrun a device's small UDP receive buffer (which drops packets, including a
+// following runlevel sync barrier). sendPacket() enforces a minimum gap between
+// consecutive configuration-channel sends; streaming/data packets are exempt.
+// std::this_thread::sleep_for only ever sleeps AT LEAST the requested time, so
+// the lower-bound timing assertion below cannot be flaky.
+///////////////////////////////////////////////////////////////////////////////////////////////////
+
+TEST_F(DeviceSessionTest, ConfigSends_ArePaced_DataSends_AreNot) {
+    auto config = ConnectionParams::custom("127.0.0.1", "0.0.0.0", 51045, 51046);
+    auto result = createDeviceSession(config, ProtocolVersion::PROTOCOL_CURRENT);
+    ASSERT_TRUE(result.isOk()) << result.error();
+    auto& session = result.value();
+
+    constexpr int kBurst = 40;  // 39 gaps * 200us >= ~7.8 ms guaranteed for config
+
+    // Data-channel packets (chid without the configuration bit) are not throttled.
+    cbPKT_GENERIC data{};
+    data.cbpkt_header.chid = 1;  // a real acquisition channel, not configuration
+    data.cbpkt_header.type = 0x0006;
+    const auto data_start = std::chrono::steady_clock::now();
+    for (int i = 0; i < kBurst; ++i) ASSERT_TRUE(session->sendPacket(data).isOk());
+    const auto data_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
+        std::chrono::steady_clock::now() - data_start).count();
+
+    // Configuration-channel packets are paced.
+    cbPKT_GENERIC cfg{};
+    cfg.cbpkt_header.chid = cbPKTCHAN_CONFIGURATION;
+    cfg.cbpkt_header.type = cbPKTTYPE_CHANSET;
+    const auto cfg_start = std::chrono::steady_clock::now();
+    for (int i = 0; i < kBurst; ++i) ASSERT_TRUE(session->sendPacket(cfg).isOk());
+    const auto cfg_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
+        std::chrono::steady_clock::now() - cfg_start).count();
+
+    // Config burst is meaningfully paced; data burst is effectively free.
+    EXPECT_GE(cfg_ms, 4) << "configuration sends were not throttled";
+    EXPECT_LT(data_ms, 4) << "data sends should not be throttled";
+    EXPECT_GT(cfg_ms, data_ms) << "config sends should be slower than data sends";
+}