2

したがって、相互に通信する必要がある 2 つの Python3.2 プロセスがあります。伝達する必要がある情報のほとんどは、標準的な辞書です。名前付きパイプを使用する方法が適しているように思われたので、両方のプロセスでインスタンス化できるパイプ クラスを作成しました。このクラスは、情報を取得するための非常に基本的なプロトコルを実装しています。

私の問題は、機能する場合と機能しない場合があることです。コードが失敗する場所を除いて、この動作にはパターンがないようです。

Pipe クラスの重要な部分は次のとおりです。さらにコードが必要な場合は叫んでください:

class Pipe:
    """
    there are a bunch of constants set up here. I dont think it would be useful to include them. Just think like this: Pipe.WHATEVER = 'WHATEVER'
    """
    def __init__(self,sPath):
        """
        create the fifo. if it already exists just associate with it
        """
        self.sPath = sPath
        if not os.path.exists(sPath):
            os.mkfifo(sPath)
        self.iFH = os.open(sPath,os.O_RDWR | os.O_NONBLOCK)
        self.iFHBlocking = os.open(sPath,os.O_RDWR)

    def write(self,dMessage):
        """
        write the dict to the fifo
        if dMessage is not a dictionary then there will be an exception here.  There never is 
        """
        self.writeln(Pipe.MESSAGE_START)
        for k in dMessage:
            self.writeln(Pipe.KEY)
            self.writeln(k)
            self.writeln(Pipe.VALUE)
            self.writeln(dMessage[k])
        self.writeln(Pipe.MESSAGE_END)

    def writeln(self,s):
        os.write(self.iFH,bytes('{0} : {1}\n'.format(Pipe.LINE_START,len(s)+1),'utf-8'))
        os.write(self.iFH,bytes('{0}\n'.format(s), 'utf-8'))
        os.write(self.iFH,bytes(Pipe.LINE_END+'\n','utf-8'))

    def readln(self):
        """
        look for LINE_START, get line size
        read until LINE_END
        clean up
        return string
        """
        iLineStartBaseLength = len(self.LINE_START)+3  #'{0} : '
        try:
            s = os.read(self.iFH,iLineStartBaseLength).decode('utf-8')
        except:
            return Pipe.READLINE_FAIL

        if Pipe.LINE_START in s:
            #get the length of the line
            sLineLen = ''
            while True:
                try:
                    sCurrent = os.read(self.iFH,1).decode('utf-8')
                except:
                    return Pipe.READLINE_FAIL
                if sCurrent == '\n':
                    break
                sLineLen += sCurrent
            try:
                iLineLen = int(sLineLen.strip(string.punctuation+string.whitespace))
            except:
                raise Exception('Not a valid line length: "{0}"'.format(sLineLen))
            #read the line
            sLine = os.read(self.iFHBlocking,iLineLen).decode('utf-8')

            #read the line terminator
            sTerm = os.read(self.iFH,len(Pipe.LINE_END+'\n')).decode('utf-8')
            if sTerm == Pipe.LINE_END+'\n':
                return sLine
            return Pipe.READLINE_FAIL

        else:
            return Pipe.READLINE_FAIL

    def read(self):
        """
        read from the fifo, make a dict
        """
        dRet        = {}
        sKey        = ''
        sValue      = ''
        sCurrent    = None

        def value_flush():
            nonlocal dRet, sKey, sValue, sCurrent
            if sKey:
                dRet[sKey.strip()] = sValue.strip()
            sKey = ''
            sValue = ''
            sCurrent = ''

        if self.message_start():
            while True:
                sLine = self.readln()
                if Pipe.MESSAGE_END in sLine:
                    value_flush()
                    return dRet
                elif Pipe.KEY in sLine:
                    value_flush()
                    sCurrent = Pipe.KEY
                elif Pipe.VALUE in sLine:
                    sCurrent = Pipe.VALUE
                else:
                    if sCurrent == Pipe.VALUE:
                        sValue += sLine
                    elif sCurrent == Pipe.KEY:
                        sKey += sLine
        else:
            return Pipe.NO_MESSAGE

ここで失敗することがあります（readlnで）：

        try:
            iLineLen = int(sLineLen.strip(string.punctuation+string.whitespace))
        except:
            raise Exception('Not a valid line length: "{0}"'.format(sLineLen))

他の場所では失敗しません。

エラーの例は次のとおりです。

Not a valid line length: "KE 17"

断続的であるという事実は、ある種の競合状態が原因であると私に言っています。私はそれが何であるかを理解するのに苦労しています. 何か案は？

プロセスの呼び出しに関する追加の編集

パイプの使用方法は、同じパスでコンストラクターを呼び出すことにより、processA と ProcessB でインスタンス化されます。次に、プロセス A がパイプに断続的に書き込み、プロセス B がそこから読み取ろうとします。物事を双方向として機能させようとすることは決してありません。

これは、状況のより長い説明です。私は質問を短くしようとしてきましたが、それをあきらめる時が来たと思います。とにかく、いい感じにプレイする必要があるデーモンと Pyramid プロセスがあります。使用中の Pipe インスタンスは 2 つあります。1 つは Pyramid のみが書き込み、もう 1 つはデーモンのみが書き込みます。Pyramid が書いたものは本当に短いです。私はこのパイプでエラーを経験したことがありません。デーモンが書くものはずっと長く、これは私に悲しみを与えているパイプです. どちらのパイプも同じ方法で実装されます。どちらのプロセスも、それぞれのパイプに辞書を書き込むだけです (そうでない場合は、Pipe.write に例外が発生します)。

基本的なアルゴリズムは次のとおりです: Pyramid はデーモンを生成し、デーモンは運命のクレイズ オブジェクト階層と膨大な RAM 消費をロードします。Pyramid は POST リクエストをデーモンに送信し、デーモンは大量の計算を行い、データを Pyramid に送信して、人間に優しいページをレンダリングできるようにします。その後、人間は HTML フォームなどに入力することで階層内にあるものに応答できるため、ピラミッドは別の辞書をデーモンに送信し、デーモンは辞書の応答を返します。

つまり、1 つのパイプだけが問題を示しており、問題のあるパイプには他のパイプよりも多くのトラフィックがあり、辞書のみがいずれかに書き込まれることが保証されています。

質問とコメントへの応答として編集

試してみるように言う前に... 読み物を除いて。例外がまったく発生するという事実は、私を悩ませているものです。iLineLengh = int(stuff) は、常に整数のように見える文字列を渡す必要があるように見えます。これはほとんどの場合であり、すべてではありません。したがって、おそらく整数ではないことについてコメントしたいという衝動を感じた場合は、コメントしないでください。

私の質問を言い換えると、競合状態を見つければ、あなたは私のヒーローになります。

少し例を編集します。

process_1.py:

oP = Pipe(some_path)
while 1:
    oP.write({'a':'foo','b':'bar','c':'erm...','d':'plop!','e':'etc'})

プロセス_2.py:

oP = Pipe(same_path_as_before)
while 1:
    print(oP.read())
4

2 に答える 2

1

コードをいじってみたところ、問題はファイルの読み取り方法にあると思われます。

具体的には、次のような行です。

os.read(self.iFH, iLineStartBaseLength)

その呼び出しは必ずしもiLineStartBaseLengthバイトを返すとは限りません - を消費"LI"し、戻っREADLINE_FAILて再試行する可能性があります。2 回目の試行で、行の残りを取得し、何らかの方法で数値以外の文字列をint()呼び出しに渡すことになります。

予測できないのは、FIFO がフラッシュされる方法にある可能性があります。完全な行が書き込まれたときにフラッシュが発生した場合は、すべて問題ありません。行が中途半端な時にフラッシュしたらおかしい。

少なくとも、私が最終的にハッキングしたバージョンのスクリプトでは、oP.read()呼び出しはprocess_2.py送信されたものとは異なる口述を取得することがよくありました (以前の奇妙なKEYことに出血する可能性があります)。VALUE

OS X でコードを実行するために多くの変更を加えなければならなかったので、私は間違っているかもしれません。私の変更されたコードはこちら

正確な修正方法はわかりませんが、..jsonモジュールなどを使用すると、プロトコル/解析を大幅に簡素化できます-改行で区切られたJSONデータは解析がはるかに簡単です:

import os
import time
import json
import errno


def retry_write(*args, **kwargs):
    """Like os.write, but retries until EAGAIN stops appearing
    """

    while True:
        try:
            return os.write(*args, **kwargs)
        except OSError as e:
            if e.errno == errno.EAGAIN:
                time.sleep(0.5)
            else:
                raise


class Pipe(object):
    """FIFO based IPC based on newline-separated JSON
    """

    ENCODING = 'utf-8'

    def __init__(self,sPath):
        self.sPath = sPath
        if not os.path.exists(sPath):
            os.mkfifo(sPath)

        self.fd = os.open(sPath,os.O_RDWR | os.O_NONBLOCK)
        self.file_blocking = open(sPath, "r", encoding=self.ENCODING)

    def write(self, dmsg):
        serialised = json.dumps(dmsg) + "\n"
        dat = bytes(serialised.encode(self.ENCODING))

        # This blocks until data can be read by other process.
        # Can just use os.write and ignore EAGAIN if you want
        # to drop the data
        retry_write(self.fd, dat)

    def read(self):
        serialised = self.file_blocking.readline()
        return json.loads(serialised)
于 2012-10-30T12:27:34.507 に答える
-1

try:、ブロックを取り除き、except:実際にスローされている例外を確認してください。

したがって、サンプルを次のように置き換えます。

iLineLen = int(sLineLen.strip(string.punctuation+string.whitespace))

これで がスローされるに違いありValueErrorません。それは、「KE 17」を にキャストしようとしているからですint。

文字列string.whitespaceを.string.punctuationint

于 2012-10-29T15:12:33.413 に答える