Gluon-ts: Multiprocessing Pipeline

[jaheba]

This is more a report, then it is a proposal.

---

I'm not a fan of the current multiprocessing approach we have in our input-pipeline. I would like to replace it with a more generic approach.

Over the last days, I've did some exploration into the space of using pipelines and multiprocessing.

What I came up with, is something like this:

def add_1(arg):
    return arg + 1

steps = add_1, add_1
pipeline = Pipeline(steps, num_workers=4)
pipeline.start()

pipeline.enqueue(3)
assert pipeline.dequeue() == 5

The idea is simple, we define a pipeline, which consists of steps, which are executed one after another. In this way, it is really similar to our Transformation interface.

To enqueue data, I've used separate "emitter" processes, which read in data:

# split the dataset in half, and use two readers which start at different points in the dataset
line_starts = split_seeks(path, splits=2)
readers = [FileReader(path, pos, num_lines) for pos, num_lines in line_starts]

# creates a process for each reader, which enqueues data into the pipeline
emit(readers, pipeline)

I found that using more than two reader processes did not improve throughput.

Sentinel Values

To indicate the end of the input, I used sentinel values, which are sent instead of actual data.

    def __iter__(self):
        num_sentinels = 0
        while True:
            batch = self.pipeline.dequeue()

            if batch == self.sentinel:
                num_sentinels += 1
                if num_sentinels == len(self.emitter.processes):
                    break
                continue
            else:
                yield batch

That way, we know when we have processed all input data.

Other Findings

Batching

Instead of enqueueing single items, I've used data-batches. This significantly improved performance.

However, larger values (1000) also led to a slow down.

Single-Queue vs Multi-Queue

I tried two approaches:

• Each process fetches a value from the input-queue, runs all steps on that value, and then submits the result to the output-queue:

[Q] -> a, b, c -> [Q]

• Similar to the first approach, but now there are queues between steps, and each process only ever applies one step at once. There is a second task queue, describing what needs to be done:

[Q] -> a -> [Q] -> b -> [Q] -> c [Q]

The idea was, that workers would spend more time on slower steps. For example, if b takes much more time than a and c, a single worker might execute a and c, while the rest executes b.

However, when comparing the throughput of both approaches using our transformation-pipeline as a workload, the first one was much faster.

Shared Memory

Python 3.8 offers multiprocessing.shared_memory. I wanted to see if it makes a difference, if instead of copying numpy arrays between processes, we use shared memory instead.

Although I saw improvements when using really large arrays (1 million entries), throughput was much slower when running on our datasets, which have smaller arrays.

[AaronSpieler]

Great idea!

Comments:

• I like the emitter pattern, seems to make sense in this context
• Interesting that performance did not improve with regards to more workers, that might indicate a significant overhead
• Yeah, [Q] -> a, b, c -> [Q] seems more logical, since it doesn't require copying data to other processes memory
• The shared memory could be beneficial when transforming the dataset to MXNet native arrays, since copying matrices from shared memory into GPU is faster that from non shared memory

Questions:

• Can we assign specific parts of the dataset to specific workers?
• Can a single worker work on different parts of the dataset?
• What are the overall benefits with regards to the current implementation?
• Is this faster than the current implementation?
• How would shuffling on a time series basis look like (i.e. shuffling dataset ids)?

[lostella]

• Can we assign specific parts of the dataset to specific workers?

• Can a single worker work on different parts of the dataset?

My understanding is that the proposal is to completely decouple the file reading from the items processing. @jaheba correct me if I’m wrong: essentially “someone” would go through the dataset and fill in a queue, and then other processes would just consume from that queue regardless of where in the dataset the stuff is from, or who put it in the queue, right?

I quite like the idea (provided I understood correctly).

My question is: would everything work fine if some of the pipeline steps (say, the last one, which in the model transformation chains would be the instance splitter) produces zero or more outputs?

[jaheba]

My understanding is that the proposal is to completely decouple the file reading from the items processing.

Right, I think we want some control over how files are read and make that independent. For example, if we add another data-source which is not easily parallelisable our whole setup would work just as well. And, in my experience, just reading in data is relatively fast.

essentially “someone” would go through the dataset and fill in a queue, and then other processes would just consume from that queue regardless of where in the dataset the stuff is from, or who put it in the queue, right?

Yes. There is no guaranteed order.

My question is: would everything work fine if some of the pipeline steps (say, the last one, which in the model transformation chains would be the instance splitter) produces zero or more outputs?

The pipeline itself just runs its step. The problem I see is that it's hard to know on the other side of the pipeline when all items have been processed. The pipeline itself doesn't know either, since there can always be more items which have not yet been submitted to it.

Another issue is that we don't have a guaranteed order, so even my current sentinel approach might not just work as I want it to be.

[jaheba]

• Interesting that performance did not improve with regards to more workers, that might indicate
  a significant overhead

It could be that we are bound by file IO at some point.

• Yeah, [Q] -> a, b, c -> [Q] seems more logical, since it doesn't require copying data to other processes memory

I'm not sure copying is as much of an issue as synchronisation.

Questions:

• Can we assign specific parts of the dataset to specific workers?
• Can a single worker work on different parts of the dataset?

There is no connection between worker and parts of the dataset.

• What are the overall benefits with regards to the current implementation?

Separation of concerns. Neither the dataset layer nor the pipeline-steps have to know anything about multiprocessing. That makes it easier to add or change things.

• Is this faster than the current implementation?

I think in theory it shouldn't, but there could be some benefit of splitting dataset loading from the pipeline.

• How would shuffling on a time series basis look like (i.e. shuffling dataset ids)?

Depends on the suffling strategy. If we read in batches and then shuffle them, it should "just work".