Changeset 13943

Timestamp:

06/28/16 17:03:36 (8 years ago)

Author:

thasling

Message:

#2615:
implemented lock for timers, because if elapsed-method takes longer than the timer to elapse again, the method is called by another thread conccurently

Location:

branches/thasling/DistributedGA

Files:

• DistributedGA.ContactServer/ContactServiceImpl.cs (modified) (2 diffs)
• DistributedGA.ContactServer/DistributedGA.ContactServer.csproj (modified) (1 diff)
• DistributedGA.Core/Implementation/WcfMessageSender.cs (modified) (2 diffs)
• DistributedGA.Core/Implementation/WcfPeerListManager.cs (modified) (3 diffs)
• DistributedGA.Hive/P2PMigrationAnalyzer.cs (modified) (1 diff)

branches/thasling/DistributedGA/DistributedGA.ContactServer/ContactServiceImpl.cs

@@ -18 +18 @@
 
     private Object logLock = new Object();
+
+    private Object timerLock = new Object();//if elapsed takes too long, another thread also enters the method
 
     public ContactServiceImpl() {
@@ -78 +80 @@
 
     private void CleanUpContactTable(object sender, ElapsedEventArgs e) {
-      DateTime deadline = DateTime.Now;
-      //collect items to remove
-      List<PeerInfo> itemsToDelete = new List<PeerInfo>();
-      foreach (PeerInfo pi in allPeers.Keys) {
-        DateTime tmp;
-        if (allPeers.TryGetValue(pi, out tmp)) {
-          //if (tmp.AddHours(1f) < deadline)
-          if (tmp.AddMinutes(1) < deadline)  //TODO
+      lock (timerLock) {
+        DateTime deadline = DateTime.Now;
+        //collect items to remove
+        List<PeerInfo> itemsToDelete = new List<PeerInfo>();
+        foreach (PeerInfo pi in allPeers.Keys) {
+          DateTime tmp;
+          if (allPeers.TryGetValue(pi, out tmp)) {
+            //if (tmp.AddHours(1f) < deadline)
+            if (tmp.AddMinutes(1) < deadline)  //TODO
                     {
-            //if (tmp < deadline.AddHours(1f)) {
-            itemsToDelete.Add(pi);
+              //if (tmp < deadline.AddHours(1f)) {
+              itemsToDelete.Add(pi);
+            }
           }
         }
-      }
-      //remove items
-      foreach (PeerInfo pi in itemsToDelete) {
-        DateTime tmp;
-        allPeers.TryRemove(pi, out tmp);
-        Console.WriteLine(string.Format("Removed peer {0}:{1} from dictionary because last access was: {2}", pi.IpAddress, pi.Port, tmp));
+        //remove items
+        foreach (PeerInfo pi in itemsToDelete) {
+          DateTime tmp;
+          allPeers.TryRemove(pi, out tmp);
+          Console.WriteLine(string.Format("Removed peer {0}:{1} from dictionary because last access was: {2}", pi.IpAddress, pi.Port, tmp));
+        }
       }
     }

branches/thasling/DistributedGA/DistributedGA.ContactServer/DistributedGA.ContactServer.csproj

@@ -55 +55 @@
   </ItemGroup>
   <ItemGroup>
-    <None Include="App.config" />
+    <None Include="App.config">
+      <SubType>Designer</SubType>
+    </None>
   </ItemGroup>
   <ItemGroup>

branches/thasling/DistributedGA/DistributedGA.Core/Implementation/WcfMessageSender.cs

@@ -18 +18 @@
     private Timer timer; //sends cached messages to network in background
 
+    private Object timerLock = new Object();
+
     public void Init(PeerInfo source) {
       myself = source;
       bufferedMessages = new SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>>();
-      bufferedMessages.Limit = 1000;
+      bufferedMessages.Limit = 100000;
       timer = new Timer(1000 * 60); //each 5 minutes
       timer.Elapsed += GenerateSendingTasks;
@@ -52 +54 @@
     /// <param name="data">data to send</param>
     private void SendDataFromQueue(PeerInfo destination, byte[][] data) {
-      try {
-        Console.WriteLine(string.Format("Sending data to {0}:{1} in the background...", destination.IpAddress, destination.Port));
-        var serviceUrl = "DistributedGA.svc";
-        var baseUri = new Uri(string.Concat("net.tcp://", destination.IpAddress, ":", destination.Port, "/DistributedGA"));
-        var serviceUri = new Uri(baseUri, serviceUrl);
+      lock (timerLock) {
+        try {
+          Console.WriteLine(string.Format("Sending data to {0}:{1} in the background...", destination.IpAddress, destination.Port));
+          var serviceUrl = "DistributedGA.svc";
+          var baseUri = new Uri(string.Concat("net.tcp://", destination.IpAddress, ":", destination.Port, "/DistributedGA"));
+          var serviceUri = new Uri(baseUri, serviceUrl);
 
-        var binding = new NetTcpBinding();
-        var endpoint = new EndpointAddress(serviceUri);
-        using (var myChannelFactory = new ChannelFactory<IMessageContract>(binding, endpoint)) {
-          using (IClientChannel client = (IClientChannel)myChannelFactory.CreateChannel()) {
-            ((IMessageContract)client).SendData(myself, data); //maybe timout exception...
+          var binding = new NetTcpBinding();
+          var endpoint = new EndpointAddress(serviceUri);
+          using (var myChannelFactory = new ChannelFactory<IMessageContract>(binding, endpoint)) {
+            using (IClientChannel client = (IClientChannel)myChannelFactory.CreateChannel()) {
+              ((IMessageContract)client).SendData(myself, data); //maybe timout exception...
+            }
           }
         }
+        catch { } //if maybe sending failed (because of connection lost, etc.): just ignore
       }
-      catch { } //if maybe sending failed (because of connection lost, etc.): just ignore
     }
   }

branches/thasling/DistributedGA/DistributedGA.Core/Implementation/WcfPeerListManager.cs

@@ -15 +15 @@
 
     private Timer timer = null; //sends heartbeat to contact-server
+
+    private Object timerLock = new Object();
 
     private ChannelFactory<IContactService> myChannelFactory;
@@ -73 +75 @@
         res.Add(allPeers.ElementAt(index));
       }
-      return res;
+      return allPeers;
     }
 
@@ -90 +92 @@
 
     private void SendHeartbeatToServer(object sender, ElapsedEventArgs e) {
-      try {
-        heartbeatClient.UpdateHeartbeat(myself);
+      lock (timerLock) {
+        try {
+          heartbeatClient.UpdateHeartbeat(myself);
+        }
+        catch { } //nothing to do
       }
-      catch { } //nothing to do
     }
 

branches/thasling/DistributedGA/DistributedGA.Hive/P2PMigrationAnalyzer.cs

@@ -97 +97 @@
       // init P2P
       h = new PeerNetworkMessageHandler();
-      var peer = h.GetPeerInfo();
-      peer.ProblemInstance = ((StringValue)Parameters["JobGUID"].ActualValue).Value;
       var lanIpPrefix = ((StringValue)(Parameters["LanIpPrefix"].ActualValue)).Value;
       var contactServerUri = ((StringValue)(Parameters["ContactServerURL"].ActualValue)).Value;
       h.Init(lanIpPrefix, contactServerUri);
+      var peer = h.GetPeerInfo();
+      peer.ProblemInstance = ((StringValue)Parameters["JobGUID"].ActualValue).Value;
     }