mputils.py
3.99 KB
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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
# ==========================================================================
# Utility functions for multiprocessing handling
#
# Copyright (C) 2018 Luigi Tibaldo
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# ==========================================================================
import gammalib
# ===================== #
# Get number of threads #
# ===================== #
def nthreads(cls):
"""
Determines the number of parallel processes to use
The number is based on the user parameter "nthreads" and the availability
of the multiprocessing module.
Parameters
----------
cls : `~ctools.cscript`
cscript class
Returns
-------
nthreads : int
Number of threads
"""
# Log multiprocessing configuration
cls._log_header1(gammalib.NORMAL, 'Multiprocessing')
# Try getting multiprocessing support
try:
from multiprocessing import cpu_count
ncpus = cpu_count()
cls._log_value(gammalib.NORMAL, 'Multiprocessing', 'available')
cls._log_value(gammalib.EXPLICIT, 'CPUs available', ncpus)
# Set processes to number of CPUs if requested by the user
if cls['nthreads'].integer() == 0:
nthreads = ncpus
# ... otherwise use the number requested
else:
cls._log_value(gammalib.EXPLICIT, 'Processes requested',
cls['nthreads'].integer())
nthreads = cls['nthreads'].integer()
except:
nthreads = 1
cls._log_value(gammalib.NORMAL, 'Multiprocessing', 'not available')
# Log number of processes
cls._log_value(gammalib.NORMAL, 'Processes available', nthreads)
# Return number of processes to be used
return nthreads
# ============================ #
# Execute function in parallel #
# ============================ #
def process(nthreads, function, args):
"""
Execute function in parallel
Parameters
----------
nthreads : int
Number of parallel threads
function : Python function
Function to be executed in parallel
args : list of tuples
Function arguments for each evaluation
Returns
-------
results : list of dicts
List of function evaluation results
"""
# Import Pool module from multiprocessing module
from multiprocessing import Pool
# Create a pool of processes
pool = Pool(processes=nthreads)
# Run function in parallel
results = pool.map(function, args)
# Close pool and join
pool.close()
pool.join()
# Return results
return results
# ======== #
# Function #
# ======== #
def mpfunc(args):
"""
Multiprocessing function
Parameters
----------
args : tuple
Tuple comprised of class, function name and function argument
Returns
-------
result, info : tuple
Tuple of function result and information
"""
# Extract
cls = args[0] # Class
fct = args[1] # Class method name
i = args[2] # Function argument
# Initialise thread logger
cls._log.clear()
cls._log.buffer_size(100000)
# Compute function for argument
cstart = cls.celapse()
result = getattr(cls, fct)(i)
celapse = cls.celapse() - cstart
log = cls._log.buffer()
# Close logger
cls._log.close()
# Collect thread information
info = {'celapse': celapse, 'log': log}
# Return light curve bin result and thread information
return result, info