python - multitarea - ¿Cómo uso el subproceso.Popen para conectar múltiples procesos por tuberías?

procesos concurrentes python

Question

¿Cómo ejecuto el siguiente comando de shell utilizando el módulo de subprocess Python?

echo "input data" | awk -f script.awk | sort > outfile.txt

Los datos de entrada vendrán de una cadena, por lo que realmente no necesito echo . He llegado tan lejos, ¿alguien puede explicar cómo lo hago para canalizarlo también?

p_awk = subprocess.Popen(["awk","-f","script.awk"],
                          stdin=subprocess.PIPE,
                          stdout=file("outfile.txt", "w"))
p_awk.communicate( "input data" )

ACTUALIZACIÓN : Tenga en cuenta que, si bien la respuesta aceptada a continuación no responde realmente la pregunta formulada, creo que S.Lott tiene razón y es mejor evitar tener que resolver ese problema en primer lugar.

Answer 1

Estarías un poco más feliz con lo siguiente.

import subprocess

awk_sort = subprocess.Popen( "awk -f script.awk | sort > outfile.txt",
    stdin=subprocess.PIPE, shell=True )
awk_sort.communicate( b"input data/n" )

Delegar parte del trabajo al shell. Permita que conecte dos procesos con una tubería.

Sería mucho más feliz reescribir ''script.awk'' en Python, eliminando awk y el pipeline.

Editar . Algunas de las razones para sugerir que awk no ayudan.

[Hay demasiadas razones para responder a través de comentarios.]

1. Awk está agregando un paso sin valor significativo. No hay nada único en el procesamiento de awk que Python no maneje.

2. La canalización de awk a sort, para grandes conjuntos de datos, puede mejorar el tiempo de procesamiento transcurrido. Para conjuntos cortos de datos, no tiene un beneficio significativo. Una medición rápida de awk >file ; sort file awk >file ; sort file y awk | sort awk | sort revelará que la concurrencia ayuda. Con sort, raramente ayuda porque sort no es un filtro de una sola pasada.

3. La simplicidad del procesamiento de "Python to sort" (en lugar de "Python to awk to sort") previene el tipo exacto de preguntas que se hacen aquí.

4. Python - while wordier than awk - también es explícito donde awk tiene ciertas reglas implícitas que son opacas para los novatos, y confunde a los no especialistas.

5. Awk (como el script de shell en sí) agrega Yet Another Another Programming language. Si todo esto se puede hacer en un idioma (Python), al eliminar el shell y la programación awk se eliminan dos lenguajes de programación, lo que permite que alguien se concentre en las partes de la tarea que producen el valor.

En pocas palabras: awk no puede agregar un valor significativo. En este caso, awk es un costo neto; Agregó suficiente complejidad que era necesario hacer esta pregunta. Eliminar awk será una ganancia neta.

Barra lateral ¿Por qué construir una tubería ( a | b ) es tan difícil?

Cuando el caparazón se enfrenta con a | b a | b tiene que hacer lo siguiente.

1. Tenedor un proceso de niño del shell original. Esto eventualmente se convertirá en b.

2. Construye una tubería de os. (no es un subproceso de os.pipe() ) sino que llama a os.pipe() que devuelve dos nuevos descriptores de archivo que están conectados a través del búfer común. En este punto, el proceso tiene stdin, stdout, stderr de su padre, más un archivo que será "a stdout" y "b''s stdin".

3. Tenedor un niño. El niño reemplaza su stdout con el stdout de la nueva a. Ejecute el proceso a.

4. El b niño cierra reemplaza su stdin con el nuevo b''s stdin. Ejecute el proceso b .

5. El b niño espera que se complete.

6. El padre está esperando que b se complete.

Creo que lo anterior se puede usar recursivamente para generar a | b | c a | b | c a | b | c , pero tiene que hacer una paréntesis implícita entre tuberías largas, tratándolas como si fueran a | (b | c) a | (b | c) .

Como Python tiene os.pipe() , os.exec() y os.fork() , y puede reemplazar sys.stdin y sys.stdout , hay una forma de hacer lo anterior en Python puro. De hecho, es posible que pueda resolver algunos accesos directos usando os.pipe() y subprocess.Popen .

Sin embargo, es más fácil delegar esa operación al shell.

Answer 2

Inspirado por la respuesta de @ Cristian. Me encontré con el mismo problema, pero con un comando diferente. Así que estoy poniendo mi ejemplo probado, que creo que podría ser útil:

grep_proc = subprocess.Popen(["grep", "rabbitmq"],
                             stdin=subprocess.PIPE, 
                             stdout=subprocess.PIPE)
subprocess.Popen(["ps", "aux"], stdout=grep_proc.stdin)
out, err = grep_proc.communicate()

Esto es probado.

Lo que se ha hecho

• Acción declarada grep pereza con stdin de la tubería. Este comando se ejecutará en la ejecución del comando ps cuando el conducto se llene con la salida estándar de ps .
• Llamado el comando principal ps con stdout dirigido a la tubería utilizada por el comando grep .
• Grep se comunicó para obtener el stdout de la tubería.

Me gusta de esta manera porque es una concepción de tubería natural envuelta suavemente con interfaces de subprocess .

Answer 3

Las respuestas anteriores omitieron un punto importante. La sustitución de la tubería de la carcasa es básicamente correcta, como lo señala Geocar. Es casi suficiente para ejecutar la communicate sobre el último elemento de la tubería.

El problema restante es pasar los datos de entrada a la tubería. Con múltiples subprocesos, una communicate(input_data) simple communicate(input_data) en el último elemento no funciona; se cuelga para siempre. Necesitas crear una tubería y un hijo manualmente de esta manera:

import os
import subprocess

input = """/
input data
more input
""" * 10

rd, wr = os.pipe()
if os.fork() != 0: # parent
    os.close(wr)
else:              # child
    os.close(rd)
    os.write(wr, input)
    os.close(wr)
    exit()

p_awk = subprocess.Popen(["awk", "{ print $2; }"],
                         stdin=rd,
                         stdout=subprocess.PIPE)
p_sort = subprocess.Popen(["sort"], 
                          stdin=p_awk.stdout,
                          stdout=subprocess.PIPE)
p_awk.stdout.close()
out, err = p_sort.communicate()
print (out.rstrip())

Ahora el niño proporciona la entrada a través del conducto y el padre llama a comunicar (), que funciona como se esperaba. Con este enfoque, puede crear tuberías largas arbitrarias sin recurrir a "delegar parte del trabajo en el shell". Lamentablemente, la documentación del subproceso no menciona esto.

Hay formas de lograr el mismo efecto sin tuberías:

from tempfile import TemporaryFile
tf = TemporaryFile()
tf.write(input)
tf.seek(0, 0)

Ahora usa stdin=tf para p_awk . Es una cuestión de gusto lo que prefieres.

Lo anterior aún no es 100% equivalente a las tuberías de bash porque el manejo de la señal es diferente. Puede ver esto si agrega otro elemento de tubería que trunca la salida de sort , por ejemplo, la head -n 10 . Con el código anterior, la sort imprimirá un mensaje de error de "pipa rota" en stderr . No verá este mensaje cuando ejecute la misma canalización en el shell. (Esa es la única diferencia, el resultado en stdout es el mismo). La razón parece ser que Popen de Popen establece SIG_IGN para SIGPIPE , mientras que el shell lo deja en SIG_DFL , y el manejo de la señal de sort es diferente en estos dos casos.

Answer 4

Para emular una canalización de shell:

from subprocess import check_call

check_call(''echo "input data" | a | b > outfile.txt'', shell=True)

sin invocar el shell (consulte 17.1.4.2. Sustitución de la canalización del shell ):

#!/usr/bin/env python
from subprocess import Popen, PIPE

a = Popen(["a"], stdin=PIPE, stdout=PIPE)
with a.stdin:
    with a.stdout, open("outfile.txt", "wb") as outfile:
        b = Popen(["b"], stdin=a.stdout, stdout=outfile)
    a.stdin.write(b"input data")
statuses = [a.wait(), b.wait()] # both a.stdin/stdout are closed already

plumbum proporciona algo de azúcar sintáctica:

#!/usr/bin/env python
from plumbum.cmd import a, b # magic

(a << "input data" | b > "outfile.txt")()

El análogo de:

#!/bin/sh
echo "input data" | awk -f script.awk | sort > outfile.txt

es:

#!/usr/bin/env python
from plumbum.cmd import awk, sort

(awk["-f", "script.awk"] << "input data" | sort > "outfile.txt")()

Answer 5

http://www.python.org/doc/2.5.2/lib/node535.html cubrió esto bastante bien. ¿Hay alguna parte de esto que no entendiste?

Su programa sería bastante similar, pero el segundo Popen tendría stdout = en un archivo, y no necesitaría la salida de su .communicate() .

Answer 6

EDITAR: pipes está disponible en Windows pero, lo que es más importante, no parece funcionar en Windows. Ver comentarios a continuación.

La biblioteca estándar de Python ahora incluye el módulo de pipes para manejar esto:

https://docs.python.org/2/library/pipes.html , https://docs.python.org/3.4/library/pipes.html

No estoy seguro de cuánto tiempo ha existido este módulo, pero este enfoque parece ser mucho más simple que el de subprocess .

Answer 7

import subprocess

some_string = b''input_data''

sort_out = open(''outfile.txt'', ''wb'', 0)
sort_in = subprocess.Popen(''sort'', stdin=subprocess.PIPE, stdout=sort_out).stdin
subprocess.Popen([''awk'', ''-f'', ''script.awk''], stdout=sort_in, 
                 stdin=subprocess.PIPE).communicate(some_string)