Me gustaría concurrent.futures.ProcessPoolExecutor.map() para llamar a una función que consta de 2 o más argumentos. En el siguiente ejemplo, recurrí al uso de una función lambda y definí ref como una matriz del mismo tamaño que la lista de numberlist con un valor idéntico.
1ra pregunta: ¿Hay una mejor manera de hacer esto? En el caso de que el tamaño de la lista de números pueda tener un tamaño de entre un millón y un billón de elementos, por lo tanto, el tamaño de la referencia tendría que seguir a la lista de números, este enfoque ocupa innecesariamente memoria valiosa, lo que me gustaría evitar. Hice esto porque leí que la función de map terminará su mapeo hasta que se alcance el extremo más corto de la matriz.
import concurrent.futures as cf nmax = 10 numberlist = range(nmax) ref = [5, 5, 5, 5, 5, 5, 5, 5, 5, 5] workers = 3 def _findmatch(listnumber, ref): print('def _findmatch(listnumber, ref):') x='' listnumber=str(listnumber) ref = str(ref) print('listnumber = {0} and ref = {1}'.format(listnumber, ref)) if ref in listnumber: x = listnumber print('x = {0}'.format(x)) return xa = map(lambda x, y: _findmatch(x, y), numberlist, ref) for n in a: print(n) if str(ref[0]) in n: print('match') with cf.ProcessPoolExecutor(max_workers=workers) as executor: #for n in executor.map(_findmatch, numberlist): for n in executor.map(lambda x, y: _findmatch(x, ref), numberlist, ref): print(type(n)) print(n) if str(ref[0]) in n: print('match') Al ejecutar el código anterior, descubrí que la función de map podía lograr el resultado deseado. Sin embargo, cuando transfirí los mismos términos a concurrent.futures.ProcessPoolExecutor.map(), python3.5 falló con este error:
Traceback (most recent call last): File "/usr/lib/python3.5/multiprocessing/queues.py", line 241, in _feed obj = ForkingPickler.dumps(obj) File "/usr/lib/python3.5/multiprocessing/reduction.py", line 50, in dumps cls(buf, protocol).dump(obj) _pickle.PicklingError: Can't pickle <function <lambda> at 0x7fd2a14db0d0>: attribute lookup <lambda> on __main__ failedPregunta 2 : ¿Por qué ocurrió este error y cómo obtengo concurrent.futures.ProcessPoolExecutor.map() para llamar a una función con más de 1 argumento?
Para responder primero a su segunda pregunta, está obteniendo una excepción porque una función lambda como la que está usando no es seleccionable. Dado que Python usa el protocolo pickle para serializar los datos que pasan entre el proceso principal y los procesos de trabajo de ProcessPoolExecutor , esto es un problema. No está claro por qué está usando una lambda . La lambda que tenía toma dos argumentos, al igual que la función original. Podría usar _findmatch directamente en lugar de la lambda y debería funcionar.
with cf.ProcessPoolExecutor(max_workers=workers) as executor: for n in executor.map(_findmatch, numberlist, ref): ... En cuanto al primer problema sobre pasar el segundo argumento constante sin crear una lista gigante, podría resolverlo de varias maneras. Un enfoque podría ser usar itertools.repeat para crear un objeto iterable que repita el mismo valor para siempre cuando se repite.
Pero un mejor enfoque probablemente sería escribir una función adicional que pase el argumento constante por usted. (¿Quizás es por eso que estaba tratando de usar una función lambda ?) Debería funcionar si la función que usa es accesible en el espacio de nombres de nivel superior del módulo:
def _helper(x): return _findmatch(x, 5) with cf.ProcessPoolExecutor(max_workers=workers) as executor: for n in executor.map(_helper, numberlist): ...(1) No es necesario hacer una lista. Puede usar itertools.repeat para crear un iterador que simplemente repita algún valor.
(2) Debe pasar una función con nombre al map porque se pasará al subproceso para su ejecución. map usa el protocolo pickle para enviar cosas, las lambdas no se pueden decapar y, por lo tanto, no pueden ser parte del mapa. Pero es totalmente innecesario. Todo lo que hizo su lambda fue llamar a una función de 2 parámetros con 2 parámetros. Quítelo por completo.
El código de trabajo es
import concurrent.futures as cf import itertools nmax = 10 numberlist = range(nmax) workers = 3 def _findmatch(listnumber, ref): print('def _findmatch(listnumber, ref):') x='' listnumber=str(listnumber) ref = str(ref) print('listnumber = {0} and ref = {1}'.format(listnumber, ref)) if ref in listnumber: x = listnumber print('x = {0}'.format(x)) return x with cf.ProcessPoolExecutor(max_workers=workers) as executor: #for n in executor.map(_findmatch, numberlist): for n in executor.map(_findmatch, numberlist, itertools.repeat(5)): print(type(n)) print(n) #if str(ref[0]) in n: # print('match')Con respecto a su primera pregunta, ¿entiendo correctamente que desea pasar un argumento cuyo valor se determina solo en el momento en que llama al map pero constante para todas las instancias de la función asignada? Si es así, haría el map con una función derivada de una "función de plantilla" con el segundo argumento ( ref en su ejemplo) integrado usando functools.partial :
from functools import partial refval = 5 def _findmatch(ref, listnumber): # arguments swapped ... with cf.ProcessPoolExecutor(max_workers=workers) as executor: for n in executor.map(partial(_findmatch, refval), numberlist): ... Re. pregunta 2, primera parte: no he encontrado el fragmento de código exacto que intenta encurtir (serializar) la función que luego debe ejecutarse en paralelo, pero parece natural que eso tenga que suceder, no solo los argumentos sino también la función debe transferirse a los trabajadores de alguna manera, y probablemente deba serializarse para esta transferencia. El hecho de que las funciones partial se puedan encurtir mientras que las lambda no se mencionan en otra parte, por ejemplo aquí: https://stackoverflow.com/a/19279016/6356764 .
Re. pregunta 2, segunda parte: si quisiera llamar a una función con más de un argumento en ProcessPoolExecutor.map , le pasaría la función como primer argumento, seguida de una iteración de primeros argumentos para la función, seguida de una iteración de sus segundos argumentos, etc. En su caso:
for n in executor.map(_findmatch, numberlist, ref): ...