Skip to content

Commit 7a7f8a6

Browse files
committed
Fixed blocking io calls.
1 parent 4907a68 commit 7a7f8a6

2 files changed

Lines changed: 14 additions & 2 deletions

File tree

‎python/natsrpy/_natsrpy_rs/js/object_store.pyi‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ class ObjectStore:
3131
self,
3232
name: str,
3333
writer: Writer[bytes],
34-
buf_size: int | None = None,
34+
chunk_size: int | None = 24576, # 24MB
3535
) -> None: ...
3636
async def put(
3737
self,

‎src/js/object_store.rs‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,7 @@ impl ObjectStore {
108108
chunk_size: Option<usize>,
109109
) -> NatsrpyResult<Bound<'py, PyAny>> {
110110
let ctx_guard = self.object_store.clone();
111+
let arc_writer = Arc::new(writer);
111112
natsrpy_future(py, async move {
112113
let mut object = ctx_guard.read().await.get(name).await?;
113114
let mut buf =
@@ -117,7 +118,18 @@ impl ObjectStore {
117118
if read == 0 {
118119
break;
119120
}
120-
Python::attach(|gil| writer.call_method1(gil, "write", (&buf[..read],)))?;
121+
// Buffer is cheap to clone. Since
122+
// it copies only pointer to memory.
123+
let to_write = buf.clone();
124+
// Writer is wrapped into Arc, so it's also
125+
// cheap to clone. So its fine.
126+
let writer_ref = arc_writer.clone();
127+
tokio::task::spawn_blocking(move || {
128+
Python::attach(|gil| {
129+
writer_ref.call_method1(gil, "write", (&to_write[..read],))
130+
})
131+
})
132+
.await??;
121133
buf.clear();
122134
}
123135
Ok(())

0 commit comments

Comments
 (0)