3
\$\begingroup\$

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.

\$\endgroup\$
2
\$\begingroup\$

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
}
\$\endgroup\$
  • \$\begingroup\$ I see...a functional aproach. Some questions: 1.- I guess scoped threads (crossbeam) would get rid of Arc. 2.- 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. 3.- can we filter on value references instead of key objects? Thanks for the review, much clearer code \$\endgroup\$ – francesc Dec 5 '18 at 12:37
  • \$\begingroup\$ @francesc For the first question; yes in scoped thread you'll not need an Arc but i don't know how crossbeam handles the sharing inside. It would be great if someone knows \$\endgroup\$ – Ömer Erden Dec 6 '18 at 14:51
  • \$\begingroup\$ Pretty sure that your unsafe example invokes undefined behavior. When you move a value out of a location, corresponding memory location is considered uninitialized. Here you take a pointer to that location, and then dereference it after moving the value out of there - so you dereference a pointer to uninitialized memory. \$\endgroup\$ – user3447239 Jan 3 at 10:33
1
\$\begingroup\$

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?

\$\endgroup\$
  • \$\begingroup\$ We could say it needs an improvement, there is some working on feature called NLL to improve borrow checker, but it looks like still there is no solution for this problem( using reference after move ). For the restrictive you may want to read this, Scenario can be safe but currently Rust is not able to understand all of it because there is a lot of unsafe scenario out there, for this Rust provides unsafe(probably you know), it says if you are sure about it then do it. \$\endgroup\$ – Ömer Erden Dec 7 '18 at 18:22
  • \$\begingroup\$ Additionally i haven't deeply analyzed for thread safety but for the memory safety code is safe so i'll update your post with an unsafe solution. \$\endgroup\$ – Ömer Erden Dec 7 '18 at 18:25
  • \$\begingroup\$ @fransesc it turns out my edit was rejected on your post, so i am adding it in my post. \$\endgroup\$ – Ömer Erden Dec 8 '18 at 11:57

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.