Repository: noxdafox/pebble

Issue:
Include worker PID in ProcessExpired exception
👋 , we are utilizing the pebble `ProcessPool` quite heavily and occasionally we'll find ourselves needing to debug an abnormal exit on a worker. When `ProcessExpired` is raised in the main processes `done_callback` we get the exit code, which is helpful, but lack the process id of the worker which would make finding logs and sourcing the underlying reason much easier.

Ask:
 - Can we update `handle_worker_expiration` to include the `worker_id` in the `ProcessExpired` exception [here](https://github.com/noxdafox/pebble/blob/master/pebble/pool/process.py#L267)?

Discussion:
@noxdafox: Hello,

thanks for reporting this issue. I will add the `PID` onto the exception.@charettes: Thanks for your consideration @noxdafox 🙇 Would submitting a PR with tests tweaks ourselves be appreciated or you'd prefer to tackle the feature request yourself?

I see there's [already a test we could adjust for this purpose](https://github.com/noxdafox/pebble/blob/6445f693bc4449bd0467c30af6fa7f4729404163/test/test_process_pool_fork.py#L426-L430)@noxdafox: Issue resolved in `5.1.0`.@charettes: Thanks @noxdafox!

Solution patch:
diff --git a/doc/index.rst b/doc/index.rst
index f703376..696ae4d 100644
--- a/doc/index.rst
+++ b/doc/index.rst
@@ -269,6 +269,10 @@ Pebble aims to help managing threads and processes in an easier way. It wraps Py
 
       Integer representing the process' exit code.
 
+   .. data:: pid
+
+      Integer representing the process identifier or `None`.
+
 
 Programming Guidelines
 ----------------------
diff --git a/pebble/asynchronous/process.py b/pebble/asynchronous/process.py
index 1a72478..0480c9c 100644
--- a/pebble/asynchronous/process.py
+++ b/pebble/asynchronous/process.py
@@ -121,6 +121,7 @@ async def _worker_handler(
     else:
         if result.status == common.ResultStatus.ERROR:
             result.value.exitcode = worker.exitcode
+            result.value.pid = worker.pid
         if not isinstance(result.value, asyncio.CancelledError):
             future.set_exception(result.value)
 
diff --git a/pebble/common/types.py b/pebble/common/types.py
index 224c670..5f4df8e 100644
--- a/pebble/common/types.py
+++ b/pebble/common/types.py
@@ -22,9 +22,10 @@ from concurrent.futures import Future
 
 class ProcessExpired(OSError):
     """Raised when process dies unexpectedly."""
-    def __init__(self, msg, code=0):
+    def __init__(self, msg, code=0, pid=None):
         super(ProcessExpired, self).__init__(msg)
         self.exitcode = code
+        self.pid = pid
 
 
 class PebbleFuture(Future):
diff --git a/pebble/concurrent/process.py b/pebble/concurrent/process.py
index 6e4ce5f..3617132 100644
--- a/pebble/concurrent/process.py
+++ b/pebble/concurrent/process.py
@@ -122,6 +122,7 @@ def _worker_handler(
     else:
         if result.status == common.ResultStatus.ERROR:
             result.value.exitcode = worker.exitcode
+            result.value.pid = worker.pid
         if not isinstance(result.value, CancelledError):
             future.set_exception(result.value)
 
diff --git a/pebble/pool/process.py b/pebble/pool/process.py
index 3694d3e..05f8979 100644
--- a/pebble/pool/process.py
+++ b/pebble/pool/process.py
@@ -264,7 +264,7 @@ class PoolManager:
         except LookupError:
             return
         else:
-            error = ProcessExpired('Abnormal termination', code=exitcode)
+            error = ProcessExpired('Abnormal termination', code=exitcode, pid=worker_id)
             self.task_manager.task_done(
                 task.id, Result(ResultStatus.ERROR, error))
 


Test patch:
diff --git a/test/test_asynchronous_process_fork.py b/test/test_asynchronous_process_fork.py
index 7002f3a..564e1c4 100644
--- a/test/test_asynchronous_process_fork.py
+++ b/test/test_asynchronous_process_fork.py
@@ -358,8 +358,10 @@ class TestProcessAsynchronous(unittest.TestCase):
         async def test():
             return await critical_decorated()
 
-        with self.assertRaises(ProcessExpired):
+        with self.assertRaises(ProcessExpired) as exc_ctx:
             asyncio.run(test())
+        self.assertEqual(exc_ctx.exception.exitcode, 123)
+        self.assertIsInstance(exc_ctx.exception.pid, int)
 
     def test_timeout_decorated_callback(self):
         """Process Fork ProcessExpired is forwarded to callback."""
diff --git a/test/test_asynchronous_process_forkserver.py b/test/test_asynchronous_process_forkserver.py
index a6c5877..eecebb1 100644
--- a/test/test_asynchronous_process_forkserver.py
+++ b/test/test_asynchronous_process_forkserver.py
@@ -341,8 +341,10 @@ class TestProcessAsynchronous(unittest.TestCase):
         async def test():
             return await critical_decorated()
 
-        with self.assertRaises(ProcessExpired):
+        with self.assertRaises(ProcessExpired) as exc_ctx:
             asyncio.run(test())
+        self.assertEqual(exc_ctx.exception.exitcode, 123)
+        self.assertIsInstance(exc_ctx.exception.pid, int)
 
     def test_timeout_decorated_callback(self):
         """Process Forkserver ProcessExpired is forwarded to callback."""
diff --git a/test/test_asynchronous_process_spawn.py b/test/test_asynchronous_process_spawn.py
index af5c093..bc0b0ea 100644
--- a/test/test_asynchronous_process_spawn.py
+++ b/test/test_asynchronous_process_spawn.py
@@ -341,8 +341,10 @@ class TestProcessAsynchronous(unittest.TestCase):
         async def test():
             return await critical_decorated()
 
-        with self.assertRaises(ProcessExpired):
+        with self.assertRaises(ProcessExpired) as exc_ctx:
             asyncio.run(test())
+        self.assertEqual(exc_ctx.exception.exitcode, 123)
+        self.assertIsInstance(exc_ctx.exception.pid, int)
 
     def test_timeout_decorated_callback(self):
         """Process Spawn ProcessExpired is forwarded to callback."""
diff --git a/test/test_concurrent_process_fork.py b/test/test_concurrent_process_fork.py
index 5d0dc5d..8dbffda 100644
--- a/test/test_concurrent_process_fork.py
+++ b/test/test_concurrent_process_fork.py
@@ -294,8 +294,10 @@ class TestProcessConcurrent(unittest.TestCase):
     def test_decorated_dead_process(self):
         """Process Fork ProcessExpired is raised if process dies."""
         future = critical_decorated()
-        with self.assertRaises(ProcessExpired):
+        with self.assertRaises(ProcessExpired) as exc_ctx:
             future.result()
+        self.assertEqual(exc_ctx.exception.exitcode, 123)
+        self.assertIsInstance(exc_ctx.exception.pid, int)
 
     def test_timeout_decorated_callback(self):
         """Process Fork ProcessExpired is forwarded to callback."""
diff --git a/test/test_concurrent_process_forkserver.py b/test/test_concurrent_process_forkserver.py
index bb3f956..f78b4d9 100644
--- a/test/test_concurrent_process_forkserver.py
+++ b/test/test_concurrent_process_forkserver.py
@@ -255,8 +255,10 @@ class TestProcessConcurrent(unittest.TestCase):
     def test_decorated_dead_process(self):
         """Process Forkserver ProcessExpired is raised if process dies."""
         future = critical_decorated()
-        with self.assertRaises(ProcessExpired):
+        with self.assertRaises(ProcessExpired) as exc_ctx:
             future.result()
+        self.assertEqual(exc_ctx.exception.exitcode, 123)
+        self.assertIsInstance(exc_ctx.exception.pid, int)
 
     def test_timeout_decorated_callback(self):
         """Process Forkserver ProcessExpired is forwarded to callback."""
diff --git a/test/test_concurrent_process_spawn.py b/test/test_concurrent_process_spawn.py
index 56bb185..a25b6a8 100644
--- a/test/test_concurrent_process_spawn.py
+++ b/test/test_concurrent_process_spawn.py
@@ -255,8 +255,10 @@ class TestProcessConcurrent(unittest.TestCase):
     def test_decorated_dead_process(self):
         """Process Spawn ProcessExpired is raised if process dies."""
         future = critical_decorated()
-        with self.assertRaises(ProcessExpired):
+        with self.assertRaises(ProcessExpired) as exc_ctx:
             future.result()
+        self.assertEqual(exc_ctx.exception.exitcode, 123)
+        self.assertIsInstance(exc_ctx.exception.pid, int)
 
     def test_timeout_decorated_callback(self):
         """Process Spawn ProcessExpired is forwarded to callback."""
diff --git a/test/test_process_pool_fork.py b/test/test_process_pool_fork.py
index 83ccb35..a2d8b09 100644
--- a/test/test_process_pool_fork.py
+++ b/test/test_process_pool_fork.py
@@ -427,7 +427,11 @@ class TestProcessPool(unittest.TestCase):
         """Process Pool Fork unexpect death of worker raises ProcessExpired."""
         with ProcessPool(max_workers=1, context=mp_context) as pool:
             future = pool.schedule(suicide_function)
-            self.assertRaises(ProcessExpired, future.result)
+            worker_pid = list(pool._pool_manager.worker_manager.workers)[0]
+            with self.assertRaises(ProcessExpired) as exc_ctx:
+                future.result()
+            self.assertEqual(exc_ctx.exception.exitcode, 1)
+            self.assertEqual(exc_ctx.exception.pid, worker_pid)
 
     def test_process_pool_map(self):
         """Process Pool Fork map simple."""
diff --git a/test/test_process_pool_forkserver.py b/test/test_process_pool_forkserver.py
index 336d063..6ec6258 100644
--- a/test/test_process_pool_forkserver.py
+++ b/test/test_process_pool_forkserver.py
@@ -429,7 +429,11 @@ class TestProcessPool(unittest.TestCase):
         """Process Pool Forkserver unexpect death of worker raises ProcessExpired."""
         with ProcessPool(max_workers=1, context=mp_context) as pool:
             future = pool.schedule(suicide_function)
-            self.assertRaises(ProcessExpired, future.result)
+            worker_pid = list(pool._pool_manager.worker_manager.workers)[0]
+            with self.assertRaises(ProcessExpired) as exc_ctx:
+                future.result()
+            self.assertEqual(exc_ctx.exception.exitcode, 1)
+            self.assertEqual(exc_ctx.exception.pid, worker_pid)
 
     def test_process_pool_map(self):
         """Process Pool Forkserver map simple."""
diff --git a/test/test_process_pool_spawn.py b/test/test_process_pool_spawn.py
index 2d7914b..8b40c2b 100644
--- a/test/test_process_pool_spawn.py
+++ b/test/test_process_pool_spawn.py
@@ -427,7 +427,11 @@ class TestProcessPool(unittest.TestCase):
         """Process Pool Spawn unexpect death of worker raises ProcessExpired."""
         with ProcessPool(max_workers=1, context=mp_context) as pool:
             future = pool.schedule(suicide_function)
-            self.assertRaises(ProcessExpired, future.result)
+            worker_pid = list(pool._pool_manager.worker_manager.workers)[0]
+            with self.assertRaises(ProcessExpired) as exc_ctx:
+                future.result()
+            self.assertEqual(exc_ctx.exception.exitcode, 1)
+            self.assertEqual(exc_ctx.exception.pid, worker_pid)
 
     def test_process_pool_map(self):
         """Process Pool Spawn map simple."""