Remove the mutex module.

This commit is contained in:
Brett Cannon 2008-05-08 18:55:02 +00:00
parent 39143f8156
commit d7cce26f7c
5 changed files with 1 additions and 150 deletions

View File

@ -26,7 +26,6 @@ The following modules are documented in this chapter:
bisect.rst
array.rst
sched.rst
mutex.rst
queue.rst
weakref.rst
types.rst

View File

@ -1,63 +0,0 @@
:mod:`mutex` --- Mutual exclusion support
=========================================
.. module:: mutex
:synopsis: Lock and queue for mutual exclusion.
.. sectionauthor:: Moshe Zadka <moshez@zadka.site.co.il>
The :mod:`mutex` module defines a class that allows mutual-exclusion via
acquiring and releasing locks. It does not require (or imply)
:mod:`threading` or multi-tasking, though it could be useful for those
purposes.
The :mod:`mutex` module defines the following class:
.. class:: mutex()
Create a new (unlocked) mutex.
A mutex has two pieces of state --- a "locked" bit and a queue. When the mutex
is not locked, the queue is empty. Otherwise, the queue contains zero or more
``(function, argument)`` pairs representing functions (or methods) waiting to
acquire the lock. When the mutex is unlocked while the queue is not empty, the
first queue entry is removed and its ``function(argument)`` pair called,
implying it now has the lock.
Of course, no multi-threading is implied -- hence the funny interface for
:meth:`lock`, where a function is called once the lock is acquired.
.. _mutex-objects:
Mutex Objects
-------------
:class:`mutex` objects have following methods:
.. method:: mutex.test()
Check whether the mutex is locked.
.. method:: mutex.testandset()
"Atomic" test-and-set, grab the lock if it is not set, and return ``True``,
otherwise, return ``False``.
.. method:: mutex.lock(function, argument)
Execute ``function(argument)``, unless the mutex is locked. In the case it is
locked, place the function and argument on the queue. See :meth:`unlock` for
explanation of when ``function(argument)`` is executed in that case.
.. method:: mutex.unlock()
Unlock the mutex if queue is empty, otherwise execute the first element in the
queue.

View File

@ -8,7 +8,7 @@
This module constructs higher-level threading interfaces on top of the lower
level :mod:`thread` module.
See also the :mod:`mutex` and :mod:`Queue` modules.
See also the :mod:`Queue` module.
The :mod:`dummy_threading` module is provided for situations where
:mod:`threading` cannot be used because :mod:`thread` is missing.

View File

@ -1,52 +0,0 @@
"""Mutual exclusion -- for use with module sched
A mutex has two pieces of state -- a 'locked' bit and a queue.
When the mutex is not locked, the queue is empty.
Otherwise, the queue contains 0 or more (function, argument) pairs
representing functions (or methods) waiting to acquire the lock.
When the mutex is unlocked while the queue is not empty,
the first queue entry is removed and its function(argument) pair called,
implying it now has the lock.
Of course, no multi-threading is implied -- hence the funny interface
for lock, where a function is called once the lock is aquired.
"""
from collections import deque
class mutex:
def __init__(self):
"""Create a new mutex -- initially unlocked."""
self.locked = 0
self.queue = deque()
def test(self):
"""Test the locked bit of the mutex."""
return self.locked
def testandset(self):
"""Atomic test-and-set -- grab the lock if it is not set,
return True if it succeeded."""
if not self.locked:
self.locked = 1
return True
else:
return False
def lock(self, function, argument):
"""Lock a mutex, call the function with supplied argument
when it is acquired. If the mutex is already locked, place
function and argument in the queue."""
if self.testandset():
function(argument)
else:
self.queue.append((function, argument))
def unlock(self):
"""Unlock a mutex. If the queue is not empty, call the next
function with its argument."""
if self.queue:
function, argument = self.queue.popleft()
function(argument)
else:
self.locked = 0

View File

@ -1,33 +0,0 @@
import mutex
import unittest
import test.test_support
class MutexTest(unittest.TestCase):
def setUp(self):
self.mutex = mutex.mutex()
def called_by_mutex(self, some_data):
self.assert_(self.mutex.test(), "mutex not held")
# Nested locking
self.mutex.lock(self.called_by_mutex2, "eggs")
def called_by_mutex2(self, some_data):
self.assert_(self.ready_for_2,
"called_by_mutex2 called too soon")
def test_lock_and_unlock(self):
self.read_for_2 = False
self.mutex.lock(self.called_by_mutex, "spam")
self.ready_for_2 = True
# unlock both locks
self.mutex.unlock()
self.mutex.unlock()
self.failIf(self.mutex.test(), "mutex still held")
def test_main():
test.test_support.run_unittest(MutexTest)
if __name__ == "__main__":
test_main()