How can I merge two sorted named pipes with uneven sizes?

Question

My question is similar to "Merge two sorted files based on sorting values in the same field" but extending it to named pipes.

Say I have two text files with sorted integers and I want to merge them. I can use sort -nm file1.txt file2.txt > merged.txt to do a one-pass, non-blocking merge.

Now, say these files are actually named pipes (FIFOs) that I'm making and then populating from within python. As long as I alternate writing to one pipe and then next, I can do this just fine. This code works to generate two ordered lists of integers, write them to named pipes read by a sort subprocess, which outputs the merged result in a single file:

import tempfile
import subprocess
import os
import sys


# Make temporary fifos
tempdir = tempfile.mkdtemp()
tempdir = "/tmp/tmph1ilvegn"  # hard-code tempdir for repeated runs
fifo_path1 = os.path.join(tempdir, "fifo1")
fifo_path2 = os.path.join(tempdir, "fifo2")
pos_fifo = os.mkfifo(fifo_path1)
neg_fifo = os.mkfifo(fifo_path2)


# Output will be a sorted merge from 2 inlines2ut streams.
outfile = "sorted_merge.txt"
sortProcess = subprocess.Popen('sort -snm ' +  fifo_path1 + " " + fifo_path2 + " > " +
    outfile, shell=True)


fifo_writer1 = open(fifo_path1, 'w')
fifo_writer2 = open(fifo_path2, 'w')

nlines1 = 0
nlines2 = 0

# Simulate 2 sorted lists by just going iterating through a sorted list and
# printing some numbers to one list and some to the other.

for i in range(1,100000):
    print("i: {}; n1: {}; n2: {}; imbalance:{}".format(i, nlines1, nlines2, nlines1-nlines2))
    line_to_write = (str(i) + "\n")
    if i % 2:
        nlines1 +=1
        fifo_writer2.write(line_to_write)
    else:
        nlines2 +=1
        fifo_writer1.write(line_to_write)

# clean up fifos:
fifo_writer1.close()
fifo_writer2.close()
os.remove(fifo_path1)
os.remove(fifo_path2)
sortProcess.communicate()

I get a sorted result. BUT -- now let's make the lists imbalance by changing the i % 2 to i % 3. In the original, this just prints to fifo1 then fifo2 then fifo1 then fifo2, etc. In the modified version, it prints twice as many lines to one of the two pipes.

Running this with i % 3 I get the following output:

...
i: 16182; n1: 10788; n2: 5393; imbalance:5395
i: 16183; n1: 10788; n2: 5394; imbalance:5394
i: 16184; n1: 10789; n2: 5394; imbalance:5395
i: 16185; n1: 10790; n2: 5394; imbalance:5396
i: 16186; n1: 10790; n2: 5395; imbalance:5395
i: 16187; n1: 10791; n2: 5395; imbalance:5396
i: 16188; n1: 10792; n2: 5395; imbalance:5397
i: 16189; n1: 10792; n2: 5396; imbalance:5396
i: 16190; n1: 10793; n2: 5396; imbalance:5397
i: 16191; n1: 10794; n2: 5396; imbalance:5398
i: 16192; n1: 10794; n2: 5397; imbalance:5397
i: 16193; n1: 10795; n2: 5397; imbalance:5398

It always stops at the same spot. Using strace I can see:

The python process has hung up at a write call to spot 4: write(4, "9\n15170\n15172\n15173\n15175\n15176\n"..., 4100

But the sort process is hung up at a read call to spot 3: read(3,

Looking at the lsof -n -p output for the sort process shows that it's waiting for a value to come to fifo1, while the write process is waiting to write to a value to fifo2:

sort    23330 nsheff  txt    REG  259,2   110040 10769142 /usr/bin/sort
sort    23330 nsheff  mem    REG  259,2  2981280 10752335 /usr/lib/locale/locale-archive
sort    23330 nsheff  mem    REG  259,2  1868984  6031544 /lib/x86_64-linux-gnu/libc-2.23.so
sort    23330 nsheff  mem    REG  259,2   138696  6031518 /lib/x86_64-linux-gnu/libpthread-2.23.so
sort    23330 nsheff  mem    REG  259,2   162632  6031516 /lib/x86_64-linux-gnu/ld-2.23.so
sort    23330 nsheff    0u   CHR  136,1      0t0        4 /dev/pts/1
sort    23330 nsheff    1w   REG  259,2        0  4719615 /home/nsheff/code/bamSitesToWig/sorted_merge.txt
sort    23330 nsheff    2u   CHR  136,1      0t0        4 /dev/pts/1
sort    23330 nsheff    3r  FIFO  259,2      0t0   786463 /tmp/tmph1ilvegn/fifo1
sort    23330 nsheff    4r  FIFO  259,2      0t0   786465 /tmp/tmph1ilvegn/fifo2

So for some reason, the sort process has *stopped listening to fifo2, causing the process to hang.

Now, if I go put a separate listener on fifo2 by just issuing cat fifo2... the process starts again and continues for thousands of iterations, until... it stops now at another random point (iteration 53733).

I think there must be something I don't understand going on with buffering pipes and how sort is changing from reading from one stream to the next. What is strange to me is that it's deterministic, failing at exactly the same spot, and only seems to fail when the lists are imbalanced.

Is there any way I can solve this?

Answer 1

Obviously your program creates a deadlock when you write different amounts of data to the two named pipes. Your program blocks on a write for one fifo2 (with buffer full) while the sort process blocks on a read for fifo1 (with buffer empty).

You don't know how sort is implemented. It probably wants to read the files in larger blocks and then process the data in memory for efficiency. Buffering may even happen automatically if sort uses functions from stdio.h for reading the data.

Named (and unnamed) pipes use a buffer for the data.
If the buffer is full, a writing process will block until a reading process has read some data or closed its end.
If the buffer is empty, a reading process will block until the writing process has written some data or closed its end.

If you write one line to fifo1 and two lines to fifo2 in every cycle, you will fill up the fifo2's buffer while fifo1's buffer is only half-filled.

Depending on how much data your program writes to the fifos and how much sort wants to read, this obviously ends up in a situation when sort wants to read something from fifo1 which just has an empty buffer while your program wants to write to fifo2 with a full buffer.

The result is deterministic because the pipe buffer has a fixed size and probably also your program and sort use fixed buffer sizes for reading or writing the data.

You can look at the source code of GNU sort at
https://github.com/wertarbyte/coreutils/blob/master/src/sort.c

In the beginning it tries to fill the input buffers for all input files in a loop over all files using function fillbuf.

Later under some conditions it calls fillbuf again for an input file.

In function fillbuf there is a comment

          /* Read as many bytes as possible, but do not read so many
             bytes that there might not be enough room for the
             corresponding line array.  The worst case is when the
             rest of the input file consists entirely of newlines,
             except that the last byte is not a newline.  */

Apparently sort selects one of the input files and wants a certain amount of data. It doesn't switch the input files if reading blocks.

The implementation works well for normal files because a read operation will either return some data or EOF after some time, so it will not block permanently.

---

It is always difficult to avoid deadlocks if you have more than one thing that can block between two processes/threads. In your case you should only use one pipe. Using non-blocking operations might help if you always have data to write to fifo1 if fifo2 would block or vice versa.

Using two pipes might work if you would use two separate threads/processes for writing to the pipes, but only if the threads/processes work independently from each other. It would not help if thread A which should write to pipe1 would somehow wait for thread B which just blocks on writing to pipe2.