From 8fada4e68a2bf8e9bbd3fd3840ca801351560853 Mon Sep 17 00:00:00 2001 From: Arnaud Taffanel Date: Tue, 1 Sep 2026 11:41:42 +0200 Subject: [PATCH 1/7] Fix log configuration ID lifecycle (#577) --- cflib/crazyflie/log.py | 232 ++++++++++++++++++++++++++++--------- test/crazyflie/test_log.py | 215 ++++++++++++++++++++++++++++++++++ 2 files changed, 395 insertions(+), 52 deletions(-) create mode 100644 test/crazyflie/test_log.py diff --git a/cflib/crazyflie/log.py b/cflib/crazyflie/log.py index 021e230e0..5e7659c9a 100644 --- a/cflib/crazyflie/log.py +++ b/cflib/crazyflie/log.py @@ -52,6 +52,8 @@ import errno import logging import struct +from collections import deque +from threading import Lock from .toc import Toc from .toc import TocFetcher @@ -60,7 +62,7 @@ from cflib.utils.callbacks import Caller __author__ = 'Bitcraze AB' -__all__ = ['Log', 'LogTocElement'] +__all__ = ['Log', 'LogConfigError', 'LogTocElement'] # Channels used for the logging port CHAN_TOC = 0 @@ -92,6 +94,10 @@ logger = logging.getLogger(__name__) +class LogConfigError(Exception): + """Raised when a log configuration cannot change lifecycle state.""" + + class LogVariable(): """A logging variable""" @@ -143,8 +149,8 @@ def __init__(self, name, period_in_ms): self.added_cb = Caller() self.err_no = 0 - # These 3 variables are set by the log subsystem when the bock is added - self.id = 0 + # These 3 variables are set by the log subsystem when the block is added + self.id = None self.cf = None self.useV2 = False @@ -152,12 +158,17 @@ def __init__(self, name, period_in_ms): self.period_in_ms = period_in_ms self._added = False self._started = False + self._delete_pending = False self.pending = False self.valid = False self.variables = [] self.default_fetch_as = [] + self._resolved_default_variables = [] self.name = name + def _get_effective_variables(self): + return self.variables + self._resolved_default_variables + def add_variable(self, name, fetch_as=None): """Add a new variable to the configuration. @@ -220,9 +231,10 @@ def _cmd_append_block(self): return CMD_APPEND_BLOCK def _setup_log_elements(self, pk, next_to_add): + variables = self._get_effective_variables() i = next_to_add - for i in range(next_to_add, len(self.variables)): - var = self.variables[i] + for i in range(next_to_add, len(variables)): + var = variables[i] if (var.is_toc_variable() is False): # Memory location logger.debug('Logging to raw memory %d, 0x%04X', var.get_storage_and_fetch_byte(), var.address) @@ -260,14 +272,15 @@ def create(self): for block in self.cf.log.log_blocks: if block.pending or block.added or block.started: pending += 1 - num_variables += len(block.variables) + num_variables += len(block._get_effective_variables()) if pending < Log.MAX_BLOCKS: # # The Crazyflie firmware can only handle 128 variables before # erroring out with ENOMEM. # - if num_variables + len(self.variables) > Log.MAX_VARIABLES: + if (num_variables + len(self._get_effective_variables()) > + Log.MAX_VARIABLES): raise AttributeError( ('Adding this configuration would exceed max number ' 'of variables (%d)' % Log.MAX_VARIABLES) @@ -291,7 +304,11 @@ def create(self): def start(self): """Start the logging for this entry""" - if (self.cf.link is not None): + cf = self.cf + if cf is None or self.id is None: + raise LogConfigError( + 'Log configuration must be added before it can be started') + if (cf.link is not None): if (self._added is False): self.create() logger.debug('First time block is started, add block') @@ -306,37 +323,30 @@ def start(self): def stop(self): """Stop the logging for this entry""" - if (self.cf.link is not None): - if (self.id is None): - logger.warning('Stopping block, but no block registered') - else: - logger.debug('Sending stop logging for block id=%d', self.id) - pk = CRTPPacket() - pk.set_header(5, CHAN_SETTINGS) - pk.data = (CMD_STOP_LOGGING, self.id) - self.cf.send_packet( - pk, expected_reply=(CMD_STOP_LOGGING, self.id)) + cf = self.cf + block_id = self.id + if cf is None or block_id is None: + return + if (cf.link is not None): + logger.debug('Sending stop logging for block id=%d', block_id) + pk = CRTPPacket() + pk.set_header(5, CHAN_SETTINGS) + pk.data = (CMD_STOP_LOGGING, block_id) + cf.send_packet( + pk, expected_reply=(CMD_STOP_LOGGING, block_id)) def delete(self): """Delete this entry in the Crazyflie""" - if (self.cf.link is not None): - if (self.id is None): - logger.warning('Delete block, but no block registered') - else: - logger.debug('LogEntry: Sending delete logging for block id=%d' - % self.id) - pk = CRTPPacket() - pk.set_header(5, CHAN_SETTINGS) - pk.data = (CMD_DELETE_BLOCK, self.id) - self.cf.send_packet( - pk, expected_reply=(CMD_DELETE_BLOCK, self.id)) + cf = self.cf + if cf is not None and self.id is not None: + cf.log._delete_config(self) def unpack_log_data(self, log_data, timestamp): """Unpack received logging data so it represent real values according to the configuration in the entry""" ret_data = {} data_index = 0 - for var in self.variables: + for var in self._get_effective_variables(): size = LogTocElement.get_size_from_id(var.fetch_as) name = var.name unpackstring = LogTocElement.get_unpack_string_from_id( @@ -415,6 +425,7 @@ class Log(): """Create log configuration""" MAX_BLOCKS = 16 + MAX_CONFIG_IDS = 256 MAX_VARIABLES = 128 # These codes can be decoded using os.stderror, but @@ -436,6 +447,7 @@ def __init__(self, crazyflie=None): self.cf = crazyflie self.toc = None self.cf.add_port_callback(CRTPPort.LOGGING, self._new_packet_cb) + self.cf.disconnected.add_callback(self._disconnected) self.toc_updated = Caller() self.state = IDLE @@ -444,7 +456,10 @@ def __init__(self, crazyflie=None): self._refresh_callback = None self._toc_cache = None - self._config_id_counter = 1 + self._registration_lock = Lock() + self._available_config_ids = deque() + self._ids_ready = False + self._reset_pending = False self._useV2 = False @@ -459,13 +474,21 @@ def add_config(self, logconf): connected when calling this method, otherwise it will fail.""" if not self.cf.link: - logger.error('Cannot add configs without being connected to a ' - 'Crazyflie!') - return + raise LogConfigError( + 'Cannot add log configurations without a connection') + + with self._registration_lock: + if logconf.id is not None or logconf.cf is not None: + raise LogConfigError( + 'Log configuration is already registered') + if not self._ids_ready: + raise LogConfigError( + 'Log configuration IDs are not ready') # If the log configuration contains variables that we added without # type (i.e we want the stored as type for fetching as well) then # resolve this now and add them to the block again. + resolved_default_variables = [] for name in logconf.default_fetch_as: var = self.toc.get_element_by_complete_name(name) if not var: @@ -473,15 +496,14 @@ def add_config(self, logconf): '%s not in TOC, this block cannot be used!', name) logconf.valid = False raise KeyError('Variable {} not in TOC'.format(name)) - # Now that we know what type this variable has, add it to the log - # config again with the correct type - logconf.add_variable(name, var.ctype) + resolved_default_variables.append(LogVariable(name, var.ctype)) # Now check that all the added variables are in the TOC and that # the total size constraint of a data packet with logging data is # not size = 0 - for var in logconf.variables: + effective_variables = logconf.variables + resolved_default_variables + for var in effective_variables: size += LogTocElement.get_size_from_id(var.fetch_as) # Check that we are able to find the variable in the TOC so # we can return error already now and not when the config is sent @@ -495,12 +517,23 @@ def add_config(self, logconf): if (size <= LogConfig.MAX_LEN and (logconf.period > 0 and logconf.period < 0xFF)): - logconf.valid = True - logconf.cf = self.cf - logconf.id = self._config_id_counter - logconf.useV2 = self._useV2 - self._config_id_counter = (self._config_id_counter + 1) % 255 - self.log_blocks.append(logconf) + with self._registration_lock: + if logconf.id is not None or logconf.cf is not None: + raise LogConfigError( + 'Log configuration is already registered') + if not self._ids_ready: + raise LogConfigError( + 'Log configuration IDs are not ready') + if not self._available_config_ids: + raise LogConfigError('No log configuration IDs available') + logconf.valid = True + logconf.cf = self.cf + logconf.id = self._available_config_ids.popleft() + logconf._delete_pending = False + logconf._resolved_default_variables = ( + resolved_default_variables) + logconf.useV2 = self._useV2 + self.log_blocks.append(logconf) self.block_added_cb.call(logconf) else: logconf.valid = False @@ -512,7 +545,6 @@ def reset(self): """ Reset the log system and remove all log blocks """ - self.log_blocks = [] self._send_reset_packet() def refresh_toc(self, refresh_done_callback, toc_cache): @@ -527,17 +559,101 @@ def refresh_toc(self, refresh_done_callback, toc_cache): self._send_reset_packet() def _send_reset_packet(self): + with self._registration_lock: + if self._reset_pending: + return + self._reset_pending = True + self._ids_ready = False + self._available_config_ids.clear() + pk = CRTPPacket() pk.set_header(CRTPPort.LOGGING, CHAN_SETTINGS) pk.data = (CMD_RESET_LOGGING,) self.cf.send_packet(pk, expected_reply=(CMD_RESET_LOGGING,)) + def _detach_all_configs(self, restore_ids, require_reset_pending=False): + with self._registration_lock: + if require_reset_pending and not self._reset_pending: + return False + blocks = self.log_blocks + self.log_blocks = [] + if restore_ids: + self._available_config_ids = deque( + range(self.MAX_CONFIG_IDS)) + else: + self._available_config_ids.clear() + self._ids_ready = restore_ids + self._reset_pending = False + + callbacks = [] + for block in blocks: + callbacks.append((block, block.started, block.added)) + block._started = False + block._added = False + block._delete_pending = False + block.pending = False + block.id = None + block.cf = None + block._resolved_default_variables = [] + + for block, was_started, was_added in callbacks: + if was_started: + block.started_cb.call(block, False) + if was_added: + block.added_cb.call(block, False) + return True + + def _disconnected(self, uri): + self._detach_all_configs(restore_ids=False) + def _find_block(self, id): - for block in self.log_blocks: - if block.id == id: - return block + with self._registration_lock: + for block in self.log_blocks: + if block.id == id: + return block return None + def _retire_config(self, logconf, block_id): + with self._registration_lock: + if (logconf not in self.log_blocks or + logconf.id != block_id or + not logconf._delete_pending): + return False + + was_started = logconf.started + was_added = logconf.added + self.log_blocks.remove(logconf) + self._available_config_ids.append(block_id) + logconf._started = False + logconf._added = False + logconf._delete_pending = False + logconf.pending = False + logconf.id = None + logconf.cf = None + logconf._resolved_default_variables = [] + + if was_started: + logconf.started_cb.call(logconf, False) + if was_added: + logconf.added_cb.call(logconf, False) + return True + + def _delete_config(self, logconf): + with self._registration_lock: + if logconf not in self.log_blocks or logconf.id is None: + return + if logconf._delete_pending: + return + logconf._delete_pending = True + block_id = logconf.id + + logger.debug('LogEntry: Sending delete logging for block id=%d', + block_id) + pk = CRTPPacket() + pk.set_header(CRTPPort.LOGGING, CHAN_SETTINGS) + pk.data = (CMD_DELETE_BLOCK, block_id) + self.cf.send_packet(pk, expected_reply=(CMD_DELETE_BLOCK, block_id)) + def _new_packet_cb(self, packet): """Callback for newly arrived packets with TOC information""" chan = packet.channel @@ -604,15 +720,27 @@ def _new_packet_cb(self, packet): if error_status == 0x00 or error_status == errno.ENOENT: logger.info('Have successfully deleted id=%d', id) if block: - block.started = False - block.added = False + self._retire_config(block, id) + elif block: + with self._registration_lock: + if block.id != id or not block._delete_pending: + return + block._delete_pending = False + block.err_no = error_status + msg = self._err_codes[error_status] + block.error_cb.call(block, msg) if (cmd == CMD_RESET_LOGGING): + if error_status != 0x00: + return + + reset_completed = self._detach_all_configs( + restore_ids=True, require_reset_pending=True) + if not reset_completed: + return # Guard against multiple responses due to re-sending if not self.toc: logger.debug('Logging reset, continue with TOC download') - self.log_blocks = [] - self.toc = Toc() toc_fetcher = TocFetcher(self.cf, LogTocElement, CRTPPort.LOGGING, diff --git a/test/crazyflie/test_log.py b/test/crazyflie/test_log.py new file mode 100644 index 000000000..bb9da780b --- /dev/null +++ b/test/crazyflie/test_log.py @@ -0,0 +1,215 @@ +# -*- coding: utf-8 -*- +# +# || ____ _ __ +# +------+ / __ )(_) /_______________ _____ ___ +# | 0xBC | / __ / / __/ ___/ ___/ __ `/_ / / _ \ +# +------+ / /_/ / / /_/ /__/ / / /_/ / / /_/ __/ +# || || /_____/\___/\___/_/ \__,_/ /___/\___/ +# +# Copyright (C) 2026 Bitcraze AB +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public License +# as published by the Free Software Foundation; either version 2 +# of the License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +import errno +import struct +import unittest +from unittest.mock import MagicMock + +from cflib.crazyflie import Crazyflie +from cflib.crazyflie.log import CHAN_SETTINGS +from cflib.crazyflie.log import CMD_DELETE_BLOCK +from cflib.crazyflie.log import CMD_RESET_LOGGING +from cflib.crazyflie.log import Log +from cflib.crazyflie.log import LogConfig +from cflib.crazyflie.log import LogConfigError +from cflib.crazyflie.toc import Toc +from cflib.crtp.crtpstack import CRTPPacket +from cflib.crtp.crtpstack import CRTPPort +from cflib.utils.callbacks import Caller + + +class LogTest(unittest.TestCase): + + def setUp(self): + self.cf = MagicMock(spec=Crazyflie) + self.cf.link = object() + self.cf.disconnected = Caller() + self.log = Log(self.cf) + self.cf.log = self.log + self.log.toc = Toc() + + def _acknowledge(self, command, block_id=0, error_status=0): + packet = CRTPPacket() + packet.set_header(CRTPPort.LOGGING, CHAN_SETTINGS) + packet.data = (command, block_id, error_status) + self.log._new_packet_cb(packet) + + def _make_config(self, name): + config = LogConfig(name, 100) + config.add_memory('value', 'uint8_t', 'uint8_t', 0x1000) + return config + + def test_all_byte_values_are_available_as_log_config_ids(self): + self.log.reset() + self._acknowledge(CMD_RESET_LOGGING) + + configs = [self._make_config('config-{}'.format(i)) for i in range(256)] + for config in configs: + self.log.add_config(config) + + self.assertEqual(list(range(256)), [config.id for config in configs]) + with self.assertRaises(LogConfigError): + self.log.add_config(self._make_config('one-too-many')) + + def test_deleted_id_is_released_after_acknowledgement(self): + self.log.reset() + self._acknowledge(CMD_RESET_LOGGING) + deleted_config = self._make_config('deleted') + self.log.add_config(deleted_config) + deleted_config.delete() + for i in range(1, 256): + self.log.add_config(self._make_config('config-{}'.format(i))) + + with self.assertRaises(LogConfigError): + self.log.add_config(self._make_config('before-ack')) + + self._acknowledge(CMD_DELETE_BLOCK, deleted_config.id) + + self.assertIsNone(deleted_config.id) + self.assertIsNone(deleted_config.cf) + self.assertNotIn(deleted_config, self.log.log_blocks) + self.log.add_config(deleted_config) + self.assertEqual(0, deleted_config.id) + + def test_delete_is_idempotent_until_a_failed_acknowledgement(self): + self.log.reset() + self._acknowledge(CMD_RESET_LOGGING) + config = self._make_config('config') + self.log.add_config(config) + self.cf.send_packet.reset_mock() + + config.delete() + config.delete() + + self.assertEqual(1, self.cf.send_packet.call_count) + self._acknowledge(CMD_DELETE_BLOCK, config.id, errno.ENOMEM) + self.assertEqual(0, config.id) + self.assertIn(config, self.log.log_blocks) + + config.delete() + self.assertEqual(2, self.cf.send_packet.call_count) + + def test_reset_acknowledgement_detaches_configs_and_restores_ids(self): + self.log.reset() + self._acknowledge(CMD_RESET_LOGGING) + config = self._make_config('old-config') + self.log.add_config(config) + + self.log.reset() + + self.assertEqual(0, config.id) + with self.assertRaises(LogConfigError): + self.log.add_config(self._make_config('during-reset')) + + self._acknowledge(CMD_RESET_LOGGING) + + self.assertIsNone(config.id) + self.assertIsNone(config.cf) + self.assertEqual([], self.log.log_blocks) + new_config = self._make_config('new-config') + self.log.add_config(new_config) + self.assertEqual(0, new_config.id) + + def test_disconnect_detaches_configs_without_restoring_ids(self): + self.log.reset() + self._acknowledge(CMD_RESET_LOGGING) + config = self._make_config('config') + self.log.add_config(config) + + self.cf.disconnected.call('radio://test') + + self.assertIsNone(config.id) + self.assertIsNone(config.cf) + self.assertEqual([], self.log.log_blocks) + self.cf.link = None + with self.assertRaises(LogConfigError): + self.log.add_config(self._make_config('after-disconnect')) + + def test_detached_config_requires_registration_before_start(self): + config = self._make_config('config') + + config.stop() + config.delete() + with self.assertRaises(LogConfigError): + config.start() + + def test_config_cannot_be_registered_twice(self): + self.log.reset() + self._acknowledge(CMD_RESET_LOGGING) + config = self._make_config('config') + self.log.add_config(config) + + with self.assertRaises(LogConfigError): + self.log.add_config(config) + + other_config = self._make_config('other-config') + self.log.add_config(other_config) + self.assertEqual(1, other_config.id) + + def test_untyped_variables_are_resolved_fresh_when_reregistered(self): + self.log.reset() + self._acknowledge(CMD_RESET_LOGGING) + toc_element = MagicMock() + toc_element.ctype = 'uint8_t' + self.log.toc = MagicMock() + self.log.toc.get_element_by_complete_name.return_value = toc_element + config = LogConfig('config', 100) + config.add_variable('group.value') + self.log.add_config(config) + config.delete() + self._acknowledge(CMD_DELETE_BLOCK, config.id) + + toc_element.ctype = 'uint16_t' + self.log.add_config(config) + received = [] + config.data_received_cb.add_callback( + lambda timestamp, data, logconf: received.append(data)) + + config.unpack_log_data(struct.pack(' Date: Tue, 1 Sep 2026 11:45:53 +0200 Subject: [PATCH 2/7] Handle log lifecycle recovery edges --- cflib/crazyflie/log.py | 66 +++++++++++++++++++++----------------- test/crazyflie/test_log.py | 36 +++++++++++++++++++++ 2 files changed, 73 insertions(+), 29 deletions(-) diff --git a/cflib/crazyflie/log.py b/cflib/crazyflie/log.py index 5e7659c9a..aa10770f5 100644 --- a/cflib/crazyflie/log.py +++ b/cflib/crazyflie/log.py @@ -169,6 +169,23 @@ def __init__(self, name, period_in_ms): def _get_effective_variables(self): return self.variables + self._resolved_default_variables + def _detach(self): + previous_state = (self.started, self.added) + self._started = False + self._added = False + self._delete_pending = False + self.pending = False + self.id = None + self.cf = None + self._resolved_default_variables = [] + return previous_state + + def _call_detached_callbacks(self, was_started, was_added): + if was_started: + self.started_cb.call(self, False) + if was_added: + self.added_cb.call(self, False) + def add_variable(self, name, fetch_as=None): """Add a new variable to the configuration. @@ -587,20 +604,10 @@ def _detach_all_configs(self, restore_ids, require_reset_pending=False): callbacks = [] for block in blocks: - callbacks.append((block, block.started, block.added)) - block._started = False - block._added = False - block._delete_pending = False - block.pending = False - block.id = None - block.cf = None - block._resolved_default_variables = [] - - for block, was_started, was_added in callbacks: - if was_started: - block.started_cb.call(block, False) - if was_added: - block.added_cb.call(block, False) + callbacks.append((block, block._detach())) + + for block, previous_state in callbacks: + block._call_detached_callbacks(*previous_state) return True def _disconnected(self, uri): @@ -620,22 +627,12 @@ def _retire_config(self, logconf, block_id): not logconf._delete_pending): return False - was_started = logconf.started - was_added = logconf.added self.log_blocks.remove(logconf) - self._available_config_ids.append(block_id) - logconf._started = False - logconf._added = False - logconf._delete_pending = False - logconf.pending = False - logconf.id = None - logconf.cf = None - logconf._resolved_default_variables = [] + if self._ids_ready: + self._available_config_ids.append(block_id) + previous_state = logconf._detach() - if was_started: - logconf.started_cb.call(logconf, False) - if was_added: - logconf.added_cb.call(logconf, False) + logconf._call_detached_callbacks(*previous_state) return True def _delete_config(self, logconf): @@ -652,7 +649,15 @@ def _delete_config(self, logconf): pk = CRTPPacket() pk.set_header(CRTPPort.LOGGING, CHAN_SETTINGS) pk.data = (CMD_DELETE_BLOCK, block_id) - self.cf.send_packet(pk, expected_reply=(CMD_DELETE_BLOCK, block_id)) + try: + self.cf.send_packet( + pk, expected_reply=(CMD_DELETE_BLOCK, block_id)) + except Exception: + with self._registration_lock: + if (logconf.id == block_id and + logconf._delete_pending): + logconf._delete_pending = False + raise def _new_packet_cb(self, packet): """Callback for newly arrived packets with TOC information""" @@ -732,6 +737,9 @@ def _new_packet_cb(self, packet): if (cmd == CMD_RESET_LOGGING): if error_status != 0x00: + with self._registration_lock: + if self._reset_pending: + self._reset_pending = False return reset_completed = self._detach_all_configs( diff --git a/test/crazyflie/test_log.py b/test/crazyflie/test_log.py index bb9da780b..5f73bdbf0 100644 --- a/test/crazyflie/test_log.py +++ b/test/crazyflie/test_log.py @@ -22,6 +22,7 @@ import errno import struct import unittest +from concurrent.futures import ThreadPoolExecutor from unittest.mock import MagicMock from cflib.crazyflie import Crazyflie @@ -210,6 +211,41 @@ def test_duplicate_reset_ack_does_not_detach_new_config(self): self.assertEqual(0, config.id) self.assertIn(config, self.log.log_blocks) + def test_delete_can_be_retried_when_sending_fails(self): + self.log.reset() + self._acknowledge(CMD_RESET_LOGGING) + config = self._make_config('config') + self.log.add_config(config) + self.cf.send_packet.reset_mock() + self.cf.send_packet.side_effect = [RuntimeError('send failed'), None] + + with self.assertRaises(RuntimeError): + config.delete() + config.delete() + + self.assertEqual(2, self.cf.send_packet.call_count) + + def test_reset_can_be_retried_after_failed_acknowledgement(self): + self.log.reset() + self._acknowledge(CMD_RESET_LOGGING, error_status=errno.ENOEXEC) + self.cf.send_packet.reset_mock() + + self.log.reset() + + self.assertEqual(1, self.cf.send_packet.call_count) + + def test_ids_are_unique_when_configs_are_registered_concurrently(self): + self.log.reset() + self._acknowledge(CMD_RESET_LOGGING) + configs = [self._make_config('config-{}'.format(i)) + for i in range(256)] + + with ThreadPoolExecutor(max_workers=16) as executor: + list(executor.map(self.log.add_config, configs)) + + self.assertEqual(list(range(256)), sorted( + config.id for config in configs)) + if __name__ == '__main__': unittest.main() From 17e4da787007f438e6dbea19acf15e3f6237e63e Mon Sep 17 00:00:00 2001 From: Arnaud Taffanel Date: Tue, 1 Sep 2026 11:49:55 +0200 Subject: [PATCH 3/7] Serialize log reset and delete commands --- cflib/crazyflie/log.py | 73 ++++++++++++++++++++++---------------- test/crazyflie/test_log.py | 45 +++++++++++++++++++++++ 2 files changed, 87 insertions(+), 31 deletions(-) diff --git a/cflib/crazyflie/log.py b/cflib/crazyflie/log.py index aa10770f5..b5bf41a28 100644 --- a/cflib/crazyflie/log.py +++ b/cflib/crazyflie/log.py @@ -474,6 +474,7 @@ def __init__(self, crazyflie=None): self._toc_cache = None self._registration_lock = Lock() + self._command_lock = Lock() self._available_config_ids = deque() self._ids_ready = False self._reset_pending = False @@ -576,17 +577,24 @@ def refresh_toc(self, refresh_done_callback, toc_cache): self._send_reset_packet() def _send_reset_packet(self): - with self._registration_lock: - if self._reset_pending: - return - self._reset_pending = True - self._ids_ready = False - self._available_config_ids.clear() + with self._command_lock: + with self._registration_lock: + if self._reset_pending: + return + self._reset_pending = True + self._ids_ready = False + self._available_config_ids.clear() - pk = CRTPPacket() - pk.set_header(CRTPPort.LOGGING, CHAN_SETTINGS) - pk.data = (CMD_RESET_LOGGING,) - self.cf.send_packet(pk, expected_reply=(CMD_RESET_LOGGING,)) + pk = CRTPPacket() + pk.set_header(CRTPPort.LOGGING, CHAN_SETTINGS) + pk.data = (CMD_RESET_LOGGING,) + try: + self.cf.send_packet( + pk, expected_reply=(CMD_RESET_LOGGING,)) + except Exception: + with self._registration_lock: + self._reset_pending = False + raise def _detach_all_configs(self, restore_ids, require_reset_pending=False): with self._registration_lock: @@ -636,28 +644,31 @@ def _retire_config(self, logconf, block_id): return True def _delete_config(self, logconf): - with self._registration_lock: - if logconf not in self.log_blocks or logconf.id is None: - return - if logconf._delete_pending: - return - logconf._delete_pending = True - block_id = logconf.id - - logger.debug('LogEntry: Sending delete logging for block id=%d', - block_id) - pk = CRTPPacket() - pk.set_header(CRTPPort.LOGGING, CHAN_SETTINGS) - pk.data = (CMD_DELETE_BLOCK, block_id) - try: - self.cf.send_packet( - pk, expected_reply=(CMD_DELETE_BLOCK, block_id)) - except Exception: + with self._command_lock: with self._registration_lock: - if (logconf.id == block_id and - logconf._delete_pending): - logconf._delete_pending = False - raise + if (not self._ids_ready or + logconf not in self.log_blocks or + logconf.id is None): + return + if logconf._delete_pending: + return + logconf._delete_pending = True + block_id = logconf.id + + logger.debug('LogEntry: Sending delete logging for block id=%d', + block_id) + pk = CRTPPacket() + pk.set_header(CRTPPort.LOGGING, CHAN_SETTINGS) + pk.data = (CMD_DELETE_BLOCK, block_id) + try: + self.cf.send_packet( + pk, expected_reply=(CMD_DELETE_BLOCK, block_id)) + except Exception: + with self._registration_lock: + if (logconf.id == block_id and + logconf._delete_pending): + logconf._delete_pending = False + raise def _new_packet_cb(self, packet): """Callback for newly arrived packets with TOC information""" diff --git a/test/crazyflie/test_log.py b/test/crazyflie/test_log.py index 5f73bdbf0..a2a05a8b6 100644 --- a/test/crazyflie/test_log.py +++ b/test/crazyflie/test_log.py @@ -21,6 +21,7 @@ # along with this program. If not, see . import errno import struct +import threading import unittest from concurrent.futures import ThreadPoolExecutor from unittest.mock import MagicMock @@ -246,6 +247,50 @@ def test_ids_are_unique_when_configs_are_registered_concurrently(self): self.assertEqual(list(range(256)), sorted( config.id for config in configs)) + def test_reset_waits_for_in_flight_delete_command(self): + self.log.reset() + self._acknowledge(CMD_RESET_LOGGING) + config = self._make_config('config') + self.log.add_config(config) + delete_send_started = threading.Event() + allow_delete_send = threading.Event() + reset_finished = threading.Event() + + def send_packet(packet, expected_reply): + if packet.data[0] == CMD_DELETE_BLOCK: + delete_send_started.set() + allow_delete_send.wait() + + self.cf.send_packet.side_effect = send_packet + delete_thread = threading.Thread(target=config.delete) + delete_thread.start() + self.assertTrue(delete_send_started.wait(1.0)) + + def reset(): + self.log.reset() + reset_finished.set() + + reset_thread = threading.Thread(target=reset) + reset_thread.start() + + try: + self.assertFalse(reset_finished.wait(0.1)) + finally: + allow_delete_send.set() + delete_thread.join(1.0) + reset_thread.join(1.0) + self.assertFalse(delete_thread.is_alive()) + self.assertFalse(reset_thread.is_alive()) + + def test_reset_can_be_retried_when_sending_fails(self): + self.cf.send_packet.side_effect = [RuntimeError('send failed'), None] + + with self.assertRaises(RuntimeError): + self.log.reset() + self.log.reset() + + self.assertEqual(2, self.cf.send_packet.call_count) + if __name__ == '__main__': unittest.main() From 9cc01fee8fefac44e3feb2a44d6876d950f23a1e Mon Sep 17 00:00:00 2001 From: Arnaud Taffanel Date: Tue, 1 Sep 2026 11:53:41 +0200 Subject: [PATCH 4/7] Serialize log configuration commands --- cflib/crazyflie/log.py | 68 ++++++++++++++++++++++++++------------ test/crazyflie/test_log.py | 39 ++++++++++++++++++++++ 2 files changed, 86 insertions(+), 21 deletions(-) diff --git a/cflib/crazyflie/log.py b/cflib/crazyflie/log.py index b5bf41a28..9769b4f0b 100644 --- a/cflib/crazyflie/log.py +++ b/cflib/crazyflie/log.py @@ -280,6 +280,15 @@ def _setup_log_elements(self, pk, next_to_add): def create(self): """Save the log configuration in the Crazyflie""" + cf = self.cf + if cf is None or self.id is None: + raise LogConfigError( + 'Log configuration must be added before it can be created') + with cf.log._command_lock: + cf.log._require_registered(self) + self._create() + + def _create(self): command = self._cmd_create_block() next_to_add = 0 is_done = False @@ -325,32 +334,38 @@ def start(self): if cf is None or self.id is None: raise LogConfigError( 'Log configuration must be added before it can be started') - if (cf.link is not None): - if (self._added is False): - self.create() - logger.debug('First time block is started, add block') - else: - logger.debug('Block already registered, starting logging' - ' for id=%d', self.id) - pk = CRTPPacket() - pk.set_header(5, CHAN_SETTINGS) - pk.data = (CMD_START_LOGGING, self.id, self.period) - self.cf.send_packet(pk, expected_reply=( - CMD_START_LOGGING, self.id)) + with cf.log._command_lock: + cf.log._require_registered(self) + if (cf.link is not None): + if (self._added is False): + self._create() + logger.debug('First time block is started, add block') + else: + logger.debug( + 'Block already registered, starting logging for id=%d', + self.id) + pk = CRTPPacket() + pk.set_header(5, CHAN_SETTINGS) + pk.data = (CMD_START_LOGGING, self.id, self.period) + cf.send_packet(pk, expected_reply=( + CMD_START_LOGGING, self.id)) def stop(self): """Stop the logging for this entry""" cf = self.cf - block_id = self.id - if cf is None or block_id is None: + if cf is None or self.id is None: return - if (cf.link is not None): - logger.debug('Sending stop logging for block id=%d', block_id) - pk = CRTPPacket() - pk.set_header(5, CHAN_SETTINGS) - pk.data = (CMD_STOP_LOGGING, block_id) - cf.send_packet( - pk, expected_reply=(CMD_STOP_LOGGING, block_id)) + with cf.log._command_lock: + if not cf.log._is_registered(self): + return + block_id = self.id + if (cf.link is not None): + logger.debug('Sending stop logging for block id=%d', block_id) + pk = CRTPPacket() + pk.set_header(5, CHAN_SETTINGS) + pk.data = (CMD_STOP_LOGGING, block_id) + cf.send_packet( + pk, expected_reply=(CMD_STOP_LOGGING, block_id)) def delete(self): """Delete this entry in the Crazyflie""" @@ -628,6 +643,17 @@ def _find_block(self, id): return block return None + def _is_registered(self, logconf): + with self._registration_lock: + return (self._ids_ready and + logconf in self.log_blocks and + logconf.cf is self.cf and + logconf.id is not None) + + def _require_registered(self, logconf): + if not self._is_registered(logconf): + raise LogConfigError('Log configuration is not registered') + def _retire_config(self, logconf, block_id): with self._registration_lock: if (logconf not in self.log_blocks or diff --git a/test/crazyflie/test_log.py b/test/crazyflie/test_log.py index a2a05a8b6..1d5dad180 100644 --- a/test/crazyflie/test_log.py +++ b/test/crazyflie/test_log.py @@ -28,6 +28,7 @@ from cflib.crazyflie import Crazyflie from cflib.crazyflie.log import CHAN_SETTINGS +from cflib.crazyflie.log import CMD_CREATE_BLOCK from cflib.crazyflie.log import CMD_DELETE_BLOCK from cflib.crazyflie.log import CMD_RESET_LOGGING from cflib.crazyflie.log import Log @@ -291,6 +292,44 @@ def test_reset_can_be_retried_when_sending_fails(self): self.assertEqual(2, self.cf.send_packet.call_count) + def test_reset_waits_for_in_flight_start_command(self): + self.log.reset() + self._acknowledge(CMD_RESET_LOGGING) + self.log.toc = MagicMock() + self.log.toc.get_element_by_complete_name.return_value = MagicMock() + self.log.toc.get_element_id.return_value = 1 + config = LogConfig('config', 100) + config.add_variable('group.value', 'uint8_t') + self.log.add_config(config) + create_send_started = threading.Event() + allow_create_send = threading.Event() + reset_finished = threading.Event() + + def send_packet(packet, expected_reply): + if packet.data[0] == CMD_CREATE_BLOCK: + create_send_started.set() + allow_create_send.wait() + + def reset(): + self.log.reset() + reset_finished.set() + + self.cf.send_packet.side_effect = send_packet + start_thread = threading.Thread(target=config.start) + start_thread.start() + self.assertTrue(create_send_started.wait(1.0)) + reset_thread = threading.Thread(target=reset) + reset_thread.start() + + try: + self.assertFalse(reset_finished.wait(0.1)) + finally: + allow_create_send.set() + start_thread.join(1.0) + reset_thread.join(1.0) + self.assertFalse(start_thread.is_alive()) + self.assertFalse(reset_thread.is_alive()) + if __name__ == '__main__': unittest.main() From 62b635cfd72212aed7e893f185c79069f6a35b31 Mon Sep 17 00:00:00 2001 From: Arnaud Taffanel Date: Tue, 1 Sep 2026 11:56:08 +0200 Subject: [PATCH 5/7] Guard log commands across lifecycle transitions --- cflib/crazyflie/log.py | 36 ++++++++++++++------------- test/crazyflie/test_log.py | 51 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+), 17 deletions(-) diff --git a/cflib/crazyflie/log.py b/cflib/crazyflie/log.py index 9769b4f0b..b8e7b3cf0 100644 --- a/cflib/crazyflie/log.py +++ b/cflib/crazyflie/log.py @@ -53,6 +53,7 @@ import logging import struct from collections import deque +from contextlib import contextmanager from threading import Lock from .toc import Toc @@ -284,8 +285,7 @@ def create(self): if cf is None or self.id is None: raise LogConfigError( 'Log configuration must be added before it can be created') - with cf.log._command_lock: - cf.log._require_registered(self) + with cf.log._config_command(self): self._create() def _create(self): @@ -334,8 +334,7 @@ def start(self): if cf is None or self.id is None: raise LogConfigError( 'Log configuration must be added before it can be started') - with cf.log._command_lock: - cf.log._require_registered(self) + with cf.log._config_command(self): if (cf.link is not None): if (self._added is False): self._create() @@ -355,8 +354,8 @@ def stop(self): cf = self.cf if cf is None or self.id is None: return - with cf.log._command_lock: - if not cf.log._is_registered(self): + with cf.log._config_command(self, required=False) as registered: + if not registered: return block_id = self.id if (cf.link is not None): @@ -634,7 +633,8 @@ def _detach_all_configs(self, restore_ids, require_reset_pending=False): return True def _disconnected(self, uri): - self._detach_all_configs(restore_ids=False) + with self._command_lock: + self._detach_all_configs(restore_ids=False) def _find_block(self, id): with self._registration_lock: @@ -643,16 +643,18 @@ def _find_block(self, id): return block return None - def _is_registered(self, logconf): - with self._registration_lock: - return (self._ids_ready and - logconf in self.log_blocks and - logconf.cf is self.cf and - logconf.id is not None) - - def _require_registered(self, logconf): - if not self._is_registered(logconf): - raise LogConfigError('Log configuration is not registered') + @contextmanager + def _config_command(self, logconf, required=True): + with self._command_lock: + with self._registration_lock: + registered = (self._ids_ready and + logconf in self.log_blocks and + logconf.cf is self.cf and + logconf.id is not None and + not logconf._delete_pending) + if required and not registered: + raise LogConfigError('Log configuration is not registered') + yield registered def _retire_config(self, logconf, block_id): with self._registration_lock: diff --git a/test/crazyflie/test_log.py b/test/crazyflie/test_log.py index 1d5dad180..8d66783db 100644 --- a/test/crazyflie/test_log.py +++ b/test/crazyflie/test_log.py @@ -330,6 +330,57 @@ def reset(): self.assertFalse(start_thread.is_alive()) self.assertFalse(reset_thread.is_alive()) + def test_start_is_rejected_while_delete_is_pending(self): + self.log.reset() + self._acknowledge(CMD_RESET_LOGGING) + config = self._make_config('config') + self.log.add_config(config) + config.delete() + + with self.assertRaises(LogConfigError): + config.start() + + self.assertEqual(2, self.cf.send_packet.call_count) + + def test_disconnect_waits_for_in_flight_start_command(self): + self.log.reset() + self._acknowledge(CMD_RESET_LOGGING) + self.log.toc = MagicMock() + self.log.toc.get_element_by_complete_name.return_value = MagicMock() + self.log.toc.get_element_id.return_value = 1 + config = LogConfig('config', 100) + config.add_variable('group.value', 'uint8_t') + self.log.add_config(config) + create_send_started = threading.Event() + allow_create_send = threading.Event() + disconnect_finished = threading.Event() + + def send_packet(packet, expected_reply): + if packet.data[0] == CMD_CREATE_BLOCK: + create_send_started.set() + allow_create_send.wait() + + def disconnect(): + self.cf.disconnected.call('radio://test') + disconnect_finished.set() + + self.cf.send_packet.side_effect = send_packet + start_thread = threading.Thread(target=config.start) + start_thread.start() + self.assertTrue(create_send_started.wait(1.0)) + disconnect_thread = threading.Thread(target=disconnect) + disconnect_thread.start() + + try: + self.assertFalse(disconnect_finished.wait(0.1)) + finally: + allow_create_send.set() + start_thread.join(1.0) + disconnect_thread.join(1.0) + self.assertFalse(start_thread.is_alive()) + self.assertFalse(disconnect_thread.is_alive()) + self.assertIsNone(config.id) + if __name__ == '__main__': unittest.main() From 7854889e8d72bd4fc017fba5b4a27852039eae85 Mon Sep 17 00:00:00 2001 From: Arnaud Taffanel Date: Tue, 1 Sep 2026 11:59:06 +0200 Subject: [PATCH 6/7] Handle synchronous log disconnects --- cflib/crazyflie/log.py | 16 +++++++++++----- test/crazyflie/test_log.py | 29 +++++++++++++++++++++++++++++ 2 files changed, 40 insertions(+), 5 deletions(-) diff --git a/cflib/crazyflie/log.py b/cflib/crazyflie/log.py index b8e7b3cf0..ae23a7845 100644 --- a/cflib/crazyflie/log.py +++ b/cflib/crazyflie/log.py @@ -55,6 +55,7 @@ from collections import deque from contextlib import contextmanager from threading import Lock +from threading import RLock from .toc import Toc from .toc import TocFetcher @@ -289,13 +290,15 @@ def create(self): self._create() def _create(self): + cf = self.cf + block_id = self.id command = self._cmd_create_block() next_to_add = 0 is_done = False num_variables = 0 pending = 0 - for block in self.cf.log.log_blocks: + for block in cf.log.log_blocks: if block.pending or block.added or block.started: pending += 1 num_variables += len(block._get_effective_variables()) @@ -319,11 +322,14 @@ def _create(self): while not is_done: pk = CRTPPacket() pk.set_header(5, CHAN_SETTINGS) - pk.data = (command, self.id) + pk.data = (command, block_id) is_done, next_to_add = self._setup_log_elements(pk, next_to_add) - logger.debug('Adding/appending log block id {}'.format(self.id)) - self.cf.send_packet(pk, expected_reply=(command, self.id)) + logger.debug('Adding/appending log block id {}'.format(block_id)) + cf.send_packet(pk, expected_reply=(command, block_id)) + if self.cf is not cf or self.id != block_id: + raise LogConfigError( + 'Log configuration was detached while being created') # Use append if we have to add more variables command = self._cmd_append_block() @@ -488,7 +494,7 @@ def __init__(self, crazyflie=None): self._toc_cache = None self._registration_lock = Lock() - self._command_lock = Lock() + self._command_lock = RLock() self._available_config_ids = deque() self._ids_ready = False self._reset_pending = False diff --git a/test/crazyflie/test_log.py b/test/crazyflie/test_log.py index 8d66783db..4fd4be5c1 100644 --- a/test/crazyflie/test_log.py +++ b/test/crazyflie/test_log.py @@ -381,6 +381,35 @@ def disconnect(): self.assertFalse(disconnect_thread.is_alive()) self.assertIsNone(config.id) + def test_synchronous_disconnect_during_start_does_not_deadlock(self): + self.log.reset() + self._acknowledge(CMD_RESET_LOGGING) + self.log.toc = MagicMock() + self.log.toc.get_element_by_complete_name.return_value = MagicMock() + self.log.toc.get_element_id.return_value = 1 + config = LogConfig('config', 100) + config.add_variable('group.value', 'uint8_t') + self.log.add_config(config) + errors = [] + + def send_packet(packet, expected_reply): + self.cf.disconnected.call('radio://test') + + def start(): + try: + config.start() + except LogConfigError as error: + errors.append(error) + + self.cf.send_packet.side_effect = send_packet + start_thread = threading.Thread(target=start, daemon=True) + start_thread.start() + start_thread.join(1.0) + + self.assertFalse(start_thread.is_alive()) + self.assertEqual(1, len(errors)) + self.assertIsNone(config.id) + if __name__ == '__main__': unittest.main() From 510c4ab2c651b9561d79063f12c32771120f1cd9 Mon Sep 17 00:00:00 2001 From: Arnaud Taffanel Date: Tue, 1 Sep 2026 12:01:46 +0200 Subject: [PATCH 7/7] Validate log registration after sends --- cflib/crazyflie/log.py | 21 ++++++++++------ test/crazyflie/test_log.py | 50 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 8 deletions(-) diff --git a/cflib/crazyflie/log.py b/cflib/crazyflie/log.py index ae23a7845..1c6ee6c3d 100644 --- a/cflib/crazyflie/log.py +++ b/cflib/crazyflie/log.py @@ -327,9 +327,9 @@ def _create(self): logger.debug('Adding/appending log block id {}'.format(block_id)) cf.send_packet(pk, expected_reply=(command, block_id)) - if self.cf is not cf or self.id != block_id: + if not cf.log._is_current_registration(self, cf, block_id): raise LogConfigError( - 'Log configuration was detached while being created') + 'Log configuration changed while being created') # Use append if we have to add more variables command = self._cmd_append_block() @@ -652,16 +652,21 @@ def _find_block(self, id): @contextmanager def _config_command(self, logconf, required=True): with self._command_lock: - with self._registration_lock: - registered = (self._ids_ready and - logconf in self.log_blocks and - logconf.cf is self.cf and - logconf.id is not None and - not logconf._delete_pending) + registered = self._is_current_registration( + logconf, self.cf, logconf.id) if required and not registered: raise LogConfigError('Log configuration is not registered') yield registered + def _is_current_registration(self, logconf, cf, block_id): + with self._registration_lock: + return (self._ids_ready and + logconf in self.log_blocks and + logconf.cf is cf and + logconf.id == block_id and + block_id is not None and + not logconf._delete_pending) + def _retire_config(self, logconf, block_id): with self._registration_lock: if (logconf not in self.log_blocks or diff --git a/test/crazyflie/test_log.py b/test/crazyflie/test_log.py index 4fd4be5c1..eae26789e 100644 --- a/test/crazyflie/test_log.py +++ b/test/crazyflie/test_log.py @@ -29,6 +29,7 @@ from cflib.crazyflie import Crazyflie from cflib.crazyflie.log import CHAN_SETTINGS from cflib.crazyflie.log import CMD_CREATE_BLOCK +from cflib.crazyflie.log import CMD_CREATE_BLOCK_V2 from cflib.crazyflie.log import CMD_DELETE_BLOCK from cflib.crazyflie.log import CMD_RESET_LOGGING from cflib.crazyflie.log import Log @@ -61,6 +62,17 @@ def _make_config(self, name): config.add_memory('value', 'uint8_t', 'uint8_t', 0x1000) return config + def _make_multi_packet_config(self): + self.log._useV2 = True + self.log.toc = MagicMock() + self.log.toc.get_element_by_complete_name.return_value = MagicMock() + self.log.toc.get_element_id.return_value = 1 + config = LogConfig('multi-packet', 100) + for i in range(20): + config.add_variable('group.value{}'.format(i), 'uint8_t') + self.log.add_config(config) + return config + def test_all_byte_values_are_available_as_log_config_ids(self): self.log.reset() self._acknowledge(CMD_RESET_LOGGING) @@ -410,6 +422,44 @@ def start(): self.assertEqual(1, len(errors)) self.assertIsNone(config.id) + def test_synchronous_reset_during_create_prevents_append(self): + self.log.reset() + self._acknowledge(CMD_RESET_LOGGING) + config = self._make_multi_packet_config() + sent_commands = [] + + def send_packet(packet, expected_reply): + sent_commands.append(packet.data[0]) + if packet.data[0] == CMD_CREATE_BLOCK_V2: + self.log.reset() + + self.cf.send_packet.side_effect = send_packet + + with self.assertRaises(LogConfigError): + config.start() + + self.assertEqual( + [CMD_CREATE_BLOCK_V2, CMD_RESET_LOGGING], sent_commands) + + def test_synchronous_delete_during_create_prevents_append(self): + self.log.reset() + self._acknowledge(CMD_RESET_LOGGING) + config = self._make_multi_packet_config() + sent_commands = [] + + def send_packet(packet, expected_reply): + sent_commands.append(packet.data[0]) + if packet.data[0] == CMD_CREATE_BLOCK_V2: + config.delete() + + self.cf.send_packet.side_effect = send_packet + + with self.assertRaises(LogConfigError): + config.start() + + self.assertEqual( + [CMD_CREATE_BLOCK_V2, CMD_DELETE_BLOCK], sent_commands) + if __name__ == '__main__': unittest.main()