Browse Source

Enhance omsimulator runner

Change-Id: I8adb7dae86f43aaf539648d93e27d0fae0b987ed
Ilya Shakhat 3 years ago
parent
commit
c5d91c68ab
2 changed files with 33 additions and 35 deletions
  1. 31
    35
      performa/modules/omsimulator.py
  2. 2
    0
      performa/scenarios/mq/rabbitmq.yaml

+ 31
- 35
performa/modules/omsimulator.py View File

@@ -2,7 +2,7 @@
2 2
 import copy
3 3
 import multiprocessing
4 4
 import os
5
-import signal
5
+import random
6 6
 import tempfile
7 7
 
8 8
 SERVER_PID = os.path.join(tempfile.gettempdir(), 'performa.oms.pid')
@@ -49,10 +49,15 @@ def make_client_cmd(params, i):
49 49
     params['client_file'] = make_file_name(CLIENT_FILE_NAME, i)
50 50
     params['url'] = params['client_url'] or params['url']
51 51
 
52
+    params['this_topic'] = params['topic']
53
+    if params['unique_topic_per_pair']:
54
+        params['this_topic'] += '_%s' % i
55
+
52 56
     client = ('%(python)s simulator.py '
53 57
               '--url %(url)s '
54 58
               '--json %(client_file)s '
55 59
               '-l %(duration)s '
60
+              '--topic %(this_topic)s '
56 61
               '%(client_tool)s '
57 62
               '--timeout %(timeout)s '
58 63
               '-w %(sending_delay)s '
@@ -70,60 +75,47 @@ def make_client_cmd(params, i):
70 75
 def make_server_cmd(params, i):
71 76
     params['server_file'] = make_file_name(SERVER_FILE_NAME, i)
72 77
     params['url'] = params['server_url'] or params['url']
78
+    params['server_duration'] = (params['duration'] +
79
+                                 params['server_teardown_duration'])
80
+    params['this_topic'] = params['topic']
81
+    if params['unique_topic_per_pair']:
82
+        params['this_topic'] += '_%s' % i
73 83
 
74 84
     server = ('%(python)s simulator.py '
75 85
               '--url %(url)s '
76 86
               '--json %(server_file)s '
87
+              '-l %(server_duration)s '
88
+              '--topic %(this_topic)s '
77 89
               '%(server_tool)s ') % params
78 90
 
79 91
     return server
80 92
 
81 93
 
82
-def run_client(module, command):
83
-    module.run_command(command)
84
-
85
-
86
-def run_client_pool(module, params):
94
+def run_pool(module, params):
87 95
     processes = []
88 96
 
89 97
     for i in range(params['processes']):
90 98
         cmd = make_client_cmd(params, i)
91
-        p = multiprocessing.Process(target=run_client, args=(module, cmd))
99
+        p = multiprocessing.Process(target=module.run_command, args=(cmd,))
92 100
         processes.append(p)
93 101
         p.start()
94 102
 
95
-    for p in processes:
96
-        p.join()
97
-
98
-
99
-def run_server(module, command):
100
-    module.run_command(command)
101
-
102
-
103
-def start_server_pool(module, params):
104
-    processes = []
105
-
106
-    for i in range(params['processes']):
107 103
         cmd = make_server_cmd(params, i)
108
-        p = multiprocessing.Process(target=run_client, args=(module, cmd))
104
+        p = multiprocessing.Process(target=module.run_command, args=(cmd,))
109 105
         processes.append(p)
110 106
         p.start()
111 107
 
112
-    return processes
113
-
114
-
115
-def stop_server_pool(module, processes):
116 108
     for p in processes:
117
-        rc, stdout, stderr = module.run_command('pgrep -P %s' % p.pid)
118
-
119
-        for child in (int(p) for p in stdout.split('\n') if p):
120
-            os.kill(child, signal.SIGINT)
109
+        p.join()
121 110
 
122
-    time.sleep(3)  # let simulators handle the signal
123 111
 
124
-    for p in processes:
125
-        os.kill(p.pid, signal.SIGINT)
126
-        p.join()
112
+def cleanup_old_files(module, params):
113
+    for i in range(params['processes']):
114
+        try:
115
+            os.unlink(make_file_name(CLIENT_FILE_NAME, i))
116
+            os.unlink(make_file_name(SERVER_FILE_NAME, i))
117
+        except OSError:
118
+            pass  # ignore
127 119
 
128 120
 
129 121
 def collect_results(params):
@@ -167,11 +159,12 @@ def run(module):
167 159
     params['server_tool'] = server_tool
168 160
     params['client_tool'] = client_tool
169 161
 
170
-    server_processes = start_server_pool(module, params)
162
+    if not params['topic']:
163
+        params['topic'] = 'performa_%s' % random.random()
171 164
 
172
-    run_client_pool(module, params)
165
+    cleanup_old_files(module, params)
173 166
 
174
-    stop_server_pool(module, server_processes)
167
+    run_pool(module, params)
175 168
 
176 169
     try:
177 170
         result = collect_results(params)
@@ -194,6 +187,9 @@ def main():
194 187
             duration=dict(type='int', default=10),
195 188
             timeout=dict(type='int', default=5),
196 189
             sending_delay=dict(type='float', default=-1.0),
190
+            topic=dict(),
191
+            unique_topic_per_pair=dict(type='bool', default=False),
192
+            server_teardown_duration=dict(type='int', default=15),
197 193
         ))
198 194
 
199 195
     chdir(module)

+ 2
- 0
performa/scenarios/mq/rabbitmq.yaml View File

@@ -58,6 +58,7 @@ execution:
58 58
         url: {{ rabbit_url }}
59 59
         client_url: {{ client_url }}
60 60
         server_url: {{ server_url }}
61
+        topic: profiler_topic
61 62
   -
62 63
     hosts: {{ tester_hosts }}
63 64
     matrix:
@@ -70,6 +71,7 @@ execution:
70 71
         url: {{ rabbit_url }}
71 72
         client_url: {{ client_url }}
72 73
         server_url: {{ server_url }}
74
+        topic: profiler_topic
73 75
   -
74 76
     hosts: {{ tester_hosts }}
75 77
     matrix:

Loading…
Cancel
Save