Changeset 3931 for branches/3.2/sources/HeuristicLab.Hive.Server.Core/3.2/ClientCommunicator.cs

Timestamp:

06/19/10 09:17:24 (14 years ago)

Author:

kgrading

Message:

added minor speedups and better transaction handling to the server (#828)

File:

• branches/3.2/sources/HeuristicLab.Hive.Server.Core/3.2/ClientCommunicator.cs (modified) (5 diffs)

branches/3.2/sources/HeuristicLab.Hive.Server.Core/3.2/ClientCommunicator.cs

@@ -120 +120 @@
                 Logger.Info("Client timed out and is on RESET");
                 foreach (JobDto job in DaoLocator.JobDao.FindActiveJobsOfClient(client)) {
-                  jobManager.ResetJobsDependingOnResults(job);
+                  DaoLocator.JobDao.SetJobOffline(job);
                   lock (newAssignedJobs) {
                     if (newAssignedJobs.ContainsKey(job.Id))
@@ -149 +149 @@
               lastHeartbeats.Remove(client.Id);
             foreach (JobDto job in DaoLocator.JobDao.FindActiveJobsOfClient(client)) {
-              jobManager.ResetJobsDependingOnResults(job);
+              DaoLocator.JobDao.SetJobOffline(job);
             }
             heartbeatLock.ExitWriteLock();
@@ -428 +428 @@
       JobDto job2Calculate = scheduler.GetNextJobForClient(clientId);
       if (job2Calculate != null) {
-        response.Job = job2Calculate;
+        response.Job = job2Calculate;        
         response.Job.PluginsNeeded = DaoLocator.PluginInfoDao.GetPluginDependenciesForJob(response.Job);
         response.Success = true;
@@ -481 +481 @@
 
       if (response.Success) {
-        jobStream = DaoLocator.JobDao.GetSerializedJobStream(result.JobId);
-
+        Logger.Debug("Trying to aquire WCF Job Stream");
+        //jobStream = DaoLocator.JobDao.GetSerializedJobStream(result.JobId);
+        //Logger.Debug("Job Stream Aquired");
         byte[] buffer = new byte[3024];
+        List<byte> serializedJob = new List<byte>();
         int read = 0;
-        while ((read = stream.Read(buffer, 0, buffer.Length)) > 0) {
-          jobStream.Write(buffer, 0, read);
-        }
-        jobStream.Close();
+        int i = 0;        
+        while ((read = stream.Read(buffer, 0, buffer.Length)) > 0) {          
+          for (int j = 0; j < read; j++) {
+            serializedJob.Add(buffer[j]);
+          }
+          if (i% 100 == 0)
+            Logger.Debug("Writing to stream: " + i);         
+          //jobStream.Write(buffer, 0, read);
+          i++;
+        }
+        Logger.Debug("Done Writing, closing the stream!");               
+        //jobStream.Close();
+
+        DaoLocator.JobDao.SetBinaryJobFile(result.JobId, serializedJob.ToArray());
       }
       Logger.Info("END Job received for Storage:");
+      stream.Dispose();
       return response;
     }
@@ -656 +669 @@
         foreach (JobDto job in jobsOfClient) {
           if (job.State != State.finished)
-            jobManager.ResetJobsDependingOnResults(job);
+            DaoLocator.JobDao.SetJobOffline(job);
         }
       }