Export (0) Print
Expand All

Walkthrough: Creating a Custom Dataflow Block Type

.NET Framework 4.5

Although the TPL Dataflow Library provides several dataflow block types that enable a variety of functionality, you can also create custom block types. This document describes how to create a dataflow block type that implements custom behavior.

Read Dataflow (Task Parallel Library) before you read this document.

Tip Tip

The TPL Dataflow Library (System.Threading.Tasks.Dataflow namespace) is not distributed with the .NET Framework 4.5. To install the System.Threading.Tasks.Dataflow namespace, open your project in Visual Studio 2012, choose Manage NuGet Packages from the Project menu, and search online for the Microsoft.Tpl.Dataflow package.

Consider a dataflow application that requires that input values be buffered and then output in a sliding window manner. For example, for the input values {0, 1, 2, 3, 4, 5} and a window size of three, a sliding window dataflow block produces the output arrays {0, 1, 2}, {1, 2, 3}, {2, 3, 4}, and {3, 4, 5}. The following sections describe two ways to create a dataflow block type that implements this custom behavior. The first technique uses the Encapsulate(Of TInput, TOutput) method to combine the functionality of an ISourceBlock(Of TOutput) object and an ITargetBlock(Of TInput) object into one propagator block. The second technique defines a class that derives from IPropagatorBlock(Of TInput, TOutput) and combines existing functionality to perform custom behavior.

The following example uses the Encapsulate(Of TInput, TOutput) method to create a propagator block from a target and a source. A propagator block enables a source block and a target block to act as a receiver and sender of data.

This technique is useful when you require custom dataflow functionality, but you do not require a type that provides additional methods, properties, or fields.

' Creates a IPropagatorBlock<T, T[]> object propagates data in a  
' sliding window fashion. 
Public Shared Function CreateSlidingWindow(Of T)(ByVal windowSize As Integer) As IPropagatorBlock(Of T, T())
    ' Create a queue to hold messages. 
    Dim queue = New Queue(Of T)()

    ' The source part of the propagator holds arrays of size windowSize 
    ' and propagates data out to any connected targets. 
    Dim source = New BufferBlock(Of T())()

    ' The target part receives data and adds them to the queue. 
    Dim target = New ActionBlock(Of T)(Sub(item)
        ' Add the item to the queue. 
        ' Remove the oldest item when the queue size exceeds the window size. 
        ' Post the data in the queue to the source block when the queue size 
        ' equals the window size.
        queue.Enqueue(item)
        If queue.Count > windowSize Then
            queue.Dequeue()
        End If 
        If queue.Count = windowSize Then
            source.Post(queue.ToArray())
        End If 
    End Sub)

    ' When the target is set to the completed state, propagate out any 
    ' remaining data and set the source to the completed state.
    target.Completion.ContinueWith(Sub()
        If queue.Count > 0 AndAlso queue.Count < windowSize Then
            source.Post(queue.ToArray())
        End If
        source.Complete()
    End Sub)

    ' Return a IPropagatorBlock<T, T[]> object that encapsulates the  
    ' target and source blocks. 
    Return DataflowBlock.Encapsulate(target, source)
End Function

The following example shows the SlidingWindowBlock class. This class derives from IPropagatorBlock(Of TInput, TOutput) so that it can act as both a source and a target of data. As in the previous example, the SlidingWindowBlock class is built on existing dataflow block types. However, the SlidingWindowBlock class also implements the methods that are required by the ISourceBlock(Of TOutput), ITargetBlock(Of TInput), and IDataflowBlock interfaces. These methods all forward work to the predefined dataflow block type members. For example, the Post method defers work to the m_target data member, which is also an ITargetBlock(Of TInput) object.

This technique is useful when you require custom dataflow functionality, and also require a type that provides additional methods, properties, or fields. For example, the SlidingWindowBlock class also derives from IReceivableSourceBlock(Of TOutput) so that it can provide the TryReceive and TryReceiveAll methods. The SlidingWindowBlock class also demonstrates extensibility by providing the WindowSize property, which retrieves the number of elements in the sliding window.

    ' Propagates data in a sliding window fashion. 
    Public Class SlidingWindowBlock(Of T)
        Implements IPropagatorBlock(Of T, T()), IReceivableSourceBlock(Of T())
        ' The size of the window. 
        Private ReadOnly m_windowSize As Integer 
        ' The target part of the block. 
        Private ReadOnly m_target As ITargetBlock(Of T)
        ' The source part of the block. 
        Private ReadOnly m_source As IReceivableSourceBlock(Of T())

        ' Constructs a SlidingWindowBlock object. 
        Public Sub New(ByVal windowSize As Integer)
            ' Create a queue to hold messages. 
            Dim queue = New Queue(Of T)()

            ' The source part of the propagator holds arrays of size windowSize 
            ' and propagates data out to any connected targets. 
            Dim source = New BufferBlock(Of T())()

            ' The target part receives data and adds them to the queue. 
            Dim target = New ActionBlock(Of T)(Sub(item)
                ' Add the item to the queue. 
                ' Remove the oldest item when the queue size exceeds the window size. 
                ' Post the data in the queue to the source block when the queue size 
                ' equals the window size.
                queue.Enqueue(item)
                If queue.Count > windowSize Then
                    queue.Dequeue()
                End If 
                If queue.Count = windowSize Then
                    source.Post(queue.ToArray())
                End If 
            End Sub)

            ' When the target is set to the completed state, propagate out any 
            ' remaining data and set the source to the completed state.
            target.Completion.ContinueWith(Sub()
                If queue.Count > 0 AndAlso queue.Count < windowSize Then
                    source.Post(queue.ToArray())
                End If
                source.Complete()
            End Sub)

            m_windowSize = windowSize
            m_target = target
            m_source = source
        End Sub 

        ' Retrieves the size of the window. 
        Public ReadOnly Property WindowSize() As Integer 
            Get 
                Return m_windowSize
            End Get 
        End Property 

        '#Region "IReceivableSourceBlock<TOutput> members" 

        ' Attempts to synchronously receive an item from the source. 
        Public Function TryReceive(ByVal filter As Predicate(Of T()), <System.Runtime.InteropServices.Out()> ByRef item() As T) As Boolean Implements IReceivableSourceBlock(Of T()).TryReceive
            Return m_source.TryReceive(filter, item)
        End Function 

        ' Attempts to remove all available elements from the source into a new  
        ' array that is returned. 
        Public Function TryReceiveAll(<System.Runtime.InteropServices.Out()> ByRef items As IList(Of T())) As Boolean Implements IReceivableSourceBlock(Of T()).TryReceiveAll
            Return m_source.TryReceiveAll(items)
        End Function 

        '#End Region

#Region "ISourceBlock<TOutput> members" 

        ' Links this dataflow block to the provided target. 
        Public Function LinkTo(ByVal target As ITargetBlock(Of T()), ByVal linkOptions As DataflowLinkOptions) As IDisposable Implements ISourceBlock(Of T()).LinkTo
            Return m_source.LinkTo(target, linkOptions)
        End Function 

        ' Called by a target to reserve a message previously offered by a source  
        ' but not yet consumed by this target. 
        Private Function ReserveMessage(ByVal messageHeader As DataflowMessageHeader, ByVal target As ITargetBlock(Of T())) As Boolean Implements ISourceBlock(Of T()).ReserveMessage
            Return m_source.ReserveMessage(messageHeader, target)
        End Function 

        ' Called by a target to consume a previously offered message from a source. 
        Private Function ConsumeMessage(ByVal messageHeader As DataflowMessageHeader, ByVal target As ITargetBlock(Of T()), ByRef messageConsumed As Boolean) As T() Implements ISourceBlock(Of T()).ConsumeMessage
            Return m_source.ConsumeMessage(messageHeader, target, messageConsumed)
        End Function 

        ' Called by a target to release a previously reserved message from a source. 
        Private Sub ReleaseReservation(ByVal messageHeader As DataflowMessageHeader, ByVal target As ITargetBlock(Of T())) Implements ISourceBlock(Of T()).ReleaseReservation
            m_source.ReleaseReservation(messageHeader, target)
        End Sub

#End Region

#Region "ITargetBlock<TInput> members" 

        ' Asynchronously passes a message to the target block, giving the target the  
        ' opportunity to consume the message. 
        Private Function OfferMessage(ByVal messageHeader As DataflowMessageHeader, ByVal messageValue As T, ByVal source As ISourceBlock(Of T), ByVal consumeToAccept As Boolean) As DataflowMessageStatus Implements ITargetBlock(Of T).OfferMessage
            Return m_target.OfferMessage(messageHeader, messageValue, source, consumeToAccept)
        End Function

#End Region

#Region "IDataflowBlock members" 

        ' Gets a Task that represents the completion of this dataflow block. 
        Public ReadOnly Property Completion() As Task Implements IDataflowBlock.Completion
            Get 
                Return m_source.Completion
            End Get 
        End Property 

        ' Signals to this target block that it should not accept any more messages,  
        ' nor consume postponed messages.  
        Public Sub Complete() Implements IDataflowBlock.Complete
            m_target.Complete()
        End Sub 

        Public Sub Fault(ByVal [error] As Exception) Implements IDataflowBlock.Fault
            m_target.Fault([error])
        End Sub

#End Region
    End Class

The following example shows the complete code for this walkthrough. It also demonstrates how to use the both sliding window blocks in a method that writes to the block, reads from it, and prints the results to the console.

Imports System
Imports System.Collections.Generic
Imports System.Linq
Imports System.Threading.Tasks
Imports System.Threading.Tasks.Dataflow

' Demonstrates how to create a custom dataflow block type. 
Friend Class Program
    ' Creates a IPropagatorBlock<T, T[]> object propagates data in a  
    ' sliding window fashion. 
    Public Shared Function CreateSlidingWindow(Of T)(ByVal windowSize As Integer) As IPropagatorBlock(Of T, T())
        ' Create a queue to hold messages. 
        Dim queue = New Queue(Of T)()

        ' The source part of the propagator holds arrays of size windowSize 
        ' and propagates data out to any connected targets. 
        Dim source = New BufferBlock(Of T())()

        ' The target part receives data and adds them to the queue. 
        Dim target = New ActionBlock(Of T)(Sub(item)
            ' Add the item to the queue. 
            ' Remove the oldest item when the queue size exceeds the window size. 
            ' Post the data in the queue to the source block when the queue size 
            ' equals the window size.
            queue.Enqueue(item)
            If queue.Count > windowSize Then
                queue.Dequeue()
            End If 
            If queue.Count = windowSize Then
                source.Post(queue.ToArray())
            End If 
        End Sub)

        ' When the target is set to the completed state, propagate out any 
        ' remaining data and set the source to the completed state.
        target.Completion.ContinueWith(Sub()
            If queue.Count > 0 AndAlso queue.Count < windowSize Then
                source.Post(queue.ToArray())
            End If
            source.Complete()
        End Sub)

        ' Return a IPropagatorBlock<T, T[]> object that encapsulates the  
        ' target and source blocks. 
        Return DataflowBlock.Encapsulate(target, source)
    End Function 

    ' Propagates data in a sliding window fashion. 
    Public Class SlidingWindowBlock(Of T)
        Implements IPropagatorBlock(Of T, T()), IReceivableSourceBlock(Of T())
        ' The size of the window. 
        Private ReadOnly m_windowSize As Integer 
        ' The target part of the block. 
        Private ReadOnly m_target As ITargetBlock(Of T)
        ' The source part of the block. 
        Private ReadOnly m_source As IReceivableSourceBlock(Of T())

        ' Constructs a SlidingWindowBlock object. 
        Public Sub New(ByVal windowSize As Integer)
            ' Create a queue to hold messages. 
            Dim queue = New Queue(Of T)()

            ' The source part of the propagator holds arrays of size windowSize 
            ' and propagates data out to any connected targets. 
            Dim source = New BufferBlock(Of T())()

            ' The target part receives data and adds them to the queue. 
            Dim target = New ActionBlock(Of T)(Sub(item)
                ' Add the item to the queue. 
                ' Remove the oldest item when the queue size exceeds the window size. 
                ' Post the data in the queue to the source block when the queue size 
                ' equals the window size.
                queue.Enqueue(item)
                If queue.Count > windowSize Then
                    queue.Dequeue()
                End If 
                If queue.Count = windowSize Then
                    source.Post(queue.ToArray())
                End If 
            End Sub)

            ' When the target is set to the completed state, propagate out any 
            ' remaining data and set the source to the completed state.
            target.Completion.ContinueWith(Sub()
                If queue.Count > 0 AndAlso queue.Count < windowSize Then
                    source.Post(queue.ToArray())
                End If
                source.Complete()
            End Sub)

            m_windowSize = windowSize
            m_target = target
            m_source = source
        End Sub 

        ' Retrieves the size of the window. 
        Public ReadOnly Property WindowSize() As Integer 
            Get 
                Return m_windowSize
            End Get 
        End Property 

        '#Region "IReceivableSourceBlock<TOutput> members" 

        ' Attempts to synchronously receive an item from the source. 
        Public Function TryReceive(ByVal filter As Predicate(Of T()), <System.Runtime.InteropServices.Out()> ByRef item() As T) As Boolean Implements IReceivableSourceBlock(Of T()).TryReceive
            Return m_source.TryReceive(filter, item)
        End Function 

        ' Attempts to remove all available elements from the source into a new  
        ' array that is returned. 
        Public Function TryReceiveAll(<System.Runtime.InteropServices.Out()> ByRef items As IList(Of T())) As Boolean Implements IReceivableSourceBlock(Of T()).TryReceiveAll
            Return m_source.TryReceiveAll(items)
        End Function 

        '#End Region

#Region "ISourceBlock<TOutput> members" 

        ' Links this dataflow block to the provided target. 
        Public Function LinkTo(ByVal target As ITargetBlock(Of T()), ByVal linkOptions As DataflowLinkOptions) As IDisposable Implements ISourceBlock(Of T()).LinkTo
            Return m_source.LinkTo(target, linkOptions)
        End Function 

        ' Called by a target to reserve a message previously offered by a source  
        ' but not yet consumed by this target. 
        Private Function ReserveMessage(ByVal messageHeader As DataflowMessageHeader, ByVal target As ITargetBlock(Of T())) As Boolean Implements ISourceBlock(Of T()).ReserveMessage
            Return m_source.ReserveMessage(messageHeader, target)
        End Function 

        ' Called by a target to consume a previously offered message from a source. 
        Private Function ConsumeMessage(ByVal messageHeader As DataflowMessageHeader, ByVal target As ITargetBlock(Of T()), ByRef messageConsumed As Boolean) As T() Implements ISourceBlock(Of T()).ConsumeMessage
            Return m_source.ConsumeMessage(messageHeader, target, messageConsumed)
        End Function 

        ' Called by a target to release a previously reserved message from a source. 
        Private Sub ReleaseReservation(ByVal messageHeader As DataflowMessageHeader, ByVal target As ITargetBlock(Of T())) Implements ISourceBlock(Of T()).ReleaseReservation
            m_source.ReleaseReservation(messageHeader, target)
        End Sub

#End Region

#Region "ITargetBlock<TInput> members" 

        ' Asynchronously passes a message to the target block, giving the target the  
        ' opportunity to consume the message. 
        Private Function OfferMessage(ByVal messageHeader As DataflowMessageHeader, ByVal messageValue As T, ByVal source As ISourceBlock(Of T), ByVal consumeToAccept As Boolean) As DataflowMessageStatus Implements ITargetBlock(Of T).OfferMessage
            Return m_target.OfferMessage(messageHeader, messageValue, source, consumeToAccept)
        End Function

#End Region

#Region "IDataflowBlock members" 

        ' Gets a Task that represents the completion of this dataflow block. 
        Public ReadOnly Property Completion() As Task Implements IDataflowBlock.Completion
            Get 
                Return m_source.Completion
            End Get 
        End Property 

        ' Signals to this target block that it should not accept any more messages,  
        ' nor consume postponed messages.  
        Public Sub Complete() Implements IDataflowBlock.Complete
            m_target.Complete()
        End Sub 

        Public Sub Fault(ByVal [error] As Exception) Implements IDataflowBlock.Fault
            m_target.Fault([error])
        End Sub

#End Region
    End Class 

    ' Demonstrates usage of the sliding window block by sending the provided 
    ' values to the provided propagator block and printing the output of  
    ' that block to the console. 
    Private Shared Sub DemonstrateSlidingWindow(Of T)(ByVal slidingWindow As IPropagatorBlock(Of T, T()), ByVal values As IEnumerable(Of T))
        ' Create an action block that prints arrays of data to the console. 
        Dim windowComma As String = String.Empty
        Dim printWindow = New ActionBlock(Of T())(Sub(window)
            Console.Write(windowComma)
            Console.Write("{")
            Dim comma As String = String.Empty
            For Each item As T In window
                Console.Write(comma)
                Console.Write(item)
                comma = "," 
            Next item
            Console.Write("}")
            windowComma = ", " 
        End Sub)

        ' Link the printer block to the sliding window block.
        slidingWindow.LinkTo(printWindow)

        ' Set the printer block to the completed state when the sliding window 
        ' block completes.
        slidingWindow.Completion.ContinueWith(Sub() printWindow.Complete())

        ' Print an additional newline to the console when the printer block completes. 
        Dim completion = printWindow.Completion.ContinueWith(Sub() Console.WriteLine())

        ' Post the provided values to the sliding window block and then wait 
        ' for the sliding window block to complete. 
        For Each value As T In values
            slidingWindow.Post(value)
        Next value
        slidingWindow.Complete()

        ' Wait for the printer to complete and perform its final action.
        completion.Wait()
    End Sub 

    Shared Sub Main(ByVal args() As String)

        Console.Write("Using the DataflowBlockExtensions.Encapsulate method ")
        Console.WriteLine("(T=int, windowSize=3):")
        DemonstrateSlidingWindow(CreateSlidingWindow(Of Integer)(3), Enumerable.Range(0, 10))

        Console.WriteLine()

        Dim slidingWindow = New SlidingWindowBlock(Of Char)(4)

        Console.Write("Using SlidingWindowBlock<T> ")
        Console.WriteLine("(T=char, windowSize={0}):", slidingWindow.WindowSize)
        DemonstrateSlidingWindow(slidingWindow, _
            From n In Enumerable.Range(65, 10) _
            Select ChrW(n))
    End Sub 
End Class 

' Output: 
'Using the DataflowBlockExtensions.Encapsulate method (T=int, windowSize=3): 
'{0,1,2}, {1,2,3}, {2,3,4}, {3,4,5}, {4,5,6}, {5,6,7}, {6,7,8}, {7,8,9} 

'Using SlidingWindowBlock<T> (T=char, windowSize=4): 
'{A,B,C,D}, {B,C,D,E}, {C,D,E,F}, {D,E,F,G}, {E,F,G,H}, {F,G,H,I}, {G,H,I,J} 
' 

Copy the example code and paste it in a Visual Studio project, or paste it in a file that is named SlidingWindowBlock.cs (SlidingWindowBlock.vb for Visual Basic) and then run the following command in a Visual Studio Command Prompt window.

Visual C#

csc.exe /r:System.Threading.Tasks.Dataflow.dll SlidingWindowBlock.cs

Visual Basic

vbc.exe /r:System.Threading.Tasks.Dataflow.dll SlidingWindowBlock.vb

Show:
© 2014 Microsoft