How to use the readerwriterlock.rwlock.RWLockFairD function in readerwriterlock

To help you get started, we’ve selected a few readerwriterlock examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github elarivie / pyReaderWriterLock / tests / test_rwlock.py View on Github external
def test_read_vs_downgrade_fair(self) -> None:
		"""
		# Given: Instance of RWLockFairD.

		# When: A reader lock is acquired OR A writer lock is downgraded.

		# Then: The internal state should be the same.
		"""
		# ## Arrange
		c_rwlock_1 = rwlock.RWLockFairD()
		c_rwlock_2 = rwlock.RWLockFairD()

		def assert_internal_state():
			"""Assert internal."""
			self.assertEqual(int(c_rwlock_1.v_read_count), int(c_rwlock_2.v_read_count))
			self.assertEqual(bool(c_rwlock_1.c_lock_read_count.locked()), bool(c_rwlock_2.c_lock_read_count.locked()))
			self.assertEqual(bool(c_rwlock_1.c_lock_read.locked()), bool(c_rwlock_2.c_lock_read.locked()))
			self.assertEqual(bool(c_rwlock_1.c_lock_write.locked()), bool(c_rwlock_2.c_lock_write.locked()))

		# ## Assume
		assert_internal_state()

		# ## Act
		a_read_lock = c_rwlock_1.gen_rlock()
		a_read_lock.acquire()
		a_downgrade_lock: Union[rwlock.LockableD, rwlock.Lockable] = c_rwlock_2.gen_wlock()
github elarivie / pyReaderWriterLock / tests / test_rwlock.py View on Github external
def test_read_vs_downgrade_fair(self) -> None:
		"""
		# Given: Instance of RWLockFairD.

		# When: A reader lock is acquired OR A writer lock is downgraded.

		# Then: The internal state should be the same.
		"""
		# ## Arrange
		c_rwlock_1 = rwlock.RWLockFairD()
		c_rwlock_2 = rwlock.RWLockFairD()

		def assert_internal_state():
			"""Assert internal."""
			self.assertEqual(int(c_rwlock_1.v_read_count), int(c_rwlock_2.v_read_count))
			self.assertEqual(bool(c_rwlock_1.c_lock_read_count.locked()), bool(c_rwlock_2.c_lock_read_count.locked()))
			self.assertEqual(bool(c_rwlock_1.c_lock_read.locked()), bool(c_rwlock_2.c_lock_read.locked()))
			self.assertEqual(bool(c_rwlock_1.c_lock_write.locked()), bool(c_rwlock_2.c_lock_write.locked()))

		# ## Assume
		assert_internal_state()

		# ## Act
		a_read_lock = c_rwlock_1.gen_rlock()
		a_read_lock.acquire()
		a_downgrade_lock: Union[rwlock.LockableD, rwlock.Lockable] = c_rwlock_2.gen_wlock()
		a_downgrade_lock.acquire()