Changeset 13982

Timestamp:

07/02/16 15:23:57 (8 years ago)

Author:

thasling

Message:

#2615:
made minor changes

Location:

branches/thasling/DistributedGA

Files:

• DistributedGA.ContactServer/ContactServiceImpl.cs (modified) (2 diffs)
• DistributedGA.Core/Implementation/PeerNetworkMessageHandler.cs (modified) (6 diffs)
• DistributedGA.Core/Implementation/WcfMessageSender.cs (modified) (2 diffs)
• DistributedGA.Core/Implementation/WcfPeerListManager.cs (modified) (4 diffs)
• DistributedGA.Core/Interface/IPeerListManager.cs (modified) (1 diff)
• DistributedGA.Hive/P2PMigrationAnalyzer.cs (modified) (1 diff)

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

@@ -24 +24 @@
       allPeers = new ConcurrentDictionary<PeerInfo, DateTime>();
 
-      timer = new Timer(1000); //each hour
+      timer = new Timer(1000); //each minute
       timer.Elapsed += CleanUpContactTable;
       timer.Start();
@@ -87 +87 @@
           DateTime tmp;
           if (allPeers.TryGetValue(pi, out tmp)) {
-            //if (tmp.AddHours(1f) < deadline)
-            if (tmp.AddMinutes(1) < deadline)  //TODO
-                    {
-              //if (tmp < deadline.AddHours(1f)) {
+            if (tmp.AddMinutes(1) < deadline) {
               itemsToDelete.Add(pi);
             }

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

@@ -28 +28 @@
     private IMessageService host = null;
 
+    private double communicationRate;
+    private Random rand;
+
     public event EventHandler<Exception> ExceptionOccurend;
 
@@ -37 +40 @@
           ProblemInstance = problemInstance
         }; // TODO: get own peerinfo
+
+        this.communicationRate = communicationRate;
+        rand = new Random();
 
         writeQueue = new SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[]>>();
@@ -48 +54 @@
 
         peerListManager = new WcfPeerListManager();
-        peerListManager.Init(ownInstance, contactServerUrl, communicationRate);
+        peerListManager.Init(ownInstance, contactServerUrl);
 
         sender = new WcfMessageSender();
@@ -72 +78 @@
     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...
           try {
@@ -129 +138 @@
     }
 
+    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 string GetInternalIpAddress(string ipPrefix) {
       try {
@@ -141 +159 @@
       }
       catch { return null; }
+    }
+
+    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;
+      }
     }
 

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

@@ -102 +102 @@
     private void SendDataFromQueue(PeerInfo destination, byte[][] data) {
       try {
-        //int arrayLength = 3;
-        //if (data.GetUpperBound(0) > arrayLength) {
-        //  //HACK: SEND MAX 10 items
-        //  byte[][] fake = new byte[arrayLength][];
-        //  for (int i = 0; i < arrayLength; i++) {
-        //    fake[i] = data[i];
-        //  }
-        //  data = fake;
-        //}
-
-
         Console.WriteLine(string.Format("Sending data to {0}:{1} in the background...", destination.IpAddress, destination.Port));
         var serviceUrl = "DistributedGA.svc";
@@ -130 +119 @@
           using (IClientChannel client = (IClientChannel)myChannelFactory.CreateChannel()) {
             for (int i = 0; i < data.GetUpperBound(0) + 1; i++) {
-              ((IMessageContract)client).SendData(myself, data[i]); //maybe timout exception...
+              ((IMessageContract)client).SendData(myself, data[i]); //maybe exception...
             }
           }

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

@@ -24 +24 @@
     private IContactService heartbeatClient;
 
-    private Random rand;
 
-    private double communicationRate; //how many peers are contacted by this peer in percent
 
-    public void Init(PeerInfo source, string contactServerUrl, double communicationRate) {
+    public void Init(PeerInfo source, string contactServerUrl) {
       serverString = contactServerUrl;
-      this.communicationRate = communicationRate;
       myself = source;
-      rand = new Random();
-     
+
       //Init ChannelFactory and Clients
       var binding = new NetTcpBinding();
@@ -43 +39 @@
       client.RegisterPeer(source);
       //Start heartbeat timer
-      timer = new Timer(1000); //each 5 minutes
+      timer = new Timer(1000 * 20); //each 20 seconds
       timer.Elapsed += SendHeartbeatToServer;
       timer.Start();
@@ -51 +47 @@
       try {
         var allPeers = client.GetPeerList(myself); //maybe timout exception...
-        var peersForMessaging = ChoosePeersForMessaging(ref allPeers);
-        return peersForMessaging;
+        return allPeers;
       }
       catch { } //if maybe sending failed (because of connection lost, etc.): just ignore
@@ -72 +67 @@
     }
 
-
-    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 SendHeartbeatToServer(object sender, ElapsedEventArgs e) {
       lock (timerLock) {

branches/thasling/DistributedGA/DistributedGA.Core/Interface/IPeerListManager.cs

@@ -6 +6 @@
   public interface IPeerListManager {
 
-    void Init(PeerInfo source, string contactServerUrl, double communicationRate); //Registers own instance at the contact-server
+    void Init(PeerInfo source, string contactServerUrl); //Registers own instance at the contact-server
 
     List<PeerInfo> GetPeerList(); //Recieves all peers in the network from contact-server

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

@@ -136 +136 @@
       base.InitializeState();
       // init P2P
-      Init();
+      if (h == null) {
+        //otherwhise old object is not disposed correctly
+        Init();
+      }
     }