Changeset 14009 for branches/thasling/DistributedGA/DistributedGA.Core/Implementation

Timestamp:

07/06/16 16:47:47 (8 years ago)

Author:

thasling

Message:

#2615:
made minor changes,
wrong commit happende last time, so still tbd:
-sort List
-dispose of anaylzer is never called

Location:

branches/thasling/DistributedGA/DistributedGA.Core/Implementation

Files:

• PeerNetworkMessageHandler.cs (modified) (9 diffs)
• WcfMessageSender.cs (modified) (2 diffs)
• WcfPeerListManager.cs (modified) (2 diffs)

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

@@ -22 +22 @@
     //two queues are used to gather and and provide population more efficiently
     private object activeQueueLocker = new Object();
-    private SizedConcurrentQueue<KeyValuePair<PeerInfo,byte[]>> writeQueue;
+    private SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[]>> writeQueue;
     private SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[]>> readQueue;
 
@@ -29 +29 @@
 
     private double communicationRate;
-    private Random rand;
 
     public event EventHandler<Exception> ExceptionOccurend;
@@ -42 +41 @@
 
         this.communicationRate = communicationRate;
-        rand = new Random();
 
         writeQueue = new SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[]>>();
-        readQueue = new SizedConcurrentQueue<KeyValuePair<PeerInfo,byte[]>>();
+        readQueue = new SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[]>>();
         writeQueue.Limit = messageCacheCapacity;
         readQueue.Limit = writeQueue.Limit;
@@ -54 +52 @@
 
         peerListManager = new WcfPeerListManager();
-        peerListManager.Init(ownInstance, contactServerUrl);
+        peerListManager.Init(ownInstance, contactServerUrl, communicationRate);
 
         sender = new WcfMessageSender();
@@ -78 +76 @@
     public void PublishDataToNetwork(byte[] data) {
       try {
-        var peers = peerListManager.GetPeerList();
-        foreach (PeerInfo peer in peerListManager.GetPeerList()) {
-          var peersForMessaging = ChoosePeersForMessaging(ref peers);
-
-          //maybe something will go wrong during network communication...
+        var allPeers = peerListManager.GetPeerList();
+        foreach (PeerInfo peer in allPeers) {
           try {
             sender.SendData(peer, data);
@@ -90 +85 @@
           }
         }
+
       }
       catch (Exception ex) {
@@ -99 +95 @@
       try {
         List<KeyValuePair<PeerInfo, byte[]>> res = new List<KeyValuePair<PeerInfo, byte[]>>();
-        KeyValuePair<PeerInfo,byte[]> item;
+        KeyValuePair<PeerInfo, byte[]> item;
         lock (activeQueueLocker) {
           //changing the current queue for storing items to send
@@ -131 +127 @@
 
     private void PropagateException(Exception ex) {
-      //if (CountdownCompleted != null)
-      //  CountdownCompleted(this, e);
       if (ExceptionOccurend != null) {
         ExceptionOccurend(this, ex);
       }
-    }
-
-    private List<PeerInfo> ChoosePeersForMessaging(ref List<PeerInfo> allPeers) {
-      Shuffle<PeerInfo>(allPeers);
-      int toTake = Convert.ToInt32(allPeers.Count * communicationRate) + 1;
-      if (allPeers.Count > 0 && toTake == 0) {
-        toTake = 1;
-      }
-      return allPeers.Take(toTake).ToList(); ;
     }
 
@@ -161 +146 @@
     }
 
-    private void Shuffle<T>(IList<T> list) {
-      int n = list.Count;
-      while (n > 1) {
-        n--;
-        int k = rand.Next(n + 1);
-        T value = list[k];
-        list[k] = list[n];
-        list[n] = value;
-      }
-    }
-
     private void OnDataRecieved(object sender, MessageRecieveEventArgs e) {
       if (e != null && e.Sender.ProblemInstance.Equals(ownInstance.ProblemInstance)) {
         lock (activeQueueLocker) {
-            writeQueue.Enqueue(new KeyValuePair<PeerInfo, byte[]>(e.Sender, e.data));
+          writeQueue.Enqueue(new KeyValuePair<PeerInfo, byte[]>(e.Sender, e.data));
         }
 

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

@@ -17 +17 @@
     private SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[]>> bufferedMessagesWrite;
     private SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[]>> bufferedMessagesRead;
-
 
     private Timer timer; //sends cached messages to network in background
@@ -80 +79 @@
         }
         //for (int i = 0; i <= messagePackage.Value.GetUpperBound(0); i++) {
-          cache[messagePackage.Key].Add(messagePackage.Value);
+        cache[messagePackage.Key].Add(messagePackage.Value);
         //}
       }

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

@@ -22 +22 @@
     private IContactService client;
 
-    private IContactService heartbeatClient;
+    private List<PeerInfo> cachedPeerList;
 
+    private double communicationRate;
 
+    private Random rand;
 
-    public void Init(PeerInfo source, string contactServerUrl) {
+    public void Init(PeerInfo source, string contactServerUrl, double communicationRate) {
       serverString = contactServerUrl;
+      this.communicationRate = communicationRate;
       myself = source;
-
-      //Init ChannelFactory and Clients
+      cachedPeerList = new List<PeerInfo>();
+      rand = new Random();
+      //Init ChannelFactory and Client
       var binding = new NetTcpBinding();
       var endpoint = new EndpointAddress(serverString);
       myChannelFactory = new ChannelFactory<IContactService>(binding, endpoint);
       client = myChannelFactory.CreateChannel();
-      heartbeatClient = myChannelFactory.CreateChannel();
       //Register Peer
-      client.RegisterPeer(source);
+      client.RegisterPeer(myself);
       //Start heartbeat timer
       timer = new Timer(1000 * 20); //each 20 seconds
-      timer.Elapsed += SendHeartbeatToServer;
+      timer.Elapsed += RefreshPeerList;
       timer.Start();
     }
 
     public List<PeerInfo> GetPeerList() {
-      try {
-        var allPeers = client.GetPeerList(myself); //maybe timout exception...
-        return allPeers;
-      }
-      catch { } //if maybe sending failed (because of connection lost, etc.): just ignore
-      return new List<PeerInfo>();
+      return cachedPeerList;
     }
 
@@ -61 +59 @@
       timer.Dispose();
       ((IClientChannel)client).Close();
-      ((IClientChannel)heartbeatClient).Close();
       myChannelFactory.Close();
-      client = null;
       myChannelFactory = null;
     }
 
-    private void SendHeartbeatToServer(object sender, ElapsedEventArgs e) {
+    private List<PeerInfo> ChoosePeersForMessaging(ref List<PeerInfo> allPeers) {
+      Shuffle<PeerInfo>(allPeers);
+      int toTake = Convert.ToInt32(allPeers.Count * communicationRate) + 1;
+      if (allPeers.Count > 0 && toTake == 0) {
+        toTake = 1;
+      }
+      return allPeers.Take(toTake).ToList(); ;
+    }
+
+    private void Shuffle<T>(IList<T> list) {
+      int n = list.Count;
+      while (n > 1) {
+        n--;
+        int k = rand.Next(n + 1);
+        T value = list[k];
+        list[k] = list[n];
+        list[n] = value;
+      }
+    }
+
+    private void RefreshPeerList(object sender, ElapsedEventArgs e) {
       lock (timerLock) {
         try {
-          heartbeatClient.UpdateHeartbeat(myself);
+          var allPeers = client.GetPeerList(myself);
+          cachedPeerList = ChoosePeersForMessaging(ref allPeers);
         }
         catch { } //nothing to do