Changeset 1154 for trunk/sources/HeuristicLab.Hive.Server.Core/ClientCommunicator.cs

Timestamp:

01/18/09 12:58:11 (15 years ago)

Author:

svonolfe

Message:

Improved locking to allow more concurrency in the ClientCommunicator (#399)

File:

• trunk/sources/HeuristicLab.Hive.Server.Core/ClientCommunicator.cs (modified) (16 diffs)

trunk/sources/HeuristicLab.Hive.Server.Core/ClientCommunicator.cs

@@ -32 +32 @@
 using System.Reflection;
 using HeuristicLab.Hive.JobBase;
-using System.Runtime.CompilerServices;
 using HeuristicLab.Hive.Server.Core.InternalInterfaces;
+using System.Threading;
 
 namespace HeuristicLab.Hive.Server.Core {
@@ -40 +40 @@
   /// </summary>
   public class ClientCommunicator: IClientCommunicator {
-    Dictionary<Guid, DateTime> lastHeartbeats = 
+    private static Dictionary<Guid, DateTime> lastHeartbeats = 
       new Dictionary<Guid,DateTime>();
+
+    private static ReaderWriterLockSlim heartbeatLock =
+      new ReaderWriterLockSlim();
+
+    private static Mutex jobLock =
+      new Mutex();
 
     IClientAdapter clientAdapter;
@@ -64 +70 @@
       lifecycleManager.RegisterHeartbeat( 
         new EventHandler(lifecycleManager_OnServerHeartbeat));
-
-      lastHeartbeats = new Dictionary<Guid, DateTime>();
     }
 
@@ -74 +78 @@
     /// <param name="sender"></param>
     /// <param name="e"></param>
-    [MethodImpl(MethodImplOptions.Synchronized)]
     void lifecycleManager_OnServerHeartbeat(object sender, EventArgs e) {
       List<ClientInfo> allClients = new List<ClientInfo>(clientAdapter.GetAll());
@@ -81 +84 @@
       foreach (ClientInfo client in allClients) {
         if (client.State != State.offline && client.State != State.nullState) {
+          heartbeatLock.EnterUpgradeableReadLock();
+
           if (!lastHeartbeats.ContainsKey(client.ClientId)) {
             client.State = State.offline;
@@ -86 +91 @@
           } else {
             DateTime lastHbOfClient = lastHeartbeats[client.ClientId];
+
             TimeSpan dif = DateTime.Now.Subtract(lastHbOfClient);
             // check if time between last hearbeat and now is greather than HEARTBEAT_MAX_DIF
@@ -102 +108 @@
               client.State = State.offline;
               clientAdapter.Update(client);
+
+              heartbeatLock.EnterWriteLock();
               lastHeartbeats.Remove(client.ClientId);
+              heartbeatLock.ExitWriteLock();
             }
           }
+
+          heartbeatLock.ExitUpgradeableReadLock();
         } else {
+          heartbeatLock.EnterWriteLock();
           if (lastHeartbeats.ContainsKey(client.ClientId))
             lastHeartbeats.Remove(client.ClientId);
+          heartbeatLock.ExitWriteLock();
         }
       }
@@ -120 +133 @@
     /// <param name="clientInfo"></param>
     /// <returns></returns>
-    [MethodImpl(MethodImplOptions.Synchronized)]
     public Response Login(ClientInfo clientInfo) {
       Response response = new Response();
 
+      heartbeatLock.EnterWriteLock();
       if (lastHeartbeats.ContainsKey(clientInfo.ClientId)) {
         lastHeartbeats[clientInfo.ClientId] = DateTime.Now;
@@ -129 +142 @@
         lastHeartbeats.Add(clientInfo.ClientId, DateTime.Now);
       }
+      heartbeatLock.ExitWriteLock();
 
       ICollection<ClientInfo> allClients = clientAdapter.GetAll();
@@ -152 +166 @@
     /// <param name="hbData"></param>
     /// <returns></returns>
-    [MethodImpl(MethodImplOptions.Synchronized)]
     public ResponseHB SendHeartBeat(HeartBeatData hbData) {
       ResponseHB response = new ResponseHB();
@@ -165 +178 @@
       }
 
+      heartbeatLock.EnterWriteLock();
       if (lastHeartbeats.ContainsKey(hbData.ClientId)) {
         lastHeartbeats[hbData.ClientId] = DateTime.Now;
@@ -170 +184 @@
         lastHeartbeats.Add(hbData.ClientId, DateTime.Now);
       }
+      heartbeatLock.ExitWriteLock();
 
       response.Success = true;
@@ -196 +211 @@
     /// <param name="clientId"></param>
     /// <returns></returns>
-    [MethodImpl(MethodImplOptions.Synchronized)]
     public ResponseJob PullJob(Guid clientId) {
       ResponseJob response = new ResponseJob();
-      lock (this) {
-        LinkedList<Job> allOfflineJobs = new LinkedList<Job>(jobAdapter.GetJobsByState(State.offline));
-        if (allOfflineJobs != null && allOfflineJobs.Count > 0) {
-          Job job2Calculate = allOfflineJobs.First.Value;
-          job2Calculate.State = State.calculating;
-          job2Calculate.Client = clientAdapter.GetById(clientId);
-          response.Job = job2Calculate;
-          jobAdapter.Update(job2Calculate);
-          response.Success = true;
-          response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_PULLED;
-          return response;
-        } 
-      }
+      
+      /// Critical section ///
+      jobLock.WaitOne();
+
+      LinkedList<Job> allOfflineJobs = new LinkedList<Job>(jobAdapter.GetJobsByState(State.offline));
+      if (allOfflineJobs != null && allOfflineJobs.Count > 0) {
+        Job job2Calculate = allOfflineJobs.First.Value;
+        job2Calculate.State = State.calculating;
+        job2Calculate.Client = clientAdapter.GetById(clientId);
+        job2Calculate.Client.State = State.calculating;
+
+        response.Job = job2Calculate;
+        jobAdapter.Update(job2Calculate);
+        response.Success = true;
+        response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_JOB_PULLED;
+        return response;
+      }
+
+      jobLock.ReleaseMutex();
+
       response.Success = true;
       response.StatusMessage = ApplicationConstants.RESPONSE_COMMUNICATOR_NO_JOBS_LEFT;
@@ -228 +249 @@
     /// <param name="finished"></param>
     /// <returns></returns>
-    [MethodImpl(MethodImplOptions.Synchronized)]
     public ResponseResultReceived SendJobResult(Guid clientId, 
       long jobId, 
@@ -269 +289 @@
         jobAdapter.Update(job);
 
+        client.State = State.idle;
+        clientAdapter.Update(client);
+
         List<JobResult> jobResults = new List<JobResult>(jobResultAdapter.GetResultsOf(job));
         foreach (JobResult currentResult in jobResults) 
@@ -298 +321 @@
     /// </summary>
     /// <param name="clientId"></param>
-    /// <returns></returns>
-    [MethodImpl(MethodImplOptions.Synchronized)]                       
+    /// <returns></returns>                       
     public Response Logout(Guid clientId) {
       Response response = new Response();
 
+      heartbeatLock.EnterWriteLock();
       if (lastHeartbeats.ContainsKey(clientId))
         lastHeartbeats.Remove(clientId);
+      heartbeatLock.ExitWriteLock();
 
       ClientInfo client = clientAdapter.GetById(clientId);