multiprocessing.pool.map and function with two arguments

Question

I am using multiprocessing.Pool()

here is what i want to Pool:

def insert_and_process(file_to_process,db):
    db = DAL("path_to_mysql" + db)
    #Table Definations
    db.table.insert(**parse_file(file_to_process))
    return True

if __name__=="__main__":
    file_list=os.listdir(".")
    P = Pool(processes=4)
    P.map(insert_and_process,file_list,db) # here having problem.

I want to pass 2 arguments What i want to do is to initialize only 4 DB connections (here will try to create connection on every function call so possibly millions of them and cause IO Freezed to death) . if i can create 4 db connections and 1 for each processes it will be ok.

Is there any solution for Pool ? or should i abandon it ?

EDIT:

From help of both of you i got this by doing this:

args=zip(f,cycle(dbs))
Out[-]: 
[('f1', 'db1'),
 ('f2', 'db2'),
 ('f3', 'db3'),
 ('f4', 'db4'),
 ('f5', 'db1'),
 ('f6', 'db2'),
 ('f7', 'db3'),
 ('f8', 'db4'),
 ('f9', 'db1'),
 ('f10', 'db2'),
 ('f11', 'db3'),
 ('f12', 'db4')]

So here it how it gonna work , i gonna move DB connection code out to the main level and do this:

def process_and_insert(args):

    #Table Definations
    args[1].table.insert(**parse_file(args[0]))
    return True

if __name__=="__main__":
    file_list=os.listdir(".")
    P = Pool(processes=4)

    dbs = [DAL("path_to_mysql/database") for i in range(0,3)]
    args=zip(file_list,cycle(dbs))
    P.map(insert_and_process,args) # here having problem.

Yeah , i going to test it out and let you guys know.

score 29 · Accepted Answer · edited Sep 08 '14 at 17:57

29

The Pool documentation does not say of a way of passing more than one parameter to the target function - I've tried just passing a sequence, but does not get unfolded (one item of the sequence for each parameter).

However, you can write your target function to expect the first (and only) parameter to be a tuple, in which each element is one of the parameters you are expecting:

from itertools import repeat

def insert_and_process((file_to_process,db)):
    db = DAL("path_to_mysql" + db)
    #Table Definations
    db.table.insert(**parse_file(file_to_process))
    return True

if __name__=="__main__":
    file_list=os.listdir(".")
    P = Pool(processes=4)
    P.map(insert_and_process,zip(file_list,repeat(db)))

(note the extra parentheses in the definition of insert_and_process - python treat that as a single parameter that should be a 2-item sequence. The first element of the sequence is attributed to the first variable, and the other to the second)

edited Sep 08 '14 at 17:57

crizCraig

8,487
6
54
53

answered Dec 15 '11 at 16:57

jsbueno

99,910
10
151
209

4

Note that the syntax `def f((arg1, arg2)):` is gone in Python 3. – Ferdinand Beyer Dec 15 '11 at 17:01
1

@FerdinandBeyer: I had forgotten that. Well, unless the implementation of multiprocessing.Pool.map is different there, the way to go would be to assign to a single argument, and unpack it inside the function then. – jsbueno Dec 15 '11 at 22:52
Thanks i got it working! i got it by doing zip(file_list,cycle(dbs)). But i do not use f((arg1, arg2)). as i used more of your code i selected you! – Phyo Arkar Lwin Dec 16 '11 at 10:23

Ferdinand Beyer · Answer 2 · 2011-12-15T17:02:49.950

Your pool will spawn four processes, each run by it's own instance of the Python interpreter. You can use a global variable to hold your database connection object, so that exactly one connection is created per process:

global_db = None

def insert_and_process(file_to_process, db):
    global global_db
    if global_db is None:
        # If this is the first time this function is called within this
        # process, create a new connection.  Otherwise, the global variable
        # already holds a connection established by a former call.
        global_db = DAL("path_to_mysql" + db)
    global_db.table.insert(**parse_file(file_to_process))
    return True

Since Pool.map() and friends only support one-argument worker functions, you need to create a wrapper that forwards the work:

def insert_and_process_helper(args):
    return insert_and_process(*args)

if __name__ == "__main__":
    file_list=os.listdir(".")
    db = "wherever you get your db"
    # Create argument tuples for each function call:
    jobs = [(file, db) for file in file_list]
    P = Pool(processes=4)
    P.map(insert_and_process_helper, jobs)

Thanks Ferdinand , this is close to what i want. What i want to do is to create 4 DB Connections . One connection for each processes, but not for every function calls . `DAL("Path To db")` will create a db connection. Single connection would be slower than Quad connections at once. — Phyo Arkar Lwin, Dec 16 '11 at 09:30
I've tried those examples and it worked fine when the function doesn't have to return...; Can't we do something like my_var=P.map(insert_and_process_helper, jobs)? — neverMind, Aug 07 '13 at 05:27

score 5 · Answer 3 · answered Sep 09 '12 at 16:10

No need to use zip. If for example you have 2 parameters, x and y, and each of them can get several values, like:

X=range(1,6)
Y=range(10)

The function should get only one parameter, and unpack it inside:

def func(params):
    (x,y)=params
    ...

And you call it like that:

params = [(x,y) for x in X for y in Y]
pool.map(func, params)

Mehmet nuri · Answer 4 · 2016-06-24T09:58:07.030

3

You can use

from functools import partial

library for this purpose

like

func = partial(rdc, lat, lng)
r = pool.map(func, range(8))

and

def rdc(lat,lng,x):
    pass

edited Jun 24 '16 at 09:58

answered Jun 13 '16 at 08:48

Mehmet nuri

840
7
7

score 2 · Answer 5 · edited Sep 19 '12 at 14:47

2

Using

params=[(x,y) for x in X for y in Y]

you create a full copy of x and y, and that may be slower than using

from itertools import repeat
P.map(insert_and_process,zip(file_list,repeat(db)))

edited Sep 19 '12 at 14:47

dario_ramos

7,118
9
61
108

answered Sep 19 '12 at 09:21

user1682580

21
2

multiprocessing.pool.map and function with two arguments

5 Answers5

Linked