GitLab

mentioned in merge request !6 (merged)

So that I understand, the plan about this TODO, it would be something more or less like this ?

index 838cbec..5f40379 100755
--- a/nxdtest/__init__.py
+++ b/nxdtest/__init__.py
@@ -58,7 +58,7 @@ from __future__ import print_function, absolute_import
 from erp5.util.taskdistribution import TaskDistributor
 from subprocess import Popen, PIPE
 from time import time, strftime, gmtime, localtime
-import os, sys, argparse, logging, traceback, re, pwd, socket
+import os, sys, argparse, logging, traceback, re, pwd, socket, fcntl, select
 import six
 from golang import b
 from golang import context, sync
@@ -267,6 +267,12 @@ def main():
 
 # tee, similar to tee(1) utility, copies data from fin to fout appending them to buf.
 def tee(ctx, fin, fout, buf):
+    # set file descriptors non blocking
+    for f in (fin, fout):
+        fd = f.fileno()
+        fl = fcntl.fcntl(fd, fcntl.F_GETFL)
+        fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
+
     while 1:
         # poll for cancellation periodically and stop if requested.
         # FIXME handle ↓↓↓ os.read and .write in non-blocking way so that we
@@ -282,13 +288,23 @@ def tee(ctx, fin, fout, buf):
         #
         #   besides when a lot of output is available it would be a waste to
         #   read/flush it line-by-line. )
-        data = os.read(fin.fileno(), 4096)
-        if not(data):
-            return  # EOF
-
-        fout.write(data)
-        fout.flush()
-        buf.append(data)
+        data = None
+        readable, _, _ = select.select([fin], [], [], 0)
+        if fin in readable:
+            data = os.read(fin.fileno(), 4096)
+            if not(data):
+                return  # EOF
+            while 1:
+                e = ctx.err()
+                if e is not None:
+                    raise e
+                _, writable, _ = select.select([], [fout], [], 0)
+                if fout in writable:
+                    fout.write(data)
+                    fout.flush()
+                    break
+
+            buf.append(data)

I'm not saying "let's do this", some tests are failing because they use BytesIO (with pytest's capsys), I just want to make sure I generally understood the idea.

@jerome, yes, you understood this idea correctly.

My preference is 1) indeed "let's not do this now" and 2) for library support for such IO to reside in pygolang runtime and be represented to user as e.g. xio.read(ctx, f) and xio.write(ctx, f, data) utilities. Note also that in general it is not as easy as it sounds: on Linux select/epoll always reports IO as being ready for regular files, and it is only sockets and pipes that support polling for real.

Once again, even if with adding this TODO/FIXME in place we are not doing any step backwards: the previous version of the code could also block forever in read/write - only there was no any cancellation attempt nor any comment about that.

P.S. for IO that supports contexts on Go side please see e.g. https://pkg.go.dev/lab.nexedi.com/kirr/go123@v0.0.0-20201018154837-96046edf8211/xio , kirr/go123@7ad867a3 and kirr/go123@d2dc6c09.

P.P.S. standard Go runtime automatically supports polling for sockets and pipes, but there is no (yet ?) user-visible API for contexts.

https://golang.org/doc/go1.11#os
https://golang.org/pkg/internal/poll/

Switch tee from threading.Thread to sync.WorkGroup