Questions tagged [tpl-dataflow]

TPL Dataflow (TDF) is a .NET library for building concurrent applications. It promotes actor/agent-oriented designs through primitives for in-process message passing, dataflow, and pipelining. TDF builds upon the TPL (Task Parallel Library) in .NET 4 and integrates with async language support in C#, Visual Basic, and F#. TDF lacks join/merge by key (like SSIS) and time-based windowing (available in Rx).

Filter by
Sorted by
Tagged with
1 vote
1 answer
17 views

Dataflow with BatchBlock - producer runs ahead

I'm designing a pretty simple producer-consumer with Dataflow. I need a BatchBlock for batching a rather slow consumer (which can only be single-threaded due to EF DbContext). Everything works as ...
smoksnes's user avatar
  • 10.4k
0 votes
1 answer
21 views

VS 2022 CPU profiling not giving meaningful results when using TPL Dataflow

I am trying to figure out where my bottlenecks are in an application that uses TPL Dataflow, and I'm not finding any meaningful results. When I break in my application and look at the CPU usage this ...
TJ Rockefeller's user avatar
1 vote
0 answers
22 views

Rabbit MQ thread safety in an IHostedService

I have three c# classes. A rabbitmq subscriber has a connect() method, subscribe(Action messageHandler) method, and an Ack(ulong messageId) method. This class is instantiated as a singleton and passed ...
csharpguy's user avatar
2 votes
1 answer
46 views

Why does my ActionBlock get a state of completed without me setting it?

I am puzzled at the moment and can't think of any reasonable explanation. Maybe someone here has an explanation. Setup: I have a class that generates a lot of workitems to be computed with an ...
Iaman Swtrse's user avatar
0 votes
1 answer
29 views

TPL DataFlow not executing

I have the following code: using System; using System.Net; using System.Threading.Tasks.Dataflow; using System.Collections.Generic; public class Program { public static void ...
DavidS's user avatar
  • 2,169
0 votes
0 answers
21 views

TPL DataFlow: Can I get away without a complete custom block?

I'm building up a data flow pipeline that receives live data from a service and eventually uploads it to the cloud using the azure device client. Minimizing the number of messages sent is one of the ...
Eddie's user avatar
  • 1
2 votes
0 answers
44 views

How to stream result from n parallel requests send to x clients concurrently?

I have several identical api services available on different clients to which I would like to send several http-requests. This data is can be transformed into a map (Lookup<TKey, TSource>). ...
André Nøbbe's user avatar
1 vote
1 answer
32 views

TPL Dataflow with Rx: Observer misses messages

In the sample coding below, the source and IObservable and also part of a pipeline. The sink at the end of the pipeline gets all the messages. But the observer gets only the first and last messages. ...
DavidY's user avatar
  • 347
1 vote
1 answer
94 views

TPL Dataflow block that modifies state and sends a single message after it completes

I just started to learn about TPL Dataflow and have a question as described below: "Block"s 1, 2, 3 holds references to states. They modify the states and send messages downstream each time ...
DavidY's user avatar
  • 347
0 votes
1 answer
233 views

HttpClient taking too long

I created a .Net Framework 4.7.2 console app that concurrently makes many requests in an API hosted in AWS. My problem is that the requests are taking too long. The API's response time is usually ...
Carlos Siestrup's user avatar
1 vote
1 answer
57 views

Enabling Await and Exception Capture with TPL Dataflow

I have designed a simple JobProcessor using TPL Data flow (my first time using it). I want to be able to create jobs, and have them invoked and placed on a priority queue (the PriorityBufferBlock). My ...
FluffMaster4000's user avatar
0 votes
0 answers
61 views

Asynchronous operation of the BatchBlock in BackgroundService

I need to receive messages from a queue broker and, after some processing, store them in a database. In order to save records in blocks, i use a BatchBlock. All this works through the ...
Alex's user avatar
  • 1
0 votes
0 answers
50 views

Passing HttpContext in a ActionBlock

I am doing AzureAD implict flow for authenticating on a ASP.NET Core (2.2) api. For a specific reference to a library the application is in .net 4.8 but the api is hosted inside asp.net core. The API ...
Auroops's user avatar
  • 11
1 vote
0 answers
82 views

Is it a problem that BufferBlock is always on?

I need log provider for my .NET Core project (especially, I want to send logs to Graylog with HTTP) and I searched all log repos on GitHub to find a good idea. At the end of the day I found a few ...
dev's user avatar
  • 21
-1 votes
1 answer
34 views

How can I "requeue" messages in a TPL Pipeline?

I have a fire-and-forget Web Service that processes incoming requests. The request needs to go through number of processes that interact with the database via a single-threaded TPL pipeline. If any ...
ilitirit's user avatar
  • 15.9k
1 vote
1 answer
78 views

Microsoft TPL Dataflow Library | MaxDegreeOfParallelism and IO-bound work

My use case is this: send 100,000+ web requests to our application server and wait the results. Here, most of the delay is IO-bound, not CPU-bound, so I understand the Dataflow libraries may not be ...
thomasrea0113's user avatar
0 votes
1 answer
107 views

How to implement a sorted buffer?

I need to traverse a collection of disjoint folders; each folder is associated to a visited time configurated somewhere in the folder. I then sort the folders, and process the one with the earliest ...
NilNul's user avatar
  • 19
1 vote
1 answer
74 views

Custom TPL Block: how to test Consume, Release, Reserve

Suppose I want to create a custom ISourceBlock<T>. From the MSDN sliding window example, I could begin the construction of my custom block like so: public sealed class SomeCustomBlock<T> : ...
Jacob Shanley's user avatar
1 vote
1 answer
110 views

Custom Dataflow transform block with dynamic MaxDegreeOfParallelism

I receive a sequence of objects (e.g., ItemGroup[]) that each contains multiple jobs (e.g., Item[]) and a max degree of parallelism value, for example: public record Item(string Name); public record ...
Ritmo2k's user avatar
  • 973
1 vote
1 answer
83 views

How to avoid reset timeout in batchblock

I'm using TPL DataFlow with BatchBlock which is triggered automatically when the number of currently queued or postponed items is less than the BatchSize after a timeout as given below : Timer ...
Julien Martin's user avatar
1 vote
0 answers
164 views

Use EF Core as scoped service from within parallel executing actionblocks

I want to use TPL DataFlow to execute some complex validation rules and transformations on multiple items with a degree of parallelism greater than 1. In certain occasions, the code needs to access ...
Thanasis Ioannidis's user avatar
1 vote
2 answers
288 views

Conform IAsyncEnumerable to Dataflow ISourceBlock

I have an existing transformation function Func<IAsyncEnumerable<TIn>, IAsyncEnumerable<TOut>> that I want to use inside a Dataflow pipeline. In order to link this transformation ...
Bouke's user avatar
  • 11.6k
0 votes
2 answers
148 views

TPL Dataflow: How to start the next async action when the current one hasn't finished yet, preserving the execution order?

Consider the following program, which uses TPL Dataflow. Hence, ActionBlock comes from the Dataflow library. internal static class Program { public static async Task Main(string[] args) { ...
Tohnmeister's user avatar
1 vote
1 answer
72 views

How to parallelize a list using multiple ActionBlocks?

I have a large folder structure that I'm trying to download from a shared drive. The shared drive is slow, but it also has several mirrors. To speed up the process I'm trying to make a little ...
Trygve's user avatar
  • 591
0 votes
2 answers
71 views

always allocates and returns a new instance of resource class System.Threading.Tasks.Dataflow.Internal.Disposables

alloc_fn: Method LinkTo<X.Offers.Core.DataFlow.RefreshOfferMatching.RefreshUsersOffersContext> always allocates and returns a new instance of the resource class System.Threading.Tasks.Dataflow....
duckGame's user avatar
3 votes
3 answers
342 views

How to properly wait for completion when having forked TPL Dataflow pipeline? [duplicate]

I have a TPL Dataflow pipeline where a target block is linked to two propagating blocks, which are then both linked to a source block. All are linked with PropagateCompletion = true. The first ...
Tohnmeister's user avatar
0 votes
0 answers
174 views

How can I unit test a TPL Dataflow pipeline?

Given the following pipeline class I want to test the actual LinkTo(s) such that if I pass bad input into the DeserializeBlock I can check that DeserializeFailedBlock was called. I'm looking for best ...
csharpguy's user avatar
0 votes
1 answer
31 views

Is MaxDegreeOfParallelism best effort or a hard limit?

Just wondering about the MaxDegreeOfParallelism setting in the dataflow library. I was writing a unit test to ensure that the number of parallel tasks didn't exceed the specified amount, when I found ...
NickL's user avatar
  • 1,871
0 votes
2 answers
65 views

TPL Dataflow block with an inner life

I've been doing a bit of TPL dataflow coding and am quite happy with the basics. The question I have is, how would I go about doing a TPL block that, besides reacting to its queue, also has a life on ...
Volker's user avatar
  • 1,743
2 votes
1 answer
134 views

MapReduce using DataFlow library

I am trying to implement a classic map-reduce problem using System.Threading.Tasks.Dataflow, and although I can get something (sort of) working I'm struggling to see how to generalise this ...
Jamiec's user avatar
  • 133k
1 vote
1 answer
204 views

How to ensure BatchBlock has completed

I'm trying to get an understanding of using BatchBlock as it seems to be what I'll need for a project I'm working on. However, I'm getting hung up on one part in particular, and that's how to ensure ...
Chris's user avatar
  • 13
2 votes
1 answer
214 views

C# TPL Dataflow ReceiveAsync() is not completed but task shows completed

I've created TPL Dataflow pipeline as shown below BufferBlock --> TransformBlock --> TransformBlock--> TransformBlock --> BufferBlock For all the data blocks are initialised with ...
mohammed mazin's user avatar
1 vote
1 answer
57 views

TPL DataFlow is being idle with no reason

Consider the following DataFlow pipeline on a 16 cores machine Extract - TransformBlock. Bound Capacity: 32, DOP: 16 Download - TransformBlock. Bound Capacity: 1,024, DOP: 64 Process - TransformBlock. ...
boger's user avatar
  • 553
0 votes
0 answers
41 views

TPL DataFlow not behaving as expected

Consider the following pipeline: public class Program { private static int firstBlockactualDOP = 0; private static int lastBlockactualDOP = 0; private static ActionBlock<int> ...
boger's user avatar
  • 553
0 votes
1 answer
276 views

TPL: Dispose processed items

In C#, I am using Task Parallel Library (TPL) to download an image, process the image, and save the analysis results. A simplified code reads as the following. var getImage = new TransformBlock<int,...
Dr. Strangelove's user avatar
1 vote
1 answer
211 views

Conditional multiple LinkTo in TPL

I have a buffer block that links to multiple action blocks: bufferBlock.LinkTo(consumerActionBlock1, linkOptions); bufferBlock.LinkTo(consumerActionBlock2, linkOptions); bufferBlock.LinkTo(...
Maxim Zabolotskikh's user avatar
1 vote
1 answer
533 views

TPL Dataflow Transform block post to two action block in parallel

I have a TPL Dataflow, that worked fine using only a transform block, then an action block. I've added a new action block to be executed simultaneously with the existing action block, but my new ...
NHARI Med's user avatar
  • 141
0 votes
0 answers
36 views

Why TransformBlock get different results?

Why TransformBlock get different results each time it executes? Expect: 1,2,3 4,5,6,7,8,9 Result: 4,2,3 1,5,6,7,8,9 1,2,3 4,5,6,7,8,9 4,2,3 1,5,6,7,8,9 1,2,3 4,5,6,7,8,9 1,2,3 4,5,6,7,8,9 1,5,3 4,2,...
zhusp's user avatar
  • 129
2 votes
2 answers
223 views

Report when input to first dataflow block finishes all linked blocks

I am using TPL Dataflow to download data from a ticketing system. The system takes the ticket number as the input, calls an API and receives a nested JSON response with various information. Once ...
Ninja's user avatar
  • 346
1 vote
1 answer
50 views

How to complete correctly a branched pipeline with BatchBlock?

Consider the following pipeline: public static void Main() { var firstBlock = new TransformBlock<string, string>(s => { Console.WriteLine("FirstBlock"); ...
boger's user avatar
  • 553
0 votes
1 answer
38 views

ActionBlock misses actions sometimes

I'm using ActionBlock and I tested if it's working properly like below and sometime Actionblock misses actions and it shouldn't happen at all Why is this happening and how can i fix it? var n = 0; var ...
TYFA's user avatar
  • 313
0 votes
0 answers
243 views

TPL BufferBlock Not Completing When Bounded Capacity is Greater Than 1

I currently have a pipeline setup as such: BufferBlock ==> BatchBlock ==> TransformBlock ==> TransformManyBlock ==> BufferBlock The BoundedCapacity is set to 2,000. However, if I provide ...
Dominick's user avatar
  • 312
2 votes
1 answer
764 views

How to buffer and batch REST get queries with asp.net web API 2 (using TPL/Reactive)?

I am runnning a public front facing api that hardly scale for now. This API is built upon Asp.Net web API 2 (.Net 4.7.2). This The problem that I have is that from this entry point the service have to ...
Dypso's user avatar
  • 563
0 votes
1 answer
140 views

Async thread dies on TPL await

I'm writing a simple producer/consumer application, but I'm noticing a really strange behaviour..This is the code: private Thread _timelineThread = null; private BufferBlock<RtpPacket> _queue = ...
Mario Vaccaro's user avatar
1 vote
1 answer
71 views

Does Complete execute Trigger for DataFlow blocks?

I have two pipelines using multiple Buffer, Batch, and Action blocks from the System.Threading.Tasks.Dataflow namespace. I understand once all work items have been added to the pipeline that the ...
Dominick's user avatar
  • 312
0 votes
0 answers
277 views

TPL DataFlow memory leak

I've developed a multiple data pipelines using TPL DataFlow v5.0 which any pipeline contains TPL blocks. Each pipeline looks like this: BufferBlock - receives udp packets from multiple clients and ...
Chen Ben Hamo's user avatar
0 votes
2 answers
126 views

How to search vast code base for multiple literal strings efficiently?

This question is a follow up on How to optimize performance in a simple TPL DataFlow pipeline? The source code is here - https://github.com/MarkKharitonov/LearningTPLDataFlow Given: Several solutions ...
mark's user avatar
  • 58.2k
1 vote
1 answer
553 views

How to optimize performance in a simple TPL DataFlow pipeline?

Given: Hundreds of .NET projects Thousands of C# files across all the projects A string literal I want to output all the matches of the given literal in all the files across all the projects. I ...
mark's user avatar
  • 58.2k
1 vote
1 answer
649 views

How to do error handling with connected TPL dataflow blocks?

it seems that I do not understand TPL Dataflow error handling. Lets assume I have a list of items I wanna process and I use a ActionBlock for that: var actionBlock = new ActionBlock<int[]>(async ...
SebastianStehle's user avatar
2 votes
1 answer
181 views

How to process thousands of C# files across hundreds of projects efficiently?

Introduction I am solving the following problem: Given a C# type X with one or more publicly visible constants what are all the C# types in the solution that depend on the constants in the type X? ...
mark's user avatar
  • 58.2k

1
2 3 4 5
13