Unable to multi-thread a scalable method

Question

UPDATE: To help clarify what I'm asking I have posted a little java code that gets the idea across.

A while ago I asked a question on how to get an algorithm to break down a set of numbers, the idea was to give it a list of numbers (1,2,3,4,5) and a total(10) and it would figure out all the multiples of each number that would add up to the total('1*10' or '1*1,1*2,1*3,1*4' or '2*5',etc..). It was the first programming exercise I ever did so it took me a while and I got it working but now I want to try to see if I can scale it. The person in the original question said it was scalable but I'm a bit confused at how to do it. The recursive part is the area I'm stuck at scaling the part that combines all the results(the table it is referring to is not scalable but applying caching I am able to make it fast)

I have the following algorithm(pseudo code):

//generates table
for i = 1 to k
    for z = 0 to sum:
        for c = 1 to z / x_i:
            if T[z - c * x_i][i - 1] is true:
                set T[z][i] to true

//uses table to bring all the parts together
function RecursivelyListAllThatWork(k, sum) // Using last k variables, make sum
    /* Base case: If we've assigned all the variables correctly, list this
     * solution.
     */
    if k == 0:
        print what we have so far
        return

    /* Recursive step: Try all coefficients, but only if they work. */
    for c = 0 to sum / x_k:
       if T[sum - c * x_k][k - 1] is true:
           mark the coefficient of x_k to be c
           call RecursivelyListAllThatWork(k - 1, sum - c * x_k)
           unmark the coefficient of x_k

I'm really at a loss at how to thread/multiprocess the RecursivelyListAllThatWork function. I know if I send it a smaller K( which is int of total number of items in list) it will process that subset but I don't know how to do ones that combine results across the subset. For example, if list is [1,2,3,4,5,6,7,8,9,10] and I send it K=3 then only the 1,2,3 get processed which is fine but what about if I need results that include 1 and 10? I have tried to modify the table(variable T) so only the subset I want are there but still doesn't work because, like the solution above, it does a subset but cannot process answers that require a wider range.

I don't need any code just if someone can explain how to conceptually break this recursive step to so other cores/machines can be used.

UPDATE: I still can't seem to figure out how to turn RecursivelyListAllThatWork into a runnable(I know technically how to do it, but I don't understand how to change the RecursivelyListAllThatWork algorithm so it can be ran in parallel. The other parts are just here to make the example work, I only need to implement runnable on RecursivelyListAllThatWork method). Here's the java code:

import java.awt.Point;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;


public class main
{
    public static void main(String[] args)
    {
        System.out.println("starting..");
        int target_sum = 100;
        int[] data = new int[] { 10, 5, 50, 20, 25, 40 };
        List T = tableGeneator(target_sum, data);
        List<Integer> coeff = create_coeff(data.length);
        RecursivelyListAllThatWork(data.length, target_sum, T, coeff, data);
    }

    private static List<Integer> create_coeff(int i) {
        // TODO Auto-generated method stub
        Integer[] integers = new Integer[i];
        Arrays.fill(integers, 0);
        List<Integer> integerList = Arrays.asList(integers);
        return integerList;
    }


    private static void RecursivelyListAllThatWork(int k, int sum, List T, List<Integer> coeff, int[] data) {
        // TODO Auto-generated method stub
        if (k == 0) {
            //# print what we have so far
            for (int i = 0; i < coeff.size(); i++) {
                System.out.println(data[i] + " = " + coeff.get(i));
            }

            System.out.println("*******************");
            return;
        }

        Integer x_k = data[k-1];
        //  Recursive step: Try all coefficients, but only if they work. 
        for (int c = 0; c <= sum/x_k; c++) { //the c variable caps the percent
            if (T.contains(new Point((sum - c * x_k), (k-1))))
            {
                    // mark the coefficient of x_k to be c
                    coeff.set((k-1), c);
                    RecursivelyListAllThatWork((k - 1), (sum - c * x_k), T, coeff, data);
                    // unmark the coefficient of x_k
                    coeff.set((k-1), 0);
            }

        }

    }

    public static List tableGeneator(int target_sum, int[] data) {
        List T = new ArrayList();
        T.add(new Point(0, 0));

        float max_percent = 1;
        int R = (int) (target_sum * max_percent * data.length);
        for (int i = 0; i < data.length; i++)
        {
            for (int s = -R; s < R + 1; s++)
            {
                int max_value = (int) Math.abs((target_sum * max_percent)
                        / data[i]);
                for (int c = 0; c < max_value + 1; c++)
                {
                    if (T.contains(new Point(s - c * data[i], i)))
                    {
                        Point p = new Point(s, i + 1);
                        if (!T.contains(p))
                        {
                            T.add(p);
                        }
                    }
                }
            }
        }
        return T;
    }
} 

Answer 1

The general answer to multi-threading is to de-recursivate a recursive implementation thanks to a stack (LIFO or FIFO). When implementing such an algorithm, the number of threads is a fixed parameter for the algorithm (number of cores for instance).

To implement it, the language call stack is replaced by a stack storing last context as a checkpoint when the tested condition ends the recursivity. In your case it is either k=0 or coeff values matchs targeted sum.

After de-recursivation, a first implementation is to run multiple threads to consume the stack BUT the stack access becomes a contention point because it may require synchronization.

A better scalable solution is to dedicate a stack for each thread but an initial production of contexts in the stack is required.

I propose a mix approach with a first thread working recursively for a limited number of k as a maximum recursion depth: 2 for the small data set in example, but I recommend 3 if larger. Then this first part delegates the generated intermediate contexts to a pool of threads which will process remaining k with a non-recursive implementation. This code is not based on the complex algorithm you use but on a rather "basic" implementation:

import java.util.Arrays;
import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class MixedParallel
{
    // pre-requisite: sorted values !!
    private static final int[] data = new int[] { 5, 10, 20, 25, 40, 50 };

    // Context to store intermediate computation or a solution
    static class Context {
        int k;
        int sum;
        int[] coeff;
        Context(int k, int sum, int[] coeff) {
            this.k = k;
            this.sum = sum;
            this.coeff = coeff;
        }
    }

    // Thread pool for parallel execution
    private static ExecutorService executor;
    // Queue to collect solutions
    private static Queue<Context> solutions;

    static {
        final int numberOfThreads = 2;
        executor =
            new ThreadPoolExecutor(numberOfThreads, numberOfThreads, 1000, TimeUnit.SECONDS,
                                   new LinkedBlockingDeque<Runnable>());
        // concurrent because of multi-threaded insertions
        solutions = new ConcurrentLinkedQueue<Context>();
    }


    public static void main(String[] args)
    {
        int target_sum = 100;
        // result vector, init to 0
        int[] coeff = new int[data.length];
        Arrays.fill(coeff, 0);
        mixedPartialSum(data.length - 1, target_sum, coeff);

        executor.shutdown();
        // System.out.println("Over. Dumping results");
        while(!solutions.isEmpty()) {
            Context s = solutions.poll();
            printResult(s.coeff);
        }
    }

    private static void printResult(int[] coeff) {
        StringBuffer sb = new StringBuffer();
        for (int i = coeff.length - 1; i >= 0; i--) {
            if (coeff[i] > 0) {
                sb.append(data[i]).append(" * ").append(coeff[i]).append("   ");
            }
        }
        System.out.println(sb.append("from ").append(Thread.currentThread()));
    }

    private static void mixedPartialSum(int k, int sum, int[] coeff) {
        int x_k = data[k];
        for (int c = sum / x_k; c >= 0; c--) {
            coeff[k] = c;
            int[] newcoeff = Arrays.copyOf(coeff, coeff.length);
            if (c * x_k == sum) {
                //printResult(newcoeff);
                solutions.add(new Context(0, 0, newcoeff));
                continue;
            } else if (k > 0) {
                if (data.length - k < 2) {
                    mixedPartialSum(k - 1, sum - c * x_k, newcoeff);
                    // for loop on "c" goes on with previous coeff content
                } else {
                    // no longer recursive. delegate to thread pool
                    executor.submit(new ComputePartialSum(new Context(k - 1, sum - c * x_k, newcoeff)));
                }
            }
        }
    }

    static class ComputePartialSum implements Callable<Void> {
        // queue with contexts to process
        private Queue<Context> contexts;

        ComputePartialSum(Context request) {
            contexts = new ArrayDeque<Context>();
            contexts.add(request);
        }

        public Void call() {
            while(!contexts.isEmpty()) {
                Context current = contexts.poll();
                int x_k = data[current.k];
                for (int c = current.sum / x_k; c >= 0; c--) {
                    current.coeff[current.k] = c;
                    int[] newcoeff = Arrays.copyOf(current.coeff, current.coeff.length);
                    if (c * x_k == current.sum) {
                        //printResult(newcoeff);
                        solutions.add(new Context(0, 0, newcoeff));
                        continue;
                    } else if (current.k > 0) {
                        contexts.add(new Context(current.k - 1, current.sum - c * x_k, newcoeff));
                    }
                }
            }
            return null;
        }
    }
}

You can check which thread has found outputted result and check all are involed: the main thread in recursive mode and the two thread from the pool in context stack mode.

Now this implementation is scalable when data.length is high:

• the maximum recursion depth is limited to the main thread at a low level
• each thread from the pool works with its own context stack without contention with others
• the parameters to tune now are numberOfThreads and maxRecursionDepth

So the answer is yes, your algorithm can be parallelized. Here is a fully recursive implementation based on your code:

import java.awt.Point;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class OriginalParallel
{
    static final int numberOfThreads = 2;
    static final int maxRecursionDepth = 3;

    public static void main(String[] args)
    {
        int target_sum = 100;
        int[] data = new int[] { 50, 40, 25, 20, 10, 5 };
        List T = tableGeneator(target_sum, data);
        int[] coeff = new int[data.length];
        Arrays.fill(coeff, 0);
        RecursivelyListAllThatWork(data.length, target_sum, T, coeff, data);
        executor.shutdown();
    }

    private static void printResult(int[] coeff, int[] data) {
        StringBuffer sb = new StringBuffer();
        for (int i = coeff.length - 1; i >= 0; i--) {
            if (coeff[i] > 0) {
                sb.append(data[i]).append(" * ").append(coeff[i]).append("   ");
            }
        }
        System.out.println(sb.append("from ").append(Thread.currentThread()));
    }

    // Thread pool for parallel execution
    private static ExecutorService executor;
    static {
        executor =
            new ThreadPoolExecutor(numberOfThreads, numberOfThreads, 1000, TimeUnit.SECONDS,
                                   new LinkedBlockingDeque<Runnable>());
    }

    private static void RecursivelyListAllThatWork(int k, int sum, List T, int[] coeff, int[] data) {
        if (k == 0) {
            printResult(coeff, data);
            return;
        }
        Integer x_k = data[k-1];
        //  Recursive step: Try all coefficients, but only if they work. 
        for (int c = 0; c <= sum/x_k; c++) { //the c variable caps the percent
            if (T.contains(new Point((sum - c * x_k), (k-1)))) {
                    // mark the coefficient of x_k to be c
                    coeff[k-1] = c;
                    if (data.length - k != maxRecursionDepth) {
                        RecursivelyListAllThatWork((k - 1), (sum - c * x_k), T, coeff, data);
                    } else {
                        // delegate to thread pool when reaching depth 3
                        int[] newcoeff = Arrays.copyOf(coeff, coeff.length);
                        executor.submit(new RecursiveThread(k - 1, sum - c * x_k, T, newcoeff, data)); 
                    }
                    // unmark the coefficient of x_k
                    coeff[k-1] = 0;
            }
        }
    }

    static class RecursiveThread implements Callable<Void> {
        int k;
        int sum;
        int[] coeff;
        int[] data;
        List T;

        RecursiveThread(int k, int sum, List T, int[] coeff, int[] data) {
            this.k = k;
            this.sum = sum;
            this.T = T;
            this.coeff = coeff;
            this.data = data;
            System.out.println("New job for k=" + k);
        }

        public Void call() {
            RecursivelyListAllThatWork(k, sum, T, coeff, data);
            return null;
        }
    }

    public static List tableGeneator(int target_sum, int[] data) {
        List T = new ArrayList();
        T.add(new Point(0, 0));

        float max_percent = 1;
        int R = (int) (target_sum * max_percent * data.length);
        for (int i = 0; i < data.length; i++) {
            for (int s = -R; s < R + 1; s++) {
                int max_value = (int) Math.abs((target_sum * max_percent) / data[i]);
                for (int c = 0; c < max_value + 1; c++) {
                    if (T.contains(new Point(s - c * data[i], i))) {
                        Point p = new Point(s, i + 1);
                        if (!T.contains(p)) {
                            T.add(p);
                        }
                    }
                }
            }
        }
        return T;
    }
}

Answer 2

1) Instead of

if k == 0:
    print what we have so far
    return

you can check to see how many coefficients are non-zero; if that count is greater than a certain threshold (3 in your example), then just don't print it. (Hint: this would be closely related to the

mark the coefficient of x_k to be c

line.)

2) Recursive functions are generally exponential in nature, which means that as you scale higher, the runtime will grow sharply larger.

With that in mind, you can apply multithreading to both calculating the table and the recursive function.

When considering the table, think about which parts of the loop affect each other and must be done in sequence; the converse, of course, is finding which parts don't affect each other and can be run in parallel.

As for the recursive function, your best bet would probably be to apply the multithreading to the branching part.

Answer 3

They key to making this multithreaded is just to make sure that you don't have unnecessary global data structures, like your "marks" on the coefficients.

Let's say you have K numbers n[0] ... n[K-1] in your table and the sum you want to reach is S. I assume below that the array n[] is sorted from smallest to largest number.

A simple enumeration algorithm is here. i is index to the list of numbers, s is the current sum already built, and cs is a list of coefficients for the numbers 0 .. i - 1:

function enumerate(i, s, cs):
  if (s == S):
     output_solution(cs)
  else if (i == K):
     return // dead end
  else if ((S - s) < n[i]):
     return // no solution can be found
  else:
     for c in 0 .. floor((S - s) / n[i]): // note: floor(...) > 0
        enumerate(i + 1, s + c * n[i], append(cs, c))

To run the process:

 enumerate(0, 0, make_empty_list())

Now here are no global data structures anymore, except the table n[] (constant data), and 'enumerate' also does not return anything, so you can change the recursive call to run in its own thread at your will. E.g. you can spawn a new thread to a recursive enumerate() call unless you have too many threads running already, in which case you wait.