coroutine: kinda works now

src/mesh/MeshService.cpp

@@ -49,14 +49,14 @@ MeshService service;
 
 #include "Router.h"
 
-static uint32_t sendOwnerCb()
+static int32_t sendOwnerCb()
 {
     service.sendOurOwner();
 
     return getPref_send_owner_interval() * getPref_position_broadcast_secs() * 1000;
 }
 
-static concurrency::Periodic sendOwnerPeriod("SendOwner", sendOwnerCb);
+static concurrency::Periodic *sendOwnerPeriod;
 
 MeshService::MeshService() : toPhoneQueue(MAX_RX_TOPHONE)
 {
@@ -65,16 +65,18 @@ MeshService::MeshService() : toPhoneQueue(MAX_RX_TOPHONE)
 
 void MeshService::init()
 {
+    sendOwnerPeriod = new concurrency::Periodic("SendOwner", sendOwnerCb);
+
     nodeDB.init();
 
     if (gps)
         gpsObserver.observe(&gps->newStatus);
-    packetReceivedObserver.observe(&router.notifyPacketReceived);
+    packetReceivedObserver.observe(&router->notifyPacketReceived);
 }
 
 void MeshService::sendOurOwner(NodeNum dest, bool wantReplies)
 {
-    MeshPacket *p = router.allocForSending();
+    MeshPacket *p = router->allocForSending();
     p->to = dest;
     p->decoded.want_response = wantReplies;
     p->decoded.which_payload = SubPacket_user_tag;
@@ -121,7 +123,7 @@ const MeshPacket *MeshService::handleFromRadioUser(const MeshPacket *mp)
         sendOurOwner(mp->from);
 
         String lcd = String("Joined: ") + mp->decoded.user.long_name + "\n";
-        screen.print(lcd.c_str());
+        screen->print(lcd.c_str());
     }
 
     return mp;
@@ -257,7 +259,7 @@ void MeshService::sendToMesh(MeshPacket *p)
     }
 
     // Note: We might return !OK if our fifo was full, at that point the only option we have is to drop it
-    router.sendLocal(p);
+    router->sendLocal(p);
 }
 
 void MeshService::sendNetworkPing(NodeNum dest, bool wantReplies)
@@ -279,7 +281,7 @@ void MeshService::sendOurPosition(NodeNum dest, bool wantReplies)
     assert(node->has_position);
 
     // Update our local node info with our position (even if we don't decide to update anyone else)
-    MeshPacket *p = router.allocForSending();
+    MeshPacket *p = router->allocForSending();
     p->to = dest;
     p->decoded.which_payload = SubPacket_position_tag;
     p->decoded.position = node->position;
@@ -293,7 +295,7 @@ int MeshService::onGPSChanged(const meshtastic::GPSStatus *unused)
 {
 
     // Update our local node info with our position (even if we don't decide to update anyone else)
-    MeshPacket *p = router.allocForSending();
+    MeshPacket *p = router->allocForSending();
     p->decoded.which_payload = SubPacket_position_tag;
 
     Position &pos = p->decoded.position;

src/mesh/RadioLibInterface.cpp

@@ -24,12 +24,6 @@ RadioLibInterface::RadioLibInterface(RADIOLIB_PIN_TYPE cs, RADIOLIB_PIN_TYPE irq
     instance = this;
 }
 
-bool RadioLibInterface::init()
-{
-    setup(); // init our timer
-    return RadioInterface::init();
-}
-
 #ifndef NO_ESP32
 // ESP32 doesn't use that flag
 #define YIELD_FROM_ISR(x) portYIELD_FROM_ISR()
@@ -42,7 +36,7 @@ void INTERRUPT_ATTR RadioLibInterface::isrLevel0Common(PendingISR cause)
     instance->disableInterrupt();
 
     BaseType_t xHigherPriorityTaskWoken;
-    instance->notifyFromISR(&xHigherPriorityTaskWoken, cause, eSetValueWithOverwrite);
+    instance->notifyFromISR(&xHigherPriorityTaskWoken, cause, true);
 
     /* Force a context switch if xHigherPriorityTaskWoken is now set to pdTRUE.
     The macro used to do this is dependent on the port and may be called
@@ -206,6 +200,8 @@ void RadioLibInterface::onNotify(uint32_t notification)
         startTransmitTimer();
         break;
     case TRANSMIT_DELAY_COMPLETED:
+        // DEBUG_MSG("delay done\n");
+
         // If we are not currently in receive mode, then restart the timer and try again later (this can happen if the main thread
         // has placed the unit into standby)  FIXME, how will this work if the chipset is in sleep mode?
         if (!txQueue.isEmpty()) {
@@ -232,8 +228,7 @@ void RadioLibInterface::startTransmitTimer(bool withDelay)
     if (!txQueue.isEmpty()) {
         uint32_t delay =
             !withDelay ? 1 : random(MIN_TX_WAIT_MSEC, MAX_TX_WAIT_MSEC); // See documentation for loop() wrt these values
-                                                                         // DEBUG_MSG("xmit timer %d\n", delay);
-        DEBUG_MSG("delaying %u\n", delay);
+        // DEBUG_MSG("xmit timer %d\n", delay);
         notifyLater(delay, TRANSMIT_DELAY_COMPLETED, false); // This will implicitly enable
     }
 }

src/mesh/RadioLibInterface.h

@@ -161,10 +161,6 @@ class RadioLibInterface : public RadioInterface
     virtual void startSend(MeshPacket *txp);
 
   protected:
-    /// Initialise the Driver transport hardware and software.
-    /// Make sure the Driver is properly configured before calling init().
-    /// \return true if initialisation succeeded.
-    virtual bool init();
 
     /** Do any hardware setup needed on entry into send configuration for the radio.  Subclasses can customize */
     virtual void configHardwareForSend() {}

src/mesh/ReliableRouter.cpp

@@ -160,9 +160,10 @@ PendingPacket *ReliableRouter::startRetransmission(MeshPacket *p)
 /**
  * Do any retransmissions that are scheduled (FIXME - for the time being called from loop)
  */
-void ReliableRouter::doRetransmissions()
+int32_t ReliableRouter::doRetransmissions()
 {
     uint32_t now = millis();
+    int32_t d = INT32_MAX;
 
     // FIXME, we should use a better datastructure rather than walking through this map.
     // for(auto el: pending) {
@@ -192,5 +193,13 @@ void ReliableRouter::doRetransmissions()
                 p.setNextTx();
             }
         }
+        else {
+            // Not yet time
+            int32_t t = p.nextTxMsec - now;
+
+            d = min(t, d);
+        }
     }
+
+    return d;
 }

src/mesh/ReliableRouter.h

@@ -79,10 +79,13 @@ class ReliableRouter : public FloodingRouter
     virtual ErrorCode send(MeshPacket *p);
 
     /** Do our retransmission handling */
-    virtual uint32_t runOnce()
+    virtual int32_t runOnce()
     {
-        doRetransmissions();
-        return FloodingRouter::runOnce();
+        auto d = FloodingRouter::runOnce();
+
+        int32_t r = doRetransmissions();
+
+        return min(d, r);
     }
 
   protected:
@@ -123,6 +126,8 @@ class ReliableRouter : public FloodingRouter
 
     /**
      * Do any retransmissions that are scheduled (FIXME - for the time being called from loop)
+     *
+     * @return the number of msecs until our next retransmission or MAXINT if none scheduled
      */
-    void doRetransmissions();
+    int32_t doRetransmissions();
 };

src/mesh/Router.cpp

@@ -41,20 +41,22 @@ Router::Router() : concurrency::OSThread("Router"), fromRadioQueue(MAX_RX_FROMRA
     /* DEBUG_MSG("Size of NodeInfo %d\n", sizeof(NodeInfo));
     DEBUG_MSG("Size of SubPacket %d\n", sizeof(SubPacket));
     DEBUG_MSG("Size of MeshPacket %d\n", sizeof(MeshPacket)); */
+
+    fromRadioQueue.setReader(this);
 }
 
 /**
  * do idle processing
  * Mostly looking in our incoming rxPacket queue and calling handleReceived.
  */
-uint32_t Router::runOnce()
+int32_t Router::runOnce()
 {
     MeshPacket *mp;
     while ((mp = fromRadioQueue.dequeuePtr(0)) != NULL) {
         perhapsHandleReceived(mp);
     }
 
-    return 0;
+    return INT32_MAX; // Wait a long time - until we get woken for the message queue
 }
 
 /// Generate a unique packet id

src/mesh/Router.h

@@ -45,7 +45,7 @@ class Router : protected concurrency::OSThread
      * do idle processing
      * Mostly looking in our incoming rxPacket queue and calling handleReceived.
      */
-    virtual uint32_t runOnce();
+    virtual int32_t runOnce();
 
     /**
      * Works like send, but if we are sending to the local node, we directly put the message in the receive queue
@@ -114,7 +114,7 @@ class Router : protected concurrency::OSThread
     void handleReceived(MeshPacket *p);
 };
 
-extern Router &router;
+extern Router *router;
 
 /// Generate a unique packet id
 // FIXME, move this someplace better

src/mesh/TypedQueue.h

@@ -3,6 +3,7 @@
 #include <cassert>
 #include <type_traits>
 
+#include "concurrency/OSThread.h"
 #include "freertosinc.h"
 
 #ifdef HAS_FREE_RTOS
@@ -15,6 +16,7 @@ template <class T> class TypedQueue
 {
     static_assert(std::is_pod<T>::value, "T must be pod");
     QueueHandle_t h;
+    concurrency::OSThread *reader = NULL;
 
   public:
     TypedQueue(int maxElements)
@@ -29,13 +31,35 @@ template <class T> class TypedQueue
 
     bool isEmpty() { return uxQueueMessagesWaiting(h) == 0; }
 
-    bool enqueue(T x, TickType_t maxWait = portMAX_DELAY) { return xQueueSendToBack(h, &x, maxWait) == pdTRUE; }
+    bool enqueue(T x, TickType_t maxWait = portMAX_DELAY)
+    {
+        if (reader) {
+            reader->setInterval(0);
+            concurrency::mainDelay.interrupt();
+        }
+        return xQueueSendToBack(h, &x, maxWait) == pdTRUE;
+    }
 
-    bool enqueueFromISR(T x, BaseType_t *higherPriWoken) { return xQueueSendToBackFromISR(h, &x, higherPriWoken) == pdTRUE; }
+    bool enqueueFromISR(T x, BaseType_t *higherPriWoken)
+    {
+        if (reader) {
+            reader->setInterval(0);
+            concurrency::mainDelay.interruptFromISR(higherPriWoken);
+        }
+        return xQueueSendToBackFromISR(h, &x, higherPriWoken) == pdTRUE;
+    }
 
     bool dequeue(T *p, TickType_t maxWait = portMAX_DELAY) { return xQueueReceive(h, p, maxWait) == pdTRUE; }
 
     bool dequeueFromISR(T *p, BaseType_t *higherPriWoken) { return xQueueReceiveFromISR(h, p, higherPriWoken); }
+
+    /**
+     * Set a thread that is reading from this queue
+     * If a message is pushed to this queue that thread will be scheduled to run ASAP.
+     *
+     * Note: thread will not be automatically enabled, just have its interval set to 0
+     */
+    void setReader(concurrency::OSThread *t) { reader = t; }
 };
 
 #else