From 55006119c8fb2cd62d51d7be2cc0508dd1ebc671 Mon Sep 17 00:00:00 2001
From: Robert Jordens <jordens@gmail.com>
Date: Mon, 1 Feb 2016 01:52:31 -0700
Subject: [PATCH] subprocesses: unify termination logic

---
 artiq/devices/ctlmgr.py | 54 ++++++++++++++++++++++++-----------------
 artiq/master/worker.py  | 39 +++++++++++++++++------------
 2 files changed, 55 insertions(+), 38 deletions(-)

diff --git a/artiq/devices/ctlmgr.py b/artiq/devices/ctlmgr.py
index 83295cf69..785f67267 100644
--- a/artiq/devices/ctlmgr.py
+++ b/artiq/devices/ctlmgr.py
@@ -106,30 +106,40 @@ class Controller:
             await self._terminate()
 
     async def _terminate(self):
-        logger.info("Terminating controller %s", self.name)
-        if self.process is not None and self.process.returncode is None:
+        if self.process is None or self.process.returncode is not None:
+            logger.info("Controller %s already terminated", self.name)
+            return
+        logger.debug("Terminating controller %s", self.name)
+        try:
+            await asyncio.wait_for(self.call("terminate"), self.term_timeout)
+            await asyncio.wait_for(self.process.wait(), self.term_timeout)
+            logger.info("Controller %s terminated", self.name)
+            return
+        except:
+            logger.warning("Controller %s did not exit on request, "
+                           "ending the process", self.name)
+        if os.name != "nt":
             try:
-                await asyncio.wait_for(self.call("terminate"),
-                                       self.term_timeout)
-            except:
-                logger.warning("Controller %s did not respond to terminate "
-                               "command, killing", self.name)
-                try:
-                    self.process.kill()
-                except ProcessLookupError:
-                    pass
+                self.process.terminate()
+            except ProcessLookupError:
+                pass
             try:
-                await asyncio.wait_for(self.process.wait(),
-                                       self.term_timeout)
-            except:
-                logger.warning("Controller %s failed to exit, killing",
-                               self.name)
-                try:
-                    self.process.kill()
-                except ProcessLookupError:
-                    pass
-                await self.process.wait()
-        logger.debug("Controller %s terminated", self.name)
+                await asyncio.wait_for(self.process.wait(), self.term_timeout)
+                logger.info("Controller process %s terminated", self.name)
+                return
+            except asyncio.TimeoutError:
+                logger.warning("Controller process %s did not terminate, "
+                               "killing", self.name)
+        try:
+            self.process.kill()
+        except ProcessLookupError:
+            pass
+        try:
+            await asyncio.wait_for(self.process.wait(), self.term_timeout)
+            logger.info("Controller process %s killed", self.name)
+            return
+        except asyncio.TimeoutError:
+            logger.warning("Controller process %s failed to die", self.name)
 
 
 def get_ip_addresses(host):
diff --git a/artiq/master/worker.py b/artiq/master/worker.py
index fb1752aed..12d05c796 100644
--- a/artiq/master/worker.py
+++ b/artiq/master/worker.py
@@ -115,30 +115,37 @@ class Worker:
                                    " (RID %s)", self.ipc.process.returncode,
                                    self.rid)
                 return
-            obj = {"action": "terminate"}
             try:
-                await self._send(obj, cancellable=False)
+                await self._send({"action": "terminate"}, cancellable=False)
+                await asyncio.wait_for(self.ipc.process.wait(), term_timeout)
+                logger.debug("worker exited on request (RID %s)", self.rid)
+                return
             except:
-                logger.debug("failed to send terminate command to worker"
-                             " (RID %s), killing", self.rid, exc_info=True)
+                logger.debug("worker failed to exit on request"
+                             " (RID %s), ending the process", self.rid,
+                             exc_info=True)
+            if os.name != "nt":
                 try:
-                    self.ipc.process.kill()
+                    self.ipc.process.terminate()
                 except ProcessLookupError:
                     pass
-                await self.ipc.process.wait()
-                return
+                try:
+                    await asyncio.wait_for(self.ipc.process.wait(), term_timeout)
+                    logger.debug("worker terminated (RID %s)", self.rid)
+                    return
+                except asyncio.TimeoutError:
+                    logger.warning("worker did not terminate (RID %s), killing",
+                                   self.rid)
+            try:
+                self.ipc.process.kill()
+            except ProcessLookupError:
+                pass
             try:
                 await asyncio.wait_for(self.ipc.process.wait(), term_timeout)
+                logger.debug("worker killed (RID %s)", self.rid)
+                return
             except asyncio.TimeoutError:
-                logger.debug("worker did not exit by itself (RID %s), killing",
-                             self.rid)
-                try:
-                    self.ipc.process.kill()
-                except ProcessLookupError:
-                    pass
-                await self.ipc.process.wait()
-            else:
-                logger.debug("worker exited by itself (RID %s)", self.rid)
+                logger.warning("worker refuses to die (RID %s)", self.rid)
         finally:
             self.io_lock.release()