Multiple iterations of function with multiple arguments returning multiple values using Multiprocessing in python

Viewed 114

I doing 100 iterations of the function model so, i tried using multiprocessing to distribute the tasks and for getting the final output I tried using queue but it takes too much time, failing the purpose of multiprocessing. How to solve this problem?

def model(X,Y):
  ada_clf={}
  pred1={}
  auc_final=[]
  for iteration in range(100):
    ada_clf[iteration] = AdaBoostClassifier(DecisionTreeClassifier(),n_estimators=1000,learning_rate=0.001)
    ada_clf[iteration].fit(X,Y)
    pred1[iteration]=ada_clf[iteration].predict(test1)
   
  individuallabelsfromada1=[]
  for i in range(len(test1)):
    individuallabelsfromada1.append([])
    for j in range(100):
      individuallabelsfromada1[i].append(pred1[j][i])
  
  final_labels_ada1=[]
  for each in individuallabelsfromada1:
    final_labels_ada1.append(find_majority(each))
 
  final=pd.Series(final_labels_ada1)
  temp_arr=np.array(final)
  total_labels2=pd.Series(temp_arr)

  fpr, tpr, thresholds = roc_curve(y_test, total_labels2, pos_label=1)
  auc_final.append(auc(fpr,tpr))
  q.put(total_labels2)
  q1.put(auc_final)
  q2.put(ada_clf)
  
  print('done')

  
overall_labels={}
final_auc={}
final_ada_clf={}

processes=[]
q=Queue()
q1=Queue()
q2=Queue()
for iteration in range(100):
  if __name__=='__main__':
    p=multiprocessing.Process(target=model,args=(x_train,y_labels,q,q1,q2,))
    overall_labels[iteration]=q.get()
    final_auc[iteration]=q1.get()
    final_ada_clf[iteration]=q2.get()
    p.start()
    processes.append(p)
for each in processes:
  each.join()

Below is my edited version, but returns only single output, i tried using multiple output but could not get it, so settled for only single output i.e. total_labels2:-

    ##code before this is same as before, only thing changed is arguments of model from def model(X,Y) to def model(repeat,X,Y)
    total_labels2 = pd.Series(temp_arr)

    return (repeat,total_labels2)


def get_result(total_labels2):
    global testover_forall
    testover_forall.append(total_labels2)


if __name__ == '__main__':
    import multiprocessing as mp

    testover_forall = []

    pool = mp.Pool(40)
    for repeat in range(100):
        pool.apply_async(bound_model, args= repeat, x_train, y_train), callback= get_result)
    pool.close()
    pool.join()


repetations_index=[]
for i in range(100):
  repetations_index.append(testover_forall[i][0])

final_last_labels = {}
for i in range(100):
    temp = str(i)
    final_last_labels[temp] = testover_forall[repetations_index[i]][1]

totally_last_labels=[]
for each in final_last_labels:
  temp=np.array(final_last_labels[each])
  totally_last_labels.append(temp) 
1 Answers

See my comments (actually questions) to your post.

You should be using a multiprocessing pool to limit the number of processes that you create to the number of CPU cores that you have. This will also make it easier to get return values back from your model function instead of writing results to 3 different queues (and you could have written a tuple of 3 values to just one queue). You will, of course, require other import statements and code. Given your use of numpy and other libraries, which may be implemented in C Language, you could also retry running this using threading to see if that helps or hurts performance. Do this by replacing ProcessPoolExecutor with ThreadPoolExecutor in the two places it is referenced.

Note

Any changes that model makes to passed arguments X and Y will not be reflected back to the main process. So if model is called repeatedly with the same arguments over and over, as it appears to be, it's not clear whether each call will return different values, especially if the calls are being done in parallel.

from concurrent.futures import ProcessPoolExecutor

def model(X,Y):
  ada_clf={}
  pred1={}
  auc_final=[]
  for iteration in range(100):
    ada_clf[iteration] = AdaBoostClassifier(DecisionTreeClassifier(),n_estimators=1000,learning_rate=0.001)
    ada_clf[iteration].fit(X,Y)
    pred1[iteration]=ada_clf[iteration].predict(test1)
   
  individuallabelsfromada1=[]
  for i in range(len(test1)):
    individuallabelsfromada1.append([])
    for j in range(100):
      individuallabelsfromada1[i].append(pred1[j][i])
  
  final_labels_ada1=[]
  for each in individuallabelsfromada1:
    final_labels_ada1.append(find_majority(each))
 
  final=pd.Series(final_labels_ada1)
  temp_arr=np.array(final)
  total_labels2=pd.Series(temp_arr)

  fpr, tpr, thresholds = roc_curve(y_test, total_labels2, pos_label=1)
  auc_final.append(auc(fpr,tpr))
  #q.put(total_labels2)
  #q1.put(auc_final)
  #q2.put(ada_clf)
  return total_labels2, auc_final, ada_clf
  
  #print('done')


if __name__ == '__main__':
    with ProcessPoolExecutor() as executor:
        futures = [executor.submit(model, x_train, y_labels) for iteration in range(100)]
        # simple lists will suffice:
        overall_labels = []
        final_auc = []
        final_ada_clf = []
        for future in futures:
            # get return value and store
            total_labels2, auc_final, ada_clf = future.result()
            overall_labels.append(total_labels2)
            final_auc.append(auc_final)
            final_ada_clf.append(ada_clf)

Update

It wasn't clear from the problem specification that the returned results are based on a random number generator and if successive calls to the worker function, model, do not employ a single random number generator across all processes in the multiprocessing pool, then the multiprocessing implementation will clearly return different results than the non-multiprocessing version. And it is not clear from the code provided where the random number generator is being used; it may be in library code that you have no access to. If that is the case, you have two options: (1) Use multithreading instead by changing the import statement as I have indicated in the code below; you may still achieve performance benefits as I have already mentioned or (2) Update the signature to model as follows. You will be passed a new argument, random_generator, that currently supports two methods, randint (like random.randint and random (like random.random), although it should be easy enough to modify the code if you need a different method from module random. You will use this random number generator in place of module random if you are able to. But note that this random generator will run much more slowly than the standard one; this is the price you pay.

Since we are also adding a repetition argument to model (it now has to be the final argument -- note the updated signature below), we can now use method map (no need to use a callback):

def model(X,Y, random_generator, repetition):
...
etc.

from multiprocessing import Pool
# or use the following import instead to use multithreading (but then use standard random generator):
# from multiprocessing.dummy import Pool
import random
from functools import partial
from multiprocessing.managers import BaseManager

class RandomGeneratorManager(BaseManager):
    pass

class RandomGenerator:
    def __init__(self):
        random.seed(0)

    def randint(self, a, b):
        return random.randint(a, b)

    def random(self):
        return random.random()

    # add other functions if needed


if __name__ == '__main__':
    RandomGeneratorManager.register('RandomGenerator', RandomGenerator)
    with RandomGeneratorManager() as manager:
        random_generator = manager.RandomGenerator()
        # why 40? why not use default, which is the number of cpu cores you have?:
        pool = Pool(40): 
        worker = partial(model, x_train, y_labels, random_generator)
        results = pool.map(worker, range(100))
Related