Changeset 32

Timestamp:

02/29/08 19:22:27 (17 years ago)

Author:

gkronber

Message:

worked on #2

Location:

trunk/sources

Files:

• HeuristicLab.DistributedEngine/DistributedEngine.cs (modified) (5 diffs)
• HeuristicLab.Grid/ClientForm.cs (modified) (7 diffs)
• HeuristicLab.Grid/EngineStore.cs (modified) (11 diffs)
• HeuristicLab.Grid/GridServer.cs (modified) (1 diff)
• HeuristicLab.Grid/HeuristicLab.Grid.csproj (modified) (1 diff)
• HeuristicLab.Grid/IClient.cs (added)
• HeuristicLab.Grid/IEngineStore.cs (modified) (1 diff)

trunk/sources/HeuristicLab.DistributedEngine/DistributedEngine.cs ¶

@@ -33 +33 @@
 namespace HeuristicLab.DistributedEngine {
   public class DistributedEngine : EngineBase, IEditable {
-    // currently executed operators
-    private IOperator[] currentOperators;
-    private int operatorIndex;
     private IGridServer server;
+    private Dictionary<Guid, AtomicOperation> runningEngines = new Dictionary<Guid, AtomicOperation>();
 
     private string serverAddress;
@@ -47 +45 @@
       }
     }
-
-    public DistributedEngine() {
-      currentOperators = new IOperator[1000];
-    }
-
 
     public override object Clone(IDictionary<Guid, object> clonedObjects) {
@@ -84 +77 @@
     public override void Abort() {
       base.Abort();
-      for(int i = 0; i < currentOperators.Length; i++) {
-        if(currentOperators[i] != null)
-          currentOperators[i].Abort();
+      foreach(Guid engineGuid in runningEngines.Keys) {
+        server.AbortEngine(engineGuid);
       }
     }
 
     protected override void ProcessNextOperation() {
-      operatorIndex = 1;
       ProcessNextOperation(myExecutionStack, 0);
     }
@@ -100 +91 @@
         IOperation next = null;
         try {
-          currentOperators[currentOperatorIndex] = atomicOperation.Operator;
           next = atomicOperation.Operator.Execute(atomicOperation.Scope);
         } catch(Exception ex) {
@@ -115 +105 @@
         CompositeOperation compositeOperation = (CompositeOperation)operation;
         if(compositeOperation.ExecuteInParallel) {
-          Dictionary<Guid, AtomicOperation> runningEngines = new Dictionary<Guid, AtomicOperation>();
           foreach(AtomicOperation parOperation in compositeOperation.Operations) {
             ProcessingEngine engine = new ProcessingEngine(OperatorGraph, GlobalScope, parOperation); // OperatorGraph not needed?

trunk/sources/HeuristicLab.Grid/ClientForm.cs ¶

@@ -34 +34 @@
 using System.IO;
 using System.IO.Compression;
+using System.Net;
 
 namespace HeuristicLab.Grid {
-  public partial class ClientForm : Form {
-
+  [ServiceBehavior(InstanceContextMode = InstanceContextMode.Single, ConcurrencyMode = ConcurrencyMode.Multiple, UseSynchronizationContext = false)]
+  public partial class ClientForm : Form, IClient {
     private ChannelFactory<IEngineStore> factory;
+    private ServiceHost clientHost;
     private System.Timers.Timer fetchOperationTimer;
     private IEngineStore engineStore;
+    private Guid currentGuid;
+    private ProcessingEngine currentEngine;
+    private string clientUrl;
 
     public ClientForm() {
@@ -48 +53 @@
       fetchOperationTimer.Elapsed += new System.Timers.ElapsedEventHandler(fetchOperationTimer_Elapsed);
       statusTextBox.Text = "Stopped";
+      currentGuid = Guid.Empty;
     }
 
     private void startButton_Click(object sender, EventArgs e) {
+      clientUrl = "net.tcp://" + Dns.GetHostAddresses(Dns.GetHostName())[0] + ":8002/Grid/Client";
+      clientHost = new ServiceHost(this, new Uri(clientUrl));
       try {
         NetTcpBinding binding = new NetTcpBinding();
@@ -56 +64 @@
         binding.ReaderQuotas.MaxStringContentLength = 100000000; // also 100M chars
         binding.ReaderQuotas.MaxArrayLength = 100000000; // also 100M elements;
-        binding.Security.Mode = SecurityMode.None;        
+        binding.Security.Mode = SecurityMode.None;
+
+        clientHost.AddServiceEndpoint(typeof(IClient), binding, clientUrl);
+        clientHost.Open();
+
         factory = new ChannelFactory<IEngineStore>(binding);
         engineStore = factory.CreateChannel(new EndpointAddress(addressTextBox.Text));
@@ -65 +77 @@
         statusTextBox.Text = "Waiting for engine";
 
-      } catch (Exception ex) {
+      } catch (CommunicationException ex) {
         MessageBox.Show("Exception while connecting to the server: " + ex.Message);
+        clientHost.Abort();
         startButton.Enabled = true;
         stopButton.Enabled = false;
@@ -76 +89 @@
       fetchOperationTimer.Stop();
       factory.Abort();
+      clientHost.Close();
       statusTextBox.Text = "Stopped";
       stopButton.Enabled = false;
@@ -83 +97 @@
     private void fetchOperationTimer_Elapsed(object sender, System.Timers.ElapsedEventArgs e) {
       byte[] engineXml;
-      Guid guid;
       fetchOperationTimer.Stop();
-      if (engineStore.TryTakeEngine(out guid, out engineXml)) {
-        ProcessingEngine engine = RestoreEngine(engineXml);
+      if (engineStore.TryTakeEngine(clientUrl, out currentGuid, out engineXml)) {
+        currentEngine = RestoreEngine(engineXml);
         if (InvokeRequired) { Invoke((MethodInvoker)delegate() { statusTextBox.Text = "Executing engine"; }); } else statusTextBox.Text = "Executing engine";
-        engine.Finished += delegate(object src, EventArgs args) {
-          byte[] resultScopeXml = SaveScope(engine.InitialOperation.Scope);
-          engineStore.StoreResult(guid, resultScopeXml);
+        currentEngine.Finished += delegate(object src, EventArgs args) {
+          byte[] resultScopeXml = SaveScope(currentEngine.InitialOperation.Scope);
+          engineStore.StoreResult(currentGuid, resultScopeXml);
+          currentGuid = Guid.Empty;
+          currentEngine = null;
           fetchOperationTimer.Interval = 100;
           fetchOperationTimer.Start();
         };
-        engine.Execute();
+        currentEngine.Execute();
       } else {
         if(InvokeRequired) { Invoke((MethodInvoker)delegate() { statusTextBox.Text = "Waiting for engine"; }); } else statusTextBox.Text = "Waiting for engine";
@@ -100 +115 @@
         fetchOperationTimer.Start();
       }
+    }
+    public void Abort(Guid guid) {
+      if(!IsRunningEngine(guid)) return;
+      currentEngine.Abort();
+    }
+    public bool IsRunningEngine(Guid guid) {
+      return currentGuid == guid;
     }
     private ProcessingEngine RestoreEngine(byte[] engine) {

trunk/sources/HeuristicLab.Grid/EngineStore.cs ¶

@@ -27 +27 @@
 
 namespace HeuristicLab.Grid {
-  [ServiceBehavior(InstanceContextMode = InstanceContextMode.Single, ConcurrencyMode = ConcurrencyMode.Multiple, UseSynchronizationContext=false)]
+  [ServiceBehavior(InstanceContextMode = InstanceContextMode.Single, ConcurrencyMode = ConcurrencyMode.Multiple, UseSynchronizationContext = false)]
   public class EngineStore : IEngineStore {
     private Queue<Guid> engineQueue;
@@ -33 +33 @@
     private Dictionary<Guid, byte[]> runningEngines;
     private Dictionary<Guid, byte[]> results;
+    private Dictionary<Guid, string> runningClients;
     private object bigLock;
+    private ChannelFactory<IClient> clientChannelFactory;
 
     private event EventHandler ResultRecieved;
@@ -59 +61 @@
       waitingEngines = new Dictionary<Guid, byte[]>();
       runningEngines = new Dictionary<Guid, byte[]>();
+      runningClients = new Dictionary<Guid, string>();
       results = new Dictionary<Guid, byte[]>();
       bigLock = new object();
+
+      NetTcpBinding binding = new NetTcpBinding();
+      binding.MaxReceivedMessageSize = 100000000; // 100Mbytes
+      binding.ReaderQuotas.MaxStringContentLength = 100000000; // also 100M chars
+      binding.ReaderQuotas.MaxArrayLength = 100000000; // also 100M elements;
+      binding.Security.Mode = SecurityMode.None;
+
+      clientChannelFactory = new ChannelFactory<IClient>(binding);
     }
 
-    public bool TryTakeEngine(out Guid guid, out byte[] engine) {
-      lock (bigLock) {
-        if (engineQueue.Count == 0) {
+    public bool TryTakeEngine(string clientUrl, out Guid guid, out byte[] engine) {
+      lock(bigLock) {
+        if(engineQueue.Count == 0) {
           guid = Guid.Empty;
           engine = null;
@@ -74 +85 @@
           waitingEngines.Remove(guid);
           runningEngines[guid] = engine;
+          runningClients[guid] = clientUrl;
           return true;
         }
@@ -80 +92 @@
 
     public void StoreResult(Guid guid, byte[] result) {
-      lock (bigLock) {
-        if (!runningEngines.ContainsKey(guid)) return; // ignore result when the engine is not known to be running
+      lock(bigLock) {
+        if(!runningEngines.ContainsKey(guid)) return; // ignore result when the engine is not known to be running
 
         runningEngines.Remove(guid);
+        runningClients.Remove(guid);
         results[guid] = result;
         OnResultRecieved(guid);
@@ -90 +103 @@
 
     internal void AddEngine(Guid guid, byte[] engine) {
-      lock (bigLock) {
+      lock(bigLock) {
         engineQueue.Enqueue(guid);
         waitingEngines.Add(guid, engine);
@@ -97 +110 @@
 
     internal byte[] RemoveResult(Guid guid) {
-      lock (bigLock) {
+      lock(bigLock) {
         byte[] result = results[guid];
         results.Remove(guid);
@@ -106 +119 @@
     internal byte[] GetResult(Guid guid) {
       ManualResetEvent waitHandle = new ManualResetEvent(false);
-      lock (bigLock) {
-        if (results.ContainsKey(guid)) {
+      lock(bigLock) {
+        if(results.ContainsKey(guid)) {
           byte[] result = results[guid];
           results.Remove(guid);
@@ -114 +127 @@
           ResultRecieved += delegate(object source, EventArgs args) {
             ResultRecievedEventArgs resultArgs = (ResultRecievedEventArgs)args;
-            if (resultArgs.resultGuid == guid) {
+            if(resultArgs.resultGuid == guid) {
               waitHandle.Set();
             }
@@ -124 +137 @@
       waitHandle.Close();
 
-      lock (bigLock) {
+      lock(bigLock) {
         byte[] result = results[guid];
         results.Remove(guid);
@@ -131 +144 @@
     }
 
+    internal void AbortEngine(Guid guid) {
+      string clientUrl = "";
+      lock(bigLock) {
+        if(runningClients.ContainsKey(guid)) {
+          clientUrl = runningClients[guid];
+        }
+
+        if(clientUrl != "") {
+          IClient client = clientChannelFactory.CreateChannel(new EndpointAddress(clientUrl));
+          client.Abort(guid);
+        }
+      }
+    }
+
     private void OnResultRecieved(Guid guid) {
       ResultRecievedEventArgs args = new ResultRecievedEventArgs();
       args.resultGuid = guid;
-      if (ResultRecieved != null) {
+      if(ResultRecieved != null) {
         ResultRecieved(this, args);
       }

trunk/sources/HeuristicLab.Grid/GridServer.cs ¶

@@ -46 +46 @@
 
     public void AbortEngine(Guid engine) {
-      throw new NotImplementedException();
+      engineStore.AbortEngine(engine);
     }
   }

trunk/sources/HeuristicLab.Grid/HeuristicLab.Grid.csproj ¶

@@ -63 +63 @@
     <Compile Include="GridServerApplication.cs" />
     <Compile Include="HeuristicLabGridPlugin.cs" />
+    <Compile Include="IClient.cs" />
     <Compile Include="IEngineStore.cs" />
     <Compile Include="IGridServer.cs" />

trunk/sources/HeuristicLab.Grid/IEngineStore.cs ¶

@@ -29 +29 @@
   [ServiceContract(Namespace = "http://HeuristicLab.Grid")]
   interface IEngineStore {
-
     [OperationContract]
-    bool TryTakeEngine(out Guid guid, out byte[] engine);
+    bool TryTakeEngine(string clientUrl, out Guid guid, out byte[] engine);
 
     [OperationContract]