State, subscriptions, and real-time control
Snapshots read once, subscriptions pull periodically, and Servo sessions send continuous targets.
| API | Direction | Use |
|---|---|---|
state.snapshot() | Device to application | Preflight and post-command verification |
state.subscribe() | Device to application | Continuous state monitoring and logging |
touch.subscribe() | Device to application | Tactile monitoring and algorithm input |
health.subscribe() | Device to application | Runtime and fault monitoring |
open_servo() | Application to device | Continuous position or impedance targets |
Subscription lifecycle
subscription = hand.state.subscribe(period=0.02)
try:
while True:
state = await subscription.next()
process_state(state)
finally:
subscription.close()Subscriptions do not retain history. Use bounded queues and define a drop or downsampling policy when consumers are slower than acquisition; do not allow telemetry to accumulate without a limit.
Bounded queues and drop policies
Size each queue from the subscription period, the consumer's worst-case processing time, and the maximum acceptable buffering latency. For example, a 20 ms State period with at most 640 ms of buffering gives ceil(0.64 / 0.02) = 32. The value 32 is an example calculation, not a fixed recommendation for every device or workload.
| Data use | Isolation | Default policy when full | Requirement |
|---|---|---|---|
| State control input or UI | Dedicated small queue | Drop the oldest frame and retain the latest state | Count drops and detect gaps from timestamps |
| Real-time Touch processing | Bounded queue separate from State | Drop the oldest frame or downsample at a declared ratio | Detect gaps with sequence; do not share the Health queue |
| Health and faults | Dedicated high-priority queue or event channel | Never silently drop fault onset, recovery, or safety-state transitions | Alert and enter the application's safe path on overflow |
| Raw data recording | Dedicated writer and persistent buffer | Do not silently drop by default | If backpressure is impossible, alert, stop recording, or mark the gap explicitly |
Do not put all telemetry in one queue
High-rate State or Touch samples can keep a shared queue full. Health, fault, and safety-state changes require a separate channel so ordinary telemetry cannot evict them.
The following Python and C++ examples use the same drop-oldest policy and track received, processed, dropped, and max_depth:
import asyncio
from dataclasses import dataclass
@dataclass
class QueueMetrics:
received: int = 0
processed: int = 0
dropped: int = 0
max_depth: int = 0
def offer_latest(
queue: asyncio.Queue,
sample,
metrics: QueueMetrics,
) -> None:
metrics.received += 1
if queue.full():
queue.get_nowait() # Drop the oldest State sample.
queue.task_done()
metrics.dropped += 1
queue.put_nowait(sample)
metrics.max_depth = max(metrics.max_depth, queue.qsize())
async def forward_states(hand) -> None:
queue = asyncio.Queue(maxsize=32)
metrics = QueueMetrics()
subscription = hand.state.subscribe(period=0.02)
async def read_subscription() -> None:
while True:
state = await subscription.next()
offer_latest(queue, state, metrics)
async def process_states() -> None:
while True:
state = await queue.get()
try:
# Replace this with the application's non-blocking handoff.
print(state.timestamp, state.positions_deg)
metrics.processed += 1
finally:
queue.task_done()
tasks = [
asyncio.create_task(read_subscription()),
asyncio.create_task(process_states()),
]
try:
await asyncio.gather(*tasks)
finally:
subscription.close()
for task in tasks:
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
print(metrics)#include <revo3/revo3.hpp>
#include <algorithm>
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <cstdio>
#include <deque>
#include <exception>
#include <mutex>
#include <optional>
#include <stdexcept>
#include <thread>
#include <utility>
struct QueueMetrics {
std::size_t received = 0;
std::size_t processed = 0;
std::size_t dropped = 0;
std::size_t max_depth = 0;
};
template <typename T>
class BoundedLatestQueue {
public:
explicit BoundedLatestQueue(std::size_t capacity) : capacity_(capacity) {
if (capacity_ == 0) {
throw std::invalid_argument("queue capacity must be positive");
}
}
bool push_latest(T value) {
std::lock_guard<std::mutex> lock(mutex_);
if (closed_) {
return false;
}
++metrics_.received;
if (queue_.size() == capacity_) {
queue_.pop_front();
++metrics_.dropped;
}
queue_.push_back(std::move(value));
metrics_.max_depth = std::max(metrics_.max_depth, queue_.size());
ready_.notify_one();
return true;
}
std::optional<T> pop() {
std::unique_lock<std::mutex> lock(mutex_);
ready_.wait(lock, [this] { return closed_ || !queue_.empty(); });
if (queue_.empty()) {
return std::nullopt;
}
T value = std::move(queue_.front());
queue_.pop_front();
return value;
}
void mark_processed() {
std::lock_guard<std::mutex> lock(mutex_);
++metrics_.processed;
}
void close() {
{
std::lock_guard<std::mutex> lock(mutex_);
closed_ = true;
}
ready_.notify_all();
}
QueueMetrics metrics() const {
std::lock_guard<std::mutex> lock(mutex_);
return metrics_;
}
private:
const std::size_t capacity_;
mutable std::mutex mutex_;
std::condition_variable ready_;
std::deque<T> queue_;
QueueMetrics metrics_;
bool closed_ = false;
};
int main() {
revo3::Manager manager;
auto hand = manager.connect_auto();
auto subscription = hand.state().subscribe(std::chrono::milliseconds(20));
using State = decltype(subscription.next());
BoundedLatestQueue<State> queue(32);
std::exception_ptr reader_error;
std::thread reader([&] {
try {
for (std::size_t index = 0; index < 500; ++index) {
if (!queue.push_latest(subscription.next())) {
break;
}
}
} catch (...) {
reader_error = std::current_exception();
}
queue.close();
});
std::thread worker([&] {
while (auto state = queue.pop()) {
std::printf("timestamp=%lld.%09lld J0=%.2f degree\n",
static_cast<long long>(state->timestamp.sec),
static_cast<long long>(state->timestamp.nsec),
state->motors.positions_deg[0]);
queue.mark_processed();
}
});
reader.join();
worker.join();
subscription.close();
const auto metrics = queue.metrics();
std::printf("received=%zu processed=%zu dropped=%zu max_depth=%zu\n",
metrics.received, metrics.processed,
metrics.dropped, metrics.max_depth);
if (reader_error) {
std::rethrow_exception(reader_error);
}
return 0;
}Python: offer_latest() is appropriate for State/UI/control consumers that only need the latest value. It is not appropriate for lossless logging. Move blocking file, database, or network writes to a worker thread or separate process. If the recording path cannot keep up, alert or stop recording instead of silently applying the drop-oldest policy.
C++: The finite read count lets the example terminate naturally. A long-running service should close the subscription and queue after receiving its stop signal, then join() both threads. When callback-backed data crosses threads, copy the required fields according to the lifetime documented by the SDK. Calling close() wakes a waiting consumer, and a reader exception is rethrown after both threads finish.
Do not reuse this BoundedLatestQueue for Health and fault events. Use a separate queue or event channel that preserves fault onset, recovery, and safety-state transitions. On overflow, enter an alert or safe-stop path instead of calling pop_front().
In every language, make the drop policy application configuration and expose it through logs or monitoring; do not leave it as implicit queue behavior.
Servo session
A Servo session owns control state. Complete layout and Health checks before opening it, define send timing and command timeout, monitor telemetry without assuming its rate equals the control rate, and close the session on every exit. No unconditional fixed frequency is guaranteed across hosts, adapters, bus load, and firmware.