Don't drop any trailing data after multipliexing; redirect to stderr

[Rob Browning]

See the commit messages for further information.

Pushed to main.

--
Rob Browning
rlb @defaultvalue.org and @debian.org
GPG as of 2011-07-10 E6A9 DA3C C9FD 1FF8 C676 D2C4 C0F0 39E9 ED1B 597A
GPG as of 2002-11-03 14DD 432F AE39 534D B592 F9A0 25C8 D377 8C7E 73A4

[Rob Browning]

Keep multiplexing until both of the input streams reach EOF, and stop
paying attention to the process status. Leave the process lifetime up
to the caller.

This avoids losing a backtrace, for example if the output stream ends
before the error stream.

Thanks to Johannes Berg for help fixing some issues in previous
versions.

Signed-off-by: Rob Browning <r...@defaultvalue.org>
Tested-by: Rob Browning <r...@defaultvalue.org>
---
lib/bup/cmd/mux.py | 2 +-
lib/bup/helpers.py | 16 ++++++++++------
2 files changed, 11 insertions(+), 7 deletions(-)

diff --git a/lib/bup/cmd/mux.py b/lib/bup/cmd/mux.py
index 1ae43603..9d7ddea5 100644
--- a/lib/bup/cmd/mux.py
+++ b/lib/bup/cmd/mux.py
@@ -43,7 +43,7 @@ def main(argv):
out = byte_stream(sys.stdout)
out.write(b'BUPMUX')
out.flush()
- mux(p, out.fileno(), outr, errr)
+ mux(out.fileno(), outr, errr)
os.close(outr)
os.close(errr)
prv = p.wait()
diff --git a/lib/bup/helpers.py b/lib/bup/helpers.py
index e7c56e15..bc36e19f 100644
--- a/lib/bup/helpers.py
+++ b/lib/bup/helpers.py
@@ -524,20 +524,24 @@ def checked_reader(fd, n):


MAX_PACKET = 128 * 1024
-def mux(p, outfd, outr, errr):
+def mux(outfd, outr, errr):
try:
fds = [outr, errr]
- while p.poll() is None:
+ while fds:
rl, _, _ = select.select(fds, [], [])
for fd in rl:
if fd == outr:
buf = os.read(outr, MAX_PACKET)
- if not buf: break
- os.writev(outfd, (struct.pack('!IB', len(buf), 1), buf))
+ if not buf:
+ fds.remove(outr)
+ else:
+ os.writev(outfd, (struct.pack('!IB', len(buf), 1), buf))
elif fd == errr:
buf = os.read(errr, 1024)
- if not buf: break
- os.writev(outfd, (struct.pack('!IB', len(buf), 2), buf))
+ if not buf:
+ fds.remove(errr)
+ else:
+ os.writev(outfd, (struct.pack('!IB', len(buf), 2), buf))
finally:
os.write(outfd, struct.pack('!IB', 0, 3))

--
2.47.3

[Rob Browning]

After we're finished multiplexing, write any further input data to
stderr. This might include debug messages or backtraces or... For
example, anything after the mux() in bup.cmd.mux, including exceptions
that make it out of main.

For now, just do this unconditionally in the DemuxConn itself since
there are only two dependants (bup demux, and client ViaBup), and it
should be fine for both.

The omission of this was hiding an exception that caused intermittent
test-on failures, leaving the only evidence an exit status of 120,
which is what python does when it tries to flush stdout/stderr on exit
and can't: https://bugs.python.org/issue5319

Thanks to Johannes Berg for finding the cause, and help devising the
solution.

Signed-off-by: Rob Browning <r...@defaultvalue.org>
Tested-by: Rob Browning <r...@defaultvalue.org>
---

lib/bup/helpers.py | 15 ++++++++++++++-
1 file changed, 14 insertions(+), 1 deletion(-)

diff --git a/lib/bup/helpers.py b/lib/bup/helpers.py
index fda49211..18402d4a 100644
--- a/lib/bup/helpers.py
+++ b/lib/bup/helpers.py
@@ -552,7 +552,11 @@ def mux(outfd, outr, errr):


class DemuxConn(BaseConn):
- """A helper class for bup's client-server protocol."""
+ """A helper class for bup's client-server protocol. For now, it
+ always assumes it takes responsbility for all of the remaining
+ content in infd, and infd will be at EOF after the last read.
+
+ """
def __init__(self, infd, outp):
BaseConn.__init__(self, outp)
# Anything that comes through before the sync string was not
@@ -603,6 +607,15 @@ class DemuxConn(BaseConn):
elif fdw == 3:
self.closed = True
debug2("DemuxConn: marked closed\n")
+ # Write any remaining output to stderr. This might
+ # include debug messages or backtraces or... For example,
+ # anything after the mux() in bup.cmd.mux, including
+ # exceptions that make it out of main.
+ buf = os.read(self.infd, 1024)
+ while buf:
+ sys.stderr.buffer.write(buf)
+ buf = os.read(self.infd, 1024)
+ sys.stderr.buffer.flush()
return True

def _load_buf(self, timeout):
--
2.47.3

[Rob Browning]

Move sending the "BUPMUX" start indicator into mux() so that it's
responsible for both starting and ending the multiplexed stream, and
can ensure that whenever we start multiplexing, we always eventually
end it too.

This means that any output generated after a mux() call (say a
backtrace that made it to the top level) won't be at risk of being
written to a stream that's still multiplexed from the receiver's
perspective.

This is relevant when "bup mux" is invoked by on--server, because
on--server combines stderr and stdout before exec'ing mux. In
practice, it wasn't a problem because we always called mux() in
bup.cmd.mux (and entered the try/finally) immediately after sending
BUPMUX.

Signed-off-by: Rob Browning <r...@defaultvalue.org>
Tested-by: Rob Browning <r...@defaultvalue.org>
---

lib/bup/cmd/mux.py | 2 --
lib/bup/helpers.py | 5 +++++
2 files changed, 5 insertions(+), 2 deletions(-)

diff --git a/lib/bup/cmd/mux.py b/lib/bup/cmd/mux.py
index 9d7ddea5..eeb5a346 100644
--- a/lib/bup/cmd/mux.py
+++ b/lib/bup/cmd/mux.py
@@ -41,8 +41,6 @@ def main(argv):
os.close(errw)
sys.stdout.flush()
out = byte_stream(sys.stdout)
- out.write(b'BUPMUX')
- out.flush()

mux(out.fileno(), outr, errr)
os.close(outr)
os.close(errr)

diff --git a/lib/bup/helpers.py b/lib/bup/helpers.py
index bc36e19f..fda49211 100644
--- a/lib/bup/helpers.py
+++ b/lib/bup/helpers.py
@@ -525,6 +525,11 @@ def checked_reader(fd, n):

MAX_PACKET = 128 * 1024

def mux(outfd, outr, errr):
+ """Multiplex fds outr and errr onto outfd until exhausted. Always
+ terminate the multiplexed "session" before returning.
+
+ """
+ os.write(outfd, b'BUPMUX')

try:
fds = [outr, errr]

while fds:
--
2.47.3