diff --git a/src/mesh/StreamAPI.cpp b/src/mesh/StreamAPI.cpp index 2c3362877..9785b7f5e 100644 --- a/src/mesh/StreamAPI.cpp +++ b/src/mesh/StreamAPI.cpp @@ -52,7 +52,8 @@ void StreamAPI::writeStream() do { // Send every packet we can len = getFromRadio(txBuf + HEADER_LEN); - emitTxBuffer(len); + if (len != 0 && !emitTxBuffer(len)) + break; } while (len); } } @@ -169,21 +170,36 @@ int32_t StreamAPI::readStream() /** * Send the current txBuffer over our stream */ -void StreamAPI::emitTxBuffer(size_t len) +bool StreamAPI::writeFrame(uint8_t *buf, size_t len) { - if (len != 0) { - txBuf[0] = START1; - txBuf[1] = START2; - txBuf[2] = (len >> 8) & 0xff; - txBuf[3] = len & 0xff; + if (len == 0 || !canWrite) + return false; - auto totalLen = len + HEADER_LEN; - // Serialize stream writes against `emitLogRecord` so a LOG_ firing - // mid-packet-emission can't interleave bytes on the wire. - concurrency::LockGuard guard(&streamLock); - stream->write(txBuf, totalLen); + buf[0] = START1; + buf[1] = START2; + buf[2] = (len >> 8) & 0xff; + buf[3] = len & 0xff; + + auto totalLen = len + HEADER_LEN; + // Serialize write-readiness checks, writes and write-failure handling + // against concurrent stream writes/close. + concurrency::LockGuard guard(&streamLock); + if (!canWriteFrame(totalLen)) + return false; + + size_t written = stream->write(buf, totalLen); + if (written == totalLen) { stream->flush(); + return true; } + + onFrameWriteFailed(totalLen, written); + return false; +} + +bool StreamAPI::emitTxBuffer(size_t len) +{ + return writeFrame(txBuf, len); } void StreamAPI::emitRebooted() @@ -221,20 +237,7 @@ void StreamAPI::emitLogRecord(meshtastic_LogRecord_Level level, const char *src, size_t len = pb_encode_to_bytes(txBufLog + HEADER_LEN, meshtastic_FromRadio_size, &meshtastic_FromRadio_msg, &fromRadioScratchLog); - if (len != 0) { - txBufLog[0] = START1; - txBufLog[1] = START2; - txBufLog[2] = (len >> 8) & 0xff; - txBufLog[3] = len & 0xff; - - auto totalLen = len + HEADER_LEN; - // Serialize stream writes against `emitTxBuffer` so a packet - // emission in flight on another task doesn't interleave bytes - // with this log record. - concurrency::LockGuard guard(&streamLock); - stream->write(txBufLog, totalLen); - stream->flush(); - } + writeFrame(txBufLog, len); } /// Hookable to find out when connection changes @@ -249,4 +252,4 @@ void StreamAPI::onConnectionChanged(bool connected) // received a packet in a while powerFSM.trigger(EVENT_SERIAL_DISCONNECTED); } -} \ No newline at end of file +} diff --git a/src/mesh/StreamAPI.h b/src/mesh/StreamAPI.h index d3ad9ba79..65a2758af 100644 --- a/src/mesh/StreamAPI.h +++ b/src/mesh/StreamAPI.h @@ -80,7 +80,7 @@ class StreamAPI : public PhoneAPI /** * Send the current txBuffer over our stream */ - void emitTxBuffer(size_t len); + bool emitTxBuffer(size_t len); /// Are we allowed to write packets to our output stream (subclasses can turn this off - i.e. SerialConsole) bool canWrite = true; @@ -91,7 +91,12 @@ class StreamAPI : public PhoneAPI /// Low level function to emit a protobuf encapsulated log record void emitLogRecord(meshtastic_LogRecord_Level level, const char *src, const char *format, va_list arg); + virtual bool canWriteFrame(size_t frameLen) { return true; } + virtual void onFrameWriteFailed(size_t frameLen, size_t writtenLen) {} + private: + bool writeFrame(uint8_t *buf, size_t len); + /// Dedicated scratch + tx buffer for LogRecord emission. /// /// The main packet emission path (`writeStream` -> `getFromRadio` -> @@ -113,4 +118,4 @@ class StreamAPI : public PhoneAPI meshtastic_FromRadio fromRadioScratchLog = {}; uint8_t txBufLog[MAX_STREAM_BUF_SIZE] = {0}; concurrency::Lock streamLock; -}; \ No newline at end of file +}; diff --git a/src/mesh/api/ServerAPI.cpp b/src/mesh/api/ServerAPI.cpp index f3e7854ca..af6790cd2 100644 --- a/src/mesh/api/ServerAPI.cpp +++ b/src/mesh/api/ServerAPI.cpp @@ -22,12 +22,38 @@ template void ServerAPI::close() StreamAPI::close(); } -/// Check the current underlying physical link to see if the client is currently connected +/// Check the current underlying physical link to see if the client is currently +/// connected template bool ServerAPI::checkIsConnected() { return client.connected(); } +template bool ServerAPI::canWriteFrame(size_t) +{ + // Only a dropped link is a reason to refuse a write up front. A full transmit + // buffer (availableForWrite() == 0) is normal backpressure, not a dead socket, + // so we must not close the connection on it. A genuinely failed write is + // detected after the fact in onFrameWriteFailed(). + if (!client.connected()) { + canWrite = false; + enabled = false; + LOG_WARN("TCP client disconnected before write, closing API service"); + close(); + return false; + } + + return true; +} + +template void ServerAPI::onFrameWriteFailed(size_t frameLen, size_t writtenLen) +{ + canWrite = false; + enabled = false; + LOG_WARN("TCP client write short (%lu/%lu bytes), closing API service", (unsigned long)writtenLen, (unsigned long)frameLen); + close(); +} + template int32_t ServerAPI::runOnce() { if (client.connected()) { @@ -95,4 +121,4 @@ template int32_t APIServerPort::runOnce() waitTime = 100; #endif return 100; // only check occasionally for incoming connections -} \ No newline at end of file +} diff --git a/src/mesh/api/ServerAPI.h b/src/mesh/api/ServerAPI.h index 2da77c8e9..ece8e0ba2 100644 --- a/src/mesh/api/ServerAPI.h +++ b/src/mesh/api/ServerAPI.h @@ -29,6 +29,8 @@ template class ServerAPI : public StreamAPI, private concurrency::OSTh /// We override this method to prevent publishing EVENT_SERIAL_CONNECTED/DISCONNECTED for wifi links (we want the board to /// stay in the POWERED state to prevent disabling wifi) virtual void onConnectionChanged(bool connected) override {} + virtual bool canWriteFrame(size_t frameLen) override; + virtual void onFrameWriteFailed(size_t frameLen, size_t writtenLen) override; virtual int32_t runOnce() override; // Check for dropped client connections };