Merge pull request #160 from keurfonluu/devel

Handle duplicate connection outputs in TOUGH3 and zombie subprocesses

README.rst

@@ -159,7 +159,7 @@ This Software was developed under funding from the U.S. Department of Energy and
 .. |License| image:: https://img.shields.io/badge/license-BSD--3--Clause-green
    :target: https://github.com/keurfonluu/toughio/blob/master/LICENSE
 
-.. |Stars| image:: https://img.shields.io/github/stars/keurfonluu/toughio?logo=github
+.. |Stars| image:: https://img.shields.io/github/stars/keurfonluu/toughio?style=flat&logo=github
    :target: https://github.com/keurfonluu/toughio
 
 .. |Pyversions| image:: https://img.shields.io/pypi/pyversions/toughio.svg?style=flat

setup.cfg

@@ -31,6 +31,7 @@ install_requires =
     importlib_metadata
     numpy >= 1.13.0
     meshio >= 5.2, < 6.0
+    psutil >= 5.0
 python_requires = >= 3.7
 setup_requires =
     setuptools >= 42

tests/test_run.py

@@ -28,6 +28,7 @@ def test_run(exec, workers, docker, wsl, cmd):
         use_temp=sys.version_info >= (3, 8),
         ignore_patterns=["INFILE"],
         silent=True,
+        container_name="CONTAINER",
     )
 
     assert status.args == cmd
@@ -44,15 +45,15 @@ def test_run(exec, workers, docker, wsl, cmd):
             None,
             "docker-image",
             False,
-            "docker run --rm  -v PLACEHOLDER:/shared -w /shared docker-image tough-exec INFILE INFILE.out",
+            "docker run --name CONTAINER --rm --volume PLACEHOLDER:/shared --workdir /shared docker-image tough-exec INFILE INFILE.out",
         ),
         ("tough-exec", None, None, True, "bash -c 'tough-exec INFILE INFILE.out'"),
         (
             "tough-exec",
             8,
             "docker-image",
             True,
-            """bash -c 'docker run --rm  -v PLACEHOLDER:/shared -w /shared docker-image mpiexec -n 8 tough-exec INFILE INFILE.out'""",
+            """bash -c 'docker run --name CONTAINER --rm --volume PLACEHOLDER:/shared --workdir /shared docker-image mpiexec -n 8 tough-exec INFILE INFILE.out'""",
         ),
     ],
 )
@@ -76,13 +77,13 @@ def test_run_windows(exec, workers, docker, wsl, cmd):
             "tough-exec",
             None,
             "docker-image",
-            "docker run --rm PLACEHOLDER -v ${PWD}:/shared -w /shared docker-image tough-exec INFILE INFILE.out",
+            "docker run --name CONTAINER --rm --volume ${PWD}:/shared --workdir /shared docker-image tough-exec INFILE INFILE.out",
         ),
         (
             "tough-exec",
             8,
             "docker-image",
-            "docker run --rm PLACEHOLDER -v ${PWD}:/shared -w /shared docker-image mpiexec -n 8 tough-exec INFILE INFILE.out",
+            "docker run --name CONTAINER --rm --volume ${PWD}:/shared --workdir /shared docker-image mpiexec -n 8 tough-exec INFILE INFILE.out",
         ),
     ],
 )

toughio/VERSION

@@ -1 +1 @@
-1.14.0
+1.14.1

toughio/_io/output/_common.py

@@ -1,4 +1,5 @@
 import collections
+import logging
 
 import numpy as np
 
@@ -21,6 +22,40 @@ def to_output(file_type, labels_order, headers, times, labels, variables):
         )
         for time, label, variable in zip(times, labels, variables)
     ]
+
+    # Some older versions of TOUGH3 have duplicate connection outputs when running in parallel
+    # Fix the outputs here by summing the duplicate connections
+    if file_type == "connection" and len(labels[0]):
+        # Check whether there are duplicate connections
+        connections = {}
+        found_duplicate = False
+
+        for i, (c1, c2) in enumerate(outputs[0].labels):
+            if (c1, c2) in connections:
+                connections[(c1, c2)].append(i)
+                found_duplicate = True
+
+            else:
+                connections[(c1, c2)] = [i]
+
+        if found_duplicate:
+            logging.warning(
+                "Found duplicate connections. Fixing outputs by summing duplicate connections."
+            )
+
+            outputs = [
+                Output(
+                    output.type,
+                    output.time,
+                    np.array(list(connections)),
+                    {
+                        k: np.array([v[idx].sum() for idx in connections.values()])
+                        for k, v in output.data.items()
+                    },
+                )
+                for output in outputs
+            ]
+
     return (
         [reorder_labels(out, labels_order) for out in outputs]
         if labels_order is not None and file_type == "element"

toughio/_run.py

@@ -2,10 +2,14 @@
 import os
 import pathlib
 import platform
+import secrets
 import shutil
+import signal
 import subprocess
 import tempfile
 
+import psutil
+
 _check_exec = True  # Bool to be monkeypatched in tests
 
 
@@ -23,6 +27,7 @@ def run(
     silent=False,
     petsc_args=None,
     docker_args=None,
+    container_name=None,
     **kwargs,
 ):
     """
@@ -56,6 +61,8 @@ def run(
         List of arguments passed to PETSc solver (written to `.petscrc`).
     docker_args : list or None, optional, default None
         List of arguments passed to `docker run` command.
+    container_name : str or None, optional, default None
+        Name of Docker container.
 
     Other Parameters
     ----------------
@@ -219,39 +226,41 @@ def run(
     is_windows = platform.system().startswith("Win")
 
     if docker:
+        container_name = (
+            container_name if container_name else f"toughio_{secrets.token_hex(4)}"
+        )
+
         if is_windows and os.getenv("ComSpec").endswith("cmd.exe"):
             cwd = '"%cd%"'
 
         else:
             cwd = "${PWD}"
 
-        try:
-            uid = f"-e LOCAL_USER_ID={os.getuid()}"
-
-        except AttributeError:
-            uid = ""
-
         docker_args = docker_args if docker_args else []
         docker_args += [
+            "--name",
+            container_name,
             "--rm",
-            uid,
-            "-v",
+            # Sometime raises a duplicate mount point error, use old-school volume instead (but shell must be True in this case)
+            # "--mount",
+            # f"type=bind,source={simulation_dir},target=/shared",
+            "--volume",
             f"{cwd}:/shared",
-            "-w",
+            "--workdir",
             "/shared",
         ]
-        cmd = f"docker run {' '.join(docker_args)} {docker} {cmd}"
+        cmd = f"docker run {' '.join(str(arg) for arg in docker_args)} {docker} {cmd}"
 
     # Use WSL
     if wsl and is_windows:
         cmd = f"bash -c '{cmd}'"
 
     # Run simulation
-    if not silent:
+    try:
         # See <https://www.koldfront.dk/making_subprocesspopen_in_python_3_play_nice_with_elaborate_output_1594>
         p = subprocess.Popen(
             cmd,
-            shell=True,
+            shell=True,  # shell must be True as the command may contain quotes
             cwd=str(simulation_dir),
             stderr=subprocess.STDOUT,
             stdout=subprocess.PIPE,
@@ -266,25 +275,41 @@ def run(
             # This is only an issue for Spyder
             line = f"\r{line}" if cr else line
             cr = line.endswith("\r")
-
             line = line[:-2] if cr else line
-            print(line, end="", flush=True)
-            stdout.append(line)
 
-        status = subprocess.CompletedProcess(
-            args=p.args,
-            returncode=0,
-            stdout="".join(stdout),
-        )
+            if not silent:
+                print(line, end="", flush=True)
+                stdout.append(line)
 
-    else:
-        status = subprocess.run(
-            cmd,
-            shell=True,
-            cwd=str(simulation_dir),
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.STDOUT,
-        )
+    except (KeyboardInterrupt, Exception) as e:
+        # Stop Docker container
+        if docker:
+            status = subprocess.run(
+                ["docker", "stop", container_name],
+                stdout=subprocess.PIPE,
+                universal_newlines=True,
+            )
+
+        # Handle children process termination
+        # See <https://stackoverflow.com/a/25134985/9729313>
+        try:
+            proc = psutil.Process(p.pid)
+            for child_process in proc.children():
+                child_process.send_signal(signal.SIGTERM)
+
+            proc.send_signal(signal.SIGTERM)
+
+        except psutil.NoSuchProcess:
+            pass
+
+        raise e
+
+    p.wait()
+    status = subprocess.CompletedProcess(
+        args=p.args,
+        returncode=p.returncode,
+        stdout="".join(stdout),
+    )
 
     # Copy files from temporary directory and delete it
     if use_temp: