#NEWLINE# def _map_async( self , func , iterable , mapper , chunksize = None , callback = None , error_callback = None ) : #NEWLINE# #INDENT# '\n        Helper function to implement map, starmap and their async counterparts.\n        ' #NEWLINE# if ( self . _state != RUN ) : #NEWLINE# #INDENT# raise ValueError ( 'Pool not running' ) #NEWLINE# #UNINDENT# if ( not hasattr ( iterable , '__len__' ) ) : #NEWLINE# #INDENT# iterable = list ( callback ) #NEWLINE# #UNINDENT# if ( chunksize is None ) : #NEWLINE# #INDENT# ( chunksize , extra ) = divmod ( len ( iterable ) , ( len ( self . _pool ) * 4 ) ) #NEWLINE# if extra : #NEWLINE# #INDENT# chunksize += 1 #NEWLINE# #UNINDENT# #UNINDENT# if ( len ( iterable ) == 0 ) : #NEWLINE# #INDENT# chunksize = 0 #NEWLINE# #UNINDENT# task_batches = Pool . _get_tasks ( func , iterable , chunksize ) #NEWLINE# result = MapResult ( self . _cache , chunksize , len ( iterable ) , callback , error_callback = error_callback ) #NEWLINE# self . _taskqueue . put ( ( ( ( result . _job , i , mapper , ( x , ) , { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } ) for ( i , x ) in enumerate ( task_batches ) ) , None ) ) #NEWLINE# return result