Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 0 additions & 2 deletions gama.api/src/gama/api/runtime/GeneralSynchronizer.java
Original file line number Diff line number Diff line change
Expand Up @@ -157,7 +157,6 @@ public boolean acquire() {
semaphore.acquire();
return true;
} catch (InterruptedException e) {
e.printStackTrace();
Thread.currentThread().interrupt();
return false;
}
Expand Down Expand Up @@ -218,7 +217,6 @@ public boolean acquire(final int n) {
semaphore.acquire(n);
return true;
} catch (InterruptedException e) {
e.printStackTrace();
Thread.currentThread().interrupt();
return false;
}
Expand Down
23 changes: 22 additions & 1 deletion gama.api/src/gama/api/runtime/ISimulationRunner.java
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,9 @@
package gama.api.runtime;

import java.util.Set;
import java.util.function.BooleanSupplier;
import java.util.function.Consumer;
import java.util.function.Predicate;

import gama.api.kernel.simulation.ISimulationAgent;

Expand Down Expand Up @@ -138,4 +141,22 @@ public interface ISimulationRunner {
*/
boolean hasSimulations();

}
/**
* Lets an already added simulation run on its own thread, without any synchronization with the other simulations:
* it steps as many times as needed until {@code isOver} is true, then {@code onOver} is called (from the
* simulation's thread) and the thread terminates. The simulation is not taken into account by {@link #step()}
* anymore.
*
* @param agent
* a simulation previously added with {@link #add(ISimulationAgent)}
* @param isOver
* evaluated after each step (and in case of an error); true when the simulation must stop
* @param isPaused
* evaluated before each step; the simulation waits as long as it is true (unless it is over)
* @param onOver
* called once, when the simulation is over
*/
void runAutonomously(ISimulationAgent agent, Predicate<ISimulationAgent> isOver, BooleanSupplier isPaused,
Consumer<ISimulationAgent> onOver);

}
134 changes: 100 additions & 34 deletions gama.api/src/gama/api/runtime/SimulationRunner.java
Original file line number Diff line number Diff line change
Expand Up @@ -17,12 +17,16 @@
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BooleanSupplier;
import java.util.function.Consumer;
import java.util.function.Predicate;

import gama.api.kernel.agent.IPopulation;
import gama.api.kernel.simulation.ISimulationAgent;
import gama.api.kernel.species.IExperimentSpecies;
import gama.api.runtime.GamaExecutorService.Caller;
import gama.dev.DEBUG;
import gama.dev.THREADS;

/**
* Default implementation of {@link ISimulationRunner} for managing concurrent simulation execution.
Expand Down Expand Up @@ -169,51 +173,111 @@ private SimulationRunner(final int concurrency) {
*/
@Override
public void remove(final ISimulationAgent agent) {
if (runnables.remove(agent) != null) { activeCount.decrementAndGet(); }
final Thread t = runnables.remove(agent);
if (t == null) return;
// Autonomous simulations have already been taken out of the count
if (!(t instanceof SimulationThread st) || st.autonomy == null) { activeCount.decrementAndGet(); }
// A thread idling on the semaphore would otherwise never terminate (see issues #198 and #282)
if (t instanceof SimulationThread st && st.waiting) { st.interrupt(); }
}

/** A dedicated thread running the steps of one simulation, which can be woken up when it is removed. */
private final class SimulationThread extends Thread {

/** The agent. */
private final ISimulationAgent agent;

/** True while the thread is idle, waiting for the permission to step. */
volatile boolean waiting;

/** Set when the simulation must run on its own, without synchronization. */
volatile Autonomy autonomy;

/**
* Instantiates a new simulation thread.
*
* @param agent
* the agent
*/
SimulationThread(final ISimulationAgent agent) {
super("Thread of " + agent.getName());
this.agent = agent;
setDaemon(true);
}

/** Steps the simulation as fast as possible until it is over. */
private void runFreely(final Autonomy a) {
boolean over = false;
while (!over && !shutdown && runnables.get(agent) == this) {
try {
while (a.isPaused.getAsBoolean() && !a.isOver.test(agent) && !shutdown) { THREADS.WAIT(10); }
if (!a.isOver.test(agent) && !shutdown) { agent.step(); }
Comment thread
Copilot marked this conversation as resolved.
Outdated
} catch (Throwable tg) {
EXCEPTION_HANDLER.uncaughtException(Thread.currentThread(), tg);
}
over = agent.dead() || a.isOver.test(agent);
}
Comment thread
Copilot marked this conversation as resolved.
Outdated
if (!shutdown && runnables.get(agent) == this) { a.onOver.accept(agent); }
}

@Override
public void run() {
while (!shutdown && !agent.dead() && runnables.get(agent) == this) {
waiting = true;
final boolean acquired = autonomy != null || simulationsSemaphore.acquire();
waiting = false;
if (autonomy != null) {
Thread.interrupted();
runFreely(autonomy);
return;
}
if (!acquired) { break; }
// Recheck after waking up: dispose() may have released the permit just to
// unblock this thread for shutdown — we must not step a dead/cleared runner.
if (shutdown || agent.dead()) { break; }
try {
agent.step();
} catch (Throwable tg) {
EXCEPTION_HANDLER.uncaughtException(Thread.currentThread(), tg);
} finally {
// Released whatever the outcome: a step that throws still consumed the permit granted by
// step(), which would otherwise wait for it forever.
experimentSemaphore.release();
}
}
}
}

/** The way an autonomous simulation is run. */
private record Autonomy(Predicate<ISimulationAgent> isOver, BooleanSupplier isPaused,
Consumer<ISimulationAgent> onOver) {}

@Override
public void runAutonomously(final ISimulationAgent agent, final Predicate<ISimulationAgent> isOver,
final BooleanSupplier isPaused, final Consumer<ISimulationAgent> onOver) {
if (!(runnables.get(agent) instanceof SimulationThread st)) return;
// The simulation does not take part in the synchronized steps anymore
activeCount.decrementAndGet();
st.autonomy = new Autonomy(isOver, isPaused, onOver);
Comment thread
AlexisDrogoul marked this conversation as resolved.
// The thread is idle (or about to be) on the semaphore: wake it up so that it notices its new mode
st.interrupt();
}

/**
* Adds a simulation agent to the runner and starts its dedicated execution thread.
*
* <p>
* Creates a new thread for the simulation that will:
* </p>
* <ol>
* <li>Wait for a permit from simulationsSemaphore</li>
* <li>Execute the simulation's step</li>
* <li>Release a permit to experimentSemaphore</li>
* <li>Repeat until the simulation dies or the runner shuts down</li>
* </ol>
* Adds a simulation agent to the runner and starts its dedicated execution thread, which waits for a permit,
* steps the simulation, signals the experiment, and repeats until the simulation dies, is removed or the runner
* shuts down.
*
* @param agent
* the simulation agent to add
*/
@Override
public void add(final ISimulationAgent agent) {
Thread t = new Thread("Thread of " + agent.getName()) {
@Override
public void run() {
while (!shutdown && !agent.dead()) {
// DEBUG.OUT("Waiting for " + agent);
if (!simulationsSemaphore.acquire()) { break; }
// Recheck after waking up: dispose() may have released the permit just to
// unblock this thread for shutdown — we must not step a dead/cleared runner.
if (shutdown || agent.dead()) { break; }
try {
agent.step();
} catch (Throwable tg) {
EXCEPTION_HANDLER.uncaughtException(Thread.currentThread(), tg);
} finally {
// Released whatever the outcome: a step that throws still consumed the permit granted by
// step(), which would otherwise wait for it forever.
experimentSemaphore.release();
}
}
}
};
t.start();
final Thread t = new SimulationThread(agent);
// Registered before starting, as the thread checks that it is still registered
runnables.put(agent, t);
activeCount.incrementAndGet();
t.start();
}

/**
Expand All @@ -236,6 +300,8 @@ public void run() {
@Override
public void step() {
// DEBUG.OUT("Releasing to all simulations");
// Dead simulations never consume permits: counting them would block the experiment forever
for (final ISimulationAgent sim : runnables.keySet()) { if (sim.dead()) { remove(sim); } }
int nb = activeCount.get();
simulationsSemaphore.release(nb);
// If the experiment thread is interrupted while waiting, abort the step.
Expand Down
21 changes: 21 additions & 0 deletions gama.core/src/gama/core/simulation/SimulationPopulation.java
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,9 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.BooleanSupplier;
import java.util.function.Consumer;
import java.util.function.Predicate;

import gama.annotations.constants.IKeyword;
import gama.api.GAMA;
Expand Down Expand Up @@ -281,6 +284,24 @@ public void unscheduleSimulation(final ISimulationAgent sim) {
runner.remove(sim);
}

/**
* Lets a scheduled simulation run on its own thread, not synchronized with the other simulations, until it is
* over.
*
* @param sim
* a scheduled simulation
* @param isOver
* tells when the simulation must stop
* @param isPaused
* tells if the simulation should wait before its next step
* @param onOver
* called (from the simulation's thread) when the simulation is over
*/
public void runAutonomously(final ISimulationAgent sim, final Predicate<ISimulationAgent> isOver,
final BooleanSupplier isPaused, final Consumer<ISimulationAgent> onOver) {
runner.runAutonomously(sim, isOver, isPaused, onOver);
}

/**
* Gets the number of active stepables.
*
Expand Down
Loading
Loading