ノンブロッキングFIFO

Question

2つのpythonプロセスの間でfifoを作ることができます。リーダーが入力を処理できない場合、行を削除することはできますか？

• 読者はライター、その後速くreadまたはreadlineしようとした場合には、それをブロックする必要があり、書き込みます。
• リーダーがライターの書き込み速度より速く動作しない場合、ライターはブロックしないでください。ラインは一度に1ラインを除いてバッファされるべきではなく、書かれた最後のラインのみが次のreadlineの試みで読者によって受信されるべきである。

これは名前付きFIFOで可能ですか、これを実現する他の簡単な方法はありますか？

Answer 1

次のコードでは、名前付きFIFOを使用して2つのスクリプト間の通信を可能にしています。

• リーダーがライターよりも速くreadにしようとすると、ブロックされます。
• リーダーがライターに追いつかない場合、ライターはブロックしません。
• 操作はバッファ指向です。ライン指向の操作は現在実装されていません。
• このコードは概念実証と考えるべきです。遅延とバッファサイズは任意です。

コード

import argparse 
import errno 
import os 
from select import select 
import time 

class OneFifo(object): 
    def __init__(self, name): 
     self.name = name 

    def __enter__(self): 
     if os.path.exists(self.name): 
      os.unlink(self.name) 
     os.mkfifo(self.name) 
     return self 

    def __exit__(self, exc_type, exc_value, exc_traceback): 
     if os.path.exists(self.name): 
      os.unlink(self.name) 

    def write(self, data): 
     print "Waiting for client to open FIFO..." 
     try: 
      server_file = os.open(self.name, os.O_WRONLY | os.O_NONBLOCK) 
     except OSError as exc: 
      if exc.errno == errno.ENXIO: 
       server_file = None 
      else: 
       raise 
     if server_file is not None: 
      print "Writing line to FIFO..." 
      try: 
       os.write(server_file, data) 
       print "Done." 
      except OSError as exc: 
       if exc.errno == errno.EPIPE: 
        pass 
       else: 
        raise 
      os.close(server_file) 

    def read_nonblocking(self): 
     result = None 
     try: 
      client_file = os.open(self.name, os.O_RDONLY | os.O_NONBLOCK) 
     except OSError as exc: 
      if exc.errno == errno.ENOENT: 
       client_file = None 
      else: 
       raise 
     if client_file is not None: 
      try: 
       rlist = [client_file] 
       wlist = [] 
       xlist = [] 
       rlist, wlist, xlist = select(rlist, wlist, xlist, 0.01) 
       if client_file in rlist: 
        result = os.read(client_file, 1024) 
      except OSError as exc: 
       if exc.errno == errno.EAGAIN or exc.errno == errno.EWOULDBLOCK: 
        result = None 
       else: 
        raise 
      os.close(client_file) 
     return result 

    def read(self): 
     try: 
      with open(self.name, 'r') as client_file: 
       result = client_file.read() 
     except OSError as exc: 
      if exc.errno == errno.ENOENT: 
       result = None 
      else: 
       raise 
     if not len(result): 
      result = None 
     return result 

def parse_argument(): 
    parser = argparse.ArgumentParser() 
    parser.add_argument('-c', '--client', action='store_true', 
         help='Set this flag for the client') 
    parser.add_argument('-n', '--non-blocking', action='store_true', 
         help='Set this flag to read without blocking') 
    result = parser.parse_args() 
    return result 

if __name__ == '__main__': 
    args = parse_argument() 
    if not args.client: 
     with OneFifo('known_name') as one_fifo: 
      while True: 
       one_fifo.write('one line') 
       time.sleep(0.1) 
    else: 
     one_fifo = OneFifo('known_name') 
     while True: 
      if args.non_blocking: 
       result = one_fifo.read_nonblocking() 
      else: 
       result = one_fifo.read() 
      if result is not None: 
       print result 

clientはFIFOを開いた場合serverチェック。 clientがFIFOを開いている場合、serverは行を書き込みます。それ以外の場合は、serverが実行を継続します。ブロックされている読み取りが問題を引き起こすため、非ブロック読み取りを実装しました。serverが再起動すると、ほとんどの場合clientはブロックされたままになり、回復しません。ノンブロッキングclientを使用すると、再起動がより容易になります。server起動時に

出力

[user@machine:~] python onefifo.py 
Waiting for client to open FIFO... 
Waiting for client to open FIFO... 
Writing line to FIFO...   
Done. 
Waiting for client to open FIFO... 
Writing line to FIFO... 
Done. 

[user@machine:~] python onefifo.py -c 
one line 
one line 

ノート

serverは、FIFOが既に存在していることを検出した場合、それを削除します。これはclientsにserverが再開したことを通知する最も簡単な方法です。この通知は通常、ブロックバージョンclientによって無視されます。

Answer 2

私が知る限り、それは実際にはFIFO（キュー）ではなく、単一の変数です。キューまたはパイプを最大サイズ1に設定すると実装可能かもしれませんが、プロセスの1つのオブジェクトでLockを使用する方が効果的ですproxy object。読者は読む度にNoneに設定し、書き込むたびにその内容を上書きします。

オブジェクトのプロキシとロックのプロキシをすべての関連プロセスの引数として渡すことで、それらを他のプロセスに渡すことができます。これを少し便利にするために、Managerを使うことができます。これは渡すことができるプロキシを持つ単一のオブジェクトを提供し、そこに入れたい他のオブジェクト（ロックを含む）のプロキシを含みます。 This answerは、オブジェクトを新しいプロセスに渡すためのManagerの適切な使用例を提供します。