I am a newbie on TPL Dataflow
. I have a list of project numbers that I need to process. A project could have about 8000
items and I need to get the data for each item in the project and then push this data into 5 separate servers.
Here is what I have coded thus far. I'm stuck at the step of how to load this data into the 5 servers. I am not sure if this is coded correctly. Any advice is much appreciated.
public static bool PushData(string projectId)
{
var linkCompletion = new DataflowLinkOptions
{
PropagateCompletion = true
};
var projectItems = new TransformBlock<ProjectDTO, ProjectDTO>(
dto => dto.GetItemData(dto), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5 });
var itemData = new ActionBlock<ProjectDTO>(
dto => PostEachServerAsync(dto, "server1", "setmemcache"), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5 });
projectItems.LinkTo(projectRules, linkCompletion);
IList<ProjectDTO> dtoList = new List<ProjectDTO>();
dtoList = MemcachedDTO.GetDataByProject(projectId);
foreach (ProjectDTOd in dtoList)
{
projectItems.Post(d);
}
projectItems.Complete();
projectItems.Completion.Wait();
return false;
}
Here is my code now - but it does not complete properly - could anyone please tell me what I am doing wrong?
[HttpGet]
public HttpResponseMessage ReloadItem(string projectQuery)
{
try
{
var linkCompletion = new DataflowLinkOptions
{
PropagateCompletion = true
};
IList<string> projectIds = projectQuery.Split(',').ToList();
IEnumerable<string> serverList = ConfigurationManager.AppSettings["ServerList"].Split(',').Cast<string>();
var iR = new TransformBlock<MemcachedDTO, MemcachedDTO>(
dto => dto.GetIR(dto), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 3 });
var pR = serverList.Select(
i => new { Id = i, Action = new ActionBlock<MemcachedDTO>(dto => PostEachServerAsync(dto, i, "set"), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 3 }) });
List<MemcachedDTO> dtoList = new List<MemcachedDTO>();
foreach (string pid in projectIds)
{
IList<MemcachedDTO> dtoTemp = new List<MemcachedDTO>();
dtoTemp = MemcachedDTO.GetItemIdsByProject(pid);
dtoList.AddRange(dtoTemp);
}
foreach (var action in pR)
{
iR.LinkTo(action.Action, linkCompletion);
}
foreach (MemcachedDTO d in dtoList)
{
iR.Post(d);
}
iR.Complete();
foreach (var action in pR)
{
action.Action.Completion.Wait();
}
return Request.CreateResponse(HttpStatusCode.OK, new { message = projectIds.ToString() + " reload success" });
}
catch (Exception ex)
{
return Request.CreateResponse(HttpStatusCode.InternalServerError, new { message = ex.Message.ToString() });
}
}
Your code doesn't compile at all, how do you run it?
First of all, do not block your thread with
.Wait()
, useasync/await
pattern here. Second, you need aBroadcastBlock
to notify more than 1 block with your data. Third, you need 5 differentActionBlock
s, not 1 with degree of parallelism 5. Fourth, you're awaiting wrongCompletion
task - wait for the last block completion, not the first one, so in your case you need to wait for 5 block completions withWhenAll
method.So your code could be like this (I assume that
projectRules
anditemsData
is the same block):