fix unittest

Author: chenyushuo

trinity/buffer/reader/queue_reader.py

@@ -18,7 +18,6 @@ def __init__(self, config: StorageConfig):
         self.timeout = config.max_read_timeout
         self.read_batch_size = config.batch_size
         self.queue = QueueStorage.get_wrapper(config)
-        ray.get(self.queue.acquire.remote())
 
     def read(self, batch_size: Optional[int] = None, **kwargs) -> List:
         try:
@@ -48,6 +47,3 @@ def state_dict(self) -> Dict:
     def load_state_dict(self, state_dict):
         # Queue Not supporting state dict yet
         return None
-
-    def __del__(self):
-        ray.get(self.queue.release.remote())

trinity/buffer/reader/sql_reader.py

@@ -17,7 +17,6 @@ def __init__(self, config: StorageConfig) -> None:
         assert config.storage_type == StorageType.SQL.value
         self.wrap_in_ray = config.wrap_in_ray
         self.storage = SQLStorage.get_wrapper(config)
-        ray.get(self.storage.acquire.remote())
 
     def read(self, batch_size: Optional[int] = None, **kwargs) -> List:
         if self.wrap_in_ray:
@@ -41,6 +40,3 @@ def state_dict(self) -> Dict:
     def load_state_dict(self, state_dict):
         # SQL Not supporting state dict yet
         return None
-
-    def __del__(self):
-        ray.get(self.storage.release.remote())

trinity/explorer/explorer.py

@@ -448,6 +448,9 @@ async def shutdown(self) -> None:
             self.scheduler = None
         if self.experience_pipeline:
             await self.experience_pipeline.close.remote()
+            # reserve `experience_pipeline.output` for trainer
+            # TODO: refactor the lifecycle of buffer actor
+            self._old_experience_pipeline = self.experience_pipeline
             self.experience_pipeline = None
         if self.monitor:
             self.monitor.close()