Changeset 13959

Timestamp:

06/29/16 22:23:58 (8 years ago)

Author:

thasling

Message:

#2615:
it works

Location:

branches/thasling/DistributedGA

Files:

• DistributedGA.Core/Implementation/WcfMessageSender.cs (modified) (5 diffs)
• DistributedGA.Core/Implementation/WcfPeerListManager.cs (modified) (2 diffs)
• DistributedGA.Hive/P2PMigrationAnalyzer.cs (modified) (1 diff)

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

@@ -14 +14 @@
     private PeerInfo myself;
 
-    private SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>> bufferedMessages;
+    //providing two queues for faster access
+    private SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>> bufferedMessagesWrite;
+    private SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>> bufferedMessagesRead;
+
 
     private Timer timer; //sends cached messages to network in background
@@ -22 +25 @@
     public void Init(PeerInfo source, int messageCacheCapacity) {
       myself = source;
-      bufferedMessages = new SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>>();
-      bufferedMessages.Limit = messageCacheCapacity;
+      bufferedMessagesRead = new SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>>();
+      bufferedMessagesRead.Limit = messageCacheCapacity;
+      bufferedMessagesWrite = new SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>>();
+      bufferedMessagesWrite.Limit = bufferedMessagesRead.Limit;
       timer = new Timer(1000 * 60); //each 5 minutes
       timer.Elapsed += GenerateSendingTasks;
@@ -30 +35 @@
 
     public void SendData(PeerInfo destination, byte[][] data) {
-      bufferedMessages.Enqueue(new KeyValuePair<PeerInfo, byte[][]>(destination, data));
+      bufferedMessagesWrite.Enqueue(new KeyValuePair<PeerInfo, byte[][]>(destination, data));
     }
 
@@ -40 +45 @@
 
     private void GenerateSendingTasks(object sender, ElapsedEventArgs e) {
-      while (!bufferedMessages.IsEmpty) {
-        KeyValuePair<PeerInfo, byte[][]> message;
-        if (bufferedMessages.TryDequeue(out message)) {
-          Task.Factory.StartNew(() => SendDataFromQueue(message.Key, message.Value), TaskCreationOptions.LongRunning);
+      lock (timerLock) {
+        //changing the queues...
+        var tmp = bufferedMessagesRead;
+        bufferedMessagesRead = bufferedMessagesWrite;
+        bufferedMessagesWrite = tmp;
+        List<KeyValuePair<PeerInfo, byte[][]>> messages = new List<KeyValuePair<PeerInfo, byte[][]>>();
+        while (!bufferedMessagesRead.IsEmpty) {
+          KeyValuePair<PeerInfo, byte[][]> message;
+          if (bufferedMessagesRead.TryDequeue(out message)) {
+            messages.Add(message);
+            //Task.Factory.StartNew(() => SendDataFromQueue(message.Key, message.Value), TaskCreationOptions.LongRunning);
+          }
+        }
+        //now: merge them and start sending tasks
+
+        List<KeyValuePair<PeerInfo, byte[][]>> mergedMessages = MergeMessages(messages);
+        foreach (var item in mergedMessages) {
+          Task.Factory.StartNew(() => SendDataFromQueue(item.Key, item.Value), TaskCreationOptions.LongRunning);
         }
       }
+    }
+
+    /// <summary>
+    /// Merges and resorts all Messages by their destination
+    /// </summary>
+    /// <param name="messages">a list with multiple messages to the same  destination</param>
+    /// <returns></returns>
+    private List<KeyValuePair<PeerInfo, byte[][]>> MergeMessages(List<KeyValuePair<PeerInfo, byte[][]>> messages) {
+      List<KeyValuePair<PeerInfo, byte[][]>> res = new List<KeyValuePair<PeerInfo, byte[][]>>();
+      Dictionary<PeerInfo, List<byte[]>> cache = new Dictionary<PeerInfo, List<byte[]>>();
+      foreach (var messagePackage in messages) {
+        if (!cache.ContainsKey(messagePackage.Key)) {
+          cache.Add(messagePackage.Key, new List<byte[]>());
+        }
+        for (int i = 0; i <= messagePackage.Value.GetUpperBound(0); i++) {
+          cache[messagePackage.Key].Add(messagePackage.Value[i]);
+        }
+      }
+      //now we have a dictionary with all messages per destionation
+      //so: create a byte[][] again and return
+      foreach (var dest in cache.Keys) {
+        byte[][] messagesPerDest = new byte[cache[dest].Count][];
+        for (int i = 0; i <= messagesPerDest.GetUpperBound(0); i++) {
+          messagesPerDest[i] = cache[dest][i];
+        }
+        res.Add(new KeyValuePair<PeerInfo, byte[][]>(dest, messagesPerDest));
+      }
+      return res;
     }
 
@@ -54 +101 @@
     /// <param name="data">data to send</param>
     private void SendDataFromQueue(PeerInfo destination, byte[][] data) {
-      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);
+      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;
+        }
 
-          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...
-            }
+
+        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();
+        binding.ReaderQuotas.MaxArrayLength = 2147483647;
+        binding.ReaderQuotas.MaxNameTableCharCount = 2147483647;
+        binding.ReaderQuotas.MaxBytesPerRead = 2147483647;
+        binding.ReaderQuotas.MaxStringContentLength = 2147483647;
+
+        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 (Exception ex) {
+        Console.WriteLine(ex.Message);
       }
     }

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

@@ -72 +72 @@
     private List<PeerInfo> ChoosePeersForMessaging(List<PeerInfo> allPeers) {
       //communicate with 10% of the network
-      int noOfPeers = allPeers.Count / (100 /communicationRate);
+      int noOfPeers = allPeers.Count / (100 / communicationRate);
       List<int> indexList = GetRandomItemIndexes(noOfPeers, 0, allPeers.Count - 1);
       List<PeerInfo> res = new List<PeerInfo>();
@@ -86 +86 @@
       int tmp = -1;
       while (res.Count < noOfItems) {
-        tmp = rnd.Next(minValue, maxValue);
+        tmp = rnd.Next(minValue, maxValue + 1);
         if (!res.Contains(tmp)) {
           res.Add(tmp);

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

@@ -53 +53 @@
       get { return (ILookupParameter<IntValue>)Parameters["CommunicationRate"]; }
     }
-    public ILookupParameter<IntValue> MessageCacheCapacityParameter {                         
+    public ILookupParameter<IntValue> MessageCacheCapacityParameter {
       get { return (ILookupParameter<IntValue>)Parameters["MessageCacheCapacity"]; }
     }