Executive Service and Runnable

Question

I have a list containing the data to calculate for each pixel (e.g. list size = 1024x768). Now I want to repeat the multithreading in the list and save the calculation for each pixel in the HashMap. But no matter what I do, I cannot do everything right. I tried several ways, my last was the following:

ConcurrentMap<T, Color> map = new ConcurrentHashMap<T, Color>(); ExecutorService pool = Executors.newFixedThreadPool(4); Iterator<T> it = camera.iterator(); while (it.hasNext()) { Runnable run = () -> { int i = 0; while (it.hasNext() && i < 1000) { i++; T cameraRay = it.next(); if (object.collide(cameraRay.getRay()) == null) map.put(cameraRay, BG_COLOR); else map.put(cameraRay, this.shader.shade(cameraRay.getRay(), object.collide(cameraRay.getRay())).getColor()); } }; pool.execute(run); } pool.shutdown(); try { if (pool.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS)) { System.out.println("Mapsize: " + map.size()); // Draw Image: map.forEach((ray, color) -> {image.setColor(ray, color);}); } } catch (InterruptedException e) { e.printStackTrace(); } 

Note that hasNext() iterators hasNext() synchronized. The problem is sometimes related to the heap problem or simply that the HashMap is smaller than the size of the list.

I assume that I did not understand something about the proper execution of Runnables or ExecutorService.

I appreciate any help here.

EDIT: I added System.out.println(i) immediately before the i++ statement. Despite the fact that at some point i < 1000 unexpectedly detected, the following appears:

 507 169 86624 625 626 Exception in thread "pool-2-thread-2" java.lang.OutOfMemoryError: Java heap space Exception in thread "pool-2-thread-3" java.lang.reflect.InvocationTargetException at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) at java.lang.reflect.Method.invoke(Unknown Source) at com.sun.javafx.application.LauncherImpl.launchApplicationWithArgs(Unknown Source) at com.sun.javafx.application.LauncherImpl.launchApplication(Unknown Source) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) at java.lang.reflect.Method.invoke(Unknown Source) at sun.launcher.LauncherHelper$FXHelper.main(Unknown Source) Caused by: java.lang.OutOfMemoryError: Java heap space at java.util.concurrent.ThreadPoolExecutor.addWorker(Unknown Source) at java.util.concurrent.ThreadPoolExecutor.execute(Unknown Source) at raytracer.impl.ParallelRenderer.render(ParallelRenderer.java:78) at raytracer.ImageViewer.main(ImageViewer.java:118) ... 11 more Exception in thread "pool-2-thread-4" java.lang.OutOfMemoryError: Java heap space java.lang.OutOfMemoryError: Java heap space at raytracer.impl.TriangleImpl.collide(TriangleImpl.java:87) at raytracer.impl.SimpleScene.collide(SimpleScene.java:27) at raytracer.impl.ParallelRenderer.lambda$0(ParallelRenderer.java:71) at raytracer.impl.ParallelRenderer$$Lambda$48/24559708.run(Unknown Source) at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) at java.lang.Thread.run(Unknown Source) 

EDIT 2: According to Warst's answer, I tried the following

 Iterator<T> it = camera.iterator(); List<T> buffer = new ArrayList<T>(1000); while (it.hasNext()) { buffer.add(it.next()); if (buffer.size() >= 1000 || !it.hasNext()) { Runnable run = () -> { for (T cameraRay : buffer) { if (object.collide(cameraRay.getRay()) == null) // No collision map.put(cameraRay, BG_COLOR); else map.put(cameraRay, this.shader.shade(cameraRay.getRay(), object.collide(cameraRay.getRay())).getColor()); } }; pool.execute(run); buffer.clear(); } } 

But it is very strange that the Runnable block is never entered, why?

Answer 1

What bothers me is that your runnables use the same iterator. What surprises me even more is that you run executable files, iterate over the iterator, but these tags also control the iterator. This code can (and will, as your question has proved) lead to a bunch of race conditions and subsequent headaches.

I would suggest the following:

• Get camera iterator
• Make an empty buffer
• Read the first x (e.g. 1000) samples from the iterator to the buffer
• Create a runnable with a buffer that will work with thousands of records
• Send runnable to the service and go back to 2. Repeat until the iterator is next.

Assuming your data processing is (significantly) slower than repeating over the camera, this should do the trick. If this is not the case, there is no reason for multithreading.

update 2

I updated my sample code to what works:

 public static void main(String[] args) throws InterruptedException { ConcurrentHashMap<Integer, String> map = new ConcurrentHashMap<>(); ExecutorService pool = Executors.newFixedThreadPool(4); Iterator<Integer> it = getIt(); Task t = new Task(map); while (it.hasNext()) { t.add(it.next()); if (t.size()>=1000 || !it.hasNext()) { pool.submit(t); t = new Task(map); } } pool.shutdown(); pool.awaitTermination(1, TimeUnit.DAYS); // Breakpoint here to inspect map System.out.println("Done!"); } 

FROM

 private static Iterator<Integer> getIt(){ return new Iterator<Integer>() { private int nr = 0; @Override public boolean hasNext() { return nr < 20000; } @Override public Integer next() { return nr++; } }; } 

AND

 private static class Task extends ArrayList<Integer> implements Runnable{ private final ConcurrentHashMap<Integer, String> map; public Task(ConcurrentHashMap<Integer, String> map) { this.map = map; } @Override public void run() { try{ for (Integer i : this) { // Simulate some crunching: write to the stdout in 10 iterations for each number: 10 000 prints for each Runnable for (int j = 0; j < 10; j++) { System.out.println("Iteration "+j+" for "+i); } // Store something in the map, namely that this Integer, or T in your case, has been processed map.put(i, "Done"); } } catch(Exception e){ e.printStackTrace(); } } } 

The breakpoint hits after 20-30 seconds, and the card contains all integers paired with the line "Finish".