Audit cases of "tokio::mpsc::channel, size of 1"

[smklein]

For background, see: #9259

Any case where we create a tokio::mpsc::channel of size 1, we are subsequently vulnerable to deadlock with the following code:

loop {
  tokio::select! {
    // Suppose this is queued first, but is only returning `Poll::pending`
    _ = rx.send(...) => { return }
    // Suppose instead, this select! branch triggers...
    _ = yield_interval.tick() => {
      // ... Then this call to `send` will be stuck forever, queued behind the first rx.send,
      // which is no longer polled.
      let result = rx.send(...).await;
      if result = ... { return }
    }
  }
}

This issue will document cases where we're using mpscs, and risking this behavior.

[smklein]

To answer the question of "what about mpscs of size > 1, would they hit this issue?":

Using the following labels:

// Someone outside this loop may be calling `rx.send`, let's call that "send A".

loop {
  tokio::select! {
    // Let's call this "send B"
    _ = rx.send(...) => { return }
    _ = yield_interval.tick() => {
      // Lets call this "send C"
      let result = rx.send(...).await;
      if result = ... { return }
    }
  }
}

The very expanded view of events is the following:

• The mpsc must exist, originally created with a capacity of one
• Someone else must be using the mpsc, sending a message that hasn't been read yet, so the capacity becomes zero
• rx.send(...) in the select! arm starts, but returns Poll::pending, because the capacity is still zero. Critically, this reserves a spot in the queue, and keeps it, because it is not cancelled.
• Inside the yield_interval.tick() arm, a different rx.send(...) call is made.

Event	Channel Contents	Channel Semaphore Queue
mpsc created	Empty, capacity = 1	[]
Send A (placed message immediately)	Contains message A, capacity = 0	[]
Send B (cannot place message, go to queue)	Contains message A, capacity = 0	[B]
Receive A	Empty, capacity = 1	[B]
Send C (cannot place message, go to queue)	Empty, capacity = 1	[B, C]
All futures sends are stuck waiting for B, but they cannot execute, because B isn't being polled	Empty, capacity = 1	[B, C, ... ]

At the end of this sequence, even though the channel is empty, it created a position in a queue for the first call to send() which it received. In the above table, that's send B, which is no longer being polled, and also has not been cancelled. This means: It won't make it onto the queue, and it won't get out of the way.

If the mpsc was "one size larger", this problem would be pushed back slightly:

Event	Channel Contents	Channel Semaphore Queue
mpsc created	Empty, capacity = 2	[]
Send A (placed message immediately)	Contains message A, capacity = 1	[]
Send B (placed message immediately)	Contains message [A, B] capacity = 0	[]
Receive A	Contains message [B], capacity = 1	[]
Send C (placed message immediately)	Contains message [B, C], capacity = 0	[]

Basically: We'd be able to unblock the task by immediately depositing the message in the mpsc, without relying on any queue ordering of stuck messages. However, the problem still exists, when the mpsc queue becomes full, for any reason.

The Channel Semaphore Queue grants permits to waiters as soon as capacity becomes available - and granting that capacity is not necessarily blocking, if the sender is a future that isn't being polled - but a larger mpsc can still reproduce this issue if there are N futures which are inert and blocked and have been granted capacity. This results in an mpsc::Sender::capacity of zero, which means that other send() calls could block, until one of those prior futures (now holding the capacity) gets unblocked, and can relinquish the space (via cancellation, or via sending and having the value be received).

[smklein]

omicron/wicket/src/wicketd.rs

Line 111 in af3b89f

let (poll_interval_now_tx, poll_interval_now_rx) = mpsc::channel(1);

• Uses try_send, not send, is not blocking.

---

omicron/installinator/src/reporter.rs

Line 109 in af3b89f

let (completion_tx, mut completion_rx) = mpsc::channel(1);

• Sender is cloned, and send is invoked in report_task_loop, which is spawned from a task in ReportTask::new. I think this cannot block like this issue describes because each sender exists in an independent tokio::task.

---

omicron/wicketd/src/mgs/inventory.rs

Lines 67 to 75 in af3b89f

	// We only want one outstanding ignition request at a time; if our
	// consumer is behind, we don't need to hit MGS until they can handle
	// our results.
	let (tx, rx) = mpsc::channel(1);
	// "Fetch immediately" also only needs a channel depth of 1: if there is
	// already a message in this channel, we're already trying to fetch
	// ASAP.
	let (fetch_now_tx, fetch_now_rx) = mpsc::channel(1);

• fetch_now_tx uses try_send, so it doesn't have any issues
• I believe there is only one sender

---

omicron/wicketd/src/mgs/inventory.rs

Lines 200 to 209 in af3b89f

	) -> (Self, ReceiverStream<FetchedSpData>) {
	// We only want one outstanding request at a time; if our consumer is
	// behind, we don't need to request new state MGS until they can handle
	// our results.
	let (data_tx, data_rx) = mpsc::channel(1);
	// "Fetch immediately" also only needs a channel depth of 1: if there is
	// already a message in this channel, we're already trying to fetch
	// ASAP.
	let (fetch_now_tx, fetch_now_rx) = mpsc::channel(1);

• Same as above, only one sender

---

omicron/oximeter/instruments/src/kstat/sampler.rs

Line 1245 in af3b89f

let (outbox, inbox) = mpsc::channel(1);

• ⚠️ This appears to expose an interface of async functions, which call send.outbox.send. I think these interfaces could be vulnerable to the same issue exposed by qorb - a pending call to the API (add_target, remove_target, update_target) would block a subsequent call to any other part of the API.

---

omicron/gateway-cli/src/usart.rs

Lines 167 to 168 in af3b89f

	let (tx0, mut rx0) = mpsc::channel(1);
	let (tx1, rx1) = mpsc::channel(1);

• tx1 only has one non-cloned spot where send is used
• tx0 is used as part of the start_if_unstartedmethod - however, this requires &mut self, and blocks out other calls to send().

---

omicron/sled-agent/src/artifact_store.rs

Line 627 in af3b89f

let (tx, mut rx) = mpsc::channel(1);

• I think these mpscs are actually used more like one-shots - they appear to be fully read and then dropped in write_stream.
  

  omicron/sled-agent/src/artifact_store.rs

  Line 680 in af3b89f

  sender.send(chunk).await.ok().map(|()| sender)

---

omicron/sled-agent/src/bootstrap/server.rs

Line 205 in af3b89f

let (sled_reset_tx, sled_reset_rx) = mpsc::channel(1);

• sled_reset_tx only uses try_send

omicron/sled-agent/src/bootstrap/server.rs

Line 223 in af3b89f

let (sled_init_tx, sled_init_rx) = mpsc::channel(1);

• sled_init_tx is used in sprockets_server, with a single tx_requests.send used within handle_start_sled_agent_request. I don't think the tokio task calling this thread can currently switch away from the task, so it will poll until there is progress.

---

omicron/sled-agent/src/instance.rs

Line 1597 in af3b89f

let (tx_monitor, rx_monitor) = mpsc::channel(1);

• tx_monitor.send is invoked by an InstanceMonitorRunner, but cannot be cancelled.

---

omicron/sled-agent/config-reconciler/src/dump_setup_task.rs

Line 34 in af3b89f

let (archive_tx, archive_rx) = mpsc::channel(1);

• (RISKY, BUT I THINK OKAY?) archive_tx.send is invoked by archive_former_zone_root, which is a pub async method. We call this in the DumpSetupTask, in a tokio::select loop, but we only call this within the body of one of the select! arms -- not at a top level. Functionally this means it's "only getting called once".

---

omicron/wicket-dbg/src/runner.rs

Lines 67 to 68 in af3b89f

	let (tx, rx) = mpsc::channel(1);
	let (event_tx, event_rx) = mpsc::channel(1);

• I think these are okay - the calling contexts are &mut self, and fully blocking the task. There is a tokio::select here, but the futures in each arm are cancellable recv futures -- all the send future block the calling task.