5
$\begingroup$

Assume I have three tasks that must be performed in order:

Do[
    productA = TaskA[taskId];   (*consumes disk I/O*)
    productB = TaskB[productA]; (*consumes CPU*)
    TaskC[productB],            (*consumes network bandwidth*)
{taskId,10000}]

The three tasks above involve usage of different resources. For example, TaskA imports an image file from disk. TaskB performs some data augmentation via image processing. TaskC sends augmented data to another node via WSTP on TCP.

I would like to improve performance by making a asynchronous pipeline. That is, starting three threads for each task. Each task puts/retrieves intermediate results to/from a queue with fixed size.

(*hypothetical code*)
queueAB = MakeQueue[16];
queueBC = MakeQueue[16];
taskA = MakeThread[
    Do[
        With[{productA=TaskA[taskId]},EnQueue[queueAB, productA]];
        EnQueue[queueAB, $TaskEnded],
{taskId, 10000}]
];
taskB = MakeThread[
Module[{productA},
    While[True,
        productA = DeQueue[queueAB];
        If[productA === $TaskEnded,
                EnQueue[queueBC, $TaskEnded];
            Break[],
            EnQueue[queueBC, TaskB[productA]]
        ]
    ]
]
];
taskC = MakeThread[
Module[{productB},
    While[True,
        productB = DeQueue[queueBC];
        If[productB === $TaskEnded,
                Break[],
                TaskC[productB]
            ]
        ]
    ]
];
ParallelRunThread[{taskA, taskB, taskC}]

How to implement the above pipeline?

$\endgroup$
1

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.