multiprocessing pool: determine process names (unable to terminate its processes)

Question

I have some code in which I attempt to create 4 processes within a Pool.

Once I get any exception (eg the database it is trying to connect to is down), I want to kill the pool, sleep for 10secs and then create a new pool with 4 processes.

However it seems that the Pool is never killed because the processes names keep getting incremented each time. Does the pool have a cache where it keeps name count?

def connect_db() 
  pass


while True: 
 p = Pool(4)
 for process in multiprocessing.active_children():
  print(process.name) #why is the name incremented by 1 each time while loop iterates? 
 try:
  r = p.map(connect_db, ())
 except Exception as e:
  pool.close()
  pool.join()
  time.sleep(10)

First four processes are SpawnPoolWorker-1 to 4 and next 4 are SpawnPoolWorker-5 to 8. How does it know that I have already created 4 processes before? I am creating a new instance of Pool each time or am I doing something wrong?

Answer 1

The main reason you were not seeing what you expect is the following line of code:

r = p.map(connect_db, ())

You're calling multiprocess.map with an empty iterable, so connect_db isn't being called at all, and you don't reach the except part of the code, not closing the pool, etc.

Here's a skeleton that works, with a bunch of print statements for debugging. I'm attaching the output below, and as you can see, there are exactly four child process at every round.

import multiprocessing
import time 
import random

def connect_db(i):
    print(f"Trying to connect {i}")
    time.sleep(random.random() * 2)
    raise Exception("Failed to connect")

while True: 
    p = multiprocessing.Pool(4)
    print("active children are:")
    for idx, process in enumerate(multiprocessing.active_children()):
        print(f"Child number {idx} is {process.name}") #why is the name incremented by 1 each time while loop iterates? 
    try:
        print("About to create a pool")
        r = p.map(connect_db, range(4))
        print("Created a pool")
    except Exception as e:
        print(e)
        print("terminating threads")

        p.terminate()
    p.close()
    p.join()
    time.sleep(5)

Output:

active children are:
Child number 0 is ForkPoolWorker-2
Child number 1 is ForkPoolWorker-1
Child number 2 is ForkPoolWorker-4
Child number 3 is ForkPoolWorker-3
About to create a pool
Trying to connect 0
Trying to connect 1
Trying to connect 2
Trying to connect 3
Failed to connect
terminating threads
active children are:
Child number 0 is ForkPoolWorker-5
Child number 1 is ForkPoolWorker-6
Child number 2 is ForkPoolWorker-8
Child number 3 is ForkPoolWorker-7
About to create a pool
Trying to connect 0
Trying to connect 1
...

One last note - if the use-case is indeed database connections, there are readymade connection pools, and you should probably use one of them. Plus, I'm not sure one can share database connections across processes.

Controling process names in a pool

If, for some reason, you'd like to control the process names in a pool, you can do that by creating your own pool context:

import multiprocessing
from multiprocessing import context
import time 
import random

process_counter = 0

class MyForkProcess(multiprocessing.context.ForkProcess):
    def __init__(self, *args, **kwargs):
        global process_counter
        name = f"MyForkProcess-{process_counter}"
        process_counter += 1
        super(MyForkProcess, self).__init__(*args, name = name, **kwargs)

class MyContext(multiprocessing.context.ForkContext):
    _name = 'MyForkContext'
    Process = MyForkProcess 

def connect_db(i):
    print(f"Trying to connect {i}")
    cp = multiprocessing.current_process()
    print(f"The name of the child process is {cp.name}")
    time.sleep(random.random() * 2)
    raise Exception("Failed to connect")

context = MyContext()
while True: 
    p = context.Pool(4)
    print("active children are:")
    for idx, process in enumerate(multiprocessing.active_children()):
        print(f"Child number {idx} is {process.name}") #why is the name incremented by 1 each time while loop iterates? 
    try:
        print("About to create a pool")
        r = p.map(connect_db, range(4))
        print("Created a pool")
    except Exception as e:
        print(e)
        print("terminating threads")

        p.terminate()
        process_counter = 0

    p.close()
    p.join()
    time.sleep(5)

The output now is:

active children are:
Child number 0 is MyForkPoolWorker-2
Child number 1 is MyForkPoolWorker-0
Child number 2 is MyForkPoolWorker-3
Child number 3 is MyForkPoolWorker-1
About to create a pool
Trying to connect 0
The name of the child process is MyForkPoolWorker-0
Trying to connect 1
The name of the child process is MyForkPoolWorker-1
Trying to connect 2
The name of the child process is MyForkPoolWorker-2
Trying to connect 3
The name of the child process is MyForkPoolWorker-3
Failed to connect
terminating threads
active children are:
Child number 0 is MyForkPoolWorker-2
Child number 1 is MyForkPoolWorker-0
Child number 2 is MyForkPoolWorker-1
Child number 3 is MyForkPoolWorker-3
About to create a pool
...