threading
This commit is contained in:
1 parent
8752ac7011
commit
5d32213922
7 files changed
+230
-44
No files matched your search
@@ -246,7 +246,7 @@ public class ListValue extends AbstractListValue implements ContainerValueInterf
|
||||
return items;
|
||||
}
|
||||
|
||||
public Iterator<Value> iterator() { return new ArrayList<>(items).iterator(); }
|
||||
public Iterator<Value> iterator() { return new ArrayList<>(items).iterator(); } // should be thread safe
|
||||
|
||||
public void extend(List<Value> subList)
|
||||
{
|
||||
|
||||
@@ -1,29 +1,49 @@
|
||||
package carpet.script.value;
|
||||
|
||||
import carpet.script.Context;
|
||||
import carpet.script.LazyValue;
|
||||
import carpet.script.Expression;
|
||||
import carpet.script.Tokenizer;
|
||||
import carpet.script.exception.InternalExpressionException;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ThreadPoolExecutor;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
public class ThreadValue extends Value
|
||||
{
|
||||
CompletableFuture<Value> taskFuture;
|
||||
private CompletableFuture<Value> taskFuture;
|
||||
private long id;
|
||||
private final Object lock;
|
||||
private final boolean[] started;
|
||||
private Supplier<Value>task;
|
||||
private static long sequence = 0L;
|
||||
|
||||
private static final ExecutorService executorService = Executors.newCachedThreadPool();
|
||||
private static final Map<Value,ThreadPoolExecutor> executorServices = new HashMap<>();
|
||||
|
||||
public ThreadValue(LazyValue expression, Context ctx)
|
||||
public ThreadValue(Value pool, FunctionValue function, Expression expr, Tokenizer.Token token, Context ctx)
|
||||
{
|
||||
id = sequence++;
|
||||
lock = new Object();
|
||||
started = new boolean[]{false};
|
||||
task = () ->
|
||||
{
|
||||
if (!started[0])
|
||||
{
|
||||
started[0] = true;
|
||||
return function.lazyEval(ctx, Context.NONE, expr, token, Collections.emptyList()).evalValue(ctx);
|
||||
}
|
||||
return null;
|
||||
};
|
||||
taskFuture = CompletableFuture.supplyAsync(
|
||||
() -> expression.evalValue(ctx),
|
||||
executorService
|
||||
() -> { synchronized (lock) { return task.get(); } },
|
||||
executorServices.computeIfAbsent(pool, (v) -> (ThreadPoolExecutor)Executors.newCachedThreadPool())
|
||||
);
|
||||
Thread.yield();
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -40,6 +60,18 @@ public class ThreadValue extends Value
|
||||
|
||||
public Value join()
|
||||
{
|
||||
if (!started[0])
|
||||
{
|
||||
synchronized (lock)
|
||||
{
|
||||
if (!started[0])
|
||||
{
|
||||
Value res = task.get();
|
||||
taskFuture.complete(res);
|
||||
return res;
|
||||
}
|
||||
}
|
||||
}
|
||||
try
|
||||
{
|
||||
return taskFuture.get();
|
||||
@@ -57,7 +89,14 @@ public class ThreadValue extends Value
|
||||
|
||||
public void stop()
|
||||
{
|
||||
taskFuture.complete(Value.NULL);
|
||||
synchronized (lock)
|
||||
{
|
||||
if (!taskFuture.isDone())
|
||||
{
|
||||
started[0] = true;
|
||||
taskFuture.complete(Value.NULL);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -72,7 +111,27 @@ public class ThreadValue extends Value
|
||||
public int compareTo(Value o)
|
||||
{
|
||||
if (!(o instanceof ThreadValue))
|
||||
throw new InternalExpressionException("Cannot compare ");
|
||||
throw new InternalExpressionException("Cannot compare tasks to other types");
|
||||
return (int) (this.id - ((ThreadValue) o).id);
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode()
|
||||
{
|
||||
return Long.hashCode(id);
|
||||
}
|
||||
|
||||
public static int taskCount()
|
||||
{
|
||||
return executorServices.values().stream().map(ThreadPoolExecutor::getActiveCount).reduce(0, Integer::sum);
|
||||
}
|
||||
|
||||
public static int taskCount(Value pool)
|
||||
{
|
||||
if (executorServices.containsKey(pool))
|
||||
{
|
||||
return executorServices.get(pool).getActiveCount();
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
@@ -110,7 +110,7 @@ public abstract class Value implements Comparable<Value>, Cloneable
|
||||
@Override
|
||||
public int compareTo(final Value o)
|
||||
{
|
||||
if (o instanceof NumericValue || o instanceof ListValue)
|
||||
if (o instanceof NumericValue || o instanceof ListValue || o instanceof ThreadValue)
|
||||
{
|
||||
return -o.compareTo(this);
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user