Changeset 13937 for branches/thasling/DistributedGA/DistributedGA.Core

Timestamp:

06/24/16 21:23:47 (8 years ago)

Author:

thasling

Message:

#2615:
implemented ConcurrentQueue
made minor changes in Dispose-methods

Location:

branches/thasling/DistributedGA/DistributedGA.Core

Files:

• DistributedGA.Core.csproj (modified) (1 diff)
• Implementation/PeerNetworkMessageHandler.cs (modified) (4 diffs)
• Implementation/WcfMessageSender.cs (modified) (3 diffs)
• Implementation/WcfPeerListManager.cs (modified) (1 diff)
• Util (added)
• Util/SizedConcurrentQueue.cs (added)

branches/thasling/DistributedGA/DistributedGA.Core/DistributedGA.Core.csproj

@@ -64 +64 @@
     <Compile Include="Interface\IPeerListManager.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
+    <Compile Include="Util\SizedConcurrentQueue.cs" />
   </ItemGroup>
   <ItemGroup>

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

@@ -6 +6 @@
 using DistributedGA.Core.Domain;
 using DistributedGA.Core.Interface;
+using DistributedGA.Core.Util;
 
 namespace DistributedGA.Core.Implementation {
@@ -21 +22 @@
     //to queues are used to gather and and provide population more efficiently
     private object activeQueueLocker = new object();
-    private ConcurrentQueue<byte[]> writeQueue;
-    private ConcurrentQueue<byte[]> readQueue;
+    private SizedConcurrentQueue<byte[]> writeQueue;
+    private SizedConcurrentQueue<byte[]> readQueue;
 
     //uses IMessageService for recieving population from one peer at once
@@ -37 +38 @@
         }; // TODO: get own peerinfo
 
-        writeQueue = new ConcurrentQueue<byte[]>();
-        readQueue = new ConcurrentQueue<byte[]>();
-
+        writeQueue = new SizedConcurrentQueue<byte[]>();
+        readQueue = new SizedConcurrentQueue<byte[]>();
+        writeQueue.Limit = 1000;
+        readQueue.Limit = writeQueue.Limit;
 
         host = new WcfMessageService();
         ownInstance.Port = host.Init(ownInstance.IpAddress); //getting port, on which service is hostet
-        host.OnDataRecieved += new EventHandler<MessageRecieveEventArgs>(OnPopulationRecieved);
+        host.OnDataRecieved += new EventHandler<MessageRecieveEventArgs>(OnDataRecieved);
 
         peerListManager = new WcfPeerListManager();
@@ -131 +133 @@
     }
 
-    private void OnPopulationRecieved(object sender, MessageRecieveEventArgs e) {
+    private void OnDataRecieved(object sender, MessageRecieveEventArgs e) {
       if (e != null) {
         lock (activeQueueLocker) {

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

@@ -7 +7 @@
 using DistributedGA.Core.Domain;
 using DistributedGA.Core.Interface;
+using DistributedGA.Core.Util;
 
 namespace DistributedGA.Core.Implementation {
@@ -13 +14 @@
     private PeerInfo myself;
 
-    private ConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>> bufferedMessages;
+    private SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>> bufferedMessages;
 
     private Timer timer; //sends cached messages to network in background
@@ -19 +20 @@
     public void Init(PeerInfo source) {
       myself = source;
-      bufferedMessages = new ConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>>();
+      bufferedMessages = new SizedConcurrentQueue<KeyValuePair<PeerInfo, byte[][]>>();
+      bufferedMessages.Limit = 1000;
       timer = new Timer(1000 * 60); //each 5 minutes
       timer.Elapsed += GenerateSendingTasks;

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

@@ -60 +60 @@
       ((IClientChannel)heartbeatClient).Close();
       myChannelFactory.Close();
+      client = null;
+      myChannelFactory = null;
     }