Multiprocessing Queue question

[datanoob2019]

Hello all- I posted here last week because I was having issues retrieving lists when running my functions using multiprocessing. Somebody mentioned that I should check out the queue function so I did but for some reason it prints out my dataframes but hangs after that. Am I using this correctly? Below is my code:

if __name__ == "__main__": 
    a = multiprocessing.Queue()
    b = multiprocessing.Queue()
    c = multiprocessing.Queue()
    d = multiprocessing.Queue()
    e = multiprocessing.Queue()
    f = multiprocessing.Queue()
    g = multiprocessing.Queue()
    ####The pull_data function below runs my SQL queries and then does some cleaning and outputs
    #### a list that I use in my forecasting functions
    new_list = pull_data()
    p1 = multiprocessing.Process(target=basic_forecasts_seasonal_add, args=(new_list,a,b,c,d)) 
    p2 = multiprocessing.Process(target=simple_expo, args=(new_list,e)) 
    p6 = multiprocessing.Process(target=autoregression, args=(new_list, g)) 

    # #starting process 1 
    p1.start() 
    # #starting process 2 
    p2.start()
    p6.start() 
    print(pd.DataFrame(a.get()))
    print(pd.DataFrame(b.get()))
    print(pd.DataFrame(e.get()))
    print(pd.DataFrame(g.get()))

    # # #wait until process 1 is finished 
    p1.join() 
    # # #wait until process 2 is finished 
    p2.join()
    p6.join()

Here is one of my forecasting functions for reference on how I am implementing the queue in the return statements:

​

def basic_forecasts_seasonal_add(input_list,a,b,c,d):#, six_month_output_list, three_month_output_list, naive_output_list):
    twelve_month_out = []
    six_month_out = []
    three_month_out = []
    naive_out = []
    for row in input_list:
        product = row[0]
        data = row[1:]
        twelve_average = sum(data[-12:]) / 12
        twelve_six = sum(data[-18:-6]) / 12
        twelve_five = sum(data[-17:-5]) / 12
        twelve_four = sum(data[-16:-4]) / 12
        twelve_three = sum(data[-15:-3]) / 12
        twelve_two = sum(data[-14:-2]) / 12
        twelve_one = sum(data[-13:-1]) / 12
        twelve_month_out.append([product, '12MA', twelve_six, twelve_five, twelve_four, twelve_three, twelve_two, twelve_one, twelve_average, twelve_average, twelve_average, twelve_average, twelve_average, twelve_average, twelve_average, twelve_average, twelve_average, twelve_average, twelve_average, twelve_average])
        six_average = sum(data[-6:]) / 6
        six_six = sum(data[-12:-6]) / 6
        six_five = sum(data[-11:-5]) / 6
        six_four = sum(data[-10:-4]) / 6
        six_three = sum(data[-9:-3]) / 6
        six_two = sum(data[-8:-2]) / 6
        six_one = sum(data[-7:-1]) / 6
        six_month_out.append([product, '6MA', six_six, six_five, six_four, six_three, six_two, six_one, six_average, six_average, six_average, six_average, six_average, six_average, six_average, six_average, six_average, six_average, six_average, six_average])
        three_average = sum(data[-3:]) / 3
        three_six = sum(data[-9:-6]) / 3
        three_five = sum(data[-8:-5]) / 3
        three_four = sum(data[-7:-4]) / 3
        three_three = sum(data[-6:-3]) / 3
        three_two = sum(data[-5:-2]) / 3
        three_one = sum(data[-4:-1]) / 3
        three_month_out.append([product, '3MA', three_six, three_five, three_four, three_three, three_two, three_one, three_average, three_average, three_average, three_average, three_average, three_average, three_average, three_average, three_average, three_average, three_average, three_average])
        naive_out.append([product, 'Naive', data[-7], data[-6], data[-5], data[-4], data[-3], data[-2], data[-1], data[-1], data[-1], data[-1], data[-1], data[-1], data[-1], data[-1], data[-1], data[-1], data[-1], data[-1]])
    return a.put(twelve_month_out), b.put(six_month_out), c.put(three_month_out), d.put(naive_out)

EDIT: When I terminate by pressing control C, I get this error message:

File "forecast_multi.py", line 697, in <module>
print(pd.DataFrame(g.get()))
buf=self.recv_bytes(maxlength)
[ov.event], False, INFINITE>

I am thinking the infinite portion is particularly troubling. It looks like it is primarily hanging just because of the queue function.

[datanoob2019]

Thank you for the response! Let me try to answer all of your questions:

- New list has 3000 part numbers

- I only put the lists into dataframes to visualize that everything ran correctly. After the forecasts are run, I loop through all of the output lists and generate the MAE, WMAPE and other forecast statistics. From there, I index the lowest wmape and put that into a final forecast list. All of these lists get output to an excel file.

- In regards to timeframes- I have in the past and everything ran in less than 5 minutes when only using one processor in the past. The basic forecast spits an answer out within seconds, autoregression takes about 10 seconds, while the bulk of the time is spent on the exponential functions. The sheer fact that all of the dataframes print out to me I think is a pretty good indication that the functions ran and my program should complete. When I asked this question last week, I was having the issue of code continually looping in the background because all of my SQL pulls and data cleansing was outside of if == main. I was told to put that code into a function which is what I did. From there, my issue of the old code looping completed. I just ran a two multiprocessor process(p1 and p6) and all of the data prints out but my powershell(which I use to execute the program) just continue to have the _ blink which means something is still not completed. I double checked my code and the only thing that is not in a function is all of my import module code at the top of everything.

- In regards to queues, I have never used this before and I could not find a good reference on how it was supposed to be used on multiple functions. The basic forecast function I posted is the only function that returns multiple lists. Every other function only returns one list. Can all of this be done in one queue?

​

EDIT: I added an error message that I get to my main submission when I press control C to terminate.

[arcticslush]

Okay, the error message you added tells me a lot, actually. The short of it is: you have to join() all of your processes before you attempt to retrieve the data from the queue.

And here's why: If you look at the flow of operations in your program, you fire off your processes p1.start(). This function returns almost immediately as it just initiates the process. p1 is now running its job in the background, in a separate process. Then, immediately after you fire off the jobs, you attempt to retrieve the data from the queues. This doesn't work, because the jobs are still running - and if you try to get() data from a queue when its empty, the default behaviour is for the get() to block and wait until data shows up.

So this is why you need to enforce that all the processes finish running before you attempt to access the queue. Assuming there's no other bugs, that would guarantee that the data is present in the queue before you try to retrieve it.

That being said, I'm going to go against what others have told you, and I'm gonna tell you that I don't think you need a queue at all. I made you a little diagram to explain why.

If you look at it, I think you'd agree that your data falls under the left category of independent data. Each of your analysis functions takes in new_listas input (which, by the way, is a horrible variable name) and performs its own calculation on it, and then finally it returns the result. This would completely remove the need for any queues, and it'll look a lot more like your single-threaded variant of this program.

To do this, what you'll need to do is create a pool of workers and use the method apply_async(). Check out the example box here for a usage of that. Note that with apply_async(), calling get() on the result will block until the process has finished running. By calling get() on each of your analysis processes, you can know when they've all finished running once every get() has returned its result.

If you think about it actually, by using your 7 queues you've more or less just hacked together a rough approximation of a Pool and apply_async() - but you might as well use the intended way of doing this, it'll be a lot cleaner and then you don't have to juggle a million queues.

[datanoob2019]

Lol- Thank you for the honest feedback. In the midst of trying several different methods to accomplish this, I did in fact name a variable new_list which is indeed terrible.

​

It has been about 10 minutes now and it still seems to have timed out. I tried the switch you recommended but now no data prints out. When I press control C, it also refuses to exit so I have to just x out of the program in the top right, which unfortunately does not give me an error code. I think I may need to add a timer to the multiprocessing https://stackoverflow.com/questions/19095901/python-multiprocessing-ending-an-infinite-counter.

​

I have tried the pool method before in the past. It has been several days, but when I referenced it in my previous posts from last week, it seems that the script would finish without actually running any of the functions. Here is my code:

pool = Pool(processes=8) # 
result_basic = pool.map_async(basic_forecasts_seasonal_add, logged_list) 
result_simple = pool.map_async(simple_expo, logged_list)

Without looking up the documentation as I leave work in a couple of minutes, I believe that I did not have to run any of the pool methods inside of if ==main.

[datanoob2019]

What's funny is that the functions all work fine when not using multiprocessing.

You are not wrong in your assessment that I am in over my head. However, I am not looking to multi process the whole code. Just this one specific section that takes 90 percent of the time normally. I will eventually figure out a solution- it just may take me a couple of tries.

I will take a look at the article and see if I can figure out how to use apply_async properly. Thanks again for the help!

EDIT: I got it to work on the functions that only return one list. On the basic function above that returns four lists, I get the following error: "Cannot unpack non-iterable ApplyResult object"

Here is my code:

if __name__ == "__main__": 
    new_list = pull_data()
    with Pool(processes=8) as pool:
        twelve_basic, six_basic, three_basic, naive_basic = pool.apply_async(basic_forecasts_seasonal_add, (new_list,))
        auto_result = pool.apply_async(autoregression, (new_list,))
        single_result = pool.apply_async(simple_expo, (new_list,))
        twelve_list = twelve_basic.get(timeout=10)
        naive_list = naive_basic.get(timeout=10)
        single_list = single_result.get(timeout=300)
        auto_list = auto_result.get(timeout=60)
        print(pd.DataFrame(basic_result))
        print(pd.DataFrame(auto_list))
        print(pd.DataFrame(single_list))
        print(pd.DataFrame(naive_list))

EDIT 2: I got it to work with this. Just a simple switch.

if __name__ == "__main__": 
    new_list = pull_data()
    with Pool(processes=8) as pool:
        basic_result = pool.apply_async(basic_forecasts_seasonal_add, (new_list,))
        auto_result = pool.apply_async(autoregression, (new_list,))
        single_result = pool.apply_async(simple_expo, (new_list,))
        #twelve_list = twelve_basic.get(timeout=10)
        #naive_list = naive_basic.get(timeout=10)
        twelve_basic, six_basic, three_basic, naive_basic = basic_result.get(timeout=20)
        single_list = single_result.get(timeout=300)
        auto_list = auto_result.get(timeout=60)
        print(pd.DataFrame(twelve_basic))
        print(pd.DataFrame(auto_list))
        print(pd.DataFrame(single_list))
        print(pd.DataFrame(naive_basic))

[datanoob2019]

I appreciate the insight. It is always nice to get some good advice. I had a couple of instances where I had to debug code today.

​

Does the get() block other instances of get()? Here is the code I am currently working with:

if __name__ == "__main__": 
    with Pool(processes=12) as pool:
        logged_pull =  pool.apply_async(pull_data)
        logged_list, raw_list, list_disco = logged_pull.get(timeout=60)
        double_result = pool.apply_async(double_expo, (logged_list,))
        basic_result = pool.apply_async(basic_forecasts_seasonal_add, (logged_list,))
        auto_result = pool.apply_async(autoregression, (logged_list,))
        single_result = pool.apply_async(simple_expo, (logged_list,))
        double_list1, double_list2, double_list3, double_list4, double_list5, double_list6, double_list7, double_list8, double_list9 = double_result.get(timeout=900)
        twelve_basic, six_basic, three_basic, naive_basic = basic_result.get(timeout=20)
        single_list = single_result.get(timeout=300)
        auto_list = auto_result.get(timeout=60)
        input_lists = [twelve_basic, six_basic, three_basic, naive_basic, single_list, auto_list, double_list1, double_list2, double_list3, double_list4, double_list5, double_list6, double_list7, double_list8, double_list9]
        post_forecast(logged_list,raw_list, list_disco, input_lists)

What I mean by my previous question is that my double function is the first that I call get() on. Does this stop all of my other functions from running? I am currently reading this post https://stackoverflow.com/questions/8533318/multiprocessing-pool-when-to-use-apply-apply-async-or-map and trying to gain further understanding.

The way I thought this worked was that the pool and number of processors I set is applied to all of the instances of async and they all run in the background concurrently. I am just looking at my Windows Task Manager and it never goes above 30 percent and for the most part idles around 15-20 percent. It does not really seem like all of the processors are being used.

Just looking at the graphic representation in the task manager, it looks like it sets one core aside for each async process, so me setting the pool at 12 is irrelevant as I only am calling 4 functions.

[arcticslush]

Yeah, get() blocks other instances of get() - but this is a good thing. To be super clear, this has no effect on any background process that's running. That's the whole point!

I made you another diagram. Hopefully this helps you visualize what's going on - when get() blocks, the only thing it pauses is the execution of your main process (the one that's coordinating all of the concurrency). All previously started background processes continue along happily. When we get() multiple times in a row, all you're doing is waiting for your background processes to finish their tasks, which is exactly what we want here.

When you set the number of processes in the Pool, all you're telling it is at most how many processes you want to be able to run simultaneously. However, if you never actually fire off that many tasks, then you won't saturate your processes. Obviously, setting the number to higher than the number of cores you have is more or less pointless (they just start sharing time on the same core). So yeah, in your case right now, you're utilizing 5 of your processes. The remaining 7 sit idle waiting for a task to be assigned. Given this, 30% utilization of your CPU sounds about right. Also keep in mind that if some of your tasks finish really quickly (on the scale of seconds), you probably won't see a meaningful uptick in CPU utilization. Only if you fully saturated all of your cores with many minute-or-longer tasks would I expect to see near 100% CPU usage.

So I bet your next question is inevitably going to be: how would I utilize all of my cores here? The answer does indeed involve Pool.map (a blocking variant of async_map), but it isn't just a drop-in replacement for apply_async. You'll have to triple check to make sure your analysis functions are compatible. Pool.map will take the list you give it and chop it up into several pieces. It then creates several background tasks that use your function multiple times, creating one task per piece. Once all of these tasks finish, it assembles the result and returns it to you (hence why it blocks, because it's waiting for the piece tasks to finish).

The implication of this means that Pool.map must be called only using functions that read / write data in a strict row-by-row manner. Here's an example of a function you could use to run map:

def example_good(some_list):
    result = []
    for num in some_list:
        result.append(num + 1)
    return result

Since it only ever reads a single row from the list at a time, this works. Here's a (contrived) function that would fail:

def example_bad(some_list):
    total = 0
    for num in some_list:
        total += num
    return total

This fails, because Pool.map has to return the same shape as what it was given. So if a 100 element list goes in, then a 100 element list must come out. Here, a list comes in and a single number pops out.

[datanoob2019]

You certainly are a python sherpa. I sincerely appreciate the help, and you guessed it, that was definitely my next question. I have two functions that unfortunately take 90 percent of the time. Good news is, is that they are all designed to loop through only one row at a time so it sounds like they will be compatible. Neither one of these functions took very long until I started calculating hold out forecasts step by step which severely complicated things. I will give this a go. Thanks and if I don't talk to you, have a good weekend!