Support admin dashboard in multi-process deployments

panel/command/serve.py

@@ -20,7 +20,7 @@
 from bokeh.command.subcommands.serve import Serve as _BkServe
 from bokeh.command.util import build_single_handler_applications
 from bokeh.server.contexts import ApplicationContext
-from tornado.ioloop import IOLoop, PeriodicCallback
+from tornado.ioloop import PeriodicCallback
 from tornado.web import StaticFileHandler
 
 from ..auth import OAuthProvider
@@ -72,10 +72,9 @@ def parse_vars(items):
 class AdminApplicationContext(ApplicationContext):
 
     def __init__(self, application, unused_timeout=15000, **kwargs):
-        super().__init__(application, io_loop=IOLoop.current(), **kwargs)
+        super().__init__(application, **kwargs)
         self._unused_timeout = unused_timeout
         self._cleanup_cb = None
-        self._loop.add_callback(self.run_load_hook)
 
     async def cleanup_sessions(self):
         await self._cleanup_sessions(self._unused_timeout)
@@ -308,7 +307,9 @@ def customize_kwargs(self, args, server_kwargs):
             config._admin = True
             app = Application(FunctionHandler(admin_panel))
             unused_timeout = args.check_unused_sessions or 15000
-            app_ctx = AdminApplicationContext(app, unused_timeout=unused_timeout, url='/admin')
+            state._admin_context = app_ctx = AdminApplicationContext(
+                app, unused_timeout=unused_timeout, url='/admin'
+            )
             if all(not isinstance(handler, DocumentLifecycleHandler) for handler in app._handlers):
                 app.add(DocumentLifecycleHandler())
             app_patterns = []

panel/io/admin.py

@@ -34,11 +34,7 @@
 from .server import set_curdoc
 from .state import state
 
-try:
-    import psutil
-    process = psutil.Process(os.getpid())
-except Exception:
-    process = None
+PROCESSES = {}
 
 log_sessions = []
 
@@ -324,11 +320,19 @@ def get_version_info():
     Param: {param.__version__}</br>
     </code>""", width=300, height=300, margin=(0, 5))
 
+def get_process():
+    import psutil
+    if os.getpid() in PROCESSES:
+        process = PROCESSES[os.getpid()]
+    else:
+        PROCESSES[os.getpid()] = process = psutil.Process(os.getpid())
+    return process
+
 def get_mem():
-    return pd.DataFrame([(time.time(), process.memory_info().rss/1024/1024)], columns=['time', 'memory'])
+    return pd.DataFrame([(time.time(), get_process().memory_info().rss/1024/1024)], columns=['time', 'memory'])
 
 def get_cpu():
-    return pd.DataFrame([(time.time(), process.cpu_percent())], columns=['time', 'cpu'])
+    return pd.DataFrame([(time.time(), get_process().cpu_percent())], columns=['time', 'cpu'])
 
 def get_process_info():
     memory = Trend(
@@ -339,14 +343,12 @@ def get_process_info():
         data=get_cpu(), plot_x='time', plot_y='cpu', plot_type='step',
         title='CPU Usage (%)', width=300, height=300
     )
-    def update_memory(): memory.stream(get_mem())
-    def update_cpu(): cpu.stream(get_cpu())
-    mem_cb = state.add_periodic_callback(update_memory, period=1000, start=False)
-    cpu_cb = state.add_periodic_callback(update_cpu, period=1000, start=False)
-    mem_cb.log = False
-    cpu_cb.log = False
-    mem_cb.start()
-    cpu_cb.start()
+    def update_stats():
+        memory.stream(get_mem())
+        cpu.stream(get_cpu())
+    stats_cb = state.add_periodic_callback(update_stats, period=1000, start=False)
+    stats_cb.log = False
+    stats_cb.start()
     return memory, cpu
 
 def get_session_data():
@@ -412,11 +414,14 @@ def _unwatch_session_info(session_context):
 def get_overview(doc=None):
     layout = FlexBox(*get_session_info(doc), margin=0, sizing_mode='stretch_width')
     info = get_version_info()
-    if process is None:
+    try:
+        import psutil  # noqa
+    except Exception:
         layout.append(info)
         return layout
-    layout.extend([*get_process_info(), info])
-    return layout
+    else:
+        layout.extend([*get_process_info(), info])
+        return layout
 
 
 def log_component():
@@ -476,7 +481,7 @@ def _remove_log_session(session_context):
     return template
 
 def admin_panel(doc):
-    template = admin_template(doc)
     with set_curdoc(doc):
+        template = admin_template(doc)
         template.server_doc(doc)
     return doc

panel/io/server.py

@@ -239,7 +239,7 @@ def destroy_document(self, session):
     state.schedule_task('gc.collect', gc.collect, at=at)
 
 
-# Patch Srrver to attach task factory to asyncio loop
+# Patch Server to attach task factory to asyncio loop and handle Admin server context
 class Server(BokehServer):
 
     def __init__(self, *args, **kwargs):
@@ -248,9 +248,22 @@ def __init__(self, *args, **kwargs):
             _add_task_factory(self.io_loop.asyncio_loop) # type: ignore
         except Exception:
             pass
+        if state._admin_context:
+            state._admin_context._loop = self._loop
+
+    def start(self) -> None:
+        super().start()
+        if state._admin_context:
+            self._loop.add_callback(state._admin_context.run_load_hook)
+
+    def stop(self, wait: bool = True) -> None:
+        super().stop(wait=wait)
+        if state._admin_context:
+            state._admin_context.run_unload_hook()
 
 bokeh.server.server.Server = Server
 
+
 # Patch Application to handle session callbacks
 class Application(BkApplication):
 

panel/io/state.py

@@ -122,6 +122,9 @@ class _state(param.Parameterized):
     _thread_id_: ClassVar[WeakKeyDictionary[Document, int]] = WeakKeyDictionary()
     _thread_pool = None
 
+    # Admin application (remove in Panel 1.0 / Bokeh 3.0)
+    _admin_context = None
+
     # Jupyter communication
     _comm_manager = _CommManager
     _kernels = {}