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

Timestamp:

06/19/16 21:37:24 (8 years ago)

Author:

thasling

Message:

#2615:
re-implemented WcfPeerListManager
also sending messages is now done in the background async

Location:

branches/thasling/DistributedGA/DistributedGA.Core/Implementation

Files:

• WcfMessageSender.cs (modified) (2 diffs)
• WcfMessageService.cs (modified) (2 diffs)
• WcfPeerListManager.cs (modified) (1 diff)

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

@@ -1 +1 @@
 using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
 using System.ServiceModel;
+using System.Threading.Tasks;
+using System.Timers;
 using DistributedGA.Core.Domain;
 using DistributedGA.Core.Interface;
@@ -6 +10 @@
 namespace DistributedGA.Core.Implementation {
   public class WcfMessageSender : IMessageSender {
+
     private PeerInfo myself;
+
+    private ConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>> bufferedMessages;
+
+    private Timer timer; //sends cached messages to network in background
 
     public void Init(PeerInfo source) {
       myself = source;
+      bufferedMessages = new ConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>>();
+      timer = new Timer(1000 * 60); //each 5 minutes
+      timer.Elapsed += GenerateSendingTasks;
+      timer.Start();
     }
 
     public void SendData(PeerInfo destination, byte[][] data) {
+      bufferedMessages.Enqueue(new KeyValuePair<PeerInfo, byte[][]>(destination, data));
+    }
 
-      var serviceUrl = "DistributedGA.svc";
-      var baseUri = new Uri(string.Concat("net.tcp://", destination.IpAddress, ":", destination.Port, "/DistributedGA"));
-      var serviceUri = new Uri(baseUri, serviceUrl);
+    public void Dispose() {
+      timer.Stop();
+      timer.Dispose();
+      timer = null;
+    }
 
-      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...
+    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);
         }
       }
     }
 
-    public void Dispose() {
+    /// <summary>
+    /// sends data to another peer, is used by a single thread to run in background
+    /// </summary>
+    /// <param name="destination">destination peer</param>
+    /// <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);
 
+        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
     }
   }

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

@@ -31 +31 @@
                 var serviceUri = new Uri(baseUri, serviceUrl);
                 NetTcpBinding binding = new NetTcpBinding();
-                //using (var host = new ServiceHost(typeof(MessageContractImpl), serviceUri))
 
                 using (var host = new ServiceHost(messageContract, serviceUri))
@@ -42 +41 @@
                 }
             }).Start();
-            //close service again:
-            //  _ResetEvent.Set();
             return port;
         }

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

@@ -7 +7 @@
 using System.Timers;
 
-namespace DistributedGA.Core.Implementation
-{
-    public class WcfPeerListManager : IPeerListManager
-    {
-        private string serverString = null;
+namespace DistributedGA.Core.Implementation {
+  public class WcfPeerListManager : IPeerListManager {
 
-        private IContactService client = null;
+    private string serverString = null;
 
-        private IContactService heartbeatClient = null;
+    private PeerInfo myself = null;
 
-        private PeerInfo myself = null;
+    private Timer timer = null; //sends heartbeat to contact-server
 
-        private Timer timer = null; //sends heartbeat to contact-server
+    private ChannelFactory<IContactService> myChannelFactory;
 
-        public void Init(PeerInfo source, string contactServerUrl)
-        {
-            serverString = contactServerUrl;
-            client = CreateClient();
-            heartbeatClient = CreateClient();
-            myself = source;
-            client.RegisterPeer(source);
-            timer = new Timer(1000 ); //each 5 minutes
-            timer.Elapsed += SendHeartbeatToServer;
-            timer.Start();
-        }
+    private IContactService client;
+
+    private IContactService heartbeatClient;
+
+    public void Init(PeerInfo source, string contactServerUrl) {
+      serverString = contactServerUrl;
+      myself = source;
+      //Init ChannelFactory and Clients
+      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);
+      //Start heartbeat timer
+      timer = new Timer(1000); //each 5 minutes
+      timer.Elapsed += SendHeartbeatToServer;
+      timer.Start();
+    }
+
+    public List<PeerInfo> GetPeerList() {
+      try {
+        var allPeers = client.GetPeerList(myself); //maybe timout exception...
+        var peersForMessaging = ChoosePeersForMessaging(allPeers);
+        //return peersForMessaging;
+        return allPeers; //TODO: Enable 10% list communication
+      }
+      catch { } //if maybe sending failed (because of connection lost, etc.): just ignore
+      return new List<PeerInfo>();
+    }
+
+    public void SendLogToServer(string msg) {
+      client.MakeLog(myself, msg);
+    }
+
+    public void Dispose() {
+      timer.Stop();
+      timer.Dispose();
+      ((IClientChannel)client).Close();
+      ((IClientChannel)heartbeatClient).Close();
+      myChannelFactory.Close();
+    }
 
 
+    private List<PeerInfo> ChoosePeersForMessaging(List<PeerInfo> allPeers) {
+      //communicate with 10% of the network
+      int noOfPeers = allPeers.Count / 10;
+      List<int> indexList = GetRandomItemIndexes(noOfPeers, 0, allPeers.Count - 1);
+      List<PeerInfo> res = new List<PeerInfo>();
+      foreach (int index in indexList) {
+        res.Add(allPeers.ElementAt(index));
+      }
+      return res;
+    }
 
-        public List<PeerInfo> GetPeerList()
-        {
-            var allPeers = client.GetPeerList(myself);
-            var peersForMessaging = ChoosePeersForMessaging(allPeers);
-            //return peersForMessaging;
-            return allPeers; //TODO: Enable 10% list communication
+    private List<int> GetRandomItemIndexes(int noOfItems, int minValue, int maxValue) {
+      List<int> res = new List<int>();
+      Random rnd = new Random();
+      int tmp = -1;
+      while (res.Count < noOfItems) {
+        tmp = rnd.Next(minValue, maxValue);
+        if (!res.Contains(tmp)) {
+          res.Add(tmp);
         }
+      }
+      return res;
+    }
 
-        public void SendLogToServer(string msg)
-        {
-            client.MakeLog(myself, msg);
-        }
+    private void SendHeartbeatToServer(object sender, ElapsedEventArgs e) {
+      try {
+        heartbeatClient.UpdateHeartbeat(myself);
+      }
+      catch { } //nothing to do
+    }
 
-        private IContactService CreateClient()
-        {
-            var binding = new NetTcpBinding();
-            var endpoint = new EndpointAddress(serverString);
-            var myChannelFactory = new ChannelFactory<IContactService>(binding, endpoint);
-
-            IContactService client = null;
-            client = myChannelFactory.CreateChannel();
-            return client;
-        }
-
-        private List<PeerInfo> ChoosePeersForMessaging(List<PeerInfo> allPeers)
-        {
-            //communicate with 10% of the network
-            int noOfPeers = allPeers.Count / 10;
-            List<int> indexList = GetRandomItemIndexes(noOfPeers, 0, allPeers.Count - 1);
-            List<PeerInfo> res = new List<PeerInfo>();
-            foreach (int index in indexList)
-            {
-                res.Add(allPeers.ElementAt(index));
-            }
-            return res;
-        }
-
-        private List<int> GetRandomItemIndexes(int noOfItems, int minValue, int maxValue)
-        {
-            List<int> res = new List<int>();
-            Random rnd = new Random();
-            int tmp = -1;
-            while (res.Count < noOfItems)
-            {
-                tmp = rnd.Next(minValue, maxValue);
-                if (!res.Contains(tmp))
-                {
-                    res.Add(tmp);
-                }
-            }
-            return res;
-        }
-
-        private void SendHeartbeatToServer(object sender, ElapsedEventArgs e)
-        {
-            try
-            {
-                heartbeatClient.UpdateHeartbeat(myself);
-            }
-            catch { } //nothing to do, exception is raised when getting peer list
-        }
-
-
-
-        public void Dispose()
-        {
-            timer.Stop();
-            timer.Dispose();
-            timer = null;
-        }
-    }
+  }
 }