Splice between a pipe and a socket

[pkolaczk]

I can see you can zero-copy-move data between the pipes.
Is there a way to move the data from a socket to a pipe and from a pipe to a socket by using splice/tee?

[NobodyXu]

According to the man page of [splice]:

It transfers up to len bytes of data from the file descriptor fd_in to the file descriptor fd_out, where one of the file descriptors must refer to a pipe.

You can use splice to move data from pipe into socket. or vise versa.

[NobodyXu]

There're two functions for this:

• tokio_pipe::PipeRead::splice_to
• tokio_pipe::PipeWrite::splice_from

[pkolaczk]

Ok, but how to make them work with OwnedReadHalf and OwnedWriteHalf of the TpcStream?
Looks like stream halves don't implement to AsRawFd :(

[NobodyXu]

Checkout impl AsRef for OwnedReadHalf and impl AsRef for OwnedWriteHalf, which enables you to get TcpStream from OwnedReadHalf and OwnedWriteHalf respectively.

Then, you can just use impl AsRawFd for TcpStream.

[pkolaczk]

Ok, makes sense. Thank you!

[pkolaczk]

Tried that, but I'm hitting a problem:

async fn forward(src: OwnedReadHalf, dest: OwnedWriteHalf) -> io::Result<u64> {
    use tokio_pipe::{pipe, PipeRead, PipeWrite};
    let (mut pipe_read, mut pipe_write) = pipe()?;
    let mut in_fd = src.as_ref().as_raw_fd();
    let mut out_fd = dest.as_ref().as_raw_fd();
    let mut total_count = 0;
    loop {
        let count = pipe_write.splice_from(&mut AsyncFd::new(in_fd)?, None, 4096).await?;
        if count == 0 {
            break;
        }
        pipe_read.splice_to(&mut AsyncFd::new(out_fd)?, None, count, true).await?;
    }
    Ok(total_count)

error: generator cannot be sent between threads safely
   --> src/main.rs:187:9
    |
187 |         tokio::spawn(async move {
    |         ^^^^^^^^^^^^ future created by async block is not `Send`
    |
    = help: within `impl Future<Output = Result<(), ()>>`, the trait `Send` is not implemented for `*mut i64`
note: required by a bound in `tokio::spawn`
   --> /home/pkolaczk/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.19.2/src/task/spawn.rs:127:21
    |
127 |         T: Future + Send + 'static,
    |                     ^^^^ required by this bound in `tokio::spawn`

If I remove the inner body of the loop (splice_to and splice_from calls), it compiles fine.
I guess this has something to do with those AsyncFd references, although I'm quite stumped, because the docs say it is Sync and Send.

[NobodyXu]

Aha I got it.

That's because splice_impl contains raw pointer.
I will submit a PR to fix this.

[NobodyXu]

@pkolaczk Can you try the PR linked with this issue to see if it fixes the issue for you?

[pkolaczk]

Yeah, checking....

[pkolaczk]

Looks good! Thank you!

[pkolaczk]

Huh, but the idea still doesn't work:

let mut in_fd = AsyncFd::new(src.as_ref().as_raw_fd()).expect("in_fd");

This fails with:

thread 'tokio-runtime-worker' panicked at 'in_fd: Os { code: 17, kind: AlreadyExists, message: "File exists" }', src/main.rs:102:60
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

I also tried without getting the raw_fd:

let mut in_fd = AsyncFd::new(src.as_ref()).expect("in_fd");

But that doesn't even compile:

the trait `AsRawFd` is not implemented for `&tokio::net::TcpStream`

How to get an AsRawFd from &tokio::net::TcpStream? I cannot obtain a tokio::net::TcpStream here (which would be AsRawFd), because I have only one half of it.

[NobodyXu]

Ok, I got it.

The problem is that you cannot register the same fd twice with AsyncFd (TcpStream internally registers it in the same way).