Simple multithreaded chat server in Rust

Question

I'm trying to develop a simple Rust chat server. I'm not a Rust expert and come from Java and Kotlin.

This same server in Kotlin is:

import java.io.*
import java.net.*
import kotlin.concurrent.thread
import java.util.concurrent.ConcurrentHashMap

fun main(args: Array<String>) {
    val serv = ServerSocket(Integer.parseInt(args[0]))
    //val users = ConcurrentHashMap<String, PrintWriter>()
    val users = Collections.synchronizedMap(HashMap<String, PrintWriter>())

    while (true) {
        val s = serv.accept()
        thread {
            var sin = s.getInputStream().bufferedReader()
            val sout = PrintWriter(s.getOutputStream(), true)

            // read nick
            val nick = sin.readLine()
            users.put(nick, sout)

            sin.forEachLine {
                for (peer in users.values) {
                    if (peer == sout) continue
                    peer.println(it)
                }
            }
            users.remove(nick)
        }
    }
}

After many attempts I've come up with a working Rust implementation :

use std::env;
use std::io;
use std::io::Write;
use std::io::{LineWriter, BufReader, BufRead};
use std::net::{TcpListener, TcpStream};
use std::sync::{Arc, Mutex};
use std::collections::HashMap;
use std::thread;
use std::ops::DerefMut;

fn main() -> io::Result<()> {
    let args: Vec<String> = env::args().collect();
    let server_socket = TcpListener::bind(&*format!("localhost:{}", args[1]))?;
    let users: Arc<Mutex<HashMap<String, LineWriter<TcpStream>>>> = Arc::new(Mutex::new(HashMap::new()));

    for socket in server_socket.incoming() {
        let users = users.clone();
        thread::spawn(move || -> io::Result<()> {
            let socket = socket?;
            let socket_copy = socket.try_clone()?;
            let mut inp = BufReader::new(socket);
            let out = LineWriter::new(socket_copy);

            // read nick
            let mut nick = String::new();
            inp.read_line(&mut nick)?;
            nick.pop(); // discard '\n'
            let nick_copy = nick.clone();
            {
                let mut users = users.lock().unwrap();
                users.insert(nick, out);
            }

            for line in inp.lines() {
                {
                    let line = line?;
                    let mut users = users.lock().unwrap();
                    for (nick, peer) in users.deref_mut() {
                        if *nick == nick_copy { continue; }
                        writeln!(peer, "{}", line)?;
                    }
                }
            }

            // remove nick
            {
                let mut users = users.lock().unwrap();
                users.remove(&nick_copy);
            }
            Ok(())
        });
    }
    Ok(())
}

Too verbose for my taste.

I'm particularly dissatisfied with those clone ops. When to move? When to borrow?

I'd prefer to use for peer in users.values_mut() instead of for (nick, peer) in users.deref_mut() but then I've problems when comparing LineWriter references to discard sender.

I'd also prefer to use RwLock instead of Mutex.

Any pointers to simplify/enhance code will be greatly appreciated.

Answer 1

I made some changes to your code (see below). You can check the code for what I've done so far: I added some comments on it so you can compare it with your version. I'll note some important points in here. Please ask if you have any questions; I will try to explain or we can improve the code together.

• Propagating error to the main function from a spawned thread is not possible. It's only possible after joined to the main thread (you can use join handler then propagate your anonymous function's return value to the main).
• Created ConnectionsSync struct as helper to get rid of Arc<RwLock<HashMap< initialization: it was looking too messy; also with this way you can add behaviors like (add_user, broadcast_message, remove_user, ... )
• Incoming has Iterator implementation so I can use it as iterator and I used try_for_each to tell that I am going to apply fallible function, then it will be propagated to main.

Code

use std::collections::HashMap;
use std::env;
use std::io;
use std::io::{BufRead, BufReader, LineWriter, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::{Arc, RwLock};
use std::thread;

type User = Arc<String>;
struct ConnectionsSync(Arc<RwLock<HashMap<User, LineWriter<TcpStream>>>>);
impl ConnectionsSync {
    fn new() -> Self {
        ConnectionsSync(Arc::new(RwLock::new(HashMap::new())))
    }
}

fn main() -> io::Result<()> {
    let port = env::args().nth(1).unwrap_or("8080".to_string()); //default port 8080
    let connections = ConnectionsSync::new();

    TcpListener::bind(&format!("localhost:{}", port))?
        .incoming()
        .try_for_each(|socket| -> io::Result<()> {
            let socket = socket?;
            let connections = connections.0.clone(); //increases ref count

            thread::spawn(move || {
                let (mut reader, writer) = socket
                    .try_clone()
                    .map(|socket1| (BufReader::new(socket1), LineWriter::new(socket)))
                    .unwrap();

                // read nick
                let mut nick = String::new();
                reader
                    .read_line(&mut nick)
                    .and_then(|_| Ok(nick.pop()))
                    .expect("Not able to read user nick");

                let current_user = Arc::new(nick);
                //add user
                connections
                    .write()
                    .unwrap()
                    .insert(current_user.clone(), writer);

                //read user input
                reader.lines().into_iter().for_each(|line| {
                    let line = line.expect("Not able to read user input");
                    connections
                        .write()
                        .unwrap()
                        .iter_mut()
                        .filter(|(other_user, _)| **other_user != current_user)
                        .for_each(|(_, peer)| {
                            writeln!(peer, "{} \t-> {}", current_user, line);
                        })
                });

                //discard user
                connections.write().unwrap().remove(&current_user);
            });

            Ok(())
        })
}

Edit for questions in comment section

Can we get rid of nick.clone() and keep a simple reference? I guess not because cannot move out of borrowed content despite nick being immutable.

Actually cannot move out of borrowed content error is not related being immutable or mutable. Consider this; Your friend borrowed a phone from you and moved it's battery out physically to use somewhere else. Then your friend gave it your phone back to you without a battery. You would probably give an instant error. So due to safety, Rust watches your friend and warns: "Hey your friend is going to need that part in the future, so stop it!"

Besides the error you mentioned, another error will be occurred if we get rid of nick.clone() and keep a simple reference: We'll face with "borrowed value does not live long enough". This happens because we have a HashMap which it's reference counted, so we don't exactly know when it is going to die. If we put our nick's simple reference into HashMap, the compiler will tell that: "Hey HasMap is the owner of nick's reference but nick is gonna die in that block and HashMap will live after that, so what i am gonna do with this reference in the future ? ". We can solve this problem with 2 solution.

• With cloning because we have 2 ownership of nick (HashMap, Thread Block). I believe you don't want this solution due to overhead of copying String.
• With Smart Pointers Like Rc(Reference Counted), we will give nick's ownership to the Rc and share it across the HashMap and Thread. But Rc is not thread safe so we need to use Arc instead of it.i updated code with Arc version, there will be still cloning in code but it will just increase the reference count instead of copying the whole String.

Can we filter on value references instead of key objects?

Sure you can do it simply you need to change your code like this:

.filter(|(_other_user, other_users_value)| other_users_value != current_users_value)

But in our case this will not work because Eq is not implemented for TCPStream, if you implement it by yourself it will work.

---

Edit: Added unsafe version with raw pointers instead of clone

let mut users = HashMap::new();

let nick = "SomeInput".to_string();
let nick_ptr = &nick as *const String;

users.insert(nick, "SomeValue".to_string());

unsafe {
    users.remove(&*nick_ptr);
    println!("{:?}", users.get(&*nick_ptr)); //output None
}

Answer 2

On the get rid of nick.clone() question or any form of reference counting whatsoever. Problem is also ilustrated here how-can-i-keep-a-reference-to-a-key-after-it-has-been-inserted-into-a-hashmap

I illustrate it with a simpler, sequential example:

let mut users = HashMap<String, ValueType>;
let value = ValueType::new();
let reader = LineWriter::new(..);

let mut nick = String::new();
reader.read_line(&mut nick);
let nick_ref = &nick;

// attempt to move, it seems that should be valid
// if users is frozen (which is what happens in practice)
// because nick is borrowed
users.insert(nick, value);

// this remove isn't allowed (of course), but former move seems to be valid
// does the borrow checker need an inprovement?
users.remove(nick_ref);

Is the borrow checker too restrictive?