Changeset 228

Timestamp:

05/09/08 16:22:33 (17 years ago)

Author:

gkronber

Message:

bug fixing in DistributedEngine and Grid-Infrastructure

Location:

trunk/sources

Files:

• HeuristicLab.DistributedEngine/DistributedEngine.cs (modified) (1 diff)
• HeuristicLab.DistributedEngine/JobManager.cs (modified) (4 diffs)
• HeuristicLab.Grid/ClientForm.cs (modified) (5 diffs)

trunk/sources/HeuristicLab.DistributedEngine/DistributedEngine.cs

@@ -89 +89 @@
         CompositeOperation compositeOperation = (CompositeOperation)operation;
         if(compositeOperation.ExecuteInParallel) {
-          WaitHandle[] waithandles = new WaitHandle[compositeOperation.Operations.Count];
-          int i = 0;
-          foreach(AtomicOperation parOperation in compositeOperation.Operations) {
-            waithandles[i++] = jobManager.BeginExecuteOperation(OperatorGraph, GlobalScope, parOperation);
-          }
-          // WaitAll works only with maximally 64 waithandles
-          if(waithandles.Length <= 64) {
-            WaitHandle.WaitAll(waithandles);
-          } else {
-            for(i = 0; i < waithandles.Length; i++) {
-              waithandles[i].WaitOne();
+          try {
+            WaitHandle[] waithandles = new WaitHandle[compositeOperation.Operations.Count];
+            int i = 0;
+            foreach(AtomicOperation parOperation in compositeOperation.Operations) {
+              waithandles[i++] = jobManager.BeginExecuteOperation(OperatorGraph, GlobalScope, parOperation);
             }
-          }
-          if(jobManager.Exception != null) {
+            // WaitAll works only with maximally 64 waithandles
+            if(waithandles.Length <= 64) {
+              WaitHandle.WaitAll(waithandles);
+            } else {
+              for(i = 0; i < waithandles.Length; i++) {
+                waithandles[i].WaitOne();
+              }
+            }
+            if(jobManager.Exception != null) {
+              myExecutionStack.Push(compositeOperation);
+              Abort();
+              ThreadPool.QueueUserWorkItem(delegate(object state) { OnExceptionOccurred(jobManager.Exception); });
+            }
+          } catch(Exception e) {
             myExecutionStack.Push(compositeOperation);
             Abort();

trunk/sources/HeuristicLab.DistributedEngine/JobManager.cs

@@ -21 +21 @@
     private const int MAX_RESTARTS = 5;
     private Exception exception;
+    private ChannelFactory<IGridServer> factory;
     public Exception Exception {
       get { return exception; }
@@ -31 +32 @@
     internal void Reset() {
       lock(locker) {
-        // open a new channel
-        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;
-        ChannelFactory<IGridServer> factory = new ChannelFactory<IGridServer>(binding);
-        server = factory.CreateChannel(new EndpointAddress(address));
-
+        ResetConnection();
         foreach(WaitHandle wh in waithandles.Values) wh.Close();
         waithandles.Clear();
@@ -48 +41 @@
     }
 
+    private void ResetConnection() {
+      // open a new channel
+      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;
+      factory = new ChannelFactory<IGridServer>(binding);
+      server = factory.CreateChannel(new EndpointAddress(address));
+    }
+
     public WaitHandle BeginExecuteOperation(IOperatorGraph operatorGraph, IScope globalScope, AtomicOperation operation) {
       ProcessingEngine engine = new ProcessingEngine(operatorGraph, globalScope, operation); // OperatorGraph not needed?
       MemoryStream memStream = new MemoryStream();
       GZipStream stream = new GZipStream(memStream, CompressionMode.Compress, true);
-      PersistenceManager.Save(engine, stream); // Careful! Make sure that persistence is thread-safe!
+      PersistenceManager.Save(engine, stream);
       stream.Close();
+      if(factory.State != CommunicationState.Opened)
+        ResetConnection();
       Guid currentEngineGuid = server.BeginExecuteEngine(memStream.ToArray());
       lock(locker) {
@@ -69 +75 @@
       do {
         try {
+          if(factory.State != CommunicationState.Opened) ResetConnection();
           byte[] resultXml = server.TryEndExecuteEngine(engineGuid, 100);
           if(resultXml != null) {

trunk/sources/HeuristicLab.Grid/ClientForm.cs

@@ -40 +40 @@
   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() {
@@ -57 +55 @@
 
     private void startButton_Click(object sender, EventArgs e) {
-      string hostname = Dns.GetHostName();
-      IPAddress[] addresses = Dns.GetHostAddresses(hostname);
-
-      // windows XP returns the external ip on index 0 while windows vista returns the external ip on index 2
-      if(System.Environment.OSVersion.Version.Major >= 6) {
-        clientUrl = "net.tcp://" + Dns.GetHostAddresses(Dns.GetHostName())[2] + ":" + clientPort.Text + "/Grid/Client";
-      } else {
-        clientUrl = "net.tcp://" + Dns.GetHostAddresses(Dns.GetHostName())[0] + ":" + clientPort.Text + "/Grid/Client";
-      }
-
-      clientHost = new ServiceHost(this, new Uri(clientUrl));
       try {
-        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;
-
-        clientHost.AddServiceEndpoint(typeof(IClient), binding, clientUrl);
-        clientHost.Open();
-
-        factory = new ChannelFactory<IEngineStore>(binding);
-        engineStore = factory.CreateChannel(new EndpointAddress(addressTextBox.Text));
-
+        ResetConnection();
         fetchOperationTimer.Start();
         startButton.Enabled = false;
@@ -88 +64 @@
       } catch(CommunicationException ex) {
         MessageBox.Show("Exception while connecting to the server: " + ex.Message);
-        clientHost.Abort();
         startButton.Enabled = true;
         stopButton.Enabled = false;
@@ -95 +70 @@
     }
 
+    private void ResetConnection() {
+      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;
+      factory = new ChannelFactory<IEngineStore>(binding);
+      engineStore = factory.CreateChannel(new EndpointAddress(addressTextBox.Text));
+    }
+
     private void stopButton_Click(object sender, EventArgs e) {
       fetchOperationTimer.Stop();
       factory.Abort();
-      clientHost.Close();
       statusTextBox.Text = "Stopped";
       stopButton.Enabled = false;
@@ -108 +92 @@
       fetchOperationTimer.Stop();
       try {
+        if(factory.State != CommunicationState.Opened) {
+          ResetConnection();
+        }
         if(engineStore.TryTakeEngine(out currentGuid, out engineXml)) {
           currentEngine = RestoreEngine(engineXml);