Comunicazione IPC/Socket veloce in Java/Python

Question

Due processi (Java e Python) devono comunicare nella mia applicazione. Ho notato che la comunicazione socket prende il 93% del tempo di esecuzione. Perché la comunicazione è così lenta? Dovrei essere in cerca di alternative alla comunicazione socket o questo può essere reso più veloce?

Aggiornamento: ho scoperto una soluzione semplice. Sembra che il flusso di output bufferizzato non sia realmente bufferizzato per qualche motivo sconosciuto. Quindi, ora inserisco tutti i dati nei buffer di stringa in entrambi i processi client/server. Lo scrivo nel socket nel metodo flush.

Sono ancora interessato ad un esempio dell'utilizzo della memoria condivisa per lo scambio rapido di dati tra processi.

Alcune informazioni aggiuntive:

1. dimensioni Messaggio nella applicaiton è sotto 64kb maggior parte del tempo.
2. Il server è in Java, il client è scritto in Python.
3. Il socket IPC è implementato di seguito: ci vogliono 50 cicli per inviare 200 byte! Questo deve essere troppo alto. Se invio 2 byte in 5000 cicli, ci vuole molto meno tempo.
4. Entrambi i processi vengono eseguiti su una macchina Linux.
5. Nell'applicazione reale vengono eseguite ogni 10 chiamate al client iFid.write().
6. Questo è fatto su un sistema Linux.

Questo è il lato server:

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); 
     } 
    } 
} 

E il lato client è:

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

Si dispone di un numero di opzioni. Dato che stai usando Linux potresti usare socket di dominio UNIX. Oppure, è possibile serializzare i dati come ASCII o JSon o qualche altro formato e inviarlo tramite pipe, SHM (segmento di memoria condivisa), coda messaggi, DBUS o simili. Vale la pena di pensare a quale tipo di dati si dispone, in quanto questi meccanismi IPC hanno caratteristiche di prestazioni diverse. C'è uno draft USENIX paper con una buona analisi dei vari trade-off che vale la pena di leggere.

Dato che nei commenti a questa risposta si preferisce usare SHM, ecco alcuni esempi di codice per iniziare. Utilizzando la libreria 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  

Per il lato Java che si desidera creare la stessa classe, a dispetto di quello che ho detto nei commenti JTux sembra fornire la funzionalità equivalenti e l'API che serve è in UPosixIPC di classe.

Il codice seguente è uno schema del tipo di cosa che è necessario implementare. Tuttavia, mancano alcune cose: la gestione delle eccezioni è ovvia, anche alcuni flag (li trovi in ​​UConstant) e ti consigliamo di aggiungere un semaforo per proteggere i metodi put/get. Tuttavia, questo dovrebbe metterti sulla giusta strada. Ricorda che un file mmap o mappato in memoria è un'interfaccia simile a un file di un segmento di RAM. Quindi, puoi usare il suo descrittore di file come se fosse lo fd di un file normale.

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

Alcune riflessioni

• Il server è in Java, il cliente è scritto in Python.

Una combinazione dispari, ma esiste un motivo per cui non è possibile chiamare l'altro inviando tramite stdin, stdout?

• Il socket IPC è implementato di seguito: ci vogliono 50 cicli per inviare 200 byte! Questo deve essere troppo alto. Se invio 2 byte in 5000 cicli, ci vuole molto meno tempo.

Qualsiasi chiamata al sistema operativo sarà relativamente lenta (latenza). L'uso della memoria condivisa può passare il kernel. Se la velocità effettiva si verifica, ho riscontrato che è possibile raggiungere 1-2 GB/s utilizzando i socket se la latenza non è un problema di questo tipo.

• Entrambi i processi vengono eseguiti su una macchina Linux.

Fare memoria condivisa ideale.

• Nell'applicazione reale vengono eseguite ogni 10 chiamate al client iFid.write().

Non capisco perché questo è il caso. Perché non costruire una singola struttura/buffer e scriverlo una volta. Vorrei utilizzare un buffer diretto è NIO per ridurre al minimo la latenza. Usare la traduzione dei caratteri è piuttosto costoso, specialmente se hai solo bisogno di ASCII.

• Questo è fatto su un sistema Linux.

dovrebbe essere facile da ottimizzare.

ho l'uso della memoria attraverso la memoria condivisa mappata file. Questo perché ho bisogno di registrare ogni messaggio a fini di controllo. Ottengo una latenza media di circa 180 ns andata e ritorno sostenuta per milioni di messaggi e circa 490 ns in un'applicazione reale.

Un vantaggio di questo approccio è che se ci sono brevi ritardi, il lettore può recuperare molto rapidamente con lo scrittore. Supporta anche il riavvio e la replica facilmente.

Questo è implementato solo in Java, ma il principio è abbastanza semplice e sono sicuro che avrebbe funzionato in pitone pure.

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