-
Notifications
You must be signed in to change notification settings - Fork 52
/
thread_check.py
68 lines (55 loc) · 1.86 KB
/
thread_check.py
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
# -*- coding: utf-8 -*-
# Copyright (c) oct2py developers.
# Distributed under the terms of the MIT License.
from __future__ import print_function, absolute_import
import threading
import datetime
from . import Oct2Py, Oct2PyError
class ThreadClass(threading.Thread):
"""Octave instance thread
"""
def run(self):
"""
Create a unique instance of Octave and verify namespace uniqueness.
Raises
======
Oct2PyError
If the thread does not sucessfully demonstrate independence
"""
octave = Oct2Py()
# write the same variable name in each thread and read it back
octave.push('name', self.getName())
name = octave.pull('name')
now = datetime.datetime.now()
print("{0} got '{1}' at {2}".format(self.getName(), name, now))
octave.exit()
try:
assert self.getName() == name
except AssertionError: # pragma: no cover
raise Oct2PyError('Thread collision detected')
return
def thread_check(nthreads=3):
"""
Start a number of threads and verify each has a unique Octave session.
Parameters
==========
nthreads : int
Number of threads to use.
Raises
======
Oct2PyError
If the thread does not sucessfully demonstrate independence.
"""
print("Starting {0} threads at {1}".format(nthreads,
datetime.datetime.now()))
threads = []
for i in range(nthreads):
thread = ThreadClass()
thread.setDaemon(True)
thread.start()
threads.append(thread)
for thread in threads:
thread.join()
print('All threads closed at {0}'.format(datetime.datetime.now()))
if __name__ == '__main__': # pragma: no cover
thread_check()