diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index 48df406e..cbc7b1ca 100644 --- a/examples/CMakeLists.txt +++ b/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) diff --git a/examples/ClearConfigRepro/clear_config_repro.cpp b/examples/ClearConfigRepro/clear_config_repro.cpp new file mode 100644 index 00000000..71adac51 --- /dev/null +++ b/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 +#include + +#include +#include +#include +#include +#include +#include + +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 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(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= 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; +} diff --git a/src/cbdev/src/device_session.cpp b/src/cbdev/src/device_session.cpp index 10d13625..b008d4fe 100644 --- a/src/cbdev/src/device_session.cpp +++ b/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 DeviceSession::sendPacket(const cbPKT_GENERIC& pkt) { return Result::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 && diff --git a/tests/unit/test_device_session.cpp b/tests/unit/test_device_session.cpp index 7e49d67c..e97bbe8f 100644 --- a/tests/unit/test_device_session.cpp +++ b/tests/unit/test_device_session.cpp @@ -13,6 +13,7 @@ #include #include "cbdev/device_session.h" #include "cbdev/device_factory.h" +#include #include #include #include @@ -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::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::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"; +}