Yicong-Huang commented on code in PR #6916:
URL: https://github.com/apache/texera/pull/6916#discussion_r3662053730
##########
amber/src/main/python/core/models/internal_queue.py:
##########
@@ -67,12 +72,23 @@ def __init__(self):
self._queue_ids: Set[ChannelIdentity] = set()
self._queue_state: Set[InternalQueue.DisableType] = set()
self._lock = RLock()
+ # Number of queued elements that represent work (see _is_work).
Counted at
+ # put/get time because the underlying multi-queue offers no iteration,
and
+ # it must see elements in disabled (paused/backpressured) sub-queues
too.
+ self._work_count = AtomicInteger()
def is_empty(self, key=None) -> bool:
return self._queue.is_empty(key)
def get(self) -> T:
- return self._queue.get()
+ item = self._queue.get()
+ if self._is_work(item):
+ self._work_count.dec()
+ return item
+
+ def peek(self) -> Optional[T]:
+ """Non-destructively return the next available item, or None when
empty."""
+ return self._queue.peek()
Review Comment:
I feel I reviewed similar code before. Let's not make this queue peekable?
##########
amber/src/main/python/core/models/internal_queue.py:
##########
@@ -67,12 +72,23 @@ def __init__(self):
self._queue_ids: Set[ChannelIdentity] = set()
self._queue_state: Set[InternalQueue.DisableType] = set()
self._lock = RLock()
+ # Number of queued elements that represent work (see _is_work).
Counted at
+ # put/get time because the underlying multi-queue offers no iteration,
and
+ # it must see elements in disabled (paused/backpressured) sub-queues
too.
+ self._work_count = AtomicInteger()
Review Comment:
this is a queue. let's not give it specific semantic like "work" to be done.
A queue is a data structure for FIFO, please let it stay general.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
