Does foreach works similar to multi-processing? I thought if I give 5 videos at once to my inference script then it would run the inference over 5 videos simultaneously but when i did it my flow with foreach loop I got one video getting processed at a time. I am doing -

self.next(self.run_model, foreach='local_original_video_filenames')

where

run_model

flow run the inference over the video and

local_original_video_filenames

is a list with local video urls.

What compute layer are you using? You control the number of parallel jobs that are run using

--max-workers

but that is set to

16

by default. So I am guessing its an issue with the compute you are using.

The EC2 instance that I am using has a GPU so i am just using @resources over my run_model step and running it normally without passing any other arguments

So are you using AWS Batch or AWS EKS?

No i am doing it locally over my ec2, so its pretty similar to running it locally but over ec2

Ah so you are running it locally, unless you have multiple-GPUs on the AWS EC2 instance it will only have the ability to run 1 job as parallelising inference(assuming the model fits into a single GPU’s RAM) would require N GPU’s for N parallel jobs.

okayy but when i was doing it over local I wrote a script using gpt and i was able to run multiple process over my GPU, >10 process at once.

That is because you are performing batch inference with a

batch_size=1

on a GPU. Remove the use of the

foreach

and load the data with a bigger batch_size.

sorry i didn't understood the loading of data part with a batch size. For more context, my overall flow is pretty simple, download the video from s3 run the inference over these videos. I wanted to run the inference over the videos simultaneously cause locally I have tested with above script and I was able to do 10-15 videos at once. I want to replicate the same here. If possible can you give me the references which can help or give an example of it

You just need to process all the videos in a single step instead of splitting them up in a foreach:

from metaflow import FlowSpec, step


class SampleFlow(FlowSpec):
    @step
    def start(self):
        self.videos = ...
        self.next(self.a)

    @step
    def process_videos(self):
        process_videos(self.videos)
        self.next(self.end)

    @step
    def end(self):
        ...


if __name__ == "__main__":
    SampleFlow()

When you do not have enough workers to perform a foreach you are essentially doing a for loop across each video and processing them sequentially:

[process(video) for each video in videos]

okay, will try. Thank you!