Merge pull request #2943 from finos/fix-uvicorn-bug

texodus · web-flow · commit a450820eb8aa · 2025-03-08T23:10:06.000-05:00
Fix segfault in Python webservers
diff --git a/rust/perspective-python/Cargo.toml b/rust/perspective-python/Cargo.toml
@@ -71,7 +71,6 @@ pyo3 = { version = "0.23.4", features = [
     "experimental-async",
     "extension-module",
     "serde",
-    "py-clone",
 ] }
 pythonize = "0.23.0"
 tracing = { version = ">=0.1.36" }
diff --git a/rust/perspective-python/perspective/tests/async/test_websocket_client.py b/rust/perspective-python/perspective/tests/async/test_websocket_client.py
@@ -0,0 +1,140 @@
+#  ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
+#  ┃ ██████ ██████ ██████       █      █      █      █      █ █▄  ▀███ █       ┃
+#  ┃ ▄▄▄▄▄█ █▄▄▄▄▄ ▄▄▄▄▄█  ▀▀▀▀▀█▀▀▀▀▀ █ ▀▀▀▀▀█ ████████▌▐███ ███▄  ▀█ █ ▀▀▀▀▀ ┃
+#  ┃ █▀▀▀▀▀ █▀▀▀▀▀ █▀██▀▀ ▄▄▄▄▄ █ ▄▄▄▄▄█ ▄▄▄▄▄█ ████████▌▐███ █████▄   █ ▄▄▄▄▄ ┃
+#  ┃ █      ██████ █  ▀█▄       █ ██████      █      ███▌▐███ ███████▄ █       ┃
+#  ┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫
+#  ┃ Copyright (c) 2017, the Perspective Authors.                              ┃
+#  ┃ ╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌ ┃
+#  ┃ This file is part of the Perspective library, distributed under the terms ┃
+#  ┃ of the [Apache License 2.0](https://www.apache.org/licenses/LICENSE-2.0). ┃
+#  ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛
+
+import asyncio
+import threading
+import websocket
+import os
+import os.path
+
+import tornado.websocket
+import tornado.web
+import tornado.ioloop
+
+import perspective
+import perspective.handlers.tornado
+
+PORT = 8082
+
+
+def test_big_multi_thing():
+    here = os.path.abspath(os.path.dirname(__file__))
+    file_path = os.path.join(
+        here,
+        "..",
+        "..",
+        "..",
+        "..",
+        "..",
+        "node_modules",
+        "superstore-arrow",
+        "superstore.lz4.arrow",
+    )
+
+    async def init_table(client):
+        global SERVER_DATA
+        global SERVER_TABLE
+        with open(file_path, mode="rb") as file:
+            SERVER_DATA = file.read()
+            SERVER_TABLE = client.table(SERVER_DATA, name="superstore")
+
+        global ws
+        ws = websocket.WebSocketApp(
+            "ws://localhost:{}/websocket".format(PORT),
+            on_open=on_open,
+            on_message=on_message,
+            # on_error=on_error,
+            # on_close=on_close,
+        )
+
+        global ws_thread
+        ws_thread = threading.Thread(target=ws.run_forever)
+        ws_thread.start()
+
+    def server_thread():
+        def make_app(perspective_server):
+            return tornado.web.Application(
+                [
+                    (
+                        r"/websocket",
+                        perspective.handlers.tornado.PerspectiveTornadoHandler,
+                        {"perspective_server": perspective_server},
+                    ),
+                ]
+            )
+
+        perspective_server = perspective.Server()
+        app = make_app(perspective_server)
+        global server
+        server = app.listen(PORT)
+
+        global server_loop
+        server_loop = tornado.ioloop.IOLoop.current()
+        client = perspective_server.new_local_client(
+            loop_callback=server_loop.add_callback
+        )
+        server_loop.call_later(0, init_table, client)
+        server_loop.start()
+
+    server_thread = threading.Thread(target=server_thread)
+    server_thread.start()
+
+    client_loop = asyncio.new_event_loop()
+    client_loop.set_debug(True)
+    client_thread = threading.Thread(target=client_loop.run_forever)
+    client_thread.start()
+
+    async def send_request(msg):
+        global ws
+        ws.send(msg, websocket.ABNF.OPCODE_BINARY)
+
+    def on_message(ws, message):
+        async def poke_client():
+            await client.handle_response(message)
+
+        asyncio.run_coroutine_threadsafe(poke_client(), client_loop)
+
+    # def on_error(ws, error):
+    #     print(f"Error!: {error}")
+
+    # def on_close(ws, close_status_code, close_msg):
+    #     print("Connection closed")
+
+    def on_open(ws):
+        global client
+        client = perspective.AsyncClient(send_request)
+        asyncio.run_coroutine_threadsafe(test(client), client_loop)
+
+    global count
+    count = 0
+
+    def update(x):
+        global count
+        count += 1
+
+    async def test(client):
+        table = await client.open_table("superstore")
+        view = await table.view()
+        await view.on_update(update)
+        SERVER_TABLE.update(SERVER_DATA)
+        assert await table.size() == 19988
+        assert count == 1
+        await server.close_all_connections()
+        client_loop.stop()
+
+    client_thread.join()
+    client_loop.close()
+    ws.close()
+    ws_thread.join()
+    server_loop.add_callback(server_loop.stop)
+    server_thread.join()
+    server_loop.close()
diff --git a/rust/perspective-python/perspective/tests/multi_threaded/test_multi_threaded.py b/rust/perspective-python/perspective/tests/multi_threaded/test_multi_threaded.py
@@ -18,24 +18,69 @@
 
 
 class TestServer(object):
-    def test_concurrent_updates(self):
+    def test_sync_updates_with_loop_callback_are_sync(self):
         def feed(table):
             y = 1000
             while y > 0:
                 y -= 1
-                table.update([{"a": random.randint(0, 10), "index": 1}])
+                table.update([{"a": random.randint(0, 10), "index": y}])
 
         perspective_server = Server()
-        loop = asyncio.get_event_loop()
-        client = perspective_server.new_local_client(
-            loop_callback=loop.call_soon_threadsafe
-        )
+        loop = asyncio.new_event_loop()
+        thread = threading.Thread(target=loop.run_forever)
+        thread.start()
 
+        client = perspective_server.new_local_client()
         table = client.table(
             {"a": "integer", "index": "integer"}, index="index", name="test"
         )
 
-        thread = threading.Thread(target=feed, args=(table,), daemon=True)
-        thread.start()
+        view = table.view()
+        global count
+        count = 0
+
+        def update(x):
+            global count
+            count += 1
+
+        view.on_update(update)
         feed(table)
+        assert table.size() == 1000
+        assert count == 1000
+        loop.call_soon_threadsafe(loop.stop)
+        thread.join()
+        loop.close()
+
+    def test_concurrent_updates(self):
+        async def feed(table, loop):
+            y = 1000
+            while y > 0:
+                y -= 1
+                table.update([{"a": random.randint(0, 10), "index": y}])
+                await asyncio.sleep(0.001)
+
+        perspective_server = Server()
+        loop = asyncio.new_event_loop()
+        thread = threading.Thread(target=loop.run_forever)
+        thread.start()
+
+        client = perspective_server.new_local_client()
+        table = client.table(
+            {"a": "integer", "index": "integer"}, index="index", name="test"
+        )
+
+        view = table.view()
+        global count
+        count = 0
+
+        def update(x):
+            global count
+            count += 1
+
+        view.on_update(update)
+        asyncio.run_coroutine_threadsafe(feed(table, loop), loop).result()
+        assert table.size() == 1000
+        assert count == 1000
+        loop.call_soon_threadsafe(loop.stop)
         thread.join()
+        loop.close()
diff --git a/rust/perspective-python/src/client/client_async.rs b/rust/perspective-python/src/client/client_async.rs
@@ -17,9 +17,9 @@ use std::sync::Arc;
 use async_lock::RwLock;
 use futures::FutureExt;
 use perspective_client::{
-    assert_table_api, assert_view_api, Client, OnUpdateMode, OnUpdateOptions, Table, TableData,
-    TableInitOptions, TableReadFormat, UpdateData, UpdateOptions, View, ViewOnUpdateResp,
-    ViewWindow,
+    Client, OnUpdateMode, OnUpdateOptions, Table, TableData, TableInitOptions, TableReadFormat,
+    UpdateData, UpdateOptions, View, ViewOnUpdateResp, ViewWindow, assert_table_api,
+    assert_view_api,
 };
 use pyo3::exceptions::PyValueError;
 use pyo3::prelude::*;
@@ -37,16 +37,16 @@ use crate::py_err::{PyPerspectiveError, ResultTClientErrorExt};
 #[derive(Clone)]
 pub struct AsyncClient {
     pub(crate) client: Client,
-    loop_cb: Arc<RwLock<Option<Py<PyAny>>>>,
-    close_cb: Option<Py<PyAny>>,
+    loop_cb: Arc<RwLock<Arc<Option<Py<PyAny>>>>>,
+    close_cb: Arc<Option<Py<PyAny>>>,
 }
 
 impl AsyncClient {
     pub fn new_from_client(client: Client) -> Self {
         AsyncClient {
             client,
             loop_cb: Arc::default(),
-            close_cb: None,
+            close_cb: Arc::default(),
         }
     }
 }
@@ -80,7 +80,7 @@ impl AsyncClient {
         AsyncClient {
             client,
             loop_cb: Arc::default(),
-            close_cb: handle_close,
+            close_cb: handle_close.into(),
         }
     }
 
@@ -160,12 +160,12 @@ impl AsyncClient {
     }
 
     pub async fn set_loop_callback(&self, loop_cb: Py<PyAny>) -> PyResult<()> {
-        *self.loop_cb.write().await = Some(loop_cb);
+        *self.loop_cb.write().await = Some(loop_cb).into();
         Ok(())
     }
 
     pub fn terminate(&self, py: Python<'_>) -> PyResult<()> {
-        if let Some(cb) = &self.close_cb {
+        if let Some(cb) = &*self.close_cb {
             cb.call0(py)?;
         }
 
@@ -225,12 +225,12 @@ impl AsyncTable {
     }
 
     pub async fn on_delete(&self, callback_py: Py<PyAny>) -> PyResult<u32> {
-        let loop_cb = self.client.loop_cb.read().await.clone();
+        let loop_cb = (*self.client.loop_cb.read().await).clone();
         let callback = {
             let callback_py = Python::with_gil(|py| Py::clone_ref(&callback_py, py));
             Box::new(move || {
                 Python::with_gil(|py| {
-                    if let Some(loop_cb) = &loop_cb {
+                    if let Some(loop_cb) = &*loop_cb {
                         loop_cb.call1(py, (&callback_py,))?;
                     } else {
                         callback_py.call0(py)?;
@@ -402,7 +402,7 @@ impl AsyncView {
             Box::new(move || {
                 let loop_cb = loop_cb.clone();
                 Python::with_gil(|py| {
-                    if let Some(loop_cb) = &loop_cb {
+                    if let Some(loop_cb) = &*loop_cb {
                         loop_cb.call1(py, (&*callback_py,))?;
                     } else {
                         callback_py.call0(py)?;
@@ -424,8 +424,8 @@ impl AsyncView {
 
     #[pyo3(signature=(callback, mode=None))]
     pub async fn on_update(&self, callback: Py<PyAny>, mode: Option<String>) -> PyResult<u32> {
-        let locked_val = self.client.loop_cb.read().await;
-        let loop_cb = Python::with_gil(|py| locked_val.as_ref().map(|v| Py::clone_ref(v, py)));
+        let locked_val = self.client.loop_cb.read().await.clone();
+        let loop_cb = Python::with_gil(|py| (*locked_val).as_ref().map(|v| Py::clone_ref(v, py)));
         let callback = move |x: ViewOnUpdateResp| {
             let loop_cb = Python::with_gil(|py| loop_cb.as_ref().map(|v| Py::clone_ref(v, py)));
             let callback = Python::with_gil(|py| Py::clone_ref(&callback, py));
diff --git a/rust/perspective-python/src/server/server_sync.rs b/rust/perspective-python/src/server/server_sync.rs
@@ -35,7 +35,7 @@ pub struct PySyncServer {
 }
 
 #[derive(Clone)]
-struct PyConnection(Py<PyAny>);
+struct PyConnection(Arc<Py<PyAny>>);
 
 impl SessionHandler for PyConnection {
     async fn send_response<'a>(
@@ -57,7 +57,7 @@ impl PySyncServer {
     pub fn new_session(&self, _py: Python, response_cb: Py<PyAny>) -> PySyncSession {
         let session = self
             .server
-            .new_session(PyConnection(response_cb))
+            .new_session(PyConnection(response_cb.into()))
             .block_on();
 
         let session = Arc::new(RwLock::new(Some(session)));
