Well, the title is self explanatory. My code is something like this:
var clients = new ConcurrentBag<SharingClient>();
var peers = new ConcurrentDictionary<Task, SharingClient>();
clients.Add(...); // This may happen on another thread
clients.Add(...); // This may happen on another thread
var token = cts.Token;
while (!token.IsCancellationRequested)
{
if (clients.Count == 0)
{
// Waiting for more clients
await TaskEx.Delay(500, token); // Using BCL.Async as I need to support Windows XP (.Net 4)
}
else if (clients.Count == 1)
{
// Close the left-alone client
clients.FirstOrDefault()?.Close(); // Close may remove or modify **_clients**.
}
else if (clients.Any(c => c.DataAvailable))
{
// There is some data, let see if we have timed out peers from last time
foreach (var p in peers.Where(peer => !peer.Key.IsCompleted))
{
// Close the timed out clients, will also terminate the task
p.Value.Close(); // Close may remove or modify **_clients**.
}
peers.Clear();
// Checking for possible operations and filling the array
Parallel.ForEach(clients.Where(c => c.DataAvailable), (item) =>
{
Parallel.ForEach(clients.Where(c => !c.Equals(item)), (item2) =>
{
peers.TryAdd(item2.Broadcast(item.Data), item2); //Broadcast may add, remove or modify **_clients**.
});
});
// Minimum of 10 secs for operations to run before getting timedout
await TaskEx.WhenAny(TaskEx.WhenAll(peers.Keys.ToArray()), TaskEx.Delay(10000, token)); // Using BCL.Async as I need to support Windows XP (.Net 4)
// Even tho some of them may have timed out by now, as we may have no data for the next operation, we will wait for data before deciding to close them
}
else
{
// Waiting for some data to appear - Recheck as soon as OS allows us
await TaskEx.Delay(1, token); // Using BCL.Async as I need to support Windows XP (.Net 4)
}
}
As you can clearly see, I used ConcurrentBag
and ConcurrentDictionary
along with await
, Linq, Simple foreach
, Parallel.ForEach
and BCL.Async methods.
As this is the first time I am using Parallel
, ConcurrentBag
and ConcurrentDictionary
; I want to ask the people more familiar with these parts of the Framework to tell me if there is a problem in logic, thread safety and/or new ways of doing things.
I am all ears, Thank you in advance
EDIT
Based on one MSDN documentation mentioned in Ivan's answer, for more safety I should remove the .Where()
method and merge it into the Parallel.ForEach
's body as an if
.
But still, there is little about .FirstOrDefault()
and .Any()
methods and their thread safety situation. I know that this is a good practice to use lock
every time a variable is going to get accessed. But in this case as I am using a ConcurrentBag
, I want to be sure about the necessity of copying the variable before running this lines of code:
if (clients.Any(client => client.IsConnected))
{
clients.FirstOrDefault()?.Close(); // Close may remove or modify **_clients**.
EDIT 2
By decompiling the .FirstOrDefault()
and .Any()
methods, I found both of them using a simple foreach
. So I believe they should be thread safe after all. So now the question is:
Is running a Parallel.ForEach
inside an other Parallel.ForEach
with same source and then modifying this source, alright? Or should I change the logic here?
Parallel.ForEach(clients, (item) =>
{
if (item.DataAvailable)
{
Parallel.ForEach(clients, (item2) =>
{
if (!item2.Equals(item))
{
// Modify **clients**