ecos: implemented command trottling, to prevent receive buffer overflow in ECoS

server/src/hardware/protocol/ecos/iohandler/iohandler.cpp

@@ -3,7 +3,7 @@
  *
  * This file is part of the traintastic source code.
  *
- * Copyright (C) 2021 Reinder Feenstra
+ * Copyright (C) 2021-2022 Reinder Feenstra
  *
  * This program is free software; you can redistribute it and/or
  * modify it under the terms of the GNU General Public License
@@ -71,9 +71,9 @@ void IOHandler::processRead(size_t bytesTransferred)
           if(pos != std::string_view::npos)
           {
             size_t end = buffer.find('>', pos);
-            if(end != buffer.size())
+            if(end != std::string_view::npos)
             {
-              m_kernel.receive(std::string_view{m_readBuffer.data() + m_readPos, end - m_readPos + 1});
+              receive(std::string_view{m_readBuffer.data() + m_readPos, end - m_readPos + 1});
               m_readPos = end + 1;
             }
             else
@@ -100,4 +100,9 @@ void IOHandler::processRead(size_t bytesTransferred)
   }
 }
 
+void IOHandler::receive(std::string_view message)
+{
+  m_kernel.receive(message);
+}
+
 }

server/src/hardware/protocol/ecos/iohandler/iohandler.hpp

@@ -3,7 +3,7 @@
  *
  * This file is part of the traintastic source code.
  *
- * Copyright (C) 2021 Reinder Feenstra
+ * Copyright (C) 2021-2022 Reinder Feenstra
  *
  * This program is free software; you can redistribute it and/or
  * modify it under the terms of the GNU General Public License
@@ -38,12 +38,13 @@ class IOHandler
     std::array<char, 32 * 1024> m_readBuffer;
     size_t m_readBufferOffset;
     size_t m_readPos;
-    std::array<char, 1024> m_writeBuffer;
+    std::array<char, 32 * 1024> m_writeBuffer;
     size_t m_writeBufferOffset;
 
     IOHandler(Kernel& kernel);
 
     void processRead(size_t bytesTransferred);
+    virtual void receive(std::string_view message);
     virtual void write() = 0;
 
   public:

server/src/hardware/protocol/ecos/iohandler/tcpiohandler.cpp

@@ -3,7 +3,7 @@
  *
  * This file is part of the traintastic source code.
  *
- * Copyright (C) 2021 Reinder Feenstra
+ * Copyright (C) 2021-2022 Reinder Feenstra
  *
  * This program is free software; you can redistribute it and/or
  * modify it under the terms of the GNU General Public License
@@ -21,6 +21,7 @@
  */
 
 #include "tcpiohandler.hpp"
+#include "../messages.hpp"
 #include "../kernel.hpp"
 #include "../../../../core/eventloop.hpp"
 #include "../../../../log/log.hpp"
@@ -86,18 +87,48 @@ void TCPIOHandler::read()
     });
 }
 
+void TCPIOHandler::receive(std::string_view message)
+{
+  IOHandler::receive(message);
+  if(m_waitingForReply > 0 && isReply(message))
+  {
+    m_waitingForReply--;
+    if(!m_writing && m_waitingForReply < transferWindow && m_writeBufferOffset > 0)
+      write();
+  }
+}
+
 void TCPIOHandler::write()
 {
-  m_socket.async_write_some(boost::asio::buffer(m_writeBuffer.data(), m_writeBufferOffset),
+  assert(!m_writing);
+
+  m_writing = true;
+
+  const char* p = m_writeBuffer.data();
+  const char* end = m_writeBuffer.data() + m_writeBufferOffset;
+
+  while(m_waitingForReply < transferWindow && (p = std::find(p, end, '\n')) != end)
+  {
+    p++;
+    m_waitingForReply++;
+  }
+
+  if(p == m_writeBuffer.data())
+    return;
+
+  m_socket.async_write_some(boost::asio::buffer(m_writeBuffer.data(), p - m_writeBuffer.data()),
     [this](const boost::system::error_code& ec, std::size_t bytesTransferred)
     {
+      m_writing = false;
+
       if(!ec)
       {
         if(bytesTransferred < m_writeBufferOffset)
         {
           m_writeBufferOffset -= bytesTransferred;
           memmove(m_writeBuffer.data(), m_writeBuffer.data() + bytesTransferred, m_writeBufferOffset);
-          write();
+          if(m_waitingForReply < transferWindow)
+            write();
         }
         else
           m_writeBufferOffset = 0;

server/src/hardware/protocol/ecos/iohandler/tcpiohandler.hpp

@@ -3,7 +3,7 @@
  *
  * This file is part of the traintastic source code.
  *
- * Copyright (C) 2021 Reinder Feenstra
+ * Copyright (C) 2021-2022 Reinder Feenstra
  *
  * This program is free software; you can redistribute it and/or
  * modify it under the terms of the GNU General Public License
@@ -31,12 +31,16 @@ namespace ECoS {
 class TCPIOHandler final : public IOHandler
 {
   private:
+    static constexpr uint32_t transferWindow = 25;
     boost::asio::ip::tcp::socket m_socket;
     boost::asio::ip::tcp::endpoint m_endpoint;
+    bool m_writing = false;
+    uint32_t m_waitingForReply = 0;
 
     void read();
 
   protected:
+    void receive(std::string_view message) final;
     void write() final;
 
   public:

server/src/hardware/protocol/ecos/messages.cpp

@@ -28,16 +28,20 @@
 
 namespace ECoS {
 
+static const std::string_view startDelimiterReply = "<REPLY ";
 static const std::string_view endDelimiter = "<END ";
 
+bool isReply(std::string_view message)
+{
+  return startsWith(message, startDelimiterReply);
+}
+
 bool parseReply(std::string_view message, Reply& reply)
 {
-  static const std::string_view startDelimiter = "<REPLY ";
-
-  if(!startsWith(message, startDelimiter))
+  if(!isReply(message))
     return false;
 
-  size_t n = startDelimiter.size();
+  size_t n = startDelimiterReply.size();
   size_t pos;
   if((pos = message.find('(', n)) == std::string_view::npos)
     return false;

server/src/hardware/protocol/ecos/messages.hpp

@@ -148,6 +148,7 @@ inline std::string release(uint16_t objectId, std::initializer_list<std::string_
   return buildCommand(Command::release, objectId, options);
 }
 
+bool isReply(std::string_view message);
 bool parseReply(std::string_view message, Reply& reply);
 bool parseEvent(std::string_view message, Event& event);