QR + computing_units

Author: michal-choinski

.gitignore

@@ -175,3 +175,5 @@ dkms.conf
 # Ignore log files
 *.log
 
+# Ignore virtual environments
+venv

Jenkinsfile

@@ -11,7 +11,7 @@ def setGithubCommitStatus(state, description) {
 
 pipeline {
     options {
-        timeout(time: 3, unit: 'HOURS')
+        timeout(time: 4, unit: 'HOURS')
     }
     agent {
         node {

dislib/__init__.py

@@ -1,7 +1,7 @@
 import os
 
 from dislib.data.array import random_array, apply_along_axis, array, zeros, \
-    full, identity
+    full, identity, eye
 from dislib.data.io import load_svmlight_file, load_npy_file, load_txt_file, \
     load_mdcrd_file, save_txt
 from dislib.math import kron, svd
@@ -27,7 +27,7 @@
               "Probably it was not installed with setup.py.\n%s" % e)
         __version__ = 'unknown'
 
-__all__ = ['array', 'random_array', 'zeros', 'full', 'identity',
+__all__ = ['array', 'random_array', 'zeros', 'full', 'identity', 'eye',
            'load_txt_file', 'load_svmlight_file', 'load_npy_file',
            'load_mdcrd_file', 'save_txt',
            'apply_along_axis', 'kron', 'svd']

dislib/classification/csvm/base.py

@@ -3,6 +3,7 @@
 import numpy as np
 from pycompss.api.api import compss_delete_object
 from pycompss.api.api import compss_wait_on
+from pycompss.api.constraint import constraint
 from pycompss.api.parameter import COLLECTION_IN, Depth, Type
 from pycompss.api.task import task
 from scipy.sparse import hstack as hstack_sp
@@ -383,12 +384,14 @@ def _linear_kernel(x):
         return np.dot(x, x.T)
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=1)
 def _gen_ids(n_samples):
     idx = [[uuid4().int] for _ in range(n_samples)]
     return np.array(idx)
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(x_list={Type: COLLECTION_IN, Depth: 2},
       y_list={Type: COLLECTION_IN, Depth: 2},
       id_list={Type: COLLECTION_IN, Depth: 2},
@@ -414,18 +417,21 @@ def _train(x_list, y_list, id_list, random_state, **params):
     return sv, sv_labels, sv_ids, clf
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(x_list={Type: COLLECTION_IN, Depth: 2}, returns=np.array)
 def _predict(x_list, clf):
     x = Array._merge_blocks(x_list)
     return clf.predict(x).reshape(-1, 1)
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(x_list={Type: COLLECTION_IN, Depth: 2}, returns=np.array)
 def _decision_function(x_list, clf):
     x = Array._merge_blocks(x_list)
     return clf.decision_function(x).reshape(-1, 1)
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(x_list={Type: COLLECTION_IN, Depth: 2},
       y_list={Type: COLLECTION_IN, Depth: 2}, returns=tuple)
 def _score(x_list, y_list, clf):
@@ -438,6 +444,7 @@ def _score(x_list, y_list, clf):
     return np.sum(equal), x.shape[0]
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=float)
 def _merge_scores(*partials):
     total_correct = 0.

dislib/cluster/daura/base.py

@@ -1,6 +1,7 @@
 from itertools import chain
 
 from pycompss.api.api import compss_wait_on
+from pycompss.api.constraint import constraint
 from pycompss.api.parameter import COLLECTION_IN, INOUT
 from pycompss.api.task import task
 from sklearn.base import BaseEstimator
@@ -72,11 +73,13 @@ def fit_predict(self, distances):
         return clusters
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=1)
 def _get_neighbors(block, start_col, cutoff):
     return [np.flatnonzero(r <= cutoff) + start_col for r in block]
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(row_blocks_neighbors=COLLECTION_IN, returns=1)
 def _merge_neighbors(row_blocks_neighbors, start_idx):
     row_neighs = [set(chain(*nb)) for nb in zip(*row_blocks_neighbors)]
@@ -85,6 +88,7 @@ def _merge_neighbors(row_blocks_neighbors, start_idx):
     return row_neighs
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=1)
 def _find_candidate_cluster(row_neighbors, start_row):
     argmax = max(range(len(row_neighbors)),
@@ -94,11 +98,13 @@ def _find_candidate_cluster(row_neighbors, start_row):
     return cluster
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(candidates=COLLECTION_IN, returns=1)
 def _find_largest_cluster(candidates):
     return max(candidates, key=len)
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(row_neighbors=INOUT)
 def _remove_neighbors(row_neighbors, to_remove, start_row):
     for r in to_remove:

dislib/cluster/dbscan/base.py

@@ -1,5 +1,6 @@
 import numpy as np
 from pycompss.api.api import compss_wait_on
+from pycompss.api.constraint import constraint
 from pycompss.api.parameter import COLLECTION_IN, COLLECTION_OUT, \
     Type, Depth
 from pycompss.api.task import task
@@ -311,6 +312,7 @@ def _rearrange_labels(labels, indices, n_blocks):
     return sorted_blocks
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(mn={Type: COLLECTION_IN, Depth: 2},
       mx={Type: COLLECTION_IN, Depth: 2},
       returns=1)
@@ -331,6 +333,7 @@ def _generate_bins(mn, mx, dimensions, n_regions):
     return bins
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(blocks={Type: COLLECTION_IN, Depth: 2},
       samples_list={Type: COLLECTION_OUT},
       indices={Type: COLLECTION_OUT})
@@ -372,6 +375,7 @@ def _arrange_block(blocks, bins, dimensions, shape, samples_list, indices):
         indices[i] = sample_indices
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(indices=COLLECTION_IN,
       blocks=COLLECTION_OUT)
 def _rearrange_region(labels, indices, blocks):
@@ -389,6 +393,7 @@ def _rearrange_region(labels, indices, blocks):
         start = end
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(samples_list={Type: COLLECTION_IN}, returns=2)
 def _merge_samples(samples_list, sparse):
     if sparse:
@@ -399,6 +404,7 @@ def _merge_samples(samples_list, sparse):
     return samples, samples.shape[0]
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(labels_list=COLLECTION_IN, indices=COLLECTION_IN, returns=1)
 def _merge_labels(labels_list, indices):
     labels = np.vstack(labels_list)
@@ -409,6 +415,7 @@ def _merge_labels(labels_list, indices):
     return np.take(labels, idx).reshape(-1, 1)
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=1)
 def _merge_dicts(*dicts):
     merged_dict = {}
@@ -419,6 +426,7 @@ def _merge_dicts(*dicts):
     return merged_dict
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=1)
 def _get_connected_components(equiv):
     # Add inverse equivalences

dislib/cluster/dbscan/classes.py

@@ -3,6 +3,7 @@
 from itertools import chain
 
 import numpy as np
+from pycompss.api.constraint import constraint
 from pycompss.api.task import task
 from scipy.sparse import lil_matrix, vstack as vstack_sparse
 from scipy.sparse.csgraph import connected_components
@@ -89,6 +90,7 @@ def update_labels(self, components):
                                      components)
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=1)
 def _update_labels(labels_region, labels, components):
     components_map = {}
@@ -102,6 +104,7 @@ def _update_labels(labels_region, labels, components):
     return labels
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=3)
 def _compute_equivalences(n_samples, region_ids, *starred_args):
     n_regions = len(region_ids)
@@ -143,6 +146,7 @@ def _compute_equivalences(n_samples, region_ids, *starred_args):
     return label_regions, labels, equiv
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=4)
 def _compute_neighbours(epsilon, min_samples, begin_idx, end_idx, samples,
                         sparse, *neigh_samples):
@@ -196,11 +200,13 @@ def _concatenate_samples(sparse, *samples):
         return vstack_sparse(samples)
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=1)
 def _lists_to_array(*cp_list):
     return np.concatenate(cp_list)
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=1)
 def _compute_cp_labels(core_points, *in_cp_neighs):
     core_ids = np.cumsum(core_points) - 1

dislib/cluster/gm/base.py

@@ -3,6 +3,7 @@
 import numpy as np
 from numpy.random.mtrand import RandomState
 from pycompss.api.api import compss_wait_on, compss_delete_object
+from pycompss.api.constraint import constraint
 from pycompss.api.parameter import Type, COLLECTION_IN, Depth
 from pycompss.api.task import task
 from scipy import linalg
@@ -541,6 +542,7 @@ def _initialize_parameters(self, x, random_state):
                 compss_delete_object(resp_block)
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(x={Type: COLLECTION_IN, Depth: 2},
       resp={Type: COLLECTION_IN, Depth: 2},
       returns=1)
@@ -564,6 +566,7 @@ def _reduce_estimate_parameters(partials, arity):
     return _finalize_parameters(partials[0])
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=1)
 def _merge_estimate_parameters(*partials_params):
     n_samples = sum(params[0] for params in partials_params)
@@ -572,6 +575,7 @@ def _merge_estimate_parameters(*partials_params):
     return n_samples, nk, means
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=3)
 def _finalize_parameters(params):
     n_samples = params[0]
@@ -634,6 +638,7 @@ def _estimate_covariances(x, resp, nk, means, reg_covar, covar_type, arity):
     return finalize_covariances(covar_type, reg_covar, nk, means, partials[0])
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(x={Type: COLLECTION_IN, Depth: 2},
       resp={Type: COLLECTION_IN, Depth: 2},
       returns=1)
@@ -654,6 +659,7 @@ def _partial_covar_full(resp, x, means):
     return covariances
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(x={Type: COLLECTION_IN, Depth: 2},
       returns=1)
 def _partial_covar_tied(x):
@@ -665,6 +671,7 @@ def _partial_covar_tied(x):
     return avg_sample_2
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(x={Type: COLLECTION_IN, Depth: 2},
       resp={Type: COLLECTION_IN, Depth: 2},
       returns=1)
@@ -680,11 +687,13 @@ def _partial_covar_diag(resp, x, means):
     return avg_resp_sample_2 - 2 * avg_sample_means
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=1)
 def _sum_covar_partials(*covar_partials):
     return sum(covar_partials)
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=2)
 def _finalize_covar_full(covar_type, reg_covar, nk, covariances):
     n_components, n_features, _ = covariances.shape
@@ -695,6 +704,7 @@ def _finalize_covar_full(covar_type, reg_covar, nk, covariances):
     return covariances, precisions_chol
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=2)
 def _finalize_covar_tied(covar_type, reg_covar, nk, means, covariances):
     avg_means2 = np.dot(nk * means.T, means)
@@ -705,6 +715,7 @@ def _finalize_covar_tied(covar_type, reg_covar, nk, means, covariances):
     return covariances, precisions_chol
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=2)
 def _finalize_covar_diag(covar_type, reg_covar, nk, means, covariances):
     covariances /= nk[:, np.newaxis]
@@ -714,6 +725,7 @@ def _finalize_covar_diag(covar_type, reg_covar, nk, means, covariances):
     return covariances, precisions_chol
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=2)
 def _finalize_covar_spherical(covar_type, reg_covar, nk, means, covariances):
     covariances /= nk[:, np.newaxis]
@@ -775,18 +787,21 @@ def _compute_precision_cholesky(covariances, covariance_type):
     return precisions_chol
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=1)
 def _sum_log_prob_norm(*partials):
     total, count = map(sum, zip(*partials))
     return total, count
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=1)
 def _finalize_sum_log_prob_norm(*partials):
     total, count = map(sum, zip(*partials))
     return total / count
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(x={Type: COLLECTION_IN, Depth: 2}, returns=2)
 def _estimate_responsibilities(x, weights, means, precisions_cholesky,
                                covariance_type):
@@ -964,12 +979,14 @@ def _resp_argmax(resp):
     return pred
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(resp={Type: COLLECTION_IN, Depth: 2}, returns=1)
 def _partial_resp_argmax(resp):
     resp = Array._merge_blocks(resp)
     return resp.argmax(axis=1)[:, np.newaxis]
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(labels={Type: COLLECTION_IN, Depth: 2}, returns=1)
 def _resp_subset(labels, n_components):
     labels = Array._merge_blocks(labels).flatten()
@@ -979,6 +996,7 @@ def _resp_subset(labels, n_components):
     return resp_chunk
 
 
+@constraint(computing_units="${ComputingUnits}")
 @task(returns=1)
 def _random_resp_subset(n_samples, n_components, seed):
     resp_chunk = RandomState(seed).rand(n_samples, n_components)