# -*- 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))