ESPHome 2025.9.0-dev
Loading...
Searching...
No Matches
lock_free_queue.h
Go to the documentation of this file.
1#pragma once
2
3#if defined(USE_ESP32)
4
5#include <atomic>
6#include <cstddef>
7
8#include <freertos/FreeRTOS.h>
9#include <freertos/task.h>
10
11/*
12 * Lock-free queue for single-producer single-consumer scenarios.
13 * This allows one thread to push items and another to pop them without
14 * blocking each other.
15 *
16 * This is a Single-Producer Single-Consumer (SPSC) lock-free ring buffer.
17 * Available on platforms with FreeRTOS support (ESP32, LibreTiny).
18 *
19 * Common use cases:
20 * - BLE events: BLE task produces, main loop consumes
21 * - MQTT messages: main task produces, MQTT thread consumes
22 *
23 * @tparam T The type of elements stored in the queue (must be a pointer type)
24 * @tparam SIZE The maximum number of elements (1-255, limited by uint8_t indices)
25 */
26
27namespace esphome {
28
29// Base lock-free queue without task notification
30template<class T, uint8_t SIZE> class LockFreeQueue {
31 public:
33
34 bool push(T *element) {
35 bool was_empty;
36 uint8_t old_tail;
37 return push_internal_(element, was_empty, old_tail);
38 }
39
40 protected:
41 // Internal push that reports queue state - for use by derived classes
42 bool push_internal_(T *element, bool &was_empty, uint8_t &old_tail) {
43 if (element == nullptr)
44 return false;
45
46 uint8_t current_tail = tail_.load(std::memory_order_relaxed);
47 uint8_t next_tail = (current_tail + 1) % SIZE;
48
49 // Read head before incrementing tail
50 uint8_t head_before = head_.load(std::memory_order_acquire);
51
52 if (next_tail == head_before) {
53 // Buffer full
54 dropped_count_.fetch_add(1, std::memory_order_relaxed);
55 return false;
56 }
57
58 was_empty = (current_tail == head_before);
59 old_tail = current_tail;
60
61 buffer_[current_tail] = element;
62 tail_.store(next_tail, std::memory_order_release);
63
64 return true;
65 }
66
67 public:
68 T *pop() {
69 uint8_t current_head = head_.load(std::memory_order_relaxed);
70
71 if (current_head == tail_.load(std::memory_order_acquire)) {
72 return nullptr; // Empty
73 }
74
75 T *element = buffer_[current_head];
76 head_.store((current_head + 1) % SIZE, std::memory_order_release);
77 return element;
78 }
79
80 size_t size() const {
81 uint8_t tail = tail_.load(std::memory_order_acquire);
82 uint8_t head = head_.load(std::memory_order_acquire);
83 return (tail - head + SIZE) % SIZE;
84 }
85
86 uint16_t get_and_reset_dropped_count() { return dropped_count_.exchange(0, std::memory_order_relaxed); }
87
88 void increment_dropped_count() { dropped_count_.fetch_add(1, std::memory_order_relaxed); }
89
90 bool empty() const { return head_.load(std::memory_order_acquire) == tail_.load(std::memory_order_acquire); }
91
92 bool full() const {
93 uint8_t next_tail = (tail_.load(std::memory_order_relaxed) + 1) % SIZE;
94 return next_tail == head_.load(std::memory_order_acquire);
95 }
96
97 protected:
98 T *buffer_[SIZE];
99 // Atomic: written by producer (push/increment), read+reset by consumer (get_and_reset)
100 std::atomic<uint16_t> dropped_count_; // 65535 max - more than enough for drop tracking
101 // Atomic: written by consumer (pop), read by producer (push) to check if full
102 // Using uint8_t limits queue size to 255 elements but saves memory and ensures
103 // atomic operations are efficient on all platforms
104 std::atomic<uint8_t> head_;
105 // Atomic: written by producer (push), read by consumer (pop) to check if empty
106 std::atomic<uint8_t> tail_;
107};
108
109// Extended queue with task notification support
110template<class T, uint8_t SIZE> class NotifyingLockFreeQueue : public LockFreeQueue<T, SIZE> {
111 public:
112 NotifyingLockFreeQueue() : LockFreeQueue<T, SIZE>(), task_to_notify_(nullptr) {}
113
114 bool push(T *element) {
115 bool was_empty;
116 uint8_t old_tail;
117 bool result = this->push_internal_(element, was_empty, old_tail);
118
119 // Notify optimization: only notify if we need to
120 if (result && task_to_notify_ != nullptr &&
121 (was_empty || this->head_.load(std::memory_order_acquire) == old_tail)) {
122 // Notify in two cases:
123 // 1. Queue was empty - consumer might be going to sleep
124 // 2. Consumer just caught up to where tail was - might go to sleep
125 // Note: There's a benign race in case 2 - between reading head and calling
126 // xTaskNotifyGive(), the consumer could advance further. This would result
127 // in an unnecessary wake-up, but is harmless and extremely rare in practice.
128 xTaskNotifyGive(task_to_notify_);
129 }
130 // Otherwise: consumer is still behind, no need to notify
131
132 return result;
133 }
134
135 // Set the FreeRTOS task handle to notify when items are pushed to the queue
136 // This enables efficient wake-up of a consumer task that's waiting for data
137 // @param task The FreeRTOS task handle to notify, or nullptr to disable notifications
138 void set_task_to_notify(TaskHandle_t task) { task_to_notify_ = task; }
139
140 private:
141 TaskHandle_t task_to_notify_;
142};
143
144} // namespace esphome
145
146#endif // defined(USE_ESP32)
uint16_t get_and_reset_dropped_count()
bool push(T *element)
std::atomic< uint16_t > dropped_count_
bool push_internal_(T *element, bool &was_empty, uint8_t &old_tail)
std::atomic< uint8_t > tail_
std::atomic< uint8_t > head_
void set_task_to_notify(TaskHandle_t task)
Providing packet encoding functions for exchanging data with a remote host.
Definition a01nyub.cpp:7