Synch.py 7.75 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42
"""Synchronization metaclass.

This metaclass  makes it possible to declare synchronized methods.

"""

import thread

# First we need to define a reentrant lock.
# This is generally useful and should probably be in a standard Python
# library module.  For now, we in-line it.

class Lock:

    """Reentrant lock.

    This is a mutex-like object which can be acquired by the same
    thread more than once.  It keeps a reference count of the number
    of times it has been acquired by the same thread.  Each acquire()
    call must be matched by a release() call and only the last
    release() call actually releases the lock for acquisition by
    another thread.

    The implementation uses two locks internally:

    __mutex is a short term lock used to protect the instance variables
    __wait is the lock for which other threads wait

    A thread intending to acquire both locks should acquire __wait
    first.

   The implementation uses two other instance variables, protected by
   locking __mutex:

    __tid is the thread ID of the thread that currently has the lock
    __count is the number of times the current thread has acquired it

    When the lock is released, __tid is None and __count is zero.

    """

    def __init__(self):
Guido van Rossum's avatar
Guido van Rossum committed
43 44 45 46 47
        """Constructor.  Initialize all instance variables."""
        self.__mutex = thread.allocate_lock()
        self.__wait = thread.allocate_lock()
        self.__tid = None
        self.__count = 0
48 49

    def acquire(self, flag=1):
Guido van Rossum's avatar
Guido van Rossum committed
50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79
        """Acquire the lock.

        If the optional flag argument is false, returns immediately
        when it cannot acquire the __wait lock without blocking (it
        may still block for a little while in order to acquire the
        __mutex lock).

        The return value is only relevant when the flag argument is
        false; it is 1 if the lock is acquired, 0 if not.

        """
        self.__mutex.acquire()
        try:
            if self.__tid == thread.get_ident():
                self.__count = self.__count + 1
                return 1
        finally:
            self.__mutex.release()
        locked = self.__wait.acquire(flag)
        if not flag and not locked:
            return 0
        try:
            self.__mutex.acquire()
            assert self.__tid == None
            assert self.__count == 0
            self.__tid = thread.get_ident()
            self.__count = 1
            return 1
        finally:
            self.__mutex.release()
80 81

    def release(self):
Guido van Rossum's avatar
Guido van Rossum committed
82
        """Release the lock.
83

Guido van Rossum's avatar
Guido van Rossum committed
84 85
        If this thread doesn't currently have the lock, an assertion
        error is raised.
86

Guido van Rossum's avatar
Guido van Rossum committed
87 88
        Only allow another thread to acquire the lock when the count
        reaches zero after decrementing it.
89

Guido van Rossum's avatar
Guido van Rossum committed
90 91 92 93 94 95 96 97 98 99 100
        """
        self.__mutex.acquire()
        try:
            assert self.__tid == thread.get_ident()
            assert self.__count > 0
            self.__count = self.__count - 1
            if self.__count == 0:
                self.__tid = None
                self.__wait.release()
        finally:
            self.__mutex.release()
101 102 103 104 105 106 107


def _testLock():

    done = []

    def f2(lock, done=done):
Guido van Rossum's avatar
Guido van Rossum committed
108 109 110 111
        lock.acquire()
        print "f2 running in thread %d\n" % thread.get_ident(),
        lock.release()
        done.append(1)
112 113

    def f1(lock, f2=f2, done=done):
Guido van Rossum's avatar
Guido van Rossum committed
114 115 116 117 118 119 120
        lock.acquire()
        print "f1 running in thread %d\n" % thread.get_ident(),
        try:
            f2(lock)
        finally:
            lock.release()
        done.append(1)
121 122 123

    lock = Lock()
    lock.acquire()
Guido van Rossum's avatar
Guido van Rossum committed
124
    f1(lock)                            # Adds 2 to done
125 126 127
    lock.release()

    lock.acquire()
128

129 130 131 132 133 134 135 136
    thread.start_new_thread(f1, (lock,)) # Adds 2
    thread.start_new_thread(f1, (lock, f1)) # Adds 3
    thread.start_new_thread(f2, (lock,)) # Adds 1
    thread.start_new_thread(f2, (lock,)) # Adds 1

    lock.release()
    import time
    while len(done) < 9:
Guido van Rossum's avatar
Guido van Rossum committed
137 138
        print len(done)
        time.sleep(0.001)
139 140 141 142 143 144 145 146 147 148 149
    print len(done)


# Now, the Locking metaclass is a piece of cake.
# As an example feature, methods whose name begins with exactly one
# underscore are not synchronized.

from Meta import MetaClass, MetaHelper, MetaMethodWrapper

class LockingMethodWrapper(MetaMethodWrapper):
    def __call__(self, *args, **kw):
Guido van Rossum's avatar
Guido van Rossum committed
150 151 152 153 154 155 156
        if self.__name__[:1] == '_' and self.__name__[1:] != '_':
            return apply(self.func, (self.inst,) + args, kw)
        self.inst.__lock__.acquire()
        try:
            return apply(self.func, (self.inst,) + args, kw)
        finally:
            self.inst.__lock__.release()
157 158 159 160

class LockingHelper(MetaHelper):
    __methodwrapper__ = LockingMethodWrapper
    def __helperinit__(self, formalclass):
Guido van Rossum's avatar
Guido van Rossum committed
161 162
        MetaHelper.__helperinit__(self, formalclass)
        self.__lock__ = Lock()
163 164 165 166 167 168 169 170 171

class LockingMetaClass(MetaClass):
    __helper__ = LockingHelper

Locking = LockingMetaClass('Locking', (), {})

def _test():
    # For kicks, take away the Locking base class and see it die
    class Buffer(Locking):
Guido van Rossum's avatar
Guido van Rossum committed
172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209
        def __init__(self, initialsize):
            assert initialsize > 0
            self.size = initialsize
            self.buffer = [None]*self.size
            self.first = self.last = 0
        def put(self, item):
            # Do we need to grow the buffer?
            if (self.last+1) % self.size != self.first:
                # Insert the new item
                self.buffer[self.last] = item
                self.last = (self.last+1) % self.size
                return
            # Double the buffer size
            # First normalize it so that first==0 and last==size-1
            print "buffer =", self.buffer
            print "first = %d, last = %d, size = %d" % (
                self.first, self.last, self.size)
            if self.first <= self.last:
                temp = self.buffer[self.first:self.last]
            else:
                temp = self.buffer[self.first:] + self.buffer[:self.last]
            print "temp =", temp
            self.buffer = temp + [None]*(self.size+1)
            self.first = 0
            self.last = self.size-1
            self.size = self.size*2
            print "Buffer size doubled to", self.size
            print "new buffer =", self.buffer
            print "first = %d, last = %d, size = %d" % (
                self.first, self.last, self.size)
            self.put(item)              # Recursive call to test the locking
        def get(self):
            # Is the buffer empty?
            if self.first == self.last:
                raise EOFError          # Avoid defining a new exception
            item = self.buffer[self.first]
            self.first = (self.first+1) % self.size
            return item
210 211

    def producer(buffer, wait, n=1000):
Guido van Rossum's avatar
Guido van Rossum committed
212 213 214 215 216 217 218 219
        import time
        i = 0
        while i < n:
            print "put", i
            buffer.put(i)
            i = i+1
        print "Producer: done producing", n, "items"
        wait.release()
220 221

    def consumer(buffer, wait, n=1000):
Guido van Rossum's avatar
Guido van Rossum committed
222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238
        import time
        i = 0
        tout = 0.001
        while i < n:
            try:
                x = buffer.get()
                if x != i:
                    raise AssertionError, \
                          "get() returned %s, expected %s" % (x, i)
                print "got", i
                i = i+1
                tout = 0.001
            except EOFError:
                time.sleep(tout)
                tout = tout*2
        print "Consumer: done consuming", n, "items"
        wait.release()
239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256

    pwait = thread.allocate_lock()
    pwait.acquire()
    cwait = thread.allocate_lock()
    cwait.acquire()
    buffer = Buffer(1)
    n = 1000
    thread.start_new_thread(consumer, (buffer, cwait, n))
    thread.start_new_thread(producer, (buffer, pwait, n))
    pwait.acquire()
    print "Producer done"
    cwait.acquire()
    print "All done"
    print "buffer size ==", len(buffer.buffer)

if __name__ == '__main__':
    _testLock()
    _test()