brokapi_runtime/
instrument_queue.rs

1//! Per-instrument command serialization (ordered_command_channel: false).
2//!
3//! Commands on one instrument run one at a time, in arrival order (tokio mutexes are FIFO):
4//! a cancel followed by a place cannot be reordered on the wire. The guard is RAII, so the
5//! instrument is released on every exit path, including errors and dropped futures.
6
7use std::collections::HashMap;
8use std::sync::{Arc, Mutex};
9
10use brokapi_core::{ClientOrderId, InstrumentId, InstrumentKey};
11use tokio::sync::{Mutex as AsyncMutex, OwnedMutexGuard};
12
13/// Held while a command on the instrument is on the wire.
14pub type InstrumentGuard = OwnedMutexGuard<()>;
15
16/// Per-instrument command serialization: one trading command at a time on each instrument.
17///
18/// The lock is keyed by [`InstrumentKey`] (symbol and mic), not by the whole [`InstrumentId`]: the
19/// same instrument reached through a ticker, a uid or a figi is still one instrument. Waiters are
20/// served in the order they asked (the lock is FIFO), so a cancel sent before a place reaches the
21/// broker before it. Why: the broker's command channel is not ordered, and a place that overtakes
22/// a cancel would act on a different book than the caller intended.
23///
24/// Different instruments do not wait for each other. The lock is released when the guard is dropped,
25/// on every exit path, including errors and cancelled futures.
26///
27/// [`bind`](Self::bind) and [`instrument_of`](Self::instrument_of) remember which instrument a cid
28/// belongs to, for commands that carry only a cid (cancel, replace).
29///
30/// # Examples
31///
32/// ```
33/// use brokapi_core::{ClientOrderId, InstrumentId, InstrumentKey, NativeId};
34/// use brokapi_runtime::InstrumentLocks;
35///
36/// # #[tokio::main(flavor = "current_thread")]
37/// # async fn main() {
38/// let sber = InstrumentId::new(
39///     InstrumentKey::new("SBER", "TQBR"),
40///     NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
41/// );
42/// let locks = InstrumentLocks::new();
43/// let cid = ClientOrderId::new_v4();
44/// locks.bind(cid, &sber);
45///
46/// // A place holds the instrument while its command is on the wire.
47/// let guard = locks.lock(&sber).await;
48/// assert!(locks.is_busy(&sber));
49/// drop(guard);
50/// assert!(!locks.is_busy(&sber));
51///
52/// // A cancel carries only the cid. It finds the instrument and takes the same lock.
53/// let guard = locks.lock_cid(cid).await;
54/// assert!(guard.is_some() && locks.is_busy(&sber));
55/// # }
56/// ```
57#[derive(Debug, Default)]
58pub struct InstrumentLocks {
59    locks: Mutex<HashMap<InstrumentKey, Arc<AsyncMutex<()>>>>,
60    by_cid: Mutex<Bound>,
61}
62
63/// The cid -> instrument map with the order the cids came in, so that the oldest can go.
64#[derive(Debug, Default)]
65struct Bound {
66    map: HashMap<ClientOrderId, InstrumentId>,
67    order: std::collections::VecDeque<ClientOrderId>,
68}
69
70/// How many cids [`InstrumentLocks`] remembers. The oldest are forgotten first: a cancel of such an old
71/// order is no longer serialized with the other commands of its instrument.
72pub const BOUND_CIDS: usize = 100_000;
73
74impl InstrumentLocks {
75    /// Creates an empty set of locks.
76    pub fn new() -> Self {
77        Self::default()
78    }
79
80    /// Waits for earlier commands on `instrument`, then holds it until the guard drops.
81    pub async fn lock(&self, instrument: &InstrumentId) -> InstrumentGuard {
82        let lock = {
83            let mut map = self.locks.lock().expect("instrument locks poisoned");
84            Arc::clone(map.entry(instrument.key.clone()).or_default())
85        };
86        lock.lock_owned().await
87    }
88
89    /// Whether a command on the instrument is on the wire right now. For tests and diagnostics: the
90    /// answer can be stale by the time the caller acts on it.
91    pub fn is_busy(&self, instrument: &InstrumentId) -> bool {
92        let map = self.locks.lock().expect("instrument locks poisoned");
93        map.get(&instrument.key)
94            .is_some_and(|l| l.try_lock().is_err())
95    }
96
97    /// Remembers the instrument of an order we placed.
98    pub fn bind(&self, cid: ClientOrderId, instrument: &InstrumentId) {
99        let mut b = self.by_cid.lock().expect("instrument locks poisoned");
100        if b.map.insert(cid, instrument.clone()).is_none() {
101            b.order.push_back(cid);
102            while b.order.len() > BOUND_CIDS {
103                if let Some(old) = b.order.pop_front() {
104                    b.map.remove(&old);
105                }
106            }
107        }
108    }
109
110    /// Forgets the instrument of a cid, for an order that is long finished. A later cancel of that cid is
111    /// no longer serialized with the other commands of its instrument.
112    pub fn forget(&self, cid: ClientOrderId) {
113        self.by_cid
114            .lock()
115            .expect("instrument locks poisoned")
116            .map
117            .remove(&cid);
118    }
119
120    /// How many cids are bound to an instrument. For tests and metrics.
121    pub fn bound(&self) -> usize {
122        self.by_cid
123            .lock()
124            .expect("instrument locks poisoned")
125            .map
126            .len()
127    }
128
129    /// The instrument of a cid, if this process placed (or replaced) it.
130    pub fn instrument_of(&self, cid: ClientOrderId) -> Option<InstrumentId> {
131        self.by_cid
132            .lock()
133            .expect("instrument locks poisoned")
134            .map
135            .get(&cid)
136            .cloned()
137    }
138
139    /// Locks the instrument of `cid` when it is known. `None`: an order from before this process
140    /// started, its commands are not serialized with the others.
141    pub async fn lock_cid(&self, cid: ClientOrderId) -> Option<InstrumentGuard> {
142        match self.instrument_of(cid) {
143            Some(i) => Some(self.lock(&i).await),
144            None => None,
145        }
146    }
147}
brokapi_runtime/instrument_queue.rs

brokapi_runtime/
instrument_queue.rs
