Process Substitution Without Shell?

November 28th, 2023
shell, tech
While I still write a decent amount of shell I generally try to avoid it. It's hard for others to read, has a lot of sharp edges, tends to swallow errors, and handles the unusual situations poorly. But one thing that keeps me coming back to it is how easily I can set up trees of processes.

Say I have a program that reads two files together in a single pass [1] and writes something out. The inputs you have are compressed, so you'll need to decompress them, and the output needs to be compressed before you write it out to storage. You could do:

# download the files
aws s3 cp "$path1" .
aws s3 cp "$path2" .

# decompress the files
gunzip "$file1"
gunzip "$file2"

# run the command
cmd -1 "$file1" -2 "$file2" > "$fileOut"

# compress the output
gzip "$fileOut"

# upload the output
aws s3 cp "$fileOut.gz" "$pathOut"

This works, but for large files it's slow and needs too much space. We're waiting for each step to finish before starting the next, and we're storing some very large intermediate files on the local machine.

Instead, we'd like to stream the inputs down, decompress them as we go, compress the output as it comes out, and stream the output back up. In bash this is reasonably straightforward to write:

cmd -1 <(aws s3 cp "$path1" - | gunzip) \
    -2 <(aws s3 cp "$path2" - | gunzip) \
  | gzip | aws s3 cp - "$pathOut"

This uses almost no disk space and it parallelizes the decompression, command, and recompression. But it's also shell...

I tend to use python for this kind of thing, where I'm gluing things together and want it to be clear what I'm doing. It seems like it should be possible to do this sort of thing with the subprocess module, but while I've played with it a few times I haven't figured it out. I'd like an API like:

pipeline = subprocess.Pipeline()
dl1 = pipeline.process(
    ["aws", "s3", "cp", path1, "-"])
gunzip1 = pipeline.process(
    ["gunzip"], stdin=dl1.stdout)
dl2 = pipeline.process(
    ["aws", "s3", "cp", path2, "-"])
gunzip2 = pipeline.process(
    ["gunzip"], stdin=dl2.stdout)
cmd = pipeline.process(
    ["cmd", "-1", dl1.stdout,
            "-2", dl2.stdout])
gzip = pipeline.process(
    ["gzip"], stdin=cmd.stdout)
pipeline.process(
    ["aws", "s3", "cp", "-", pathOut],
    stdin=gzip.stdout)
pipeline.check_call()

Or:

from subprocess import check_call, PIPE, InputFile

check_call([
  "cmd",
  "-1", InputFile([
          "aws", "s3", "cp", path1, "-",
          PIPE, "gunzip"]),
  "-2", InputFile([
          "aws", "s3", "cp", path2, "-",
          PIPE, "gunzip"]),
  PIPE, "gzip",
  PIPE, "aws", "s3", "cp", "-", pathOut])

These are 5x and 3x the length of the bash version, but I'd be willing to put up with that for having something that's more robust. The difference would also be smaller in practice as the commands would typically have a lot of arguments.

I see these stack overflow answers suggesting named pipes, but they seem awkward, hard to read, and easy to get wrong. Is there a better way? Should I just stick with bash when doing something bash is this good a fit for, especially now that people can paste my code into an LLM and get an explanation of what it's doing?


[1] Interleaved fastq files, where the Nth record in file 1 corresponds to the Nth record in file 2.

Comment via: facebook, lesswrong, mastodon

Recent posts on blogs I like:

On The Prosperity Gospel

things I find fascinating: religion, scams

via Thing of Things February 19, 2024

Diseconomies of scale in fraud, spam, support, and moderation

If I ask myself a question like "I'd like to buy an SD card; who do I trust to sell me a real SD card and not some fake, Amazon or my local Best Buy?", of course the answer is that I trust my local Best Buy1 more than Amazon, which is notoriou…

via Posts on February 18, 2024

Raising children on the eve of AI

How do we prepare them for what we're not prepared for? The post Raising children on the eve of AI appeared first on Otherwise.

via Otherwise February 15, 2024

more     (via openring)