0
\$\begingroup\$

Hello I have written a multi-threaded implementation of the K-means clustering algorithm. The main goals are correctness and scalable performance on multi-core CPUs. I expect to code to not have race conditions and no data races, and to scale good with more CPU cores.

package bg.unisofia.fmi.rsa;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ParallelKmeans {

    private static CountDownLatch countDownLatch;
    private final int n;
    private final int k;
    public int numThreads = 1;
    List<Node> observations = new ArrayList<>();
    float[][] clusters;

    public ParallelKmeans(int n, int k) {
        this.n = n;
        this.k = k;
        clusters = new float[k][n];
        for (float[] cluster : clusters) {
            for (int i = 0; i < cluster.length; i++) {
                cluster[i] = (float) Math.random();
            }
        }
    }

    public void assignStep(ExecutorService executorService) throws InterruptedException {
        Runnable[] assignWorkers = new AssignWorker[numThreads];
        final int chunk = observations.size() / assignWorkers.length;
        countDownLatch = new CountDownLatch(numThreads);
        for (int j = 0; j < assignWorkers.length; j++) {
            assignWorkers[j] = new AssignWorker(j * chunk, (j + 1) * chunk);
            executorService.execute(assignWorkers[j]);
        }
        countDownLatch.await();

    }

    public void updateStep(ExecutorService executorService) throws InterruptedException {

        countDownLatch = new CountDownLatch(numThreads);

        UpdateWorker[] updateWorkers = new UpdateWorker[numThreads];
        final int chunk = observations.size() / updateWorkers.length;
        for (int j = 0; j < updateWorkers.length; j++) {
            updateWorkers[j] = new UpdateWorker(j * chunk, (j + 1) * chunk);
            executorService.execute(updateWorkers[j]);
        }
        countDownLatch.await();
        clusters = new float[k][n];
        int[] counts = new int[k];

        for (UpdateWorker u : updateWorkers) {
            VectorMath.add(counts, u.getCounts());
            for (int j = 0; j < k; j++) {
                VectorMath.add(clusters[j], u.getClusters()[j]);
            }
        }

        for (int j = 0; j < clusters.length; j++) {
            if (counts[j] != 0) {
                VectorMath.divide(clusters[j], counts[j]);
            }
        }
    }

    void cluster() throws InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
        for (int i = 0; i < 50; i++) {
            assignStep(executorService);
            updateStep(executorService);
        }
        executorService.shutdown();
    }

    public static class Node {
        float[] vec;
        int cluster;
    }

    class AssignWorker implements Runnable {
        int l, r;

        public AssignWorker(int l, int r) {
            this.l = l;
            this.r = r;
        }

        @Override
        public void run() {
            List<Node> chunk = observations.subList(l, r);
            for (Node ob : chunk) {
                float minDist = Float.POSITIVE_INFINITY;
                int idx = 0;
                for (int i = 0; i < clusters.length; i++) {
                    if (minDist > VectorMath.dist(ob.vec, clusters[i])) {
                        minDist = VectorMath.dist(ob.vec, clusters[i]);
                        idx = i;
                    }
                }
                ob.cluster = idx;
            }
            countDownLatch.countDown();
        }
    }

    class UpdateWorker implements Runnable {
        int[] counts;
        int l, r;
        float[][] clusters;

        UpdateWorker(int l, int r) {
            this.l = l;
            this.r = r;
        }

        int[] getCounts() {
            return counts;
        }

        public float[][] getClusters() {
            return clusters;
        }

        @Override
        public void run() {
            this.counts = new int[k];
            this.clusters = new float[k][n];
            for (Node ob : observations.subList(l, r)) {
                VectorMath.add(this.clusters[ob.cluster], ob.vec);
                this.counts[ob.cluster]++;
            }
            countDownLatch.countDown();
        }
    }

}
\$\endgroup\$
4
  • \$\begingroup\$ This code does nothing. observations never gets assigned any values. \$\endgroup\$
    – Eric Stein
    Feb 28 '21 at 21:28
  • \$\begingroup\$ Unless you're assigning it from outside this class before running it, in which case the calling context would be necessary to figure out if it works. \$\endgroup\$
    – Eric Stein
    Mar 1 '21 at 0:36
  • \$\begingroup\$ Of course they are set from otuside. I dont see how tht is relevant to the things I care about. I found my own mistake anyway. I need to use volatile keyword. \$\endgroup\$ Mar 3 '21 at 17:16
  • \$\begingroup\$ please post VectorMath class \$\endgroup\$
    – Flame239
    Mar 5 '21 at 14:32
2
+100
\$\begingroup\$

Interface

Your classes interface is confusing. You have an internal method cluster, which appears to be the main entry point into your ParallelKmeans class. However, it then calls two public methods (assignStep and updateStep) that do the actual work. This seems wrong. Particularly since assignStep and updateStep can't be run safely at the same time.

countDownLatch

You're using a static CoundDownLatch, which you're recreating in your assignStep and updateStep methods. This doesn't really make sense to me. By having it static, you're having it shared across all instances of ParallelKmeans classes. Is this really the expected behaviour? As you're reinitialising the static in both of your public methods, it creates the possibility of it being changed unexpectedly. If you want to stay with a CountDownLatch, consider making it a local variable to each of your public methods and passing it into the constructors for your workers so that they have access to it.

How many threads

You're creating a threadpool that's based around the number of processors the machine has. However, both your update/assign steps are using the member variable numThreads, which is hard coded to 1. This disconnect is strange. Consider changing your startup code to calculate how many threads you want to use, then perform construction and assignment using this number.

\$\endgroup\$
4
  • \$\begingroup\$ I agree about the static thing. It makes absolutely no sense for the latch to be static. \$\endgroup\$ Mar 8 '21 at 2:05
  • \$\begingroup\$ Also agree on the last point. Can you elaborate on the first "Interface" one though? \$\endgroup\$ Mar 8 '21 at 2:07
  • 1
    \$\begingroup\$ @TeodorDyakov When I look at a new class, I start by looking at the methods with a public scope. These are advertising that they are the way a class is expected to be interacted with. I think of this as the 'public interface'. For your class, the public methods look like they aren't designed to be called by a client. The next thing I'll look at is the methods with no defined scope, i.e. package-local visibility. These usually offer a bit more control because there's assumed knowledge from other code in the package, the 'internal interface'. \$\endgroup\$
    – forsvarir
    Mar 8 '21 at 10:00
  • \$\begingroup\$ @TeodorDyakov in your code, cluster seems like it should be public, whereas assignStep and updateStep seems like they should be at least package-local, or more probably private. \$\endgroup\$
    – forsvarir
    Mar 8 '21 at 10:01
0
\$\begingroup\$

I have local variables in the UpdateWorker class that are later accessed from the main Thread. Unless I mark these variables as volatile there may be visibility issues. The same thing also applies to the cluster field in Observation

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.