Реализация семантики потока перемещения

Я хочу написать функцию, которая будет вызываться следующим образом:

send("message","address");

Где какой-то другой поток, который делает

let k = recv("address");
println!("{}",k);

видит message.

В частности, сообщение может быть большим, и поэтому мне нужна семантика "переместить" или "ноль-копировать" для отправки сообщения.

В C решение выглядит примерно так:

  1. Размещать сообщения в куче
  2. Иметь глобальную поточно-ориентированную хеш-карту, которая сопоставляет «адрес» с некоторой ячейкой памяти.
  3. Записывайте указатели в ячейку памяти при отправке и пробуждайте получателя с помощью семафора.
  4. Считайте указатели из области памяти при получении и подождите, пока семафор обработает новые сообщения.

Но, согласно другому вопросу SO, шаг № 2 "звучит как плохая идея". Так что я хотел бы увидеть более идиоматический подход к этой проблеме.


multithreading messaging rust zero-copy
person Drew    schedule 27.01.2015    source источник

Ответы (1)


arrow_upward
2
arrow_downward

Вы автоматически получаете такую ​​семантику перемещения и получаете легковесные перемещения, помещая большие значения в Box (т. е. размещая их в куче). Используя type ConcurrentHashMap<K, V> = Mutex<HashMap<K, V>>; в качестве потокобезопасного хэш-карты (существуют различные способы его улучшения), можно было бы:

use std::collections::{HashMap, RingBuf};
use std::sync::Mutex;

type ConcurrentHashMap<K, V> = Mutex<HashMap<K, V>>;

lazy_static! {
    pub static ref MAP: ConcurrentHashMap<String, RingBuf<String>> = {
        Mutex::new(HashMap::new())
    }
}

fn send(message: String, address: String) {
    MAP.lock()
       // find the place this message goes
       .entry(address)
       .get()
       // create a new RingBuf if this address was empty
       .unwrap_or_else(|v| v.insert(RingBuf::new()))
       // add the message on the back
       .push_back(message)
}
fn recv(address: &str) -> Option<String> {
     MAP.lock()
        .get_mut(address)
        // pull the message off the front
        .and_then(|buf| buf.pop_front())
}

Этот код использует макрос lazy_static! для получения глобальной хэш-карты (возможно, лучше использовать локальный объект, который оборачивает Arc<ConcurrentHashMap<...>, fwiw, так как глобальное состояние может затруднить рассуждения о поведении программы). Он также использует RingBuf в качестве очереди, так что сообщения накапливаются для заданного address. Если вы хотите поддерживать только одно сообщение за раз, тип может быть ConcurrentHashMap<String, String>, send может стать MAP.lock().insert(address, message), а recv просто MAP.lock().remove(address).

(NB. Я не компилировал это, поэтому типы могут точно не совпадать.)

person huon    schedule 27.01.2015