Multi-stage pipe not working about sh HOT 10 CLOSED

I was able to reproduce it with those specific commands, but interestingly, other commands work for me:

from pbs import ls, sort, head, HOME
print head(sort(ls(HOME)))

I wonder if sort() isn't getting it's input fast enough from du(), and closing it's stdin? I'll investigate...

from sh.

Something rattling in my brain says "Unix processes can exit() before their pipes have been read." Sort wouldn't write output until du had completed, and head wouldn't be able to read from sort and do its job until both sort and du have exited. I wonder if there's a timing issue there. Shells have code passed down from the Ancients to handle all the corner cases, but since pbs takes on the shell's role... Could pbs or subprocess allow du to hard-exit before sort has a chance to read its output?

from sh.

@jonathaneunice It could be. I found this.... try running:

from pbs import du, sort, HOME
y = sort(du(HOME))
print y

Do you get a UnicodeDecodeError? It seems I have some weird named files in my ~/, and the piping code in pbs is trying to decode it as utf8 before passing it along, choking, and then head() may be trying to read off of a closed stdin.

from sh.

Ah my mistake, you said you tried that in your first post. What's weird is, I can get the head(sort(du())) code to work on some directories, but not others. But it doesn't seem to be dependent on the amount of lines du outputs.

from sh.

I have just over 12K lines in my du(HOME) output. There appear to be no non-ASCII characters in du's output. Unicode conversion doesn't seem to be an issue.

Odder still, tail(sort(du(HOME), "-nr")) works fine, even when head() does not. So I tried head(tail(...)) and it works. Then I tried head(tail(..., "-20000")) which doesn't work. Leading me to this code:

    try:
        n = 1
        while n < 200:
            y5 = head(tail(sort(du(HOME), "-nr"), "-{}".format(n)))
            out = open("y5.txt", "w")
            print >>out, y5
            print (len(str(y5).splitlines())), " lines in y5 at n =", n
            out.close()
            n += 1
    except:
        print "head() fails at n = ", n

Which fail AT DIFFERENT SETTINGS OF N, depending on the run. N = 18, 19, 21, 25 all saw failures. Dying at different values of N, depending on the run, IMO further suggests a timing-dependent issue. Could the process feeding head be reaped before head has had a chance to read all data in its pipe?

from sh.

Could the process feeding head be reaped before head has had a chance to read all data in its pipe?

It's possible but I have a hard time seeing how. Under the hood, each command blocks until it is done, unless explicitly put in the background: (line 133, pbs.py:

# run and block
self._stdout, self._stderr = self.process.communicate(stdin)
rc = self.process.wait()

On the piping side, a command that is receiving from another command feeds directly off of it's (completed) output: (pbs.py 363.py)

actual_stdin = first_arg.stdout

and the .stdout property is evaluated only after the command finishes. So as far as I can tell, commands finish entirely, their output is returned, and that output is fed to the next command.

A simple test I can think of is to use subprocess directly to run "head", then feed a massive amount of data in via .communicate() and see if that fails at all. I'll try to do some tests today, but I've been focusing on the rewrite_popen branch, which would do away with subprocess altogether (but perhaps not the problem). Anything else you can find will be a great help.

from sh.

I don't know the pbs or subprocess internals sufficiently to more than speculate...but two observations:

1. Subprocess is using low-level os.write() calls, which boil down to Unix write() system calls, and then aborting with a Unix error 32 (EPIPE, "broken pipe"). That's defined as "one of the processes went away, breaking the pipe, then the other process tried to do I/O on a consequently no-longer-existing file."

2. I am suspicious of the description "commands finish entirely, their output is returned, and that output is fed to the next..." That's not how pipes work, generally. The first command runs, writing data to a file, then at some point either completes OR is descheduled by the OS. At some later point, (any of the) processes further down the pipeline may be scheduled. If some data is available to them (piped into their stdin), they may read some of it and do whatever with it, including writing it to their stdout. But that later process may be descheduled before draining all data in its stdin, or before completing its own processing or output. The order that processes are scheduled in a pipeline is not deterministic. I don't mean to be pedantic, but if pbs assumes that processes do their jobs to completion, and then the next process in turn neatly has all the data available to it and all the time it needs to complete its task...that's now how Unix processes/pipes work. It's only the cleanest, simplest case. Could pbs mis-assume that an earlier pipeline stage has completed, and prematurely close its files as a result?

from sh.

@jonathaneunice

2. You're smart to be suspicious :) Typically piping does work that way. Unfortunately pbs (and subprocess in general) de-parallellizes system commands that would work cooperatively. This is the drawback of the subprocess module...it forces commands to finish before their output can be read (this isn't totally true, it is possible to read incrementally, but not out of the box, and certainly not easily). So pbs isn't assuming that the commands are finished, it is ensuring that it is. The call to self.process.wait() only returns when the process ends, and self.process.wait() happens on every command, and there is no behind-the-scenes threading.

I did do some tests however on our problem, and take a look at this:

from subprocess import Popen, PIPE
from string import letters


p_head = Popen(["head"], stdin=PIPE, stdout=PIPE)


data = [letters] * 1237
stdout, stderr = p_head.communicate("\n".join(data))

print stdout

For me, the above code yields "OSError: [Errno 32] Broken pipe" consistently. Unless I change "1237" to "1236", then it works consistently. Doing a little math, "letters" is 52 characters (uppercase and lowercase ascii letters), which is joined by newlines, so total data being written to the head proceess for 1237 lines is 65560 bytes, or just over 64k. With 1236 lines it's 65507 bytes, just shy of 64k. 64k happens to be the pipe buffer size of unix. Mystery solved?

from sh.

Just for some closure on the issue, the rewrite_popen branch should take care of issues like these when it's finished. It's a rewrite of subprocess to make processes work like they should. Piped processes will yield data cooperatively, instead of waiting for the innermost process to finish, consuming its output, waiting for the next process to finish, consuming its output, etc.

I'm still hammering out the bugs on that branch though, so until then, I guess the advice would have to be not to pipe >64k of data around. I can't even patch this with a quick fix because it seems to be fundamental to how Python's subprocess module works.

from sh.

fixed in master, v1.0

from sh.