Skip to content

[fix,serialization] Fix FieldMeta status update and remove unnecessary copy and use recv_multipart(copy=False) by default #46

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
0oshowero0 merged 9 commits into from
Mar 19, 2026
Merged

[fix,serialization] Fix FieldMeta status update and remove unnecessary copy and use recv_multipart(copy=False) by default #46

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
ddef1cf
remove unnecessary copy
0oshowero0 Mar 16, 2026
121339b
fix comments
0oshowero0 Mar 16, 2026
7297090
try: fix metadata error
0oshowero0 Mar 17, 2026
fc0ca4a
fix more metadata error
0oshowero0 Mar 17, 2026
7c55114
fix more metadata error
0oshowero0 Mar 17, 2026
7afd726
fix mooncake backend
0oshowero0 Mar 17, 2026
e8fca14
fix
0oshowero0 Mar 17, 2026
bd38912
fully remove FieldMeta if empty
0oshowero0 Mar 18, 2026
0f92c1e
solve comments
0oshowero0 Mar 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 1 addition & 3 deletions tests/test_async_simple_storage_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,10 +128,8 @@ async def test_async_storage_manager_mock_operations(mock_async_storage_manager)
manager._put_to_single_storage_unit = AsyncMock()
manager._get_from_single_storage_unit = AsyncMock(
return_value=(
[0, 1],
["test_field"],
{"test_field": [torch.tensor([1.0, 2.0]), torch.tensor([3.0, 4.0])]},
b"this is the serialized message object.",
)
)
manager._clear_single_storage_unit = AsyncMock()
Expand Down Expand Up @@ -286,7 +284,7 @@ async def fake_get(global_indexes, fields, target_storage_unit=None, **kwargs):
su = target_storage_unit
called_with[su] = list(global_indexes)
tensors = [torch.zeros(2) for _ in global_indexes]
return global_indexes, fields, {"f": tensors}, b""
return fields, {"f": tensors}

manager._get_from_single_storage_unit = fake_get

Expand Down
4 changes: 2 additions & 2 deletions tests/test_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,7 @@ def _handle_requests(self):
try:
socks = dict(poller.poll(100)) # 100ms timeout
if self.request_socket in socks:
messages = self.request_socket.recv_multipart()
messages = self.request_socket.recv_multipart(copy=False)
identity = messages.pop(0)
serialized_msg = messages
request_msg = ZMQMessage.deserialize(serialized_msg)
Expand Down Expand Up @@ -332,7 +332,7 @@ def _handle_data_requests(self):
try:
socks = dict(poller.poll(100)) # 100ms timeout
if self.data_socket in socks:
messages = self.data_socket.recv_multipart()
messages = self.data_socket.recv_multipart(copy=False)
identity = messages.pop(0)
serialized_msg = messages
msg = ZMQMessage.deserialize(serialized_msg)
Expand Down
Loading
Loading