Upload large blob fixes

packages/syft/src/syft/service/action/action_object.py

@@ -8,6 +8,7 @@
 import inspect
 from io import BytesIO
 from pathlib import Path
+import sys
 import threading
 import time
 import traceback
@@ -769,7 +770,12 @@ def _save_to_blob_storage_(self, data: Any) -> SyftError | None:
                 )
                 data.upload_to_blobstorage_from_api(api)
             else:
-                storage_entry = CreateBlobStorageEntry.from_obj(data)
+                serialized = serialize(data, to_bytes=True)
+                size = sys.getsizeof(serialized)
+                storage_entry = CreateBlobStorageEntry.from_obj(data, file_size=size)
+
+                if not TraceResultRegistry.current_thread_is_tracing():
+                    self.syft_action_data_cache = self.as_empty_data()
                 if self.syft_blob_storage_entry_id is not None:
                     # TODO: check if it already exists
                     storage_entry.id = self.syft_blob_storage_entry_id
@@ -784,9 +790,7 @@ def _save_to_blob_storage_(self, data: Any) -> SyftError | None:
                     if isinstance(blob_deposit_object, SyftError):
                         return blob_deposit_object
 
-                    result = blob_deposit_object.write(
-                        BytesIO(serialize(data, to_bytes=True))
-                    )
+                    result = blob_deposit_object.write(BytesIO(serialized))
                     if isinstance(result, SyftError):
                         return result
                     self.syft_blob_storage_entry_id = (

packages/syft/src/syft/store/blob_storage/seaweedfs.py

@@ -35,9 +35,10 @@
 from ...types.syft_object import SYFT_OBJECT_VERSION_3
 from ...util.constants import DEFAULT_TIMEOUT
 
+MAX_QUEUE_SIZE = 100
 WRITE_EXPIRATION_TIME = 900  # seconds
-DEFAULT_FILE_PART_SIZE = (1024**3) * 5  # 5GB
-DEFAULT_UPLOAD_CHUNK_SIZE = 819200
+DEFAULT_FILE_PART_SIZE = 1024**3 / 100  # 10MB
+DEFAULT_UPLOAD_CHUNK_SIZE = 1024 * 800  # 800KB
 
 
 @serializable()
@@ -94,7 +95,7 @@ def __init__(self) -> None:
                         def async_generator(
                             self, chunk_size: int = DEFAULT_UPLOAD_CHUNK_SIZE
                         ) -> Generator:
-                            item_queue: Queue = Queue()
+                            item_queue: Queue = Queue(maxsize=MAX_QUEUE_SIZE)
                             threading.Thread(
                                 target=self.add_chunks_to_queue,
                                 kwargs={"queue": item_queue, "chunk_size": chunk_size},

packages/syft/src/syft/types/blob_storage.py

@@ -327,8 +327,9 @@ class CreateBlobStorageEntry(SyftObject):
     extensions: list[str] = []
 
     @classmethod
-    def from_obj(cls, obj: SyftObject) -> Self:
-        file_size = sys.getsizeof(serialize._serialize(obj=obj, to_bytes=True))
+    def from_obj(cls, obj: SyftObject, file_size: int | None = None) -> Self:
+        if file_size is None:
+            file_size = sys.getsizeof(serialize._serialize(obj=obj, to_bytes=True))
         return cls(file_size=file_size, type_=type(obj))
 
     @classmethod