Changeset 2058 for trunk/sources/HeuristicLab.Grid/3.2

Timestamp:

06/18/09 15:01:10 (15 years ago)

Author:

gkronber

Message:

Refactored JobManager and added a plugin that contains a bridge between grid and hive. The bridge allows to use the execution engine service of Hive as a grid server. This way CEDMA job execution and DistributedEngine job execution can either use Hive or Grid as backend. #642 (Hive backend for CEDMA)

Location:

trunk/sources/HeuristicLab.Grid/3.2

Files:

• GridServerProxy.cs (added)
• HeuristicLab.Grid-3.2.csproj (modified) (2 diffs)
• HeuristicLabGridPlugin.cs (modified) (1 diff)
• JobManager.cs (modified) (8 diffs)

trunk/sources/HeuristicLab.Grid/3.2/HeuristicLab.Grid-3.2.csproj

@@ -99 +99 @@
     </Compile>
     <Compile Include="AsyncGridResult.cs" />
+    <Compile Include="GridServerProxy.cs" />
     <Compile Include="Database.cs" />
     <Compile Include="EngineRunner.cs" />
@@ -128 +129 @@
       <Project>{94186A6A-5176-4402-AE83-886557B53CCA}</Project>
       <Name>HeuristicLab.PluginInfrastructure</Name>
+    </ProjectReference>
+    <ProjectReference Include="..\..\HeuristicLab.Tracing\3.2\HeuristicLab.Tracing-3.2.csproj">
+      <Project>{EE2034D9-6E27-48A1-B855-42D45F69A4FC}</Project>
+      <Name>HeuristicLab.Tracing-3.2</Name>
     </ProjectReference>
   </ItemGroup>

trunk/sources/HeuristicLab.Grid/3.2/HeuristicLabGridPlugin.cs

@@ -30 +30 @@
   [PluginFile(Filename = "HeuristicLab.Grid-3.2.dll", Filetype = PluginFileType.Assembly)]
   [Dependency(Dependency = "HeuristicLab.Core-3.2")]
+  [Dependency(Dependency = "HeuristicLab.Tracing-3.2")]
   public class HeuristicLabGridPlugin : PluginBase {
   }

trunk/sources/HeuristicLab.Grid/3.2/JobManager.cs

@@ -39 +39 @@
   public class JobManager {
     private const int MAX_RESTARTS = 5;
-    private const int MAX_CONNECTION_RETRIES = 10;
-    private const int RETRY_TIMEOUT_SEC = 60;
     private const int RESULT_POLLING_TIMEOUT = 5;
 
@@ -49 +47 @@
     private object runningQueueLock = new object();
     private Queue<AsyncGridResult> runningJobs = new Queue<AsyncGridResult>();
-    private object connectionLock = new object();
-
     private AutoResetEvent runningWaitHandle = new AutoResetEvent(false);
     private AutoResetEvent waitingWaitHandle = new AutoResetEvent(false);
 
-    private ChannelFactory<IGridServer> factory;
 
-    public JobManager(string address) {
-      this.address = address;
+    public JobManager(IGridServer server) {
+      this.server = server;
       Thread starterThread = new Thread(StartEngines);
       Thread resultsGatheringThread = new Thread(GetResults);
@@ -65 +60 @@
 
     public void Reset() {
-      ResetConnection();
       lock (waitingQueueLock) {
         foreach (AsyncGridResult r in waitingJobs) {
@@ -80 +74 @@
     }
 
-    private void ResetConnection() {
-      Trace.TraceInformation("Reset connection in JobManager");
-      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;
-
-        factory = new ChannelFactory<IGridServer>(binding);
-        server = factory.CreateChannel(new EndpointAddress(address));
-      }
-    }
 
     public void StartEngines() {
@@ -103 +84 @@
           if (job == null) waitingWaitHandle.WaitOne(); // no jobs waiting
           else {
-            Guid currentEngineGuid = TryStartExecuteEngine(job.Engine);
+            Guid currentEngineGuid = server.BeginExecuteEngine(ZipEngine(job.Engine));
             if (currentEngineGuid == Guid.Empty) {
               // couldn't start the job -> requeue
@@ -141 +122 @@
           if (job == null) runningWaitHandle.WaitOne(); // no jobs running
           else {
-            byte[] zippedResult = TryEndExecuteEngine(server, job.Guid);
+            byte[] zippedResult = server.TryEndExecuteEngine(job.Guid);
             if (zippedResult != null) { 
               // successful => store result
@@ -149 +130 @@
             } else {
               // there was a problem -> check the state of the job and restart if necessary
-              JobState jobState = TryGetJobState(server, job.Guid);
+              JobState jobState = server.JobState(job.Guid);
               if (jobState == JobState.Unknown) {
                 job.Restarts++;
@@ -195 +176 @@
       }
     }
-
-    private Guid TryStartExecuteEngine(IEngine engine) {
-      byte[] zippedEngine = ZipEngine(engine);
-      return SavelyExecute(() => server.BeginExecuteEngine(zippedEngine));
-    }
-
-    private byte[] TryEndExecuteEngine(IGridServer server, Guid engineGuid) {
-      return SavelyExecute(() => {
-        byte[] zippedResult = server.TryEndExecuteEngine(engineGuid);
-        return zippedResult;
-      });
-    }
-
-    private JobState TryGetJobState(IGridServer server, Guid engineGuid) {
-      return SavelyExecute(() => server.JobState(engineGuid));
-    }
-
-    private TResult SavelyExecute<TResult>(Func<TResult> a) {
-      int retries = 0;
-      do {
-        try {
-          lock (connectionLock) {
-            return a();
-          }
-        }
-        catch (TimeoutException) {
-          retries++;
-          Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
-        }
-        catch (CommunicationException) {
-          ResetConnection();
-          retries++;
-          Thread.Sleep(TimeSpan.FromSeconds(RETRY_TIMEOUT_SEC));
-        }
-      } while (retries < MAX_CONNECTION_RETRIES);
-      Trace.TraceWarning("Reached max connection retries");
-      return default(TResult);
-    }
   }
 }