multiprocessing di zeromq e python, troppi file aperti

Question

Ho un modello basato su agente, in cui diversi agenti sono avviati da un processo centrale e comunicano tramite un altro processo centrale. Ogni agente e il processo di comunicazione comunicano tramite zmq. Tuttavia, quando inizio a più di 100 agenti standard_out invia:

Invalid argument (src/stream_engine.cpp: 143) Troppi file aperti (src/ipc_listener.cpp: 292)

e Mac Os richiede una segnalazione del problema:

Python si è chiuso improvvisamente mentre utilizzava il plug-in libzmq.5.dylib.

Il problema mi sembra che siano aperti troppi contesti. Ma come posso evitarlo con il multiprocessing?

I allegare una parte del codice qui sotto:

class Agent(Database, Logger, Trade, Messaging, multiprocessing.Process): 
    def __init__(self, idn, group, _addresses, trade_logging): 
     multiprocessing.Process.__init__(self) 
     .... 

    def run(self): 
     self.context = zmq.Context() 
     self.commands = self.context.socket(zmq.SUB) 
     self.commands.connect(self._addresses['command_addresse']) 
     self.commands.setsockopt(zmq.SUBSCRIBE, "all") 
     self.commands.setsockopt(zmq.SUBSCRIBE, self.name) 
     self.commands.setsockopt(zmq.SUBSCRIBE, group_address(self.group)) 

     self.out = self.context.socket(zmq.PUSH) 
     self.out.connect(self._addresses['frontend']) 
     time.sleep(0.1) 
     self.database_connection = self.context.socket(zmq.PUSH) 
     self.database_connection.connect(self._addresses['database']) 
     time.sleep(0.1) 
     self.logger_connection = self.context.socket(zmq.PUSH) 
     self.logger_connection.connect(self._addresses['logger']) 

     self.messages_in = self.context.socket(zmq.DEALER) 
     self.messages_in.setsockopt(zmq.IDENTITY, self.name) 
     self.messages_in.connect(self._addresses['backend']) 

     self.shout = self.context.socket(zmq.SUB) 
     self.shout.connect(self._addresses['group_backend']) 
     self.shout.setsockopt(zmq.SUBSCRIBE, "all") 
     self.shout.setsockopt(zmq.SUBSCRIBE, self.name) 
     self.shout.setsockopt(zmq.SUBSCRIBE, group_address(self.group)) 

     self.out.send_multipart(['!', '!', 'register_agent', self.name]) 

     while True: 
      try: 
       self.commands.recv() # catches the group adress. 
      except KeyboardInterrupt: 
       print('KeyboardInterrupt: %s,self.commands.recv() to catch own adress ~1888' % (self.name)) 
       break 
      command = self.commands.recv() 
      if command == "!": 
       subcommand = self.commands.recv() 
       if subcommand == 'die': 
        self.__signal_finished() 
        break 
      try: 
       self._methods[command]() 
      except KeyError: 
       if command not in self._methods: 
        raise SystemExit('The method - ' + command + ' - called in the agent_list is not declared (' + self.name) 
       else: 
        raise 
      except KeyboardInterrupt: 
       print('KeyboardInterrupt: %s, Current command: %s ~1984' % (self.name, command)) 
       break 

      if command[0] != '_': 
       self.__reject_polled_but_not_accepted_offers() 
       self.__signal_finished() 
     #self.context.destroy() 

l'intero codice è sotto http://www.github.com/DavoudTaghawiNejad/abce

Answer 1

probabilità sono che non sia troppo molti contesti, è troppe prese. Guardando attraverso il repository, vedo che stai usando (correttamente) IPC come tuo trasporto; IPC utilizza un descrittore di file come "indirizzo" per passare i dati avanti e indietro tra diversi processi. Se sto leggendo correttamente, stai aprendo fino a 7 socket per processo, quindi questo verrà sommato rapidamente. Scommetto che se esegui il debugging nel bel mezzo del tuo codice, vedrai che non fallisce quando viene creato l'ultimo contesto, ma quando l'ultimo socket spinge il limite del file aperto oltre il limite.

La mia comprensione è che il limite utente tipico per gli FD aperti è circa 1000, quindi in circa 100 agenti si stanno premendo 700 FD aperti solo per le proprie prese. Il resto è probabilmente solo tipico. Non ci dovrebbero essere problemi ad aumentare il limite fino a 10.000, più alto a seconda della situazione. Altrimenti dovrai riscrivere per utilizzare meno socket per processo per ottenere un limite di processo più alto.

Answer 2

Questo non ha nulla a che fare con zeromq né con python. È il sistema operativo sottostante, che consente solo fino a una certa soglia di file aperti simultaneamente. Questo limite include file normali, ma anche connessioni socket.

È possibile visualizzare il limite corrente utilizzando ulimit -n, sarà probabilmente impostato su 1024. Le macchine che eseguono server o che hanno altri motivi (come il multiprocesso) spesso richiedono di impostare questo limite più in alto o semplicemente su unlimited. - More info about ulimit.

Inoltre, c'è un altro global limit, tuttavia non ho ancora dovuto regolare nulla.

In generale, dovresti chiederti se hai davvero bisogno di tanti agenti. Di solito, i processi di lavoro X/2X devono essere sufficienti, dove X corrisponde al conteggio della CPU.