OpenOptions::open never returns

[bbergeron-xlkey]

I have this function here which creates a UNIX named pipe for writing:

use std::path::Path;
use std::io;
use nix::sys::stat;
use nix::unistd;

async fn create_and_open_named_pipe(path: &Path) -> io::Result<tokio::fs::File> {
	println!("Creating pipe...");
	unistd::mkfifo(path, stat::Mode::S_IRUSR | stat::Mode::S_IWUSR)
        	.expect("Failed to create FIFO");

	println!("Opening pipe...");
	let file = tokio::fs::OpenOptions::new()
		.write(true)
		.open(path)
		.await;

	println!("done"); // Never prints
	file
}

But it blocks indefinitely around .open(path).await. What's going on?

[sfackler]

From man mkfifo:

   Once you have created a FIFO special file in this way, any
   process can open it for reading or writing, in the same way as an
   ordinary file.  However, it has to be open at both ends
   simultaneously before you can proceed to do any input or output
   operations on it.  Opening a FIFO for reading normally blocks
   until some other process opens the same FIFO for writing, and
   vice versa.  See [fifo(7)](https://man7.org/linux/man-pages/man7/fifo.7.html) for nonblocking handling of FIFO special
   files.

[Darksonn]

Do not use tokio::fs with things that are not real files. Please see tokio::net::unix::pipe for utilities for working with fifo pipes from Tokio.

[bbergeron-xlkey]

Thanks. I rewrote the function:

fn create_and_open_named_pipe(path: &Path) -> tokio::net::unix::pipe::Sender {
	unistd::mkfifo(path, stat::Mode::S_IRUSR | stat::Mode::S_IWUSR)
         	.expect("Failed to create FIFO");

	// source: https://docs.rs/tokio/latest/tokio/net/unix/pipe/struct.OpenOptions.html
	tokio::net::unix::pipe::OpenOptions::new()
	 	.read_write(true)
		.unchecked(true) 
		.open_sender(path)
		.expect("Failed to open FIFO")
}

But now instead of blocking there, it blocks at a later 'copy' call:

let mut sender = create_and_open_named_pipe(path);
tokio::io::copy(&mut reader, &mut sender).await.unwrap();
// (send command to child process to make it read the FIFO)
self.send(path).await.unwrap()

Which makes sense with what sfackler said because the receiving end isn't configured yet. Does tokio provides anything to fix that?

[Darksonn]

You can only write to sender if another process has opened the file for reading. So why don't you open it for reading from another process? You can of course implement that other process with Tokio or any programming language you would like.

[bbergeron-xlkey]

My project uses a program called exiftool in deamon mode, and uses stdin and stdout to send commands and receive output. To be more specific, my program sends file paths to exiftool and receive exif data. To avoid storage overhead, I decided to use named pipes so that any copying from my program to exiftool is done in-memory.

The fifo stays alive as long as the child process exists. From what I understand, it would be best to do these things in that order:

1. Send the command, have exiftool wait on the fifo
2. Copy the data through the fifo
3. Read from stdout

So I rewrote my code again and made it more explicit:

use std::path::{Path, PathBuf};
use std::process::Stdio;
use tokio::net::unix::pipe::Sender;
use tokio::{io::AsyncWriteExt, process::{Child, Command}};
use tokio::io::{copy, AsyncBufRead, AsyncReadExt};
use nix::sys::stat;
use nix::unistd;

pub struct Exiftool {
	child: Child,
	pipe_path: PathBuf,
	pipe_file: Sender
}

impl Exiftool {
	pub async fn new() -> Self {
		let child = Command::new("exiftool")
			.args(["-stay_open", "true", "-@", "-"])
			.stdin(Stdio::piped())
			.stdout(Stdio::piped())
			.spawn()
			.unwrap();

		let pipe_path = PathBuf::from("./pipe");
		let pipe_file = create_named_pipe(&pipe_path);

		Self {
			child,
			pipe_path,
			pipe_file
		}
	}

	/// Query exiftol to return a file's (provided by the reader object) exif data as json
	pub async fn exec(&mut self, mut reader: impl AsyncBufRead + std::marker::Unpin) -> String {
		println!("Sending command to stdin...");
		let command = &format!("-j\n{}\n", self.pipe_path.display());
		self.child.stdin.take().unwrap().write(command.as_bytes()).await.unwrap();

		println!("Copying data to fifo...");
		copy(&mut reader, &mut self.pipe_file).await.unwrap(); // (blocks here)

		println!("Reading output from stdout...");
		let mut output = String::new();
		self.child.stdout.take().unwrap().read_to_string(&mut output).await.unwrap();
		output
	}
}

fn create_named_pipe(path: &Path) -> tokio::net::unix::pipe::Sender {
	unistd::mkfifo(path, stat::Mode::S_IRUSR | stat::Mode::S_IWUSR)
        .expect("Failed to create FIFO");

	tokio::net::unix::pipe::OpenOptions::new()
	 	.read_write(true)
		.unchecked(true)
		.open_sender(path)
		.expect("Failed to open FIFO")
}

And I have the same problem where copy blocks and never returns. I'd like to be more helpful but I'm out of my depth here.

[bbergeron-xlkey]

I tried a new version where I swapped stdin and the pipe (i.e. sending commands through the pipe, and copying files to stdin):

use std::process::Stdio;
use tokio::net::unix::pipe::Sender;
use tokio::process::{Child, Command};
use tokio::io::{copy, AsyncBufRead, AsyncReadExt};
use nix::sys::stat;
use nix::unistd;

pub struct Exiftool {
	child: Child,
	command_sender: Sender
}

impl Exiftool {
	pub async fn new() -> Self {
		let command_sender = create_named_pipe("./pipe");

		// Open exiftool in daemon mode (-stay_open true) and make it listen for commands
		// on "./pipe" (-@ ./pipe)
		let child = Command::new("exiftool")
			.args(["-stay_open", "true", "-@", "./pipe"])
			.stdin(Stdio::piped())
			.stdout(Stdio::piped())
			.spawn()
			.unwrap();

		Self { child, command_sender }
	}

	/// Query exiftool to return a file's (provided by the reader object) exif data as json
	pub async fn exec(&mut self, mut reader: impl AsyncBufRead + std::marker::Unpin) -> String {
		println!("Sending command to pipe...");
		self.command_sender.try_write("-j\n-\n".as_bytes()).unwrap();

		println!("Copying data to stdin...");
		copy(&mut reader, &mut self.child.stdin.take().unwrap()).await.unwrap(); // (still blocks!)

		println!("Reading output from stdout...");
		let mut output = String::new();
		self.child.stdout.take().unwrap().read_to_string(&mut output).await.unwrap();
		output
	}
}

fn create_named_pipe(path: &str) -> tokio::net::unix::pipe::Sender {
	unistd::mkfifo(path, stat::Mode::S_IRUSR | stat::Mode::S_IWUSR)
    	.expect("Failed to create FIFO");

	tokio::net::unix::pipe::OpenOptions::new()
	 	.read_write(true)
		.unchecked(true)
		.open_sender(path)
		.expect("Failed to open FIFO")
}

This way, the pipe works and I'm free to use stdin for sending files... Except that clone still blocks, even tho cloning from a reader to stdin worked in a previous iteration. It's such a headache. Are you fellas still with me? Should I open a new discussion? The original question is answered, after all.

[Darksonn]

Let me ask you again ... in the former example, you copy data from exiftool into a pipe. The pipe is not connected to anywhere? What did you want the code to do? Throw it away? No, what it's going to do is that it will wait for someone to receive the data, and once there is a receiver then it will send it.

[bbergeron-xlkey]

No, it's the opposite: I copy the data from my program into a pipe, and exiftool reads it. My project is the sender; exiftool is the receiver - that's how things are connected.

I was hoping that the data would be buffered until something reads it, but sflacker's first comment clarified that FIFOs work differently - they work as you described. With that understood, I fixed my code, and now I’m realizing the other half of the problem concerns exiftool and how it receives data.

TL;DR: The first two replies solved the issue, so thanks to both of you.