|
592
|
1 import unittest
|
|
|
2 import time
|
|
|
3 import os
|
|
|
4 import threading
|
|
|
5 from helpers import OrthancTestCase, Helpers
|
|
|
6
|
|
598
|
7 from orthanc_api_client import OrthancApiClient, ChangeType
|
|
|
8 from orthanc_api_client import helpers as OrthancHelpers
|
|
|
9
|
|
592
|
10 from orthanc_tools import OrthancTestDbPopulator
|
|
|
11
|
|
|
12 import pathlib
|
|
|
13 import subprocess
|
|
|
14 import glob
|
|
|
15 here = pathlib.Path(__file__).parent.resolve()
|
|
|
16
|
|
|
17
|
|
|
18 def worker_upload_folder(orthanc_root_url: str, folder: str, repeat: int, worker_id: int):
|
|
|
19 o = OrthancApiClient(orthanc_root_url)
|
|
|
20 for i in range(0, repeat):
|
|
|
21 o.upload_folder(folder, ignore_errors=True)
|
|
|
22
|
|
|
23 def worker_anonymize_study(orthanc_root_url: str, study_id: str, repeat: int, worker_id: int):
|
|
|
24 o = OrthancApiClient(orthanc_root_url)
|
|
|
25
|
|
|
26 for i in range(0, repeat):
|
|
|
27 o.studies.anonymize(orthanc_id=study_id, delete_original=False)
|
|
|
28
|
|
597
|
29 def count_changes(changes, type: ChangeType):
|
|
|
30 return len([c.change_type for c in changes if c.change_type==type])
|
|
|
31
|
|
592
|
32 def worker_upload_delete_study_part(orthanc_root_url: str, folder: str, repeat: int, workers_count: int, worker_id: int):
|
|
|
33 o = OrthancApiClient(orthanc_root_url)
|
|
|
34
|
|
|
35 all_files = glob.glob(os.path.join(folder, '*.dcm'))
|
|
|
36
|
|
|
37 for i in range(0, repeat):
|
|
|
38 instances_ids = []
|
|
|
39
|
|
|
40 for i in range(0, len(all_files)):
|
|
598
|
41 if i % workers_count == worker_id: # each thread takes a part
|
|
592
|
42 instances_ids.extend(o.upload_file(all_files[i]))
|
|
|
43
|
|
|
44 for instance_id in instances_ids:
|
|
594
|
45 o.instances.delete(orthanc_id=instance_id, ignore_errors=True)
|
|
592
|
46
|
|
|
47
|
|
598
|
48 def worker_upload_delete_test_dicoms(orthanc_root_url: str, files_count: int, worker_id: int):
|
|
|
49 o = OrthancApiClient(orthanc_root_url)
|
|
|
50
|
|
|
51 instances_ids = []
|
|
|
52 counter = 0
|
|
|
53
|
|
|
54 for i in range(0, files_count):
|
|
|
55 counter += 1
|
|
|
56 dicom_file = OrthancHelpers.generate_test_dicom_file(width=4, height=4,
|
|
|
57 tags = {
|
|
|
58 "PatientID" : f"{worker_id}",
|
|
|
59 "StudyInstanceUID" : f"{worker_id}",
|
|
|
60 "SeriesInstanceUID" : f"{worker_id}.{counter%10}"
|
|
|
61 })
|
|
|
62 instances_ids.extend(o.upload(dicom_file))
|
|
|
63
|
|
|
64 study_id = o.instances.get_parent_study_id(instances_ids[0])
|
|
|
65 o.studies.delete(orthanc_id=study_id)
|
|
|
66
|
|
|
67
|
|
592
|
68 class TestConcurrency(OrthancTestCase):
|
|
|
69
|
|
|
70 @classmethod
|
|
|
71 def terminate(cls):
|
|
|
72
|
|
|
73 if Helpers.is_docker():
|
|
|
74 subprocess.run(["docker", "rm", "-f", "pg-server"])
|
|
|
75 else:
|
|
|
76 cls.pg_service_process.terminate()
|
|
|
77
|
|
|
78
|
|
|
79 @classmethod
|
|
|
80 def prepare(cls):
|
|
|
81 test_name = "Concurrency"
|
|
|
82 cls._storage_name = "concurrency"
|
|
|
83
|
|
|
84 print(f'-------------- preparing {test_name} tests')
|
|
|
85
|
|
|
86 cls.clear_storage(storage_name=cls._storage_name)
|
|
|
87
|
|
|
88 pg_hostname = "localhost"
|
|
|
89 if Helpers.is_docker():
|
|
|
90 pg_hostname = "pg-server"
|
|
|
91 cls.create_docker_network("concurrency")
|
|
|
92
|
|
|
93 config = {
|
|
|
94 "PostgreSQL" : {
|
|
|
95 "EnableStorage": False,
|
|
|
96 "EnableIndex": True,
|
|
|
97 "Host": pg_hostname,
|
|
|
98 "Port": 5432,
|
|
|
99 "Database": "postgres",
|
|
|
100 "Username": "postgres",
|
|
|
101 "Password": "postgres",
|
|
|
102 "IndexConnectionsCount": 10,
|
|
|
103 "MaximumConnectionRetries" : 2000,
|
|
|
104 "ConnectionRetryInterval" : 5,
|
|
620
|
105 "TransactionMode": "READ COMMITTED"
|
|
617
|
106 #"TransactionMode": "Serializable",
|
|
620
|
107 # "EnableVerboseLogs": True
|
|
592
|
108 },
|
|
|
109 "AuthenticationEnabled": False,
|
|
|
110 "OverwriteInstances": True,
|
|
|
111 "JobsEngineThreadsCount" : {
|
|
|
112 "ResourceModification": 8
|
|
|
113 },
|
|
|
114 }
|
|
|
115
|
|
|
116 config_path = cls.generate_configuration(
|
|
|
117 config_name=f"{test_name}",
|
|
|
118 storage_name=cls._storage_name,
|
|
|
119 config=config,
|
|
|
120 plugins=Helpers.plugins
|
|
|
121 )
|
|
|
122
|
|
|
123 # launch the docker PG server
|
|
|
124 print('--------------- launching PostgreSQL server ------------------')
|
|
|
125
|
|
|
126 cls.pg_service_process = subprocess.Popen([
|
|
|
127 "docker", "run", "--rm",
|
|
|
128 "-p", "5432:5432",
|
|
|
129 "--network", "concurrency",
|
|
|
130 "--name", "pg-server",
|
|
|
131 "--env", "POSTGRES_HOST_AUTH_METHOD=trust",
|
|
|
132 "postgres:15"])
|
|
|
133 time.sleep(5)
|
|
|
134
|
|
|
135 if Helpers.break_before_preparation:
|
|
|
136 print(f"++++ It is now time to start your Orthanc under tests with configuration file '{config_path}' +++++")
|
|
|
137 input("Press Enter to continue")
|
|
|
138 else:
|
|
|
139 cls.launch_orthanc_under_tests(
|
|
|
140 config_name=f"{test_name}",
|
|
|
141 storage_name=cls._storage_name,
|
|
|
142 config=config,
|
|
|
143 plugins=Helpers.plugins,
|
|
|
144 docker_network="concurrency"
|
|
|
145 )
|
|
|
146
|
|
|
147 cls.o = OrthancApiClient(cls.o._root_url)
|
|
|
148 cls.o.wait_started()
|
|
|
149 cls.o.delete_all_content()
|
|
597
|
150
|
|
|
151 def check_is_empty(self):
|
|
620
|
152 print("checking is empty")
|
|
|
153
|
|
597
|
154 self.assertEqual(0, len(self.o.studies.get_all_ids()))
|
|
|
155 self.assertEqual(0, len(self.o.series.get_all_ids()))
|
|
|
156 self.assertEqual(0, len(self.o.instances.get_all_ids()))
|
|
|
157
|
|
620
|
158 print("checking is empty (2)")
|
|
597
|
159 stats = self.o.get_json("/statistics")
|
|
620
|
160 print("checking is empty (3)")
|
|
597
|
161 self.assertEqual(0, stats.get("CountPatients"))
|
|
|
162 self.assertEqual(0, stats.get("CountStudies"))
|
|
|
163 self.assertEqual(0, stats.get("CountSeries"))
|
|
|
164 self.assertEqual(0, stats.get("CountInstances"))
|
|
|
165 self.assertEqual(0, int(stats.get("TotalDiskSize")))
|
|
|
166 # time.sleep(10000)
|
|
|
167 self.assertTrue(self.is_storage_empty(self._storage_name))
|
|
|
168
|
|
|
169 # all changes shall have been deleted as well
|
|
|
170 changes, last_change, done = self.o.get_changes(since=0, limit=100000)
|
|
|
171 self.assertTrue(done)
|
|
|
172 self.assertEqual(0, len(changes))
|
|
|
173
|
|
592
|
174
|
|
|
175 def execute_workers(self, worker_func, worker_args, workers_count):
|
|
|
176 workers = []
|
|
|
177 for i in range(0, workers_count):
|
|
|
178 t = threading.Thread(target=worker_func, args=worker_args + (i, ))
|
|
|
179 workers.append(t)
|
|
|
180 t.start()
|
|
|
181
|
|
|
182 for t in workers:
|
|
|
183 t.join()
|
|
|
184
|
|
620
|
185 # def test_concurrent_uploads_same_study(self):
|
|
|
186 # self.o.delete_all_content()
|
|
|
187 # self.clear_storage(storage_name=self._storage_name)
|
|
592
|
188
|
|
620
|
189 # start_time = time.time()
|
|
|
190 # workers_count = 20
|
|
|
191 # repeat_count = 1
|
|
592
|
192
|
|
620
|
193 # # massively reupload the same study multiple times with OverwriteInstances set to true
|
|
|
194 # # Make sure the studies, series and instances are created only once
|
|
|
195 # self.execute_workers(
|
|
|
196 # worker_func=worker_upload_folder,
|
|
|
197 # worker_args=(self.o._root_url, here / "../../Database/Knee", repeat_count,),
|
|
|
198 # workers_count=workers_count)
|
|
592
|
199
|
|
620
|
200 # elapsed = time.time() - start_time
|
|
|
201 # print(f"TIMING test_concurrent_uploads_same_study with {workers_count} workers and {repeat_count}x repeat: {elapsed:.3f} s")
|
|
592
|
202
|
|
620
|
203 # self.assertTrue(self.o.is_alive())
|
|
592
|
204
|
|
620
|
205 # self.assertEqual(1, len(self.o.studies.get_all_ids()))
|
|
|
206 # self.assertEqual(2, len(self.o.series.get_all_ids()))
|
|
|
207 # self.assertEqual(50, len(self.o.instances.get_all_ids()))
|
|
592
|
208
|
|
620
|
209 # stats = self.o.get_json("/statistics")
|
|
|
210 # self.assertEqual(1, stats.get("CountPatients"))
|
|
|
211 # self.assertEqual(1, stats.get("CountStudies"))
|
|
|
212 # self.assertEqual(2, stats.get("CountSeries"))
|
|
|
213 # self.assertEqual(50, stats.get("CountInstances"))
|
|
|
214 # self.assertEqual(4118738, int(stats.get("TotalDiskSize")))
|
|
592
|
215
|
|
620
|
216 # self.o.instances.delete(orthanc_ids=self.o.instances.get_all_ids())
|
|
592
|
217
|
|
620
|
218 # self.check_is_empty()
|
|
592
|
219
|
|
|
220 def test_concurrent_anonymize_same_study(self):
|
|
|
221 self.o.delete_all_content()
|
|
|
222 self.clear_storage(storage_name=self._storage_name)
|
|
|
223
|
|
|
224 self.o.upload_folder(here / "../../Database/Knee")
|
|
|
225 study_id = self.o.studies.get_all_ids()[0]
|
|
|
226
|
|
|
227 start_time = time.time()
|
|
|
228 workers_count = 4
|
|
620
|
229 repeat_count = 2
|
|
592
|
230
|
|
|
231 # massively anonymize the same study. This generates new studies and is a
|
|
|
232 # good way to simulate ingestion of new studies
|
|
|
233 self.execute_workers(
|
|
|
234 worker_func=worker_anonymize_study,
|
|
|
235 worker_args=(self.o._root_url, study_id, repeat_count,),
|
|
|
236 workers_count=workers_count)
|
|
|
237
|
|
|
238 elapsed = time.time() - start_time
|
|
593
|
239 print(f"TIMING test_concurrent_anonymize_same_study with {workers_count} workers and {repeat_count}x repeat: {elapsed:.3f} s")
|
|
592
|
240
|
|
|
241 self.assertTrue(self.o.is_alive())
|
|
|
242
|
|
|
243 self.assertEqual(1 + workers_count * repeat_count, len(self.o.studies.get_all_ids()))
|
|
|
244 self.assertEqual(2 * (1 + workers_count * repeat_count), len(self.o.series.get_all_ids()))
|
|
|
245 self.assertEqual(50 * (1 + workers_count * repeat_count), len(self.o.instances.get_all_ids()))
|
|
|
246
|
|
620
|
247 print("get stats")
|
|
592
|
248 stats = self.o.get_json("/statistics")
|
|
620
|
249 print("get stats (2)")
|
|
592
|
250 self.assertEqual(1 + workers_count * repeat_count, stats.get("CountPatients"))
|
|
|
251 self.assertEqual(1 + workers_count * repeat_count, stats.get("CountStudies"))
|
|
|
252 self.assertEqual(2 * (1 + workers_count * repeat_count), stats.get("CountSeries"))
|
|
|
253 self.assertEqual(50 * (1 + workers_count * repeat_count), stats.get("CountInstances"))
|
|
620
|
254 print("get changes")
|
|
597
|
255 changes, last_change, done = self.o.get_changes(since=0, limit=100000)
|
|
620
|
256 print("get changes (2)")
|
|
597
|
257 self.assertTrue(done)
|
|
|
258
|
|
|
259 self.assertEqual(1 + workers_count * repeat_count, count_changes(changes, ChangeType.NEW_PATIENT))
|
|
|
260 self.assertEqual(1 + workers_count * repeat_count, count_changes(changes, ChangeType.NEW_STUDY))
|
|
|
261 self.assertEqual(2 * (1 + workers_count * repeat_count), count_changes(changes, ChangeType.NEW_SERIES))
|
|
|
262 self.assertEqual(50 * (1 + workers_count * repeat_count), count_changes(changes, ChangeType.NEW_INSTANCE))
|
|
592
|
263
|
|
593
|
264 start_time = time.time()
|
|
|
265
|
|
620
|
266 print("deleting")
|
|
|
267 all_instances_ids = self.o.instances.get_all_ids()
|
|
|
268 print("deleting (2)")
|
|
|
269 self.o.instances.delete(orthanc_ids=all_instances_ids)
|
|
|
270 print("deleted")
|
|
592
|
271
|
|
593
|
272 elapsed = time.time() - start_time
|
|
|
273 print(f"TIMING test_concurrent_anonymize_same_study deletion took: {elapsed:.3f} s")
|
|
|
274
|
|
597
|
275 self.check_is_empty()
|
|
592
|
276
|
|
|
277
|
|
|
278 def test_upload_delete_same_study_from_multiple_threads(self):
|
|
|
279 self.o.delete_all_content()
|
|
|
280 self.clear_storage(storage_name=self._storage_name)
|
|
|
281
|
|
|
282 start_time = time.time()
|
|
596
|
283 overall_repeat = 10
|
|
594
|
284
|
|
598
|
285 for i in range(0, overall_repeat):
|
|
594
|
286 workers_count = 5
|
|
596
|
287 repeat_count = 3
|
|
594
|
288
|
|
|
289 # massively upload and delete the same study. Each worker is writing a part of the instances and deleting them.
|
|
|
290 # We are trying to have multiple workers deleting the last instance of a study at the same time.
|
|
|
291 self.execute_workers(
|
|
|
292 worker_func=worker_upload_delete_study_part,
|
|
|
293 worker_args=(self.o._root_url, here / "../../Database/Knee/T1", repeat_count, workers_count, ),
|
|
|
294 workers_count=workers_count)
|
|
592
|
295
|
|
597
|
296 self.check_is_empty()
|
|
592
|
297
|
|
|
298 elapsed = time.time() - start_time
|
|
596
|
299 print(f"TIMING test_upload_delete_same_study_from_multiple_threads with {workers_count} workers and {repeat_count}x repeat ({overall_repeat}x): {elapsed:.3f} s")
|
|
592
|
300
|
|
|
301
|
|
598
|
302 def test_upload_multiple_studies_from_multiple_threads(self):
|
|
|
303 self.o.delete_all_content()
|
|
|
304 self.clear_storage(storage_name=self._storage_name)
|
|
|
305
|
|
|
306 start_time = time.time()
|
|
|
307 overall_repeat = 3
|
|
|
308
|
|
|
309 for i in range(0, overall_repeat):
|
|
|
310 files_count = 25
|
|
|
311 workers_count = 10
|
|
|
312
|
|
|
313 # massively upload and delete all studies from the test detabase. Each worker is writing all instances from a folder and then deletes them.
|
|
|
314 # This test is only measuring performances.
|
|
|
315 self.execute_workers(
|
|
|
316 worker_func=worker_upload_delete_test_dicoms,
|
|
|
317 worker_args=(self.o._root_url, files_count, ),
|
|
|
318 workers_count=workers_count)
|
|
|
319
|
|
|
320 self.check_is_empty()
|
|
|
321
|
|
|
322 elapsed = time.time() - start_time
|
|
|
323 print(f"TIMING test_upload_multiple_studies_from_multiple_threads with {workers_count} workers and {files_count} files and repeat {overall_repeat}x: {elapsed:.3f} s")
|
|
|
324
|
|
592
|
325 # transfers + dicomweb |
