I am creating a program that consists of WebsocketReader, which reads data from a websocket (basically runs a loop to read out the newly incoming data), and ConfigurationFetcher, which needs to periodically (once per 30s) make a HTTP request to a different resource from which it receives configuration data, that should be passed to the WebsocketReader, in order for it to store them internally for faster access. The Fetcher is on a different thread, since it can be triggered manually "from outside".

Question is, how to most efficiently "inject" the data into the Reader, that is running the loop. Currently I am acquiring a Mutex lock on the Reader's configuration data in each iteration of the loop in order for the Fetcher to have a change to acquire the lock as well, when the config is being updated.

But is seems incredibly inefficient to me to be acquiring the lock on every iteration inside the WebsocketReader when the ConfigurationFetcher actually need to step in only once every 30s (meaning only once in 30 s, the mutex lock is held by someone else).

Is there any more efficient way of doing this? I would sort of like something, where a lock can be held "indefinitelly" until someone else needs to borrow it.

I am creating a program that consists of WebsocketReader, which reads data from a websocket (basically runs a loop to read out the newly incoming data), and ConfigurationFetcher, which needs to periodically (once per 30s) make a HTTP request to a different resource from which it receives configuration data, that should be passed to the WebsocketReader, in order for it to store them internally for faster access. The Fetcher is on a different thread, since it can be triggered manually "from outside".

Question is, how to most efficiently "inject" the data into the Reader, that is running the loop. Currently I am acquiring a Mutex lock on the Reader's configuration data in each iteration of the loop in order for the Fetcher to have a change to acquire the lock as well, when the config is being updated.

But is seems incredibly inefficient to me to be acquiring the lock on every iteration inside the WebsocketReader when the ConfigurationFetcher actually need to step in only once every 30s (meaning only once in 30 s, the mutex lock is held by someone else).

Is there any more efficient way of doing this? I would sort of like something, where a lock can be held "indefinitelly" until someone else needs to borrow it.

• rust
• concurrency
• mutex

• 3 Uncontested mutex access is very fast. Do you have reason to believe this has a performance impact? – kmdreko Commented Feb 7 at 22:02
• You could try_send it to a bufferless channel, but idk if that would be better than your current method. – drewtato Commented Feb 7 at 22:32
• Are you using any async runtime such as tokio? – fdan Commented Feb 8 at 15:09
• 1 You might want to look at arc_swap. However, as previously pointed out, there is most like no efficiency to be gained. On modern CPUs, an uncontested mutex can be acquired really fast, and other solutions (e.g. a channel, arc_swap) have the same or nearly the same mechanics at their heart. – user2722968 Commented 2 days ago
• @kmdreko not empirical, just an intuition. I am not saying it is not fast, but that it just seems unnecesary if it can be avoided. – Lukáš Řádek Commented 2 days ago

1 Answer 1

When working with tokio, one option is to use a channel in combination with select!. This way, your websocket handler will never lock a mutex when receiving data. Instead, the configuration is stored on the stack and only updated when the configuration is fetched by the other task.

The following example uses a UDP socket instead of a websocket.

#[derive(Copy, Clone)]
struct Config;

#[tokio::main]
async fn main() {
    let (tx, rx) = watch::channel(Config);
    let _ = join!(
        tokio::spawn(fetch_config(tx)),
        tokio::spawn(handle_websocket(rx))
    );
}

async fn fetch_config(tx: Sender<Config>) {
    loop {
        // fetch config using http
        tx.send(Config).unwrap();
        time::sleep(Duration::from_secs(30)).await;
    }
}

async fn handle_websocket(mut rx: Receiver<Config>) {
    let sock = UdpSocket::bind("127.0.0.1:3000").await.unwrap();
    let mut _config = Config;

    loop {
        let mut buf = [0; 1024];
        select! {
            _ = sock.recv_from(&mut buf) => { /* work with _config */ },
            _ = rx.changed() => _config = *rx.borrow_and_update(),
        }
    }
}

Playground