Deleting a BFD peer blocks waiting for sync thread to be joined

[jgallagher]

I'm trying to wire up proptests to the omicron BFD config reconciler. My first few attempts at that resulted in extremely long tests (many minutes). I think I've narrowed this down to "deleting a BFD peer immediately after adding it blocks for about 1 second", which is easily reproducible:

% cat make-bfd-peer.json
{
    "peer": "1.2.3.4",
    "listen": "0.0.0.0",
    "required_rx": 1234,
    "detection_threshold": 10,
    "mode": "SingleHop"
}
% time curl -X PUT -H 'content-type: application/json' -H 'api-version: 8.0.0'
'http://[::1]:8889/bfd/peers' --data @make-bfd-peer.json && \
  time curl -X DELETE -H 'api-version: 8.0.0' 'http://[::1]:8889/bfd/peers/1.2.3.4'
curl -X PUT -H 'content-type: application/json' -H 'api-version: 8.0.0'     0.00s user 0.00s system 79% cpu 0.007 total
curl -X DELETE -H 'api-version: 8.0.0' 'http://[::1]:8889/bfd/peers/1.2.3.4'  0.00s user 0.00s system 0% cpu 1.002 total

The mgd logs confirm the handler latency for this delete was about 1 second as well:

14:21:57.634Z INFO slog-rs (mgd): request completed
    latency_us = 994447
    local_addr = [::1]:8889
    method = DELETE
    module = admin
    remote_addr = [::1]:36984
    req_id = ace339cc-ef54-413e-9481-cb6085d0f466
    response_code = 204
    unit = api_server
    uri = /bfd/peers/1.2.3.4

The remove_bfd_peer() function looks quite innocuous: it takes a (sync) lock to remove an entry from a HashMap (bfd.daemon.remove_peer()), then takes a different (sync) lock to remove an entry from a HashSet, set an atomic boolean, and remove some entries from other HashMaps (bfd.dispatcher.remove()). This seems like it should be fine: all these operations should be fast and there appear to be no blocking calls made.

Some extra debug logging confirmed it was the bfd.daemon.remove_peer() call that was taking ~1 second, which is extra surprising because it's literally just a single HashMap::remove() call. We were not blocked waiting to acquire the lock, rather, we were blocked dropping the Session that we removed from the HashMap. I believe the sequencing here is:

1. We acquire the bfd.daemon lock.
2. We call Daemon::remove_peer().
3. This removes the peer from the HashMap, then drops both the key (IpAddr) and value (Session).
4. One of Session's fields is _egress_thread: Option<Arc<ManagedThread>>. ManagedThread has a Drop impl that sets a "this thread has been dropped" AtomicBool, then synchronously .join()s the thread. This means we're blocked here until the egress thread ends.
5. The egress thread has a loop that's receiving on a synchronous (std lib) mpsc channel with a 1 second timeout, so it only checks the dropped flag every second (assuming no packets arrive on our channel - if we're get a packet we won't check dropped at all, so I assume we're relying on the sending half of the channel being shut down?).
6. We've been holding the bfd.daemon lock this whole time - once the implicit Drop in remove_peer() is done, we'll unlock the lock and proceed.

I confirmed this is the issue affecting the proptest time by changing this 1 second recv_timeout to 10 milliseconds, and the overall proptest time went from 5 minutes to 3 seconds. I'm not necessarily proposing that as a real fix for this, although it would be an easy one, if we think checking an atomic bool more frequently than once a second isn't too high of a price to pay.

(There's a different issue here I'm not hitting - if we're having trouble binding the socket, we only check the dropped flag every 5 seconds, not even every 1 second.)

I think in general I'd propose we rethink the architecture some here though - I don't think we should be joining synchronous threads in async functions, both because it's surprising behavior and because it means we're adding blocking operations to async contexts, which historically have caused problems with the tokio runtime. I assume if we had something like this where we were obviously joining a sync thread in an async function:

async fn remove_bfd_peer(/* args */) {
    // Shut down the session thread
    let session = ctx.context()
        .bfd
        .daemon
        .lock()
        .unwrap()
        .remove_peer(rq.addr);
    session.join();

    // ... rest of method ...
}

it would stick out like a sore thumb in code review, but even this would be better than the current implementation: at least with this we'd be joining the thread after unlocking bfd.daemon instead of while we're holding the lock.

Mixing async with sync threads and I/O is pretty precarious IMO, and I don't think we're set up for success with the current structure. I know async has its warts, but this kind of thing is where it really excels; instead of having to pick a frequency at which we check a "should we stop flag" like this:

match rx.recv_timeout(Duration::from_secs(1)) {
    Ok(result) => result,
    Err(RecvTimeoutError::Timeout) => {
        if dropped.load(Ordering::Relaxed) {
            break 'egress;
        }
        continue;
    }
    Err(RecvTimeoutError::Disconnected) => {
        bfd_log!(log, warn, "udp egress channel closed");
        break 'egress;
    }
};

an async version could select! and be notified immediately when shutdown is desired with no need to pick a timeout:

tokio::select! {
    result = rx.recv() => {
        match result {
            Some(result) => result,
            None => {
                bfd_log!(log, warn, "udp egress channel closed");
                break 'egress;
            }
        }
    },
    stop = dropped_notification => { break 'egress; },
}

I assume it would be a pretty big lift to switch things over from sync threads to async tasks, but it might be worth considering? Another option would be to go the "sans i/o" route. It looks like some of the code here (e.g., the BFD state machine) is kind of leaning in that direction, in that it doesn't do I/O directly, but it's pretty far from really embracing the pattern: it still spawns threads, has timeouts (including a type that synchronously sleeps in its Drop impl - I think I get why this exists, but we have to be really careful we never drop one of those in an async context). I'm also a little worried about the threads we spawn to handle a BFD peer that aren't being joined (either implicitly or explicitly): two in the state machine and one in Listener that are presumably left to end on their own - if we tighten up the latency of shutting down this thread, are those also guaranteed to be stopped, or could they still be active for some period of time afterwards?

Some possible paths to address this other than adjusting the timeout or rewriting the world to be async:

1. Could we remove the implicit join in drop, then explicitly join the thread inside a tokio::spawn_blocking()? If we need the remove_bfd_peer() to not return until the thread is shut down, this wouldn't help with the latency, but would at least get rid of the blocking-in-an-async-context bit. (Could combine this with dropping the channel timeout some?) If we removed ManagedThread I assume some similar issues might fall out in other places where it's used (BGP / NDP also use it, but I didn't look at them to see if their uses are crossing the sync/async boundary).
2. There are other synchronous channel implementations that provide an equivalent to select (e.g., flume or crossbeam. Could we use these instead of an atomic bool we have to poll at some interval?