MicroService handling of last tick, GPUService subscription to events, partial GPU handling of events

Author: tohanov

pom.xml

@@ -34,7 +34,7 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-jar-plugin</artifactId>
-                <version>3.0.2</version>            
+                <version>3.2.0</version>            
                 <configuration>
                     <archive>
                         <manifest>

src/main/java/bgu/spl/mics/MicroService.java

@@ -30,7 +30,7 @@ public abstract class MicroService implements Runnable {
     private boolean terminated = false;
     private final String name;
     private MessageBusImpl messageBus;
-    private HashMap<Class<? extends Object>, Callback<? extends Object>> callbackHashMap;
+    private HashMap<Class<? extends Message>, Callback<? extends Message>> callbackHashMap;
 	private ConcurrentLinkedQueue<Callback<Message>> callbackQueue;
 
 
@@ -71,7 +71,7 @@ public MicroService(String name) {
      */
     protected final <T, E extends Event<T>> void subscribeEvent(Class<E> type, Callback<E> callback) {
        messageBus.subscribeEvent(type, this);
-       callbackHashMap.put((Class<? extends Object>)type, (Callback<? extends Object>)callback);
+       callbackHashMap.put((Class<? extends Message>)type, (Callback<? extends Message>)callback);
     }
 
 
@@ -97,7 +97,7 @@ protected final <T, E extends Event<T>> void subscribeEvent(Class<E> type, Callb
      */
     protected final <B extends Broadcast> void subscribeBroadcast(Class<B> type, Callback<B> callback) {
         messageBus.subscribeBroadcast(type, this);
-        callbackHashMap.put((Class<? extends Object>)type, (Callback<? extends Object>)callback);
+        callbackHashMap.put((Class<? extends Message>)type, (Callback<? extends Message>)callback);
     }
 
 
@@ -187,20 +187,16 @@ public final void run() {
 			try {
 				Message message = messageBus.awaitMessage(this);
 				Class<? extends Message> messageClass = message.getClass();
-/*
-				if (messageClass != TickBroadcast.class) {
-					// TODO: put aside the message into an internal queue
-				}
-				else { // got a tick
-					if (((TickBroadcast)message).isLast()) {
-						terminate();
-					}
-					else {
-						// TODO: operate on the mesages that were put aside to the internal queue
-					}
+
+				// TODO: think if should let the services handle the last tick as well or terminate everything without that
+				((Callback<Message>)callbackHashMap.get(messageClass)).call(message);
+				
+				if (message instanceof TickBroadcast && ((TickBroadcast)message).isLast()) {
+					terminate();
 				}
-				*/
-				callbackHashMap.get(messageClass).call(message);
+				// else {
+				// 	((Callback<Message>)callbackHashMap.get(messageClass)).call(message);
+				// }
 			}
 			catch (InterruptedException e) { 
 				synchronized (System.out) { 

src/main/java/bgu/spl/mics/application/objects/GPU.java

@@ -1,34 +1,53 @@
 package bgu.spl.mics.application.objects;
 
+import java.util.ArrayDeque;
 import java.util.Collection;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.experimental.theories.Theories;
 
 import bgu.spl.mics.MessageBusImpl;
 import bgu.spl.mics.application.services.GPUService;
+import bgu.spl.mics.application.messages.TestModelEvent;
+import bgu.spl.mics.application.messages.TickBroadcast;
+import bgu.spl.mics.application.messages.TrainModelEvent;
+import bgu.spl.mics.Callback;
+import bgu.spl.mics.Event;
 
 /**
  * Passive object representing a single GPU.
  * Add all the fields described in the assignment as private fields.
  * Add fields and methods to this class as you see fit (including public methods and constructors).
  */
 public class GPU {
+	
     /**
      * Enum representing the type of the GPU.
      */
-    public enum Type {RTX3090, RTX2080, GTX1080}
+    public enum Type { RTX3090, RTX2080, GTX1080 }
 
 
 	// region According to assignment instructions
     private Type type;
-    private Cluster cluster;
-    private Model model;
+    private Model model; // the model being taken care of currently
+    private static Cluster cluster = Cluster.getInstance();
 	// endregion According to assignment instructions
 
 
 	// region Added fields
-    private GPUService service;
+    // private GPUService service;
 	// private Collection<DataBatch> processedBatches;
-	private final byte vRAM;
-	private int processedBatchesNum;
+	private Queue<Event<Model>> modelEventsQueue;
+	private final int trainingDelay; // according to the type of the gpu
+	private final byte vRAM; // according to the type of the gpu
+	private int storedProcessedBatchesNumber;
+	private int ticksToTrainBatch;
+	private boolean training;
+	private boolean testing;
 	// endregion Added fields
 
 	/**
@@ -58,7 +77,57 @@ public enum Type {RTX3090, RTX2080, GTX1080}
     // }
 
 
-    private Type typeFromString(String _type) {
+	/**
+     * @return The type of the GPU
+     */
+    // public Type getType () {
+    //     return type;
+    // }
+
+
+	// public void runService() { // TODO remove ?
+	// 	service.run();
+	// }
+
+
+	/**
+	 * @return Reference to the corresponding GPUService
+	 */
+	// public GPUService getService() {
+	// 	return service;
+	// }
+
+
+	// public Collection<DataBatch> getProcessedBatches() {
+	// 	return processedBatches;
+	// }
+
+
+	// region for serialization from json
+	public GPU(String _type) {
+        this.type = typeFromString(_type);
+		model = null;
+		storedProcessedBatchesNumber = 0;
+
+		switch(type) {
+			case GTX1080:
+				vRAM = 8;
+				trainingDelay = 4;
+				break;
+			case RTX2080:
+				vRAM = 16;
+				trainingDelay = 2;
+				break;
+			default : // RTX3090:
+				vRAM = 32;
+				trainingDelay = 1;
+		}
+
+		modelEventsQueue = new ArrayDeque<>();
+    }
+
+
+	private Type typeFromString(String _type) {
 		Type returnType;
 		String uppercaseType = _type.toUpperCase();
 
@@ -71,75 +140,75 @@ else if (uppercaseType == "RTX2080")
 
 		return returnType;
 	}
+	// endregion for serialization from json
 
 
-	/**
-     * @return The type of the GPU
-     */
-    public Type getType () {
-        return type;
-    }
+	public void gotTick() {
+		// TODO: treat the case of last tick??
 
+		// TODO: split to smaller functions (queries + actions)
+		if (training) { // && storedProcessedBatchesNumber != 0) {
+			--ticksToTrainBatch;
+			
+			if (ticksToTrainBatch == 0) { // if finished training batch
+				ticksToTrainBatch = trainingDelay; // reset the counter
+				finishTrainingBatch();	// 
 
-	// public void runService() { // TODO remove ?
-	// 	service.run();
-	// }
+				if (storedProcessedBatchesNumber == 0) {
+					training = false; // TODO: maybe incorrect since someone could tell me to test a model while i haven't gotten all the batches from cpu yet
+					// TODO set Future to done / send more batches to cpu for processing
+				}
+			}
+		}
+		else if (testing) {
+			// TODO
+		}
+		// else if (storedProcessedBatchesNumber != 0) {
+		// 	training = true;
+		// 	--ticksToTrainBatch;
+
+		// 	if (trainingDelay == 0) { 
+		// 		gpu.finishTrainingBatch();
+
+		// 		if (gpu.getProcessedBatchesNum() == 0) {
+		// 			// TODO set Future to done
+		// 		}
+		// 	}
+		// }
+	}
 
 
-	/**
-	 * @return Reference to the corresponding GPUService
-	 */
-	public GPUService getService() {
-		return service;
+	public void gotModelEvent(Event<Model> modelEvent) {
+		// TODO: put aside to wait for ticks
+		modelEventsQueue.add(modelEvent);
 	}
 
 
-	// public Collection<DataBatch> getProcessedBatches() {
-	// 	return processedBatches;
+	// public void gotModelToTest(TestModelEvent testModelEvent) {
+	// 	// TODO: put aside to wait for ticks
 	// }
 
 
 	/**
 	 * @return The amount of vRAM the GPU has
 	 */
-	public byte getVRAM() {
-		return vRAM;
-	}
+	// public byte getVRAM() {
+	// 	return vRAM;
+	// }
 
+	
 	/**
 	 * @return Number of processed batches currently in training
 	 */
-	public int getProcessedBatchesNum() {
-		return processedBatchesNum;
-	}
+	// public int getProcessedBatchesNum() {
+	// 	return storedProcessedBatchesNumber;
+	// }
 
 
 	/**
 	 * @inv processedBatchesNum >= 0
 	 */
 	public void finishTrainingBatch() {
-		--processedBatchesNum;
+		--storedProcessedBatchesNumber;
 	}
-
-
-	// region for serialization from json
-	public GPU(String _type) {
-        this.type = typeFromString(_type);
-		// this.service = new GPUService(_name, this);
-
-		processedBatchesNum = 0;
-
-		switch(type) {
-			case GTX1080:
-				vRAM = 8;
-				break;
-			case RTX2080:
-				vRAM = 16;
-				break;
-			default : // RTX3090:
-				vRAM = 32;
-		}
-    }
-	// endregion for serialization from json
-
-}
+}