feat(interfaces): queue redis writes by default at 0.1s interval

docs.openc3.com/docs/configuration/interfaces.md

@@ -131,7 +131,7 @@ INTERFACE INTERFACE_NAME tcpip_client_interface.rb host.docker.internal 8080 808
 
 #### UPDATE_INTERVAL
 
-Number of seconds to wait before writing packets to Redis. By default packets are written to Redis every second (both the Current Value Table (CVT) and the stream). If you have high rate data and you want faster updates you can lower this value. Note that this will increase processor utilization! This option was previously known as `OPTIMIZE_THROUGHPUT`.
+Number of seconds to wait before writing packets to Redis. Interfaces queue the raw packet stream writes they make to Redis and flush them in a single pipelined batch every 0.1 seconds by default. Decommutated telemetry and the Current Value Table (CVT) are written separately by the decom microservice and are not affected by this option. Raising this value reduces processor utilization for high rate data at the cost of slower updates. Lowering it gives faster updates but increases processor utilization. Setting it to 0 disables raw packet stream queuing and writes every packet to Redis immediately, which is the slowest option for high rate data. This option was previously known as `OPTIMIZE_THROUGHPUT`.
 
 <Tabs groupId="script-language">
 <TabItem value="python" label="Python">

openc3/lib/openc3/microservices/interface_microservice.rb

@@ -558,6 +558,7 @@ def run
 
   class InterfaceMicroservice < Microservice
     UNKNOWN_BYTES_TO_PRINT = 16
+    DEFAULT_UPDATE_INTERVAL = 0.1 # seconds
 
     # Interfaces never use descriptions so don't keep them in memory
     def load_descriptions?
@@ -599,19 +600,24 @@ def initialize(name)
         RouterStatusModel.set(@interface.as_json(), scope: @scope)
       end
 
-      @queued = false
+      # Queue Redis writes by default and flush them every update_interval seconds
+      update_interval = DEFAULT_UPDATE_INTERVAL
       @interface.options.each do |option_name, option_values|
         # OPTIMIZE_THROUGHPUT was changed to UPDATE_INTERVAL to better represent the setting
         if option_name.upcase == 'UPDATE_INTERVAL' or option_name.upcase == 'OPTIMIZE_THROUGHPUT'
-          @queued = true
           update_interval = option_values[0].to_f
-          EphemeralStoreQueued.instance.set_update_interval(update_interval)
-          StoreQueued.instance.set_update_interval(update_interval)
         end
         if option_name.upcase == 'SYNC_PACKET_COUNT_DELAY_SECONDS'
           TargetModel.sync_packet_count_delay_seconds = option_values[0].to_f
         end
       end
+      # An UPDATE_INTERVAL of 0 (or less) disables queuing and writes directly to Redis
+      @queued = update_interval > 0.0
+      if @queued
+        # Apply to every db_shard since writes are routed to the shard of each target
+        EphemeralStoreQueued.set_update_interval_all_shards(update_interval)
+        StoreQueued.set_update_interval_all_shards(update_interval)
+      end
 
       @interface_thread_sleeper = Sleeper.new
       @cancel_thread = false

openc3/lib/openc3/utilities/store_queued.rb

@@ -24,20 +24,32 @@ class StoreQueued
     # Mutex used to ensure that only one instance is created
     @@instance_mutex = Mutex.new
 
+    DEFAULT_UPDATE_INTERVAL = 1 # seconds
+
     # Get the singleton instance for the given db_shard
-    # Sets the update interval to 1 second by default
-    def self.instance(update_interval = 1, db_shard: 0) # seconds
+    # Uses the interval from set_update_interval_all_shards if not given
+    def self.instance(update_interval = nil, db_shard: 0)
       @instances ||= []
       the_instance = @instances[db_shard]
       return the_instance if the_instance
 
       @@instance_mutex.synchronize do
         @instances ||= []
-        @instances[db_shard] ||= self.new(update_interval, db_shard: db_shard)
+        @instances[db_shard] ||= self.new(update_interval || @update_interval || DEFAULT_UPDATE_INTERVAL, db_shard: db_shard)
         return @instances[db_shard]
       end
     end
 
+    # Set the update interval on every db_shard instance, including ones created later
+    def self.set_update_interval_all_shards(interval)
+      return unless interval > 0.0
+
+      @@instance_mutex.synchronize do
+        @update_interval = interval
+        (@instances || []).compact.each { |the_instance| the_instance.set_update_interval(interval) }
+      end
+    end
+
     # Delegate all unknown class methods to delegate to the instance (db_shard 0)
     def self.method_missing(message, *args, **kwargs, &)
       self.instance.public_send(message, *args, **kwargs, &)

openc3/python/openc3/microservices/interface_microservice.py

@@ -643,6 +643,7 @@ def run(self):
 class InterfaceMicroservice(Microservice):
     UNKNOWN_BYTES_TO_PRINT = 16
     DISCONNECT_WAIT_TIME = 1
+    DEFAULT_UPDATE_INTERVAL = 0.1  # seconds
 
     # Interfaces never use descriptions so don't keep them in memory
     def load_descriptions(self):
@@ -686,16 +687,20 @@ def __init__(self, name):
         else:
             RouterStatusModel.set(self.interface.as_json(), scope=self.scope)
 
-        self.queued = False
+        # Queue Redis writes by default and flush them every update_interval seconds
+        update_interval = InterfaceMicroservice.DEFAULT_UPDATE_INTERVAL
         for option_name, option_values in self.interface.options.items():
             # OPTIMIZE_THROUGHPUT was changed to UPDATE_INTERVAL to better represent the setting
             if option_name.upper() == "UPDATE_INTERVAL" or option_name.upper() == "OPTIMIZE_THROUGHPUT":
-                self.queued = True
                 update_interval = float(option_values[0])
-                EphemeralStoreQueued.instance().set_update_interval(update_interval)
-                StoreQueued.instance().set_update_interval(update_interval)
             if option_name.upper() == "SYNC_PACKET_COUNT_DELAY_SECONDS":
                 TargetModel.sync_packet_count_delay_seconds = float(option_values[0])
+        # An UPDATE_INTERVAL of 0 (or less) disables queuing and writes directly to Redis
+        self.queued = update_interval > 0.0
+        if self.queued:
+            # Apply to every db_shard since writes are routed to the shard of each target
+            EphemeralStoreQueued.set_update_interval_all_shards(update_interval)
+            StoreQueued.set_update_interval_all_shards(update_interval)
 
         if self.interface_or_router == "INTERFACE":
             self.handler_thread = InterfaceCmdHandlerThread(

openc3/python/openc3/utilities/store_implementation.py

@@ -65,6 +65,9 @@ class StoreMeta(type):
             "instance",
             "instance_mutex",
             "my_instances",
+            "set_update_interval_all_shards",
+            "all_shards_update_interval",
+            "DEFAULT_UPDATE_INTERVAL",
             "db_shard_for_target",
             "_db_shard_cache",
             "_db_shard_cache_lock",

openc3/python/openc3/utilities/store_queued.py

@@ -40,18 +40,36 @@ class StoreQueued(metaclass=StoreMeta):
     # Mutex used to ensure that only one instance is created
     instance_mutex = threading.Lock()
 
+    DEFAULT_UPDATE_INTERVAL = 1  # seconds
+
+    # Interval from set_update_interval_all_shards, used for instances created later
+    all_shards_update_interval = None
+
     # Get the singleton instance for a given db_shard
+    # Uses the interval from set_update_interval_all_shards if not given
     @classmethod
-    def instance(cls, update_interval=1, db_shard=0):
+    def instance(cls, update_interval=None, db_shard=0):
         inst = cls.my_instances.get(db_shard)
         if inst:
             return inst
 
         with cls.instance_mutex:
             if db_shard not in cls.my_instances:
+                if update_interval is None:
+                    update_interval = cls.all_shards_update_interval or cls.DEFAULT_UPDATE_INTERVAL
                 cls.my_instances[db_shard] = cls(update_interval, db_shard=db_shard)
             return cls.my_instances[db_shard]
 
+    # Set the update interval on every db_shard instance, including ones created later
+    @classmethod
+    def set_update_interval_all_shards(cls, interval):
+        if interval <= 0.0:
+            return
+        with cls.instance_mutex:
+            cls.all_shards_update_interval = interval
+            for inst in cls.my_instances.values():
+                inst.set_update_interval(interval)
+
     def __init__(self, update_interval, db_shard=0):
         self.is_shutdown = False
         self.update_interval = update_interval
@@ -143,5 +161,8 @@ class EphemeralStoreQueued(StoreQueued):
     # Variable that holds the singleton instances per db_shard
     my_instances = {}
 
+    # Interval from set_update_interval_all_shards, used for instances created later
+    all_shards_update_interval = None
+
     def store_instance(self):
         return EphemeralStore.instance(db_shard=self.db_shard)

openc3/python/test/microservices/test_interface_microservice.py

@@ -1079,6 +1079,77 @@ def test_supports_optimize_throughput_option_for_backward_compatibility(self):
         im.shutdown()
         time.sleep(0.1)  # Allow threads to exit
 
+    def test_queues_writes_by_default_with_default_update_interval(self):
+        model = InterfaceModel(
+            name="INST_INT",
+            scope="DEFAULT",
+            target_names=["INST"],
+            cmd_target_names=["INST"],
+            tlm_target_names=["INST"],
+            config_params=["test_interface.py"],
+            options=[],
+        )
+        model.update()
+
+        # Set a different interval to verify the default gets applied
+        StoreQueued.instance().set_update_interval(1)
+        EphemeralStoreQueued.instance().set_update_interval(1)
+
+        im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT")
+        self.addCleanup(im.shutdown)
+        self.assertTrue(im.queued)
+        self.assertEqual(StoreQueued.instance().update_interval, InterfaceMicroservice.DEFAULT_UPDATE_INTERVAL)
+        self.assertEqual(EphemeralStoreQueued.instance().update_interval, InterfaceMicroservice.DEFAULT_UPDATE_INTERVAL)
+        self.assertEqual(InterfaceMicroservice.DEFAULT_UPDATE_INTERVAL, 0.1)
+
+        im.shutdown()
+        time.sleep(0.1)  # Allow threads to exit
+
+    def test_update_interval_zero_disables_queued_writes(self):
+        model = InterfaceModel(
+            name="INST_INT",
+            scope="DEFAULT",
+            target_names=["INST"],
+            cmd_target_names=["INST"],
+            tlm_target_names=["INST"],
+            config_params=["test_interface.py"],
+            options=[["UPDATE_INTERVAL", "0"]],
+        )
+        model.update()
+
+        im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT")
+        self.addCleanup(im.shutdown)
+        self.assertFalse(im.queued)
+
+        im.shutdown()
+        time.sleep(0.1)  # Allow threads to exit
+
+    def test_applies_update_interval_to_every_db_shard(self):
+        model = InterfaceModel(
+            name="INST_INT",
+            scope="DEFAULT",
+            target_names=["INST"],
+            cmd_target_names=["INST"],
+            tlm_target_names=["INST"],
+            config_params=["test_interface.py"],
+            options=[["UPDATE_INTERVAL", "0.25"]],
+        )
+        model.update()
+        existing_store = StoreQueued.instance(db_shard=2)
+        existing_ephemeral = EphemeralStoreQueued.instance(db_shard=2)
+
+        im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT")
+        self.addCleanup(im.shutdown)
+        # Instances that already existed
+        self.assertEqual(existing_store.update_interval, 0.25)
+        self.assertEqual(existing_ephemeral.update_interval, 0.25)
+        # Instances first used after startup, e.g. a target on another shard
+        self.assertEqual(StoreQueued.instance(db_shard=1).update_interval, 0.25)
+        self.assertEqual(EphemeralStoreQueued.instance(db_shard=1).update_interval, 0.25)
+
+        im.shutdown()
+        time.sleep(0.1)  # Allow threads to exit
+
 
 class TestInterfaceCmdHandlerRoutedCommands(unittest.TestCase):
     def setUp(self):

openc3/python/test/test_helper.py

@@ -168,6 +168,8 @@ def mock_redis(self):
     Store._db_shard_cache = {}
     EphemeralStoreQueued.my_instances = {}
     StoreQueued.my_instances = {}
+    EphemeralStoreQueued.all_shards_update_interval = None
+    StoreQueued.all_shards_update_interval = None
     return redis
 
 

openc3/python/test/topics/test_telemetry_topic.py

@@ -0,0 +1,48 @@
+# Copyright 2026 OpenC3, Inc.
+# All Rights Reserved.
+#
+# 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 LICENSE.md for more details.
+#
+# This file may also be used under the terms of a commercial license
+# if purchased from OpenC3, Inc.
+
+import datetime
+import unittest
+from unittest.mock import MagicMock, patch
+
+from openc3.packets.packet import Packet
+from openc3.topics.telemetry_topic import TelemetryTopic
+from test.test_helper import mock_redis
+
+
+class TestTelemetryTopic(unittest.TestCase):
+    def setUp(self):
+        mock_redis(self)
+        shard_patch = patch("openc3.topics.telemetry_topic.Store.db_shard_for_target", return_value=3)
+        shard_patch.start()
+        self.addCleanup(shard_patch.stop)
+
+        self.packet = Packet("TARGET", "PACKET")
+        self.packet.received_time = datetime.datetime.now(datetime.timezone.utc)
+        self.packet.packet_time = datetime.datetime.now(datetime.timezone.utc)
+        self.packet.received_count = 1
+
+    @patch("openc3.topics.telemetry_topic.EphemeralStoreQueued.instance")
+    def test_writes_to_queued_store_for_target_shard(self, instance):
+        instance.return_value = MagicMock()
+        TelemetryTopic.write_packet(self.packet, queued=True, scope="DEFAULT")
+        instance.assert_called_once_with(db_shard=3)
+        topic, msg_hash = instance.return_value.write_topic.call_args.args
+        self.assertEqual(topic, "DEFAULT__TELEMETRY__{TARGET}__PACKET")
+        self.assertEqual(msg_hash["target_name"], "TARGET")
+
+    @patch("openc3.topics.telemetry_topic.Topic.write_topic")
+    @patch("openc3.topics.telemetry_topic.EphemeralStoreQueued.instance")
+    def test_writes_directly_to_target_shard_when_not_queued(self, instance, write_topic):
+        TelemetryTopic.write_packet(self.packet, scope="DEFAULT")
+        instance.assert_not_called()
+        self.assertEqual(write_topic.call_args.args[0], "DEFAULT__TELEMETRY__{TARGET}__PACKET")
+        self.assertEqual(write_topic.call_args.kwargs["db_shard"], 3)

openc3/python/test/utilities/test_store_queued.py

@@ -15,7 +15,8 @@
 import unittest
 from unittest.mock import MagicMock
 
-import test.test_helper  # noqa: F401
+from openc3.utilities.store_queued import EphemeralStoreQueued, StoreQueued
+from test.test_helper import mock_redis
 
 
 def load_real_store_queued():
@@ -70,5 +71,26 @@ def test_shutdown_drains_before_writing_through(self):
         self.assertTrue(self.store_queued.is_shutdown)
 
 
+class TestStoreQueuedUpdateIntervalAllShards(unittest.TestCase):
+    def setUp(self):
+        mock_redis(self)
+
+    def test_updates_existing_instances_and_later_ones(self):
+        existing = StoreQueued.instance(db_shard=0)
+        self.assertEqual(existing.update_interval, 1)
+        StoreQueued.set_update_interval_all_shards(0.25)
+        self.assertEqual(existing.update_interval, 0.25)
+        self.assertEqual(StoreQueued.instance(db_shard=1).update_interval, 0.25)
+        # Separate setting per class
+        self.assertEqual(EphemeralStoreQueued.instance(db_shard=1).update_interval, 1)
+        # An explicit interval still wins for a new instance
+        self.assertEqual(StoreQueued.instance(update_interval=0.5, db_shard=2).update_interval, 0.5)
+
+    def test_ignores_intervals_of_zero_or_less(self):
+        StoreQueued.set_update_interval_all_shards(0.25)
+        StoreQueued.set_update_interval_all_shards(0)
+        self.assertEqual(StoreQueued.instance().update_interval, 0.25)
+
+
 if __name__ == "__main__":
     unittest.main()