我遇到了一个非常奇怪的问题，当我处理从多个进程(从子进程模块开始)到文件的打印时，我无法解释。我看到的行为是我的某些输出被略微截断了，而某些则完全丢失了。我正在使用Alext Martelli的解决方案的经过稍微修改的版本，用于在这里How do I get a thread safe print in Python 2.6?找到线程安全打印。主要区别在于写方法。为了确保在写入同一文件的多个进程之间不会交错输出，我对输出进行缓冲，仅在看到换行符时才写入。
import sys
import threading
tls = threading.local()
class ThreadSafeFile(object):
"""
@author: Alex Martelli
@see: https://stackoverflow.com/questions/3029816/how-do-i-get-a-thread-safe-print-in-python-2-6
@summary: Allows for safe printing of output of multi-threaded programs to stdout.
"""
def __init__(self, f):
self.f = f
self.lock = threading.RLock()
self.nesting = 0
self.dataBuffer = ""
def _getlock(self):
self.lock.acquire()
self.nesting += 1
def _droplock(self):
nesting = self.nesting
self.nesting = 0
for i in range(nesting):
self.lock.release()
def __getattr__(self, name):
if name == 'softspace':
return tls.softspace
else:
raise AttributeError(name)
def __setattr__(self, name, value):
if name == 'softspace':
tls.softspace = value
else:
return object.__setattr__(self, name, value)
def write(self, data):
self._getlock()
self.dataBuffer += data
if data == '\n':
self.f.write(self.dataBuffer)
self.f.flush()
self.dataBuffer = ""
self._droplock()
def flush(self):
self.f.flush()
还应该注意的是，要使其异常运行，将需要大量时间或具有多个处理器或内核的机器。在报告失败之前，我在单处理器计算机上在测试套件中运行了该违规程序约7000次。我创建的用于演示我在测试套件中遇到的问题的程序似乎也可以在单处理器计算机上运行，但是在多核或多处理器计算机上执行该程序肯定会失败。
以下程序显示了该问题，它比我希望的要复杂得多，但我想尽可能保留程序的行为。
进程1 main.py的代码
import subprocess, sys, socket, time, random
from threadSafeFile import ThreadSafeFile
sys.stdout = ThreadSafeFile(sys.__stdout__)
usage = "python main.py nprocs niters"
workerFilename = "/path/to/worker.py"
def startMaster(n, iters):
host = socket.gethostname()
for i in xrange(n):
#set up ~synchronization between master and worker
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind((host,0))
sock.listen(1)
socketPort = sock.getsockname()[1]
cmd = 'ssh %s python %s %s %d %d %d' % \
(host, workerFilename, host, socketPort, i, iters)
proc = subprocess.Popen(cmd.split(), shell=False, stdout=None, stderr=None)
conn, addr = sock.accept()
#wait for worker process to start
conn.recv(1024)
for j in xrange(iters):
#do very bursty i/o
for k in xrange(iters):
print "master: %d iter: %d message: %d" % (n,i, j)
#sleep for some amount of time between .02s and .5s
time.sleep(1 * (random.randint(1,50) / float(100)))
#wait for worker to finish
conn.recv(1024)
sock.close()
proc.kill()
def main(nprocs, niters):
startMaster(nprocs, niters)
if __name__ == "__main__":
if len(sys.argv) != 3:
print usage
sys.exit(1)
nprocs = int(sys.argv[1])
niters = int(sys.argv[2])
main(nprocs, niters)
进程2的代码worker.py
import sys, socket,time, random, time
from threadSafeFile import ThreadSafeFile
usage = "python host port id iters"
sys.stdout = ThreadSafeFile(sys.__stdout__)
def main(host, port, n, iters):
#tell master to start
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((host, port))
sock.send("begin")
for i in xrange(iters):
#do bursty i/o
for j in xrange(iters):
print "worker: %d iter: %d message: %d" % (n,i, j)
#sleep for some amount of time between .02s and .5s
time.sleep(1 * (random.randint(1,50) / float(100)))
#tell master we are done
sock.send("done")
sock.close()
if __name__ == "__main__":
if len(sys.argv) != 5:
print usage
sys.exit(1)
host = sys.argv[1]
port = int(sys.argv[2])
n = int(sys.argv[3])
iters = int(sys.argv[4])
main(host,port,n,iters)
测试时，我按如下方式运行main.py:
python main.py 1 75 > main.out
结果文件的长度应为75 * 75 * 2 = 11250行，其格式为:
(母版|工作人员):％d个迭代:％d个消息:％d个
在大多数情况下，它短20至30行，但是我偶尔看到该程序具有适当的行数。在进一步研究了罕见的成功之后，其中的一些行被截断了，例如:
ter:％d消息:％d
另一个有趣的方面是，当使用多处理而不是子处理启动ssh进程时，此程序的行为符合预期。有人可能会说，为什么在多处理正常的情况下还要使用子流程呢？不幸的是，是我自己的学者真正想知道为什么它表现异常。任何想法和/或见解将不胜感激。谢谢。
***编辑
本，我知道threadSafeFile在每个进程中使用不同的锁，但是出于两个原因，我在较大的项目中需要使用它。
1)每个进程可能有多个线程，这些线程将被写入stdout，即使此示例没有这样做。因此，我需要保证线程级别和进程级别的安全。
2)如果我不确定在刷新stdout时缓冲区的末尾有一个'\n'，那么将存在一些潜在的执行跟踪，其中进程1将其缓冲区写入文件而没有尾随'\n'，然后进程2进入并写入其缓冲区。现在我们有线交错了，那不是我想要的。
我也了解到，这种机制使打印内容有些局限。现在，在我这个项目的开发阶段，限制是可以的。当我可以保证正确性时，我可以开始放宽限制。
您关于在data =='\n'内锁定条件检查内部的评论不正确。如果锁进入条件检查的内部，则通常情况下threadSafeFile不再是线程安全的。如果有任何线程可以添加到数据缓冲区，则dataBuffer + = data处将存在竞争条件，因为这不是原子操作。也许您的评论与该示例有关，在该示例中，每个进程只有1个线程，但是如果是这种情况，那么我们甚至根本不需要锁。
关于操作系统级别的锁定，我的理解是，如果要写入的字节数小于内部缓冲区的大小，则多个程序能够安全地在UNIX平台上写入同一文件。在这种情况下，操作系统是否不应该为我处理所有必要的锁定？
请您参考如下方法:
在每个过程中，您都为ThreadSafeFile创建一个sys.stdout，每个都有一个锁，但是它们是不同的锁；在所有不同过程中使用的锁都没有连接在一起。因此，您将获得与根本不使用任何锁相同的效果。因为所有进程都有自己的锁，所以永远不会被另一个进程中的锁阻塞。
在单处理器计算机上运行时，此方法唯一可行的原因是您要进行缓冲以使写入排队直到遇到换行符。这意味着输出的每一行都一次性写入。在单处理器上，操作系统不太可能决定在一系列连续调用write的过程中切换进程，这会浪费您的数据。但是，如果输出全部写在一行的大块中，而您并不关心行在文件中的结束顺序，那么上下文切换就不太可能在您关心的操作过程中发生关于。不过，从理论上讲，这并不是不可能的，所以即使对于单处理器，我也不会将此代码称为正确的代码。ThreadSafeFile非常特别是仅线程安全。它依赖于以下事实:程序对于要写入的每个文件只有一个ThreadSafeFile对象。因此，对该文件的任何写入都将通过该单个对象，并在锁上同步。
当您有多个进程时，您将没有单个进程中的线程共享的全局内存。因此，每个进程必定有其自己的单独的ThreadSafeFile(sys.stdout)对象。这与使用线程并生成N个线程(每个线程都创建了自己的ThreadSafeFile(sys.stdout))是完全相同的错误。
我不知道在使用多处理时这是如何工作的，因为您尚未发布用于执行此操作的代码。但是我的理解是，出于所有相同的原因，如果您以这样的方式使用多重处理，即每个进程都创建自己的新ThreadSafeFile，那么这样做仍然会失败。也许您不是在使用多处理的版本中这样做？
您需要做的是安排以某种方式连接同步对象(锁)。多处理模块可以为您完成此任务。注意在the example here中如何创建一次锁，然后在创建新锁时将其传递给每个新进程。 (当然，这仍然会在10个不同的过程中产生10个不同的锁对象，但是Python在后台必须做的是创建OS级锁，然后使每个复制的Python级锁对象都引用单个OS-级别锁定)。
如果要通过子处理来实现此目的，那么您只是从单独的脚本开始完全独立的工作程序命令，那么您将需要某种方式来使它们全部与单个操作系统级别的锁进行对话。我不知道标准库中有什么可以帮助您做到这一点的。我只会使用多重处理。
另一个想法是，您的缓冲和锁定代码也看起来有点可疑。如果有人调用sys.stdout.write("foo\n")会发生什么？我不确定，但是只能猜测这是有效的，因为print的实现碰巧在您要打印的内容上调用sys.stdout.write，然后使用一个换行符再次调用它。绝对没有必要这样做!它可以很容易地在内存中组合单个输出字符串，然后只调用sys.stdout.write一次。另外，如果您需要打印需要在输出中一起显示的多行块，会发生什么情况？
另一个问题是，您在第一次将进程写入缓冲区时获取了锁，在缓冲区被填满时继续持有该锁，然后写入该行，最后释放锁。如果您的锁确实有效，并且进程在开始一行与结束一行之间花费了很长时间，那么它将阻止所有其他进程甚至缓冲其写入!也许这就是您想要的，如果这样的意图是，当一个进程开始写东西时，它可以保证其输出将在下一个文件中命中。但是在那种情况下，您甚至根本不需要缓冲。我认为您应该在if data == '\n':之后获取锁，然后您也不需要跟踪嵌套级别的所有代码。
