How to use readerwriterlock - 10 common examples

def downgrader1() -> None:
					"""Downgrader using a timeout blocking acquire strategy."""
					if c_curr_lock_type not in self.c_rwlock_type_downgradable: return
					try:
						nonlocal v_value
						c_enter_time = time.time()
						while time.time() - c_enter_time <= s_period_sec:
							c_lock_w1: Union[rwlock.Lockable, rwlock.LockableD] = c_curr_rw_lock.gen_wlock()
							assert isinstance(c_lock_w1, rwlock.LockableD), type(c_lock_w1)
							time.sleep(sys.float_info.min)
							locked: bool = c_lock_w1.acquire(blocking=True, timeout=sys.float_info.min)
							if locked:
								try:
									# Asert like a writer
									v_temp = v_value
									v_value += 1
									self.assertEqual(v_value, (v_temp + 1))

									assert isinstance(c_lock_w1, rwlock.LockableD), c_lock_w1
									c_lock_w1 = c_lock_w1.downgrade()
									assert isinstance(c_lock_w1, rwlock.Lockable), c_lock_w1

									# Asert like a reader
									vv_value: int = v_value

def setUp(self) -> None:
		"""Test setup."""
		self.c_rwlock_type_downgradable = (rwlock.RWLockReadD, rwlock.RWLockWriteD, rwlock.RWLockFairD)
		self.c_rwlock_type = (rwlock.RWLockRead, rwlock.RWLockWrite, rwlock.RWLockFair) + self.c_rwlock_type_downgradable

def test_write_req15(self) -> None:
		"""
		# Given: a Downgradable RW lock type to instantiate a RW lock.

		# When: A a generated writer lock is downgraed but it wasn't in a locked state.

		# Then: the generated locks raise an exception if released while being unlocked.
		"""
		# ## Arrange
		for current_rw_lock_type in self.c_rwlock_type_downgradable:
			with self.subTest(current_rw_lock_type):
				current_rw_lock = current_rw_lock_type()
				assert isinstance(current_rw_lock, rwlock.RWLockableD)
				current_lock: Union[rwlock.LockableD] = current_rw_lock.gen_wlock()

				with self.assertRaises(rwlock.RELEASE_ERR_CLS) as err:
					current_lock.release()
				self.assertEqual(str(err.exception), str(rwlock.RELEASE_ERR_MSG))

				with self.assertRaises(rwlock.RELEASE_ERR_CLS):
					# ## Assume
					self.assertFalse((current_lock.locked()))
					# ## Act
					current_lock.downgrade()

				self.assertFalse(current_lock.locked())

# Given: a Downgradable RW lock type to instantiate a RW lock.

		# When: A a generated writer lock is downgraed but it wasn't in a locked state.

		# Then: the generated locks raise an exception if released while being unlocked.
		"""
		# ## Arrange
		for current_rw_lock_type in self.c_rwlock_type_downgradable:
			with self.subTest(current_rw_lock_type):
				current_rw_lock = current_rw_lock_type()
				assert isinstance(current_rw_lock, rwlock.RWLockableD)
				current_lock: Union[rwlock.LockableD] = current_rw_lock.gen_wlock()

				with self.assertRaises(rwlock.RELEASE_ERR_CLS) as err:
					current_lock.release()
				self.assertEqual(str(err.exception), str(rwlock.RELEASE_ERR_MSG))

				with self.assertRaises(rwlock.RELEASE_ERR_CLS):
					# ## Assume
					self.assertFalse((current_lock.locked()))
					# ## Act
					current_lock.downgrade()

				self.assertFalse(current_lock.locked())

def test_read_vs_downgrade_read(self) -> None:
		"""
		# Given: Instance of RWLockReadD.

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

		def assert_internal_state():
			self.assertEqual(int(c_rwlock_1.v_read_count), int(c_rwlock_2.v_read_count))
			self.assertEqual(bool(c_rwlock_1.c_resource.locked()), bool(c_rwlock_2.c_resource.locked()))
			self.assertEqual(bool(c_rwlock_1.c_lock_read_count.locked()), bool(c_rwlock_2.c_lock_read_count.locked()))
		# ## Assume
		assert_internal_state()

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

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

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

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()

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()
		assert isinstance(a_downgrade_lock, rwlock.LockableD)
		a_downgrade_lock = a_downgrade_lock.downgrade()
		# ## Assert
		assert_internal_state()

		a_read_lock.release()
		a_downgrade_lock.release()
		assert_internal_state()

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()
		a_read_lock.acquire()
		a_downgrade_lock: Union[rwlock.LockableD, rwlock.Lockable] = c_rwlock_2.gen_wlock()
		a_downgrade_lock.acquire()
		assert isinstance(a_downgrade_lock, rwlock.LockableD)
		a_downgrade_lock = a_downgrade_lock.downgrade()
		# ## Assert
		assert_internal_state()

		a_read_lock.release()
		a_downgrade_lock.release()
		assert_internal_state()