Changeset 14009 for branches/thasling/DistributedGA

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

Files:

• DistributedGA.Core/Implementation/PeerNetworkMessageHandler.cs (modified) (9 diffs)
• DistributedGA.Core/Implementation/WcfMessageSender.cs (modified) (2 diffs)
• DistributedGA.Core/Implementation/WcfPeerListManager.cs (modified) (2 diffs)
• DistributedGA.Core/Interface/IPeerListManager.cs (modified) (1 diff)
• DistributedGA.Hive/MigrationStrategy.cs (modified) (1 diff)
• DistributedGA.Hive/P2PMigrationAnalyzer.cs (modified) (8 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

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

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

branches/thasling/DistributedGA/DistributedGA.Hive/MigrationStrategy.cs

@@ -6 +6 @@
 
 namespace DistributedGA.Hive {
-  public enum MigrationStrategy  { TakeBestReplaceBad, TakeRandomReplaceBad, TakeBestReplaceRandom, TakeRandomReplaceRandom };
+  public enum MigrationStrategy  { Best, Worst, Random };
 }

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

@@ -72 +72 @@
     }
 
-    public IConstrainedValueParameter<EnumValue<MigrationStrategy>> MigrationStrategyParameter {
-      get { return (IConstrainedValueParameter<EnumValue<MigrationStrategy>>)Parameters["MigrationStrategy"]; }
+    public IConstrainedValueParameter<EnumValue<MigrationStrategy>> MigrationStrategySelectParameter {
+      get { return (IConstrainedValueParameter<EnumValue<MigrationStrategy>>)Parameters["MigrationStrategySelect"]; }
+    }
+
+    public IConstrainedValueParameter<EnumValue<MigrationStrategy>> MigrationStrategyReplaceParameter {
+      get { return (IConstrainedValueParameter<EnumValue<MigrationStrategy>>)Parameters["MigrationStrategyReplace"]; }
     }
 
@@ -100 +104 @@
       Parameters.Add(new LookupParameter<BoolValue>("Maximization"));
       Parameters.Add(new ScopeTreeLookupParameter<DoubleValue>("Quality", 1));
-      Parameters.Add(new ValueParameter<IntValue>("MigrationInterval", "", new IntValue(1)));
+      Parameters.Add(new ValueParameter<IntValue>("MigrationInterval", "", new IntValue(5)));
       Parameters.Add(new ValueParameter<PercentValue>("MigrationRate", "", new PercentValue(0.05)));
       Parameters.Add(new ValueParameter<PercentValue>("CommunicationRate", "", new PercentValue(0.10)));
-      Parameters.Add(new ValueParameter<IntValue>("MessageCacheCapacity", "", new IntValue(100)));
+      Parameters.Add(new ValueParameter<IntValue>("MessageCacheCapacity", "", new IntValue(1000)));
       Parameters.Add(new ValueParameter<StringValue>("LanIpPrefix", "", new StringValue("10.")));
       Parameters.Add(new LookupParameter<IRandom>("Random", "The random number generator"));
@@ -111 +115 @@
 
       var validValues = new ItemSet<EnumValue<MigrationStrategy>>();
-      validValues.Add(new EnumValue<MigrationStrategy>(MigrationStrategy.TakeBestReplaceBad));
-      validValues.Add(new EnumValue<MigrationStrategy>(MigrationStrategy.TakeBestReplaceRandom));
-      validValues.Add(new EnumValue<MigrationStrategy>(MigrationStrategy.TakeRandomReplaceBad));
-      validValues.Add(new EnumValue<MigrationStrategy>(MigrationStrategy.TakeRandomReplaceRandom));
-
-      Parameters.Add(new ConstrainedValueParameter<EnumValue<MigrationStrategy>>("MigrationStrategy", validValues, (new EnumValue<MigrationStrategy>(MigrationStrategy.TakeBestReplaceBad))));
+      validValues.Add(new EnumValue<MigrationStrategy>(MigrationStrategy.Best));
+      validValues.Add(new EnumValue<MigrationStrategy>(MigrationStrategy.Worst));
+      validValues.Add(new EnumValue<MigrationStrategy>(MigrationStrategy.Random));
+      Parameters.Add(new ConstrainedValueParameter<EnumValue<MigrationStrategy>>("MigrationStrategySelect", validValues, (new EnumValue<MigrationStrategy>(MigrationStrategy.Random))));
+      Parameters.Add(new ConstrainedValueParameter<EnumValue<MigrationStrategy>>("MigrationStrategyReplace", validValues, (new EnumValue<MigrationStrategy>(MigrationStrategy.Random))));
+
     }
 
     public override IDeepCloneable Clone(Cloner cloner) {
       return new P2PMigrationAnalyzer(this, cloner);
-    }
-
-    public override void ClearState() {
-      base.ClearState();
-      h.Dispose();
-      h = null;
     }
 
@@ -143 +141 @@
       if (h == null) {
         Init();
+        //TODO: sort population
+
       }
       if (MigrationIterationsParameter.ActualValue == null) {
@@ -160 +160 @@
         }
         var popQualities = QualityParameter.ActualValue;
-        var selectedMigStrat = MigrationStrategyParameter.Value.Value;
+        var selectedMigStratSelect = MigrationStrategySelectParameter.Value.Value;
+        var selectedMigStratReplace = MigrationStrategyReplaceParameter.Value.Value;
 
         var rand = Random;
@@ -168 +169 @@
         for (int i = 0; i < noOfEmigrants; i++) {
           //select emigrant depending on strategy
-          switch (selectedMigStrat) {
-            case MigrationStrategy.TakeBestReplaceBad:
+          switch (selectedMigStratSelect) {
+            case MigrationStrategy.Best:
               emigrants = scope.SubScopes[i];
               emigrantsList.Add(emigrants);
               break;
 
-            case MigrationStrategy.TakeBestReplaceRandom:
-              emigrants = scope.SubScopes[i];
-              emigrantsList.Add(emigrants);
-              break;
-
-            case MigrationStrategy.TakeRandomReplaceBad:
+            case MigrationStrategy.Random:
               replIdx = rand.Next(scope.SubScopes.Count);
               emigrants = scope.SubScopes[replIdx];
@@ -185 +181 @@
               break;
 
-            case MigrationStrategy.TakeRandomReplaceRandom:
-              replIdx = rand.Next(scope.SubScopes.Count);
-              emigrants = scope.SubScopes[replIdx];
+            case MigrationStrategy.Worst:
+              emigrants = scope.SubScopes[scope.SubScopes.Count - i];
               emigrantsList.Add(emigrants);
               break;
+
 
             default:
@@ -226 +222 @@
 
               // replace individual in current population
-              switch (selectedMigStrat) {
-                case MigrationStrategy.TakeBestReplaceBad:
+              switch (selectedMigStratReplace) {
+                case MigrationStrategy.Best:
                   scope.SubScopes.RemoveAt(0);
                   break;
 
-                case MigrationStrategy.TakeRandomReplaceBad:
-                  scope.SubScopes.RemoveAt(0);
-                  break;
-
-                case MigrationStrategy.TakeBestReplaceRandom:
-                  //replace random
+                case MigrationStrategy.Random:
                   replIdx = rand.Next(scope.SubScopes.Count);
                   scope.SubScopes.RemoveAt(replIdx);
                   break;
 
-                case MigrationStrategy.TakeRandomReplaceRandom:
+                case MigrationStrategy.Worst:
                   //replace random
-                  replIdx = rand.Next(scope.SubScopes.Count);
-                  scope.SubScopes.RemoveAt(replIdx);
+                  scope.SubScopes.RemoveAt(scope.SubScopes.Count);
                   break;