mirror of
https://github.com/MinecraftForge/MinecraftForge.git
synced 2026-11-13 00:37:52 +00:00
Remove usage of AsynchronousExecutor library in favor or a simpler implementation.
This commit is contained in:
LexManos committed
1 parent
d8249b7886
commit
3177d55579
5 files changed
+198
-577
No files matched your search
@@ -1,32 +1,130 @@
|
||||
package net.minecraftforge.common.chunkio;
|
||||
|
||||
import net.minecraftforge.common.util.AsynchronousExecutor;
|
||||
import java.util.Iterator;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.concurrent.ThreadFactory;
|
||||
import java.util.concurrent.ThreadPoolExecutor;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
public class ChunkIOExecutor {
|
||||
import com.google.common.collect.Maps;
|
||||
|
||||
import net.minecraft.world.World;
|
||||
import net.minecraft.world.chunk.Chunk;
|
||||
import net.minecraft.world.chunk.storage.AnvilChunkLoader;
|
||||
import net.minecraft.world.gen.ChunkProviderServer;
|
||||
import net.minecraftforge.fml.common.FMLLog;
|
||||
|
||||
public class ChunkIOExecutor
|
||||
{
|
||||
static final int BASE_THREADS = 1;
|
||||
static final int PLAYERS_PER_THREAD = 50;
|
||||
|
||||
private static final AsynchronousExecutor<QueuedChunk, net.minecraft.world.chunk.Chunk, Runnable, RuntimeException> instance = new AsynchronousExecutor<QueuedChunk, net.minecraft.world.chunk.Chunk, Runnable, RuntimeException>(new ChunkIOProvider(), BASE_THREADS);
|
||||
private static final Map<QueuedChunk, ChunkIOProvider> tasks = Maps.newConcurrentMap();
|
||||
private static final ThreadPoolExecutor pool = new ThreadPoolExecutor(BASE_THREADS, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
|
||||
new LinkedBlockingQueue<Runnable>(),
|
||||
new ThreadFactory()
|
||||
{
|
||||
private AtomicInteger count = new AtomicInteger(1);
|
||||
@Override
|
||||
public Thread newThread(Runnable r)
|
||||
{
|
||||
Thread thread = new Thread(r, "Chunk I/O Executor Thread-" + count.getAndIncrement());
|
||||
thread.setDaemon(true);
|
||||
return thread;
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
public static net.minecraft.world.chunk.Chunk syncChunkLoad(net.minecraft.world.World world, net.minecraft.world.chunk.storage.AnvilChunkLoader loader, net.minecraft.world.gen.ChunkProviderServer provider, int x, int z) {
|
||||
return instance.getSkipQueue(new QueuedChunk(x, z, loader, world, provider));
|
||||
//Load the chunk completely in this thread. Dequeue as needed...
|
||||
public static Chunk syncChunkLoad(World world, AnvilChunkLoader loader, ChunkProviderServer provider, int x, int z)
|
||||
{
|
||||
QueuedChunk key = new QueuedChunk(x, z, loader, world, provider);
|
||||
ChunkIOProvider task = tasks.get(key);
|
||||
if (task != null)
|
||||
{
|
||||
if (!pool.remove(task) && !task.runFinished()) // If it wasn't in the pool, and run hasn't finished, then wait for the async thread.
|
||||
{
|
||||
synchronized(task)
|
||||
{
|
||||
try
|
||||
{
|
||||
task.wait();
|
||||
}
|
||||
catch (InterruptedException e)
|
||||
{
|
||||
e.printStackTrace(); // Something happened? Log it?
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
task = new ChunkIOProvider(key);
|
||||
task.run();
|
||||
}
|
||||
task.syncCallback();
|
||||
return task.getChunk();
|
||||
}
|
||||
|
||||
public static void queueChunkLoad(net.minecraft.world.World world, net.minecraft.world.chunk.storage.AnvilChunkLoader loader, net.minecraft.world.gen.ChunkProviderServer provider, int x, int z, Runnable runnable) {
|
||||
instance.add(new QueuedChunk(x, z, loader, world, provider), runnable);
|
||||
//Queue the chunk to be loaded, and call the runnable when finished
|
||||
public static void queueChunkLoad(World world, AnvilChunkLoader loader, ChunkProviderServer provider, int x, int z, Runnable runnable)
|
||||
{
|
||||
QueuedChunk key = new QueuedChunk(x, z, loader, world, provider);
|
||||
ChunkIOProvider task = tasks.get(key);
|
||||
if (task == null)
|
||||
{
|
||||
task = new ChunkIOProvider(key);
|
||||
task.addCallback(runnable); // Add before calling execute for thread safety
|
||||
tasks.put(key, task);
|
||||
pool.execute(task);
|
||||
}
|
||||
else
|
||||
{
|
||||
task.addCallback(runnable);
|
||||
}
|
||||
}
|
||||
|
||||
// Abuses the fact that hashCode and equals for QueuedChunk only use world and coords
|
||||
public static void dropQueuedChunkLoad(net.minecraft.world.World world, int x, int z, Runnable runnable) {
|
||||
instance.drop(new QueuedChunk(x, z, null, world, null), runnable);
|
||||
// Remove the chunk from the queue if it's in the list.
|
||||
public static void dropQueuedChunkLoad(World world, int x, int z, Runnable runnable)
|
||||
{
|
||||
QueuedChunk key = new QueuedChunk(x, z, null, world, null);
|
||||
ChunkIOProvider task = tasks.get(key);
|
||||
if (task == null)
|
||||
{
|
||||
FMLLog.warning("Attempted to dequeue chunk that wasn't queued? %d @ (%d, %d)", world.provider.getDimension(), x, z);
|
||||
return;
|
||||
}
|
||||
|
||||
task.removeCallback(runnable);
|
||||
|
||||
if (!task.hasCallback())
|
||||
{
|
||||
tasks.remove(key);
|
||||
pool.remove(task);
|
||||
}
|
||||
}
|
||||
|
||||
public static void adjustPoolSize(int players) {
|
||||
public static void adjustPoolSize(int players)
|
||||
{
|
||||
int size = Math.max(BASE_THREADS, (int) Math.ceil(players / PLAYERS_PER_THREAD));
|
||||
instance.setActiveThreads(size);
|
||||
pool.setCorePoolSize(size);
|
||||
}
|
||||
|
||||
public static void tick() {
|
||||
instance.finishActive();
|
||||
public static void tick()
|
||||
{
|
||||
Iterator<ChunkIOProvider> itr = tasks.values().iterator();
|
||||
while (itr.hasNext())
|
||||
{
|
||||
ChunkIOProvider task = itr.next();
|
||||
if (task.runFinished() && task.hasCallback())
|
||||
{
|
||||
task.syncCallback();
|
||||
}
|
||||
itr.remove();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,64 +1,113 @@
|
||||
package net.minecraftforge.common.chunkio;
|
||||
|
||||
|
||||
import net.minecraft.world.ChunkCoordIntPair;
|
||||
import net.minecraft.world.chunk.Chunk;
|
||||
import net.minecraft.world.chunk.storage.AnvilChunkLoader;
|
||||
import net.minecraft.world.gen.ChunkProviderServer;
|
||||
import net.minecraft.nbt.NBTTagCompound;
|
||||
import net.minecraftforge.common.MinecraftForge;
|
||||
import net.minecraftforge.common.util.AsynchronousExecutor;
|
||||
import net.minecraftforge.event.world.ChunkDataEvent;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||
|
||||
class ChunkIOProvider implements AsynchronousExecutor.CallBackProvider<QueuedChunk, net.minecraft.world.chunk.Chunk, Runnable, RuntimeException> {
|
||||
private final AtomicInteger threadNumber = new AtomicInteger(1);
|
||||
class ChunkIOProvider implements Runnable
|
||||
{
|
||||
private QueuedChunk chunkInfo;
|
||||
private Chunk chunk;
|
||||
private NBTTagCompound nbt;
|
||||
private ConcurrentLinkedQueue<Runnable> callbacks = new ConcurrentLinkedQueue<Runnable>();
|
||||
private boolean ran = false;
|
||||
|
||||
// async stuff
|
||||
public net.minecraft.world.chunk.Chunk callStage1(QueuedChunk queuedChunk) throws RuntimeException {
|
||||
net.minecraft.world.chunk.storage.AnvilChunkLoader loader = queuedChunk.loader;
|
||||
Object[] data = null;
|
||||
try {
|
||||
data = loader.loadChunk__Async(queuedChunk.world, queuedChunk.x, queuedChunk.z);
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
ChunkIOProvider(QueuedChunk chunk)
|
||||
{
|
||||
this.chunkInfo = chunk;
|
||||
}
|
||||
|
||||
public void addCallback(Runnable callback)
|
||||
{
|
||||
this.callbacks.add(callback);
|
||||
}
|
||||
public void removeCallback(Runnable callback)
|
||||
{
|
||||
this.callbacks.remove(callback);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() // async stuff
|
||||
{
|
||||
synchronized(this)
|
||||
{
|
||||
AnvilChunkLoader loader = chunkInfo.loader;
|
||||
Object[] data = null;
|
||||
try
|
||||
{
|
||||
data = loader.loadChunk__Async(chunkInfo.world, chunkInfo.x, chunkInfo.z);
|
||||
}
|
||||
catch (IOException e)
|
||||
{
|
||||
e.printStackTrace();
|
||||
}
|
||||
|
||||
if (data != null)
|
||||
{
|
||||
this.nbt = (NBTTagCompound)data[1];
|
||||
this.chunk = (Chunk)data[0];
|
||||
}
|
||||
|
||||
this.ran = true;
|
||||
this.notifyAll();
|
||||
}
|
||||
|
||||
if (data != null) {
|
||||
queuedChunk.compound = (net.minecraft.nbt.NBTTagCompound) data[1];
|
||||
return (net.minecraft.world.chunk.Chunk) data[0];
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
// sync stuff
|
||||
public void callStage2(QueuedChunk queuedChunk, net.minecraft.world.chunk.Chunk chunk) throws RuntimeException {
|
||||
if(chunk == null) {
|
||||
public void syncCallback()
|
||||
{
|
||||
ChunkProviderServer provider = this.chunkInfo.provider;
|
||||
if (chunk == null)
|
||||
{
|
||||
// If the chunk loading failed just do it synchronously (may generate)
|
||||
queuedChunk.provider.originalLoadChunk(queuedChunk.x, queuedChunk.z);
|
||||
provider.originalLoadChunk(this.chunkInfo.x, this.chunkInfo.z);
|
||||
return;
|
||||
}
|
||||
|
||||
queuedChunk.loader.loadEntities(queuedChunk.world, queuedChunk.compound.getCompoundTag("Level"), chunk);
|
||||
MinecraftForge.EVENT_BUS.post(new ChunkDataEvent.Load(chunk, queuedChunk.compound)); // Don't call ChunkDataEvent.Load async
|
||||
chunk.setLastSaveTime(queuedChunk.provider.worldObj.getTotalWorldTime());
|
||||
queuedChunk.provider.id2ChunkMap.add(ChunkCoordIntPair.chunkXZ2Int(queuedChunk.x, queuedChunk.z), chunk);
|
||||
queuedChunk.provider.loadedChunks.add(chunk);
|
||||
chunk.onChunkLoad();
|
||||
// Load Entities
|
||||
this.chunkInfo.loader.loadEntities(this.chunkInfo.world, this.nbt.getCompoundTag("Level"), this.chunk);
|
||||
|
||||
if (queuedChunk.provider.chunkGenerator != null) {
|
||||
queuedChunk.provider.chunkGenerator.recreateStructures(chunk, queuedChunk.x, queuedChunk.z);
|
||||
MinecraftForge.EVENT_BUS.post(new ChunkDataEvent.Load(this.chunk, this.nbt)); // Don't call ChunkDataEvent.Load async
|
||||
|
||||
this.chunk.setLastSaveTime(provider.worldObj.getTotalWorldTime());
|
||||
provider.id2ChunkMap.add(ChunkCoordIntPair.chunkXZ2Int(this.chunkInfo.x, this.chunkInfo.z), this.chunk);
|
||||
provider.loadedChunks.add(this.chunk);
|
||||
this.chunk.onChunkLoad();
|
||||
|
||||
if (provider.chunkGenerator != null)
|
||||
{
|
||||
provider.chunkGenerator.recreateStructures(this.chunk, this.chunkInfo.x, this.chunkInfo.z);
|
||||
}
|
||||
|
||||
chunk.populateChunk(queuedChunk.provider, queuedChunk.provider.chunkGenerator);
|
||||
this.chunk.populateChunk(provider, provider.chunkGenerator);
|
||||
|
||||
for (Runnable r : this.callbacks)
|
||||
{
|
||||
r.run();
|
||||
}
|
||||
|
||||
this.callbacks.clear();
|
||||
}
|
||||
|
||||
public void callStage3(QueuedChunk queuedChunk, net.minecraft.world.chunk.Chunk chunk, Runnable runnable) throws RuntimeException {
|
||||
runnable.run();
|
||||
public Chunk getChunk()
|
||||
{
|
||||
return this.chunk;
|
||||
}
|
||||
|
||||
public Thread newThread(Runnable runnable) {
|
||||
Thread thread = new Thread(runnable, "Chunk I/O Executor Thread-" + threadNumber.getAndIncrement());
|
||||
thread.setDaemon(true);
|
||||
return thread;
|
||||
public boolean runFinished()
|
||||
{
|
||||
return this.ran;
|
||||
}
|
||||
|
||||
public boolean hasCallback()
|
||||
{
|
||||
return this.callbacks.size() > 0;
|
||||
}
|
||||
}
|
||||
@@ -7,7 +7,6 @@ class QueuedChunk {
|
||||
final net.minecraft.world.chunk.storage.AnvilChunkLoader loader;
|
||||
final net.minecraft.world.World world;
|
||||
final net.minecraft.world.gen.ChunkProviderServer provider;
|
||||
net.minecraft.nbt.NBTTagCompound compound;
|
||||
|
||||
public QueuedChunk(int x, int z, net.minecraft.world.chunk.storage.AnvilChunkLoader loader, net.minecraft.world.World world, net.minecraft.world.gen.ChunkProviderServer provider) {
|
||||
this.x = x;
|
||||
|
||||
@@ -1,360 +0,0 @@
|
||||
package net.minecraftforge.common.util;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Queue;
|
||||
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.concurrent.ThreadFactory;
|
||||
import java.util.concurrent.ThreadPoolExecutor;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
|
||||
|
||||
import net.minecraftforge.fml.common.FMLLog;
|
||||
|
||||
|
||||
/**
|
||||
* Executes tasks using a multi-stage process executor. Synchronous executions are via {@link AsynchronousExecutor#finishActive()} or the {@link AsynchronousExecutor#get(Object)} methods.
|
||||
* <li \> Stage 1 creates the object from a parameter, and is usually called asynchronously.
|
||||
* <li \> Stage 2 takes the parameter and object from stage 1 and does any synchronous processing to prepare it.
|
||||
* <li \> Stage 3 takes the parameter and object from stage 1, as well as a callback that was registered, and performs any synchronous calculations.
|
||||
*
|
||||
* @param <P> The type of parameter you provide to make the object that will be created. It should implement {@link Object#hashCode()} and {@link Object#equals(Object)} if you want to get the value early.
|
||||
* @param <T> The type of object you provide. This is created in stage 1, and passed to stage 2, 3, and returned if get() is called.
|
||||
* @param <C> The type of callback you provide. You may register many of these to be passed to the provider in stage 3, one at a time.
|
||||
* @param <E> A type of exception you may throw and expect to be handled by the main thread
|
||||
* @author Wesley Wolfe (c) 2012, 2014
|
||||
*/
|
||||
public final class AsynchronousExecutor<P, T, C, E extends Throwable> {
|
||||
|
||||
public static interface CallBackProvider<P, T, C, E extends Throwable> extends ThreadFactory {
|
||||
|
||||
/**
|
||||
* Normally an asynchronous call, but can be synchronous
|
||||
*
|
||||
* @param parameter parameter object provided
|
||||
* @return the created object
|
||||
*/
|
||||
T callStage1(P parameter) throws E;
|
||||
|
||||
/**
|
||||
* Synchronous call
|
||||
*
|
||||
* @param parameter parameter object provided
|
||||
* @param object the previously created object
|
||||
*/
|
||||
void callStage2(P parameter, T object) throws E;
|
||||
|
||||
/**
|
||||
* Synchronous call, called multiple times, once per registered callback
|
||||
*
|
||||
* @param parameter parameter object provided
|
||||
* @param object the previously created object
|
||||
* @param callback the current callback to execute
|
||||
*/
|
||||
void callStage3(P parameter, T object, C callback) throws E;
|
||||
}
|
||||
|
||||
private static class StateHolder {
|
||||
|
||||
protected static final int PENDING = 0x0;
|
||||
protected static final int STAGE_1_ASYNC = PENDING + 1;
|
||||
protected static final int STAGE_1_SYNC = STAGE_1_ASYNC + 1;
|
||||
protected static final int STAGE_1_COMPLETE = STAGE_1_SYNC + 1;
|
||||
protected static final int FINISHED = STAGE_1_COMPLETE + 1;
|
||||
|
||||
protected volatile int state = PENDING;
|
||||
|
||||
protected static final AtomicIntegerFieldUpdater<StateHolder> STATE = AtomicIntegerFieldUpdater.newUpdater(StateHolder.class, "state");
|
||||
}
|
||||
|
||||
private class Task extends StateHolder implements Runnable {
|
||||
|
||||
private final P parameter;
|
||||
private T object;
|
||||
private final List<C> callbacks = new LinkedList<C>();
|
||||
private E t = null;
|
||||
|
||||
private Task(final P parameter) {
|
||||
this.parameter = parameter;
|
||||
}
|
||||
|
||||
public void run() {
|
||||
if (initAsync()) {
|
||||
finished.add(this);
|
||||
}
|
||||
}
|
||||
|
||||
private boolean initAsync() {
|
||||
if (STATE.compareAndSet(this, PENDING, STAGE_1_ASYNC)) {
|
||||
boolean ret = true;
|
||||
|
||||
try {
|
||||
init();
|
||||
} finally {
|
||||
if (STATE.compareAndSet(this, STAGE_1_ASYNC, STAGE_1_COMPLETE)) {
|
||||
// No one is/will be waiting
|
||||
} else {
|
||||
// We know that the sync thread will be waiting
|
||||
synchronized (this) {
|
||||
if (state != STAGE_1_SYNC) {
|
||||
// They beat us to the synchronized block
|
||||
this.notifyAll();
|
||||
} else {
|
||||
// We beat them to the synchronized block
|
||||
}
|
||||
state = STAGE_1_COMPLETE; // They're already synchronized, atomic locks are not needed
|
||||
}
|
||||
// We want to return false, because we know a synchronous task already handled the finish()
|
||||
ret = false; // Don't return inside finally; VERY bad practice.
|
||||
}
|
||||
}
|
||||
|
||||
return ret;
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private void initSync() {
|
||||
if (STATE.compareAndSet(this, PENDING, STAGE_1_COMPLETE)) {
|
||||
// If we succeed that variable switch, good as done
|
||||
init();
|
||||
} else if (STATE.compareAndSet(this, STAGE_1_ASYNC, STAGE_1_SYNC)) {
|
||||
// Async thread is running, but this shouldn't be likely; we need to sync to wait on them because of it.
|
||||
synchronized (this) {
|
||||
if (STATE.compareAndSet(this, STAGE_1_SYNC, PENDING)) { // They might NOT synchronized yet, atomic lock IS needed
|
||||
// We are the first into the lock
|
||||
while (state != STAGE_1_COMPLETE) {
|
||||
try {
|
||||
this.wait();
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
throw new RuntimeException("Unable to handle interruption on " + parameter, e);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// They beat us to the synchronized block
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Async thread is not pending, the more likely situation for a task not pending
|
||||
}
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private void init() {
|
||||
try {
|
||||
object = provider.callStage1(parameter);
|
||||
} catch (final Throwable t) {
|
||||
this.t = (E) t;
|
||||
}
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private T get() throws E {
|
||||
initSync();
|
||||
if (callbacks.isEmpty()) {
|
||||
// 'this' is a placeholder to prevent callbacks from being empty during finish call
|
||||
// See get method below
|
||||
callbacks.add((C) this);
|
||||
}
|
||||
finish();
|
||||
return object;
|
||||
}
|
||||
|
||||
private void finish() throws E {
|
||||
switch (state) {
|
||||
default:
|
||||
case PENDING:
|
||||
case STAGE_1_ASYNC:
|
||||
case STAGE_1_SYNC:
|
||||
throw new IllegalStateException("Attempting to finish unprepared(" + state + ") task(" + parameter + ")");
|
||||
case STAGE_1_COMPLETE:
|
||||
try {
|
||||
if (t != null) {
|
||||
throw t;
|
||||
}
|
||||
if (callbacks.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
|
||||
final CallBackProvider<P, T, C, E> provider = AsynchronousExecutor.this.provider;
|
||||
final P parameter = this.parameter;
|
||||
final T object = this.object;
|
||||
|
||||
provider.callStage2(parameter, object);
|
||||
for (C callback : callbacks) {
|
||||
if (callback == this) {
|
||||
// 'this' is a placeholder to prevent callbacks from being empty on a get() call
|
||||
// See get method above
|
||||
continue;
|
||||
}
|
||||
provider.callStage3(parameter, object, callback);
|
||||
}
|
||||
} finally {
|
||||
tasks.remove(parameter);
|
||||
state = FINISHED;
|
||||
}
|
||||
case FINISHED:
|
||||
}
|
||||
}
|
||||
|
||||
private boolean drop() {
|
||||
if (STATE.compareAndSet(this, PENDING, FINISHED)) {
|
||||
// If we succeed that variable switch, good as forgotten
|
||||
tasks.remove(parameter);
|
||||
return true;
|
||||
} else {
|
||||
// We need the async thread to finish normally to properly dispose of the task
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private final CallBackProvider<P, T, C, E> provider;
|
||||
private final Queue<Task> finished = new ConcurrentLinkedQueue<Task>();
|
||||
private final Map<P, Task> tasks = new HashMap<P, Task>();
|
||||
private final ThreadPoolExecutor pool;
|
||||
|
||||
/**
|
||||
* Uses a thread pool to pass executions to the provider.
|
||||
* @see AsynchronousExecutor
|
||||
*/
|
||||
public AsynchronousExecutor(final CallBackProvider<P, T, C, E> provider, final int coreSize) {
|
||||
if (provider == null) {
|
||||
throw new IllegalArgumentException("Provider cannot be null");
|
||||
}
|
||||
this.provider = provider;
|
||||
|
||||
// We have an unbound queue size so do not need a max thread size
|
||||
pool = new ThreadPoolExecutor(coreSize, Integer.MAX_VALUE, 60l, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), provider);
|
||||
}
|
||||
|
||||
/**
|
||||
* Adds a callback to the parameter provided, adding parameter to the queue if needed.
|
||||
* <p>
|
||||
* This should always be synchronous.
|
||||
*/
|
||||
public void add(P parameter, C callback) {
|
||||
Task task = tasks.get(parameter);
|
||||
if (task == null) {
|
||||
tasks.put(parameter, task = new Task(parameter));
|
||||
pool.execute(task);
|
||||
}
|
||||
task.callbacks.add(callback);
|
||||
}
|
||||
|
||||
/**
|
||||
* This removes a particular callback from the specified parameter.
|
||||
* <p>
|
||||
* If no callbacks remain for a given parameter, then the {@link CallBackProvider CallBackProvider's} stages may be omitted from execution.
|
||||
* Stage 3 will have no callbacks, stage 2 will be skipped unless a {@link #get(Object)} is used, and stage 1 will be avoided on a best-effort basis.
|
||||
* <p>
|
||||
* Subsequent calls to {@link #getSkipQueue(Object)} will always work.
|
||||
* <p>
|
||||
* Subsequent calls to {@link #get(Object)} might work.
|
||||
* <p>
|
||||
* This should always be synchronous
|
||||
* @return true if no further execution for the parameter is possible, such that, no exceptions will be thrown in {@link #finishActive()} for the parameter, and {@link #get(Object)} will throw an {@link IllegalStateException}, false otherwise
|
||||
* @throws IllegalStateException if parameter is not in the queue anymore
|
||||
* @throws IllegalStateException if the callback was not specified for given parameter
|
||||
*/
|
||||
public boolean drop(P parameter, C callback) throws IllegalStateException {
|
||||
final Task task = tasks.get(parameter);
|
||||
if (task == null) {
|
||||
// Print debug info for QueuedChunk and avoid crash
|
||||
//throw new IllegalStateException("Unknown " + parameter);
|
||||
FMLLog.info("Unknown %s", parameter);
|
||||
FMLLog.info("This should not happen. Please report this error to Forge.");
|
||||
return false;
|
||||
}
|
||||
if (!task.callbacks.remove(callback)) {
|
||||
throw new IllegalStateException("Unknown " + callback + " for " + parameter);
|
||||
}
|
||||
if (task.callbacks.isEmpty()) {
|
||||
return task.drop();
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* This method attempts to skip the waiting period for said parameter.
|
||||
* <p>
|
||||
* This should always be synchronous.
|
||||
* @throws IllegalStateException if the parameter is not in the queue anymore, or sometimes if called from asynchronous thread
|
||||
*/
|
||||
public T get(P parameter) throws E, IllegalStateException {
|
||||
final Task task = tasks.get(parameter);
|
||||
if (task == null) {
|
||||
throw new IllegalStateException("Unknown " + parameter);
|
||||
}
|
||||
return task.get();
|
||||
}
|
||||
|
||||
/**
|
||||
* Processes a parameter as if it was in the queue, without ever passing to another thread.
|
||||
*/
|
||||
public T getSkipQueue(P parameter) throws E {
|
||||
return skipQueue(parameter);
|
||||
}
|
||||
|
||||
/**
|
||||
* Processes a parameter as if it was in the queue, without ever passing to another thread.
|
||||
*/
|
||||
public T getSkipQueue(P parameter, C callback) throws E {
|
||||
final T object = skipQueue(parameter);
|
||||
provider.callStage3(parameter, object, callback);
|
||||
return object;
|
||||
}
|
||||
|
||||
/**
|
||||
* Processes a parameter as if it was in the queue, without ever passing to another thread.
|
||||
*/
|
||||
public T getSkipQueue(P parameter, C... callbacks) throws E {
|
||||
final CallBackProvider<P, T, C, E> provider = this.provider;
|
||||
final T object = skipQueue(parameter);
|
||||
for (C callback : callbacks) {
|
||||
provider.callStage3(parameter, object, callback);
|
||||
}
|
||||
return object;
|
||||
}
|
||||
|
||||
/**
|
||||
* Processes a parameter as if it was in the queue, without ever passing to another thread.
|
||||
*/
|
||||
public T getSkipQueue(P parameter, Iterable<C> callbacks) throws E {
|
||||
final CallBackProvider<P, T, C, E> provider = this.provider;
|
||||
final T object = skipQueue(parameter);
|
||||
for (C callback : callbacks) {
|
||||
provider.callStage3(parameter, object, callback);
|
||||
}
|
||||
return object;
|
||||
}
|
||||
|
||||
private T skipQueue(P parameter) throws E {
|
||||
Task task = tasks.get(parameter);
|
||||
if (task != null) {
|
||||
return task.get();
|
||||
}
|
||||
T object = provider.callStage1(parameter);
|
||||
provider.callStage2(parameter, object);
|
||||
return object;
|
||||
}
|
||||
|
||||
/**
|
||||
* This is the 'heartbeat' that should be called synchronously to finish any pending tasks
|
||||
*/
|
||||
public void finishActive() throws E {
|
||||
final Queue<Task> finished = this.finished;
|
||||
while (!finished.isEmpty()) {
|
||||
finished.poll().finish();
|
||||
}
|
||||
}
|
||||
|
||||
public void setActiveThreads(final int coreSize) {
|
||||
pool.setCorePoolSize(coreSize);
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user