Ensure tasks are cancelled correctly (#7641)

panel/io/state.py

@@ -599,12 +599,13 @@ def cancel_task(self, name: str, wait: bool=False):
         wait: boolean
             Whether to wait until after the next execution.
         """
-        if name not in self._scheduled:
+        key = f"{os.getpid()}_{name}"
+        if key not in self._scheduled:
             raise KeyError(f'No task with the name {name!r} has been scheduled.')
         if wait:
-            self._scheduled[name] = (None, self._scheduled[name][1])
+            self._scheduled[key] = (None, self._scheduled[key][1])
         else:
-            del self._scheduled[name]
+            del self._scheduled[key]
 
     def clear_caches(self):
         """
@@ -878,9 +879,9 @@ def schedule_task(
           Whether the callback should be run on a thread (requires
           config.nthreads to be set).
         """
-        name = f"{os.getpid()}_{name}"
-        if name in self._scheduled:
-            if callback is not self._scheduled[name][1]:
+        key = f"{os.getpid()}_{name}"
+        if key in self._scheduled:
+            if callback is not self._scheduled[key][1]:
                 self.param.warning(
                     "A separate task was already scheduled under the "
                     f"name {name!r}. The new task will be ignored. If "
@@ -933,9 +934,9 @@ def dgen():
             call_time_seconds = (next(diter) - now)
         except StopIteration:
             return
-        self._scheduled[name] = (diter, callback)
+        self._scheduled[key] = (diter, callback)
         self._ioloop.call_later(
-            delay=call_time_seconds, callback=partial(self._scheduled_cb, name, threaded)
+            delay=call_time_seconds, callback=partial(self._scheduled_cb, key, threaded)
         )
 
     def sync_busy(self, indicator: BooleanIndicator) -> None:

panel/tests/test_server.py

@@ -323,6 +323,24 @@ def loaded():
     wait_until(lambda: len(counts) >= 5 and counts == list(range(len(counts))))
 
 
+def test_server_cancel_task(server_implementation):
+    state.cache['count'] = 0
+    def periodic_cb():
+        state.cache['count'] += 1
+
+    def app():
+        state.schedule_task('periodic', periodic_cb, period='0.1s')
+        return '# state.schedule test'
+
+    serve_and_request(app)
+
+    wait_until(lambda: state.cache['count'] > 0)
+    state.cancel_task('periodic')
+    count = state.cache['count']
+    time.sleep(0.5)
+    assert state.cache['count'] == count
+
+
 def test_server_schedule_repeat(server_implementation):
     state.cache['count'] = 0
     def periodic_cb():