move streamapi into a thread, saves power and increases responsiveness

src/mesh/MeshService.cpp

@@ -114,7 +114,7 @@ bool MeshService::reloadConfig()
 void MeshService::reloadOwner()
 {
     assert(nodeInfoPlugin);
-    if(nodeInfoPlugin)
+    if (nodeInfoPlugin)
         nodeInfoPlugin->sendOurNodeInfo();
     nodeDB.saveToDisk();
 }
@@ -172,13 +172,12 @@ void MeshService::sendNetworkPing(NodeNum dest, bool wantReplies)
     assert(node);
 
     if (node->has_position) {
-        if(positionPlugin) {
+        if (positionPlugin) {
             DEBUG_MSG("Sending position ping to 0x%x, wantReplies=%d\n", dest, wantReplies);
             positionPlugin->sendOurPosition(dest, wantReplies);
         }
-    }
-    else {
-        if(nodeInfoPlugin) {
+    } else {
+        if (nodeInfoPlugin) {
             DEBUG_MSG("Sending nodeinfo ping to 0x%x, wantReplies=%d\n", dest, wantReplies);
             nodeInfoPlugin->sendOurNodeInfo(dest, wantReplies);
         }

src/mesh/MeshService.h

@@ -84,13 +84,12 @@ class MeshService
     NodeInfo *refreshMyNodeInfo();
 
   private:
-
     /// Called when our gps position has changed - updates nodedb and sends Location message out into the mesh
     /// returns 0 to allow futher processing
     int onGPSChanged(const meshtastic::GPSStatus *arg);
 
-    /// Handle a packet that just arrived from the radio.  This method does _ReliableRouternot_ free the provided packet.  If it needs
-    /// to keep the packet around it makes a copy
+    /// Handle a packet that just arrived from the radio.  This method does _ReliableRouternot_ free the provided packet.  If it
+    /// needs to keep the packet around it makes a copy
     int handleFromRadio(const MeshPacket *p);
     friend class RoutingPlugin;
 };

src/mesh/PhoneAPI.cpp

@@ -1,10 +1,10 @@
 #include "PhoneAPI.h"
+#include "Channels.h"
 #include "GPS.h"
 #include "MeshService.h"
 #include "NodeDB.h"
 #include "PowerFSM.h"
 #include "RadioInterface.h"
-#include "Channels.h"
 #include <assert.h>
 
 #if FromRadio_size > MAX_TO_FROM_RADIO_SIZE
@@ -22,16 +22,18 @@ void PhoneAPI::init()
     observe(&service.fromNumChanged);
 }
 
-PhoneAPI::~PhoneAPI() {
+PhoneAPI::~PhoneAPI()
+{
     close();
 }
 
-void PhoneAPI::close() {
+void PhoneAPI::close()
+{
     unobserve();
     state = STATE_SEND_NOTHING;
     bool oldConnected = isConnected;
     isConnected = false;
-    if(oldConnected != isConnected)
+    if (oldConnected != isConnected)
         onConnectionChanged(isConnected);
 }
 
@@ -49,7 +51,7 @@ void PhoneAPI::checkConnectionTimeout()
 /**
  * Handle a ToRadio protobuf
  */
-void PhoneAPI::handleToRadio(const uint8_t *buf, size_t bufLength)
+bool PhoneAPI::handleToRadio(const uint8_t *buf, size_t bufLength)
 {
     powerFSM.trigger(EVENT_CONTACT_FROM_PHONE); // As long as the phone keeps talking to us, don't let the radio go to sleep
     lastContactMsec = millis();
@@ -62,12 +64,9 @@ void PhoneAPI::handleToRadio(const uint8_t *buf, size_t bufLength)
     memset(&toRadioScratch, 0, sizeof(toRadioScratch));
     if (pb_decode_from_bytes(buf, bufLength, ToRadio_fields, &toRadioScratch)) {
         switch (toRadioScratch.which_payloadVariant) {
-        case ToRadio_packet_tag: {
-            MeshPacket &p = toRadioScratch.packet;
-            printPacket("PACKET FROM PHONE", &p);
-            service.handleToRadio(p);
-            break;
-        }
+        case ToRadio_packet_tag:
+            return handleToRadioPacket(toRadioScratch.packet);
+
         case ToRadio_want_config_id_tag:
             config_nonce = toRadioScratch.want_config_id;
             DEBUG_MSG("Client wants config, nonce=%u\n", config_nonce);
@@ -86,6 +85,8 @@ void PhoneAPI::handleToRadio(const uint8_t *buf, size_t bufLength)
     } else {
         DEBUG_MSG("Error: ignoring malformed toradio\n");
     }
+
+    return false;
 }
 
 /**
@@ -127,7 +128,7 @@ size_t PhoneAPI::getFromRadio(uint8_t *buf)
         fromRadioScratch.my_info = myNodeInfo;
         state = STATE_SEND_NODEINFO;
 
-        service.refreshMyNodeInfo();  // Update my NodeInfo because the client will be asking for it soon.
+        service.refreshMyNodeInfo(); // Update my NodeInfo because the client will be asking for it soon.
         break;
 
     case STATE_SEND_NODEINFO: {
@@ -226,7 +227,13 @@ bool PhoneAPI::available()
 /**
  * Handle a packet that the phone wants us to send.  It is our responsibility to free the packet to the pool
  */
-void PhoneAPI::handleToRadioPacket(MeshPacket *p) {}
+bool PhoneAPI::handleToRadioPacket(MeshPacket &p)
+{
+    printPacket("PACKET FROM PHONE", &p);
+    service.handleToRadio(p);
+
+    return true;
+}
 
 /// If the mesh service tells us fromNum has changed, tell the phone
 int PhoneAPI::onNotify(uint32_t newValue)

src/mesh/PhoneAPI.h

@@ -67,8 +67,9 @@ class PhoneAPI
 
     /**
      * Handle a ToRadio protobuf
+     * @return true true if a packet was queued for sending (so that caller can yield)
      */
-    virtual void handleToRadio(const uint8_t *buf, size_t len);
+    virtual bool handleToRadio(const uint8_t *buf, size_t len);
 
     /**
      * Get the next packet we want to send to the phone
@@ -93,7 +94,7 @@ class PhoneAPI
     /// Hookable to find out when connection changes
     virtual void onConnectionChanged(bool connected) {}
 
-  /// If we haven't heard from the other side in a while then say not connected
+    /// If we haven't heard from the other side in a while then say not connected
     void checkConnectionTimeout();
 
     /**
@@ -103,9 +104,10 @@ class PhoneAPI
 
   private:
     /**
-     * Handle a packet that the phone wants us to send.  It is our responsibility to free the packet to the pool
+     * Handle a packet that the phone wants us to send.  We can write to it but can not keep a reference to it
+     * @return true true if a packet was queued for sending
      */
-    void handleToRadioPacket(MeshPacket *p);
+    bool handleToRadioPacket(MeshPacket &p);
 
     /// If the mesh service tells us fromNum has changed, tell the phone
     virtual int onNotify(uint32_t newValue);

src/mesh/StreamAPI.cpp

@@ -5,48 +5,62 @@
 #define START2 0xc3
 #define HEADER_LEN 4
 
-void StreamAPI::loop()
+int32_t StreamAPI::runOnce()
 {
+    auto result = readStream();
     writeStream();
-    readStream();
     checkConnectionTimeout();
+    return result;
 }
 
 /**
  * Read any rx chars from the link and call handleToRadio
  */
-void StreamAPI::readStream()
+int32_t StreamAPI::readStream()
 {
-    while (stream->available()) { // Currently we never want to block
-        uint8_t c = stream->read();
+    uint32_t now = millis();
+    if (!stream->available()) {
+        // Nothing available this time, if the computer has talked to us recently, poll often, otherwise let CPU sleep a long time
+        bool recentRx = (now - lastRxMsec) < 2000;
+        return recentRx ? 5 : 250;
+    } else {
+        while (stream->available()) { // Currently we never want to block
+            uint8_t c = stream->read();
 
-        // Use the read pointer for a little state machine, first look for framing, then length bytes, then payload
-        size_t ptr = rxPtr++; // assume we will probably advance the rxPtr
+            // Use the read pointer for a little state machine, first look for framing, then length bytes, then payload
+            size_t ptr = rxPtr++; // assume we will probably advance the rxPtr
 
-        rxBuf[ptr] = c; // store all bytes (including framing)
+            rxBuf[ptr] = c; // store all bytes (including framing)
 
-        if (ptr == 0) { // looking for START1
-            if (c != START1)
-                rxPtr = 0;     // failed to find framing
-        } else if (ptr == 1) { // looking for START2
-            if (c != START2)
-                rxPtr = 0;                             // failed to find framing
-        } else if (ptr >= HEADER_LEN) {                // we have at least read our 4 byte framing
-            uint32_t len = (rxBuf[2] << 8) + rxBuf[3]; // big endian 16 bit length follows framing
+            if (ptr == 0) { // looking for START1
+                if (c != START1)
+                    rxPtr = 0;     // failed to find framing
+            } else if (ptr == 1) { // looking for START2
+                if (c != START2)
+                    rxPtr = 0;                             // failed to find framing
+            } else if (ptr >= HEADER_LEN) {                // we have at least read our 4 byte framing
+                uint32_t len = (rxBuf[2] << 8) + rxBuf[3]; // big endian 16 bit length follows framing
 
-            if (ptr == HEADER_LEN) {
-                // we _just_ finished our 4 byte header, validate length now (note: a length of zero is a valid
-                // protobuf also)
-                if (len > MAX_TO_FROM_RADIO_SIZE)
-                    rxPtr = 0; // length is bogus, restart search for framing
-            }
+                if (ptr == HEADER_LEN) {
+                    // we _just_ finished our 4 byte header, validate length now (note: a length of zero is a valid
+                    // protobuf also)
+                    if (len > MAX_TO_FROM_RADIO_SIZE)
+                        rxPtr = 0; // length is bogus, restart search for framing
+                }
 
-            if (rxPtr != 0 && ptr + 1 == len + HEADER_LEN) {
-                // If we didn't just fail the packet and we now have the right # of bytes, parse it
-                handleToRadio(rxBuf + HEADER_LEN, len);
-                rxPtr = 0; // start over again
+                if (rxPtr != 0 && ptr + 1 == len + HEADER_LEN) {
+                    rxPtr = 0; // start over again on the next packet
+
+                    // If we didn't just fail the packet and we now have the right # of bytes, parse it
+                    if (handleToRadio(rxBuf + HEADER_LEN, len))
+                        return 0; // we want to be called again ASAP because we still have more work to do
+                }
             }
         }
+
+        // we had packets available this time, so assume we might have them next time also
+        lastRxMsec = now;
+        return 0;
     }
 }
 
@@ -71,7 +85,7 @@ void StreamAPI::writeStream()
 void StreamAPI::emitTxBuffer(size_t len)
 {
     if (len != 0) {
-        DEBUG_MSG("emit tx %d\n", len);
+        // DEBUG_MSG("emit tx %d\n", len);
         txBuf[0] = START1;
         txBuf[1] = START2;
         txBuf[2] = (len >> 8) & 0xff;
@@ -93,6 +107,6 @@ void StreamAPI::emitRebooted()
     fromRadioScratch.which_payloadVariant = FromRadio_rebooted_tag;
     fromRadioScratch.rebooted = true;
 
-    DEBUG_MSG("Emitting reboot packet for serial shell\n");
+    // DEBUG_MSG("Emitting reboot packet for serial shell\n");
     emitTxBuffer(pb_encode_to_bytes(txBuf + HEADER_LEN, FromRadio_size, FromRadio_fields, &fromRadioScratch));
 }

src/mesh/StreamAPI.h

@@ -2,6 +2,7 @@
 
 #include "PhoneAPI.h"
 #include "Stream.h"
+#include "concurrency/OSThread.h"
 
 // A To/FromRadio packet + our 32 bit header
 #define MAX_STREAM_BUF_SIZE (MAX_TO_FROM_RADIO_SIZE + sizeof(uint32_t))
@@ -27,7 +28,7 @@ valid utf8 encoding. This makes it a bit easier to start a device outputting reg
 after it has received a valid packet from the PC, turn off unencoded debug printing and switch to this packet encoding.
 
  */
-class StreamAPI : public PhoneAPI
+class StreamAPI : public PhoneAPI, protected concurrency::OSThread
 {
     /**
      * The stream we read/write from
@@ -37,21 +38,23 @@ class StreamAPI : public PhoneAPI
     uint8_t rxBuf[MAX_STREAM_BUF_SIZE];
     size_t rxPtr = 0;
 
+    /// time of last rx, used, to slow down our polling if we haven't heard from anyone
+    uint32_t lastRxMsec = 0;
+
   public:
-    StreamAPI(Stream *_stream) : stream(_stream) {}
+    StreamAPI(Stream *_stream) : concurrency::OSThread("StreamAPI"), stream(_stream) {}
 
     /**
      * Currently we require frequent invocation from loop() to check for arrived serial packets and to send new packets to the
      * phone.
-     * FIXME, to support better power behavior instead move to a thread and block on serial reads.
      */
-    void loop();
+    virtual int32_t runOnce();
 
   private:
     /**
      * Read any rx chars from the link and call handleToRadio
      */
-    void readStream();
+    int32_t readStream();
 
     /**
      * call getFromRadio() and deliver encapsulated packets to the Stream
@@ -63,7 +66,7 @@ class StreamAPI : public PhoneAPI
      * Send a FromRadio.rebooted = true packet to the phone
      */
     void emitRebooted();
-    
+
     /**
      * Send the current txBuffer over our stream
      */

src/mesh/wifi/WiFiServerAPI.cpp

@@ -40,18 +40,20 @@ void WiFiServerAPI::onConnectionChanged(bool connected)
 }
 
 /// override close to also shutdown the TCP link
-void WiFiServerAPI::close() {
+void WiFiServerAPI::close()
+{
     client.stop(); // drop tcp connection
     StreamAPI::close();
 }
 
-bool WiFiServerAPI::loop()
+int32_t WiFiServerAPI::runOnce()
 {
     if (client.connected()) {
-        StreamAPI::loop();
-        return true;
+        return StreamAPI::runOnce();
     } else {
-        return false;       
+        DEBUG_MSG("Client dropped connection, suspending API service\n");
+        enabled = false; // we no longer need to run
+        return 0;
     }
 }
 
@@ -78,18 +80,5 @@ int32_t WiFiServerPort::runOnce()
         openAPI = new WiFiServerAPI(client);
     }
 
-    if (openAPI) {
-        // Allow idle processing so the API can read from its incoming stream
-        if(!openAPI->loop()) {
-            // If our API link was up, shut it down
-
-            DEBUG_MSG("Client dropped connection, closing API client\n");
-            // Note: we can't call delete here because this object includes other state
-            // besides the stream API.  Instead kill it later when we start a new instance
-            delete openAPI;
-            openAPI = NULL;
-        }
-        return 0; // run fast while our API server is running
-    } else
-        return 100; // only check occasionally for incoming connections
+    return 100; // only check occasionally for incoming connections
 }

src/mesh/wifi/WiFiServerAPI.h

@@ -1,7 +1,6 @@
 #pragma once
 
 #include "StreamAPI.h"
-#include "concurrency/OSThread.h"
 #include <WiFi.h>
 
 /**
@@ -18,15 +17,14 @@ class WiFiServerAPI : public StreamAPI
 
     virtual ~WiFiServerAPI();
 
-    /// @return true if we want to keep running, or false if we are ready to be destroyed
-    virtual bool loop(); // Check for dropped client connections
-
     /// override close to also shutdown the TCP link
     virtual void close();
 
   protected:
     /// Hookable to find out when connection changes
     virtual void onConnectionChanged(bool connected);
+
+    virtual int32_t runOnce(); // Check for dropped client connections
 };
 
 /**