How to handle many connections in one tokio task

[canoriz]

I am writing a program (specifically, a BitTorrent client) and it has many logical jobs (download jobs). For any job, many net connections are associated with the job, and new connections may join or leave the job. Every connection of a job basically do the same operations, for example, for every connection, there are 2 operations:

1. Receive a message from peer, then tell job controller the message.
2. Job controller gives a message to a connection, then the connection sends that message to peer.

I can think of 2 approaches to implement this.

Approach 1:

Spawn a tokio task for every connection, and use some sync mechanisms to sync states with the job controller.

async fn connection_task(c: TcpConn, controller: Arc<JobController>) {
    let rx = controller.command_handle_of(c);
    select! {
        msg = receive_income_msg_from(c) => {
            some_sync_operation(msg, controller);
        }
        msg = outcome_msg_from_controller(rx) => {
            send_msg_through(c, msg);
        }
    }
}

Approach 2:

Use an event loop, like epoll, which tracks all the interested connections (file descriptors). Once a connection can read or write, do operations on that connection. By this way, all read and write are done in only one task, and do not require sync mechanism between connections. Below is a highly simplified pseudocode.

async fn job_controller(cmd_rx: Receiver<Cmd>) {
    let conns = some_epoll_like_struct::new();
    loop {
        select! {
            c = conns.is_ready() => {
               if write_ready(c) {
                    send_data_through(c, pending_send(c).pop_one());
                    if no_more_pending_send(c) {
                        conns.set_interest(c, Interest::Read);
                    }
               }
               if read_ready(c) {
                   read_data_from(c);
               }
              
               do_some_controller_work();
               // maybe plan to send data through other connection
               if let (other_conn, msg) = maybe_need_send_new_data() {
                   pending_send(other_conn).append(msg);
                   conns.set_interest(other_conn, Interest::Read | Interest::Write);
               }
           }
           cmd{connection: c, cmd: cmd} = cmd_rx.recv() => {
               match cmd {
                   AddConn => conn.add(c);
                   RemoveConn => conn.remove(c);
               }
           }
        }
    }
}

My question is:

Is approach 2 suggested or not in tokio? And if viable, how to implement that?

One implementation I come up with is combining many TcpConn::Ready() into one large future (maybe by using JoinSet?), then poll the large future, the poll result will be the ready connection, like epoll. But this requires writing everything in the event loop style, which is (to me) not rust's async style, and seems to be re-creating what epoll has already done.

A more "rust async style" way is having a conns.concurrent_in_one_task(handler_of_every_connection: async fn(c: Conn, s: StateOfNow)) API, which runs all connections in one task concurrently. And APIs to add/remove connections of conns.

To my knowledge, it is not that simple as the pseudocode, send() and recv() system calls may send/recv only part of the buffer, and the caller should send/recv the remaining part when connection is ready again, so some additional states are required.

I have some experience in C and sync Rust, but not much in tokio or async Rust. Any discussion will be helpful.

[Darksonn]

Spawn a task per connection. You may find this article and talk useful for how to design programs with one task per connection.

[canoriz]

Spawn a task per connection. You may find this article and talk useful for how to design programs with one task per connection.

Your article is really helpful, I will use one task per connection.
But just curious, do you mean users are not suggested to handle many connections in one tokio task?

[Darksonn]

Yes, handling many connections in one task makes no sense. If you spawn one task per connection, then you are using an event loop like epoll which tracks all the interested connections. That's literally what Tokio does. The entire point of async/await is to make spawning tasks really cheap, so why complicate your life by not taking advantage of that?