Add clean-room implementation of Executor framework and integration tests

Implemented a clean-room, minimal version of java.util.concurrent Executor framework (Executor, ExecutorService, Future, Executors, FutureTask, LinkedBlockingQueue) in vm/JavaAPI. This implementation avoids Oracle-copyrighted code and complex dependencies to facilitate usage in the ParparVM native test environment.

Added ExecutorIntegrationTest in vm/tests to verify the framework compiles and executes correctly when translated to C. The test includes extensive native stubs to support the minimal runtime required for execution.

Author: google-labs-jules[bot]

vm/JavaAPI/src/java/util/concurrent/AbstractExecutorService.java

@@ -0,0 +1,36 @@
+package java.util.concurrent;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public abstract class AbstractExecutorService implements ExecutorService {
+
+    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
+        return new FutureTask<T>(runnable, value);
+    }
+
+    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
+        return new FutureTask<T>(callable);
+    }
+
+    public Future<?> submit(Runnable task) {
+        if (task == null) throw new NullPointerException();
+        RunnableFuture<Void> ftask = newTaskFor(task, null);
+        execute(ftask);
+        return ftask;
+    }
+
+    public <T> Future<T> submit(Runnable task, T result) {
+        if (task == null) throw new NullPointerException();
+        RunnableFuture<T> ftask = newTaskFor(task, result);
+        execute(ftask);
+        return ftask;
+    }
+
+    public <T> Future<T> submit(Callable<T> task) {
+        if (task == null) throw new NullPointerException();
+        RunnableFuture<T> ftask = newTaskFor(task);
+        execute(ftask);
+        return ftask;
+    }
+}

vm/JavaAPI/src/java/util/concurrent/BlockingQueue.java

@@ -0,0 +1,16 @@
+package java.util.concurrent;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Queue;
+
+public interface BlockingQueue<E> extends Queue<E> {
+    void put(E e) throws InterruptedException;
+    boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException;
+    E take() throws InterruptedException;
+    E poll(long timeout, TimeUnit unit) throws InterruptedException;
+    int remainingCapacity();
+    int drainTo(Collection<? super E> c);
+    int drainTo(Collection<? super E> c, int maxElements);
+}

vm/JavaAPI/src/java/util/concurrent/Callable.java

@@ -0,0 +1,5 @@
+package java.util.concurrent;
+
+public interface Callable<V> {
+    V call() throws Exception;
+}

vm/JavaAPI/src/java/util/concurrent/CancellationException.java

@@ -0,0 +1,6 @@
+package java.util.concurrent;
+
+public class CancellationException extends IllegalStateException {
+    public CancellationException() { }
+    public CancellationException(String message) { super(message); }
+}

vm/JavaAPI/src/java/util/concurrent/ExecutionException.java

@@ -0,0 +1,8 @@
+package java.util.concurrent;
+
+public class ExecutionException extends Exception {
+    public ExecutionException() { }
+    public ExecutionException(String message) { super(message); }
+    public ExecutionException(String message, Throwable cause) { super(message, cause); }
+    public ExecutionException(Throwable cause) { super(cause); }
+}

vm/JavaAPI/src/java/util/concurrent/Executor.java

@@ -0,0 +1,5 @@
+package java.util.concurrent;
+
+public interface Executor {
+    void execute(Runnable command);
+}

vm/JavaAPI/src/java/util/concurrent/ExecutorCompletionService.java

@@ -0,0 +1,48 @@
+package java.util.concurrent;
+
+public class ExecutorCompletionService<V> {
+    private final Executor executor;
+    private final BlockingQueue<Future<V>> completionQueue;
+
+    public ExecutorCompletionService(Executor executor) {
+        this(executor, new LinkedBlockingQueue<Future<V>>());
+    }
+
+    public ExecutorCompletionService(Executor executor, BlockingQueue<Future<V>> completionQueue) {
+        if (executor == null || completionQueue == null) throw new NullPointerException();
+        this.executor = executor;
+        this.completionQueue = completionQueue;
+    }
+
+    public Future<V> submit(Callable<V> task) {
+        if (task == null) throw new NullPointerException();
+        RunnableFuture<V> f = new QueueingFuture(task);
+        executor.execute(f);
+        return f;
+    }
+
+    public Future<V> submit(Runnable task, V result) {
+        if (task == null) throw new NullPointerException();
+        RunnableFuture<V> f = new QueueingFuture(task, result);
+        executor.execute(f);
+        return f;
+    }
+
+    public Future<V> take() throws InterruptedException {
+        return completionQueue.take();
+    }
+
+    public Future<V> poll() {
+        return completionQueue.poll();
+    }
+
+    public Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException {
+        return completionQueue.poll(timeout, unit);
+    }
+
+    private class QueueingFuture extends FutureTask<V> {
+        QueueingFuture(Callable<V> c) { super(c); }
+        QueueingFuture(Runnable t, V r) { super(t, r); }
+        protected void done() { completionQueue.add(this); }
+    }
+}

vm/JavaAPI/src/java/util/concurrent/ExecutorService.java

@@ -0,0 +1,20 @@
+package java.util.concurrent;
+
+import java.util.Collection;
+import java.util.List;
+
+public interface ExecutorService extends Executor {
+    void shutdown();
+    List<Runnable> shutdownNow();
+    boolean isShutdown();
+    boolean isTerminated();
+    boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
+    <T> Future<T> submit(Callable<T> task);
+    <T> Future<T> submit(Runnable task, T result);
+    Future<?> submit(Runnable task);
+    // invokeAll and invokeAny are complex to implement correctly in a minimal way without full infrastructure
+    // but the test uses submit. I will skip invokeAll/Any for now or add stubs if needed.
+    // The previous copied interface had them. If I omit them, I break source compatibility with full JDK,
+    // but the requirement is "subset".
+    // I'll add them if tests fail or if I can implement them simply.
+}

vm/JavaAPI/src/java/util/concurrent/Executors.java

@@ -0,0 +1,131 @@
+package java.util.concurrent;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class Executors {
+
+    public static ExecutorService newFixedThreadPool(int nThreads) {
+        return new SimpleThreadPool(nThreads);
+    }
+
+    public static ExecutorService newSingleThreadExecutor() {
+        return new SimpleThreadPool(1);
+    }
+
+    public static <T> Callable<T> callable(Runnable task, T result) {
+        if (task == null) throw new NullPointerException();
+        return new RunnableAdapter<T>(task, result);
+    }
+
+    public static Callable<Object> callable(Runnable task) {
+        if (task == null) throw new NullPointerException();
+        return new RunnableAdapter<Object>(task, null);
+    }
+
+    static final class RunnableAdapter<T> implements Callable<T> {
+        final Runnable task;
+        final T result;
+        RunnableAdapter(Runnable task, T result) {
+            this.task = task;
+            this.result = result;
+        }
+        public T call() {
+            task.run();
+            return result;
+        }
+    }
+
+    static class SimpleThreadPool extends AbstractExecutorService {
+        private final List<Runnable> workQueue = new ArrayList<Runnable>();
+        private final List<Worker> workers = new ArrayList<Worker>();
+        private boolean isShutdown;
+
+        SimpleThreadPool(int nThreads) {
+            for (int i = 0; i < nThreads; i++) {
+                Worker w = new Worker();
+                workers.add(w);
+                w.start();
+            }
+        }
+
+        public void execute(Runnable command) {
+            synchronized(workQueue) {
+                if (isShutdown) throw new RejectedExecutionException();
+                workQueue.add(command);
+                workQueue.notify();
+            }
+        }
+
+        public void shutdown() {
+            synchronized(workQueue) {
+                isShutdown = true;
+                workQueue.notifyAll();
+            }
+        }
+
+        public List<Runnable> shutdownNow() {
+            List<Runnable> pending;
+            synchronized(workQueue) {
+                isShutdown = true;
+                pending = new ArrayList<Runnable>(workQueue);
+                workQueue.clear();
+                workQueue.notifyAll();
+            }
+            return pending;
+        }
+
+        public boolean isShutdown() {
+            synchronized(workQueue) {
+                return isShutdown;
+            }
+        }
+
+        public boolean isTerminated() {
+            synchronized(workQueue) {
+                if (!isShutdown || !workQueue.isEmpty()) return false;
+                for (Worker w : workers) {
+                    if (w.isAlive()) return false;
+                }
+                return true;
+            }
+        }
+
+        public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
+            long millis = unit.toMillis(timeout);
+            long end = System.currentTimeMillis() + millis;
+            while (!isTerminated()) {
+                long delay = end - System.currentTimeMillis();
+                if (delay <= 0) return false;
+                Thread.sleep(Math.min(delay, 100));
+            }
+            return true;
+        }
+
+        class Worker extends Thread {
+            public void run() {
+                while (true) {
+                    Runnable task = null;
+                    synchronized(workQueue) {
+                        while (workQueue.isEmpty() && !isShutdown) {
+                            try {
+                                workQueue.wait();
+                            } catch (InterruptedException e) {}
+                        }
+                        if (workQueue.isEmpty() && isShutdown) {
+                            return;
+                        }
+                        task = workQueue.remove(0);
+                    }
+                    if (task != null) {
+                        try {
+                            task.run();
+                        } catch (RuntimeException e) {
+                            e.printStackTrace();
+                        }
+                    }
+                }
+            }
+        }
+    }
+}

vm/JavaAPI/src/java/util/concurrent/Future.java

@@ -0,0 +1,9 @@
+package java.util.concurrent;
+
+public interface Future<V> {
+    boolean cancel(boolean mayInterruptIfRunning);
+    boolean isCancelled();
+    boolean isDone();
+    V get() throws InterruptedException, ExecutionException;
+    V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
+}