Changeset 103 for java/main/src/main/java/com/framsticks/experiment/WorkPackageLogic.java

Timestamp:

07/18/13 23:52:25 (11 years ago)

Author:

psniegowski

Message:

HIGHLIGHTS:

• add auto loading and saving algorithms between

frams files format and Java classes

• respect ValueChange? events in GUI (do not reload object)
• support results of procedures in Java server
• make Experiment automatically convert between frams file and NetFile? object
• add MessageLogger? (compatible with original frams server messages)
• WorkPackageLogic? now validates results, is able to discard them, reschedule

whole package, or only uncomputed remainder

• show primes computed in PrimeExperiment? live in GUI

CHANGELOG:
Show just a short description in PrimeExperiment?.

Add primes_changed event to the PrimeExperiment?.

Make WorkPackageLogic? robust to frams server returning invalid results.

Add MessageLogger? to logics.

Add NetFile? interface. Support Messages from server.

Minor changes to connections.

Merge results in the PrimeExperiment?.

More netload class->file conversion to Simulator.

Move netsave parsing to Simulator.

Fix bug with inverted ordering of events firing in Experiment.

Minor changes.

Minor logging changes.

Use AccessOperations?.convert in NetLoadSaveLogic?

NetLoadSaveLogic? now encloses the conversion.

Use more generic AccessOperations? saveAll and loadAll in PrimePackage?.

Add Result class for enclosing of call invocations' results.

Improve feature request handling in Connections.

Use AccessOperations?.convert in RemoteTree? events parsing.

Minor change.

Add some information params to Java server root and CLI objects.

A draft implementation of loadAll algorithm.

That algorithm tries to load objects into a tree structure.

Add AccessOperationsTest? test.

Develop WorkPackageLogic?.

• add state tracking fields
• add work package generation

Add utility class SimplePrimitive?.

Meant for Java backend classes, enclose a single primitive value
and set of listeners.

Improve primitive value refresh in GUI.

When ValueChange? found in called event, do not reload whole
object, but only update GUI (no communication is performed).

Use ValueChange? in the TestClass? test.

Minor changes.

Sending all packages in PrimeExperiment? to the frams servers.

Develop AccessOperations?.loadComposites().

Remove addAccess from MultiParamLoader? interface.

There is now no default AccessProvider? in MultiParamLoader?.
User must explicitely set AccessStash? or Registry.

Improve saving algorithms in AccessOperations?.

File:

• java/main/src/main/java/com/framsticks/experiment/WorkPackageLogic.java (modified) (4 diffs)

java/main/src/main/java/com/framsticks/experiment/WorkPackageLogic.java

@@ -1 +1 @@
 package com.framsticks.experiment;
+
+import java.util.IdentityHashMap;
+import java.util.LinkedList;
+import java.util.Map;
 
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.LogManager;
 
-import com.framsticks.communication.File;
-import com.framsticks.params.ListSource;
+import com.framsticks.core.Message;
+import com.framsticks.core.ValueChange;
+import com.framsticks.params.EventListener;
+import com.framsticks.params.MessageLogger;
+import com.framsticks.params.SimplePrimitive;
 import com.framsticks.params.annotations.FramsClassAnnotation;
 import com.framsticks.params.annotations.ParamAnnotation;
@@ -11 +18 @@
 import com.framsticks.util.dispatching.FutureHandler;
 
-@FramsClassAnnotation
-public abstract class WorkPackageLogic<WP extends WorkPackage> extends AbstractExperimentLogic {
+@FramsClassAnnotation(order = {"netLoadSaveLogic", "newestTaskScheduled", "newestResultReceived"})
+public abstract class WorkPackageLogic<WP extends WorkPackage<WP>> extends AbstractExperimentLogic {
 
         private static final Logger log = LogManager.getLogger(WorkPackageLogic.class);
 
         @ParamAnnotation
-        public final NetLoadSaveLogic netLoadSaveLogic;
+        public final NetLoadSaveLogic<WP> netLoadSaveLogic;
+
+        protected final Class<WP> packageJavaClass;
+
+        protected final MessageLogger messages = new MessageLogger(WorkPackageLogic.class);
+
+        protected final Map<Simulator, WP> sentPackages = new IdentityHashMap<>();
+        protected final LinkedList<WP> queuedPackages = new LinkedList<>();
+
+        protected final SimplePrimitive<String> newestTaskSent = new SimplePrimitive<>();
+        protected final SimplePrimitive<String> newestResultReceived = new SimplePrimitive<>();
+
 
         /**
          * @param experiment
          */
-        public WorkPackageLogic(Experiment experiment) {
+        public WorkPackageLogic(Experiment experiment, Class<WP> packageJavaClass) {
                 super(experiment);
+                this.packageJavaClass = packageJavaClass;
 
-                netLoadSaveLogic = new NetLoadSaveLogic(experiment) {
+                netLoadSaveLogic = new NetLoadSaveLogic<WP>(experiment, packageJavaClass) {
 
                         @Override
-                        public void netload(Simulator simulator, final Future<File> net) {
+                        public void netload(final Simulator simulator, final Future<WP> netFuture) {
                                 assert experiment.isActive();
                                 log.debug("providing netload file for {}", simulator);
 
-                                generateNextPackage(new FutureHandler<WP> (net) {
+                                findNextPackage(new FutureHandler<WP> (netFuture) {
 
                                         @Override
-                                        protected void result(WP result) {
-                                                log.debug("sending package: {}", result);
-                                                File file = new File("test-file", ListSource.createFrom("# a test"));
+                                        protected void result(WP net) {
+                                                assert experiment.isActive();
 
-                                                net.pass(file);
+                                                if (net == null) {
+                                                        log.debug("no more packages left");
+                                                        return;
+                                                }
+                                                WorkPackageLogic.this.messages.info("netload", net.sumUpTask());
+                                                log.debug("sending package: {}", net);
+                                                newestTaskSent.set(net.sumUpTask());
+                                                sentPackages.put(simulator, net);
+
+                                                netFuture.pass(net);
                                         }
                                 });
@@ -46 +73 @@
 
                         @Override
-                        public void netsave(Simulator simulator, File net) {
+                        public void netsave(Simulator simulator, WP netResult) {
+                                assert experiment.isActive();
+
+                                log.debug("received package from {}: {}", simulator, netResult);
+                                WorkPackageLogic.this.messages.info("netsave", netResult.toString());
+
+                                WP netSent = sentPackages.get(simulator);
+                                if (netSent == null) {
+                                        log.error("no task found in {} for received result {}", simulator, netResult);
+                                        return;
+                                }
+                                sentPackages.remove(simulator);
+
+                                try {
+                                        WP netRemainder = netSent.getRemainder(netResult);
+                                        if (netRemainder != null) {
+                                                log.warn("queueing remainder: {}", netRemainder);
+                                                queuedPackages.add(netRemainder);
+                                        }
+                                } catch (InvalidWorkPackage e) {
+                                        log.error("in simulator {}, result {} is in no relation to task {} ({}), rescheduling", simulator, netResult, netSent, e.getShortMessage(new StringBuilder()));
+                                        queuedPackages.add(netSent);
+                                        return;
+                                }
+
+                                returnPackage(netResult);
                                 // processFile();
                         }
@@ -53 +105 @@
 
 
+
+        protected void findNextPackage(final Future<WP> future) {
+                if (!queuedPackages.isEmpty()) {
+                        WP workPackage = queuedPackages.pollFirst();
+                        future.pass(workPackage);
+                        return;
+                }
+
+                generateNextPackage(new FutureHandler<WP>(experiment) {
+
+                        @Override
+                        protected void result(WP result) {
+                                if (result == null) {
+                                        log.info("no more packages left");
+                                        future.pass(null);
+                                        return;
+                                }
+                                future.pass(result);
+                        }
+                });
+        }
+
         protected abstract void generateNextPackage(Future<WP> future);
+        protected abstract void returnPackage(WP workPackage);
+
+        /**
+         * @return the newestTaskScheduled
+         */
+        @ParamAnnotation(id = "newest_task_sent")
+        public String getNewestTaskSent() {
+                return newestTaskSent.get();
+        }
+
+        /**
+         * @return the newestResultReceived
+         */
+        @ParamAnnotation(id = "newest_result_received")
+        public String getNewestResultReceived() {
+                return newestResultReceived.get();
+        }
+
+        @ParamAnnotation(id = "newest_task_sent_changed")
+        public void addNewestTaskSentListener(EventListener<ValueChange> listener) {
+                newestTaskSent.addListener(listener);
+        }
+
+        @ParamAnnotation(id = "newest_task_sent_changed")
+        public void removeNewestTaskSentListener(EventListener<ValueChange> listener) {
+                newestTaskSent.removeListener(listener);
+        }
+
+        @ParamAnnotation(id = "newest_result_received_changed")
+        public void addNewestResultReceivedListener(EventListener<ValueChange> listener) {
+                newestResultReceived.addListener(listener);
+        }
+
+        @ParamAnnotation(id = "newest_result_received_changed")
+        public void removeNewestResultReceivedListener(EventListener<ValueChange> listener) {
+                newestResultReceived.removeListener(listener);
+        }
+
+        @ParamAnnotation(id = "messages")
+        public void addMessageListener(EventListener<Message> listener) {
+                messages.add(listener);
+        }
+
+        @ParamAnnotation(id = "messages")
+        public void removeMessageListener(EventListener<Message> listener) {
+                messages.remove(listener);
+        }
+
+
 
 }