Нужно ли синхронизировать вызовы TcpStream::write_all, чтобы избежать чередования?
У меня есть TcpStream (из стандартной библиотеки), и я хотел бы писать в него из нескольких потоков. TcpStream позволяет это без дополнительной синхронизации благодаря реализации Write for &TcpStream . Полезные нагрузки упакованы таким образом, что я делаю один вызов .write_all() для каждой полезной нагрузки.
use std::io::Write;
use std::net::TcpStream;
pub struct Publisher {
stream: TcpStream,
}
impl Publisher {
pub fn send(&self, payload: &[u8]) {
// ignore errors for now
let _ = (&self.stream).write_all(payload);
}
}
Но действительно ли это работает?
Меня беспокоит то, что .write_all() может включать несколько вызовов .write() для отправки полной полезной нагрузки, и, таким образом, одновременные вызовы могут в конечном итоге привести к чередованию операций записи из разных потоков. Я не вижу никакой специальной обработки для TcpStream::write_all, поэтому он просто использует реализацию типажа по умолчанию.
Обосновано ли мое беспокойство? Есть ли «умный» способ избежать этой проблемы? Или мне просто нужно обернуть его в Mutex, несмотря ни на что?
Ответы 1
Да, вы несете ответственность за синхронизацию записи с TcpStream, поэтому ваше беспокойство вполне обосновано. Тот факт, что он реализует Write для общей ссылки, более показательен для базовых реализаций, которые обычно используют только целое число для ссылки на open TcpStreams и, таким образом, не нуждаются в каких-либо ссылках или изменяемости структуры Rust для записи в них.
На самом деле, в документации нет ничего, что указывало бы на синхронизацию write_all с TcpStream. Просмотр кода также не выявил никакой внутренней синхронизации. И действительно, вы можете наблюдать чередование с помощью следующей тестовой программы с достаточно большим N:
use std::collections::HashMap;
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::thread;
const N: usize = 50_000_000;
fn main() -> Result<(), std::io::Error> {
let listener = TcpListener::bind(("127.0.0.1", 0))?;
let address = listener.local_addr()?;
let handle = thread::spawn(read(listener));
let stream = TcpStream::connect(address)?;
thread::scope(|s| {
s.spawn(write(b'a', &stream));
s.spawn(write(b'b', &stream));
});
handle.join().unwrap();
Ok(())
}
fn read(listener: TcpListener) -> impl FnOnce() {
move || {
while let Ok((mut s, _)) = listener.accept() {
let mut chars = HashMap::new();
let mut buf = [0u8; 1024];
while let Ok(n) = s.read(&mut buf) {
for &c in &buf[..n] {
chars.entry(c).and_modify(|v| *v += 1).or_insert(1);
}
if ![Some(N), None].contains(&chars.get(&b'a').copied())
&& ![Some(N), None].contains(&chars.get(&b'b').copied())
{
if chars[&b'a'] > chars[&b'b'] {
eprintln!("received 'b' before done receiving 'a' {chars:?}");
} else {
eprintln!("received 'a' before done receiving 'b' {chars:?}");
}
return;
}
}
}
}
}
fn write(c: u8, mut s: &TcpStream) -> impl FnOnce() + '_ {
let data = vec![c; N];
move || {
_ = s.write_all(&data);
}
}
Я бы использовал Mutex, чтобы добавить необходимую синхронизацию.
То же самое относится и к другим типам, имеющим impl Write for &T в стандартной библиотеке: