Comunicación Fast IPC/Socket en Java/Python

Question

Dos procesos (Java y Python) necesitan comunicarse en mi aplicación. Noté que la comunicación del socket toma el 93% del tiempo de ejecución. ¿Por qué la comunicación es tan lenta? ¿Debo buscar alternativas a la comunicación de socket o puedo hacerlo más rápido?

Actualización: Descubrí una solución simple. Parece que el flujo de salida Buffered no está realmente almacenado temporalmente por algún motivo desconocido. Por lo tanto, ahora pongo todos los datos en los búferes de cadena en ambos procesos cliente/servidor. Lo escribo en el zócalo en el método de descarga.

Todavía estoy interesado en un ejemplo del uso de memoria compartida para intercambiar datos rápidamente entre procesos.

Alguna información adicional:

1. tamaño del mensaje en el applicaiton está bajo 64kb mayor parte del tiempo.
2. El servidor está en Java, el cliente está escrito en Python.
3. El socket IPC se implementa a continuación: ¡lleva 50 ciclos y envía 200 bytes! Esto tiene que ser demasiado alto. Si envío 2 bytes en 5000 ciclos, tomará mucho menos tiempo.
4. Ambos procesos se ejecutan en una máquina Linux.
5. En la aplicación real se realizan aproximadamente 10 llamadas al iFid.write() del cliente en cada ciclo.
6. Esto se hace en un sistema Linux.

Este es el lado del servidor:

public class FastIPC{ 
    public PrintWriter out; 
    BufferedReader in; 
    Socket socket = null; 
    ServerSocket serverSocket = null; 


    public FastIPC(int port) throws Exception{ 
     serverSocket = new ServerSocket(port); 
     socket = serverSocket.accept(); 
     out = new PrintWriter(new BufferedWriter(new OutputStreamWriter(socket.getOutputStream())), true); 
     in = new BufferedReader(new InputStreamReader(socket.getInputStream())); 
    } 

    public void send(String msg){ 
     out.println(msg); // send price update to socket 
    } 

    public void flush(){ 
     out.flush(); 
    } 

    public String recv() throws Exception{ 
     return in.readLine(); 
    } 

    public static void main(String[] args){ 
     int port = 32000; 
     try{ 
      FastIPC fip = new FastIPC(port); 
      long start = new Date().getTime(); 
      System.out.println("Connected."); 
      for (int i=0; i<50; i++){ 
       for(int j=0; j<100; j++) 
        fip.send("+"); 
       fip.send("."); 
       fip.flush(); 
       String msg = fip.recv(); 
      } 
      long stop = new Date().getTime(); 
      System.out.println((double)(stop - start)/1000.); 
     }catch(Exception e){ 
      System.exit(1); 
     } 
    } 
} 

Y el lado del cliente es:

import sys 
import socket 

class IPC(object): 
    def __init__(self): 
     self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 
     self.s.connect(("localhost", 32000)) 
     self.fid = self.s.makefile() # file wrapper to read lines 
     self.listenLoop() # wait listening for updates from server 

    def listenLoop(self): 
     fid = self.fid 
     print "connected" 
     while True: 
      while True: 
       line = fid.readline() 
       if line[0]=='.': 
        break 
      fid.write('.\n') 
      fid.flush() 

if __name__ == '__main__': 
    st = IPC() 

Answer 1

Usted tiene un número de opciones. Dado que está utilizando Linux, podría usar sockets de dominio UNIX. O bien, puede serializar los datos como ASCII o JSon o algún otro formato y alimentarlos a través de un conducto, SHM (segmento de memoria compartida), cola de mensajes, DBUS o similar. Vale la pena pensar qué tipo de datos tiene, ya que estos mecanismos IPC tienen diferentes características de rendimiento. Hay un draft USENIX paper con un buen análisis de las diferentes compensaciones que vale la pena leer.

Como dices (en los comentarios a esta respuesta) que prefieres usar SHM, aquí hay algunos ejemplos de código para comenzar. El uso de la biblioteca de Python posix_ipc:

import posix_ipC# POSIX-specific IPC 
import mmap  # From Python stdlib 

class SharedMemory(object): 
    """Python interface to shared memory. 
    The create argument tells the object to create a new SHM object, 
    rather than attaching to an existing one. 
    """ 

    def __init__(self, name, size=posix_ipc.PAGE_SIZE, create=True): 
     self.name = name 
     self.size = size 
     if create: 
      memory = posix_ipc.SharedMemory(self.name, posix_ipc.O_CREX, 
              size=self.size) 
     else: 
      memory = posix_ipc.SharedMemory(self.name) 
     self.mapfile = mmap.mmap(memory.fd, memory.size) 
     os.close(memory.fd) 
     return 

    def put(self, item): 
     """Put item in shared memory. 
     """ 
     # TODO: Deal with the case where len(item) > size(self.mapfile) 
     # TODO: Guard this method with a named semaphore 
     self.mapfile.seek(0) 
     pickle.dump(item, self.mapfile, protocol=2) 
     return 

    def get(self): 
     """Get a Python object from shared memory. 
     """ 
     # TODO: Deal with the case where len(item) > size(self.mapfile) 
     # TODO: Guard this method with a named semaphore 
     self.mapfile.seek(0) 
     return pickle.load(self.mapfile) 

    def __del__(self): 
     try: 
      self.mapfile.close() 
      memory = posix_ipc.SharedMemory(self.name) 
      memory.unlink() 
     except: 
      pass 
     return  

Por el lado de Java que desea crear la misma clase, a pesar de lo que he dicho en los comentarios JTux parece proporcionar la funcionalidad equivalente y la API que necesita está en UPosixIPC clase.

El siguiente código es un resumen del tipo de cosas que necesita implementar. Sin embargo, faltan varias cosas: el manejo de excepciones es obvio, también algunos indicadores (encuéntrelos en UConstant), y deseará agregar un semáforo para proteger los métodos put/get. Sin embargo, esto debería establecerlo en el camino correcto. Recuerde que un archivo mmap o mapeado en memoria es una interfaz tipo archivo para un segmento de RAM. Por lo tanto, puede usar su descriptor de archivo como si fuera el fd de un archivo normal.

import jtux.*; 

class SHM { 

    private String name; 
    private int size; 
    private long semaphore; 
    private long mapfile; // File descriptor for mmap file 

    /* Lookup flags and perms in your system docs */ 
    public SHM(String name, int size, boolean create, int flags, int perms) { 
     this.name = name; 
     this.size = size; 
     int shm; 
     if (create) { 
      flags = flags | UConstant.O_CREAT; 
      shm = UPosixIPC.shm_open(name, flags, UConstant.O_RDWR); 
     } else { 
      shm = UPosixIPC.shm_open(name, flags, UConstant.O_RDWR); 
     } 
     this.mapfile = UPosixIPC.mmap(..., this.size, ..., flags, shm, 0); 
     return; 
    } 


    public void put(String item) { 
     UFile.lseek(this.mapfile(this.mapfile, 0, 0)); 
     UFile.write(item.getBytes(), this.mapfile); 
     return; 
    } 


    public String get() {  
     UFile.lseek(this.mapfile(this.mapfile, 0, 0)); 
     byte[] buffer = new byte[this.size]; 
     UFile.read(this.mapfile, buffer, buffer.length); 
     return new String(buffer); 
    } 


    public void finalize() { 
     UPosix.shm_unlink(this.name); 
     UPosix.munmap(this.mapfile, this.size); 
    } 

} 

Answer 2

Algunos pensamientos

• El servidor está en Java, el cliente está escrito en Python.

Una combinación extraña, pero ¿hay alguna razón por la que no se puede llamar al otro envío mediante stdin, stdout?

• El Socket IPC se implementa a continuación: ¡lleva 50 ciclos y envía 200 bytes! Esto tiene que ser demasiado alto. Si envío 2 bytes en 5000 ciclos, tomará mucho menos tiempo.

Cualquier llamada al sistema operativo va a ser relativamente lenta (latencia sabia). El uso de la memoria compartida puede pasar el kernel. Si tu problema es el rendimiento, he descubierto que puedes alcanzar 1-2 GB/s usando sockets si la latencia no es un problema para ti.

• Ambos procesos se ejecutan en una máquina Linux.

Haciendo la memoria compartida ideal.

• En la aplicación real se realizan aproximadamente 10 llamadas al iFid.write() del cliente en cada ciclo.

No estoy seguro de por qué este es el caso. ¿Por qué no construir una sola estructura/búfer y escribirlo una vez? Yo usaría un buffer directo es NIO para minimizar la latencia. Usar la traducción de caracteres es bastante caro, especialmente si solo necesitas ASCII.

• Esto se realiza en un sistema Linux.

Debe ser fácil de optimizar.

Uso la memoria compartida a través de archivos asignados de memoria. Esto se debe a que necesito registrar cada mensaje para fines de auditoría. Obtuve una latencia promedio de aproximadamente 180 ns ida y vuelta sostenida por millones de mensajes, y aproximadamente 490 ns en una aplicación real.

Una ventaja de este enfoque es que si hay pequeñas demoras, el lector puede ponerse al día rápidamente con el escritor. También es compatible con reiniciar y replicar fácilmente.

Esto solo se implementa en Java, pero el principio es bastante simple y estoy seguro de que también funcionaría en Python.

https://github.com/peter-lawrey/Java-Chronicle