[remote_transmitter] Drop a transmit while the RMT is busy instead of queueing it

This commit is contained in:
J. Nick Koston
2026-09-10 18:50:34 -05:00
parent 152bab9699
commit dcd39aa8ba
7 changed files with 85 additions and 331 deletions
@@ -29,9 +29,7 @@ _LOGGER = logging.getLogger(__name__)
AUTO_LOAD = ["remote_base"]
CONF_EOT_LEVEL = "eot_level"
CONF_MAX_PENDING = "max_pending"
CONF_NON_BLOCKING = "non_blocking"
CONF_QUEUE_DEPTH = "queue_depth"
CONF_ON_TRANSMIT = "on_transmit"
CONF_ON_COMPLETE = "on_complete"
CONF_TRANSMITTER_ID = remote_base.CONF_TRANSMITTER_ID
@@ -96,15 +94,6 @@ CONFIG_SCHEMA = (
esp32_s3=48,
): cv.All(cv.only_on_esp32, cv.int_range(min=2)),
cv.Optional(CONF_NON_BLOCKING): _validate_non_blocking_platform,
cv.SplitDefault(
CONF_QUEUE_DEPTH,
esp32=4,
esp32_c2=cv.UNDEFINED,
esp32_c61=cv.UNDEFINED,
): cv.All(cv.only_on_esp32, cv.int_range(min=1, max=16)),
cv.Optional(CONF_MAX_PENDING): cv.All(
cv.only_on_esp32, cv.int_range(min=1, max=64)
),
cv.Optional(CONF_ON_TRANSMIT): automation.validate_automation(single=True),
cv.Optional(CONF_ON_COMPLETE): automation.validate_automation(single=True),
}
@@ -118,8 +107,6 @@ CONFIG_SCHEMA = (
CONF_USE_DMA,
CONF_RMT_SYMBOLS,
CONF_NON_BLOCKING,
CONF_QUEUE_DEPTH,
CONF_MAX_PENDING,
]
)
)
@@ -140,23 +127,7 @@ def _validate_non_blocking(config: ConfigType) -> None:
config[CONF_NON_BLOCKING] = True
def _validate_queue(config: ConfigType) -> None:
if (queue_depth := config.get(CONF_QUEUE_DEPTH)) is None:
return
max_pending = config.setdefault(CONF_MAX_PENDING, 2 * queue_depth)
if max_pending < queue_depth:
raise cv.Invalid(
f"{CONF_MAX_PENDING} must be at least {CONF_QUEUE_DEPTH} ({queue_depth})",
path=[CONF_MAX_PENDING],
)
def _final_validate(config: ConfigType) -> None:
_validate_non_blocking(config)
_validate_queue(config)
FINAL_VALIDATE_SCHEMA = _final_validate
FINAL_VALIDATE_SCHEMA = _validate_non_blocking
DIGITAL_WRITE_ACTION_SCHEMA = cv.maybe_simple_value(
{
@@ -195,10 +166,6 @@ async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID], pin)
cg.add(var.set_rmt_symbols(config[CONF_RMT_SYMBOLS]))
cg.add(var.set_non_blocking(config[CONF_NON_BLOCKING]))
if config[CONF_NON_BLOCKING]:
# blocking mode never has a frame in flight when the next send starts
cg.add(var.set_queue_depth(config[CONF_QUEUE_DEPTH]))
cg.add(var.set_max_pending(config[CONF_MAX_PENDING]))
if CONF_CLOCK_RESOLUTION in config:
cg.add(var.set_clock_resolution(config[CONF_CLOCK_RESOLUTION]))
if CONF_USE_DMA in config:
@@ -2,9 +2,7 @@
#include "esphome/components/remote_base/remote_base.h"
#include "esphome/core/component.h"
#include "esphome/core/helpers.h"
#include <memory>
#include <vector>
#if defined(USE_ESP32)
@@ -40,30 +38,6 @@ struct RemoteTransmitterComponentStore {
uint32_t times{0};
uint32_t index{0};
};
// An encoded frame with its symbols in the same heap block, so a backlogged frame
// costs one allocation and moves into a queue slot without copying
struct RmtFrame {
uint32_t capacity; // symbols allocated
uint32_t count;
uint32_t offset; // first symbol after the repeat gap
uint32_t times;
uint32_t carrier_frequency;
rmt_symbol_half_t symbols[]; // NOLINT(modernize-avoid-c-arrays)
struct Deleter {
void operator()(RmtFrame *frame) const { free(frame); } // NOLINT(cppcoreguidelines-no-malloc)
};
};
using RmtFramePtr = std::unique_ptr<RmtFrame, RmtFrame::Deleter>;
// One entry of the hardware transmit queue; the encoder reads store and frame from
// the RMT interrupt until the transaction completes
struct RmtTxSlot {
RemoteTransmitterComponentStore store;
RmtFramePtr frame;
rmt_encoder_handle_t encoder{nullptr};
};
#endif
#endif
@@ -90,8 +64,6 @@ class RemoteTransmitterComponent final : public remote_base::RemoteTransmitterBa
#if defined(USE_ESP32) && SOC_RMT_SUPPORTED
void set_with_dma(bool with_dma) { this->with_dma_ = with_dma; }
void set_eot_level(bool eot_level) { this->eot_level_ = eot_level; }
void set_queue_depth(uint8_t queue_depth) { this->queue_depth_ = queue_depth; }
void set_max_pending(uint8_t max_pending) { this->max_pending_ = max_pending; }
#if ESP_IDF_VERSION >= ESP_IDF_VERSION_VAL(5, 5, 1)
void loop() override;
#endif
@@ -178,25 +150,17 @@ class RemoteTransmitterComponent final : public remote_base::RemoteTransmitterBa
#if ESP_IDF_VERSION >= ESP_IDF_VERSION_VAL(5, 5, 1)
static bool tx_done_callback(rmt_channel_handle_t channel, const rmt_tx_done_event_data_t *event, void *arg);
size_t encode_symbols_(rmt_symbol_half_t *out, uint32_t send_wait, uint32_t *offset);
bool encode_frame_(RmtFramePtr &frame, size_t count, uint32_t send_times, uint32_t send_wait);
void submit_(RmtTxSlot &slot);
void pump_backlog_();
void deliver_completions_();
void wait_all_done_();
void deliver_completion_();
FixedVector<RmtTxSlot> slots_;
OverflowQueue<RmtFrame, RmtFrame::Deleter> backlog_;
// Frames handed to the hardware queue and completions already reported; the
// interrupt advances done_count_, so frames in flight are submitted_ - done_count_
uint32_t submitted_{0};
uint32_t delivered_{0};
volatile uint32_t done_count_{0};
RemoteTransmitterComponentStore store_{};
std::vector<rmt_symbol_half_t> rmt_temp_;
#else
std::vector<rmt_symbol_word_t> rmt_temp_;
rmt_encoder_handle_t encoder_{NULL};
#endif
uint32_t current_carrier_frequency_{38000};
rmt_channel_handle_t channel_{NULL};
rmt_encoder_handle_t encoder_{NULL};
esp_err_t error_code_{ESP_OK};
std::string error_string_;
bool initialized_{false};
@@ -204,8 +168,12 @@ class RemoteTransmitterComponent final : public remote_base::RemoteTransmitterBa
bool eot_level_{false};
bool inverted_{false};
bool non_blocking_{false};
uint8_t queue_depth_{1};
uint8_t max_pending_{1};
#if ESP_IDF_VERSION >= ESP_IDF_VERSION_VAL(5, 5, 1)
// set when a frame is handed to the hardware, cleared once its completion is reported;
// the transmit done interrupt sets tx_done_
bool tx_active_{false};
volatile bool tx_done_{false};
#endif
#endif
uint8_t carrier_duty_percent_{50};
@@ -70,18 +70,10 @@ static size_t write_symbols(rmt_symbol_half_t *out, size_t pos, uint32_t ticks,
return count;
}
static RmtFrame *new_frame(uint32_t capacity) {
// malloc rather than new: with C++ exceptions disabled, new aborts instead of returning null
auto *frame = static_cast<RmtFrame *>(malloc(sizeof(RmtFrame) + capacity * sizeof(rmt_symbol_half_t))); // NOLINT
if (frame != nullptr)
frame->capacity = capacity;
return frame;
}
bool IRAM_ATTR HOT RemoteTransmitterComponent::tx_done_callback(rmt_channel_handle_t channel,
const rmt_tx_done_event_data_t *event, void *arg) {
auto *self = static_cast<RemoteTransmitterComponent *>(arg);
self->done_count_ = self->done_count_ + 1;
self->tx_done_ = true;
self->enable_loop_soon_any_context();
return false;
}
@@ -96,10 +88,8 @@ void RemoteTransmitterComponent::dump_config() {
ESP_LOGCONFIG(TAG, "Remote Transmitter:");
ESP_LOGCONFIG(TAG,
" Clock resolution: %" PRIu32 " hz\n"
" RMT symbols: %" PRIu32 "\n"
" Queue depth: %u\n"
" Max pending: %u",
this->clock_resolution_, this->rmt_symbols_, this->queue_depth_, this->max_pending_);
" RMT symbols: %" PRIu32,
this->clock_resolution_, this->rmt_symbols_);
LOG_PIN(" Pin: ", this->pin_);
if (this->current_carrier_frequency_ != 0 && this->carrier_duty_percent_ != 100) {
@@ -122,12 +112,11 @@ void RemoteTransmitterComponent::digital_write(bool value) {
memset(&config, 0, sizeof(config));
config.flags.eot_level = value;
config.flags.queue_nonblocking = 1;
// everything queued must go out first; slot 0 is then free
// a frame still on the wire finishes first and reports its completion
this->wait_for_rmt_();
RmtTxSlot &slot = this->slots_[0];
slot.store.times = 1;
slot.store.index = 0;
rmt_encoder_handle_t encoder = slot.encoder;
this->store_.times = 1;
this->store_.index = 0;
rmt_encoder_handle_t encoder = this->encoder_;
#else
rmt_symbol_word_t symbol = {
.duration0 = 1,
@@ -146,11 +135,9 @@ void RemoteTransmitterComponent::digital_write(bool value) {
this->status_set_warning();
}
#if ESP_IDF_VERSION >= ESP_IDF_VERSION_VAL(5, 5, 1)
if (error == ESP_OK)
this->submitted_++;
this->wait_all_done_();
// a level write is not a frame, so its completion is not reported
this->delivered_ = this->submitted_;
this->tx_done_ = false;
#else
error = rmt_tx_wait_all_done(this->channel_, -1);
if (error != ESP_OK) {
@@ -171,11 +158,7 @@ void RemoteTransmitterComponent::configure_rmt_() {
channel.resolution_hz = this->clock_resolution_;
channel.gpio_num = gpio_num_t(this->pin_->get_pin());
channel.mem_block_symbols = this->rmt_symbols_;
#if ESP_IDF_VERSION >= ESP_IDF_VERSION_VAL(5, 5, 1)
channel.trans_queue_depth = this->queue_depth_;
#else
channel.trans_queue_depth = 1;
#endif
channel.flags.invert_out = 0;
channel.flags.with_dma = this->with_dma_;
channel.intr_priority = 0;
@@ -219,24 +202,18 @@ void RemoteTransmitterComponent::configure_rmt_() {
return;
}
// one encoder per slot: its state is read from the interrupt while the slot's frame is in flight
this->slots_.init(this->queue_depth_);
for (uint8_t i = 0; i < this->queue_depth_; i++) {
RmtTxSlot &slot = this->slots_.emplace_back();
rmt_simple_encoder_config_t encoder;
memset(&encoder, 0, sizeof(encoder));
encoder.callback = encoder_callback;
encoder.arg = &slot.store;
encoder.min_chunk_size = 1;
error = rmt_new_simple_encoder(&encoder, &slot.encoder);
if (error != ESP_OK) {
this->error_code_ = error;
this->error_string_ = "in rmt_new_simple_encoder";
this->mark_failed();
return;
}
rmt_simple_encoder_config_t encoder;
memset(&encoder, 0, sizeof(encoder));
encoder.callback = encoder_callback;
encoder.arg = &this->store_;
encoder.min_chunk_size = 1;
error = rmt_new_simple_encoder(&encoder, &this->encoder_);
if (error != ESP_OK) {
this->error_code_ = error;
this->error_string_ = "in rmt_new_simple_encoder";
this->mark_failed();
return;
}
this->backlog_.set_capacity(this->max_pending_ - this->queue_depth_);
#else
rmt_copy_encoder_config_t encoder;
memset(&encoder, 0, sizeof(encoder));
@@ -280,7 +257,7 @@ void RemoteTransmitterComponent::configure_rmt_() {
}
#if ESP_IDF_VERSION >= ESP_IDF_VERSION_VAL(5, 5, 1)
// Blocks until the hardware queue is empty, feeding the watchdog while waiting
// Blocks until the hardware is idle, feeding the watchdog while waiting
void RemoteTransmitterComponent::wait_all_done_() {
esp_err_t error;
while ((error = rmt_tx_wait_all_done(this->channel_, RMT_WAIT_SLICE_MS)) == ESP_ERR_TIMEOUT) {
@@ -292,68 +269,24 @@ void RemoteTransmitterComponent::wait_all_done_() {
}
}
// Blocks until every queued and backlogged frame has gone out, reporting completions in order
void RemoteTransmitterComponent::deliver_completion_() {
this->tx_done_ = false;
this->tx_active_ = false;
this->complete_trigger_.trigger();
}
// Blocks until any frame on the wire has gone out and reports its completion
void RemoteTransmitterComponent::wait_for_rmt_() {
while (true) {
this->wait_all_done_();
this->deliver_completions_();
if (this->backlog_.empty())
return;
this->pump_backlog_();
}
}
void RemoteTransmitterComponent::deliver_completions_() {
uint32_t done = this->done_count_;
while (this->delivered_ != done) {
this->delivered_++;
this->complete_trigger_.trigger();
}
}
void RemoteTransmitterComponent::submit_(RmtTxSlot &slot) {
slot.store.times = slot.frame->times;
slot.store.index = slot.frame->offset;
rmt_transmit_config_t config;
memset(&config, 0, sizeof(config));
config.flags.eot_level = this->eot_level_;
config.flags.queue_nonblocking = 1;
esp_err_t error = rmt_transmit(this->channel_, slot.encoder, slot.frame->symbols,
slot.frame->count * sizeof(rmt_symbol_half_t), &config);
if (error != ESP_OK) {
ESP_LOGW(TAG, "rmt_transmit failed: %s", esp_err_to_name(error));
this->status_set_warning();
// nothing will complete, so report it now
this->complete_trigger_.trigger();
return;
}
this->status_clear_warning();
this->submitted_++;
}
// Moves backlogged frames into hardware queue slots as transmissions complete
void RemoteTransmitterComponent::pump_backlog_() {
while (!this->backlog_.empty() && this->submitted_ - this->done_count_ < this->queue_depth_) {
uint32_t carrier_frequency = this->backlog_.front().carrier_frequency;
if (carrier_frequency != this->current_carrier_frequency_) {
// the carrier applies to the whole channel, so it can only change once idle
if (this->submitted_ != this->done_count_)
return;
this->current_carrier_frequency_ = carrier_frequency;
this->configure_rmt_();
}
RmtTxSlot &slot = this->slots_[this->submitted_ % this->queue_depth_];
slot.frame = this->backlog_.pop();
this->submit_(slot);
}
this->wait_all_done_();
if (this->tx_active_)
this->deliver_completion_();
}
void RemoteTransmitterComponent::loop() {
this->deliver_completions_();
this->pump_backlog_();
// the transmit done interrupt re-enables the loop for the next completion
if (this->delivered_ == this->done_count_)
this->disable_loop();
if (this->tx_done_)
this->deliver_completion_();
// the transmit done interrupt re-enables the loop
this->disable_loop();
}
// Encodes the repeat gap followed by the frame; with out == nullptr only counts symbols.
@@ -372,76 +305,65 @@ size_t RemoteTransmitterComponent::encode_symbols_(rmt_symbol_half_t *out, uint3
return count;
}
// Encodes temp_ into frame, growing it when needed; returns false when out of memory
bool RemoteTransmitterComponent::encode_frame_(RmtFramePtr &frame, size_t count, uint32_t send_times,
uint32_t send_wait) {
if (frame == nullptr || frame->capacity < count) {
frame.reset(new_frame(count));
if (frame == nullptr)
return false;
}
frame->count = count;
frame->times = send_times;
frame->carrier_frequency = this->temp_.get_carrier_frequency();
this->encode_symbols_(frame->symbols, send_wait, &frame->offset);
return true;
}
void RemoteTransmitterComponent::send_internal(uint32_t send_times, uint32_t send_wait) {
if (this->is_failed()) {
return;
}
if (send_times == 0) {
// transmit nothing, but both triggers still fire so an on_complete-sequenced
// automation does not stall; a zero repeat count would never finish in the encoder
if (this->tx_active_ && this->tx_done_) {
// finished, but loop() has not run yet
this->deliver_completion_();
}
if (send_times == 0 || this->tx_active_) {
// nothing is sent, but both triggers still fire so an on_complete-sequenced automation
// does not stall; a zero repeat count would never finish in the encoder, and a frame that
// arrives while another is on the wire is dropped rather than blocking the loop
if (this->tx_active_) {
ESP_LOGW(TAG, "Transmitter busy, dropping");
this->status_set_warning();
}
this->transmit_trigger_.trigger();
this->complete_trigger_.trigger();
return;
}
if (this->current_carrier_frequency_ != this->temp_.get_carrier_frequency()) {
this->current_carrier_frequency_ = this->temp_.get_carrier_frequency();
this->configure_rmt_();
}
uint32_t offset;
size_t count = this->encode_symbols_(nullptr, send_wait, &offset);
if (count <= offset) {
ESP_LOGE(TAG, "Empty data");
return;
}
uint32_t carrier_frequency = this->temp_.get_carrier_frequency();
if (carrier_frequency != this->current_carrier_frequency_ && this->submitted_ == this->done_count_ &&
this->backlog_.empty()) {
this->current_carrier_frequency_ = carrier_frequency;
this->configure_rmt_();
}
// a frame goes straight to a free hardware slot unless it must wait behind the
// backlog or for a carrier change; beyond max_pending it is dropped
RmtTxSlot *slot = nullptr;
RmtFramePtr pending;
RmtFramePtr *frame = &pending;
if (carrier_frequency == this->current_carrier_frequency_ && this->backlog_.empty() &&
this->submitted_ - this->done_count_ < this->queue_depth_) {
slot = &this->slots_[this->submitted_ % this->queue_depth_];
frame = &slot->frame;
} else if (this->backlog_.full()) {
frame = nullptr;
}
if (frame == nullptr || !this->encode_frame_(*frame, count, send_times, send_wait) ||
(slot == nullptr && !this->backlog_.push(std::move(pending)))) {
ESP_LOGW(TAG, "Transmit queue full, dropping");
this->status_set_warning();
this->transmit_trigger_.trigger();
this->complete_trigger_.trigger();
return;
}
this->rmt_temp_.resize(count);
this->encode_symbols_(this->rmt_temp_.data(), send_wait, &offset);
this->store_.times = send_times;
this->store_.index = offset;
this->transmit_trigger_.trigger();
if (slot != nullptr) {
this->submit_(*slot);
rmt_transmit_config_t config;
memset(&config, 0, sizeof(config));
config.flags.eot_level = this->eot_level_;
config.flags.queue_nonblocking = 1;
this->tx_done_ = false;
this->tx_active_ = true;
esp_err_t error = rmt_transmit(this->channel_, this->encoder_, this->rmt_temp_.data(),
this->rmt_temp_.size() * sizeof(rmt_symbol_half_t), &config);
if (error != ESP_OK) {
ESP_LOGW(TAG, "rmt_transmit failed: %s", esp_err_to_name(error));
this->status_set_warning();
// nothing will complete, so report it now
this->deliver_completion_();
return;
}
if (this->non_blocking_) {
this->enable_loop();
} else {
this->status_clear_warning();
if (!this->non_blocking_) {
this->wait_for_rmt_();
}
}
-56
View File
@@ -7,7 +7,6 @@
#include <cstdarg>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <functional>
#include <iterator>
@@ -731,61 +730,6 @@ template<typename T> class FixedVector {
const T *end() const { return data_ + size_; }
};
/// Fixed-capacity circular queue of owned heap entries for rarely used backlogs.
/// Nothing is allocated until the first push, so a backlog that never fills costs only this object.
template<typename T, typename Deleter = std::default_delete<T>> class OverflowQueue {
public:
using entry_type = std::unique_ptr<T, Deleter>;
OverflowQueue() = default;
OverflowQueue(const OverflowQueue &) = delete;
OverflowQueue &operator=(const OverflowQueue &) = delete;
~OverflowQueue() {
while (!this->empty())
this->pop();
free(this->slots_); // NOLINT(cppcoreguidelines-no-malloc)
}
/// Set the maximum number of entries; must be called before the first push.
void set_capacity(uint8_t capacity) { this->capacity_ = capacity; }
bool empty() const { return this->count_ == 0; }
bool full() const { return this->count_ >= this->capacity_; }
/// Take ownership of an entry; returns false (and drops it) when full or out of memory.
bool push(entry_type entry) {
if (this->full())
return false;
if (this->slots_ == nullptr) {
// malloc rather than new: with C++ exceptions disabled, new aborts instead of returning null
this->slots_ = static_cast<T **>(malloc(this->capacity_ * sizeof(T *))); // NOLINT
if (this->slots_ == nullptr)
return false;
}
this->slots_[this->tail_] = entry.release();
this->tail_ = static_cast<uint8_t>((this->tail_ + 1) % this->capacity_);
this->count_++;
return true;
}
/// Oldest entry; caller must ensure the queue is not empty.
T &front() { return *this->slots_[this->head_]; }
/// Remove and return the oldest entry; caller must ensure the queue is not empty.
entry_type pop() {
entry_type entry(this->slots_[this->head_]);
this->head_ = static_cast<uint8_t>((this->head_ + 1) % this->capacity_);
this->count_--;
return entry;
}
protected:
T **slots_{nullptr};
uint8_t capacity_{0};
uint8_t head_{0};
uint8_t tail_{0};
uint8_t count_{0};
};
/// @brief Helper class for efficient buffer allocation - uses stack for small sizes, heap for large
/// This is useful when most operations need a small buffer but occasionally need larger ones.
/// The stack buffer avoids heap allocation in the common case, while heap fallback handles edge cases.
@@ -1,33 +0,0 @@
"""max_pending defaults to twice queue_depth and may not be smaller than it."""
import pytest
from esphome.components.remote_transmitter import (
CONF_MAX_PENDING,
CONF_QUEUE_DEPTH,
_validate_queue,
)
import esphome.config_validation as cv
def test_max_pending_defaults_to_twice_queue_depth() -> None:
config = {CONF_QUEUE_DEPTH: 4}
_validate_queue(config)
assert config[CONF_MAX_PENDING] == 8
def test_max_pending_explicit_is_kept() -> None:
config = {CONF_QUEUE_DEPTH: 4, CONF_MAX_PENDING: 4}
_validate_queue(config)
assert config[CONF_MAX_PENDING] == 4
def test_max_pending_below_queue_depth_is_rejected() -> None:
with pytest.raises(cv.Invalid, match="max_pending must be at least queue_depth"):
_validate_queue({CONF_QUEUE_DEPTH: 4, CONF_MAX_PENDING: 2})
def test_no_queue_depth_without_rmt() -> None:
config: dict[str, int] = {}
_validate_queue(config)
assert CONF_MAX_PENDING not in config
@@ -5,8 +5,6 @@ remote_transmitter:
non_blocking: true
clock_resolution: ${clock_resolution}
rmt_symbols: ${rmt_symbols}
queue_depth: 4
max_pending: 8
packages:
buttons: !include common-buttons.yaml
@@ -1,12 +0,0 @@
substitutions:
pin: GPIO2
remote_transmitter:
- id: xmitr
pin: ${pin}
carrier_duty_percent: 50%
non_blocking: true
queue_depth: 1
packages:
buttons: !include common-buttons.yaml