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
|
# -*- coding: utf-8 -*-
#
# Copyright (c) ZeroC, Inc. All rights reserved.
#
#
# Make sure IceStorm and the subscriber use the same buffer size for
# sending/receiving datagrams. This ensures the test works with bogus
# OS configurations where the reicever buffer size is smaller than the
# send buffer size (causing the received messages to be
# truncated). See also bug #6070.
#
props = {
"IceStorm.Election.MasterTimeout" : 2,
"IceStorm.Election.ElectionTimeout" : 2,
"IceStorm.Election.ResponseTimeout" : 2,
"Ice.Warn.Dispatch" : 0
}
icestorm = [ IceStorm(replica=i, nreplicas=3, props = props) for i in range(0,3) ]
class IceStormRepStressTestCase(IceStormTestCase):
def runClientSide(self, current):
def stopReplica(num):
self.icestorm[num].shutdown(current)
self.icestorm[num].stop(current, True)
def startReplica(num):
self.icestorm[num].start(current)
current.write("creating topic... ")
self.runadmin(current, "create single")
current.writeln("ok")
current.write("running subscriber... ")
subscriber = Subscriber(quiet=True)
subscriber.start(current)
subscriber.expect(current, "([^\n]+)\n")
subControl = subscriber.getMatch(current).group(1)
current.writeln("ok")
current.write("running publisher... ")
publisher = Publisher(quiet=True)
publisher.start(current)
publisher.expect(current, "([^\n]+)\n")
pubControl = publisher.getMatch(current).group(1)
current.writeln("ok")
time.sleep(2)
for i in range(0, 3):
# 0, 1
current.write("stopping replica 2 (0, 1 running)... ")
stopReplica(2)
current.writeln("ok")
time.sleep(2)
# 1, 2
current.write("starting 2, stopping 0 (1, 2 running)... ")
startReplica(2)
stopReplica(0)
current.writeln("ok")
# This waits for the replication to startup
#self.runadmin(current, "list")
time.sleep(2)
# 0, 2
current.write("starting 0, stopping 1 (0, 2 running)... ")
startReplica(0)
stopReplica(1)
current.writeln("ok")
# This waits for the replication to startup
#self.runadmin(current, "list")
time.sleep(2)
current.write("starting 1 (all running)... ")
startReplica(1)
current.writeln("ok")
# This waits for the replication to startup
#self.runadmin(current, "list")
time.sleep(2)
current.write("stopping publisher... ")
Client(exe="control", args=[pubControl]).run(current)
publisher.expect(current, "([^\n]+)\n")
publisherCount = publisher.getMatch(current).group(1)
publisher.stop(current, True)
current.writeln("ok")
current.write("stopping replicas... ")
self.stopIceStorm(current)
current.writeln("ok")
current.write("stopping subscriber... ")
Client(exe="control", args=[subControl]).run(current)
subscriber.expect(current, "([^\n]+)\n")
subscriberCount = subscriber.getMatch(current).group(1)
subscriber.stop(current, True)
current.writeln("ok")
current.writeln("publisher published %s events, subscriber received %s events" % (publisherCount, subscriberCount))
TestSuite(__file__,
[ IceStormRepStressTestCase("replicated", icestorm=icestorm) ],
options={ "ipv6" : [False] },
multihost=False, runOnMainThread=True)
|
