TeslaRel410/emulator/vpx-device/serialnamedpipe.cpp

/*
 *  VWE fork: serial-over-named-pipe backend. See serialnamedpipe.h for the
 *  wire contract. RX pacing is a faithful copy of the (patched) directserial
 *  state machine so the RIO's validated rxpollus/rxburst tuning carries over.
 */

#include "dosbox.h"

#if defined(WIN32)

#include "logging.h"
#include "pic.h"
#include "setup.h"
#include "serialport.h"
#include "serialnamedpipe.h"

bool getBituSubstring(const char* name, Bitu* data, CommandLine* cmd);

#define PIPE_RETRY_MS 500.0

enum { P_RX_IDLE = 0, P_RX_WAIT, P_RX_BLOCKED, P_RX_FASTWAIT };

// frame types (identical both directions)
enum { FRAME_DATA = 0x00, FRAME_LINES = 0x01 };

CSerialNamedPipe::CSerialNamedPipe(Bitu id, CommandLine* cmd) : CSerial(id, cmd) {
	InstallationSuccessful = false;

	std::string tmp;
	if (!cmd->FindStringBegin("pipe:", tmp, false)) {
		LOG_MSG("Serial%d: namedpipe requires pipe:<name>", (int)COMNUMBER);
		return;
	}
	// pipe:vrio -> \\.\pipe\vrio; a full \\... path passes through
	if (tmp.size() >= 2 && tmp[0] == '\\' && tmp[1] == '\\')
		pipename = tmp;
	else
		pipename = std::string("\\\\.\\pipe\\") + tmp;

	Bitu rx_poll_us = 0;
	if (getBituSubstring("rxpollus:", &rx_poll_us, cmd)) {
		if (rx_poll_us < 50) rx_poll_us = 50;
		if (rx_poll_us > 1000) rx_poll_us = 1000;
		rx_poll_ms = (float)rx_poll_us / 1000.0f;
	}
	Bitu rx_burst = 0;
	if (getBituSubstring("rxburst:", &rx_burst, cmd)) {
		if (rx_burst < 1) rx_burst = 1;
		if (rx_burst > 64) rx_burst = 64;
		rx_burst_div = (float)rx_burst;
	}
	if (getBituSubstring("rxdelay:", &rx_retry_max, cmd)) {
		if (!(rx_retry_max <= 10000)) rx_retry_max = 0;
	}

	LOG_MSG("Serial%d: namedpipe client of %s (poll %.2fms burst x%g)",
	        (int)COMNUMBER, pipename.c_str(), rx_poll_ms, rx_burst_div);

	CSerial::Init_Registers();
	// unplugged cable until the peer's 0x01 arrives
	setRI(false);
	setCD(false);
	setDSR(false);
	setCTS(false);

	InstallationSuccessful = true;
	rx_state = P_RX_IDLE;
	tryConnect();                                  // eager first attempt
	setEvent(SERIAL_POLLING_EVENT, rx_poll_ms);    // receive/reconnect tick
}

CSerialNamedPipe::~CSerialNamedPipe() {
	if (pipe != INVALID_HANDLE_VALUE) {
		CancelIo(pipe);
		CloseHandle(pipe);
	}
	// events are cleared by the framework on port destruction
}

void CSerialNamedPipe::tryConnect() {
	next_retry_ms = PIC_FullIndex() + PIPE_RETRY_MS;
	HANDLE h = CreateFileA(pipename.c_str(), GENERIC_READ | GENERIC_WRITE,
	                       0, NULL, OPEN_EXISTING, FILE_FLAG_OVERLAPPED, NULL);
	if (h == INVALID_HANDLE_VALUE) return;         // no server yet: cable stays unplugged
	DWORD mode = PIPE_READMODE_BYTE;
	SetNamedPipeHandleState(h, &mode, NULL, NULL);
	pipe = h;
	inbuf.clear();
	txq.clear();
	tx_inflight = false;
	LOG_MSG("Serial%d: namedpipe connected to %s", (int)COMNUMBER, pipename.c_str());
	sendLineState();                               // contract: one 0x01 on connect
}

void CSerialNamedPipe::dropConnection(const char* why) {
	if (pipe != INVALID_HANDLE_VALUE) {
		CancelIo(pipe);
		CloseHandle(pipe);
		pipe = INVALID_HANDLE_VALUE;
	}
	tx_inflight = false;
	txq.clear();
	inbuf.clear();                 // partial frame is meaningless now
	// disconnect = all lines low (rxq keeps already-delivered wire bytes)
	setCTS(false);
	setDSR(false);
	setCD(false);
	next_retry_ms = PIC_FullIndex() + PIPE_RETRY_MS;
	LOG_MSG("Serial%d: namedpipe disconnected (%s)", (int)COMNUMBER, why);
}

// TX is queue + single in-flight overlapped write, reaped from the poll
// tick. The emu thread NEVER waits on the pipe: a peer that isn't reading
// (e.g. both sides sending their connect hello before either reads --
// first live vRIO run, 2026-07-13) wedges the queue, not the emulator;
// after TX_WEDGE_MS of no progress the connection drops (unplugged cable).
#define TXQ_MAX      8192
#define TX_WEDGE_MS  2000.0

void CSerialNamedPipe::queueSend(const uint8_t* data, DWORD len) {
	if (pipe == INVALID_HANDLE_VALUE) return;      // unplugged: discard
	if (txq.size() + len > TXQ_MAX) {
		dropConnection("tx queue overflow (peer not reading)");
		return;
	}
	txq.insert(txq.end(), data, data + len);
	kickTx();
}

void CSerialNamedPipe::kickTx() {
	if (pipe == INVALID_HANDLE_VALUE || tx_inflight || txq.empty()) return;
	tx_pend.swap(txq);
	txq.clear();
	ZeroMemory(&tx_ov, sizeof(tx_ov));
	DWORD done = 0;
	if (WriteFile(pipe, tx_pend.data(), (DWORD)tx_pend.size(), &done, &tx_ov)) {
		return;                                    // completed synchronously
	}
	if (GetLastError() == ERROR_IO_PENDING) {
		tx_inflight = true;
		tx_wedge_ms = PIC_FullIndex() + TX_WEDGE_MS;
		return;
	}
	dropConnection("write failed");
}

void CSerialNamedPipe::pollTx() {
	if (pipe == INVALID_HANDLE_VALUE || !tx_inflight) return;
	DWORD done = 0;
	if (GetOverlappedResult(pipe, &tx_ov, &done, FALSE)) {
		tx_inflight = false;
		kickTx();
		return;
	}
	DWORD err = GetLastError();
	if (err == ERROR_IO_INCOMPLETE) {
		if (PIC_FullIndex() >= tx_wedge_ms)
			dropConnection("write wedged (peer not reading)");
		return;
	}
	tx_inflight = false;
	dropConnection("write failed");
}

void CSerialNamedPipe::sendLineState() {
	uint8_t f[2] = { FRAME_LINES,
	                 (uint8_t)((cur_dtr ? 1 : 0) | (cur_rts ? 2 : 0)) };
	queueSend(f, 2);
}

void CSerialNamedPipe::pumpPipe() {
	if (pipe == INVALID_HANDLE_VALUE) return;
	for (;;) {
		DWORD avail = 0;
		if (!PeekNamedPipe(pipe, NULL, 0, NULL, &avail, NULL)) {
			dropConnection("peer closed");
			return;
		}
		if (avail == 0) break;
		uint8_t buf[512];
		DWORD want = avail < sizeof(buf) ? avail : (DWORD)sizeof(buf);
		DWORD got = 0;
		ZeroMemory(&rx_ov, sizeof(rx_ov));
		// overlapped handle needs an OVERLAPPED; Peek guaranteed the bytes
		// are buffered, so the wait below completes immediately.
		if (!ReadFile(pipe, buf, want, &got, &rx_ov)) {
			if (GetLastError() != ERROR_IO_PENDING ||
			    !GetOverlappedResult(pipe, &rx_ov, &got, TRUE)) {
				dropConnection("read failed");
				return;
			}
		}
		if (got == 0) {
			dropConnection("read failed");
			return;
		}
		inbuf.insert(inbuf.end(), buf, buf + got);
	}
	// slice complete frames
	size_t pos = 0;
	while (pos < inbuf.size()) {
		uint8_t type = inbuf[pos];
		if (type == FRAME_DATA) {
			if (pos + 2 > inbuf.size()) break;                  // need len
			uint8_t len = inbuf[pos + 1];
			if (len == 0) {                                     // contract: len >= 1
				dropConnection("zero-length data frame");
				return;
			}
			if (pos + 2 + len > inbuf.size()) break;            // incomplete
			for (uint8_t i = 0; i < len; i++)
				rxq.push_back(inbuf[pos + 2 + i]);
			pos += 2 + (size_t)len;
		} else if (type == FRAME_LINES) {
			if (pos + 2 > inbuf.size()) break;
			uint8_t lines = inbuf[pos + 1];
			// null-modem cross: their DTR -> our DSR+CD, their RTS -> our CTS
			setDSR((lines & 1) != 0);
			setCD((lines & 1) != 0);
			setCTS((lines & 2) != 0);
			pos += 2;
		} else {
			LOG_MSG("Serial%d: namedpipe unknown frame type 0x%02x -- protocol bug",
			        (int)COMNUMBER, type);
			dropConnection("unknown frame type");
			return;
		}
	}
	inbuf.erase(inbuf.begin(), inbuf.begin() + pos);
}

/* VPX page-fault watch (cpu/paging.cpp): records each delivered RX byte with
 * the guest cs:eip at delivery, so a watched fault can say how close it was
 * to a serial delivery and where the guest was interrupted. */
extern void VPX_SerialTraceByte(double now, uint8_t b);

bool CSerialNamedPipe::doReceive() {
	if (rxq.empty()) pumpPipe();
	if (rxq.empty()) return false;
	uint8_t b = rxq.front();
	rxq.pop_front();
	VPX_SerialTraceByte(PIC_FullIndex(), b);
	receiveByteEx(b, 0);
	return true;
}

void CSerialNamedPipe::handleUpperEvent(uint16_t type) {
	switch (type) {
	case SERIAL_POLLING_EVENT: {
		setEvent(SERIAL_POLLING_EVENT, rx_poll_ms);
		if (pipe == INVALID_HANDLE_VALUE) {
			if (PIC_FullIndex() >= next_retry_ms) tryConnect();
		} else {
			pollTx();         // reap / restart the overlapped write
			pumpPipe();       // keep line-state frames fresh even when idle
		}
		// directserial's RX state machine (rxpollus/rxburst semantics)
		switch (rx_state) {
		case P_RX_IDLE:
			if (CanReceiveByte()) {
				if (doReceive()) {
					rx_state = P_RX_WAIT;
					setEvent(SERIAL_RX_EVENT, bytetime * 0.9f / rx_burst_div);
				}
			} else {
				rx_state = P_RX_BLOCKED;
				setEvent(SERIAL_RX_EVENT, bytetime * 0.9f / rx_burst_div);
			}
			break;
		case P_RX_BLOCKED:
			if (!CanReceiveByte()) {
				rx_retry++;
				if (rx_retry >= rx_retry_max) {
					rx_retry = 0;
					removeEvent(SERIAL_RX_EVENT);
					if (doReceive()) {
						//
						// OVERRUN SEMANTICS.  directserial drains its whole
						// backlog here (`while (doReceive());`) because a real
						// port's backlog is bounded by the physical line rate
						// -- there is never much to drain.  A NAMED PIPE has no
						// such bound: vRIO can have hundreds of bytes queued,
						// and this loop teleported all of them into the UART in
						// zero guest-time, each one raising receive/line-status
						// state with no instruction executed in between.
						//
						// That burst is the trigger for the pod's residual
						// page fault (2026-07-29): it fires only when the guest
						// is briefly slow to read -- which is the mission-load
						// window -- and its probability moves with anything
						// that changes guest timing, which is why our own COM3
						// logging SUPPRESSED it (quiet conf 3/3 faults vs the
						// logged conf's 25-40%).
						//
						// TESTED, NOT THE CULPRIT (2026-07-29).  Bounding the
						// drain to one byte -- real UART overrun semantics,
						// where the ARRIVING byte overwrites the unread one
						// rather than pulling the whole line forward -- did NOT
						// stop the fault (2/2 faults with it in place).  The
						// measurement that explains why: overruns DURING a run
						// are only 1-3 per report period, because vRIO sends
						// about one byte every 1-3ms, so the backlog is shallow
						// and this loop never had hundreds to teleport.  The
						// big 494/499 counts in the logs all land AFTER the
						// game exits.
						//
						// So the drain-all default stays -- it is the validated
						// directserial behaviour the RIO's rxpollus/rxburst
						// tuning was calibrated against, and changing it
						// without evidence would risk the real-cockpit timing
						// for nothing.  VPX_RX_OVERRUN_ONE=1 opts into the
						// bounded form if a future experiment wants it.
						//
						static int overrun_one = -1;
						if (overrun_one < 0)
							overrun_one = getenv("VPX_RX_OVERRUN_ONE") ? 1 : 0;
						if (!overrun_one) { while (doReceive()); }
						rx_state = P_RX_WAIT;
						setEvent(SERIAL_RX_EVENT, bytetime * 0.9f / rx_burst_div);
					} else {
						rx_state = P_RX_IDLE;
					}
				}
			} else {
				removeEvent(SERIAL_RX_EVENT);
				rx_retry = 0;
				if (doReceive()) {
					rx_state = P_RX_FASTWAIT;
					setEvent(SERIAL_RX_EVENT, bytetime * 0.65f / rx_burst_div);
				} else {
					rx_state = P_RX_IDLE;
				}
			}
			break;
		case P_RX_WAIT:
		case P_RX_FASTWAIT:
			break;
		}
		break;
	}
	case SERIAL_RX_EVENT: {
		switch (rx_state) {
		case P_RX_IDLE:
			LOG_MSG("internal error in serialnamedpipe");
			break;
		case P_RX_BLOCKED:
		case P_RX_WAIT:
		case P_RX_FASTWAIT:
			if (CanReceiveByte()) {
				rx_retry = 0;
				if (doReceive()) {
					if (rx_state == P_RX_WAIT)
						setEvent(SERIAL_RX_EVENT, bytetime * 0.9f / rx_burst_div);
					else {
						rx_state = P_RX_FASTWAIT;
						setEvent(SERIAL_RX_EVENT, bytetime * 0.65f / rx_burst_div);
					}
				} else {
					rx_state = P_RX_IDLE;
				}
			} else {
				setEvent(SERIAL_RX_EVENT, bytetime * 0.65f / rx_burst_div);
				rx_state = P_RX_BLOCKED;
			}
			break;
		}
		break;
	}
	case SERIAL_TX_EVENT: {
		if (rx_state == P_RX_IDLE && CanReceiveByte()) {
			if (doReceive()) {
				rx_state = P_RX_WAIT;
				setEvent(SERIAL_RX_EVENT, bytetime * 0.9f / rx_burst_div);
			}
		}
		ByteTransmitted();
		break;
	}
	case SERIAL_THR_EVENT: {
		ByteTransmitting();
		setEvent(SERIAL_TX_EVENT, bytetime * 1.1f);
		break;
	}
	}
}

void CSerialNamedPipe::updatePortConfig(uint16_t divider, uint8_t lcr) {
	(void)divider; (void)lcr;   // a pipe has no baud rate
}

void CSerialNamedPipe::updateMSR() {
	// modem-in lines are event-driven (0x01 frames); nothing to poll
}

void CSerialNamedPipe::transmitByte(uint8_t val, bool first) {
	if (first) setEvent(SERIAL_THR_EVENT, bytetime / 10);
	else setEvent(SERIAL_TX_EVENT, bytetime);
	uint8_t f[3] = { FRAME_DATA, 1, val };
	queueSend(f, 3);            // unconnected pipe swallows it (unplugged cable)
}

void CSerialNamedPipe::setBreak(bool value) {
	(void)value;                // not in the contract; RIO/plasma don't use break
}

void CSerialNamedPipe::setRTSDTR(bool rts, bool dtr) {
	if (rts == cur_rts && dtr == cur_dtr) return;
	cur_rts = rts;
	cur_dtr = dtr;
	sendLineState();
}

void CSerialNamedPipe::setRTS(bool val) {
	if (val == cur_rts) return;
	cur_rts = val;
	sendLineState();
}

void CSerialNamedPipe::setDTR(bool val) {
	if (val == cur_dtr) return;
	cur_dtr = val;
	sendLineState();
}

#endif // WIN32