summaryrefslogtreecommitdiffstats
path: root/Demo/metaclasses/Synch.py
blob: 1fb91602df74c95dcf394335c5d3debb24f970e3 (plain)
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
43
44
45
46
47
48
49
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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
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
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
"""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):
	"""Constructor.  Initialize all instance variables."""
	self.__mutex = thread.allocate_lock()
	self.__wait = thread.allocate_lock()
	self.__tid = None
	self.__count = 0

    def acquire(self, flag=1):
	"""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()

    def release(self):
	"""Release the lock.

	If this thread doesn't currently have the lock, an assertion
	error is raised.

	Only allow another thread to acquire the lock when the count
	reaches zero after decrementing it.

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


def _testLock():

    done = []

    def f2(lock, done=done):
	lock.acquire()
	print "f2 running in thread %d\n" % thread.get_ident(),
	lock.release()
	done.append(1)

    def f1(lock, f2=f2, done=done):
	lock.acquire()
	print "f1 running in thread %d\n" % thread.get_ident(),
	try:
	    f2(lock)
	finally:
	    lock.release()
	done.append(1)

    lock = Lock()
    lock.acquire()
    f1(lock)				# Adds 2 to done
    lock.release()

    lock.acquire()
    
    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:
	print len(done)
	time.sleep(0.001)
    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):
	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()

class LockingHelper(MetaHelper):
    __methodwrapper__ = LockingMethodWrapper
    def __helperinit__(self, formalclass):
	MetaHelper.__helperinit__(self, formalclass)
	self.__lock__ = Lock()

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

    def producer(buffer, wait, n=1000):
	import time
	i = 0
	while i < n:
	    print "put", i
	    buffer.put(i)
	    i = i+1
	print "Producer: done producing", n, "items"
	wait.release()

    def consumer(buffer, wait, n=1000):
	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()

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