runner scalability adjustments from applied testing

chanzuckerberg · Nov 25, 2019 · e149b44 · e149b44
1 parent ad90d50
commit e149b44
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 11 deletions.
diff --git a/WDL/CLI.py b/WDL/CLI.py
@@ -433,7 +433,8 @@ def runner(
 
     # configuration
     run_kwargs = dict(
-        (k, kwargs[k]) for k in ["copy_input_files", "run_dir", "runtime_cpu_max", "as_me"]
+        (k, kwargs[k])
+        for k in ["copy_input_files", "run_dir", "runtime_cpu_max", "as_me", "max_tasks"]
     )
     if runtime_memory_max:
         run_kwargs["runtime_memory_max"] = parse_byte_size(runtime_memory_max)

diff --git a/WDL/runtime/task.py b/WDL/runtime/task.py
@@ -10,6 +10,7 @@
 import traceback
 import glob
 import time
+import random
 import multiprocessing
 import threading
 import shutil
@@ -317,11 +318,15 @@ def _run(
             # stream stderr into log
             with PygtailLogger(logger, os.path.join(self.host_dir, "stderr.txt")) as poll_stderr:
                 # poll for container exit
+                running = False
                 while exit_code is None:
-                    time.sleep(1)
+                    time.sleep(random.uniform(1.0, 2.0))  # spread out work over the GIL
                     if terminating():
                         raise Terminated() from None
                     if "running" in self._observed_states:
+                        if not running:
+                            logger.notice("container running")  # pyre-fixme
+                            running = True
                         poll_stderr()
                     exit_code = self.poll_service(logger, svc)
                 logger.debug(
@@ -478,14 +483,24 @@ def chown(self, logger: logging.Logger, client: docker.DockerClient) -> None:
             chown -RP {os.geteuid()}:{os.getegid()} {shlex.quote(os.path.join(self.container_dir, 'work'))}
             """.strip()
             volumes = {self.host_dir: {"bind": self.container_dir, "mode": "rw"}}
+            logger.debug(_("post-task chown", script=script, volumes=volumes))
             try:
-                logger.debug(_("post-task chown", script=script, volumes=volumes))
-                client.containers.run(
-                    "alpine:3",
-                    command=["/bin/ash", "-c", script],
-                    volumes=volumes,
-                    auto_remove=True,
-                )
+                chowner = None
+                try:
+                    chowner = client.containers.run(
+                        "alpine:3",
+                        command=["/bin/ash", "-c", script],
+                        volumes=volumes,
+                        detach=True,
+                    )
+                    chowner_status = chowner.wait()
+                    assert (
+                        isinstance(chowner_status, dict)
+                        and chowner_status.get("StatusCode", -1) == 0
+                    ), str(chowner_status)
+                finally:
+                    if chowner:
+                        chowner.remove()
             except:
                 logger.exception("post-task chown failed")
 

diff --git a/tests/test_4taskrun.py b/tests/test_4taskrun.py
@@ -145,8 +145,8 @@ def test_logging_std_err(self, capture):
         for record in std_error_msgs:
             line_written = int(record.msg.split('=')[1])
             self.assertGreater(record.created, line_written)
-            # check line logged within 3 seconds of being written
-            self.assertGreater(line_written+3, record.created)
+            # check line logged within 4 seconds of being written
+            self.assertGreater(line_written+4, record.created)
 
     @log_capture()
     def test_logging_std_err_captures_full_line(self, capture):