No Django, como chamar um subprocesso com um tempo de arranque lento

Question

Suponha que você está executando o Django no Linux, e você tem uma visão, e você quer que vista a devolver os dados de um sub-processo chamado cmd que opera em um arquivo que o ponto de vista cria, por exemplo 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

Agora, suponha cmd tem um tempo de arranque muito lento, mas um tempo muito rápido operando, e não nativamente têm um modo daemon. Gostaria de melhorar a resposta em tempo deste ponto de vista.

Eu gostaria de fazer todo o sistema iria correr muito mais rápido, iniciando-se um número de instâncias de cmd em um trabalhador de piscina, tê-los esperar por entrada, e tendo call_process pedir um desses processos piscina trabalhador lidar com os dados.

Esta é realmente 2 partes:

Função Parte 1. A que as chamadas cmd e cmd aguarda a entrada. Isso poderia ser feito com tubos, i.

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

Parte 2. Um conjunto de trabalhadores que executam em segundo plano que estão à espera dos dados. ou seja Queremos estender acima de modo que o subprocesso já está em execução, por exemplo, quando os inicializa instância do Django, ou este call_process é chamado pela primeira vez, um conjunto desses trabalhadores é criado

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 

Deve haver alguma inicialização de trabalhadores em algum lugar, como este:

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

Agora, o que eu tenho acima torna-se algo 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

Agora, as perguntas:

1. Será que este trabalho? (Eu digitei isso em cima da minha cabeça em StackOverflow, então eu tenho certeza que existem problemas, mas conceitualmente, será que funciona)

2. Quais são os problemas que procurar?

3. Existem alternativas melhores para isso? por exemplo. threads poderiam funcionar tão bem (é Debian Lenny Linux)? Existem quaisquer bibliotecas que lidam com processos paralelos trabalhadores-piscinas como este?

4. Existem interações com Django que eu deveria estar ciente de?

Obrigado por ler! Eu espero que você encontrar este tão interessante um problema como eu.

Brian

Answer 1

Pode parecer que eu estou punting este produto como este é a segunda vez que têm respondido com uma recomendação deste.

Mas parece que você precisa de um serviço de mensagem Queing, em particular, uma fila de mensagens distribuído.

ere é como vai funcionar:

1. O seu Django App solicita CMD
2. CMD é adicionado a uma fila
3. CMD é empurrado para várias obras
4. É executado e os resultados voltaram a montante

A maior parte deste código existe, e você não tem que ir sobre a construção de seu próprio sistema.

Tenha um olhar em Aipo que foi inicialmente construído com Django.

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

Answer 2

Issy já mencionado Aipo, mas desde que os comentários não funciona bem com amostras de código, eu vou responder como uma resposta em seu lugar.

Você deve tentar usar aipo sincronicamente com a loja resultado AMQP. Você poderia distribuir a execução real para outro processo ou mesmo outra máquina. Executar de forma síncrona no aipo é fácil, por exemplo:.

>>> 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

A loja resultado AMQP torna o envio de volta o resultado muito rápido, mas ele só está disponível na versão de desenvolvimento atual (em code-congelamento para se tornar 0.8.0)

Answer 3

Como sobre "daemonizing" a chamada subprocesso usando python-daemon ou seu sucessor, grisalho .