We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Race Condition with threaded Periodic Callback can result in additional executions
#### ALL software version info

panel==1.2.1

#### Description of expected behavior and the observed behavior

Looks like due to a race condition, by scheduling a "heavy" periodic callback, it can be executed more than the maximum count when letting panel handle the multithreading. This is because, I believe, the counter is incremented after the callback is finished executing, and with multiple threads many instances can be scheduled while waiting for the count to be met.

In the below example, when you toggle on the periodic with a 0 delay it runs as expected, but increasing the delay beyond the period will result in additional executions of the callback. The larger the delay the worse the effect.

#### Complete, minimal, self-contained example code that reproduces the issue

```python
# app.py
import time
import panel as pn
pn.extension(nthreads=0)  # or > 1

def heavy():
    print(cb.counter)
    time.sleep(delay.value)  # simulate lots of work

delay = pn.widgets.IntInput(name='Delay', value=0, start=0, end=5, step=1)

cb = pn.state.add_periodic_callback(heavy, 200, count=5, start=False)
toggle = pn.widgets.Toggle(name='Run', value=False)
toggle.link(cb, bidirectional=True, value='running')

pn.Column(delay, toggle).servable()
```

Run with:
```shell
panel serve app.py
```

#### Stack traceback and/or browser JavaScript console output

N/A

#### Screenshots or screencasts of the bug in action

N/A

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `panel/io/callbacks.py`
Content:
```
1 """
2 Defines callbacks to be executed on a thread or by scheduling it
3 on a running bokeh server.
4 """
5 import asyncio
6 import inspect
7 import logging
8 import time
9 
10 from functools import partial
11 
12 import param
13 
14 from ..util import edit_readonly, function_name
15 from .logging import LOG_PERIODIC_END, LOG_PERIODIC_START
16 from .state import curdoc_locked, state
17 
18 log = logging.getLogger('panel.callbacks')
19 _periodic_logger = logging.getLogger(f'{__name__}.PeriodicCallback')
20 
21 class PeriodicCallback(param.Parameterized):
22     """
23     Periodic encapsulates a periodic callback which will run both
24     in tornado based notebook environments and on bokeh server. By
25     default the callback will run until the stop method is called,
26     but count and timeout values can be set to limit the number of
27     executions or the maximum length of time for which the callback
28     will run. The callback may also be started and stopped by setting
29     the running parameter to True or False respectively.
30     """
31 
32     callback = param.Callable(doc="""
33         The callback to execute periodically.""")
34 
35     counter = param.Integer(default=0, doc="""
36         Counts the number of executions.""")
37 
38     count = param.Integer(default=None, doc="""
39         Number of times the callback will be executed, by default
40         this is unlimited.""")
41 
42     log = param.Boolean(default=True, doc="""
43         Whether the periodic callback should log its actions.""")
44 
45     period = param.Integer(default=500, doc="""
46         Period in milliseconds at which the callback is executed.""")
47 
48     timeout = param.Integer(default=None, doc="""
49         Timeout in milliseconds from the start time at which the callback
50         expires.""")
51 
52     running = param.Boolean(default=False, doc="""
53         Toggles whether the periodic callback is currently running.""")
54 
55     def __init__(self, **params):
56         self._background = params.pop('background', False)
57         super().__init__(**params)
58         self._start_time = None
59         self._cb = None
60         self._updating = False
61         self._doc = None
62 
63     @param.depends('running', watch=True)
64     def _start(self):
65         if not self.running or self._updating:
66             return
67         self.start()
68 
69     @param.depends('running', watch=True)
70     def _stop(self):
71         if self.running or self._updating:
72             return
73         self.stop()
74 
75     @param.depends('period', watch=True)
76     def _update_period(self):
77         if self._cb:
78             self.stop()
79             self.start()
80 
81     def _exec_callback(self, post=False):
82         from .state import set_curdoc
83         try:
84             with set_curdoc(self._doc):
85                 cb = self.callback()
86         except Exception:
87             cb = None
88         if post:
89             self._post_callback()
90         return cb
91 
92     def _post_callback(self):
93         cbname = function_name(self.callback)
94         if self._doc and self.log:
95             _periodic_logger.info(
96                 LOG_PERIODIC_END, id(self._doc), cbname, self.counter
97             )
98         if not self._background:
99             with edit_readonly(state):
100                 state._busy_counter -= 1
101         self.counter += 1
102         if self.timeout is not None:
103             dt = (time.time() - self._start_time) * 1000
104             if dt > self.timeout:
105                 self.stop()
106         if self.counter == self.count:
107             self.stop()
108 
109     async def _periodic_callback(self):
110         if not self._background:
111             with edit_readonly(state):
112                 state._busy_counter += 1
113         cbname = function_name(self.callback)
114         if self._doc and self.log:
115             _periodic_logger.info(
116                 LOG_PERIODIC_START, id(self._doc), cbname, self.counter
117             )
118         is_async = (
119             inspect.isasyncgenfunction(self.callback) or
120             inspect.iscoroutinefunction(self.callback)
121         )
122         if state._thread_pool and not is_async:
123             future = state._thread_pool.submit(self._exec_callback, True)
124             future.add_done_callback(partial(state._handle_future_exception, doc=self._doc))
125             return
126         try:
127             cb = self._exec_callback()
128             if inspect.isawaitable(cb):
129                 await cb
130         except Exception:
131             log.exception('Periodic callback failed.')
132             raise
133         finally:
134             self._post_callback()
135 
136     async def _async_repeat(self, func):
137         """
138         Run func every interval seconds.
139 
140         If func has not finished before *interval*, will run again
141         immediately when the previous iteration finished.
142         """
143         while True:
144             start = time.monotonic()
145             await func()
146             timeout = (self.period/1000.) - (time.monotonic()-start)
147             if timeout > 0:
148                 await asyncio.sleep(timeout)
149 
150     def _cleanup(self, session_context):
151         self.stop()
152 
153     def start(self):
154         """
155         Starts running the periodic callback.
156         """
157         if self._cb is not None:
158             raise RuntimeError('Periodic callback has already started.')
159         if not self.running:
160             try:
161                 self._updating = True
162                 self.running = True
163             finally:
164                 self._updating = False
165         self._start_time = time.time()
166         if state._is_pyodide:
167             self._cb = asyncio.create_task(
168                 self._async_repeat(self._periodic_callback)
169             )
170         elif state.curdoc and state.curdoc.session_context:
171             self._doc = state.curdoc
172             if state._unblocked(state.curdoc):
173                 self._cb = self._doc.add_periodic_callback(self._periodic_callback, self.period)
174             else:
175                 self._doc.add_next_tick_callback(self.start)
176         else:
177             from tornado.ioloop import PeriodicCallback
178             self._cb = PeriodicCallback(lambda: asyncio.create_task(self._periodic_callback()), self.period)
179             self._cb.start()
180 
181     def stop(self):
182         """
183         Stops running the periodic callback.
184         """
185         if self.running:
186             try:
187                 self._updating = True
188                 self.running = False
189             finally:
190                 self._updating = False
191         with param.discard_events(self):
192             self.counter = 0
193         self._timeout = None
194         if state._is_pyodide:
195             self._cb.cancel()
196         elif self._doc:
197             if self._doc._session_context:
198                 self._doc.callbacks.remove_session_callback(self._cb)
199             else:
200                 self._doc.callbacks._session_callbacks.remove(self._cb)
201         elif self._cb:
202             self._cb.stop()
203         self._cb = None
204         doc = self._doc or curdoc_locked()
205         if doc:
206             doc.callbacks.session_destroyed_callbacks = {
207                 cb for cb in doc.callbacks.session_destroyed_callbacks
208                 if cb is not self._cleanup
209             }
210             self._doc = None
211
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/panel/io/callbacks.py b/panel/io/callbacks.py
--- a/panel/io/callbacks.py
+++ b/panel/io/callbacks.py
@@ -82,7 +82,11 @@
         from .state import set_curdoc
         try:
             with set_curdoc(self._doc):
-                cb = self.callback()
+                if self.running:
+                    self.counter += 1
+                    if self.counter > self.count:
+                        self.stop()
+                cb = self.callback() if self.running else None
         except Exception:
             cb = None
         if post:
@@ -98,7 +102,6 @@
         if not self._background:
             with edit_readonly(state):
                 state._busy_counter -= 1
-        self.counter += 1
         if self.timeout is not None:
             dt = (time.time() - self._start_time) * 1000
             if dt > self.timeout:

{"golden_diff": "diff --git a/panel/io/callbacks.py b/panel/io/callbacks.py\n--- a/panel/io/callbacks.py\n+++ b/panel/io/callbacks.py\n@@ -82,7 +82,11 @@\n         from .state import set_curdoc\n         try:\n             with set_curdoc(self._doc):\n-                cb = self.callback()\n+                if self.running:\n+                    self.counter += 1\n+                    if self.counter > self.count:\n+                        self.stop()\n+                cb = self.callback() if self.running else None\n         except Exception:\n             cb = None\n         if post:\n@@ -98,7 +102,6 @@\n         if not self._background:\n             with edit_readonly(state):\n                 state._busy_counter -= 1\n-        self.counter += 1\n         if self.timeout is not None:\n             dt = (time.time() - self._start_time) * 1000\n             if dt > self.timeout:\n", "issue": "Race Condition with threaded Periodic Callback can result in additional executions\n#### ALL software version info\r\n\r\npanel==1.2.1\r\n\r\n#### Description of expected behavior and the observed behavior\r\n\r\nLooks like due to a race condition, by scheduling a \"heavy\" periodic callback, it can be executed more than the maximum count when letting panel handle the multithreading. This is because, I believe, the counter is incremented after the callback is finished executing, and with multiple threads many instances can be scheduled while waiting for the count to be met.\r\n\r\nIn the below example, when you toggle on the periodic with a 0 delay it runs as expected, but increasing the delay beyond the period will result in additional executions of the callback. The larger the delay the worse the effect.\r\n\r\n#### Complete, minimal, self-contained example code that reproduces the issue\r\n\r\n```python\r\n# app.py\r\nimport time\r\nimport panel as pn\r\npn.extension(nthreads=0)  # or > 1\r\n\r\ndef heavy():\r\n    print(cb.counter)\r\n    time.sleep(delay.value)  # simulate lots of work\r\n\r\ndelay = pn.widgets.IntInput(name='Delay', value=0, start=0, end=5, step=1)\r\n\r\ncb = pn.state.add_periodic_callback(heavy, 200, count=5, start=False)\r\ntoggle = pn.widgets.Toggle(name='Run', value=False)\r\ntoggle.link(cb, bidirectional=True, value='running')\r\n\r\npn.Column(delay, toggle).servable()\r\n```\r\n\r\nRun with:\r\n```shell\r\npanel serve app.py\r\n```\r\n\r\n#### Stack traceback and/or browser JavaScript console output\r\n\r\nN/A\r\n\r\n#### Screenshots or screencasts of the bug in action\r\n\r\nN/A\n", "before_files": [{"content": "\"\"\"\nDefines callbacks to be executed on a thread or by scheduling it\non a running bokeh server.\n\"\"\"\nimport asyncio\nimport inspect\nimport logging\nimport time\n\nfrom functools import partial\n\nimport param\n\nfrom ..util import edit_readonly, function_name\nfrom .logging import LOG_PERIODIC_END, LOG_PERIODIC_START\nfrom .state import curdoc_locked, state\n\nlog = logging.getLogger('panel.callbacks')\n_periodic_logger = logging.getLogger(f'{__name__}.PeriodicCallback')\n\nclass PeriodicCallback(param.Parameterized):\n    \"\"\"\n    Periodic encapsulates a periodic callback which will run both\n    in tornado based notebook environments and on bokeh server. By\n    default the callback will run until the stop method is called,\n    but count and timeout values can be set to limit the number of\n    executions or the maximum length of time for which the callback\n    will run. The callback may also be started and stopped by setting\n    the running parameter to True or False respectively.\n    \"\"\"\n\n    callback = param.Callable(doc=\"\"\"\n        The callback to execute periodically.\"\"\")\n\n    counter = param.Integer(default=0, doc=\"\"\"\n        Counts the number of executions.\"\"\")\n\n    count = param.Integer(default=None, doc=\"\"\"\n        Number of times the callback will be executed, by default\n        this is unlimited.\"\"\")\n\n    log = param.Boolean(default=True, doc=\"\"\"\n        Whether the periodic callback should log its actions.\"\"\")\n\n    period = param.Integer(default=500, doc=\"\"\"\n        Period in milliseconds at which the callback is executed.\"\"\")\n\n    timeout = param.Integer(default=None, doc=\"\"\"\n        Timeout in milliseconds from the start time at which the callback\n        expires.\"\"\")\n\n    running = param.Boolean(default=False, doc=\"\"\"\n        Toggles whether the periodic callback is currently running.\"\"\")\n\n    def __init__(self, **params):\n        self._background = params.pop('background', False)\n        super().__init__(**params)\n        self._start_time = None\n        self._cb = None\n        self._updating = False\n        self._doc = None\n\n    @param.depends('running', watch=True)\n    def _start(self):\n        if not self.running or self._updating:\n            return\n        self.start()\n\n    @param.depends('running', watch=True)\n    def _stop(self):\n        if self.running or self._updating:\n            return\n        self.stop()\n\n    @param.depends('period', watch=True)\n    def _update_period(self):\n        if self._cb:\n            self.stop()\n            self.start()\n\n    def _exec_callback(self, post=False):\n        from .state import set_curdoc\n        try:\n            with set_curdoc(self._doc):\n                cb = self.callback()\n        except Exception:\n            cb = None\n        if post:\n            self._post_callback()\n        return cb\n\n    def _post_callback(self):\n        cbname = function_name(self.callback)\n        if self._doc and self.log:\n            _periodic_logger.info(\n                LOG_PERIODIC_END, id(self._doc), cbname, self.counter\n            )\n        if not self._background:\n            with edit_readonly(state):\n                state._busy_counter -= 1\n        self.counter += 1\n        if self.timeout is not None:\n            dt = (time.time() - self._start_time) * 1000\n            if dt > self.timeout:\n                self.stop()\n        if self.counter == self.count:\n            self.stop()\n\n    async def _periodic_callback(self):\n        if not self._background:\n            with edit_readonly(state):\n                state._busy_counter += 1\n        cbname = function_name(self.callback)\n        if self._doc and self.log:\n            _periodic_logger.info(\n                LOG_PERIODIC_START, id(self._doc), cbname, self.counter\n            )\n        is_async = (\n            inspect.isasyncgenfunction(self.callback) or\n            inspect.iscoroutinefunction(self.callback)\n        )\n        if state._thread_pool and not is_async:\n            future = state._thread_pool.submit(self._exec_callback, True)\n            future.add_done_callback(partial(state._handle_future_exception, doc=self._doc))\n            return\n        try:\n            cb = self._exec_callback()\n            if inspect.isawaitable(cb):\n                await cb\n        except Exception:\n            log.exception('Periodic callback failed.')\n            raise\n        finally:\n            self._post_callback()\n\n    async def _async_repeat(self, func):\n        \"\"\"\n        Run func every interval seconds.\n\n        If func has not finished before *interval*, will run again\n        immediately when the previous iteration finished.\n        \"\"\"\n        while True:\n            start = time.monotonic()\n            await func()\n            timeout = (self.period/1000.) - (time.monotonic()-start)\n            if timeout > 0:\n                await asyncio.sleep(timeout)\n\n    def _cleanup(self, session_context):\n        self.stop()\n\n    def start(self):\n        \"\"\"\n        Starts running the periodic callback.\n        \"\"\"\n        if self._cb is not None:\n            raise RuntimeError('Periodic callback has already started.')\n        if not self.running:\n            try:\n                self._updating = True\n                self.running = True\n            finally:\n                self._updating = False\n        self._start_time = time.time()\n        if state._is_pyodide:\n            self._cb = asyncio.create_task(\n                self._async_repeat(self._periodic_callback)\n            )\n        elif state.curdoc and state.curdoc.session_context:\n            self._doc = state.curdoc\n            if state._unblocked(state.curdoc):\n                self._cb = self._doc.add_periodic_callback(self._periodic_callback, self.period)\n            else:\n                self._doc.add_next_tick_callback(self.start)\n        else:\n            from tornado.ioloop import PeriodicCallback\n            self._cb = PeriodicCallback(lambda: asyncio.create_task(self._periodic_callback()), self.period)\n            self._cb.start()\n\n    def stop(self):\n        \"\"\"\n        Stops running the periodic callback.\n        \"\"\"\n        if self.running:\n            try:\n                self._updating = True\n                self.running = False\n            finally:\n                self._updating = False\n        with param.discard_events(self):\n            self.counter = 0\n        self._timeout = None\n        if state._is_pyodide:\n            self._cb.cancel()\n        elif self._doc:\n            if self._doc._session_context:\n                self._doc.callbacks.remove_session_callback(self._cb)\n            else:\n                self._doc.callbacks._session_callbacks.remove(self._cb)\n        elif self._cb:\n            self._cb.stop()\n        self._cb = None\n        doc = self._doc or curdoc_locked()\n        if doc:\n            doc.callbacks.session_destroyed_callbacks = {\n                cb for cb in doc.callbacks.session_destroyed_callbacks\n                if cb is not self._cleanup\n            }\n            self._doc = None\n", "path": "panel/io/callbacks.py"}], "after_files": [{"content": "\"\"\"\nDefines callbacks to be executed on a thread or by scheduling it\non a running bokeh server.\n\"\"\"\nimport asyncio\nimport inspect\nimport logging\nimport time\n\nfrom functools import partial\n\nimport param\n\nfrom ..util import edit_readonly, function_name\nfrom .logging import LOG_PERIODIC_END, LOG_PERIODIC_START\nfrom .state import curdoc_locked, state\n\nlog = logging.getLogger('panel.callbacks')\n_periodic_logger = logging.getLogger(f'{__name__}.PeriodicCallback')\n\nclass PeriodicCallback(param.Parameterized):\n    \"\"\"\n    Periodic encapsulates a periodic callback which will run both\n    in tornado based notebook environments and on bokeh server. By\n    default the callback will run until the stop method is called,\n    but count and timeout values can be set to limit the number of\n    executions or the maximum length of time for which the callback\n    will run. The callback may also be started and stopped by setting\n    the running parameter to True or False respectively.\n    \"\"\"\n\n    callback = param.Callable(doc=\"\"\"\n        The callback to execute periodically.\"\"\")\n\n    counter = param.Integer(default=0, doc=\"\"\"\n        Counts the number of executions.\"\"\")\n\n    count = param.Integer(default=None, doc=\"\"\"\n        Number of times the callback will be executed, by default\n        this is unlimited.\"\"\")\n\n    log = param.Boolean(default=True, doc=\"\"\"\n        Whether the periodic callback should log its actions.\"\"\")\n\n    period = param.Integer(default=500, doc=\"\"\"\n        Period in milliseconds at which the callback is executed.\"\"\")\n\n    timeout = param.Integer(default=None, doc=\"\"\"\n        Timeout in milliseconds from the start time at which the callback\n        expires.\"\"\")\n\n    running = param.Boolean(default=False, doc=\"\"\"\n        Toggles whether the periodic callback is currently running.\"\"\")\n\n    def __init__(self, **params):\n        self._background = params.pop('background', False)\n        super().__init__(**params)\n        self._start_time = None\n        self._cb = None\n        self._updating = False\n        self._doc = None\n\n    @param.depends('running', watch=True)\n    def _start(self):\n        if not self.running or self._updating:\n            return\n        self.start()\n\n    @param.depends('running', watch=True)\n    def _stop(self):\n        if self.running or self._updating:\n            return\n        self.stop()\n\n    @param.depends('period', watch=True)\n    def _update_period(self):\n        if self._cb:\n            self.stop()\n            self.start()\n\n    def _exec_callback(self, post=False):\n        from .state import set_curdoc\n        try:\n            with set_curdoc(self._doc):\n                if self.running:\n                    self.counter += 1\n                    if self.counter > self.count:\n                        self.stop()\n                cb = self.callback() if self.running else None\n        except Exception:\n            cb = None\n        if post:\n            self._post_callback()\n        return cb\n\n    def _post_callback(self):\n        cbname = function_name(self.callback)\n        if self._doc and self.log:\n            _periodic_logger.info(\n                LOG_PERIODIC_END, id(self._doc), cbname, self.counter\n            )\n        if not self._background:\n            with edit_readonly(state):\n                state._busy_counter -= 1\n        if self.timeout is not None:\n            dt = (time.time() - self._start_time) * 1000\n            if dt > self.timeout:\n                self.stop()\n        if self.counter == self.count:\n            self.stop()\n\n    async def _periodic_callback(self):\n        if not self._background:\n            with edit_readonly(state):\n                state._busy_counter += 1\n        cbname = function_name(self.callback)\n        if self._doc and self.log:\n            _periodic_logger.info(\n                LOG_PERIODIC_START, id(self._doc), cbname, self.counter\n            )\n        is_async = (\n            inspect.isasyncgenfunction(self.callback) or\n            inspect.iscoroutinefunction(self.callback)\n        )\n        if state._thread_pool and not is_async:\n            future = state._thread_pool.submit(self._exec_callback, True)\n            future.add_done_callback(partial(state._handle_future_exception, doc=self._doc))\n            return\n        try:\n            cb = self._exec_callback()\n            if inspect.isawaitable(cb):\n                await cb\n        except Exception:\n            log.exception('Periodic callback failed.')\n            raise\n        finally:\n            self._post_callback()\n\n    async def _async_repeat(self, func):\n        \"\"\"\n        Run func every interval seconds.\n\n        If func has not finished before *interval*, will run again\n        immediately when the previous iteration finished.\n        \"\"\"\n        while True:\n            start = time.monotonic()\n            await func()\n            timeout = (self.period/1000.) - (time.monotonic()-start)\n            if timeout > 0:\n                await asyncio.sleep(timeout)\n\n    def _cleanup(self, session_context):\n        self.stop()\n\n    def start(self):\n        \"\"\"\n        Starts running the periodic callback.\n        \"\"\"\n        if self._cb is not None:\n            raise RuntimeError('Periodic callback has already started.')\n        if not self.running:\n            try:\n                self._updating = True\n                self.running = True\n            finally:\n                self._updating = False\n        self._start_time = time.time()\n        if state._is_pyodide:\n            self._cb = asyncio.create_task(\n                self._async_repeat(self._periodic_callback)\n            )\n        elif state.curdoc and state.curdoc.session_context:\n            self._doc = state.curdoc\n            if state._unblocked(state.curdoc):\n                self._cb = self._doc.add_periodic_callback(self._periodic_callback, self.period)\n            else:\n                self._doc.add_next_tick_callback(self.start)\n        else:\n            from tornado.ioloop import PeriodicCallback\n            self._cb = PeriodicCallback(lambda: asyncio.create_task(self._periodic_callback()), self.period)\n            self._cb.start()\n\n    def stop(self):\n        \"\"\"\n        Stops running the periodic callback.\n        \"\"\"\n        if self.running:\n            try:\n                self._updating = True\n                self.running = False\n            finally:\n                self._updating = False\n        with param.discard_events(self):\n            self.counter = 0\n        self._timeout = None\n        if state._is_pyodide:\n            self._cb.cancel()\n        elif self._doc:\n            if self._doc._session_context:\n                self._doc.callbacks.remove_session_callback(self._cb)\n            else:\n                self._doc.callbacks._session_callbacks.remove(self._cb)\n        elif self._cb:\n            self._cb.stop()\n        self._cb = None\n        doc = self._doc or curdoc_locked()\n        if doc:\n            doc.callbacks.session_destroyed_callbacks = {\n                cb for cb in doc.callbacks.session_destroyed_callbacks\n                if cb is not self._cleanup\n            }\n            self._doc = None\n", "path": "panel/io/callbacks.py"}]}