10

背景

我有两个需要相互通信的 python 进程。通信由一个名为 Pipe 的类处理。我为此创建了一个单独的类，因为需要传达的大部分信息都以字典的形式出现，因此 Pipe 实现了一个非常简单的协议来执行此操作。

这是管道构造函数：

def __init__(self,sPath):
    """
    create the fifo. if it already exists just associate with it
    """
    self.sPath = sPath
    if not os.path.exists(sPath):
        try:
            os.mkfifo(sPath)
        except:
            raise Exception('cannot mkfifo at path \n {0}'.format(sPath))
    self.iFH = os.open(sPath,os.O_RDWR | os.O_NONBLOCK)
    self.iFHBlocking = os.open(sPath,os.O_RDWR)

所以理想情况下，我会在每个进程中构建一个具有相同路径的管道，他们就能很好地交谈。

我将跳过有关协议的内容，因为我认为这在很大程度上是不必要的。

所有读取和写入操作都使用以下“基本”函数：

def base_read_blocking(self,iLen):
    self.lock()
    lBytes = os.read(self.iFHBlocking,iLen)
    self.unlock()
    return lBytes

def base_read(self,iLen):
    print('entering base read')
    self.lock()
    lBytes = os.read(self.iFH,iLen)
    self.unlock()
    print('exiting base read')
    return lBytes

def base_write_blocking(self,lBytes):
    self.lock()
    safe_write(self.iFHBlocking,lBytes)
    self.unlock()

def base_write(self,lBytes):
    print('entering base write')
    self.lock()
    safe_write(self.iFH,lBytes)
    self.unlock()
    print('exiting base write')

在另一篇文章中建议使用 safe_write

def safe_write(*args, **kwargs):
    while True:
        try:
            return os.write(*args, **kwargs)
        except OSError as e:
            if e.errno == 35:
                import time
                print(".")
                time.sleep(0.5)
            else:
                raise

锁定和解锁是这样处理的：

def lock(self):
    print('locking...')
    while True:
        try:
            os.mkdir(self.get_lock_dir())
            print('...locked')
            return
        except OSError as e:
            if e.errno != 17:
                raise e

def unlock(self):
    try:
        os.rmdir(self.get_lock_dir())
    except OSError as e:
        if e.errno != 2:
            raise e
    print('unlocked')

问题

这有时会发生：

....in base_read
lBytes = os.read(self.iFH,iLen)
OSError: [Errno 11] Resource temporarily unavailable

有时这很好。

神奇的解决方案

我似乎已经阻止了问题的发生。请注意，这不是我在回答我自己的问题。我的问题将在下一节中解释。

我将读取函数更改为看起来更像这样，它整理了一些东西：

def base_read(self,iLen):
    while not self.ready_for_reading():
        import time
        print('.')
        time.sleep(0.5)

    lBytes = ''.encode('utf-8')
    while len(lBytes)<iLen:
        self.lock()
        try:
            lBytes += os.read(self.iFH,iLen)
        except OSError as e:
            if e.errno == 11:
                import time
                print('.')
                time.sleep(0.5)
        finally:
            self.unlock()
        return lBytes


def ready_for_reading(self):
    lR,lW,lX = select.select([self.iFH,],[],[],self.iTimeout)
    if not lR:
        return False
    lR,lW,lX = select.select([self.iFHBlocking],[],[],self.iTimeout)
    if not lR:
        return False
    return True

问题

我正在努力找出它暂时不可用的确切原因。由于锁定机制，这两个进程无法同时访问实际的命名管道（除非我弄错了？）所以这是由于我的程序没有考虑到 fifos 更基本的东西吗？

我真正想要的只是一个解释......我找到的解决方案有效，但它看起来很神奇。任何人都可以提供解释吗？

系统

  • Ubuntu 12.04，
  • Python3.2.3
4

1 回答 1

3

我之前在Java 上也遇到过类似的问题。看看接受的答案——问题是我在循环中创建新线程。我建议您查看创建管道的代码，并确保您没有创建多个管道。

于 2012-11-12T09:53:18.840 回答