Agent avoid lock by returning a bool if they need to reschedule

AJdeMooij · AJdeMooij · commit a2cbd4baa804 · 2020-10-04T16:08:21.000+02:00
diff --git a/src/nl/uu/cs/iss/ga/sim2apl/core/deliberation/DeliberationRunnable.java b/src/nl/uu/cs/iss/ga/sim2apl/core/deliberation/DeliberationRunnable.java
@@ -1,13 +1,11 @@
 package nl.uu.cs.iss.ga.sim2apl.core.deliberation;
- 
-import nl.uu.cs.iss.ga.sim2apl.core.platform.Platform;
+
 import nl.uu.cs.iss.ga.sim2apl.core.agent.Agent;
 import nl.uu.cs.iss.ga.sim2apl.core.agent.AgentID;
 import nl.uu.cs.iss.ga.sim2apl.core.plan.PlanExecutionError;
+import nl.uu.cs.iss.ga.sim2apl.core.platform.Platform;
 
 import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
 import java.util.concurrent.Callable;
 
 /**
@@ -17,7 +15,7 @@
  *
  * @author Bas Testerink
  */
-public final class DeliberationRunnable<T> implements Callable<List<T>> {
+public final class DeliberationRunnable<T> implements Callable<ReschedulableResult<T>> {
 	/** Interface to obtain the relevant agent's data. */
 	private final nl.uu.cs.iss.ga.sim2apl.core.agent.Agent<T> agent;
 	/** Interface to the relevant platform functionalities. */
@@ -34,7 +32,7 @@ public final class DeliberationRunnable<T> implements Callable<List<T>> {
 	public DeliberationRunnable(final nl.uu.cs.iss.ga.sim2apl.core.agent.Agent<T> agent, final Platform platform) {
 		this.agent = agent;
 		this.platform = platform;
-		this.agent.setSelfRescheduler(new SelfRescheduler(this));
+		this.agent.setSelfRescheduler(new SelfRescheduler<>(this));
 	}
 
 	/**
@@ -46,10 +44,11 @@ public DeliberationRunnable(final nl.uu.cs.iss.ga.sim2apl.core.agent.Agent<T> ag
 	 * killed and removed from the platform.
 	 */
 	@Override
-	public List<T> call(){
+	public ReschedulableResult<T> call(){
 		if(!this.agent.isDone()){ // Check first if agent was killed outside of this runnable
             // Clear intended actions potential previous deliberation cycle
             this.intendedActions = new ArrayList<>();
+			boolean reschedule = false;
 
 			try {
 				// Go through the cycle and execute each step.
@@ -71,7 +70,8 @@ public List<T> call(){
 					initiateShutdown(this.agent);
 				} else {
 					if (!this.agent.checkSleeping()) { // If the agents goes to sleep then it will be woken upon any external input (message, external trigger)
-						reschedule();
+//						reschedule();
+						reschedule = true;
 					} else {
 						Platform.getLogger().log(DeliberationRunnable.class, String.format("Agent %s going to sleep",
 								agent.getAID().getName()));
@@ -87,12 +87,13 @@ public List<T> call(){
 			}
 
             // Produce the set of intended actions
-            return intendedActions;
+            return new ReschedulableResult<T>(this, intendedActions, reschedule);
 		} else {
 			initiateShutdown(agent);
 
 			// An agent that shuts down will no longer perform actions
-            return Collections.emptyList();
+//            return Collections.emptyList();
+			return new ReschedulableResult<>(this, intendedActions, false);
 		}
 	}
 
@@ -115,7 +116,7 @@ private void initiateShutdown(Agent<T> agent) {
 	public final AgentID getAgentID(){ return this.agent.getAID(); }
 	
 	/** Reschedule this deliberation runnable so it will be executed again in the future. */
-	public final synchronized void reschedule(){
+	public final void reschedule(){
 		this.platform.scheduleForExecution(this);
-	} 
+	}
 }
diff --git a/src/nl/uu/cs/iss/ga/sim2apl/core/deliberation/ReschedulableResult.java b/src/nl/uu/cs/iss/ga/sim2apl/core/deliberation/ReschedulableResult.java
@@ -0,0 +1,36 @@
+package nl.uu.cs.iss.ga.sim2apl.core.deliberation;
+
+import nl.uu.cs.iss.ga.sim2apl.core.agent.AgentID;
+
+import java.util.List;
+
+public class ReschedulableResult<T> {
+
+    final AgentID agentID;
+    final DeliberationRunnable<T> deliberationRunnable;
+    final boolean reschedule;
+    final List<T> result;
+
+    public ReschedulableResult(DeliberationRunnable<T> deliberationRunnable, List<T> result, boolean reschedule) {
+        this.deliberationRunnable = deliberationRunnable;
+        this.agentID = deliberationRunnable.getAgentID();
+        this.reschedule = reschedule;
+        this.result = result;
+    }
+
+    public AgentID getAgentID() {
+        return agentID;
+    }
+
+    public DeliberationRunnable<T> getDeliberationRunnable() {
+        return deliberationRunnable;
+    }
+
+    public boolean isReschedule() {
+        return reschedule;
+    }
+
+    public List<T> getResult() {
+        return result;
+    }
+}
diff --git a/src/nl/uu/cs/iss/ga/sim2apl/core/deliberation/SelfRescheduler.java b/src/nl/uu/cs/iss/ga/sim2apl/core/deliberation/SelfRescheduler.java
@@ -5,11 +5,11 @@
  * 
  * @author Bas Testerink
  */
-public final class SelfRescheduler {
+public final class SelfRescheduler<T> {
 	/** The deliberation runnable that can be rescheduled. */
-	private final DeliberationRunnable deliberationRunnable;
+	private final DeliberationRunnable<T> deliberationRunnable;
 	
-	public SelfRescheduler(final DeliberationRunnable deliberationRunnable){
+	public SelfRescheduler(final DeliberationRunnable<T> deliberationRunnable){
 		this.deliberationRunnable = deliberationRunnable; 
 	}
 	
diff --git a/src/nl/uu/cs/iss/ga/sim2apl/core/platform/Platform.java b/src/nl/uu/cs/iss/ga/sim2apl/core/platform/Platform.java
@@ -275,9 +275,9 @@ public Set<AgentID> getLocalDirectoryFacilitators() {
      * @param deliberationRunnable Deliberation cycle to be executed sometime in the future.
      */
     public final void scheduleForExecution(final DeliberationRunnable deliberationRunnable) {
-        synchronized (this.tickExecutor) {
+//        synchronized (this.tickExecutor) {
             this.tickExecutor.scheduleForNextTick(deliberationRunnable);
-        }
+//        }
     }
 
     /**
diff --git a/src/nl/uu/cs/iss/ga/sim2apl/core/tick/DefaultBlockingTickExecutor.java b/src/nl/uu/cs/iss/ga/sim2apl/core/tick/DefaultBlockingTickExecutor.java
@@ -2,12 +2,10 @@
 
 import nl.uu.cs.iss.ga.sim2apl.core.agent.AgentID;
 import nl.uu.cs.iss.ga.sim2apl.core.deliberation.DeliberationRunnable;
+import nl.uu.cs.iss.ga.sim2apl.core.deliberation.ReschedulableResult;
 
 import java.util.*;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
+import java.util.concurrent.*;
 import java.util.stream.Collectors;
 
 /**
@@ -29,15 +27,15 @@ public class DefaultBlockingTickExecutor<T> implements TickExecutor<T> {
     private final ExecutorService executor;
 
     /** The list of agents scheduled for the next tick **/
-    private final ArrayList<DeliberationRunnable<T>> scheduledRunnables;
+    private final Map<AgentID, DeliberationRunnable<T>> scheduledRunnables;
 
     /**
      * Default constructor
      * @param nThreads Number of threads to use to execute the agent's sense-reason-act cycles.
      */
     public DefaultBlockingTickExecutor(int nThreads) {
         this.executor = Executors.newFixedThreadPool(nThreads);
-        this.scheduledRunnables = new ArrayList<>();
+        this.scheduledRunnables = new ConcurrentHashMap<>();
     }
 
     /**
@@ -60,11 +58,13 @@ public DefaultBlockingTickExecutor(int nThreads, Random random) {
      */
     @Override
     public boolean scheduleForNextTick(DeliberationRunnable<T> agentDeliberationRunnable) {
-        if (!this.scheduledRunnables.contains(agentDeliberationRunnable)) {
-            this.scheduledRunnables.add(agentDeliberationRunnable);
-            return true;
-        }
-        return false;
+        this.scheduledRunnables.put(agentDeliberationRunnable.getAgentID(), agentDeliberationRunnable);
+//        if (!this.scheduledRunnables.containsKey(agentDeliberationRunnable.getAgentID())) {
+//            this.scheduledRunnables.add(agentDeliberationRunnable);
+//            return true;
+//        }
+//        return false;
+        return true;
     }
 
     /**
@@ -75,7 +75,7 @@ public HashMap<AgentID, List<T>> doTick() {
         ArrayList<DeliberationRunnable<T>> runnables;
         // TODO make sure running can only happen once with some sort of mutex? How to verify if a tick is currently being executed?
         synchronized (this.scheduledRunnables) {
-            runnables = new ArrayList<>(this.scheduledRunnables);
+            runnables = new ArrayList<>(this.scheduledRunnables.values());
             this.scheduledRunnables.clear();
         }
 
@@ -88,30 +88,21 @@ public HashMap<AgentID, List<T>> doTick() {
 
         long startTime = System.currentTimeMillis();
         try {
-            List<Future<List<T>>> currentAgentFutures = this.executor.invokeAll(runnables);
-            for(int i = 0; i < currentAgentFutures.size(); i++) {
-                agentPlanActions.put(runnables.get(i).getAgentID(),
-                        currentAgentFutures.get(i).get().stream().filter(Objects::nonNull).collect(Collectors.toList())); // TODO will this work?
+            List<Future<ReschedulableResult<T>>> currentAgentFutures = this.executor.invokeAll(runnables);
+//            for(int i = 0; i < currentAgentFutures.size(); i++) {
+//                agentPlanActions.put(runnables.get(i).getAgentID(),
+//                        currentAgentFutures.get(i).get().getResult().stream().filter(Objects::nonNull).collect(Collectors.toList())); // TODO will this work?
+//            }
+            for(Future<ReschedulableResult<T>> futures : currentAgentFutures) {
+                ReschedulableResult<T> result = futures.get();
+                agentPlanActions.put(result.getAgentID(), result.getResult().stream().filter(Objects::nonNull).collect(Collectors.toList()));
+                if(result.isReschedule()) {
+                    this.scheduleForNextTick(result.getDeliberationRunnable());
+                }
             }
         } catch (InterruptedException | ExecutionException e) {
             e.printStackTrace();
         }
-//        for(DeliberationRunnable<T> dr : runnables) {
-//            try {
-//                List<T> currentAgentActions = this.executor.submit(dr).get();
-//                currentAgentActions = currentAgentActions.stream().filter(Objects::nonNull).collect(Collectors.toList());
-//
-//                List<String> currentAgentActionStrings = new ArrayList<>();
-//                for (Object action: currentAgentActions) {
-//                    currentAgentActionStrings.add((String) action);
-//                }
-//
-//                agentPlanActions.put(dr.getAgentID(), currentAgentActionStrings);
-//
-//            } catch (InterruptedException | ExecutionException e) {
-//                e.printStackTrace();
-//            }
-//        }
         this.stepDuration = (int) (System.currentTimeMillis() - startTime);
 
         tick++;
@@ -150,11 +141,12 @@ public int getLastTickDuration() {
     public List<AgentID> getScheduledAgents() {
         List<AgentID> scheduledAgents = new ArrayList<>();
         synchronized (this.scheduledRunnables) {
-            for(DeliberationRunnable runnable : this.scheduledRunnables) {
-                scheduledAgents.add(runnable.getAgentID());
-            }
+//            for(DeliberationRunnable<T> runnable : this.scheduledRunnables) {
+//                scheduledAgents.add(runnable.getAgentID());
+//            }
+            return new ArrayList<>(this.scheduledRunnables.keySet());
         }
-        return scheduledAgents;
+//        return scheduledAgents;
     }
 
     /**
diff --git a/src/nl/uu/cs/iss/ga/sim2apl/core/tick/matrix/MatrixAgentThread.java b/src/nl/uu/cs/iss/ga/sim2apl/core/tick/matrix/MatrixAgentThread.java
@@ -13,6 +13,7 @@
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
 import nl.uu.cs.iss.ga.sim2apl.core.deliberation.DeliberationRunnable;
+import nl.uu.cs.iss.ga.sim2apl.core.deliberation.ReschedulableResult;
 
 /**
  * An Agent Thread is responsible for producing events on behalf of a specific set of agents.
@@ -63,14 +64,25 @@ public void run() {
                 long startTime = System.currentTimeMillis();
                 
                 List<DeliberationRunnable<T>> runnables = inq.take();
-                List<Future<List<T>>> agentActionFutures = this.executor.invokeAll(runnables);
+                List<Future<ReschedulableResult<T>>> agentActionFutures = this.executor.invokeAll(runnables);
 
                 try {
-                    for(int i = 0; i < runnables.size(); i++) {
-                        List<String> agentActions = agentActionFutures.get(i).get().stream()
-                                .filter(Objects::nonNull).map(gson::toJson).collect(Collectors.toList());
+//                    for(int i = 0; i < runnables.size(); i++) {
+//                        List<String> agentActions = agentActionFutures.get(i).get().stream()
+//                                .filter(Objects::nonNull).map(gson::toJson).collect(Collectors.toList());
+//                        JsonObject update = new JsonObject();
+//                        update.addProperty("agentID", runnables.get(i).getAgentID().toString());
+//                        update.add("actions", gson.toJsonTree(agentActions, arrayListStringType));
+//                        JsonArray updates = new JsonArray();
+//                        updates.add(update);
+//                        this.proxy.register_events(agentproc_id, updates);
+//                    }
+                    for(Future<ReschedulableResult<T>> future : agentActionFutures) {
+                        ReschedulableResult<T> result = future.get();
+                        List<String> agentActions = result.getResult().stream().filter(Objects::nonNull)
+                                .map(gson::toJson).collect(Collectors.toList());
                         JsonObject update = new JsonObject();
-                        update.addProperty("agentID", runnables.get(i).getAgentID().toString());
+                        update.addProperty("agentID", result.getAgentID().toString());
                         update.add("actions", gson.toJsonTree(agentActions, arrayListStringType));
                         JsonArray updates = new JsonArray();
                         updates.add(update);

Original file line number	Diff line number	Diff line change
`@@ -5,11 +5,11 @@`
`5`	`5`	`*`
`6`	`6`	`* @author Bas Testerink`
`7`	`7`	`*/`
`8`		`-public final class SelfRescheduler {`
	`8`	`+public final class SelfRescheduler<T> {`
`9`	`9`	`/** The deliberation runnable that can be rescheduled. */`
`10`		`- private final DeliberationRunnable deliberationRunnable;`
	`10`	`+ private final DeliberationRunnable<T> deliberationRunnable;`
`11`	`11`
`12`		`- public SelfRescheduler(final DeliberationRunnable deliberationRunnable){`
	`12`	`+ public SelfRescheduler(final DeliberationRunnable<T> deliberationRunnable){`
`13`	`13`	`this.deliberationRunnable = deliberationRunnable;`
`14`	`14`	`}`
`15`	`15`