All geek questions in one place

Parallel computing for an element Iterator in Java

I had the same need several times, and I wanted to get other thoughts on the right way to structure the solution. It is necessary to perform some operation on many elements in many threads, without immediately requiring all the elements in memory, but only those that are under calculation. As in Iterables.partition, it’s not enough, because it displays all the elements in memory in front.

Expressing this in code, I want to write BulkCalc2, which does the same as BulkCalc1, in parallel. Below is an example code that illustrates my best attempt. I am not satisfied because it is big and ugly, but it seems to fulfill my tasks of maintaining threads, which will be used until the work is done, spreading any exceptions in the calculation and having no more NumThreads BigThing instances are required in memory immediately.

I agree with the answer that meets the set goals in the most concise way, be it a way to improve my BulkCalc2 or a completely different solution.

interface BigThing {

    int getId();

    String getString();
}

class Calc {

    // somewhat expensive computation
    double calc(BigThing bigThing) {
        Random r = new Random(bigThing.getString().hashCode());
        double d = 0;
        for (int i = 0; i < 100000; i++) {
            d += r.nextDouble();
        }
        return d;
    }
}

class BulkCalc1 {

    final Calc calc;

    public BulkCalc1(Calc calc) {
        this.calc = calc;
    }

    public TreeMap<Integer, Double> calc(Iterator<BigThing> in) {
        TreeMap<Integer, Double> results = Maps.newTreeMap();
        while (in.hasNext()) {
            BigThing o = in.next();
            results.put(o.getId(), calc.calc(o));
        }
        return results;
    }
}

class SafeIterator<T> {

    final Iterator<T> in;

    SafeIterator(Iterator<T> in) {
        this.in = in;
    }

    synchronized T nextOrNull() {
        if (in.hasNext()) {
            return in.next();
        }
        return null;
    }
}

class BulkCalc2 {

    final Calc calc;
    final int numThreads;

    public BulkCalc2(Calc calc, int numThreads) {
        this.calc = calc;
        this.numThreads = numThreads;
    }

    public TreeMap<Integer, Double> calc(Iterator<BigThing> in) {
        ExecutorService e = Executors.newFixedThreadPool(numThreads);
        List<Future<?>> futures = Lists.newLinkedList();

        final Map<Integer, Double> results = new MapMaker().concurrencyLevel(numThreads).makeMap();
        final SafeIterator<BigThing> it = new SafeIterator<BigThing>(in);
        for (int i = 0; i < numThreads; i++) {
            futures.add(e.submit(new Runnable() {

                @Override
                public void run() {
                    while (true) {
                        BigThing o = it.nextOrNull();
                        if (o == null) {
                            return;
                        }
                        results.put(o.getId(), calc.calc(o));
                    }
                }
            }));
        }

        e.shutdown();

        for (Future<?> future : futures) {
            try {
                future.get();
            } catch (InterruptedException ex) {
                // swallowing is OK
            } catch (ExecutionException ex) {
                throw Throwables.propagate(ex.getCause());
            }
        }

        return new TreeMap<Integer, Double>(results);
    }
}
+3
source share
4 answers

, , , . BulkCalc3 .

class BulkCalc3
{
    final Calc calc;

    public BulkCalc3(Calc calc)
    {
        this.calc = calc;
    }

    public TreeMap<Integer, Double> calc(Iterator<BigThing> in)
    {
        final ConcurrentMap<Integer, Double> resultMap = new MapMaker().makeMap();
        ThreadedIteratorProcessor<BigThing> processor = new ThreadedIteratorProcessor<BigThing>();
        processor.processIterator(in, new ThreadedIteratorProcessor.ElementProcessor<BigThing>()
        {
            @Override
            public void processElement(BigThing o)
            {
                resultMap.put(o.getId(), calc.calc(o));
            }
        });
        return new TreeMap<Integer, Double>(resultMap);
    }
}

:

import com.google.common.collect.Lists;
import com.google.common.util.concurrent.MoreExecutors;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * A utility class to process each element in an iterator in an efficient manner.
 */
public class ThreadedIteratorProcessor<T>
{
    public static interface ElementProcessor<T>
    {
        /**
         * Process an element.
         * @param element The element to process.
         */
        public void processElement(T element);
    }
    private final int numThreads;

    /**
     * Create an instance which uses a specified number of threads.
     * @param numThreads The number of processing threads.
     */
    public ThreadedIteratorProcessor(int numThreads)
    {
        this.numThreads = numThreads;
    }

    /**
     * Create an instance which uses a number of threads equal to the number of system processors.
     */
    public ThreadedIteratorProcessor()
    {
        this(Runtime.getRuntime().availableProcessors());
    }

    /**
     * Process each element in an iterator in parallel.  The number of worker threads depends on how this object was
     * constructed.  This method will re-throw any exception thrown in the supplied ElementProcessor.  An element will
     * not be requested from the iterator any earlier than is absolutely necessary.  In other words, the last element in
     * the iterator will not be consumed until all of the other elements are completely processed, excluding elements
     * currently being processed by the worker threads.
     * @param iterator The iterator from which to get elements.  This iterator need not be thread-safe.
     * @param elementProcessor The element processor.
     */
    public void processIterator(Iterator<T> iterator, ElementProcessor<T> elementProcessor)
    {
        // Use an ExecutorService for proper exception handling.
        ExecutorService e = Executors.newFixedThreadPool(numThreads, MoreExecutors.daemonThreadFactory());
        List<Future<?>> futures = Lists.newLinkedList();

        // Get a thread-safe iterator
        final SafeIterator<T> safeIterator = new SafeIterator<T>(iterator);

        // Submit numThreads new worker threads to pull work from the iterator.
        for (int i = 0; i < numThreads; i++)
        {
            futures.add(e.submit(new Consumer<T>(safeIterator, elementProcessor)));
        }

        e.shutdown();

        // Calling .get() on the futures accomplishes two things:
        // 1. awaiting completion of the work
        // 2. discovering an exception during calculation, and rethrowing to the client in this thread.
        for (Future<?> future : futures)
        {
            try
            {
                future.get();
            }
            catch (InterruptedException ex)
            {
                // swallowing is OK
            }
            catch (ExecutionException ex)
            {
                // Re-throw the underlying exception to the client.
                throw Throwables.propagate(ex.getCause());
            }
        }
    }

    // A runnable that sits in a loop consuming and processing elements from an iterator.
    private static class Consumer<T> implements Runnable
    {
        private final SafeIterator<T> it;
        private final ElementProcessor<T> elementProcessor;

        public Consumer(SafeIterator<T> it, ElementProcessor<T> elementProcessor)
        {
            this.it = it;
            this.elementProcessor = elementProcessor;
        }

        @Override
        public void run()
        {
            while (true)
            {
                T o = it.nextOrNull();
                if (o == null)
                {
                    return;
                }
                elementProcessor.processElement(o);
            }
        }
    }

    // a thread-safe iterator-like object.
    private static class SafeIterator<T>
    {
        private final Iterator<T> in;

        SafeIterator(Iterator<T> in)
        {
            this.in = in;
        }

        synchronized T nextOrNull()
        {
            if (in.hasNext())
            {
                return in.next();
            }
            return null;
        }
    }
}
-1

, : (, , )

33 , . - , ( ) ( ) . , , . , , .

/** More succinct */
public static Map<Integer, Double> bulkCalcSuccincter(final Iterator<BigThing> it, final Calc calc, final int numThreads) {
    final ConcurrentHashMap<Integer, Double> results = new ConcurrentHashMap<Integer, Double>();
    final java.util.List<Future> futures = new ArrayList<Future>();
    final ExecutorService e = Executors.newFixedThreadPool(numThreads);

    for (int i = 0; i < numThreads; i++) {
        futures.add(e.submit(new Runnable() {
            public void run() {
                while (true) {
                    BigThing thing = null;
                    synchronized (it) {
                        thing = (it.hasNext()) ? it.next() : null;
                    }
                    if (thing == null) {
                        break;
                    }
                    results.put(thing.getId(), calc.calc(thing));
                }
            }
        }));
    }
    e.shutdown();

    for (Future f : futures) {
        try {
            f.get();
        } catch (InterruptedException ex) {
        // swallowing is better than spitting it out
        } catch (ExecutionException ex) {
            throw Throwables.propagate(ex.getCause());
        }
    }
    return results;
}
+2

: ,

: , . BulkCalcRunner.runBulkCalc(Iterator, Calc) . , , , , .

:

  • - HashMap - . . ( HashMaps L2 , ).
  • HashMap .

  • Errors are combined into a collection for further processing. With a thread pool, each Exception requires a thread to die and recreate

    BigThing interface {int getId (); String getString (); }

    class Calc {
    // somewhat expensive computation
    double calc(BigThing bigThing) {
        Random r = new Random(bigThing.getString().hashCode());
        double d = 0;
        for (int i = 0; i < 100000; i++) {
            d += r.nextDouble();
        }
        return d;
    }
}

static class BulkCalcRunner implements Runnable {
    Calc calc;
    CountDownLatch latch;
    Iterator<BigThing> it;
    Collection<Throwable> errors;
    Map<Integer,Double> results;

    public BulkCalcRunner (Calc calc, Iterator<BigThing> it, CountDownLatch latch, Map<Integer,Double> results, Collection<Throwable> errors) {
        this.calc = calc;
        this.latch = latch;
        this.errors = errors;
        this.results = results;
    }

    public void run() {
        ArrayList<Throwable> errorLocal = new ArrayList<Throwable>();
        HashMap<Integer,Double> resultsLocal = new HashMap<Integer,Double>();
        while (true) {
            BigThing thing = null;
            try {
                synchronized (it) {
                    if (it.hasNext()) {
                        thing = it.next();
                    }
                }
            } catch (Exception e) { //prevents iterator errors from causing endless loop
                thing = null;
            }
            //finished when first null BigThing encountered
            if (thing == null) {
                synchronized (errors) {
                    errors.addAll(errorLocal);
                }
                synchronized(results) {
                    results.putAll(resultsLocal);
                }
                latch.countDown();
                break;
            }
            try {
                resultsLocal.put(thing.getId(), calc.calc(thing));
            } catch (Exception e) {
                errorLocal.add(e);
            }
        }
    }

    public static Map<Integer,Double> runBulkCalc(Iterator<BigThing> iterator, Calc calculation, int numThreads) {
        final ConcurrentHashMap<Integer, Double> results = new ConcurrentHashMap<Integer, Double>();
        final ArrayList<Throwable> errors = new ArrayList<Throwable>();
        final CountDownLatch latch = new CountDownLatch(numThreads);

        //start up the worker threads
        for (int i = 0; i < numThreads; i++) {
            new Thread(new BulkCalcRunner(calculation,iterator,latch, results, errors)).start();
        }

        try {
            //Latch waits for all the worker threads to check in as "done"
            latch.await();
        } catch (InterruptedException ex) {
            // swallowing is better than spitting it out...
        }

        //finally, propagate errors!
        for (Throwable th : errors) {
            throw Throwables.propagate(th.getCause());
        }
        return results;
    }

    public static Map<Integer,Double> runBulkCalc(Iterator<BigThing> iterator, Calc calculation) {
        return runBulkCalc(iterator,calculation,Runtime.getRuntime().availableProcessors());
    }
}
+1
source

Source: https://habr.com/ru/post/1726115/

More articles:


All Articles