Changeset 279 for branches/3.0/sources/HeuristicLab.DistributedEngine/JobManager.cs

Timestamp:

05/30/08 13:41:27 (16 years ago)

Author:

gkronber

Message:

merged changesets r219:r228, r240, r241:258, r263:265, r267,r268, r269 from trunk into the HL3 stable branch

File:

• branches/3.0/sources/HeuristicLab.DistributedEngine/JobManager.cs (copied) (copied from trunk/sources/HeuristicLab.DistributedEngine/JobManager.cs) (4 diffs)

branches/3.0/sources/HeuristicLab.DistributedEngine/JobManager.cs

@@ -1 @@
-using System;
+#region License Information
+/* HeuristicLab
+ * Copyright (C) 2002-2008 Heuristic and Evolutionary Algorithms Laboratory (HEAL)
+ *
+ * This file is part of HeuristicLab.
+ *
+ * HeuristicLab is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * HeuristicLab is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with HeuristicLab. If not, see <http://www.gnu.org/licenses/>.
+ */
+#endregion
+
+using System;
 using System.Collections.Generic;
 using System.Linq;
@@ -15 +36 @@
     private IGridServer server;
     private string address;
-    private Dictionary<Guid, AtomicOperation> engineOperations = new Dictionary<Guid, AtomicOperation>();
-    private Dictionary<Guid, byte[]> runningEngines = new Dictionary<Guid, byte[]>();
+    private Dictionary<Guid, ProcessingEngine> engines = new Dictionary<Guid, ProcessingEngine>();
     private Dictionary<Guid, ManualResetEvent> waithandles = new Dictionary<Guid, ManualResetEvent>();
-    private object locker = new object();
+    private Dictionary<AtomicOperation, byte[]> results = new Dictionary<AtomicOperation, byte[]>();
+    private object connectionLock = new object();
+    private object dictionaryLock = new object();
+
     private const int MAX_RESTARTS = 5;
-    private Exception exception;
+    private const int MAX_CONNECTION_RETRIES = 10;
+    private const int RETRY_TIMEOUT_SEC = 10;
+    private const int CHECK_RESULTS_TIMEOUT = 10;
+
     private ChannelFactory<IGridServer> factory;
-    public Exception Exception {
-      get { return exception; }
-    }
 
     public JobManager(string address) {
@@ -31 +54 @@
 
     internal void Reset() {
-      lock(locker) {
-        ResetConnection();
+      ResetConnection();
+      lock(dictionaryLock) {
         foreach(WaitHandle wh in waithandles.Values) wh.Close();
         waithandles.Clear();
-        engineOperations.Clear();
-        runningEngines.Clear();
-        exception = null;
+        engines.Clear();
+        results.Clear();
       }
     }
 
     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?
+      lock(connectionLock) {
+        // 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(IScope globalScope, AtomicOperation operation) {
+      ProcessingEngine engine = new ProcessingEngine(globalScope, operation);
+      byte[] zippedEngine = ZipEngine(engine);
+      Guid currentEngineGuid = Guid.Empty;
+      bool success = false;
+      int retryCount = 0;
+      do {
+        lock(connectionLock) {
+          try {
+            currentEngineGuid = server.BeginExecuteEngine(zippedEngine);
+            success = true;
+          } catch(TimeoutException timeoutException) {
+            if(retryCount < MAX_CONNECTION_RETRIES) {
+              Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
+              retryCount++;
+            } else {
+              throw new ApplicationException("Max retries reached.", timeoutException);
+            }
+          } catch(CommunicationException communicationException) {
+            ResetConnection();
+            // wait some time and try again (limit with maximal retries if retry count reached throw exception -> engine can decide to stop execution)
+            if(retryCount < MAX_CONNECTION_RETRIES) {
+              Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
+              retryCount++;
+            } else {
+              throw new ApplicationException("Max retries reached.", communicationException);
+            }
+          }
+        }
+      } while(!success);
+      lock(dictionaryLock) {
+        engines[currentEngineGuid] = engine;
+        waithandles[currentEngineGuid] = new ManualResetEvent(false);
+      }
+      ThreadPool.QueueUserWorkItem(new WaitCallback(TryGetResult), currentEngineGuid);
+      return waithandles[currentEngineGuid];
+    }
+
+    private byte[] ZipEngine(ProcessingEngine engine) {
       MemoryStream memStream = new MemoryStream();
       GZipStream stream = new GZipStream(memStream, CompressionMode.Compress, true);
       PersistenceManager.Save(engine, stream);
       stream.Close();
-      if(factory.State != CommunicationState.Opened)
-        ResetConnection();
-      Guid currentEngineGuid = server.BeginExecuteEngine(memStream.ToArray());
-      lock(locker) {
-        runningEngines[currentEngineGuid] = memStream.ToArray();
-        engineOperations[currentEngineGuid] = operation;
-        waithandles[currentEngineGuid] = new ManualResetEvent(false);
-        ThreadPool.QueueUserWorkItem(new WaitCallback(TryGetResult), currentEngineGuid);
-      }
-      return waithandles[currentEngineGuid];
+      byte[] zippedEngine = memStream.ToArray();
+      memStream.Close();
+      return zippedEngine;
+    }
+
+    public IScope EndExecuteOperation(AtomicOperation operation) {
+      byte[] zippedResult = null;
+      lock(dictionaryLock) {
+        zippedResult = results[operation];
+        results.Remove(operation);
+      }
+      // restore the engine 
+      using(GZipStream stream = new GZipStream(new MemoryStream(zippedResult), CompressionMode.Decompress)) {
+        ProcessingEngine resultEngine = (ProcessingEngine)PersistenceManager.Load(stream);
+        return resultEngine.InitialOperation.Scope;
+      }      
     }
 
@@ -74 +142 @@
       int restartCounter = 0;
       do {
-        try {
-          if(factory.State != CommunicationState.Opened) ResetConnection();
-          byte[] resultXml = server.TryEndExecuteEngine(engineGuid, 100);
-          if(resultXml != null) {
-            // restore the engine 
-            GZipStream stream = new GZipStream(new MemoryStream(resultXml), CompressionMode.Decompress);
-            ProcessingEngine resultEngine = (ProcessingEngine)PersistenceManager.Load(stream);
-
-            // merge the results
-            IScope oldScope = engineOperations[engineGuid].Scope;
-            oldScope.Clear();
-            foreach(IVariable variable in resultEngine.InitialOperation.Scope.Variables) {
-              oldScope.AddVariable(variable);
-            }
-            foreach(IScope subScope in resultEngine.InitialOperation.Scope.SubScopes) {
-              oldScope.AddSubScope(subScope);
-            }
-            foreach(KeyValuePair<string, string> alias in resultEngine.InitialOperation.Scope.Aliases) {
-              oldScope.AddAlias(alias.Key, alias.Value);
-            }
-
-            lock(locker) {
-              // signal the wait handle and clean up then return
-              waithandles[engineGuid].Set();
-              engineOperations.Remove(engineGuid);
-              waithandles.Remove(engineGuid);
-              runningEngines.Remove(engineGuid);
-            }
-            return;
-          } else {
-            // check if the server is still working on the job
-            JobState jobState = server.JobState(engineGuid);
-            if(jobState == JobState.Unkown) {
-              // restart job
-              byte[] packedEngine;
-              lock(locker) {
-                packedEngine = runningEngines[engineGuid];
+        Thread.Sleep(TimeSpan.FromSeconds(CHECK_RESULTS_TIMEOUT));
+        byte[] zippedResult = null;
+        lock(connectionLock) {
+          bool success = false;
+          int retries = 0;
+          do {
+            try {
+              zippedResult = server.TryEndExecuteEngine(engineGuid, 100);
+              success = true;
+            } catch(TimeoutException timeoutException) {
+              success = false;
+              retries++;
+              Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
+            } catch(CommunicationException communicationException) {
+              ResetConnection();
+              success = false;
+              retries++;
+              Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
+            }
+
+          } while(!success && retries < MAX_CONNECTION_RETRIES);
+        }
+        if(zippedResult != null) {
+          lock(dictionaryLock) {
+            // store result
+            results[engines[engineGuid].InitialOperation] = zippedResult;
+
+            // signal the wait handle and clean up then return
+            engines.Remove(engineGuid);
+            waithandles[engineGuid].Set();
+            waithandles.Remove(engineGuid);
+          }
+          return;
+        } else {
+          // check if the server is still working on the job
+          bool success = false;
+          int retries = 0;
+          JobState jobState = JobState.Unkown;
+          do {
+            try {
+              lock(connectionLock) {
+                jobState = server.JobState(engineGuid);
               }
-              server.BeginExecuteEngine(packedEngine);
-              restartCounter++;
-            }
+              success = true;
+            } catch(TimeoutException timeoutException) {
+              retries++;
+              success = false;
+              Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
+            } catch(CommunicationException communicationException) {
+              ResetConnection();
+              retries++;
+              success = false;
+              Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
+            }
+          } while(!success && retries < MAX_CONNECTION_RETRIES);
+          if(jobState == JobState.Unkown) {
+            // restart job
+            ProcessingEngine engine;
+            lock(dictionaryLock) {
+              engine = engines[engineGuid];
+            }
+            byte[] zippedEngine = ZipEngine(engine);
+            success = false;
+            retries = 0;
+            do {
+              try {
+                lock(connectionLock) {
+                  server.BeginExecuteEngine(zippedEngine);
+                }
+                success = true;
+              } catch(TimeoutException timeoutException) {
+                success = false;
+                retries++;
+                Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
+              } catch(CommunicationException communicationException) {
+                ResetConnection();
+                success = false;
+                retries++;
+                Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
+              }
+            } while(!success && retries < MAX_CONNECTION_RETRIES);
+            restartCounter++;
           }
-        } catch(Exception e) {
-          // catch all exceptions set an exception flag, signal the wait-handle and exit the routine
-          this.exception = new Exception("There was a problem with parallel execution", e);
-          waithandles[engineGuid].Set();
-          return;
         }
 
         // when we reach a maximum amount of restarts => signal the wait-handle and set a flag that there was a problem
         if(restartCounter > MAX_RESTARTS) {
-          this.exception = new Exception("Maximal number of parallel job restarts reached");
-          waithandles[engineGuid].Set();
-          return;
-        }
-
-        Thread.Sleep(TimeSpan.FromSeconds(10.0));
+          throw new ApplicationException("Maximum number of job restarts reached.");
+        }
       } while(true);
     }