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
|
# Python modules
import hashlib
# 3rd party modules
import sysv_ipc
# Utils for this demo
import utils
utils.say("Oooo 'ello, I'm Mrs. Conclusion!")
params = utils.read_params()
semaphore = sysv_ipc.Semaphore(params["KEY"])
memory = sysv_ipc.SharedMemory(params["KEY"])
utils.say("memory attached at %d" % memory.address)
what_i_wrote = ""
s = ""
for i in range(0, params["ITERATIONS"]):
utils.say("i = %d" % i)
if not params["LIVE_DANGEROUSLY"]:
# Wait for Mrs. Premise to free up the semaphore.
utils.say("acquiring the semaphore...")
semaphore.acquire()
s = utils.read_from_memory(memory)
while s == what_i_wrote:
if not params["LIVE_DANGEROUSLY"]:
# Release the semaphore...
utils.say("releasing the semaphore")
semaphore.release()
# ...and wait for it to become available again.
utils.say("acquiring for the semaphore...")
semaphore.acquire()
s = utils.read_from_memory(memory)
if what_i_wrote:
what_i_wrote = what_i_wrote.encode()
try:
assert(s == hashlib.md5(what_i_wrote).hexdigest())
except AssertionError:
raise AssertionError("Shared memory corruption after %d iterations." % i)
s = s.encode()
what_i_wrote = hashlib.md5(s).hexdigest()
utils.write_to_memory(memory, what_i_wrote)
if not params["LIVE_DANGEROUSLY"]:
utils.say("releasing the semaphore")
semaphore.release()
|
