Skip to content

dev: Remove Pathos and multiprocess dependencies #225

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 4 commits into from
May 23, 2017
Merged

dev: Remove Pathos and multiprocess dependencies #225

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
de0105c
dev: Remove Pathos and multiprocess dependencies
mihaic Apr 27, 2017
6051110
Merge branch 'master' into use-multiprocessing
mihaic May 15, 2017
026a5f4
dev: Ignore .mypy_cache
mihaic May 15, 2017
7263ddf
Use context manager to minimize Pool memory usage
mihaic May 15, 2017
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
2 changes: 1 addition & 1 deletion .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ data

brainiak.egg-info
build
.cache
.*cache
.coverage*
coverage.xml
dist
Expand Down
31 changes: 19 additions & 12 deletions brainiak/fcma/mvpa_voxelselector.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,24 @@
]


def _sfn(l, mask, myrad, bcast_var):
"""Score classifier on searchlight data using cross-validation.

The classifier is in `bcast_var[2]`. The labels are in `bast_var[0]`. The
number of cross-validation folds is in `bast_var[1].
"""
clf = bcast_var[2]
data = l[0][mask, :].T
# print(l[0].shape, mask.shape, data.shape)
skf = model_selection.StratifiedKFold(n_splits=bcast_var[1],
shuffle=False)
accuracy = np.mean(model_selection.cross_val_score(clf, data,
y=bcast_var[0],
cv=skf,
n_jobs=1))
return accuracy


class MVPAVoxelSelector:
"""Activity-based voxel selection component of FCMA

Expand Down Expand Up @@ -94,23 +112,12 @@ def run(self, clf):
'running activity-based voxel selection via Searchlight'
)
self.sl.distribute([self.data], self.mask)
self.sl.broadcast((self.labels, self.num_folds))
self.sl.broadcast((self.labels, self.num_folds, clf))
if rank == 0:
logger.info(
'data preparation done'
)

# Searchlight kernel function
def _sfn(l, mask, myrad, bcast_var):
data = l[0][mask, :].T
# print(l[0].shape, mask.shape, data.shape)
skf = model_selection.StratifiedKFold(n_splits=bcast_var[1],
shuffle=False)
accuracy = np.mean(model_selection.cross_val_score(clf, data,
y=bcast_var[0],
cv=skf,
n_jobs=1))
return accuracy
# obtain a 3D array with accuracy numbers
result_volume = self.sl.run_searchlight(_sfn)
# get result tuple list from the volume
Expand Down
37 changes: 19 additions & 18 deletions brainiak/fcma/voxelselector.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@
from . import fcma_extension # type: ignore
from . import cython_blas as blas # type: ignore
import logging
import multiprocess
import multiprocessing

logger = logging.getLogger(__name__)

Expand All @@ -37,6 +37,21 @@
]


def _cross_validation_for_one_voxel(clf, vid, num_folds, subject_data, labels):
"""Score classifier on data using cross validation."""
# no shuffling in cv
skf = model_selection.StratifiedKFold(n_splits=num_folds,
shuffle=False)
scores = model_selection.cross_val_score(clf, subject_data,
y=labels,
cv=skf, n_jobs=1)
logger.debug(
'cross validation for voxel %d is done' %
vid
)
return (vid, scores.mean())


class VoxelSelector:
"""Correlation-based voxel selection component of FCMA.

Expand Down Expand Up @@ -415,31 +430,17 @@ def _do_cross_validation(self, clf, data, task):
"""
time1 = time.time()

def _cross_validation_for_one_voxel(vid, num_folds,
subject_data, labels):
# no shuffling in cv
skf = model_selection.StratifiedKFold(n_splits=num_folds,
shuffle=False)
scores = model_selection.cross_val_score(clf, subject_data,
y=labels,
cv=skf, n_jobs=1)
logger.debug(
'cross validation for voxel %d is done' %
vid
)
return (vid, scores.mean())

if isinstance(clf, sklearn.svm.SVC) and clf.kernel == 'precomputed':
inlist = [(i + task[0], self.num_folds, data[i, :, :],
inlist = [(clf, i + task[0], self.num_folds, data[i, :, :],
self.labels) for i in range(task[1])]

with multiprocess.Pool(self.process_num) as pool:
with multiprocessing.Pool(self.process_num) as pool:
results = list(pool.starmap(_cross_validation_for_one_voxel,
inlist))
else:
results = []
for i in range(task[1]):
result = _cross_validation_for_one_voxel(i + task[0],
result = _cross_validation_for_one_voxel(clf, i + task[0],
self.num_folds,
data[i, :, :],
self.labels)
Expand Down
110 changes: 71 additions & 39 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.

from multiprocessing import Pool
import numpy as np
from mpi4py import MPI
import sys
Expand Down Expand Up @@ -337,7 +338,7 @@ def broadcast(self, bcast_var):

self.bcast_var = self.comm.bcast(bcast_var)

def run_block_function(self, block_fn):
def run_block_function(self, block_fn, extra_block_fn_params=None):
"""Perform a function for each block in a volume.

Parameters
Expand All @@ -357,20 +358,26 @@ def run_block_function(self, block_fn):

bcast_var: shared data which is broadcast to all processes

extra_params: extra parameters


Returns

3D array which is the same size as the mask
input with padding removed

extra_block_fn_params: tuple
Extra parameters to pass to the block function


"""
rank = self.comm.rank

# Apply searchlight
local_outputs = [(mypt[0],
block_fn([d[idx] for d in self.subproblems],
self.submasks[idx], self.sl_rad, self.bcast_var))
self.submasks[idx], self.sl_rad, self.bcast_var,
extra_block_fn_params))
for (idx, mypt) in enumerate(self.blocks)]

# Collect results
Expand Down Expand Up @@ -400,6 +407,9 @@ def run_searchlight(self, voxel_fn, pool_size=None):

voxel_fn: function to apply at each voxel

Must be `serializeable using pickle
<https://docs.python.org/3/library/pickle.html#what-can-be-pickled-and-unpickled>`_.

Parameters

subj: list of 4D arrays containing subset of subject data
Expand Down Expand Up @@ -430,40 +440,62 @@ def run_searchlight(self, voxel_fn, pool_size=None):

"""

def _singlenode_searchlight(l, msk, mysl_rad, bcast_var):

outmat = np.empty(msk.shape, dtype=np.object)[mysl_rad:-mysl_rad,
mysl_rad:-mysl_rad,
mysl_rad:-mysl_rad]

import pathos.multiprocessing
inlist = [([ll[i:i+2*mysl_rad+1,
j:j+2*mysl_rad+1,
k:k+2*mysl_rad+1,
:]
for ll in l],
msk[i:i+2*mysl_rad+1,
j:j+2*mysl_rad+1,
k:k+2*mysl_rad+1] * self.shape,
mysl_rad,
bcast_var)
if msk[i+mysl_rad, j+mysl_rad, k+mysl_rad] else None
for i in range(0, outmat.shape[0])
for j in range(0, outmat.shape[1])
for k in range(0, outmat.shape[2])]
outlist = list(pathos.multiprocessing.ProcessingPool(pool_size)
.map(lambda x:
voxel_fn(x[0], x[1], x[2], x[3])
if x is not None else None, inlist))

cnt = 0
for i in range(0, outmat.shape[0]):
for j in range(0, outmat.shape[1]):
for k in range(0, outmat.shape[2]):
if outlist[cnt] is not None:
outmat[i, j, k] = outlist[cnt]
cnt = cnt + 1

return outmat

return self.run_block_function(_singlenode_searchlight)
# Reuse a single `Pool` for all blocks of the MPI process.
with Pool(pool_size) as pool:
extra_block_fn_params = (pool, voxel_fn, self.shape)
block_fn_result = self.run_block_function(_singlenode_searchlight,
extra_block_fn_params)
return block_fn_result


def _singlenode_searchlight(l, msk, mysl_rad, bcast_var, extra_params):
"""Run searchlight function on block data in parallel.

`extra_params` contains:

- `Pool`.
- Searchlight function.
- `Shape` mask.
"""

pool = extra_params[0]
voxel_fn = extra_params[1]
shape_mask = extra_params[2]
outmat = np.empty(msk.shape, dtype=np.object)[mysl_rad:-mysl_rad,
mysl_rad:-mysl_rad,
mysl_rad:-mysl_rad]

inlist = []
inlist_where = []
where_idx = 0
for i in range(0, outmat.shape[0]):
for j in range(0, outmat.shape[1]):
for k in range(0, outmat.shape[2]):
if msk[i+mysl_rad, j+mysl_rad, k+mysl_rad]:
inlist.append((
[ll[i:i+2*mysl_rad+1,
j:j+2*mysl_rad+1,
k:k+2*mysl_rad+1,
:]
for ll in l],
msk[i:i+2*mysl_rad+1,
j:j+2*mysl_rad+1,
k:k+2*mysl_rad+1
] * shape_mask,
mysl_rad,
bcast_var))
inlist_where.append(where_idx)
where_idx += 1
outlist_all = np.array([None] * where_idx)
outlist = list(pool.starmap(voxel_fn, inlist))
outlist_all[inlist_where] = outlist

cnt = 0
for i in range(0, outmat.shape[0]):
for j in range(0, outmat.shape[1]):
for k in range(0, outmat.shape[2]):
if outlist_all[cnt] is not None:
outmat[i, j, k] = outlist_all[cnt]
cnt = cnt + 1

return outmat
1 change: 0 additions & 1 deletion setup.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,6 @@ def finalize_options(self):
'pymanopt',
'theano',
'pybind11>=1.7',
'pathos',
'nibabel',
'typing',
],
Expand Down
Loading