python3 - multiprocesamiento en python

Multiprocesamiento de Python: TypeError: cadena esperada o objeto Unicode, no se encontró NoneType

Question

Estoy intentando descargar un directorio ftp completo en paralelo.

#!/usr/bin/python
import sys
import datetime
import os
from multiprocessing import Process, Pool
from ftplib import FTP
curYear=""
remotePath =""
localPath = ""

def downloadFiles (remotePath,localPath):
        splitted = remotePath.split(''/'');
        host= splitted[2]
        path=''/''+''/''.join(splitted[3:])
        ftp = FTP(host)
        ftp.login()
        ftp.cwd(path)
        filenames =  ftp.nlst()
        total=len(filenames)
        i=0
        pool = Pool()
        for filename in filenames:
                        local_filename = os.path.join(localPath,filename)
                        pool.apply_async(downloadFile, (filename,local_filename,ftp))
                        #downloadFile(filename,local_filename,ftp);
                        i=i+1

        pool.close()
        pool.join()
        ftp.close()

def downloadFile(filename,local_filename,ftp):
        file = open(local_filename, ''wb'')
        ftp.retrbinary(''RETR ''+ filename, file.write)
        file.close()

def getYearFromArgs():
        if len(sys.argv) >= 2 and sys.argv[1] == "Y":
                year = sys.argv[2]
                del sys.argv[1:2]
        else:
                year = str(datetime.datetime.now().year)
        return year

def assignGlobals():
        global p
        global remotePath
        global localPath
        global URL
        global host
        global user
        global password
        global sqldb
        remotePath = ''ftp://ftp3.ncdc.noaa.gov/pub/data/noaa/isd-lite/%s/'' % (curYear)
        localPath = ''/home/isd-lite/%s/'' % (curYear)

def main():
        global curYear
        curYear=getYearFromArgs()
        assignGlobals()
        downloadFiles(remotePath,localPath)

if __name__ == "__main__":
        main()

Pero me sale esta excepción:

Exception in thread Thread-1:
Traceback (most recent call last):
  File "/usr/lib64/python2.6/threading.py", line 532, in __bootstrap_inner
    self.run()
  File "/usr/lib64/python2.6/threading.py", line 484, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/usr/lib64/python2.6/multiprocessing/pool.py", line 225, in _handle_tasks
    put(task)
TypeError: expected string or Unicode object, NoneType found

Si comento fuera de esta línea:

pool.apply_async(downloadFile, (filename,local_filename,ftp)

y quitar el comentario en esta línea:

downloadFile(filename,local_filename,ftp);

Entonces funciona bien pero es lento y no multiproceso.

Answer 1

Has probado:

pool.apply_async(downloadFile, args=(filename,local_filename,ftp))

El prototipo es:

apply_async(func, args=(), kwds={}, callback=None)

Answer 2

Actualización, 9 de mayo de 2014:

He determinado la limitación precisa. Es posible enviar objetos a través de los límites del proceso a los procesos de trabajo, siempre y cuando los objetos puedan ser decapados por la instalación de pickle de Python . El problema que describí en mi respuesta original ocurrió porque intentaba enviar un identificador de archivo a los trabajadores. Un experimento rápido demuestra por qué esto no funciona:

>>> f = open("/dev/null")
>>> import pickle
>>> pickle.dumps(f)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/lib/python2.7/pickle.py", line 1374, in dumps
    Pickler(file, protocol).dump(obj)
  File "/usr/lib/python2.7/pickle.py", line 224, in dump
    self.save(obj)
  File "/usr/lib/python2.7/pickle.py", line 306, in save
    rv = reduce(self.proto)
  File "/usr/lib/python2.7/copy_reg.py", line 70, in _reduce_ex
    raise TypeError, "can''t pickle %s objects" % base.__name__
TypeError: can''t pickle file objects

Por lo tanto, si te encuentras con el error de Python que te llevó a encontrar esta pregunta de desbordamiento de pila, asegúrate de que todas las cosas que estás enviando a través de los límites del proceso pueden ser decapadas.

Respuesta original:

Estoy un poco tarde para responder. Sin embargo, encontré el mismo mensaje de error que el póster original al intentar utilizar el módulo de multiprocesamiento de Python. Registraré mis hallazgos para que cualquier otra persona que se tope con este hilo tenga algo que probar.

En mi caso, el error se produjo debido a lo que estaba tratando de enviar al grupo de trabajadores: estaba tratando de pasar una serie de objetos de archivo para que los trabajadores del grupo los masticaran. Aparentemente, eso es demasiado para enviar a través de los límites del proceso en Python. Resolví el problema enviando los diccionarios de trabajadores de grupo que especificaban las cadenas de entrada y salida del nombre de archivo.

Así que parece que el iterable que proporciona a la función como apply_async (usé map() y imap_unordered() ) puede contener una lista de números o cadenas, o incluso una estructura de datos detallada del diccionario (siempre y cuando los valores no existan). t objetos).

En tu caso:

pool.apply_async(downloadFile, (filename,local_filename,ftp))

ftp es un objeto, que podría estar causando el problema. Como solución alternativa, recomendaría enviar los parámetros al trabajador (se parece a host y path en este caso) y dejar que el trabajador ejemplifique el objeto y se ocupe de la limpieza.