Multiprocessing on a list being passed to a function

Question

I have a function that processes one url at a time:

def sanity(url):
    try:
       if 'media' in url[:10]:
           url = "http://dummy.s3.amazonaws.com" + url
       req = urllib.request.Request(url, headers={'User-Agent' : "Magic Browser"})
       ret = urllib.request.urlopen(req)
       allurls.append(url)
       return 1
    except (urllib.request.HTTPError,urllib.request.URLError,http.client.HTTPException, ValueError) as e:
       print(e, url)
       allurls.append(url)
       errors.append(url)
       return 0

In the main function, I have a list of URLs that need to be processed by the above function. I have tried but doesn't work.

start=0
allurls=[]
errors=[]
#arr=[0,100,200...]
for i in arr:
    p=Process(target=sanity,args=(urls[start:i],))
    p.start()
    p.join()

The above code is supposed to process the URLs in a batch of 100. But it doesn't work. I know it's not working because I am writing the lists allurls and errors to two different files and they are empty when they should not be. I have found that the lists are empty. I don't understand this behavior.

Answer 1

If I understand you correctly, you want to process chunks of a list at a time, but process those chunks in parallel? Secondly you want to store the answers in a global variable. Problem is processes are not threads, so you much more involved to share memory between them.

So the alternative is to return the answer, the below code helps you do just that. First you need to convert your list to a list of lists, each list containing the data you would want to process in that chunk. You can then pass that list of lists to a function that processes each of those. The output of each chunk is a list of answers, and a list of errors (I'd recommend to convert this to a dict to keep track of which one threw an error). Then after the processes returns you can untangle the list of lists the create your list of answers and list of errors.

Here is the code that would achieve the above:

from multiprocessing import Pool

def f(x):
    try:
        return [x*x, None]  # 0 for sucess
    except Exception as e:
        return [None, e]  # 1 for failure

def chunk_f(x):
    output = []
    errors = []
    for xi in x:
        ans, err = f(xi)
        if ans:
            output.append(ans)
        if err:
            errors.append(err)
    return [output, errors]

n = 10  # chunk size
data = list(range(95))  # test data
data.extend(['a', 'b'])

l = [data[k*n:(k+1)*n] for k in range(int(len(data)/n+1))]

p = Pool(8)
d = p.map(chunk_f, l)

new_data = []
all_errors = []
for da, de in d:
    new_data.extend(da)
    all_errors.extend(de)
print(new_data)
print(all_errors)

You can also look at this stack overflow answer on different methods of chunking your data.