new cyclic writer

.gitignore

@@ -0,0 +1,16 @@
+# Build outputs
+capture
+*.exe
+*.obj
+
+# Large capture spools and generated data dumps
+*.capture_spool.bin
+capture.csv
+capture.svg
+packets.csv
+packets.svg
+test.csv
+test.svg
+internal_default.csv
+internal_default.svg
+*_test.*

main.cpp

@@ -381,7 +381,7 @@ void print_help(const char* exe_name) {
         << "                [internal_ref_hz:2000000]\n"
         << "                [di1:zero|trace|ignore]\n"
         << "                [duration_ms:100] [packet_limit:0] [csv:capture.csv] [svg:capture.svg]\n"
-        << "                [live_html:live_plot.html] [live_json:live_plot.json] [tty:/dev/ttyADC_data]\n"
+        << "                [live_html:live_plot.html] [live_json:live_plot.json] [tty:/tmp/ttyADC_data]\n"
         << "                [recv_block:32768] [stats_period_ms:1000] [live_update_period_ms:1000] [svg_history_packets:50] [start_wait_ms:10000]\n"
         << "                [buffer_words:8388608] [step_words:32768]\n"
         << "                [pullup_syn1] [pullup_syn2] [pulldown_conv_in] [pulldown_start_in]\n"
@@ -412,7 +412,7 @@ void print_help(const char* exe_name) {
         << "  buffer_words:8388608 -> input stream buffer size in 32-bit words\n"
         << "  step_words:32768 -> input stream transfer step in 32-bit words\n"
         << "  live_html/live_json -> live graph files updated as packets arrive\n"
-        << "  tty:/dev/ttyADC_data -> write binary step frames to a Linux/POSIX tty path\n"
+        << "  tty:/tmp/ttyADC_data -> write a continuous legacy 4-word CH1/CH2 stream to a Linux/POSIX tty or PTY link path\n"
         << "  If sample_clock_hz is omitted together with clock:internal, the maximum ADC speed is used\n"
         << "\n"
         << "Differential physical channel mapping:\n"
@@ -747,6 +747,19 @@ Fn load_symbol(ModuleHandle module, const char* name) {
 #endif
 }
 
+template <typename Fn>
+Fn try_load_symbol(ModuleHandle module, const char* name) {
+#ifdef _WIN32
+    const auto addr = GetProcAddress(module, name);
+    return reinterpret_cast<Fn>(addr);
+#else
+    dlerror();
+    void* addr = dlsym(module, name);
+    (void) dlerror();
+    return reinterpret_cast<Fn>(addr);
+#endif
+}
+
 struct Api {
     ModuleHandle x502_module = nullptr;
     ModuleHandle e502_module = nullptr;
@@ -755,6 +768,7 @@ struct Api {
     decltype(&X502_Free) Free = nullptr;
     decltype(&X502_Close) Close = nullptr;
     decltype(&X502_GetErrorString) GetErrorString = nullptr;
+    decltype(&X502_GetDevInfo) GetDevInfo = nullptr;
     decltype(&X502_GetDevInfo2) GetDevInfo2 = nullptr;
     decltype(&X502_SetMode) SetMode = nullptr;
     decltype(&X502_StreamsStop) StreamsStop = nullptr;
@@ -804,7 +818,11 @@ struct Api {
         Free = load_symbol<decltype(Free)>(x502_module, "X502_Free");
         Close = load_symbol<decltype(Close)>(x502_module, "X502_Close");
         GetErrorString = load_symbol<decltype(GetErrorString)>(x502_module, "X502_GetErrorString");
-        GetDevInfo2 = load_symbol<decltype(GetDevInfo2)>(x502_module, "X502_GetDevInfo2");
+        GetDevInfo = try_load_symbol<decltype(GetDevInfo)>(x502_module, "X502_GetDevInfo");
+        GetDevInfo2 = try_load_symbol<decltype(GetDevInfo2)>(x502_module, "X502_GetDevInfo2");
+        if ((GetDevInfo == nullptr) && (GetDevInfo2 == nullptr)) {
+            fail("Neither X502_GetDevInfo nor X502_GetDevInfo2 is available in x502 API library");
+        }
         SetMode = load_symbol<decltype(SetMode)>(x502_module, "X502_SetMode");
         StreamsStop = load_symbol<decltype(StreamsStop)>(x502_module, "X502_StreamsStop");
         StreamsDisable = load_symbol<decltype(StreamsDisable)>(x502_module, "X502_StreamsDisable");
@@ -859,6 +877,8 @@ void expect_ok(const Api& api, int32_t err, const std::string& what) {
 constexpr uint32_t kE502DiSyn2Mask =
     (static_cast<uint32_t>(1U) << 13U) | (static_cast<uint32_t>(1U) << 17U);
 constexpr uint32_t kE502Digital1Mask = (static_cast<uint32_t>(1U) << 0U);
+constexpr uint32_t kStreamInputAdcFlag = 0x80000000U;
+constexpr uint32_t kStreamInputCalibratedAdcFlag = 0x40000000U;
 
 using TickMs = uint64_t;
 
@@ -970,38 +990,10 @@ struct PacketAccumulator {
     }
 };
 
-struct TtyStepAccumulator {
-    double sum_ch1 = 0.0;
-    double sum_ch2 = 0.0;
-    uint32_t count_ch1 = 0;
-    uint32_t count_ch2 = 0;
-    uint32_t next_step_index = 1;
-
-    void clear_step() {
-        sum_ch1 = 0.0;
-        sum_ch2 = 0.0;
-        count_ch1 = 0;
-        count_ch2 = 0;
-    }
-
-    void reset_packet() {
-        clear_step();
-        next_step_index = 1;
-    }
-
-    void add_sample(uint32_t lch, double raw_code) {
-        if (lch == 0U) {
-            sum_ch1 += raw_code;
-            ++count_ch1;
-        } else if (lch == 1U) {
-            sum_ch2 += raw_code;
-            ++count_ch2;
-        }
-    }
-
-    bool has_complete_step() const {
-        return (count_ch1 != 0U) && (count_ch2 != 0U);
-    }
+struct TtyContinuousState {
+    bool pending_ch1_valid = false;
+    int16_t pending_ch1 = 0;
+    uint16_t next_index = 1;
 };
 
 int16_t pack_raw_code_to_int16(double avg_raw_code) {
@@ -1012,6 +1004,30 @@ int16_t pack_raw_code_to_int16(double avg_raw_code) {
     return static_cast<int16_t>(clamped);
 }
 
+bool try_extract_raw_adc_code(uint32_t word, double& raw_code) {
+    if ((word & kStreamInputAdcFlag) == 0U) {
+        return false;
+    }
+
+    int32_t value = 0;
+    if ((word & kStreamInputCalibratedAdcFlag) != 0U) {
+        const uint32_t payload = word & 0x00FFFFFFU;
+        value = static_cast<int32_t>(payload);
+        if ((payload & 0x00800000U) != 0U) {
+            value |= static_cast<int32_t>(0xFF000000U);
+        }
+    } else {
+        const uint32_t payload = word & 0x0000FFFFU;
+        value = static_cast<int32_t>(payload);
+        if ((payload & 0x00008000U) != 0U) {
+            value |= static_cast<int32_t>(0xFFFF0000U);
+        }
+    }
+
+    raw_code = static_cast<double>(value);
+    return true;
+}
+
 struct ConsoleCtrlGuard {
     bool installed = false;
 
@@ -1113,7 +1129,11 @@ int run(const Config& cfg) {
     device.opened = true;
 
     t_x502_info info{};
+    if (api.GetDevInfo2 != nullptr) {
         err = api.GetDevInfo2(device.hnd, &info, sizeof(info));
+    } else {
+        err = api.GetDevInfo(device.hnd, &info);
+    }
     expect_ok(api, err, "Get device info");
     print_device_info(info);
 
@@ -1230,16 +1250,29 @@ int run(const Config& cfg) {
               << "  ADC range: +/-" << range_to_volts(cfg.range) << " V\n"
               << "  max frames per packet per channel: " << target_frames << "\n";
 
+    const uint32_t read_capacity_words = std::max(cfg.recv_block_words, cfg.input_step_words);
+    std::size_t tty_ring_capacity_bytes = 0;
     std::unique_ptr<TtyProtocolWriter> tty_writer;
     if (cfg.tty_path) {
-        tty_writer = std::make_unique<TtyProtocolWriter>(*cfg.tty_path);
+        const uint64_t typical_tty_batch_frames = (static_cast<uint64_t>(read_capacity_words) + 1U) / 2U;
+        const uint64_t typical_tty_batch_bytes = typical_tty_batch_frames * sizeof(uint16_t) * 4U;
+        const uint64_t required_tty_ring_bytes = typical_tty_batch_bytes * 16U;
+        const uint64_t tty_ring_bytes_u64 = std::max<uint64_t>(1024U * 1024U, required_tty_ring_bytes);
+        if (tty_ring_bytes_u64 > static_cast<uint64_t>(std::numeric_limits<std::size_t>::max())) {
+            fail("TTY ring buffer size overflowed size_t");
+        }
+        tty_ring_capacity_bytes = static_cast<std::size_t>(tty_ring_bytes_u64);
+        tty_writer = std::make_unique<TtyProtocolWriter>(*cfg.tty_path, tty_ring_capacity_bytes);
+        tty_writer->emit_packet_start();
     }
     CaptureFileWriter writer(cfg.csv_path, cfg.svg_path, cfg.live_html_path, cfg.live_json_path);
     writer.initialize_live_plot();
     writer.initialize_csv(cfg.channel_count, cfg.di1_mode == Di1Mode::Trace);
     std::cout << "  live plot html: " << writer.live_html_path() << "\n"
               << "  live plot data: " << writer.live_json_path() << "\n"
-              << "  tty step output: " << (cfg.tty_path ? *cfg.tty_path : std::string("disabled")) << "\n"
+              << "  tty stream output: " << (cfg.tty_path ? *cfg.tty_path : std::string("disabled")) << "\n"
+              << "  tty ring buffer bytes: "
+              << (cfg.tty_path ? std::to_string(tty_ring_capacity_bytes) : std::string("disabled")) << "\n"
               << "  recv block words: " << cfg.recv_block_words << "\n"
               << "  input step words: " << cfg.input_step_words << "\n"
               << "  input buffer words: " << cfg.input_buffer_words << "\n"
@@ -1255,17 +1288,17 @@ int run(const Config& cfg) {
     expect_ok(api, api.StreamsStart(device.hnd), "Start streams");
     device.streams_started = true;
 
-    const uint32_t read_capacity_words = std::max(cfg.recv_block_words, cfg.input_step_words);
     std::vector<uint32_t> raw(read_capacity_words);
     std::vector<double> adc_buffer(read_capacity_words);
     std::vector<double> adc_raw_buffer(read_capacity_words);
     std::vector<uint32_t> din_buffer(read_capacity_words);
     std::deque<double> pending_adc;
-    std::deque<double> pending_adc_raw;
     std::deque<uint32_t> pending_din;
     std::deque<CapturePacket> packets;
     PacketAccumulator current_packet;
-    TtyStepAccumulator tty_step;
+    TtyContinuousState tty_state;
+    std::vector<uint16_t> tty_frame_words;
+    tty_frame_words.reserve(static_cast<std::size_t>(read_capacity_words) * 2U);
     current_packet.reset(target_frames, cfg.channel_count);
     std::size_t csv_global_frame_index = 0;
 
@@ -1298,6 +1331,8 @@ int run(const Config& cfg) {
     uint64_t stats_zeroed_samples = 0;
     uint64_t stats_completed_frames = 0;
     uint64_t stats_completed_packets = 0;
+    TtyProtocolWriter::StatsSnapshot tty_stats_window_start {};
+    bool tty_overflow_warning_printed = false;
 
     auto print_stats = [&](bool final_report) {
         const TickMs now = tick_count_ms();
@@ -1310,6 +1345,11 @@ int run(const Config& cfg) {
         const double zeroed_fraction = (stats_stored_adc_samples == 0U)
             ? 0.0
             : (100.0 * static_cast<double>(stats_zeroed_samples) / static_cast<double>(stats_stored_adc_samples));
+        const TtyProtocolWriter::StatsSnapshot tty_stats_now =
+            tty_writer ? tty_writer->stats() : TtyProtocolWriter::StatsSnapshot {};
+        const uint64_t tty_frames_written = tty_stats_now.frames_written - tty_stats_window_start.frames_written;
+        const uint64_t tty_frames_dropped = tty_stats_now.frames_dropped - tty_stats_window_start.frames_dropped;
+        const uint64_t tty_ring_overflows = tty_stats_now.ring_overflows - tty_stats_window_start.ring_overflows;
 
         std::cout << std::fixed << std::setprecision(3)
                   << (final_report ? "Final stats: " : "Online stats: ")
@@ -1324,6 +1364,11 @@ int run(const Config& cfg) {
         } else if (cfg.di1_mode == Di1Mode::Trace) {
             std::cout << ", DI1 trace=enabled";
         }
+        if (tty_writer) {
+            std::cout << ", tty_frames_written=" << tty_frames_written
+                      << ", tty_frames_dropped=" << tty_frames_dropped
+                      << ", tty_ring_overflows=" << tty_ring_overflows;
+        }
         std::cout << "\n";
 
         if (!final_report) {
@@ -1336,6 +1381,7 @@ int run(const Config& cfg) {
             stats_zeroed_samples = 0;
             stats_completed_frames = 0;
             stats_completed_packets = 0;
+            tty_stats_window_start = tty_stats_now;
         }
     };
 
@@ -1344,29 +1390,9 @@ int run(const Config& cfg) {
             return;
         }
         current_packet.reset(target_frames, cfg.channel_count);
-        if (tty_writer) {
-            tty_step.reset_packet();
-            tty_writer->emit_packet_start();
-        }
         packet_active = true;
     };
 
-    auto emit_tty_step = [&]() {
-        if (!tty_writer || !tty_step.has_complete_step()) {
-            return;
-        }
-        if (tty_step.next_step_index >= 0xFFFFU) {
-            fail("TTY protocol step index overflow within packet");
-        }
-
-        const double ch1_avg = tty_step.sum_ch1 / static_cast<double>(tty_step.count_ch1);
-        const double ch2_avg = tty_step.sum_ch2 / static_cast<double>(tty_step.count_ch2);
-        tty_writer->emit_step(static_cast<uint16_t>(tty_step.next_step_index),
-                              pack_raw_code_to_int16(ch1_avg),
-                              pack_raw_code_to_int16(ch2_avg));
-        ++tty_step.next_step_index;
-    };
-
     auto finish_packet = [&](PacketCloseReason reason) {
         const std::size_t frames = current_packet.frame_count(cfg.channel_count);
         if (frames != 0U) {
@@ -1449,10 +1475,12 @@ int run(const Config& cfg) {
 
         packet_active = false;
         current_packet.reset(target_frames, cfg.channel_count);
-        tty_step.clear_step();
     };
 
     while (!stop_loop_requested) {
+        if (tty_writer) {
+            tty_writer->throw_if_failed();
+        }
         if ((cfg.packet_limit != 0U) && (total_completed_packets >= cfg.packet_limit)) {
             stop_loop_requested = true;
             break;
@@ -1531,18 +1559,51 @@ int run(const Config& cfg) {
 
         uint32_t raw_adc_count = 0;
         if (tty_writer) {
-            raw_adc_count = static_cast<uint32_t>(adc_raw_buffer.size());
-            const int32_t raw_process_err = api.ProcessData(device.hnd,
-                                                            raw.data(),
-                                                            static_cast<uint32_t>(recvd),
-                                                            0,
-                                                            adc_raw_buffer.data(),
-                                                            &raw_adc_count,
-                                                            nullptr,
-                                                            nullptr);
-            expect_ok(api, raw_process_err, "Process raw ADC data");
+            for (std::size_t i = 0; i < recvd; ++i) {
+                double raw_code = 0.0;
+                if (!try_extract_raw_adc_code(raw[i], raw_code)) {
+                    continue;
+                }
+                if (raw_adc_count >= adc_raw_buffer.size()) {
+                    fail("Raw ADC parsing overflowed the temporary buffer");
+                }
+                adc_raw_buffer[raw_adc_count++] = raw_code;
+            }
             if (raw_adc_count != adc_count) {
-                fail("Raw ADC processing returned a different sample count than voltage processing");
+                fail("Raw ADC parsing returned a different sample count than voltage processing");
+            }
+
+            tty_frame_words.clear();
+            tty_frame_words.reserve(((static_cast<std::size_t>(raw_adc_count) + 1U) / 2U) * 4U);
+            for (uint32_t i = 0; i < raw_adc_count; ++i) {
+                const int16_t sample = pack_raw_code_to_int16(adc_raw_buffer[i]);
+                if (!tty_state.pending_ch1_valid) {
+                    tty_state.pending_ch1 = sample;
+                    tty_state.pending_ch1_valid = true;
+                    continue;
+                }
+
+                tty_frame_words.push_back(0x000A);
+                tty_frame_words.push_back(tty_state.next_index);
+                tty_frame_words.push_back(static_cast<uint16_t>(tty_state.pending_ch1));
+                tty_frame_words.push_back(static_cast<uint16_t>(sample));
+                tty_state.pending_ch1_valid = false;
+
+                if (tty_state.next_index >= 0xFFFEU) {
+                    tty_state.next_index = 1U;
+                } else {
+                    ++tty_state.next_index;
+                }
+            }
+
+            if (!tty_frame_words.empty()) {
+                tty_writer->enqueue_encoded_frames(tty_frame_words.data(), tty_frame_words.size() / 4U);
+
+                const auto tty_stats = tty_writer->stats();
+                if (!tty_overflow_warning_printed && (tty_stats.ring_overflows != 0U)) {
+                    std::cerr << "Warning: TTY ring buffer overflowed; dropping oldest frames to keep the stream continuous\n";
+                    tty_overflow_warning_printed = true;
+                }
             }
         }
 
@@ -1558,31 +1619,20 @@ int run(const Config& cfg) {
         for (uint32_t i = 0; i < adc_count; ++i) {
             pending_adc.push_back(adc_buffer[i]);
         }
-        if (tty_writer) {
-            for (uint32_t i = 0; i < raw_adc_count; ++i) {
-                pending_adc_raw.push_back(adc_raw_buffer[i]);
-            }
-        }
         for (uint32_t i = 0; i < din_count; ++i) {
             pending_din.push_back(din_buffer[i]);
         }
 
         if ((pending_adc.size() > 1000000U) ||
-            (pending_din.size() > 1000000U) ||
-            (tty_writer && (pending_adc_raw.size() > 1000000U))) {
+            (pending_din.size() > 1000000U)) {
             fail("Internal backlog grew too large while aligning ADC and DIN samples");
         }
 
         while (!pending_adc.empty() &&
                !pending_din.empty() &&
-               (!tty_writer || !pending_adc_raw.empty()) &&
                !stop_loop_requested) {
             const double adc_value = pending_adc.front();
             pending_adc.pop_front();
-            const double adc_raw_value = tty_writer ? pending_adc_raw.front() : 0.0;
-            if (tty_writer) {
-                pending_adc_raw.pop_front();
-            }
 
             const uint32_t din_value = pending_din.front();
             pending_din.pop_front();
@@ -1624,7 +1674,14 @@ int run(const Config& cfg) {
                 start_packet();
             }
 
-            if (!packet_active && start_edge) {
+            if (packet_active && start_edge) {
+                finish_packet(PacketCloseReason::ExternalStopEdge);
+                if ((cfg.packet_limit != 0U) && (total_completed_packets >= cfg.packet_limit)) {
+                    stop_loop_requested = true;
+                    continue;
+                }
+                start_packet();
+            } else if (!packet_active && start_edge) {
                 start_packet();
             }
 
@@ -1640,11 +1697,6 @@ int run(const Config& cfg) {
                 continue;
             }
 
-            if (tty_writer && di1_changed) {
-                emit_tty_step();
-                tty_step.clear_step();
-            }
-
             const uint32_t lch = next_lch;
             next_lch = (next_lch + 1U) % cfg.channel_count;
 
@@ -1663,9 +1715,6 @@ int run(const Config& cfg) {
 
             if (current_packet.channels[lch].size() < target_frames) {
                 current_packet.channels[lch].push_back(stored_value);
-                if (tty_writer) {
-                    tty_step.add_sample(lch, adc_raw_value);
-                }
                 ++current_packet.stored_samples;
                 ++total_stored_adc_samples;
                 ++stats_stored_adc_samples;
@@ -1710,6 +1759,13 @@ int run(const Config& cfg) {
         print_stats(false);
     }
 
+    TtyProtocolWriter::StatsSnapshot tty_final_stats {};
+    if (tty_writer) {
+        tty_writer->shutdown();
+        tty_writer->throw_if_failed();
+        tty_final_stats = tty_writer->stats();
+    }
+
     const std::vector<CapturePacket> svg_packets(packets.begin(), packets.end());
     writer.write(svg_packets, actual_frame_freq_hz, range_to_volts(cfg.range));
 
@@ -1754,6 +1810,11 @@ int run(const Config& cfg) {
     } else if (cfg.di1_mode == Di1Mode::Trace) {
         std::cout << ", DI1 trace=enabled";
     }
+    if (tty_writer) {
+        std::cout << ", tty_frames_written=" << tty_final_stats.frames_written
+                  << ", tty_frames_dropped=" << tty_final_stats.frames_dropped
+                  << ", tty_ring_overflows=" << tty_final_stats.ring_overflows;
+    }
     std::cout << "\n"
               << "Final SVG packets retained in memory: " << svg_packets.size() << "\n"
               << "CSV: " << cfg.csv_path << "\n"

tty_protocol_writer.cpp

@@ -1,127 +1,410 @@
 #include "tty_protocol_writer.h"
 
+#include <algorithm>
+#include <array>
+#include <chrono>
+#include <cstring>
+#include <exception>
 #include <stdexcept>
+#include <thread>
 #include <utility>
+#include <vector>
 
 #ifdef _WIN32
 
-TtyProtocolWriter::TtyProtocolWriter(std::string path) : path_(std::move(path)) {
+struct TtyProtocolWriter::Impl {};
+
+TtyProtocolWriter::TtyProtocolWriter(std::string path, std::size_t ring_capacity_bytes)
+    : path_(std::move(path)) {
+    (void) ring_capacity_bytes;
     throw std::runtime_error("tty output is supported only on Linux/POSIX");
 }
 
 TtyProtocolWriter::~TtyProtocolWriter() = default;
 
-TtyProtocolWriter::TtyProtocolWriter(TtyProtocolWriter&& other) noexcept = default;
+void TtyProtocolWriter::emit_packet_start() {}
 
-TtyProtocolWriter& TtyProtocolWriter::operator=(TtyProtocolWriter&& other) noexcept = default;
-
-void TtyProtocolWriter::emit_packet_start() const {}
-
-void TtyProtocolWriter::emit_step(uint16_t index, int16_t ch1_avg, int16_t ch2_avg) const {
+void TtyProtocolWriter::emit_step(uint16_t index, int16_t ch1_avg, int16_t ch2_avg) {
     (void) index;
     (void) ch1_avg;
     (void) ch2_avg;
 }
 
+void TtyProtocolWriter::enqueue_encoded_frames(const uint16_t* words, std::size_t frame_count) {
+    (void) words;
+    (void) frame_count;
+}
+
+TtyProtocolWriter::StatsSnapshot TtyProtocolWriter::stats() const {
+    return {};
+}
+
 const std::string& TtyProtocolWriter::path() const {
     return path_;
 }
 
-void TtyProtocolWriter::write_frame(uint16_t word0, uint16_t word1, uint16_t word2, uint16_t word3) const {
+void TtyProtocolWriter::throw_if_failed() const {}
+
+void TtyProtocolWriter::shutdown() {}
+
+void TtyProtocolWriter::enqueue_frame(uint16_t word0, uint16_t word1, uint16_t word2, uint16_t word3) {
     (void) word0;
     (void) word1;
     (void) word2;
     (void) word3;
 }
 
-void TtyProtocolWriter::close_fd() noexcept {}
+void TtyProtocolWriter::worker_loop() {}
 
 #else
 
 #include <cerrno>
-#include <cstring>
+#include <condition_variable>
 #include <fcntl.h>
+#include <limits.h>
+#include <mutex>
+#include <optional>
+#include <pty.h>
 #include <sstream>
+#include <sys/stat.h>
 #include <sys/types.h>
+#include <termios.h>
 #include <unistd.h>
 
 namespace {
 
+constexpr std::size_t kFrameWordCount = 4U;
+constexpr std::size_t kFrameByteCount = kFrameWordCount * sizeof(uint16_t);
+
+using EncodedFrame = std::array<std::uint8_t, kFrameByteCount>;
+
 std::string io_error(const std::string& action, const std::string& path) {
     std::ostringstream out;
     out << action << " '" << path << "': " << std::strerror(errno);
     return out.str();
 }
 
+void close_fd_if_open(int& fd) noexcept {
+    if (fd >= 0) {
+        ::close(fd);
+        fd = -1;
+    }
+}
+
+void set_fd_raw(int fd) {
+    struct termios tio {};
+    if (::tcgetattr(fd, &tio) != 0) {
+        throw std::runtime_error(io_error("Cannot read tty attributes for", std::to_string(fd)));
+    }
+    ::cfmakeraw(&tio);
+    if (::tcsetattr(fd, TCSANOW, &tio) != 0) {
+        throw std::runtime_error(io_error("Cannot apply raw tty attributes to", std::to_string(fd)));
+    }
+}
+
+bool is_character_device_path(const std::string& path) {
+    struct stat st {};
+    if (::stat(path.c_str(), &st) != 0) {
+        if (errno == ENOENT) {
+            return false;
+        }
+        throw std::runtime_error(io_error("Cannot stat tty output", path));
+    }
+    return S_ISCHR(st.st_mode);
+}
+
+std::optional<std::string> read_link_target(const std::string& path) {
+    std::array<char, PATH_MAX> buf {};
+    const ssize_t len = ::readlink(path.c_str(), buf.data(), buf.size() - 1U);
+    if (len < 0) {
+        if (errno == EINVAL || errno == ENOENT) {
+            return std::nullopt;
+        }
+        throw std::runtime_error(io_error("Cannot read symlink", path));
+    }
+    buf[static_cast<std::size_t>(len)] = '\0';
+    return std::string(buf.data());
+}
+
+EncodedFrame encode_frame(uint16_t word0, uint16_t word1, uint16_t word2, uint16_t word3) {
+    const uint16_t words[kFrameWordCount] = {word0, word1, word2, word3};
+    EncodedFrame frame {};
+    std::memcpy(frame.data(), words, sizeof(words));
+    return frame;
+}
+
 } // namespace
 
-TtyProtocolWriter::TtyProtocolWriter(std::string path) : path_(std::move(path)) {
-    fd_ = ::open(path_.c_str(), O_WRONLY | O_NOCTTY);
-    if (fd_ < 0) {
+struct TtyProtocolWriter::Impl {
+    explicit Impl(std::size_t ring_capacity_bytes)
+        : capacity_frames(std::max<std::size_t>(1U, ring_capacity_bytes / kFrameByteCount)),
+          ring(capacity_frames) {}
+
+    int fd = -1;
+    int slave_fd = -1;
+    std::string slave_path;
+    bool owns_link = false;
+
+    const std::size_t capacity_frames;
+    std::vector<EncodedFrame> ring;
+    std::size_t head = 0;
+    std::size_t size = 0;
+
+    mutable std::mutex mutex;
+    std::condition_variable data_ready_cv;
+    std::thread worker;
+    bool stop_requested = false;
+    std::exception_ptr failure;
+    StatsSnapshot stats;
+};
+
+TtyProtocolWriter::TtyProtocolWriter(std::string path, std::size_t ring_capacity_bytes)
+    : path_(std::move(path)),
+      impl_(std::make_unique<Impl>(ring_capacity_bytes)) {
+    if (is_character_device_path(path_)) {
+        impl_->fd = ::open(path_.c_str(), O_WRONLY | O_NOCTTY);
+        if (impl_->fd < 0) {
             throw std::runtime_error(io_error("Cannot open tty output", path_));
         }
+    } else {
+        std::array<char, PATH_MAX> slave_name {};
+        if (::openpty(&impl_->fd, &impl_->slave_fd, slave_name.data(), nullptr, nullptr) != 0) {
+            throw std::runtime_error(io_error("Cannot create PTY bridge for", path_));
+        }
+
+        try {
+            impl_->slave_path = slave_name.data();
+            set_fd_raw(impl_->slave_fd);
+
+            struct stat st {};
+            if (::lstat(path_.c_str(), &st) == 0) {
+                if (!S_ISLNK(st.st_mode) && !S_ISREG(st.st_mode)) {
+                    throw std::runtime_error("Refusing to replace non-link path '" + path_ + "'");
+                }
+                if (::unlink(path_.c_str()) != 0) {
+                    throw std::runtime_error(io_error("Cannot remove existing tty link", path_));
+                }
+            } else if (errno != ENOENT) {
+                throw std::runtime_error(io_error("Cannot inspect tty link path", path_));
+            }
+
+            if (::symlink(impl_->slave_path.c_str(), path_.c_str()) != 0) {
+                throw std::runtime_error(io_error("Cannot create tty symlink", path_));
+            }
+            impl_->owns_link = true;
+        } catch (...) {
+            close_fd_if_open(impl_->slave_fd);
+            close_fd_if_open(impl_->fd);
+            throw;
+        }
+    }
+
+    impl_->worker = std::thread([this]() { worker_loop(); });
 }
 
 TtyProtocolWriter::~TtyProtocolWriter() {
-    close_fd();
+    try {
+        shutdown();
+    } catch (...) {
     }
 
-TtyProtocolWriter::TtyProtocolWriter(TtyProtocolWriter&& other) noexcept
-    : path_(std::move(other.path_)),
-      fd_(other.fd_) {
-    other.fd_ = -1;
+    if (!impl_) {
+        return;
     }
 
-TtyProtocolWriter& TtyProtocolWriter::operator=(TtyProtocolWriter&& other) noexcept {
-    if (this != &other) {
-        close_fd();
-        path_ = std::move(other.path_);
-        fd_ = other.fd_;
-        other.fd_ = -1;
+    if (impl_->owns_link && !path_.empty()) {
+        try {
+            const auto target = read_link_target(path_);
+            if (target && (*target == impl_->slave_path)) {
+                ::unlink(path_.c_str());
             }
-    return *this;
+        } catch (...) {
+        }
+        impl_->owns_link = false;
     }
 
-void TtyProtocolWriter::emit_packet_start() const {
-    write_frame(0x000A, 0xFFFF, 0xFFFF, 0xFFFF);
+    close_fd_if_open(impl_->slave_fd);
+    close_fd_if_open(impl_->fd);
 }
 
-void TtyProtocolWriter::emit_step(uint16_t index, int16_t ch1_avg, int16_t ch2_avg) const {
-    write_frame(0x000A,
+void TtyProtocolWriter::emit_packet_start() {
+    enqueue_frame(0x000A, 0xFFFF, 0xFFFF, 0xFFFF);
+}
+
+void TtyProtocolWriter::emit_step(uint16_t index, int16_t ch1_avg, int16_t ch2_avg) {
+    enqueue_frame(0x000A,
                   index,
                   static_cast<uint16_t>(ch1_avg),
                   static_cast<uint16_t>(ch2_avg));
 }
 
+void TtyProtocolWriter::enqueue_encoded_frames(const uint16_t* words, std::size_t frame_count) {
+    if ((frame_count == 0U) || (words == nullptr)) {
+        return;
+    }
+
+    throw_if_failed();
+
+    std::lock_guard<std::mutex> lock(impl_->mutex);
+    if (impl_->failure) {
+        std::rethrow_exception(impl_->failure);
+    }
+    if (impl_->stop_requested) {
+        throw std::runtime_error("tty writer is already shut down");
+    }
+
+    std::size_t start_frame = 0;
+    std::size_t frames_to_copy = frame_count;
+    std::size_t dropped_frames = 0;
+
+    if (frame_count >= impl_->capacity_frames) {
+        start_frame = frame_count - impl_->capacity_frames;
+        frames_to_copy = impl_->capacity_frames;
+        dropped_frames = impl_->size + start_frame;
+        impl_->head = 0;
+        impl_->size = 0;
+    } else {
+        const std::size_t available_frames = impl_->capacity_frames - impl_->size;
+        if (frame_count > available_frames) {
+            dropped_frames = frame_count - available_frames;
+            impl_->head = (impl_->head + dropped_frames) % impl_->capacity_frames;
+            impl_->size -= dropped_frames;
+        }
+    }
+
+    if (dropped_frames != 0U) {
+        impl_->stats.frames_dropped += static_cast<std::uint64_t>(dropped_frames);
+        ++impl_->stats.ring_overflows;
+    }
+
+    for (std::size_t i = 0; i < frames_to_copy; ++i) {
+        const std::size_t src_index = (start_frame + i) * kFrameWordCount;
+        const std::size_t dst_index = (impl_->head + impl_->size) % impl_->capacity_frames;
+        impl_->ring[dst_index] = encode_frame(words[src_index + 0U],
+                                              words[src_index + 1U],
+                                              words[src_index + 2U],
+                                              words[src_index + 3U]);
+        ++impl_->size;
+    }
+
+    impl_->data_ready_cv.notify_one();
+}
+
+TtyProtocolWriter::StatsSnapshot TtyProtocolWriter::stats() const {
+    if (!impl_) {
+        return {};
+    }
+
+    std::lock_guard<std::mutex> lock(impl_->mutex);
+    return impl_->stats;
+}
+
 const std::string& TtyProtocolWriter::path() const {
     return path_;
 }
 
-void TtyProtocolWriter::write_frame(uint16_t word0, uint16_t word1, uint16_t word2, uint16_t word3) const {
-    const uint16_t frame[4] = {word0, word1, word2, word3};
-    const std::uint8_t* bytes = reinterpret_cast<const std::uint8_t*>(frame);
-    std::size_t remaining = sizeof(frame);
+void TtyProtocolWriter::throw_if_failed() const {
+    if (!impl_) {
+        return;
+    }
 
+    std::exception_ptr failure;
+    {
+        std::lock_guard<std::mutex> lock(impl_->mutex);
+        failure = impl_->failure;
+    }
+
+    if (failure) {
+        std::rethrow_exception(failure);
+    }
+}
+
+void TtyProtocolWriter::shutdown() {
+    if (!impl_) {
+        return;
+    }
+
+    {
+        std::lock_guard<std::mutex> lock(impl_->mutex);
+        impl_->stop_requested = true;
+    }
+
+    close_fd_if_open(impl_->fd);
+    impl_->data_ready_cv.notify_all();
+
+    if (impl_->worker.joinable()) {
+        impl_->worker.join();
+    }
+}
+
+void TtyProtocolWriter::enqueue_frame(uint16_t word0, uint16_t word1, uint16_t word2, uint16_t word3) {
+    const uint16_t words[kFrameWordCount] = {word0, word1, word2, word3};
+    enqueue_encoded_frames(words, 1U);
+}
+
+void TtyProtocolWriter::worker_loop() {
+    for (;;) {
+        EncodedFrame frame {};
+        {
+            std::unique_lock<std::mutex> lock(impl_->mutex);
+            impl_->data_ready_cv.wait(lock, [this]() {
+                return impl_->stop_requested || impl_->failure || (impl_->size != 0U);
+            });
+
+            if (impl_->failure || impl_->stop_requested) {
+                return;
+            }
+
+            frame = impl_->ring[impl_->head];
+            impl_->head = (impl_->head + 1U) % impl_->capacity_frames;
+            --impl_->size;
+        }
+
+        const std::uint8_t* bytes = frame.data();
+        std::size_t remaining = frame.size();
         while (remaining != 0U) {
-        const ssize_t written = ::write(fd_, bytes, remaining);
+            const ssize_t written = ::write(impl_->fd, bytes, remaining);
             if (written < 0) {
                 if (errno == EINTR) {
                     continue;
                 }
-            throw std::runtime_error(io_error("Cannot write tty frame to", path_));
+                if ((errno == EAGAIN) || (errno == EWOULDBLOCK) || (errno == EIO)) {
+                    {
+                        std::lock_guard<std::mutex> lock(impl_->mutex);
+                        if (impl_->stop_requested) {
+                            return;
+                        }
+                    }
+                    std::this_thread::sleep_for(std::chrono::milliseconds(5));
+                    continue;
+                }
+
+                std::lock_guard<std::mutex> lock(impl_->mutex);
+                if (!impl_->stop_requested) {
+                    impl_->failure = std::make_exception_ptr(
+                        std::runtime_error(io_error("Cannot write tty frame to", path_)));
+                }
+                impl_->data_ready_cv.notify_all();
+                return;
             }
             if (written == 0) {
-            throw std::runtime_error("tty write returned 0 bytes for '" + path_ + "'");
+                std::lock_guard<std::mutex> lock(impl_->mutex);
+                if (!impl_->stop_requested) {
+                    impl_->failure = std::make_exception_ptr(
+                        std::runtime_error("tty write returned 0 bytes for '" + path_ + "'"));
                 }
+                impl_->data_ready_cv.notify_all();
+                return;
+            }
+
             bytes += static_cast<std::size_t>(written);
             remaining -= static_cast<std::size_t>(written);
         }
-}
 
-void TtyProtocolWriter::close_fd() noexcept {
-    if (fd_ >= 0) {
-        ::close(fd_);
-        fd_ = -1;
+        {
+            std::lock_guard<std::mutex> lock(impl_->mutex);
+            ++impl_->stats.frames_written;
+        }
     }
 }
 

tty_protocol_writer.h

@@ -1,29 +1,41 @@
 #pragma once
 
+#include <cstddef>
 #include <cstdint>
+#include <memory>
 #include <string>
 
 class TtyProtocolWriter {
 public:
-    explicit TtyProtocolWriter(std::string path);
+    struct StatsSnapshot {
+        std::uint64_t frames_written = 0;
+        std::uint64_t frames_dropped = 0;
+        std::uint64_t ring_overflows = 0;
+    };
+
+    TtyProtocolWriter(std::string path, std::size_t ring_capacity_bytes);
     ~TtyProtocolWriter();
 
     TtyProtocolWriter(const TtyProtocolWriter&) = delete;
     TtyProtocolWriter& operator=(const TtyProtocolWriter&) = delete;
-    TtyProtocolWriter(TtyProtocolWriter&& other) noexcept;
-    TtyProtocolWriter& operator=(TtyProtocolWriter&& other) noexcept;
+    TtyProtocolWriter(TtyProtocolWriter&& other) noexcept = delete;
+    TtyProtocolWriter& operator=(TtyProtocolWriter&& other) noexcept = delete;
 
-    void emit_packet_start() const;
-    void emit_step(uint16_t index, int16_t ch1_avg, int16_t ch2_avg) const;
+    void emit_packet_start();
+    void emit_step(uint16_t index, int16_t ch1_avg, int16_t ch2_avg);
+    void enqueue_encoded_frames(const uint16_t* words, std::size_t frame_count);
+    StatsSnapshot stats() const;
 
     const std::string& path() const;
+    void throw_if_failed() const;
+    void shutdown();
 
 private:
-    void write_frame(uint16_t word0, uint16_t word1, uint16_t word2, uint16_t word3) const;
-    void close_fd() noexcept;
+    void enqueue_frame(uint16_t word0, uint16_t word1, uint16_t word2, uint16_t word3);
+    void worker_loop();
 
     std::string path_;
-#ifndef _WIN32
-    int fd_ = -1;
-#endif
+
+    struct Impl;
+    std::unique_ptr<Impl> impl_;
 };