Skip to content

Restrict multiprocessing to available CPUs #293

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mihaic merged 1 commit into from
Nov 8, 2017
Merged

Restrict multiprocessing to available CPUs #293

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 9 additions & 4 deletions brainiak/fcma/voxelselector.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
# (Intel Labs), 2016

import numpy as np
import os
import time
from mpi4py import MPI
from scipy.stats.mstats import zscore
Expand Down Expand Up @@ -89,10 +90,10 @@ class VoxelSelector:
The number of voxels assigned to a worker each time

process_num: Optional[int]
The number of processes used in cross validation.
The maximum number of processes used in cross validation.
If None, the number of processes will equal
the number of hardware threads,
i.e. the number returned by cpu_count().
the number of available hardware threads,
i.e. len(os.sched_getaffinity(0)), falling back to os.cpu_count().
If 0, cross validation will not use python multiprocessing.

master_rank: int, default 0
Expand All @@ -116,7 +117,11 @@ def __init__(self,
self.num_voxels2 = raw_data2[0].shape[1] \
if raw_data2 is not None else self.num_voxels
self.voxel_unit = voxel_unit
self.process_num = np.min((process_num, multiprocessing.cpu_count()))
try:
usable_cpus = len(os.sched_getaffinity(0))
except AttributeError:
usable_cpus = os.cpu_count()
self.process_num = np.min((process_num, usable_cpus))
if self.process_num == 0:
self.use_multiprocessing = False
else:
Expand Down
18 changes: 15 additions & 3 deletions brainiak/searchlight/searchlight.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.

import os
from multiprocessing import Pool
import numpy as np
from mpi4py import MPI
Expand Down Expand Up @@ -387,14 +388,25 @@ def run_block_function(self, block_fn, extra_block_fn_params=None,
extra_block_fn_params: tuple
Extra parameters to pass to the block function

pool_size: Number of parallel processes in shared memory
process pool
pool_size: int
Maximum number of processes running the block function in parallel.
The actual number of processes created is
min(pool_size, len(os.sched_getaffinity(0))) (falling back to
min(pool_size, os.cpu_count()).

"""
rank = self.comm.rank

results = []
with Pool(pool_size) as pool:
try:
usable_cpus = len(os.sched_getaffinity(0))
except AttributeError:
usable_cpus = os.cpu_count()
if pool_size is None:
processes = usable_cpus
else:
processes = min(pool_size, usable_cpus)
with Pool(processes) as pool:
for idx, block in enumerate(self.blocks):
result = pool.apply_async(
block_fn,
Expand Down
1 change: 1 addition & 0 deletions docs/newsfragments/293.feature
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Restricted multiprocessing to the available number of CPUs.