(self, config: StorageConfig)
| 321 | """An wrapper of a async queue.""" |
| 322 | |
| 323 | def __init__(self, config: StorageConfig) -> None: |
| 324 | self.logger = get_logger(f"queue_{config.name}", in_ray_actor=True) |
| 325 | self.config = config |
| 326 | self.capacity = config.capacity |
| 327 | self.queue = QueueBuffer.get_queue(config) |
| 328 | st_config = deepcopy(config) |
| 329 | st_config.wrap_in_ray = False |
| 330 | if st_config.path: |
| 331 | if is_database_url(st_config.path): |
| 332 | from trinity.buffer.writer.sql_writer import SQLWriter |
| 333 | |
| 334 | st_config.storage_type = StorageType.SQL.value |
| 335 | self.writer = SQLWriter(st_config) |
| 336 | elif is_json_file(st_config.path): |
| 337 | from trinity.buffer.writer.file_writer import JSONWriter |
| 338 | |
| 339 | st_config.storage_type = StorageType.FILE.value |
| 340 | self.writer = JSONWriter(st_config) |
| 341 | else: |
| 342 | self.logger.warning("Unknown supported storage path: %s", st_config.path) |
| 343 | self.writer = None |
| 344 | else: |
| 345 | from trinity.buffer.writer.file_writer import JSONWriter |
| 346 | |
| 347 | st_config.storage_type = StorageType.FILE.value |
| 348 | self.writer = JSONWriter(st_config) |
| 349 | self.logger.warning(f"Save experiences in {st_config.path}.") |
| 350 | self.ref_count = 0 |
| 351 | self.exp_pool = deque() # A pool to store experiences |
| 352 | self.closed = False |
| 353 | |
| 354 | async def acquire(self) -> int: |
| 355 | self.ref_count += 1 |
nothing calls this directly
no test coverage detected