python通过管道进行进程通信：竞争条件

Posted 2023-02-16

【中文标题】python通过管道进行进程通信：竞争条件

【英文标题】：python process communications via pipes: Race condition

【发布时间】：2012-10-29 15:04:11

【问题描述】：

所以我有两个 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"

它是间歇性的事实告诉我这是由于某种比赛条件，我只是在努力弄清楚它可能是什么。有什么想法吗？

EDIT添加了有关调用进程的内容

Pipe 的使用方式是在 processA 和 ProcessB 中通过调用具有相同路径的构造函数来实例化它。然后进程 A 将间歇性地写入管道，进程 B 将尝试从中读取。在任何时候，我都不会试图让这件事成为一种双向的方式。

这是对这种情况的更冗长的解释。我一直试图让这个问题简短，但我认为现在是我放弃的时候了。 Anyhoo，我有一个守护进程和一个 Pyramid 进程需要玩得很好。有两个 Pipe 实例正在使用：一个只有 Pyramid 写入，另一个只有守护进程写入。 Pyramid 写的东西很短，我在这个管道上没有遇到任何错误。守护进程写的东西要长得多，这是让我伤心的管道。两个管道都以相同的方式实现。两个进程只将字典写入各自的管道（如果不是这种情况，那么 Pipe.write 中就会出现异常）。

基本算法是：Pyramid 产生守护进程，守护进程加载疯狂的厄运对象层次结构和巨大的 ram 消耗。 Pyramid 将 POST 请求发送到守护进程，然后守护进程进行大量计算并将数据发送到 Pyramid，以便呈现人性化的页面。然后，人类可以通过填写 html 表单等来响应层次结构中的内容，从而导致金字塔向守护进程发送另一个字典，并且守护进程发送回字典响应。

所以：只有一个管道出现任何问题，问题管道的流量比另一个管道多得多，并且保证只有字典写入其中一个

编辑作为对问题和评论的回应

在你告诉我试一试之前……除了继续阅读的内容。 异常被提出的事实让我感到困扰。 iLineLengh = int(stuff) 在我看来应该总是传递一个看起来像整数的字符串。这种情况只是大多数时候，而不是全部。因此，如果您想评论它可能不是整数，请不要。

套用我的问题：找出比赛条件，你将成为我的英雄。

编辑一个小例子：

process_1.py：

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

process_2.py：

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

【问题讨论】：

很明显引发了异常，因为"KE 17" 无法转换为整数——你知道为什么一个进程会这样写吗？

@mgilson：天哪，真的吗？我稍微编辑了我的问题，也许现在会更有意义。我很确定这是竞争条件的结果，因为它是间歇性的并且总是在同一个地方

当它失败时，是在你最初运行.read()-calling 进程时发生的吗？或者它可以在运行中途发生吗？拥有代码的完整版本也会很有用，发布的代码包含大部分代码，但缺少Pipe.message_start（readln 上的缩进也丢失了）

这真是一大堆代码……所有的常量都是同名的字符串。这只是防止愚蠢的错别字的一种方法......我现在正在处理缩进。它并不总是发生在第一次运行时，有时会发生，有时会发生在完全不同的运行中。没有我可以检测到的模式

@dbr：我进行了编辑，解释了错误的上下文以及我对此的看法。如果您对代码有任何具体问题，我将非常愿意发布所需的任何信息，只是发布全部内容将是完全不切实际的，我想不出除了我之外还有什么相关的已经发布了

【参考方案1】：

在玩弄了代码之后，我怀疑问题出在您读取文件的方式上。

具体来说，像这样的行：

os.read(self.iFH, iLineStartBaseLength)

该调用不一定返回 iLineStartBaseLength 字节 - 它可能会消耗 "LI" ，然后返回 READLINE_FAIL 并重试。在第二次尝试时，它将获取该行的其余部分，并以某种方式最终将非数字字符串提供给 int() 调用

不可预测性可能来自于 fifo 的刷新方式 - 如果在写入完整行时它恰好刷新，一切都很好。如果在行写一半时它会刷新，那就奇怪了。

至少在我最终得到的脚本的破解版本中，process_2.py 中的oP.read() 调用通常与发送的那个不同（KEY 可能会渗入之前的@987654330 @ 和其他奇怪的东西）。

我可能弄错了，因为我必须进行大量更改才能使代码在 OS X 上运行，并在进行试验时进一步进行。 My modified code here

不确定如何修复它，但是.. 使用 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)

【讨论】：

它仍然存在......：/我现在正在向管道类添加一个锁定机制（使用 mkdir 和 oserror 17 来制作一个简单的信号量）开始表现。不过感谢您的帮助，您的回答和解释很有帮助和有趣

【参考方案2】：

尝试摆脱 try:、except: 块并查看实际抛出的异常。

因此，只需将您的示例替换为：

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

我敢打赌它现在会抛出 ValueError，这是因为您试图将“KE 17”转换为 int。

如果您要将字符串转换为 int，则需要剥离 string.whitespace 和 string.punctuation 以上。

【讨论】：

是的，我知道。 try..except 的原因是我可以看到导致错误的字符串。否则异常不会提供太多信息

@Sheena -- 原始异常确实会告诉您失败的字符串，但如果您坚持自己处理异常，至少只捕获 ValueError 并重新引发 ValueError.. .

@mgilson：这真的不是重点。这是一个竞赛条件。请相信我。如果我编辑代码以满足您的规范，那么它不会改变存在间歇性错误的事实。这个问题的主题是错误的间歇性，而不是异常中的文本。我的逻辑是，给出错误的那一行应该总是输入一个看起来像整数的字符串。事实并非如此。

@Sheena - 如果这是重点，我会发布它作为答案。至于它是一个竞争条件——你可能是对的，但你的帖子不排除它可能是其他东西的可能性，因为我们不知道你实际上在写什么这些管道，或者你的程序如何决定写什么（你如何保证你的管道每次运行程序时都应该写同样的东西？

@mgilson：该进程将标准 python 字典写入管道。如果不是这种情况，就会有一个非常不同的例外。而且我可以保证，如果我一遍又一遍地写入（在进程 1）和读取（在进程 2）完全相同的字典，那么它最终会出错。请通读代码，如果有任何需要澄清的地方，我将非常乐意提供详细信息。请具体说明您需要哪些信息。