Implementation of threadsafe reading from the serial.

By duplicating received data in a separate read buffer per thread.

Author: aentinger

examples/ts_serial_reader/ts_serial_reader.ino

@@ -0,0 +1,79 @@
+/**************************************************************************************
+ * INCLUDE
+ **************************************************************************************/
+
+#include <Arduino_ThreadsafeIO.h>
+
+/**************************************************************************************
+ * CONSTANTS
+ **************************************************************************************/
+
+static size_t constexpr NUM_THREADS = 5;
+
+/**************************************************************************************
+ * FUNCTION DECLARATION
+ **************************************************************************************/
+
+void serial_thread_func();
+
+/**************************************************************************************
+ * GLOBAL VARIABLES
+ **************************************************************************************/
+
+static char thread_name[NUM_THREADS][32];
+#undef Serial
+#ifdef ARDUINO_PORTENTA_H7_M4
+  SerialDispatcher Serial(Serial1); /* No SerialUSB for Portenta H7 / M4 Core */
+#else
+  SerialDispatcher Serial(SerialUSB);
+#endif
+
+/**************************************************************************************
+ * SETUP/LOOP
+ **************************************************************************************/
+
+void setup()
+{
+  /* Fire up some threads all accessing the LSM6DSOX */
+  for(size_t i = 0; i < NUM_THREADS; i++)
+  {
+    snprintf(thread_name[i], sizeof(thread_name[i]), "Thread #%02d", i);
+    rtos::Thread * t = new rtos::Thread(osPriorityNormal, OS_STACK_SIZE, nullptr, thread_name[i]);
+    t->start(serial_thread_func);
+  }
+}
+
+void loop()
+{
+
+}
+
+/**************************************************************************************
+ * FUNCTION DEFINITION
+ **************************************************************************************/
+
+void serial_thread_func()
+{
+  Serial.begin(9600);
+
+  for(;;)
+  {
+    /* Sleep between 5 and 500 ms */
+    rtos::ThisThread::sleep_for(rtos::Kernel::Clock::duration_u32(random(5,500)));
+
+    /* Read data from the serial interface into a String. */
+    String serial_msg;
+    while (Serial.available())
+      serial_msg += (char)Serial.read();
+
+    /* Print thread id and chip id value to serial. */
+    if (serial_msg.length())
+    {
+      char msg[64] = {0};
+      snprintf(msg, sizeof(msg), "[%05lu] %s: %s ...", millis(), rtos::ThisThread::get_name(), serial_msg.c_str());
+      Serial.block();
+      Serial.println(msg);
+      Serial.unblock();
+    }
+  }
+}

src/serial/SerialDispatcher.cpp

@@ -99,19 +99,37 @@ void SerialDispatcher::end()
 int SerialDispatcher::available()
 {
   mbed::ScopedLock<rtos::Mutex> lock(_mutex);
-  return _serial.available();
+  auto iter = findThreadCustomerDataById(rtos::ThisThread::get_id());
+  if (iter == std::end(_thread_customer_list)) return 0;
+
+  prepareSerialReader(iter);
+  handleSerialReader();
+
+  return iter->rx_buffer->available();
 }
 
 int SerialDispatcher::peek()
 {
   mbed::ScopedLock<rtos::Mutex> lock(_mutex);
-  return _serial.peek();
+  auto iter = findThreadCustomerDataById(rtos::ThisThread::get_id());
+  if (iter == std::end(_thread_customer_list)) return 0;
+
+  prepareSerialReader(iter);
+  handleSerialReader();
+
+  return iter->rx_buffer->peek();
 }
 
 int SerialDispatcher::read()
 {
   mbed::ScopedLock<rtos::Mutex> lock(_mutex);
-  return _serial.read();
+  auto iter = findThreadCustomerDataById(rtos::ThisThread::get_id());
+  if (iter == std::end(_thread_customer_list)) return 0;
+
+  prepareSerialReader(iter);
+  handleSerialReader();
+
+  return iter->rx_buffer->read_char();
 }
 
 void SerialDispatcher::flush()
@@ -203,4 +221,32 @@ std::list<SerialDispatcher::ThreadCustomerData>::iterator SerialDispatcher::find
   return std::find_if(std::begin(_thread_customer_list), 
                       std::end  (_thread_customer_list),
                       [thread_id](ThreadCustomerData const d) -> bool { return (d.thread_id == thread_id); });
-}
+}
+
+void SerialDispatcher::prepareSerialReader(std::list<ThreadCustomerData>::iterator & iter)
+{
+  iter->is_reader = true;
+  if (!iter->rx_buffer)
+    iter->rx_buffer.reset(new arduino::RingBuffer());
+}
+
+void SerialDispatcher::handleSerialReader()
+{
+  while (_serial.available())
+  {
+    int const c = _serial.read();
+
+    std::for_each(std::begin(_thread_customer_list),
+                  std::end  (_thread_customer_list),
+                  [c](ThreadCustomerData & d)
+                  {
+                    if (!d.is_reader)
+                      return;
+
+                    if (!d.rx_buffer->availableForStore())
+                      return;
+
+                    d.rx_buffer->store_char(c);
+                  });
+  }
+}

src/serial/SerialDispatcher.h

@@ -29,6 +29,8 @@
 
 #include <list>
 
+#include <SharedPtr.h>
+
 /**************************************************************************************
  * CLASS DECLARATION
  **************************************************************************************/
@@ -72,13 +74,16 @@ class SerialDispatcher : public arduino::HardwareSerial
     osThreadId_t thread_id;
     arduino::RingBuffer tx_buffer;
     bool block_tx_buffer;
+    bool is_reader;                                 /* This thread has expressed interest to read from Serial via a call to either read() or available(). */
+    mbed::SharedPtr<arduino::RingBuffer> rx_buffer; /* Only when a thread has expressed interested to read from serial a receive ringbuffer is allocated. */
   } ThreadCustomerData;
 
   std::list<ThreadCustomerData> _thread_customer_list;
 
   void threadFunc();
   std::list<ThreadCustomerData>::iterator findThreadCustomerDataById(osThreadId_t const thread_id);
-
+  void prepareSerialReader(std::list<ThreadCustomerData>::iterator & iter);
+  void handleSerialReader();
 };
 
 #endif /* SERIAL_DISPATCHER_H_ */