Improving concurrently, synchronizing as little as possible.

TODO.txt

@@ -1,14 +1,16 @@
 CODE IMPROVEMENTS
 =================
 
-- Profiling & benchmarking.
 - Add multithreading support on the training for the following models: 
     - OrdinalRegression
     - SoftMaxRegression
     - HierarchicalAgglomerative
     - Kmeans
     - NLMS
-- Search for getFeatureIds() and see it the estimation should change. We can use the Dataframe.xDataTypes instead?
+- Try if possible to use atomic int in dataframe: a.updateAndGet((x) -> max(x+1,rId+1));
+- Add a from-to or limit option on the Dataframe.Builder
+- Consider changing the policy of copying the data on the constructor of Record and start using unmodifiable without copy instead.
+- Profiling & benchmarking.
 
 - Update all maven plugins and dependencies to their latest versions.
     - Add support for MapDB 2.0 once a stable version is released.

src/main/java/com/datumbox/common/concurrency/StreamMethods.java

@@ -72,10 +72,9 @@ public static <T> Stream<T> stream(Iterator<T> iterator, boolean parallel) {
      * 
      * @param <T>
      * @param stream
-     * @param parallel
      * @return 
      */
-    public static <T> Stream<Map.Entry<Integer, T>> enumerate(Stream<T> stream, boolean parallel) {
+    public static <T> Stream<Map.Entry<Integer, T>> enumerate(Stream<T> stream) {
         Iterator<Map.Entry<Integer, T>> iterator = new Iterator<Map.Entry<Integer, T>>() {
             private int counter = 0;
 
@@ -99,7 +98,7 @@ public void remove() {
                 throw new UnsupportedOperationException("This is a read-only iterator, remove operation is not supported.");
             }
         };
-        return stream(iterator, parallel);
+        return stream(iterator, stream.isParallel());
     }
 
 }

src/main/java/com/datumbox/common/concurrency/ThreadMethods.java

@@ -1,62 +1,62 @@
-/**
- * Copyright (C) 2013-2016 Vasilis Vryniotis <bbriniotis@datumbox.com>
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package com.datumbox.common.concurrency;
-
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import java.util.function.Consumer;
-import java.util.stream.Stream;
-
-/**
- * This class contains a number of helper methods for Java 8 Threads.
- *
- * @author Vasilis Vryniotis <bbriniotis@datumbox.com>
- */
-public class ThreadMethods {
-
-    /**
-     * Takes the items of the stream in a throttled way and provides them to the 
-     * consumer. It uses as many threads as the available processors and it does
-     * not start more tasks than 2 times the previous number.
-     * 
-     * @param <T>
-     * @param stream
-     * @param consumer 
-     */
-    public static <T> void throttledExecution(Stream<T> stream, Consumer<T> consumer) {
-        int maxThreads = Runtime.getRuntime().availableProcessors();
-        int maxTasks = 2*maxThreads; 
-
-        ExecutorService executorService = Executors.newFixedThreadPool(maxThreads);
-        ThrottledExecutor executor = new ThrottledExecutor(executorService, maxTasks);
-
-        stream.forEach(i -> {
-            executor.execute(() -> {
-                consumer.accept(i);
-            });
-        });
-
-        executorService.shutdown();
-        try {
-            executorService.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
-        } 
-        catch (InterruptedException ex) {
-            throw new RuntimeException(ex);
-        }
-    }
-
-}
+/**
+ * Copyright (C) 2013-2016 Vasilis Vryniotis <bbriniotis@datumbox.com>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datumbox.common.concurrency;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
+
+/**
+ * This class contains a number of helper methods for Java 8 Threads.
+ *
+ * @author Vasilis Vryniotis <bbriniotis@datumbox.com>
+ */
+public class ThreadMethods {
+
+    /**
+     * Takes the items of the stream in a throttled way and provides them to the 
+     * consumer. It uses as many threads as the available processors and it does
+     * not start more tasks than 2 times the previous number.
+     * 
+     * @param <T>
+     * @param stream
+     * @param consumer 
+     */
+    public static <T> void throttledExecution(Stream<T> stream, Consumer<T> consumer) {
+        int maxThreads = Runtime.getRuntime().availableProcessors();
+        int maxTasks = 2*maxThreads; 
+
+        ExecutorService executorService = Executors.newFixedThreadPool(maxThreads);
+        ThrottledExecutor executor = new ThrottledExecutor(executorService, maxTasks);
+
+        stream.sequential().forEach(i -> {
+            executor.execute(() -> {
+                consumer.accept(i);
+            });
+        });
+
+        executorService.shutdown();
+        try {
+            executorService.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
+        } 
+        catch (InterruptedException ex) {
+            throw new RuntimeException(ex);
+        }
+    }
+
+}

src/main/java/com/datumbox/common/concurrency/ThrottledExecutor.java

@@ -1,73 +1,73 @@
-/**
- * Copyright (C) 2013-2016 Vasilis Vryniotis <bbriniotis@datumbox.com>
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package com.datumbox.common.concurrency;
-
-import java.util.concurrent.Executor;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.Semaphore;
-
-/**
- * The ThrottledExecutor enables us to throttle the input of the executor. This
- * can be useful when we don't wish to submit all the tasks at once in order to
- * preserve memory.
- *
- * @author Vasilis Vryniotis <bbriniotis@datumbox.com>
- */
-public class ThrottledExecutor implements Executor {
-
-    private final Executor wrappedExecutor;
-
-    private final Semaphore semaphore;
-
-    /**
-     * This Executor will block the main thread (when execute() is called) if the 
-     * number of submitted and unfinished tasks reaches the provided limit. This
-     * enabled us to pace the input and reduce memory consumption.
-     * 
-     * @param executor
-     * @param maxConcurrentTasks 
-     */
-    public ThrottledExecutor(Executor executor, int maxConcurrentTasks) {
-        this.wrappedExecutor = executor;
-        this.semaphore = new Semaphore(maxConcurrentTasks);
-    }
-
-    /** {@inheritDoc} */
-    @Override
-    public void execute(final Runnable command) {
-        try {
-            semaphore.acquire();
-        } 
-        catch (InterruptedException ex) {
-            throw new RuntimeException(ex);
-        }
-
-        try {
-            wrappedExecutor.execute(() -> {
-                try {
-                    command.run();
-                } 
-                finally {
-                    semaphore.release();
-                }
-            });
-        } 
-        catch (RejectedExecutionException ex) {
-            semaphore.release();
-            throw new RuntimeException(ex);
-        }
-    }
+/**
+ * Copyright (C) 2013-2016 Vasilis Vryniotis <bbriniotis@datumbox.com>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datumbox.common.concurrency;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.Semaphore;
+
+/**
+ * The ThrottledExecutor enables us to throttle the input of the executor. This
+ * can be useful when we don't wish to submit all the tasks at once in order to
+ * preserve memory.
+ *
+ * @author Vasilis Vryniotis <bbriniotis@datumbox.com>
+ */
+public class ThrottledExecutor implements Executor {
+
+    private final Executor wrappedExecutor;
+
+    private final Semaphore semaphore;
+
+    /**
+     * This Executor will block the main thread (when execute() is called) if the 
+     * number of submitted and unfinished tasks reaches the provided limit. This
+     * enabled us to pace the input and reduce memory consumption.
+     * 
+     * @param executor
+     * @param maxConcurrentTasks 
+     */
+    public ThrottledExecutor(Executor executor, int maxConcurrentTasks) {
+        this.wrappedExecutor = executor;
+        this.semaphore = new Semaphore(maxConcurrentTasks);
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public void execute(final Runnable command) {
+        try {
+            semaphore.acquire();
+        } 
+        catch (InterruptedException ex) {
+            throw new RuntimeException(ex);
+        }
+
+        try {
+            wrappedExecutor.execute(() -> {
+                try {
+                    command.run();
+                } 
+                finally {
+                    semaphore.release();
+                }
+            });
+        } 
+        catch (RejectedExecutionException ex) {
+            semaphore.release();
+            throw new RuntimeException(ex);
+        }
+    }
 }

src/main/java/com/datumbox/common/dataobjects/Dataframe.java

@@ -98,7 +98,7 @@ public static Dataframe parseTextFiles(Map<Object, URI> textFilesMap, AbstractTe
 
                 try (BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(new File(datasetURI)), "UTF8"))) {
                     final int baseCounter = dataset.size(); //because we read multiple files we need to keep track of all records added earlier
-                    ThreadMethods.throttledExecution(StreamMethods.enumerate(br.lines(), false), e -> { //WARNING: Do NOT turn on the parallel flags. Any parallelist is taken care of by the throttledExecution() method
+                    ThreadMethods.throttledExecution(StreamMethods.enumerate(br.lines()), e -> { 
                         Integer rId = baseCounter + e.getKey();
                         String line = e.getValue();
 
@@ -164,7 +164,7 @@ public static Dataframe parseCSVFile(Reader reader, String yVariable, Map<String
                                 .withRecordSeparator(recordSeparator);
 
             try (final CSVParser parser = new CSVParser(reader, format)) { 
-                ThreadMethods.throttledExecution(StreamMethods.enumerate(StreamMethods.stream(parser.spliterator(), false), false), e -> { //WARNING: Do NOT turn on the parallel flags. Any parallelist is taken care of by the throttledExecution() method
+                ThreadMethods.throttledExecution(StreamMethods.enumerate(StreamMethods.stream(parser.spliterator(), false)), e -> { 
                     Integer rId = e.getKey();
                     CSVRecord row = e.getValue();
 
@@ -311,9 +311,9 @@ public boolean contains(Object o) {
     /** {@inheritDoc} */
     @Override
     public boolean addAll(Collection<? extends Record> c) {
-        for(Record r : c) {
+        c.stream().forEach(r -> {
             add(r);
-        }
+        });
         return true;
     }