Dans Django, comment appeler un sous-processus avec un temps de démarrage lent

Question

Supposons que vous exécutez Django sous Linux et que vous avez une vue, et que vous voulez que cette vue renvoie les données d'un sous-processus appelé cmd qui fonctionne sur un fichier que la vue crée, par exemple likeso:

def call_subprocess(request): 
    response = HttpResponse() 

    with tempfile.NamedTemporaryFile("W") as f: 
     f.write(request.GET['data']) # i.e. some data 

    # cmd operates on fname and returns output 
    p = subprocess.Popen(["cmd", f.name], 
        stdout=subprocess.PIPE, 
        stderr=subprocess.PIPE) 

    out, err = p.communicate() 

    response.write(p.out) # would be text/plain... 
    return response 

maintenant, supposons que cmd a un temps de démarrage très lent, mais un temps de fonctionnement très rapide, et n'a pas nativement un mode démon. Je voudrais améliorer le temps de réponse de cette vue.

Je voudrais faire tout le système serait beaucoup plus vite en lançant un certain nombre d'instances de cmd un travailleur à la piscine, demandez-leur d'attendre l'entrée, et ayant call_process demander un de ces processus de pool de travailleurs gèrent les données.

Ceci est vraiment 2 parties:

Partie 1. Une fonction qui appelle cmd et cmd attend l'entrée. Cela pourrait se faire avec des tuyaux, à savoir

def _run_subcmd(): 
    p = subprocess.Popen(["cmd", fname], 
     stdout=subprocess.PIPE, stderr=subprocess.PIPE) 

    out, err = p.communicate() 
    # write 'out' to a tmp file 
    o = open("out.txt", "W") 
    o.write(out) 
    o.close() 
    p.close() 
    exit() 

def _run_cmd(data): 
    f = tempfile.NamedTemporaryFile("W") 
    pipe = os.mkfifo(f.name) 

    if os.fork() == 0: 
     _run_subcmd(fname) 
    else: 
     f.write(data) 

    r = open("out.txt", "r") 
    out = r.read() 
    # read 'out' from a tmp file 
    return out 

def call_process(request): 
    response = HttpResponse() 

    out = _run_cmd(request.GET['data']) 

    response.write(out) # would be text/plain... 
    return response 

Partie 2. Un ensemble de travailleurs en cours d'exécution en arrière-plan qui sont en attente sur les données. c'est-à-dire que nous souhaitons étendre ce qui précède afin que le sous-processus soit déjà exécuté, par ex. lorsque l'instance Django initialise, ou ce call_process est d'abord appelé, un ensemble de ces travailleurs est créé

WORKER_COUNT = 6 
WORKERS = [] 

class Worker(object): 
    def __init__(index): 
     self.tmp_file = tempfile.NamedTemporaryFile("W") # get a tmp file name 
     os.mkfifo(self.tmp_file.name) 
     self.p = subprocess.Popen(["cmd", self.tmp_file], 
      stdout=subprocess.PIPE, stderr=subprocess.PIPE) 
     self.index = index 

    def run(out_filename, data): 
     WORKERS[self.index] = Null # qua-mutex?? 
     self.tmp_file.write(data) 
     if (os.fork() == 0): # does the child have access to self.p?? 
      out, err = self.p.communicate() 
      o = open(out_filename, "w") 
      o.write(out) 
      exit() 

     self.p.close() 
     self.o.close() 
     self.tmp_file.close() 
     WORKERS[self.index] = Worker(index) # replace this one 
     return out_file 

    @classmethod 
    def get_worker() # get the next worker 
    # ... static, incrementing index 

Il devrait y avoir une initialisation des travailleurs quelque part, comme ceci:

def init_workers(): # create WORKERS_COUNT workers 
    for i in xrange(0, WORKERS_COUNT): 
     tmp_file = tempfile.NamedTemporaryFile() 
     WORKERS.push(Worker(i)) 

Maintenant, ce que j'ai ci-dessus devient quelque chose likeso:

def _run_cmd(data): 
    Worker.get_worker() # this needs to be atomic & lock worker at Worker.index 

    fifo = open(tempfile.NamedTemporaryFile("r")) # this stores output of cmd 

    Worker.run(fifo.name, data) 
    # please ignore the fact that everything will be 
    # appended to out.txt ... these will be tmp files, too, but named elsewhere. 

    out = fifo.read() 
    # read 'out' from a tmp file 
    return out 


def call_process(request): 
    response = HttpResponse() 

    out = _run_cmd(request.GET['data']) 

    response.write(out) # would be text/plain... 
    return response 

Maintenant , les questions:

1. Est-ce que cela fonctionnera? (Je l'ai juste tapé du haut de ma tête dans StackOverflow, donc je suis sûr qu'il y a des problèmes, mais conceptuellement, ça va marcher)

2. Quels sont les problèmes à rechercher?

3. Y a-t-il de meilleures alternatives à cela? par exemple. Les threads pourraient-ils fonctionner aussi bien (c'est Debian Lenny Linux)? Existe-t-il des bibliothèques qui gèrent des pools de travailleurs de processus parallèles comme celui-ci?

4. Y a-t-il des interactions avec Django dont je devrais être conscient?

Merci d'avoir lu! J'espère que vous trouvez cela aussi intéressant que moi.

Brian

Answer 1

Il peut sembler que je pique ce produit comme c'est la deuxième fois que j'ai répondu avec une recommandation de ceci.

Mais il semble que vous ayez besoin d'un service Message Queing, en particulier d'une file d'attente de messages distribués.

avant est de savoir comment il fonctionnera:

1. Votre demande Django App CMD
2. CMD est ajouté à une file d'attente
3. CMD est poussé à plusieurs œuvres
4. Il est exécuté et les résultats retournés en amont

La plupart de ce code existe, et vous n'avez pas à construire votre propre système.

Jetez un oeil à Celery qui a été initialement construit avec Django.

http://www.celeryq.org/ http://robertpogorzelski.com/blog/2009/09/10/rabbitmq-celery-and-django/

Answer 2

Que diriez-vous "daemonizing" l'appel à l'aide subprocess python-daemon ou son successeur, grizzled.

Answer 3

Issy déjà mentionné Céleri, mais étant donné que les commentaires ne fonctionne pas bien avec des exemples de code, je vais répondre comme une réponse à la place.

Vous devriez essayer d'utiliser Celery de manière synchrone avec le magasin de résultats AMQP. Vous pouvez distribuer l'exécution réelle à un autre processus ou même à une autre machine. L'exécution de manière synchrone dans le céleri est facile, par exemple:

>>> from celery.task import Task 
>>> from celery.registry import tasks 

>>> class MyTask(Task): 
... 
...  def run(self, x, y): 
...   return x * y 
>>> tasks.register(MyTask) 

>>> async_result = MyTask.delay(2, 2) 
>>> retval = async_result.get() # Now synchronous 
>>> retval 4 

Le magasin de résultat AMQP fait de renvoyer le résultat très rapide, mais il est uniquement disponible dans la version de développement actuelle (en gel du code pour devenir 0.8.0)