From a0b799a53133dccd61bd7422a9c11083c11a4ad4 Mon Sep 17 00:00:00 2001 From: Raffi Khatchadourian Date: Wed, 21 Mar 2018 19:15:33 -0400 Subject: [PATCH] Apply optimize stream refactoring. --- src/main/java/org/numenta/nupic/model/Connections.java | 2 +- .../java/org/numenta/nupic/network/sensor/FileSensor.java | 2 +- src/main/java/org/numenta/nupic/util/GroupBy2.java | 2 +- src/main/java/org/numenta/nupic/util/UniversalRandom.java | 2 +- src/test/java/org/numenta/nupic/QuickDayTest.java | 2 +- src/test/java/org/numenta/nupic/RunLayer.java | 6 +++--- 6 files changed, 8 insertions(+), 8 deletions(-) diff --git a/src/main/java/org/numenta/nupic/model/Connections.java b/src/main/java/org/numenta/nupic/model/Connections.java index 78fc101f..e10c7dfb 100644 --- a/src/main/java/org/numenta/nupic/model/Connections.java +++ b/src/main/java/org/numenta/nupic/model/Connections.java @@ -1553,7 +1553,7 @@ public void removeSynapseFromPresynapticMap(Synapse synapse) { * @return Synapse object on the segment with the minimal permanence */ private Synapse minPermanenceSynapse(DistalDendrite dd) { - List synapses = getSynapses(dd).stream().sorted().collect(Collectors.toList()); + List synapses = getSynapses(dd).parallelStream().sorted().collect(Collectors.toList()); Synapse min = null; double minPermanence = Double.MAX_VALUE; diff --git a/src/main/java/org/numenta/nupic/network/sensor/FileSensor.java b/src/main/java/org/numenta/nupic/network/sensor/FileSensor.java index 219940e3..9854f65b 100644 --- a/src/main/java/org/numenta/nupic/network/sensor/FileSensor.java +++ b/src/main/java/org/numenta/nupic/network/sensor/FileSensor.java @@ -157,7 +157,7 @@ public static Stream getJarEntryStream(String path) { innerPath = innerPath.startsWith("!") ? innerPath.substring(1) : innerPath; InputStream inStream = jar.getInputStream(jar.getEntry(innerPath)); BufferedReader br = new BufferedReader(new InputStreamReader(inStream)); - retVal = br.lines().onClose(() -> { + retVal = br.lines().parallel().onClose(() -> { try { br.close(); jar.close(); diff --git a/src/main/java/org/numenta/nupic/util/GroupBy2.java b/src/main/java/org/numenta/nupic/util/GroupBy2.java index fbacfe9e..1b70703e 100644 --- a/src/main/java/org/numenta/nupic/util/GroupBy2.java +++ b/src/main/java/org/numenta/nupic/util/GroupBy2.java @@ -237,7 +237,7 @@ private void advanceSequences() { * @return the next smallest generated key. */ private boolean nextMinKey() { - return Arrays.stream(nextList) + return Arrays.stream(nextList).parallel() .filter(opt -> opt.isPresent()) .map(opt -> opt.get().getSecond()) .min((k, k2) -> k.compareTo(k2)) diff --git a/src/main/java/org/numenta/nupic/util/UniversalRandom.java b/src/main/java/org/numenta/nupic/util/UniversalRandom.java index 91a8e8c5..e48c7dd3 100644 --- a/src/main/java/org/numenta/nupic/util/UniversalRandom.java +++ b/src/main/java/org/numenta/nupic/util/UniversalRandom.java @@ -354,7 +354,7 @@ public static void main(String[] args) { int[] selectedIndices = new int[sampleSize]; List collectedRandoms = new ArrayList<>(); int[] expectedSample = {1,2,3,7,8,9}; - List expectedRandoms = Arrays.stream(new int[] {0,0,0,5,3,3}).boxed().collect(Collectors.toList()); + List expectedRandoms = Arrays.stream(new int[] {0,0,0,5,3,3}).parallel().boxed().collect(Collectors.toList()); random.sampleWithPrintout(choices, selectedIndices, collectedRandoms); System.out.println("samples are equal ? " + Arrays.equals(expectedSample, selectedIndices)); System.out.println("used randoms are equal ? " + collectedRandoms.equals(expectedRandoms)); diff --git a/src/test/java/org/numenta/nupic/QuickDayTest.java b/src/test/java/org/numenta/nupic/QuickDayTest.java index b2320681..cc3f7da6 100644 --- a/src/test/java/org/numenta/nupic/QuickDayTest.java +++ b/src/test/java/org/numenta/nupic/QuickDayTest.java @@ -248,7 +248,7 @@ public void input(Double value, int recordNum, int sequenceNum) { ComputeCycle cc = temporalMemory.compute(memory, input, true); lastPredicted = predictedColumns; predictedColumns = getSDR(cc.predictiveCells()); //Get the predicted column indexes - int[] activeCellIndexes = Connections.asCellIndexes(cc.activeCells()).stream().mapToInt(i -> i).sorted().toArray(); //Get the active cells for classifier input + int[] activeCellIndexes = Connections.asCellIndexes(cc.activeCells()).parallelStream().mapToInt(i -> i).sorted().toArray(); //Get the active cells for classifier input System.out.println("TemporalMemory Input = " + Arrays.toString(input)); System.out.println("TemporalMemory Prediction = " + Arrays.toString(predictedColumns)); diff --git a/src/test/java/org/numenta/nupic/RunLayer.java b/src/test/java/org/numenta/nupic/RunLayer.java index 2d8da612..95c7b820 100644 --- a/src/test/java/org/numenta/nupic/RunLayer.java +++ b/src/test/java/org/numenta/nupic/RunLayer.java @@ -214,7 +214,7 @@ public void networkStep(int[] sparseSPOutput, boolean learn) { public Tuple tmStep(int[] sparseSPOutput, boolean learn, boolean isVerbose) { // Input into the Temporal Memory ComputeCycle cc = tm.compute(connections, sparseSPOutput, learn); - int[] activeCellIndices = cc.activeCells().stream().mapToInt(c -> c.getIndex()).sorted().toArray(); + int[] activeCellIndices = cc.activeCells().parallelStream().mapToInt(c -> c.getIndex()).sorted().toArray(); int[] predColumnIndices = SDR.cellsAsColumnIndices(cc.predictiveCells(), connections.getCellsPerColumn()); int[] activeColumns = Arrays.stream(activeCellIndices) .map(cell -> cell / connections.getCellsPerColumn()) @@ -478,7 +478,7 @@ public static Tuple testTemporalMemory(TemporalMemory tm, Connections conn, int[ } public static void loadSPOutputFile() { - try (Stream stream = Files.lines(Paths.get(MakeshiftLayer.readFile))) { + try (Stream stream = Files.lines(Paths.get(MakeshiftLayer.readFile)).parallel()) { MakeshiftLayer.input = stream.map(l -> { String line = l.replace("[", "").replace("]", "").trim(); int[] result = Arrays.stream(line.split("[\\s]*\\,[\\s]*")).mapToInt(i -> Integer.parseInt(i)).toArray(); @@ -490,7 +490,7 @@ public static void loadSPOutputFile() { } public static void loadRawInputFile() { - try (Stream stream = Files.lines(Paths.get(MakeshiftLayer.INPUT_PATH))) { + try (Stream stream = Files.lines(Paths.get(MakeshiftLayer.INPUT_PATH)).parallel()) { MakeshiftLayer.raw = stream.map(l -> l.trim()).collect(Collectors.toList()); }catch(Exception e) {e.printStackTrace();} }