buffer.hpp

Go to the documentation of this file.

#ifndef _PAINLESS_MESH_BUFFER_HPP_
#define _PAINLESS_MESH_BUFFER_HPP_
 
#include <list>
#include <map>
#include <queue>
 
#include "Arduino.h"
#include "painlessmesh/configuration.hpp"
 
#ifndef TCP_MSS
#define TCP_MSS 1024
#endif
 
namespace painlessmesh {
namespace buffer {
 
// Temporary buffer used by ReceiveBuffer and SentBuffer
struct temp_buffer_t {
  size_t length = TCP_MSS;
  char buffer[TCP_MSS];
};
 
template <class T>
class ReceiveBuffer {
 public:
  ReceiveBuffer() { buffer = T(); }
 
  void push(const char *cstr, size_t length, temp_buffer_t &buf) {
    auto data_ptr = cstr;
    do {
      auto len = strnlen(data_ptr, length);
      do {
        // Reserve one byte for the '\0' terminator below: read_len may be at
        // most buf.length - 1, otherwise buf.buffer[read_len] writes one
        // byte past the end of the buffer (caught by the ASan CI job).
        auto read_len = (std::min)(len, buf.length - 1);
        memcpy(buf.buffer, data_ptr, read_len);
        buf.buffer[read_len] = '\0';
        auto newBuffer = T(buf.buffer);
        stringAppend(buffer, newBuffer);
        len -= newBuffer.length();
        length -= newBuffer.length();
        data_ptr += newBuffer.length() * sizeof(char);
      } while (len > 0);
      if (length > 0) {
        // Skip/remove the '\0' between the messages
        length -= 1;
        data_ptr += 1 * sizeof(char);
        if (buffer.length() > 0) {  // skip empty buffers
          jsonStrings.push_back(buffer);
          buffer = T();
        }
      }
    } while (length > 0);
  }
 
  T front() {
    if (!empty()) return jsonStrings.front();
    return T();
  }
 
  void pop_front() { jsonStrings.pop_front(); }
 
  bool empty() { return jsonStrings.empty(); }
 
  void clear() {
    jsonStrings.clear();
    buffer = T();
  }
 
 private:
  T buffer;
  std::list<T> jsonStrings;
 
  inline void stringAppend(T &buffer, T &newBuffer) {
    buffer.concat(newBuffer);
  };
};
 
#ifdef PAINLESSMESH_ENABLE_STD_STRING
template <>
inline void ReceiveBuffer<std::string>::stringAppend(std::string &buffer,
                                                     std::string &newBuffer) {
  buffer.append(newBuffer);
}
#endif
 
// Maximum number of messages a single connection's SentBuffer may hold
// (issue #388). Without a cap, a peer that stops draining (stalled TCP
// connection) grows the outbound list until allocation fails on the ESP8266
// heap. When the buffer is full, an incoming message evicts the newest
// message of a strictly lower priority class; if nothing lower-priority is
// queued, the incoming message itself is rejected (push returns false).
#ifndef PAINLESSMESH_MAX_SENT_BUFFER_MESSAGES
#define PAINLESSMESH_MAX_SENT_BUFFER_MESSAGES 64
#endif
 
template <class T>
struct PrioritizedMessage {
  T message;
  uint8_t priority = 2;  // 0=CRITICAL, 1=HIGH, 2=NORMAL, 3=LOW
  
  PrioritizedMessage(const T& msg, uint8_t prio = 2) : message(msg), priority(prio) {}
};
 
template <class T>
class SentBuffer {
 public:
  SentBuffer() {
    current_read_iterator = prioritizedMessages.end();
  };
 
  bool push(const T &message, bool priority = false) {
    // Legacy API: map bool to uint8_t priority (false=2 NORMAL, true=1 HIGH)
    uint8_t priorityLevel = priority ? 1 : 2;
    return pushWithPriority(message, priorityLevel);
  }
 
  bool pushWithPriority(const T &message, uint8_t priorityLevel) {
    // Clamp priority to valid range
    if (priorityLevel > 3) priorityLevel = 3;
 
    if (prioritizedMessages.size() >= PAINLESSMESH_MAX_SENT_BUFFER_MESSAGES &&
        !evictLowerPriority(priorityLevel)) {
      messagesDropped++;
      return false;
    }
 
    prioritizedMessages.push_back(PrioritizedMessage<T>(message, priorityLevel));
 
    // Track statistics
    totalMessagesQueued++;
    switch(priorityLevel) {
      case 0: criticalQueued++; break;
      case 1: highQueued++; break;
      case 2: normalQueued++; break;
      case 3: lowQueued++; break;
    }
    return true;
  }
 
  size_t requestLength(size_t buffer_length) {
    // Use highest priority message available
    auto* msg = getNextMessage();
    if (!msg) return 0;
    // A caller with no room is owed nothing. The subtraction below leaves
    // space for the terminator toCharArray() always writes; on an unsigned
    // zero it wraps to SIZE_MAX instead, min() then picks the message, and
    // the answer comes back *larger* than was asked for -- which the
    // documented contract above forbids and read() then acts on, writing
    // length + 1 bytes into a buffer the caller said had none.
    if (buffer_length == 0) return 0;
    // String.toCharArray automatically turns the last character into
    // a \0, we need the extra space to deal with that annoyance
    return (std::min)(buffer_length - 1, msg->length() + 1);
  }
 
  void read(size_t length, temp_buffer_t &buf) {
    // Note that toCharrArray always null terminates
    // independent of whether the whole string was read so we use one extra
    // space
    auto* msg = getNextMessage();
    if (msg) {
      msg->toCharArray(buf.buffer, length + 1);
      last_read_size = length;
      last_read_priority = getCurrentMessagePriority();
    }
  }
 
  const char *readPtr(size_t length) {
    auto* msg = getNextMessage();
    if (msg) {
      last_read_size = length;
      last_read_priority = getCurrentMessagePriority();
      return msg->c_str();
    }
    return nullptr;
  }
 
  void freeRead() {
    if (prioritizedMessages.empty()) {
      last_read_size = 0;
      current_read_iterator = prioritizedMessages.end();
      return;
    }
    
    // Find and remove the message that was just read (highest priority)
    auto it = findHighestPriorityMessage();
    if (it != prioritizedMessages.end()) {
      auto& msg = it->message;
      
      if (last_read_size == msg.length() + 1) {
        // Whole message was read, remove it
        // Track statistics
        switch(it->priority) {
          case 0: criticalSent++; break;
          case 1: highSent++; break;
          case 2: normalSent++; break;
          case 3: lowSent++; break;
        }
        prioritizedMessages.erase(it);
        current_read_iterator = prioritizedMessages.end();  // Reset iterator
        clean = true;
      } else {
        // Partial message read, remove the read portion
        stringEraseFront(msg, last_read_size);
        // Keep current_read_iterator pointing to this message for next read
        clean = false;
      }
    }
    last_read_size = 0;
  }
 
  bool empty() { return prioritizedMessages.empty(); }
 
  void clear() {
    prioritizedMessages.clear();
    current_read_iterator = prioritizedMessages.end();
    messagesDropped = 0;
    totalMessagesQueued = 0;
    criticalQueued = highQueued = normalQueued = lowQueued = 0;
    criticalSent = highSent = normalSent = lowSent = 0;
  }
 
  size_t size() { return prioritizedMessages.size(); }
  
  uint8_t getLastReadPriority() const { return last_read_priority; }
  
  struct SendStats {
    uint32_t totalQueued = 0;
    uint32_t criticalQueued = 0;
    uint32_t highQueued = 0;
    uint32_t normalQueued = 0;
    uint32_t lowQueued = 0;
    uint32_t criticalSent = 0;
    uint32_t highSent = 0;
    uint32_t normalSent = 0;
    uint32_t lowSent = 0;
    // Messages lost to the PAINLESSMESH_MAX_SENT_BUFFER_MESSAGES cap:
    // evicted from the buffer or rejected on push (issue #388)
    uint32_t dropped = 0;
  };
 
  SendStats getStats() const {
    SendStats stats;
    stats.dropped = messagesDropped;
    stats.totalQueued = totalMessagesQueued;
    stats.criticalQueued = criticalQueued;
    stats.highQueued = highQueued;
    stats.normalQueued = normalQueued;
    stats.lowQueued = lowQueued;
    stats.criticalSent = criticalSent;
    stats.highSent = highSent;
    stats.normalSent = normalSent;
    stats.lowSent = lowSent;
    return stats;
  }
 
 private:
  size_t last_read_size = 0;
  uint8_t last_read_priority = 2;  // NORMAL default
  bool clean = true;
  std::list<PrioritizedMessage<T>> prioritizedMessages;
  typename std::list<PrioritizedMessage<T>>::iterator current_read_iterator;
  
  // Statistics tracking
  uint32_t messagesDropped = 0;
  uint32_t totalMessagesQueued = 0;
  uint32_t criticalQueued = 0;
  uint32_t highQueued = 0;
  uint32_t normalQueued = 0;
  uint32_t lowQueued = 0;
  uint32_t criticalSent = 0;
  uint32_t highSent = 0;
  uint32_t normalSent = 0;
  uint32_t lowSent = 0;
  
  typename std::list<PrioritizedMessage<T>>::iterator findHighestPriorityMessage() {
    if (prioritizedMessages.empty()) {
      return prioritizedMessages.end();
    }
    
    // If we're in the middle of a partial read (!clean), continue with current_read_iterator
    // This maintains backward compatibility with the original buffer behavior
    if (!clean && current_read_iterator != prioritizedMessages.end()) {
      return current_read_iterator;
    }
    
    // Find message with lowest priority value (0=CRITICAL is highest)
    auto highest = prioritizedMessages.begin();
    for (auto it = prioritizedMessages.begin(); it != prioritizedMessages.end(); ++it) {
      if (it->priority < highest->priority) {
        highest = it;
      }
    }
    
    // Cache this for partial reads
    current_read_iterator = highest;
    return highest;
  }
  
  T* getNextMessage() {
    auto it = findHighestPriorityMessage();
    if (it != prioritizedMessages.end()) {
      return &(it->message);
    }
    return nullptr;
  }
  
  uint8_t getCurrentMessagePriority() {
    auto it = findHighestPriorityMessage();
    if (it != prioritizedMessages.end()) {
      return it->priority;
    }
    return 2;  // NORMAL default
  }
 
  bool evictLowerPriority(uint8_t newPriority) {
    auto victim = prioritizedMessages.end();
    for (auto it = prioritizedMessages.begin();
         it != prioritizedMessages.end(); ++it) {
      if (!clean && it == current_read_iterator) continue;
      if (it->priority <= newPriority) continue;
      // >= keeps walking forward within the lowest class, so the newest
      // (last-queued) entry of that class is the one dropped
      if (victim == prioritizedMessages.end() ||
          it->priority >= victim->priority)
        victim = it;
    }
    if (victim == prioritizedMessages.end()) return false;
    messagesDropped++;
    prioritizedMessages.erase(victim);
    // The cached read iterator may point at the erased node; recompute on
    // the next read. A partial read in progress (!clean) is never evicted.
    if (clean) current_read_iterator = prioritizedMessages.end();
    return true;
  }
 
  inline void stringEraseFront(T &string, size_t length) {
    string.remove(0, length);
  };
};
 
#ifdef PAINLESSMESH_ENABLE_STD_STRING
template <>
inline void SentBuffer<std::string>::read(size_t length, temp_buffer_t &buf) {
  auto* msg = getNextMessage();
  if (msg) {
    msg->copy(buf.buffer, length);
    // Mimic String.toCharArray behaviour, which will insert
    // null termination at the end of original string and the last
    // character
    if (length == msg->length() + 1) buf.buffer[length - 1] = '\0';
    buf.buffer[length] = '\0';
    last_read_size = length;
    last_read_priority = getCurrentMessagePriority();
  }
}
 
template <>
inline void SentBuffer<std::string>::stringEraseFront(std::string &string,
                                                      size_t length) {
  string.erase(0, length);
};
#endif
 
}  // namespace buffer
}  // namespace painlessmesh
#endif