Source

cpython-withatomic / Lib / Queue.py

# A multi-producer, multi-consumer queue.

Empty = 'Queue.Empty' # Exception raised by get_nowait()

class Queue:

	# Initialize a queue object with a given maximum size
	# (If maxsize is <= 0, the maximum size is infinite)
	def init(self, maxsize):
		import thread
		self._init(maxsize)
		self.mutex = thread.allocate_lock()
		self.esema = thread.allocate_lock()
		self.esema.acquire_lock()
		self.fsema = thread.allocate_lock()
		return self

	# Get an approximation of the queue size (not reliable!)
	def qsize(self):
		self.mutex.acquire_lock()
		n = self._qsize()
		self.mutex.release_lock()
		return n

	# Check if the queue is empty (not reliable!)
	def empty(self):
		self.mutex.acquire_lock()
		n = self._empty()
		self.mutex.release_lock()
		return n

	# Check if the queue is full (not reliable!)
	def full(self):
		self.mutex.acquire_lock()
		n = self._full()
		self.mutex.release_lock()
		return n

	# Put a new item into the queue
	def put(self, item):
		self.fsema.acquire_lock()
		self.mutex.acquire_lock()
		was_empty = self._empty()
		self._put(item)
		if was_empty:
			self.esema.release_lock()
		if not self._full():
			self.fsema.release_lock()
		self.mutex.release_lock()

	# Get an item from the queue,
	# blocking if necessary until one is available
	def get(self):
		self.esema.acquire_lock()
		self.mutex.acquire_lock()
		was_full = self._full()
		item = self._get()
		if was_full:
			self.fsema.release_lock()
		if not self._empty():
			self.esema.release_lock()
		self.mutex.release_lock()
		return item

	# Get an item from the queue if one is immediately available,
	# raise Empty if the queue is empty or temporarily unavailable
	def get_nowait(self):
		locked = self.esema.acquire_lock(0)
		self.mutex.acquire_lock()
		if self._empty():
			# The queue is empyt -- we can't have esema
			self.mutex.release_lock()
			raise Empty
		if not locked:
			locked = self.esema.acquire_lock(0)
			if not locked:
				# Somebody else has esema
				# but we have mutex --
				# go out of their way
				self.mutex.release_lock()
				raise Empty
		was_full = self._full()
		item = self._get()
		if was_full:
			self.fsema.release_lock()
		if not self._empty():
			self.esema.release_lock()
		self.mutex.release_lock()
		return item

	# XXX Need to define put_nowait() as well.
		

	# Override these methods to implement other queue organizations
	# (e.g. stack or priority queue).
	# These will only be called with appropriate locks held

	# Initialize the queue representation
	def _init(self, maxsize):
		self.maxsize = maxsize
		self.queue = []

	def _qsize(self):
		return len(self.queue)

	# Check wheter the queue is empty
	def _empty(self):
		return not self.queue

	# Check whether the queue is full
	def _full(self):
		return self.maxsize > 0 and len(self.queue) == self.maxsize

	# Put a new item in the queue
	def _put(self, item):
		self.queue.append(item)

	# Get an item from the queue
	def _get(self):
		item = self.queue[0]
		del self.queue[0]
		return item
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.