Skip to content

GitLab

Unverified Commit f88d7dc7 authored by Graham King's avatar Graham King Committed by GitHub
Browse files

chore(bindings): Remove NatsQueue (#3086) 


Signed-off-by: default avatarGraham King <grahamk@nvidia.com>
parent 3de04dd9
Show whitespace changes
Inline Side-by-side
Showing with 311 additions and 164 deletions
lib/bindings/python/Cargo.lock
View file @ f88d7dc7
......@@ -1215,6 +1215,27 @@ dependencies = [
"syn 2.0.106",
]
[[package]]
name = "derive_more"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4a9b99b9cbbe49445b21764dc0625032a89b145a2642e67603e1c936f5458d05"
dependencies = [
"derive_more-impl",
]
[[package]]
name = "derive_more-impl"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cb7330aeadfbe296029522e6c40f315320aba36fc43a5b3632f3795348f3bd22"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.106",
"unicode-xid",
]
[[package]]
name = "dialoguer"
version = "0.11.0"
......@@ -1442,8 +1463,10 @@ dependencies = [
"anyhow",
"dynamo-async-openai",
"lazy_static",
"num-traits",
"openai-harmony",
"regex",
"rustpython-parser",
"serde",
"serde_json",
"tracing",
......@@ -1462,6 +1485,7 @@ dependencies = [
"dlpark",
"dynamo-async-openai",
"dynamo-llm",
"dynamo-parsers",
"dynamo-runtime",
"either",
"futures",
......@@ -2214,6 +2238,15 @@ dependencies = [
"version_check",
]
[[package]]
name = "getopts"
version = "0.2.24"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cfe4fbac503b8d1f88e6676011885f34b7174f46e59956bba534ba83abded4df"
dependencies = [
"unicode-width",
]
[[package]]
name = "getrandom"
version = "0.2.16"
......@@ -2331,6 +2364,9 @@ name = "hashbrown"
version = "0.14.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
dependencies = [
"ahash",
]
[[package]]
name = "hashbrown"
......@@ -2808,12 +2844,33 @@ dependencies = [
"serde",
]
[[package]]
name = "is-macro"
version = "0.3.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1d57a3e447e24c22647738e4607f1df1e0ec6f72e16182c4cd199f647cdfb0e4"
dependencies = [
"heck",
"proc-macro2",
"quote",
"syn 2.0.106",
]
[[package]]
name = "is_terminal_polyfill"
version = "1.70.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7943c866cc5cd64cbc25b2e01621d07fa8eb2a1a23160ee81ce38704e97b8ecf"
[[package]]
name = "itertools"
version = "0.11.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b1c173a5686ce8bfa551b3563d0c2170bf24ca44da99c7ca4bfdab5418c3fe57"
dependencies = [
"either",
]
[[package]]
name = "itertools"
version = "0.12.1"
......@@ -2903,6 +2960,12 @@ dependencies = [
"winapi-build",
]
[[package]]
name = "lalrpop-util"
version = "0.20.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "507460a910eb7b32ee961886ff48539633b788a36b65692b95f225b844c82553"
[[package]]
name = "lazy_static"
version = "1.5.0"
......@@ -3040,6 +3103,64 @@ version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "670fdfda89751bc4a84ac13eaa63e205cf0fd22b4c9a5fbfa085b63c1f1d3a30"
[[package]]
name = "malachite"
version = "0.4.22"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2fbdf9cb251732db30a7200ebb6ae5d22fe8e11397364416617d2c2cf0c51cb5"
dependencies = [
"malachite-base",
"malachite-nz",
"malachite-q",
]
[[package]]
name = "malachite-base"
version = "0.4.22"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5ea0ed76adf7defc1a92240b5c36d5368cfe9251640dcce5bd2d0b7c1fd87aeb"
dependencies = [
"hashbrown 0.14.5",
"itertools 0.11.0",
"libm",
"ryu",
]
[[package]]
name = "malachite-bigint"
version = "0.2.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d149aaa2965d70381709d9df4c7ee1fc0de1c614a4efc2ee356f5e43d68749f8"
dependencies = [
"derive_more",
"malachite",
"num-integer",
"num-traits",
"paste",
]
[[package]]
name = "malachite-nz"
version = "0.4.22"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "34a79feebb2bc9aa7762047c8e5495269a367da6b5a90a99882a0aeeac1841f7"
dependencies = [
"itertools 0.11.0",
"libm",
"malachite-base",
]
[[package]]
name = "malachite-q"
version = "0.4.22"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "50f235d5747b1256b47620f5640c2a17a88c7569eebdf27cd9cb130e1a619191"
dependencies = [
"itertools 0.11.0",
"malachite-base",
"malachite-nz",
]
[[package]]
name = "matchers"
version = "0.1.0"
......@@ -3737,6 +3858,44 @@ dependencies = [
"indexmap 2.11.0",
]
[[package]]
name = "phf"
version = "0.11.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fd6780a80ae0c52cc120a26a1a42c1ae51b247a253e4e06113d23d2c2edd078"
dependencies = [
"phf_shared",
]
[[package]]
name = "phf_codegen"
version = "0.11.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "aef8048c789fa5e851558d709946d6d79a8ff88c0440c587967f8e94bfb1216a"
dependencies = [
"phf_generator",
"phf_shared",
]
[[package]]
name = "phf_generator"
version = "0.11.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3c80231409c20246a13fddb31776fb942c38553c51e871f8cbd687a4cfb5843d"
dependencies = [
"phf_shared",
"rand 0.8.5",
]
[[package]]
name = "phf_shared"
version = "0.11.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "67eabc2ef2a60eb7faa00097bd1ffdb5bd28e62bf39990626a582201b7a754e5"
dependencies = [
"siphasher",
]
[[package]]
name = "pin-project"
version = "1.1.10"
......@@ -4780,6 +4939,63 @@ dependencies = [
"untrusted",
]
[[package]]
name = "rustpython-ast"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4cdaf8ee5c1473b993b398c174641d3aa9da847af36e8d5eb8291930b72f31a5"
dependencies = [
"is-macro",
"malachite-bigint",
"rustpython-parser-core",
"static_assertions",
]
[[package]]
name = "rustpython-parser"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "868f724daac0caf9bd36d38caf45819905193a901e8f1c983345a68e18fb2abb"
dependencies = [
"anyhow",
"is-macro",
"itertools 0.11.0",
"lalrpop-util",
"log",
"malachite-bigint",
"num-traits",
"phf",
"phf_codegen",
"rustc-hash 1.1.0",
"rustpython-ast",
"rustpython-parser-core",
"tiny-keccak",
"unic-emoji-char",
"unic-ucd-ident",
"unicode_names2",
]
[[package]]
name = "rustpython-parser-core"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b4b6c12fa273825edc7bccd9a734f0ad5ba4b8a2f4da5ff7efe946f066d0f4ad"
dependencies = [
"is-macro",
"memchr",
"rustpython-parser-vendored",
]
[[package]]
name = "rustpython-parser-vendored"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "04fcea49a4630a3a5d940f4d514dc4f575ed63c14c3e3ed07146634aed7f67a6"
dependencies = [
"memchr",
"once_cell",
]
[[package]]
name = "rustversion"
version = "1.0.22"
......@@ -5119,6 +5335,12 @@ dependencies = [
"quote",
]
[[package]]
name = "siphasher"
version = "1.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "56199f7ddabf13fe5074ce809e7d3f42b42ae711800501b5b16ea82ad029c39d"
[[package]]
name = "slab"
version = "0.4.11"
......@@ -5445,6 +5667,15 @@ dependencies = [
"time-core",
]
[[package]]
name = "tiny-keccak"
version = "2.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2c9d3793400a45f954c52e73d068316d76b6f4e36977e3fcebb13a2721e80237"
dependencies = [
"crunchy",
]
[[package]]
name = "tinystr"
version = "0.8.1"
......@@ -5908,6 +6139,58 @@ version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "eeba86d422ce181a719445e51872fa30f1f7413b62becb52e95ec91aa262d85c"
[[package]]
name = "unic-char-property"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a8c57a407d9b6fa02b4795eb81c5b6652060a15a7903ea981f3d723e6c0be221"
dependencies = [
"unic-char-range",
]
[[package]]
name = "unic-char-range"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0398022d5f700414f6b899e10b8348231abf9173fa93144cbc1a43b9793c1fbc"
[[package]]
name = "unic-common"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "80d7ff825a6a654ee85a63e80f92f054f904f21e7d12da4e22f9834a4aaa35bc"
[[package]]
name = "unic-emoji-char"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0b07221e68897210270a38bde4babb655869637af0f69407f96053a34f76494d"
dependencies = [
"unic-char-property",
"unic-char-range",
"unic-ucd-version",
]
[[package]]
name = "unic-ucd-ident"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e230a37c0381caa9219d67cf063aa3a375ffed5bf541a452db16e744bdab6987"
dependencies = [
"unic-char-property",
"unic-char-range",
"unic-ucd-version",
]
[[package]]
name = "unic-ucd-version"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "96bd2f2237fe450fcd0a1d2f5f4e91711124f7857ba2e964247776ebeeb7b0c4"
dependencies = [
"unic-common",
]
[[package]]
name = "unicase"
version = "2.8.1"
......@@ -5947,12 +6230,40 @@ version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4a1a07cc7db3810833284e8d372ccdc6da29741639ecc70c9ec107df0fa6154c"
[[package]]
name = "unicode-xid"
version = "0.2.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853"
[[package]]
name = "unicode_categories"
version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "39ec24b3121d976906ece63c9daad25b85969647682eee313cb5779fdd69e14e"
[[package]]
name = "unicode_names2"
version = "1.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d1673eca9782c84de5f81b82e4109dcfb3611c8ba0d52930ec4a9478f547b2dd"
dependencies = [
"phf",
"unicode_names2_generator",
]
[[package]]
name = "unicode_names2_generator"
version = "1.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b91e5b84611016120197efd7dc93ef76774f4e084cd73c9fb3ea4a86c570c56e"
dependencies = [
"getopts",
"log",
"phf_codegen",
"rand 0.8.5",
]
[[package]]
name = "unindent"
version = "0.2.4"
......
lib/bindings/python/rust/lib.rs
View file @ f88d7dc7
......@@ -102,7 +102,6 @@ fn _core(m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_class::<llm::kv::ZmqKvEventPublisher>()?;
m.add_class::<llm::kv::ZmqKvEventPublisherConfig>()?;
m.add_class::<llm::kv::KvRecorder>()?;
m.add_class::<llm::nats::NatsQueue>()?;
m.add_class::<http::HttpService>()?;
m.add_class::<http::HttpError>()?;
m.add_class::<http::HttpAsyncEngine>()?;
......
lib/bindings/python/rust/llm.rs
View file @ f88d7dc7
......@@ -32,7 +32,6 @@ pub mod entrypoint;
pub mod kv;
pub mod local_model;
pub mod model_card;
pub mod nats;
pub mod preprocessor;
#[cfg(feature = "block-manager")]
......
lib/bindings/python/rust/llm/nats.rs deleted 100644 → 0
View file @ 3de04dd9
// SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0
use super::*;
#[pyclass(subclass)]
pub(crate) struct NatsQueue {
inner: Arc<Mutex<crate::rs::transports::nats::NatsQueue>>,
}
#[pymethods]
impl NatsQueue {
#[new]
#[pyo3(signature = (stream_name, nats_server, dequeue_timeout))]
fn new(stream_name: String, nats_server: String, dequeue_timeout: f64) -> PyResult<Self> {
let inner = Arc::new(Mutex::new(crate::rs::transports::nats::NatsQueue::new(
stream_name,
nats_server,
std::time::Duration::from_secs(dequeue_timeout as u64),
)));
Ok(Self { inner })
}
fn connect<'p>(&mut self, py: Python<'p>) -> PyResult<Bound<'p, PyAny>> {
let queue = self.inner.clone();
pyo3_async_runtimes::tokio::future_into_py(py, async move {
queue.lock().await.connect().await.map_err(to_pyerr)?;
Ok(())
})
}
fn ensure_connection<'p>(&mut self, py: Python<'p>) -> PyResult<Bound<'p, PyAny>> {
let queue = self.inner.clone();
pyo3_async_runtimes::tokio::future_into_py(py, async move {
queue
.lock()
.await
.ensure_connection()
.await
.map_err(to_pyerr)?;
Ok(())
})
}
fn close<'p>(&mut self, py: Python<'p>) -> PyResult<Bound<'p, PyAny>> {
let queue = self.inner.clone();
pyo3_async_runtimes::tokio::future_into_py(py, async move {
queue.lock().await.close().await.map_err(to_pyerr)?;
Ok(())
})
}
fn enqueue_task<'p>(
&mut self,
py: Python<'p>,
task_data: Py<PyBytes>,
) -> PyResult<Bound<'p, PyAny>> {
let bytes = task_data.as_bytes(py).to_vec();
let queue = self.inner.clone();
pyo3_async_runtimes::tokio::future_into_py(py, async move {
queue
.lock()
.await
.enqueue_task(bytes.into())
.await
.map_err(to_pyerr)?;
Ok(())
})
}
#[pyo3(signature = (timeout=None))]
fn dequeue_task<'p>(
&mut self,
py: Python<'p>,
timeout: Option<f64>,
) -> PyResult<Bound<'p, PyAny>> {
let queue = self.inner.clone();
pyo3_async_runtimes::tokio::future_into_py(py, async move {
let timeout_duration = timeout.map(std::time::Duration::from_secs_f64);
Ok(queue
.lock()
.await
.dequeue_task(timeout_duration)
.await
.map_err(to_pyerr)?
.map(|bytes| bytes.to_vec()))
})
}
fn get_queue_size<'p>(&mut self, py: Python<'p>) -> PyResult<Bound<'p, PyAny>> {
let queue = self.inner.clone();
pyo3_async_runtimes::tokio::future_into_py(py, async move {
queue.lock().await.get_queue_size().await.map_err(to_pyerr)
})
}
}
lib/bindings/python/src/dynamo/_core.pyi
View file @ f88d7dc7
......@@ -893,71 +893,6 @@ async def run_input(runtime: DistributedRuntime, input: str, engine_config: Engi
"""Start an engine, connect it to an input, and run until stopped."""
...
class NatsQueue:
"""
A queue implementation using NATS JetStream for task distribution
"""
def __init__(self, stream_name: str, nats_server: str, dequeue_timeout: float) -> None:
"""
Create a new NatsQueue instance.
Args:
stream_name: Name of the NATS JetStream stream
nats_server: URL of the NATS server
dequeue_timeout: Default timeout in seconds for dequeue operations
"""
...
async def connect(self) -> None:
"""
Connect to the NATS server
"""
...
async def ensure_connection(self) -> None:
"""
Ensure connection to the NATS server, connecting if not already connected
"""
...
async def close(self) -> None:
"""
Close the connection to the NATS server
"""
...
async def enqueue_task(self, task_data: bytes) -> None:
"""
Enqueue a task to the NATS JetStream
Args:
task_data: The task data as bytes
"""
...
async def dequeue_task(self, timeout: Optional[float] = None) -> Optional[bytes]:
"""
Dequeue a task from the NATS JetStream
Args:
timeout: Optional timeout in seconds for this specific dequeue operation.
If None, uses the default timeout specified during initialization.
Returns:
The task data as bytes if available, None if no task is available
"""
...
async def get_queue_size(self) -> int:
"""
Get the current size of the queue
Returns:
The number of messages in the queue
"""
...
class Layer:
"""
A KV cache block layer
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment