From b318499666872050aa07b078aba02753da2623fa Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Thu, 10 Sep 2026 18:19:35 -0500 Subject: [PATCH] [remote_transmitter] Queue RMT transmissions instead of blocking the main loop --- .../components/remote_transmitter/__init__.py | 35 +- .../remote_transmitter/remote_transmitter.h | 58 ++- .../remote_transmitter_rmt.cpp | 345 +++++++++++++----- esphome/core/helpers.h | 56 +++ .../remote_transmitter/test_queue.py | 33 ++ .../remote_transmitter/esp32-common.yaml | 2 + .../validate-queue-depth-1.esp32-idf.yaml | 12 + 7 files changed, 441 insertions(+), 100 deletions(-) create mode 100644 tests/component_tests/remote_transmitter/test_queue.py create mode 100644 tests/components/remote_transmitter/validate-queue-depth-1.esp32-idf.yaml diff --git a/esphome/components/remote_transmitter/__init__.py b/esphome/components/remote_transmitter/__init__.py index 58392c48ab..b4dac54498 100644 --- a/esphome/components/remote_transmitter/__init__.py +++ b/esphome/components/remote_transmitter/__init__.py @@ -29,7 +29,9 @@ _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 @@ -94,6 +96,15 @@ 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), } @@ -107,6 +118,8 @@ CONFIG_SCHEMA = ( CONF_USE_DMA, CONF_RMT_SYMBOLS, CONF_NON_BLOCKING, + CONF_QUEUE_DEPTH, + CONF_MAX_PENDING, ] ) ) @@ -127,7 +140,23 @@ def _validate_non_blocking(config: ConfigType) -> None: config[CONF_NON_BLOCKING] = True -FINAL_VALIDATE_SCHEMA = _validate_non_blocking +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 DIGITAL_WRITE_ACTION_SCHEMA = cv.maybe_simple_value( { @@ -166,6 +195,10 @@ 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: diff --git a/esphome/components/remote_transmitter/remote_transmitter.h b/esphome/components/remote_transmitter/remote_transmitter.h index 4db4e80a60..2beb19cecf 100644 --- a/esphome/components/remote_transmitter/remote_transmitter.h +++ b/esphome/components/remote_transmitter/remote_transmitter.h @@ -2,7 +2,9 @@ #include "esphome/components/remote_base/remote_base.h" #include "esphome/core/component.h" +#include "esphome/core/helpers.h" +#include #include #if defined(USE_ESP32) @@ -38,6 +40,30 @@ 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; + +// 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 @@ -64,6 +90,11 @@ 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 #endif #if (defined(USE_ESP32) && SOC_RMT_SUPPORTED) || defined(USE_LIBRETINY_VARIANT_RTL8720C) || \ defined(REMOTE_TRANSMITTER_BK_PWM) @@ -145,21 +176,36 @@ class RemoteTransmitterComponent final : public remote_base::RemoteTransmitterBa void wait_for_rmt_(); #if ESP_IDF_VERSION >= ESP_IDF_VERSION_VAL(5, 5, 1) - RemoteTransmitterComponentStore store_{}; - std::vector rmt_temp_; + 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_(); + + FixedVector slots_; + OverflowQueue 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}; #else std::vector rmt_temp_; + rmt_encoder_handle_t encoder_{NULL}; #endif uint32_t current_carrier_frequency_{38000}; + rmt_channel_handle_t channel_{NULL}; + esp_err_t error_code_{ESP_OK}; + std::string error_string_; bool initialized_{false}; bool with_dma_{false}; bool eot_level_{false}; - rmt_channel_handle_t channel_{NULL}; - rmt_encoder_handle_t encoder_{NULL}; - esp_err_t error_code_{ESP_OK}; - std::string error_string_; bool inverted_{false}; bool non_blocking_{false}; + uint8_t queue_depth_{1}; + uint8_t max_pending_{1}; #endif uint8_t carrier_duty_percent_{50}; diff --git a/esphome/components/remote_transmitter/remote_transmitter_rmt.cpp b/esphome/components/remote_transmitter/remote_transmitter_rmt.cpp index 3c9a12d472..2ddaad4c9b 100644 --- a/esphome/components/remote_transmitter/remote_transmitter_rmt.cpp +++ b/esphome/components/remote_transmitter/remote_transmitter_rmt.cpp @@ -15,6 +15,9 @@ static const char *const TAG = "remote_transmitter"; static constexpr uint32_t RMT_SYMBOL_DURATION_MAX = 0x7FFF; #if ESP_IDF_VERSION >= ESP_IDF_VERSION_VAL(5, 5, 1) +// How long a blocking wait sleeps between watchdog feeds +static constexpr int RMT_WAIT_SLICE_MS = 50; + static size_t IRAM_ATTR HOT encoder_callback(const void *data, size_t size, size_t written, size_t free, rmt_symbol_word_t *symbols, bool *done, void *arg) { auto *store = static_cast(arg); @@ -49,6 +52,39 @@ static size_t IRAM_ATTR HOT encoder_callback(const void *data, size_t size, size *done = false; return count; } + +// Splits a duration into 15-bit symbols; with out == nullptr only counts them +static size_t write_symbols(rmt_symbol_half_t *out, size_t pos, uint32_t ticks, bool level) { + size_t count = 0; + while (ticks > 0) { + uint32_t duration = std::min(ticks, RMT_SYMBOL_DURATION_MAX); + if (out != nullptr) { + out[pos + count] = { + .duration = static_cast(duration), + .level = static_cast(level), + }; + } + ticks -= duration; + count++; + } + 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(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(arg); + self->done_count_++; + self->enable_loop_soon_any_context(); + return false; +} #endif void RemoteTransmitterComponent::setup() { @@ -60,8 +96,10 @@ void RemoteTransmitterComponent::dump_config() { ESP_LOGCONFIG(TAG, "Remote Transmitter:"); ESP_LOGCONFIG(TAG, " Clock resolution: %" PRIu32 " hz\n" - " RMT symbols: %" PRIu32, - this->clock_resolution_, this->rmt_symbols_); + " RMT symbols: %" PRIu32 "\n" + " Queue depth: %u\n" + " Max pending: %u", + this->clock_resolution_, this->rmt_symbols_, this->queue_depth_, this->max_pending_); LOG_PIN(" Pin: ", this->pin_); if (this->current_carrier_frequency_ != 0 && this->carrier_duty_percent_ != 100) { @@ -83,8 +121,13 @@ void RemoteTransmitterComponent::digital_write(bool value) { rmt_transmit_config_t config; memset(&config, 0, sizeof(config)); config.flags.eot_level = value; - this->store_.times = 1; - this->store_.index = 0; + config.flags.queue_nonblocking = 1; + // everything queued must go out first; slot 0 is then free + this->wait_for_rmt_(); + RmtTxSlot &slot = this->slots_[0]; + slot.store.times = 1; + slot.store.index = 0; + rmt_encoder_handle_t encoder = slot.encoder; #else rmt_symbol_word_t symbol = { .duration0 = 1, @@ -95,17 +138,26 @@ void RemoteTransmitterComponent::digital_write(bool value) { rmt_transmit_config_t config; memset(&config, 0, sizeof(config)); config.flags.eot_level = value; + rmt_encoder_handle_t encoder = this->encoder_; #endif - esp_err_t error = rmt_transmit(this->channel_, this->encoder_, &symbol, sizeof(symbol), &config); + esp_err_t error = rmt_transmit(this->channel_, encoder, &symbol, sizeof(symbol), &config); if (error != ESP_OK) { ESP_LOGW(TAG, "rmt_transmit failed: %s", esp_err_to_name(error)); 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_; +#else error = rmt_tx_wait_all_done(this->channel_, -1); if (error != ESP_OK) { ESP_LOGW(TAG, "rmt_tx_wait_all_done failed: %s", esp_err_to_name(error)); this->status_set_warning(); } +#endif } void RemoteTransmitterComponent::configure_rmt_() { @@ -119,7 +171,11 @@ 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; @@ -152,18 +208,35 @@ void RemoteTransmitterComponent::configure_rmt_() { } #if ESP_IDF_VERSION >= ESP_IDF_VERSION_VAL(5, 5, 1) - 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_); + rmt_tx_event_callbacks_t callbacks; + memset(&callbacks, 0, sizeof(callbacks)); + callbacks.on_trans_done = tx_done_callback_; + error = rmt_tx_register_event_callbacks(this->channel_, &callbacks, this); if (error != ESP_OK) { this->error_code_ = error; - this->error_string_ = "in rmt_new_simple_encoder"; + this->error_string_ = "in rmt_tx_register_event_callbacks"; this->mark_failed(); 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; + } + } + this->backlog_.set_capacity(this->max_pending_ - this->queue_depth_); #else rmt_copy_encoder_config_t encoder; memset(&encoder, 0, sizeof(encoder)); @@ -206,6 +279,173 @@ 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 +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) { + App.feed_wdt(); + } + if (error != ESP_OK) { + ESP_LOGW(TAG, "rmt_tx_wait_all_done failed: %s", esp_err_to_name(error)); + this->status_set_warning(); + } +} + +// Blocks until every queued and backlogged frame has gone out, reporting completions in order +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); + } +} + +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(); +} + +// Encodes the repeat gap followed by the frame; with out == nullptr only counts symbols. +// The gap leads the buffer so the encoder skips it on the first pass and replays it +// before every repeat; offset receives the index of the first frame symbol. +size_t RemoteTransmitterComponent::encode_symbols_(rmt_symbol_half_t *out, uint32_t send_wait, uint32_t *offset) { + size_t count = write_symbols(out, 0, this->from_microseconds_(send_wait), this->eot_level_); + *offset = count; + for (int32_t value : this->temp_.get_data()) { + bool level = value >= 0; + if (!level) { + value = -value; + } + count += write_symbols(out, count, this->from_microseconds_(static_cast(value)), level ^ this->inverted_); + } + 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 + this->transmit_trigger_.trigger(); + this->complete_trigger_.trigger(); + return; + } + + 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->transmit_trigger_.trigger(); + if (slot != nullptr) { + this->submit_(*slot); + } + if (this->non_blocking_) { + this->enable_loop(); + } else { + this->wait_for_rmt_(); + } +} +#else void RemoteTransmitterComponent::wait_for_rmt_() { esp_err_t error = rmt_tx_wait_all_done(this->channel_, -1); if (error != ESP_OK) { @@ -216,87 +456,6 @@ void RemoteTransmitterComponent::wait_for_rmt_() { this->complete_trigger_.trigger(); } -#if ESP_IDF_VERSION >= ESP_IDF_VERSION_VAL(5, 5, 1) -void RemoteTransmitterComponent::send_internal(uint32_t send_times, uint32_t send_wait) { - uint64_t total_duration = 0; - - if (this->is_failed()) { - return; - } - - // if the timeout was cancelled, block until the tx is complete - if (this->non_blocking_ && this->cancel_timeout("complete")) { - this->wait_for_rmt_(); - } - - if (this->current_carrier_frequency_ != this->temp_.get_carrier_frequency()) { - this->current_carrier_frequency_ = this->temp_.get_carrier_frequency(); - this->configure_rmt_(); - } - - this->rmt_temp_.clear(); - this->rmt_temp_.reserve(this->temp_.get_data().size() + 1); - - // encode any delay at the start of the buffer to simplify the encoder callback - // this will be skipped the first time around - total_duration += send_wait * (send_times - 1); - send_wait = this->from_microseconds_(static_cast(send_wait)); - while (send_wait > 0) { - int32_t duration = std::min(send_wait, uint32_t(RMT_SYMBOL_DURATION_MAX)); - this->rmt_temp_.push_back({ - .duration = static_cast(duration), - .level = static_cast(this->eot_level_), - }); - send_wait -= duration; - } - - // encode data - size_t offset = this->rmt_temp_.size(); - for (int32_t value : this->temp_.get_data()) { - bool level = value >= 0; - if (!level) { - value = -value; - } - total_duration += value * send_times; - value = this->from_microseconds_(static_cast(value)); - while (value > 0) { - int32_t duration = std::min(value, int32_t(RMT_SYMBOL_DURATION_MAX)); - this->rmt_temp_.push_back({ - .duration = static_cast(duration), - .level = static_cast(level ^ this->inverted_), - }); - value -= duration; - } - } - - if ((this->rmt_temp_.data() == nullptr) || this->rmt_temp_.size() <= offset) { - ESP_LOGE(TAG, "Empty data"); - return; - } - - this->transmit_trigger_.trigger(); - - rmt_transmit_config_t config; - memset(&config, 0, sizeof(config)); - config.flags.eot_level = this->eot_level_; - this->store_.times = send_times; - this->store_.index = offset; - 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(); - } else { - this->status_clear_warning(); - } - - if (this->non_blocking_) { - this->set_timeout("complete", total_duration / 1000, [this]() { this->wait_for_rmt_(); }); - } else { - this->wait_for_rmt_(); - } -} -#else void RemoteTransmitterComponent::send_internal(uint32_t send_times, uint32_t send_wait) { if (this->is_failed()) return; diff --git a/esphome/core/helpers.h b/esphome/core/helpers.h index a0afb03124..7665e763bf 100644 --- a/esphome/core/helpers.h +++ b/esphome/core/helpers.h @@ -7,6 +7,7 @@ #include #include #include +#include #include #include #include @@ -730,6 +731,61 @@ template 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> class OverflowQueue { + public: + using entry_type = std::unique_ptr; + + 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(malloc(this->capacity_ * sizeof(T *))); // NOLINT + if (this->slots_ == nullptr) + return false; + } + this->slots_[this->tail_] = entry.release(); + this->tail_ = static_cast((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((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. diff --git a/tests/component_tests/remote_transmitter/test_queue.py b/tests/component_tests/remote_transmitter/test_queue.py new file mode 100644 index 0000000000..859fedbafe --- /dev/null +++ b/tests/component_tests/remote_transmitter/test_queue.py @@ -0,0 +1,33 @@ +"""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 diff --git a/tests/components/remote_transmitter/esp32-common.yaml b/tests/components/remote_transmitter/esp32-common.yaml index 79fd47ae21..178588e2b2 100644 --- a/tests/components/remote_transmitter/esp32-common.yaml +++ b/tests/components/remote_transmitter/esp32-common.yaml @@ -5,6 +5,8 @@ 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 diff --git a/tests/components/remote_transmitter/validate-queue-depth-1.esp32-idf.yaml b/tests/components/remote_transmitter/validate-queue-depth-1.esp32-idf.yaml new file mode 100644 index 0000000000..5e07a7f40c --- /dev/null +++ b/tests/components/remote_transmitter/validate-queue-depth-1.esp32-idf.yaml @@ -0,0 +1,12 @@ +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