return logs from map_async multiprocessing

Gin.Freecs 0
Viewed 32

I am trying to test a neo4j graph and this the simplified case where everything goes well and returns a list of runtimes. However my real purpose is to actually return a dict with some logs like PID , date , query response ... like if Ic could append q_resp = {"qid":q,"resp":qtest ,"avg_runtime":averageruntime ,"Date":...}... but it seems like I cannot return other than single valued function and I tried to create global variable to share the appended logs but in vain . any not too complex idea to get around this ?

by the way only map_async from pool multiprocessing worked for me , map did not do it nor other libs like asyncio maybe because I am working from azure databricks (but I could be totally wrong).

import time
from multiprocessing import Pool

qlist = ["Match (n:company) RETURN n.country_code LIMIT 100","Match (n:company) RETURN n.country_code LIMIT 100","Match (n:company) RETURN n.country_code LIMIT 100","Match (n:company) RETURN n.country_code LIMIT 100","Match (n:company) RETURN n.country_code LIMIT 100","Match (n:company) RETURN n.country_code LIMIT 100", "Match (n:company) RETURN n.country_code LIMIT 100" , "Match (n:person) RETURN n LIMIT 9300" ,"Match (n:company) RETURN n LIMIT 2" ,"Match (n:compound) RETURN n LIMIT 1"]

conn = new_connex(0) #instatiate neo4j connector
def singleq_test(q: str ):
    ar = []
    ntest = 5
    #time.sleep(ti)
    print("inside singleq")
    for n in range(ntest):
        starttime = time.time()
        qtest = conn.query(q) #getting response 
        runtime = time.time()-starttime
        ar.append(runtime)
    #
    averageruntime = sum(ar)/ntest
    return averageruntime

def qlist_test(ql: list ):
    
    with Pool() as po:
        print("inside pool")
        res = po.map_async(singleq_test,ql)
        qltimes = res.get(timeout=20)
        #print(qltimes)

    return qltimes
qlist_test(qlist)
neo4j multiprocessing logging azure-databricks python
2 Answers

You have two choices.

Simplest is that you can return pretty much anything from simpleq_test. Rather than having it return just a single value, it can return a result and values to put into the dictionary. You would then have:

for qltime, dictionary_stuff in res.get(timeout=20):
    ... append qltime to qltimes...
    do something with the dictionary stuff

An alternative is to create a multiprocessing queue. Your threads can add information to the queue, and your main thread can read items from the thread and handle them however it wants.

changes made to a variable in one process do not affect its value in another process.

Frank Yellin0

This is how I would code the program to:

  1. have the worker function, singleq_test, return both the average running time and, for example, the results of the query
  2. show how to assign to each pool process its own connection in case a single connection cannot be used across processes
  3. make the program runnable on platforms that use either the spawn method (e.g. Windows) or fork method (e.g. Linux) for starting new child processes
import time
from multiprocessing import Pool, TimeoutError

def init_pool_processes():
    global conn
    conn = new_connex(0) #instatiate neo4j connector

def singleq_test(q: str):
    ar = []
    ntest = 5
    #time.sleep(ti)
    print("inside singleq")
    for n in range(ntest):
        starttime = time.time()
        qtest = conn.query(q) #getting response 
        runtime = time.time()-starttime
        ar.append(runtime)
    #
    averageruntime = sum(ar)/ntest
    # return average runtime and query results as a tuple:
    return (averageruntime, qtest)

def qlist_test(ql: list):
    
    with Pool(initializer=init_pool_processes) as po:
        print("inside pool")
        res = po.map_async(singleq_test, ql)
        try:
            # Get list of tuples:
            tpls = res.get(timeout=20)
            for tpl in tpls:
                # Unpack each tuple:
                averageruntime, qtest = tuple
                print(averageruntime, qtest)
        except TimeoutError:
            print('map_async timed out')
            qltimes = []
            
    return qltimes

if __name__ == '__main__':
    qlist = [
        "Match (n:company) RETURN n.country_code LIMIT 100",
        "Match (n:company) RETURN n.country_code LIMIT 100",
        "Match (n:company) RETURN n.country_code LIMIT 100",
        "Match (n:company) RETURN n.country_code LIMIT 100",
        "Match (n:company) RETURN n.country_code LIMIT 100",
        "Match (n:company) RETURN n.country_code LIMIT 100",
        "Match (n:company) RETURN n.country_code LIMIT 100",
        "Match (n:person) RETURN n LIMIT 9300",
        "Match (n:company) RETURN n LIMIT 2" ,
        "Match (n:compound) RETURN n LIMIT 1"
        ]
    qlist_test(qlist)

If a single connection can be shared among all processes, then:

import time
from multiprocessing import Pool, TimeoutError

def init_pool_processes(the_connection):
    global conn
    conn = the_connection

def singleq_test(q: str):
    ar = []
    ntest = 5
    #time.sleep(ti)
    print("inside singleq")
    for n in range(ntest):
        starttime = time.time()
        qtest = conn.query(q) #getting response 
        runtime = time.time()-starttime
        ar.append(runtime)
    #
    averageruntime = sum(ar)/ntest
    # return average runtime and query results as a tuple:
    return (averageruntime, qtest)

def qlist_test(ql: list):
    
    conn = new_connex(0) #instatiate neo4j connector
    with Pool(initializer=init_pool_processes, initargs=(conn,)) as po:
        print("inside pool")
        res = po.map_async(singleq_test, ql)
        try:
            # Get list of tuples:
            tpls = res.get(timeout=20)
            for tpl in tpls:
                # Unpack each tuple:
                averageruntime, qtest = tuple
                print(averageruntime, qtest)
        except TimeoutError:
            print('map_async timed out')
            qltimes = []
            
    return qltimes

if __name__ == '__main__':
    qlist = [
        "Match (n:company) RETURN n.country_code LIMIT 100",
        "Match (n:company) RETURN n.country_code LIMIT 100",
        "Match (n:company) RETURN n.country_code LIMIT 100",
        "Match (n:company) RETURN n.country_code LIMIT 100",
        "Match (n:company) RETURN n.country_code LIMIT 100",
        "Match (n:company) RETURN n.country_code LIMIT 100",
        "Match (n:company) RETURN n.country_code LIMIT 100",
        "Match (n:person) RETURN n LIMIT 9300",
        "Match (n:company) RETURN n LIMIT 2" ,
        "Match (n:compound) RETURN n LIMIT 1"
        ]
    qlist_test(qlist)

Note

Since the work being performed by singleq_test is primarily database access, I would think that a multithreading pool would work just as well if not better -- at least it is something to try. The only change you need to make is:

from multiprocessing.pool import ThreadPool as Pool, TimeoutError

Now the question becomes whether a single connection can be shared among multiple threads running in the same process or not.

Booboo0
Related