apply reviews

Author: chenyushuo

trinity/algorithm/sample_strategy/sample_strategy.py

@@ -82,10 +82,10 @@ def __init__(self, buffer_config: BufferConfig, **kwargs):
         self.staleness_limit = kwargs.get("staleness_limit", float("inf"))
 
     async def sample(self, step: int, **kwargs) -> Tuple[Experiences, Dict, List]:
-        oldest_valid_version = max(step - self.staleness_limit, 0)
+        min_model_version = max(step - self.staleness_limit, 0)
         metrics = {}
         with Timer(metrics, "time/read_experience"):
-            exp_list = await self.exp_buffer.read_async(oldest_valid_version=oldest_valid_version)
+            exp_list = await self.exp_buffer.read_async(min_model_version=min_model_version)
             repr_samples = representative_sample(exp_list)
         self.set_model_version_metric(exp_list, metrics)
         with Timer(metrics, "time/gather_experience"):

trinity/buffer/storage/queue.py

@@ -100,8 +100,8 @@ def default_config(cls) -> Dict:
 
 
 class QueueBuffer(ABC):
-    async def set_oldest_valid_version(self, oldest_valid_version: int):
-        self.oldest_valid_version = max(oldest_valid_version, 0)
+    async def set_min_model_version(self, min_model_version: int):
+        self.min_model_version = max(min_model_version, 0)
 
     @abstractmethod
     async def put(self, exps: List[Experience]) -> None:
@@ -152,7 +152,7 @@ def __init__(self, capacity: int):
         """
         super().__init__(maxsize=capacity)
         self._closed = False
-        self.oldest_valid_version = 0
+        self.min_model_version = 0
 
     async def put(self, item: List[Experience]):
         if len(item) == 0:
@@ -163,8 +163,8 @@ async def get(self):
         while True:
             item = await super().get()
             if (
-                self.oldest_valid_version <= 0
-                or item[0].info["model_version"] >= self.oldest_valid_version
+                self.min_model_version <= 0
+                or item[0].info["model_version"] >= self.min_model_version
             ):
                 return item
 
@@ -222,7 +222,7 @@ def __init__(
         self.reuse_cooldown_time = reuse_cooldown_time
         self._condition = asyncio.Condition()  # For thread-safe operations
         self._closed = False
-        self.oldest_valid_version = 0
+        self.min_model_version = 0
 
     async def _put(self, item: List[Experience], delay: float = 0) -> None:
         """
@@ -287,8 +287,8 @@ async def get(self) -> List[Experience]:
                     self.priority_groups.popitem(index=-1)
 
                 if (
-                    self.oldest_valid_version <= 0
-                    or item[0].info["model_version"] >= self.oldest_valid_version
+                    self.min_model_version <= 0
+                    or item[0].info["model_version"] >= self.min_model_version
                 ):
                     break
 
@@ -374,17 +374,15 @@ async def put_batch(self, exp_list: List) -> None:
         if self.writer is not None:
             self.writer.write(exp_list)
 
-    async def get_batch(
-        self, batch_size: int, timeout: float, oldest_valid_version: int = 0
-    ) -> List:
+    async def get_batch(self, batch_size: int, timeout: float, min_model_version: int = 0) -> List:
         """Get batch of experience."""
-        await self.queue.set_oldest_valid_version(oldest_valid_version)
+        await self.queue.set_min_model_version(min_model_version)
         start_time = time.time()
         result = []
         while len(result) < batch_size:
             while len(self.exp_pool) > 0 and len(result) < batch_size:
                 exp = self.exp_pool.popleft()
-                if oldest_valid_version > 0 and exp.info["model_version"] < oldest_valid_version:
+                if min_model_version > 0 and exp.info["model_version"] < min_model_version:
                     continue
                 result.append(exp)
             if len(result) >= batch_size:

trinity/buffer/storage/sql.py

@@ -143,7 +143,7 @@ def _read_fifo(self, batch_size: int) -> List[Experience]:
                 time.sleep(1)
         return exp_list
 
-    def _read_priority(self, batch_size: int, oldest_valid_version: int = 0) -> List[Experience]:
+    def _read_priority(self, batch_size: int, min_model_version: int = 0) -> List[Experience]:
         exp_list = []
         start_time = time.time()
         latest_size = 0
@@ -161,8 +161,8 @@ def _read_priority(self, batch_size: int, oldest_valid_version: int = 0) -> List
                 query = session.query(self.table_model_cls).order_by(
                     asc(self.table_model_cls.consumed), desc(self.table_model_cls.id)
                 )
-                if oldest_valid_version > 0:
-                    query = query.filter(self.table_model_cls.model_version >= oldest_valid_version)
+                if min_model_version > 0:
+                    query = query.filter(self.table_model_cls.model_version >= min_model_version)
                 experiences = query.limit(batch_size).with_for_update().all()
                 if len(experiences) != batch_size:
                     if latest_size != len(experiences):