pdZddlmZddlmZddlmZddlZddlmZdZdZ Gd d Z Gd d Z y) zImplements a buffer for tasks used in task_graph_executor. See go/parallel-processing-in-gcloud-storage for more information. )absolute_import)division)unicode_literalsN)queuezBuffer Contents: zTask Buffer is empty.ceZdZdZdZdZy)_PriorityWrapperaWraps a buffered task and tracks priority information. Attributes: task (Union[task.Task, str]): A buffered item. Expected to be a task or a string (to handle shutdowns) when used by task_graph_executor. priority (int): The priority of this task. A task with a lower value will be executed before a task with a higher value, since queue.PriorityQueue uses a min-heap. c ||_||_yN)taskpriority)selfr r s ;lib/googlecloudsdk/command_lib/storage/tasks/task_buffer.py__init__z_PriorityWrapper.__init__,sDIDMc4|j|jkSr )r )r others r__lt__z_PriorityWrapper.__lt__0s ==5>> ))rN)__name__ __module__ __qualname____doc__rrrrrr!s*rrc6eZdZdZdZdZd dZdefdZdZ y) TaskBuffera'Stores and prioritizes tasks. The current implementation uses a queue.PriorityQueue under the hood, since in experiments we found that the heap it maintains did not add too much overhead. If it does end up being a bottleneck, the same API can be implemented with a collections.deque. c6tj|_yr )r PriorityQueue_queuer s rrzTaskBuffer.__init__=s%%'DKrcJ|jjjS)a6Removes and returns an item from the buffer. Calls to `get` block if there are no elements in the queue, and return prioritized items before non-prioritized items. Returns: A buffered item. Expected to be a task or a string (to handle shutdowns) when used by task_graph_executor. )rgetr rs rr zTaskBuffer.get@s ;;??  ! !!rc^|rdnd}t||}|jj|y)aHAdds an item to the buffer. Args: task (Union[task.Task, str]): A buffered item. Expected to be a task or a string (to handle shutdowns) when used by task_graph_executor. prioritize (bool): Tasks added with prioritize=True will be returned by `get` before tasks added with prioritize=False. rN)rrput)r r prioritizer prioritized_items rr#zTaskBuffer.putLs+qAH'h7KKOO$%rreturnc6|jjS)z*Returns the number of items in the buffer.)rqsizers rsizezTaskBuffer.sizeYs ;;   rc*|jdk(rtStg}tj|j j }|r8|jd}|jt|j|r8dj|S)z.Returns a string representation of the buffer.r ) r)BUFFER_EMPTY_MESSAGE BUFFER_HEADERcopydeepcopyrrpopappendstrr join)r output_lines temp_queuepriority_wrappers r__str__zTaskBuffer.__str__]sy yy{a !!"?Lt{{001J #*#.3345  99\ ""rN)F) rrrrrr r#intr)r7rrrrr4s'( " &C #rr) r __future__rrrr. six.movesrr-r,rrrrrr;s< '' % .**&5#5#r