added is_set functions

distributed/tests/test_variable.py

@@ -20,7 +20,9 @@ async def test_variable(c, s, a, b):
 
     future = c.submit(inc, 1)
 
+    assert not await x.is_set()
     await x.set(future)
+    assert await x.is_set()
     future2 = await xx.get()
     assert future.key == future2.key
 
@@ -30,6 +32,7 @@ async def test_variable(c, s, a, b):
     assert s.tasks  # future still present
 
     x.delete()
+    assert not await x.is_set()
 
     start = time()
     while s.tasks:

distributed/variable.py

@@ -34,7 +34,11 @@ def __init__(self, scheduler):
         self.started = asyncio.Condition()
 
         self.scheduler.handlers.update(
-            {"variable_set": self.set, "variable_get": self.get}
+            {
+                "variable_set": self.set,
+                "variable_is_set": self.is_set,
+                "variable_get": self.get,
+            }
         )
 
         self.scheduler.stream_handlers["variable-future-release"] = self.future_release
@@ -58,6 +62,9 @@ async def set(self, name=None, key=None, data=None, client=None):
                 self.started.notify_all()
         self.variables[name] = record
 
+    async def is_set(self, comm=None, name=None, client=None):
+        return name in self.variables
+
     async def release(self, key, name):
         while self.waiting[key, name]:
             async with self.waiting_conditions[name]:
@@ -189,6 +196,15 @@ def set(self, value, **kwargs):
         """
         return self.client.sync(self._set, value, **kwargs)
 
+    async def _is_set(self):
+        return await self.client.scheduler.variable_is_set(
+            name=self.name, client=self.client.id
+        )
+
+    def is_set(self):
+        """Check the variable has been set"""
+        return self.client.sync(self._is_set)
+
     async def _get(self, timeout=None):
         d = await self.client.scheduler.variable_get(
             timeout=timeout, name=self.name, client=self.client.id