class Async::PriorityQueue

@public Since *Async v2*.
consumers are served first when items become available.
assigns priorities to consumers (fibers waiting to dequeue). Higher priority
Unlike a traditional priority queue where items have priorities, this queue
A queue which allows items to be processed in priority order of consumers.

def <<(item)

Compatibility with {::Queue#push}. 
def <<(item)
	self.push(item)
end

def async(priority: 0, parent: (@parent or Task.current), **options, &block)

@yields {|task| ...} When the system is idle, the block will be executed in a new task.
@parameter options [Hash] The options to pass to the task.
@parameter parent [Interface(:async) | Nil] The parent task to use for async operations.
@parameter priority [Numeric] The priority for processing items.

@asynchronous Executes the given block concurrently for each item.

Process each item in the queue. 
def async(priority: 0, parent: (@parent or Task.current), **options, &block)
	while item = self.dequeue(priority: priority)
		parent.async(item, **options, &block)
	end
end

def close

Any subsequent calls to {enqueue} will raise an exception.
Close the queue, causing all waiting tasks to return `nil`. 
def close
	@mutex.synchronize do
		@closed = true
		
		# Signal all waiting fibers with nil, skipping dead/invalid ones:
		while waiter = @waiting.pop
			waiter.signal(nil)
		end
	end
end

def closed?

@returns [Boolean] Whether the queue is closed. 
def closed?
	@closed
end

def dequeue(priority: 0, timeout: nil)

@returns [Object, nil] The next item in the queue, or nil if timeout expires.
@parameter timeout [Numeric, nil] Maximum time to wait for an item. If nil, waits indefinitely. If 0, returns immediately.
@parameter priority [Numeric] The priority of this consumer (higher = served first).

items first.
Fibers are served in priority order, with higher priority fibers receiving
If the queue is empty, this method will block until an item is available or timeout expires.

Remove and return the next item from the queue. 
def dequeue(priority: 0, timeout: nil)
	@mutex.synchronize do
		# If queue is closed and empty, return nil immediately:
		if @closed && @items.empty?
			return nil
		end
		
		# Fast path: if items available and either no waiters or we have higher priority:
		unless @items.empty?
			head = @waiting.peek
			if head.nil? or priority > head.priority
				return @items.shift
			end
		end
		
		# Handle immediate timeout (non-blocking)
		return nil if timeout == 0
		
		# Need to wait - create our own condition variable and add to waiting queue:
		sequence = @sequence
		@sequence += 1
		
		condition = ConditionVariable.new
		
		begin
			waiter = Waiter.new(Fiber.current, priority, sequence, condition, nil)
			@waiting.push(waiter)
			
			# Wait for our specific condition variable to be signaled:
			return waiter.wait_for_value(@mutex, timeout)
		ensure
			waiter&.invalidate!
		end
	end
end

def each(priority: 0)

@parameter priority [Numeric] The priority for dequeuing items.

Enumerate each item in the queue. 
def each(priority: 0)
	while item = self.dequeue(priority: priority)
		yield item
	end
end

def empty?

@returns [Boolean] Whether the queue is empty. 
def empty?
	@items.empty?
end

def enqueue(*items)

@parameter items [Array] The items to add to the queue.

Add multiple items to the queue. 
def enqueue(*items)
	@mutex.synchronize do
		if @closed
			raise ClosedError, "Cannot enqueue items to a closed queue."
		end
		
		@items.concat(items)
		
		# Wake up waiting fibers in priority order, skipping dead/invalid waiters:
		while !@items.empty? && (waiter = @waiting.pop)
			if waiter.valid?
				value = @items.shift
				waiter.signal(value)
			end
			# Dead/invalid waiter discarded, continue to next one.
		end
	end
end

def initialize(parent: nil)

@parameter parent [Interface(:async) | Nil] The parent task to use for async operations.

Create a new priority queue. 
def initialize(parent: nil)
	@items = []
	@closed = false
	@parent = parent
	@waiting = IO::Event::PriorityHeap.new
	@sequence = 0
	
	@mutex = Mutex.new
end

def pop(priority: 0, timeout: nil)

@returns [Object, nil] The dequeued item, or nil if timeout expires.
@parameter timeout [Numeric, nil] Maximum time to wait for an item. If nil, waits indefinitely. If 0, returns immediately.
@parameter priority [Numeric] The priority of this consumer.

Compatibility with {::Queue#pop}. 
def pop(priority: 0, timeout: nil)
	self.dequeue(priority: priority, timeout: timeout)
end

def push(item)

@parameter item [Object] The item to add to the queue.

Add an item to the queue. 
def push(item)
	@mutex.synchronize do
		if @closed
			raise ClosedError, "Cannot push items to a closed queue."
		end
		
		@items << item
		
		# Wake up the highest priority waiter if any, skipping dead/invalid waiters:
		while waiter = @waiting.pop
			if waiter.valid?
				value = @items.shift
				waiter.signal(value)
				break
			end
			# Dead/invalid waiter discarded, try next one.
		end
	end
end

def signal(value = nil)

Signal the queue with a value, the same as {#enqueue}. 
def signal(value = nil)
	self.enqueue(value)
end

def size

@returns [Integer] The number of items in the queue. 
def size
	@items.size
end

def wait(priority: 0)

@parameter priority [Numeric] The priority of this consumer.

Wait for an item to be available, the same as {#dequeue}. 
def wait(priority: 0)
	self.dequeue(priority: priority)
end

def waiting_count

@returns [Integer] The number of fibers waiting to dequeue. 
def waiting_count
	@mutex.synchronize do
		@waiting.size
	end
end