How to use the readerwriterlock.rwlock.RWLockWriteD 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 setUp(self) -> None:
		"""Test setup."""
		self.c_rwlock_type_downgradable = (rwlock.RWLockWriteD, rwlock.RWLockFairD, rwlock.RWLockReadD)
		self.c_rwlock_type = (rwlock.RWLockRead, rwlock.RWLockWrite, rwlock.RWLockFair) + self.c_rwlock_type_downgradable
github elarivie / pyReaderWriterLock / tests / test_rwlock.py View on Github external
def test_read_vs_downgrade_write(self) -> None:
		"""
		# Given: Instance of RWLockWriteD.

		# 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.RWLockWriteD()
		c_rwlock_2 = rwlock.RWLockWriteD()

		def assert_internal_state():
			self.assertEqual(int(c_rwlock_1.v_read_count), int(c_rwlock_2.v_read_count))
			self.assertEqual(int(c_rwlock_1.v_write_count), int(c_rwlock_2.v_write_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_write_count.locked()), bool(c_rwlock_2.c_lock_write_count.locked()))

			self.assertEqual(bool(c_rwlock_1.c_lock_read_entry.locked()), bool(c_rwlock_2.c_lock_read_entry.locked()))
			self.assertEqual(bool(c_rwlock_1.c_lock_read_try.locked()), bool(c_rwlock_2.c_lock_read_try.locked()))
			self.assertEqual(bool(c_rwlock_1.c_resource.locked()), bool(c_rwlock_2.c_resource.locked()))

		# ## Assume
		assert_internal_state()

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

		# 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.RWLockWriteD()
		c_rwlock_2 = rwlock.RWLockWriteD()

		def assert_internal_state():
			self.assertEqual(int(c_rwlock_1.v_read_count), int(c_rwlock_2.v_read_count))
			self.assertEqual(int(c_rwlock_1.v_write_count), int(c_rwlock_2.v_write_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_write_count.locked()), bool(c_rwlock_2.c_lock_write_count.locked()))

			self.assertEqual(bool(c_rwlock_1.c_lock_read_entry.locked()), bool(c_rwlock_2.c_lock_read_entry.locked()))
			self.assertEqual(bool(c_rwlock_1.c_lock_read_try.locked()), bool(c_rwlock_2.c_lock_read_try.locked()))
			self.assertEqual(bool(c_rwlock_1.c_resource.locked()), bool(c_rwlock_2.c_resource.locked()))

		# ## Assume
		assert_internal_state()

		# ## Act
		a_read_lock = c_rwlock_1.gen_rlock()
github elarivie / pyReaderWriterLock / readerwriterlock / rwlock.py View on Github external
def gen_rlock(self) -> "RWLockWriteD._aReader":
		"""Generate a reader lock."""
		return RWLockWriteD._aReader(self)
github elarivie / pyReaderWriterLock / readerwriterlock / rwlock.py View on Github external
def gen_wlock(self) -> "RWLockWriteD._aWriter":
		"""Generate a writer lock."""
		return RWLockWriteD._aWriter(self)