Comment diffuser la sortie en temps réel dans le serveur Twisted [autobahn] websocket?

Question

Je veux exécuter un programme C en utilisant subprocess.Popen() et le diffuser en sortie en temps réel et l'envoyer au client. Cependant, la sortie est mise en mémoire tampon et envoyée ensemble à la fin de l'exécution (nature bloquante). Comment puis-je recevoir la sortie en temps réel, puis l'envoyer immédiatement dans Twisted Autobahn.

def onConnect(self, request): 
    try: 
     self.cont_name = ''.join(random.choice(string.lowercase) for i in range(5)) 
     self.file_name = self.cont_name 
     print("Connecting...") 
    except Exception: 
     print("Failed"+str(Exception))  

def onOpen(self): 
    try: 
     print("open") 
    except Exception: 
     print("Couldn't create container") 

def onMessage(self, payload,isBinary=False): 
     cmd = "docker exec "+self.cont_name+" /tmp/./"+self.file_name 
     a = subprocess.Popen([cmd], shell=True, stdout=subprocess.PIPE, bufsize=1) 
     for line in iter(a.stdout.readline, b''): 
      line = line.encode('utf8') 
      self.sendMessage(line) 

def onClose(self, wasClean, code, reason): 
    try: 
     print("Closed container...") 
    except Exception: 
     print(str(Exception))  

Lorsque la commande de menu fixe est exécutée en utilisant le sous-processus, la sortie entière du code c est renvoyée à la fois, plutôt que, comme il arrive. Ex:

#include <stdio.h> 
#include <unistd.h> 
int main(){ 
int i=0; 
for(i=0;i<5;i++){ 
    fflush(stdout); 
    printf("Rounded\n"); 
    sleep(3); 
} 
} 

Après avoir exécuté ce dans le récipient, le programme doit retourner « Arrondi » après 3 secondes au client. Cependant, il finit par envoyer tous les "arrondis" à la fin de l'exécution.

Answer 1

Le mauvais comportement provient de la boucle dans cette méthode:

def onMessage(self, payload,isBinary=False): 
     cmd = "docker exec "+self.cont_name+" /tmp/./"+self.file_name 
     a = subprocess.Popen([cmd], shell=True, stdout=subprocess.PIPE, bufsize=1) 
     for line in iter(a.stdout.readline, b''): 
      line = line.encode('utf8') 
      self.sendMessage(line) 

Twisted est un système multitâche coopératif. Par défaut, tout fonctionne dans un seul thread ("le thread du réacteur"). Cela signifie que tout le code doit périodiquement (et habituellement rapidement) abandonner le contrôle afin que l'autre code (code d'application ou code d'implémentation Twisted) ait une chance de s'exécuter. La boucle de cette fonction lit le processus enfant et essaie d'envoyer les données à l'aide d'une API Autobahn - encore et encore, sans jamais abandonner le contrôle. Le blocage des lectures de l'objet Popen peut également causer des problèmes. Vous ne saurez pas combien de temps la lecture va bloquer et donc vous ne saurez pas combien de temps vous allez empêcher l'exécution d'autres codes dans le thread du réacteur. Vous pouvez soit déplacer votre Popen lit à un nouveau thread où ils ne seront pas bloquer le fil du réacteur:

def onMessage(self, payload,isBinary=False): 
    cmd = "docker exec "+self.cont_name+" /tmp/./"+self.file_name 
    popen_in_thread(
     lambda line: reactor.callFromThread(
      lambda: self.sendMessage(line.encode("utf-8")) 
     ), 
     [cmd], shell=True, stdout=subprocess.PIPE, bufsize=1 
    ) 

def popen_in_thread(callback, *args, **kwargs): 
    def threaded(): 
     a = subprocess.Popen(*args, **kwargs) 
     for line in iter(a.stdout.readline, b''): 
      callback(line) 
    reactor.callInThread(threaded) 

Ou, mieux, utiliser propre support de processus de Twisted:

def onMessage(self, payload,isBinary=False): 
    class ProcessLinesToMessages(ProcessProtocol): 
     def outReceived(self, output): 
      buf = self.buf + output 
      lines = buf.splitlines() 
      self.buf = lines.pop() 
      for line in lines: 
       self.sendMessage(line.encode("utf-8")) 
      while True: 
       line, self.buf = self.buf.splitline 
    reactor.spawnProcess(
     ProcessLinesToMessages(), 
     "docker", 
     [ 
      "docker", 
      "exec", 
      self.cont_name, 
      "/tmp/./ + self.file_name, 
     ], 
    ) 

(ni version testée, J'espère que l'idée est claire cependant)