+# -*- coding: utf-8 -*-
+
+# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+import collections
+import random
+import threading
+import time
+
+from concurrent import futures
+
+import fasteners
+from fasteners import test
+
+from fasteners import _utils
+
+
+# NOTE(harlowja): Sleep a little so now() can not be the same (which will
+# cause false positives when our overlap detection code runs). If there are
+# real overlaps then they will still exist.
+NAPPY_TIME = 0.05
+
+# We will spend this amount of time doing some "fake" work.
+WORK_TIMES = [(0.01 + x / 100.0) for x in range(0, 5)]
+
+# If latches/events take longer than this to become empty/set, something is
+# usually wrong and should be debugged instead of deadlocking...
+WAIT_TIMEOUT = 300
+
+
+def _find_overlaps(times, start, end):
+ overlaps = 0
+ for (s, e) in times:
+ if s >= start and e <= end:
+ overlaps += 1
+ return overlaps
+
+
+def _spawn_variation(readers, writers, max_workers=None):
+ start_stops = collections.deque()
+ lock = fasteners.ReaderWriterLock()
+
+ def read_func(ident):
+ with lock.read_lock():
+ # TODO(harlowja): sometime in the future use a monotonic clock here
+ # to avoid problems that can be caused by ntpd resyncing the clock
+ # while we are actively running.
+ enter_time = _utils.now()
+ time.sleep(WORK_TIMES[ident % len(WORK_TIMES)])
+ exit_time = _utils.now()
+ start_stops.append((lock.READER, enter_time, exit_time))
+ time.sleep(NAPPY_TIME)
+
+ def write_func(ident):
+ with lock.write_lock():
+ enter_time = _utils.now()
+ time.sleep(WORK_TIMES[ident % len(WORK_TIMES)])
+ exit_time = _utils.now()
+ start_stops.append((lock.WRITER, enter_time, exit_time))
+ time.sleep(NAPPY_TIME)
+
+ if max_workers is None:
+ max_workers = max(0, readers) + max(0, writers)
+ if max_workers > 0:
+ with futures.ThreadPoolExecutor(max_workers=max_workers) as e:
+ count = 0
+ for _i in range(0, readers):
+ e.submit(read_func, count)
+ count += 1
+ for _i in range(0, writers):
+ e.submit(write_func, count)
+ count += 1
+
+ writer_times = []
+ reader_times = []
+ for (lock_type, start, stop) in list(start_stops):
+ if lock_type == lock.WRITER:
+ writer_times.append((start, stop))
+ else:
+ reader_times.append((start, stop))
+ return (writer_times, reader_times)
+
+
+def _daemon_thread(target):
+ t = threading.Thread(target=target)
+ t.daemon = True
+ return t
+
+
+class ReadWriteLockTest(test.TestCase):
+ THREAD_COUNT = 20
+
+ def test_no_double_writers(self):
+ lock = fasteners.ReaderWriterLock()
+ watch = _utils.StopWatch(duration=5)
+ watch.start()
+ dups = collections.deque()
+ active = collections.deque()
+
+ def acquire_check(me):
+ with lock.write_lock():
+ if len(active) >= 1:
+ dups.append(me)
+ dups.extend(active)
+ active.append(me)
+ try:
+ time.sleep(random.random() / 100)
+ finally:
+ active.remove(me)
+
+ def run():
+ me = threading.current_thread()
+ while not watch.expired():
+ acquire_check(me)
+
+ threads = []
+ for i in range(0, self.THREAD_COUNT):
+ t = _daemon_thread(run)
+ threads.append(t)
+ t.start()
+ while threads:
+ t = threads.pop()
+ t.join()
+
+ self.assertEqual([], list(dups))
+ self.assertEqual([], list(active))
+
+ def test_no_concurrent_readers_writers(self):
+ lock = fasteners.ReaderWriterLock()
+ watch = _utils.StopWatch(duration=5)
+ watch.start()
+ dups = collections.deque()
+ active = collections.deque()
+
+ def acquire_check(me, reader):
+ if reader:
+ lock_func = lock.read_lock
+ else:
+ lock_func = lock.write_lock
+ with lock_func():
+ if not reader:
+ # There should be no-one else currently active, if there
+ # is ensure we capture them so that we can later blow-up
+ # the test.
+ if len(active) >= 1:
+ dups.append(me)
+ dups.extend(active)
+ active.append(me)
+ try:
+ time.sleep(random.random() / 100)
+ finally:
+ active.remove(me)
+
+ def run():
+ me = threading.current_thread()
+ while not watch.expired():
+ acquire_check(me, random.choice([True, False]))
+
+ threads = []
+ for i in range(0, self.THREAD_COUNT):
+ t = _daemon_thread(run)
+ threads.append(t)
+ t.start()
+ while threads:
+ t = threads.pop()
+ t.join()
+
+ self.assertEqual([], list(dups))
+ self.assertEqual([], list(active))
+
+ def test_writer_abort(self):
+ lock = fasteners.ReaderWriterLock()
+ self.assertFalse(lock.owner)
+
+ def blow_up():
+ with lock.write_lock():
+ self.assertEqual(lock.WRITER, lock.owner)
+ raise RuntimeError("Broken")
+
+ self.assertRaises(RuntimeError, blow_up)
+ self.assertFalse(lock.owner)
+
+ def test_reader_abort(self):
+ lock = fasteners.ReaderWriterLock()
+ self.assertFalse(lock.owner)
+
+ def blow_up():
+ with lock.read_lock():
+ self.assertEqual(lock.READER, lock.owner)
+ raise RuntimeError("Broken")
+
+ self.assertRaises(RuntimeError, blow_up)
+ self.assertFalse(lock.owner)
+
+ def test_double_reader_abort(self):
+ lock = fasteners.ReaderWriterLock()
+ activated = collections.deque()
+
+ def double_bad_reader():
+ with lock.read_lock():
+ with lock.read_lock():
+ raise RuntimeError("Broken")
+
+ def happy_writer():
+ with lock.write_lock():
+ activated.append(lock.owner)
+
+ with futures.ThreadPoolExecutor(max_workers=20) as e:
+ for i in range(0, 20):
+ if i % 2 == 0:
+ e.submit(double_bad_reader)
+ else:
+ e.submit(happy_writer)
+
+ self.assertEqual(10, len([a for a in activated if a == 'w']))
+
+ def test_double_reader_writer(self):
+ lock = fasteners.ReaderWriterLock()
+ activated = collections.deque()
+ active = threading.Event()
+
+ def double_reader():
+ with lock.read_lock():
+ active.set()
+ while not lock.has_pending_writers:
+ time.sleep(0.001)
+ with lock.read_lock():
+ activated.append(lock.owner)
+
+ def happy_writer():
+ with lock.write_lock():
+ activated.append(lock.owner)
+
+ reader = _daemon_thread(double_reader)
+ reader.start()
+ active.wait(WAIT_TIMEOUT)
+ self.assertTrue(active.is_set())
+
+ writer = _daemon_thread(happy_writer)
+ writer.start()
+
+ reader.join()
+ writer.join()
+ self.assertEqual(2, len(activated))
+ self.assertEqual(['r', 'w'], list(activated))
+
+ def test_reader_chaotic(self):
+ lock = fasteners.ReaderWriterLock()
+ activated = collections.deque()
+
+ def chaotic_reader(blow_up):
+ with lock.read_lock():
+ if blow_up:
+ raise RuntimeError("Broken")
+ else:
+ activated.append(lock.owner)
+
+ def happy_writer():
+ with lock.write_lock():
+ activated.append(lock.owner)
+
+ with futures.ThreadPoolExecutor(max_workers=20) as e:
+ for i in range(0, 20):
+ if i % 2 == 0:
+ e.submit(chaotic_reader, blow_up=bool(i % 4 == 0))
+ else:
+ e.submit(happy_writer)
+
+ writers = [a for a in activated if a == 'w']
+ readers = [a for a in activated if a == 'r']
+ self.assertEqual(10, len(writers))
+ self.assertEqual(5, len(readers))
+
+ def test_writer_chaotic(self):
+ lock = fasteners.ReaderWriterLock()
+ activated = collections.deque()
+
+ def chaotic_writer(blow_up):
+ with lock.write_lock():
+ if blow_up:
+ raise RuntimeError("Broken")
+ else:
+ activated.append(lock.owner)
+
+ def happy_reader():
+ with lock.read_lock():
+ activated.append(lock.owner)
+
+ with futures.ThreadPoolExecutor(max_workers=20) as e:
+ for i in range(0, 20):
+ if i % 2 == 0:
+ e.submit(chaotic_writer, blow_up=bool(i % 4 == 0))
+ else:
+ e.submit(happy_reader)
+
+ writers = [a for a in activated if a == 'w']
+ readers = [a for a in activated if a == 'r']
+ self.assertEqual(5, len(writers))
+ self.assertEqual(10, len(readers))
+
+ def test_writer_reader_writer(self):
+ lock = fasteners.ReaderWriterLock()
+ with lock.write_lock():
+ self.assertTrue(lock.is_writer())
+ with lock.read_lock():
+ self.assertTrue(lock.is_reader())
+ with lock.write_lock():
+ self.assertTrue(lock.is_writer())
+
+ def test_single_reader_writer(self):
+ results = []
+ lock = fasteners.ReaderWriterLock()
+ with lock.read_lock():
+ self.assertTrue(lock.is_reader())
+ self.assertEqual(0, len(results))
+ with lock.write_lock():
+ results.append(1)
+ self.assertTrue(lock.is_writer())
+ with lock.read_lock():
+ self.assertTrue(lock.is_reader())
+ self.assertEqual(1, len(results))
+ self.assertFalse(lock.is_reader())
+ self.assertFalse(lock.is_writer())
+
+ def test_reader_to_writer(self):
+ lock = fasteners.ReaderWriterLock()
+
+ def writer_func():
+ with lock.write_lock():
+ pass
+
+ with lock.read_lock():
+ self.assertRaises(RuntimeError, writer_func)
+ self.assertFalse(lock.is_writer())
+
+ self.assertFalse(lock.is_reader())
+ self.assertFalse(lock.is_writer())
+
+ def test_writer_to_reader(self):
+ lock = fasteners.ReaderWriterLock()
+
+ def reader_func():
+ with lock.read_lock():
+ self.assertTrue(lock.is_writer())
+ self.assertTrue(lock.is_reader())
+
+ with lock.write_lock():
+ self.assertIsNone(reader_func())
+ self.assertFalse(lock.is_reader())
+
+ self.assertFalse(lock.is_reader())
+ self.assertFalse(lock.is_writer())
+
+ def test_double_writer(self):
+ lock = fasteners.ReaderWriterLock()
+ with lock.write_lock():
+ self.assertFalse(lock.is_reader())
+ self.assertTrue(lock.is_writer())
+ with lock.write_lock():
+ self.assertTrue(lock.is_writer())
+ self.assertTrue(lock.is_writer())
+
+ self.assertFalse(lock.is_reader())
+ self.assertFalse(lock.is_writer())
+
+ def test_double_reader(self):
+ lock = fasteners.ReaderWriterLock()
+ with lock.read_lock():
+ self.assertTrue(lock.is_reader())
+ self.assertFalse(lock.is_writer())
+ with lock.read_lock():
+ self.assertTrue(lock.is_reader())
+ self.assertTrue(lock.is_reader())
+
+ self.assertFalse(lock.is_reader())
+ self.assertFalse(lock.is_writer())
+
+ def test_multi_reader_multi_writer(self):
+ writer_times, reader_times = _spawn_variation(10, 10)
+ self.assertEqual(10, len(writer_times))
+ self.assertEqual(10, len(reader_times))
+ for (start, stop) in writer_times:
+ self.assertEqual(0, _find_overlaps(reader_times, start, stop))
+ self.assertEqual(1, _find_overlaps(writer_times, start, stop))
+ for (start, stop) in reader_times:
+ self.assertEqual(0, _find_overlaps(writer_times, start, stop))
+
+ def test_multi_reader_single_writer(self):
+ writer_times, reader_times = _spawn_variation(9, 1)
+ self.assertEqual(1, len(writer_times))
+ self.assertEqual(9, len(reader_times))
+ start, stop = writer_times[0]
+ self.assertEqual(0, _find_overlaps(reader_times, start, stop))
+
+ def test_multi_writer(self):
+ writer_times, reader_times = _spawn_variation(0, 10)
+ self.assertEqual(10, len(writer_times))
+ self.assertEqual(0, len(reader_times))
+ for (start, stop) in writer_times:
+ self.assertEqual(1, _find_overlaps(writer_times, start, stop))
projects / tools / fasteners.git / commitdiff
