Remove ThreadShare to avoid unnecessary copies and reduce memory usage

Use omp_size_t in parallel for loop

src/mlpack/core/optimizers/parallel_sgd/decay_policies/exponential_backoff.hpp

@@ -50,10 +50,10 @@ class ExponentialBackoff
   ExponentialBackoff(const size_t firstBackoffEpoch,
                      const double step,
                      const double beta) :
-                     firstBackoffEpoch(firstBackoffEpoch),
-                     cutoffEpoch(firstBackoffEpoch),
-                     step(step),
-                     beta(beta)
+    firstBackoffEpoch(firstBackoffEpoch),
+    cutoffEpoch(firstBackoffEpoch),
+    step(step),
+    beta(beta)
   { /* Nothing to do. */ }
 
   /**

src/mlpack/core/optimizers/parallel_sgd/parallel_sgd.hpp

@@ -116,18 +116,6 @@ class ParallelSGD
   DecayPolicyType& DecayPolicy() { return decayPolicy; }
 
  private:
-  /**
-   * Get the share of datapoint indices to be updated by the thread with given
-   * thread id.
-   *
-   * @param thread_id The id of the current thread. Range 0-OMP_NUM_THREADS.
-   * @param visitationOrder The random list of datapoint indices for the current
-   *    iteration.
-   * @return Vector of datapoint indices to be visited by the current thread.
-   */
-  arma::Col<size_t> ThreadShare(size_t threadId,
-                                const arma::Col<size_t>& visitationOrder);
-
   //! The maximum number of allowed iterations.
   size_t maxIterations;
 

src/mlpack/core/optimizers/parallel_sgd/parallel_sgd_impl.hpp

@@ -86,16 +86,18 @@ double ParallelSGD<DecayPolicyType>::Optimize(
     {
       // Each processor gets a subset of the instances.
       // Each subset is of size threadShareSize.
-      arma::Col<size_t> instances = ThreadShare(omp_get_thread_num(),
-          visitationOrder);
-      for (size_t j = 0; j < instances.n_elem; ++j)
+      size_t threadId = omp_get_thread_num();
+
+      for (size_t j = threadId * threadShareSize;
+          j < (threadId + 1) * threadShareSize && j < visitationOrder.n_elem;
+          ++j)
       {
         // Each instance affects only some components of the decision variable.
         // So the gradient is sparse.
         arma::sp_mat gradient;
 
         // Evaluate the sparse gradient.
-        function.Gradient(iterate, instances[j], gradient);
+        function.Gradient(iterate, visitationOrder[j], gradient);
 
         // Update the decision variable with non-zero components of the
         // gradient.
@@ -117,29 +119,6 @@ double ParallelSGD<DecayPolicyType>::Optimize(
   return overallObjective;
 }
 
-template <typename DecayPolicyType>
-arma::Col<size_t> ParallelSGD<DecayPolicyType>::ThreadShare(
-    size_t threadId, const arma::Col<size_t>& visitationOrder)
-{
-  if (threadId * threadShareSize >= visitationOrder.n_elem)
-  {
-    // No data for this thread.
-    return arma::Col<size_t>();
-  }
-  else if ((threadId + 1) * threadShareSize >= visitationOrder.n_elem)
-  {
-    // The last few elements.
-    return visitationOrder.subvec(threadId * threadShareSize ,
-        visitationOrder.n_elem - 1);
-  }
-  else
-  {
-    // Equal distribution of threadShareSize examples to each thread.
-    return visitationOrder.subvec(threadId * threadShareSize,
-        (threadId + 1) * threadShareSize - 1);
-  }
-}
-
 } // namespace optimization
 } // namespace mlpack
 

src/mlpack/methods/regularized_svd/regularized_svd_function_impl.hpp

@@ -244,7 +244,7 @@ inline double ParallelSGD<ExponentialBackoff>::Optimize(
     overallObjective = 0;
 
     #pragma omp parallel for reduction(+:overallObjective)
-    for (size_t j = 0; j < function.NumFunctions(); ++j)
+    for (omp_size_t j = 0; j < (omp_size_t) function.NumFunctions(); ++j)
     {
       overallObjective += function.Evaluate(iterate, j);
     }
@@ -278,18 +278,20 @@ inline double ParallelSGD<ExponentialBackoff>::Optimize(
     {
       // Each processor gets a subset of the instances.
       // Each subset is of size threadShareSize.
-      arma::Col<size_t> instances = ThreadShare(omp_get_thread_num(),
-          visitationOrder);
-      for (size_t j = 0; j < instances.n_elem; ++j)
+      size_t threadId = omp_get_thread_num();
+
+      for (size_t j = threadId * threadShareSize;
+          j < (threadId + 1) * threadShareSize && j < visitationOrder.n_elem;
+          ++j)
       {
         const size_t numUsers = function.NumUsers();
 
         // Indices for accessing the the correct parameter columns.
-        const size_t user = data(0, instances[j]);
-        const size_t item = data(1, instances[j]) + numUsers;
+        const size_t user = data(0, visitationOrder[j]);
+        const size_t item = data(1, visitationOrder[j]) + numUsers;
 
         // Prediction error for the example.
-        const double rating = data(2, instances[j]);
+        const double rating = data(2, visitationOrder[j]);
         double ratingError = rating - arma::dot(iterate.col(user),
             iterate.col(item));
 

src/mlpack/tests/parallel_sgd_test.cpp

@@ -97,42 +97,6 @@ BOOST_AUTO_TEST_CASE(GeneralizedRosenbrockTest)
   }
 }
 
-/**
- * Test if the data points are divided correctly among the threads.
- */
-BOOST_AUTO_TEST_CASE(ThreadSharingTest)
-{
-  ConstantStep decayPolicy(0);
-
-  // Each thread gets a batch of size 4.
-  ParallelSGD<ConstantStep> s(0, 4, 1e-10, decayPolicy);
-
-  // Generate a random visitation order.
-  arma::Col<size_t> visitationOrder = arma::linspace<arma::Col<size_t>>(0,
-      9, 10);
-
-  // Lets count how many times each example is handed out in an iteration.
-  arma::Col<size_t> count(10, arma::fill::zeros);
-
-  for (size_t threadId = 0; threadId < 4; ++threadId)
-  {
-    arma::Col<size_t> share = s.ThreadShare(threadId, visitationOrder);
-    for (size_t i = 0; i < share.n_elem; ++i)
-      count(share(i))++;
-
-    // The last thread to have some data.
-    if (threadId == 2)
-      BOOST_REQUIRE_EQUAL(share.n_elem, 2);
-
-    // Only the first 3 threads get data.
-    if (threadId > 2)
-      BOOST_REQUIRE_EQUAL(share.n_elem, 0);
-  }
-
-  // If everything is correct, each count should be 1 for each data point.
-  CheckMatrices(count, arma::Col<size_t>(10, arma::fill::ones));
-}
-
 /**
  * Test the correctness of the Exponential backoff stepsize decay policy.
  */