Improve error when submitting work from a closed client

distributed/client.py

@@ -734,6 +734,10 @@
     """Custom exception class to exit All(...) early."""
 
 
+class ClosedClientError(Exception):
+    """Raised when an action with a closed client can't be performed"""
+
+
 def _handle_print(event):
     _, msg = event
     if not isinstance(msg, dict):
@@ -1419,9 +1423,8 @@
         if self.status in ("running", "closing", "connecting", "newly-created"):
             self.loop.add_callback(self._send_to_scheduler_safe, msg)
         else:
-            raise Exception(
-                "Tried sending message after closing.  Status: %s\n"
-                "Message: %s" % (self.status, msg)
+            raise ClosedClientError(
+                f"Client is {self.status}. Can't send {msg['op']} message."
             )
 
     async def _start(self, timeout=no_default, **kwargs):

distributed/tests/test_client.py

@@ -63,6 +63,7 @@
 )
 from distributed.client import (
     Client,
+    ClosedClientError,
     Future,
     FutureCancelledError,
     FuturesCancelledError,
@@ -5711,6 +5712,19 @@ async def test_warn_when_submitting_large_values_memoryview(c, s):
         c.submit(lambda: a)
 
 
+@gen_cluster(client=True, nthreads=[])
+async def test_closed_client_send_message(c, s):
+    # Ensure a meaningful, but concise error is raised when
+    # a closed client attempts to send a message to the scheduler
+    await c.close()
+    with pytest.raises(ClosedClientError, match="update-graph") as exc_info:
+        c.submit(lambda x: x + 1)
+
+    msg = str(exc_info.value)
+    assert "Can't send update-graph" in msg
+    assert len(msg) < 100
+
+
 @gen_cluster(client=True)
 async def test_unhashable_function(c, s, a, b):
     func = _UnhashableCallable()