the execure optimization

Author: zhenwei-li

src/com/dvsnier/process/__init__.py

@@ -0,0 +1 @@
+# -*- coding:utf-8 -*-

src/com/dvsnier/process/execute.py

@@ -1,50 +1,60 @@
 # -- coding:utf-8 --
 
 import datetime
-# import logging
-import os
-import platform
-import time
-import sys
+import logging
 import subprocess
+import time
 
 from com.dvsnier.directory.common_dir import generate_complex_file_name
 
 
 def execute(cmds, quiet=True):
     'the process execute that a command'
+    content = ''
+    stdouts = None
     start = time.time()
-    p = subprocess.Popen(cmds[0], stdout=subprocess.PIPE, shell=True)
-    # logging.debug('the current sub process pid(cwd: %s, ppid: %s, id: %s%d).' %
-    #               (p.pid, os.getppid(), type(p), id(p)))
-    processes = [p]
-    for x in cmds[1:]:
-        p = subprocess.Popen(x, stdin=p.stdout, stdout=subprocess.PIPE, shell=True)
-        # logging.debug(
-        #     'the current sub process pid(cwd: %s, ppid: %s, id: %s%d).' %
-        #     (p.pid, os.getppid(), type(p), id(p)))
-        # logging.debug(type(p), id(p))
-        processes.append(p)
-    # logging.debug('the current run process pid(cwd: %s, ppid: %s, id: %s%d).' %
-    #               (p.pid, os.getppid(), type(p), id(p)))
-    # logging.debug(type(p), id(p))
-    output = p.communicate()[0]
-    for p in processes:
-        p.wait()
+    with subprocess.Popen(cmds[0], stdout=subprocess.PIPE, shell=True) as p:
+        # logging.debug('the current sub process pid(cwd: %s, ppid: %s, id: %s%d).' %
+        #               (p.pid, os.getppid(), type(p), id(p)))
+        processes = [p]
+        for x in cmds[1:]:
+            #
+            # the airticle link reference:
+            #
+            # 1. https://docs.python.org/2.7/library/subprocess.html#subprocess.Popen.communicate
+            # 2. https://docs.python.org/3/library/subprocess.html#subprocess.Popen.communicate
+            # 3. https://stackoverflow.com/questions/58649679/resourcewarning-unclosed-file-io-bufferedreader-name-4
+            # 4. https://python.readthedocs.io/en/stable/library/subprocess.html
+            #
+            p = subprocess.Popen(x, stdin=p.stdout, stdout=subprocess.PIPE, bufsize=1024, shell=True)
+            # logging.debug(
+            #     'the current sub process pid(cwd: %s, ppid: %s, id: %s%d).' %
+            #     (p.pid, os.getppid(), type(p), id(p)))
+            # logging.debug(type(p), id(p))
+            processes.append(p)
+            # logging.debug('the current run process pid(cwd: %s, ppid: %s, id: %s%d).' %
+            #               (p.pid, os.getppid(), type(p), id(p)))
+            # logging.debug(type(p), id(p))
+        stdouts, stderrs = p.communicate()
+        if stderrs:
+            logging.error(stderrs)
+        for p in processes:
+            p.kill()
+            p.wait()
 
     end = time.time()
     if not quiet:
-        if platform.system() == 'Linux' and os.isatty(1):
-            print('\r'),
         msg = '[%.5f] -> %s' % (end - start, ' | '.join(cmds))
-        print(msg)
-    content = None
-    if sys.version_info.major > 2:
-        content = str(output.rstrip(bytes('\n', encoding='utf-8')), encoding='utf-8')
-    else:
-        content = output.rstrip('\n')
-    # logging.debug('the current run process pid(cwd: %s, ppid: %s, id: %s%d).' %
-    #               (p.pid, os.getppid(), type(p), id(p)))
+        logging.debug(msg)
+    if stdouts:
+        if isinstance(stdouts, str):
+            content = str(stdouts).rstrip('\n')
+        elif isinstance(stdouts, bytes):
+            content = str(stdouts.rstrip(bytes('\n', encoding='utf-8')), encoding='utf-8')
+        else:
+            content = ''
+        # logging.debug('the current run process pid(cwd: %s, ppid: %s, id: %s%d).' %
+        #               (p.pid, os.getppid(), type(p), id(p)))
     return content
 
 

tests/com/dvsnier/process/test_execute.py

@@ -1,5 +1,6 @@
 # -*- coding:utf-8 -*-
 
+# import tracemalloc
 import unittest
 
 from com.dvsnier.process.execute import execute, trace
@@ -10,35 +11,50 @@ class Test_Execute(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         print("...the set up...")
-        print
+        print('')
 
     def setUp(self):
-        return super(Test_Execute, self).setUp()
+        super(Test_Execute, self).setUp()
 
     def test_execute(self):
         'the test execute'
+        # self.debug_trace_malloc_start()
         # cmds = ["git config --local --list"]
         # cmds = ["git config --local --list", "grep \"user\""]
         cmds = ["git config --local --list", "grep \"user\"", "grep \"@\""]
         result = execute(cmds, quiet=False)
         self.assertIsNotNone(result, 'test_execute is error.')
-        print(result)
+        # self.debug_trace_malloc_end()
 
     def test_trace(self):
         'the test trace'
+        # self.debug_trace_malloc_start()
         # cmds = ["git config --local --list"]
         # cmds = ["git config --local --list", "grep \"user\""]
         cmds = ["git config --local --list", "grep \"user\"", "grep \"@\""]
         trace(cmds)
+        # self.debug_trace_malloc_end()
+
+    # def debug_trace_malloc_start(self):
+    #     tracemalloc.start()
+
+    # def debug_trace_malloc_end(self):
+    #     snapshot = tracemalloc.take_snapshot()
+    #     top_stats = snapshot.statistics('lineno')
+    #     print("[ Top 10 ]")
+    #     for stat in top_stats[:10]:
+    #         print(stat)
 
     def tearDown(self):
-        return super(Test_Execute, self).tearDown()
+        super(Test_Execute, self).tearDown()
 
     @classmethod
     def tearDownClass(cls):
-        print
+        print('')
         print("...the tear down...")
 
 
 if __name__ == '__main__':
-    unittest.main()
+    suite = unittest.TestLoader().loadTestsFromTestCase(Test_Execute)
+    unittest.TextTestRunner(verbosity=2).run(suite)
+    # unittest.main()