Looking for clarity on MSDN TPL Dataflow sample, and parallelization

user_420003 0
Viewed 46

I'm looking at this MSDN sample code and made a few adjustments to figure out what would happen if the constrained resource (as defined in the demo) is "memory" with one, two, four, or more instances created.

For example, Main() has a defined quantity of these that can be increased or decreased in quantity

networkResources.Post(new NetworkResource() { Name = "eth0" });
memoryResources.Post(new MemoryResource() { Name = "Memory01" });
fileResources.Post(new FileResource() { Name = "MFMHardDrive01" });

I'm also trying to understand more about the ActionBlock, and in particular the line where the "resources are released back to their pools". Is that line just a reference to additional work needed to be done, as it simply seems to loop the message like a while() loop with no end, making this feel similar to a SpinWait() for the thread.

The lines that add this circular reference loop, (possibly meant to pull from another queue) are below. Perhaps this is acting as some kind of thread heartbeat to prevent inefficient deallocation/reallocation of threads

networkResources.Post(data.Item1);
memoryResources.Post(data.Item2);

To make figuring this out easier, I added color to the output to make the threads more easily identifiable.

My understanding and expectation is that by adding more MemoryResources is that the system should run in parallel, similar to Windows NT administration where I can assign Network Cards to individual CPUs for more efficient parallelization. When I add additional MemoryResource items, no parallelization occurs within the same domain (network) or many source domains (network + disk).

Where is the bottleneck for parallelization defined?

enter image description here

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks.Dataflow;

// Demonstrates how to use non-greedy join blocks to distribute
// resources among a dataflow network.
class Program
{
    // Represents a resource. A derived class might represent
    // a limited resource such as a memory, network, or I/O
    // device.
    abstract class Resource
    {
        public object TeapotIsResource { get; set; }

        public int TouchCount { get; set; }
    }

    // Represents a memory resource. For brevity, the details of
    // this class are omitted.
    class MemoryResource : Resource
    {
        public MemoryResource()
        {
            NetworkSequenceID = ran.Next();
        }

        Random ran = new Random((int)DateTime.Now.Ticks);
        public int NetworkSequenceID { get; set; }

        public object TeapotIsMemory { get; set; }

        public string Name { get; set; }
    }

    // Represents a network resource. For brevity, the details of
    // this class are omitted.
    class NetworkResource : Resource
    {
        public NetworkResource()
        {
            NetworkSequenceID = ran.Next();
        }

        Random ran = new Random((int)DateTime.Now.Ticks);
        public int NetworkSequenceID { get; set; }

        public object TeapotIsNetwork { get; set; }

        public string Name { get; set; }
    }

    // Represents a file resource. For brevity, the details of
    // this class are omitted.
    class FileResource : Resource
    {
        public FileResource()
        {
            NetworkSequenceID = ran.Next();
        }

        Random ran = new Random((int)DateTime.Now.Ticks);
        public int NetworkSequenceID { get; set; }

        public string Name { get; set; }
    }

    public static  DateTime TimeForFile = DateTime.UtcNow;
    public  static DateTime TimeForNetwork = DateTime.UtcNow;

    static int NetworkSequenceID = 0;
    static void Main(string[] args)
    {
        Console.ForegroundColor = ConsoleColor.White;
        colorMap.Add(867);


        // Create three BufferBlock<T> objects. Each object holds a different
        // type of resource.
        var networkResources = new BufferBlock<NetworkResource>();
        var fileResources = new BufferBlock<FileResource>();
        var memoryResources = new BufferBlock<MemoryResource>();

        // Create two non-greedy JoinBlock<T1, T2> objects.
        // The first join works with network and memory resources;
        // the second pool works with file and memory resources.

        var joinNetworkAndMemoryResources =
           new JoinBlock<NetworkResource, MemoryResource>(
              new GroupingDataflowBlockOptions
              {
                  Greedy = false,
              });

        var joinFileAndMemoryResources =
           new JoinBlock<FileResource, MemoryResource>(
              new GroupingDataflowBlockOptions
              {
                  Greedy = false,
              });

        // Create two ActionBlock<T> objects.
        // The first block acts on a network resource and a memory resource.
        // The second block acts on a file resource and a memory resource.

        var networkMemoryAction =
           new ActionBlock<Tuple<NetworkResource, MemoryResource>>(
              data =>
              {
                  // Perform some action on the resources.

                  // Print a message.
                  PrintLine("Network worker", data.Item1.Name, data.Item1.NetworkSequenceID, networkResources.Count,
                  "using resources on", data.Item2.Name, data.Item2.NetworkSequenceID, memoryResources.Count);

                  // Simulate a lengthy operation that uses the resources.
                  Thread.Sleep(2000);

                  // Print a message.
                  PrintLine("Network worker", data.Item1.Name, data.Item1.NetworkSequenceID, networkResources.Count,
                  "finished using resources on", data.Item2.Name, data.Item2.NetworkSequenceID, memoryResources.Count);

                  data.Item1.TouchCount = data.Item1.TouchCount + 1;
                  data.Item2.TouchCount = data.Item2.TouchCount + 1;

                  if (data.Item2.TouchCount == 10)
                  {
                      Console.WriteLine(DateTime.UtcNow - TimeForNetwork);
                  }

                  networkResources.Post(data.Item1);
                  memoryResources.Post(data.Item2);
              });

        var fileMemoryAction =
           new ActionBlock<Tuple<FileResource, MemoryResource>>(
              data =>
              {
                  // Perform some action on the resources.
                  // Print a message. 
                  PrintLine("File worker", data.Item1.Name, data.Item1.NetworkSequenceID, fileResources.Count,
                      "using resources on", data.Item2.Name, data.Item2.NetworkSequenceID, memoryResources.Count);

                  // Simulate a lengthy operation that uses the resources.
                  //Thread.Sleep(new Random().Next(500, 2000));
                  Thread.Sleep(2000);

                  // Print a message.
                  PrintLine("File worker", data.Item1.Name, data.Item1.NetworkSequenceID, fileResources.Count,
               "finished using resources on", data.Item2.Name, data.Item2.NetworkSequenceID, memoryResources.Count);

                  data.Item1.TouchCount = data.Item1.TouchCount + 1;
                  data.Item2.TouchCount = data.Item2.TouchCount + 1;

                  if (data.Item2.TouchCount == 10 )
                  {
                      Console.WriteLine(DateTime.UtcNow - TimeForFile);
                  }
                  // Release the resources back to their respective pools.
                  fileResources.Post(data.Item1);
                  memoryResources.Post(data.Item2);
              });

        // Link the resource pools to the JoinBlock<T1, T2> objects.
        // Because these join blocks operate in non-greedy mode, they do not
        // take the resource from a pool until all resources are available from
        // all pools.

        networkResources.LinkTo(joinNetworkAndMemoryResources.Target1);
        memoryResources.LinkTo(joinNetworkAndMemoryResources.Target2);

        fileResources.LinkTo(joinFileAndMemoryResources.Target1);
        memoryResources.LinkTo(joinFileAndMemoryResources.Target2);

        // Link the JoinBlock<T1, T2> objects to the ActionBlock<T> objects.

        joinNetworkAndMemoryResources.LinkTo(networkMemoryAction);
        joinFileAndMemoryResources.LinkTo(fileMemoryAction);

        // Populate the resource pools. In this example, network and
        // file resources are more abundant than memory resources.

        Console.WriteLine("ADDING: Allocating 4 Network interfaces");
        networkResources.Post(new NetworkResource() { Name = "eth0" });
        networkResources.Post(new NetworkResource() { Name = "eth1" });
        //networkResources.Post(new NetworkResource() { Name = "eth2" });
        //networkResources.Post(new NetworkResource() { Name = "eth3" });

        Console.WriteLine("ADDING: Allocate a small memory resource");
        memoryResources.Post(new MemoryResource() { Name = "Memory01" }); 
        memoryResources.Post(new MemoryResource() { Name = "Memory02" }); 

        Console.WriteLine("ADDING: Old disk technology simulator ");
        fileResources.Post(new FileResource() { Name = "MFMHardDrive01" });
        //fileResources.Post(new FileResource() { Name = "MFMHardDrive02" });

        // Allow data to flow through the network for several seconds.
        Thread.Sleep(10000000);
    }

    public static void MessageProcessor(string from)
    {
    }

    private static void PrintLine(string workerType, string dataItem1Name, int networkSequenceID1, int count1, string onString, string item2Name, int networkSequenceID2, int count2)
    {
        // Many threads writing to console
        lock (colorMap)
        {
            int fixAPICountingAtZero = 1;

            Console.Write(workerType + " " + dataItem1Name + "/");
            SetConsoleColor(networkSequenceID1);
            Console.Write("(" + (count1 + fixAPICountingAtZero) + "): " + onString + item2Name + "/");
            SetConsoleColor(networkSequenceID2);
            Console.WriteLine(" (" + (count2 + fixAPICountingAtZero) + ")");
        }
    }

    static  List<int> colorMap = new List<int>();
    static string SetConsoleColor(int anumber)
    {
       
            if (!colorMap.Contains(anumber))
            {
                colorMap.Add(anumber);
            }

            var original = Console.ForegroundColor;
            Console.ForegroundColor = (ConsoleColor)colorMap.IndexOf(anumber);
            Console.Write(anumber);
            Console.ForegroundColor = original;

        return "";
    }
}

/* Sample output:
File worker: using resources...
File worker: finished using resources...
Network worker: using resources...
Network worker: finished using resources...
File worker: using resources...
File worker: finished using resources...
Network worker: using resources...
Network worker: finished using resources...
File worker: using resources...
File worker: finished using resources...
File worker: using resources...
File worker: finished using resources...
Network worker: using resources...
Network worker: finished using resources...
Network worker: using resources...
Network worker: finished using resources...
File worker: using resources...
*/
tpl-dataflow task-parallel-library msdn c#
1 Answers

If I'm understanding the problem correctly.

The first problem I can see is, the ActionBlock constructor can take options that default to the following if not set.

ExecutionDataflowBlockOptions

When specific configuration options are not set, the following defaults are used:

OptionsDefault

  • TaskScheduler : Default
  • CancellationToken :None
  • MaxMessagesPerTask :DataflowBlockOptions.Unbounded (-1)
  • BoundedCapacity : DataflowBlockOptions.Unbounded (-1)
  • MaxDegreeOfParallelism : 1

So, if you don't set the options on an ActionBlock explicitly it will be constrained to processing messages serially due to the fact MaxDegreeOfParallelism is defaulted to 1

Example

var blobk = new ActionBlock<T>(
     x => {...},
     new ExecutionDataflowBlockOptions
     {
         MaxDegreeOfParallelism = <something that makes sense to your solution>
     });

Note : Your current implementation uses synchronous workflows. To really open up the performance and scalability of a Dataflow pipeline, you will want to make sure you are using the Func<Task> delegates overloads and taking advantages of the async and await pattern for any IO bound workloads you can.

user_16129750
Related