远程调试Python进程的小工具

原文链接：http://whosemario.github.io/2015/12/01/python-remote-debug-py/

起因

我们的游戏进程是用Python实现的，有时候为了调试一些游戏逻辑，我们不得不打印很多的log，并且有时候少打印了一些log还要补上这些log后重新走一遍游戏逻辑，甚是麻烦。

Python提供了pdb模块，是否可以随时调试一个Python进程的某段代码呢？

Python Pdb模块

我们先从Python的Pdb模块入手，如下是Pdb的构造函数。

class Pdb(bdb.Bdb, cmd.Cmd):

def __init__(self, completekey='tab', stdin=None, stdout=None, skip=None):
bdb.Bdb.__init__(self, skip=skip)
cmd.Cmd.__init__(self, completekey, stdin, stdout)

Pdb的构造函数会传入stdin和stdout，如果为None则是真标准输入输出。好了，我传入一个其他的文件描述符这不就可以将pdb的调试信息重定向了嘛，pdb模块完全可扩展，不需要改pdb了。

def runcall(*args, **kwds):
return Pdb().runcall(*args, **kwds)

Pdb模块有runcall方法（其实runcall的实现是在bdb模块里面的），也好了，我们可以利用这个方法来实现debug一个函数了。

进程之间通信

我的目标是在同一台机器上实现两个进程之间的通信，服务端进程就是我们游戏的Game进程，客户端是一个Python程序。这里我使用了两个FIFO（有名管道）来实现了进程间的通信，参考了reference1。

#-*- coding:utf-8 -*-

"""\
使用FIFO模拟一个双工管道用于两个进程之间的通信
"""

import os
import tempfile

__all__ = ["NamePipe"]

class NamePipe(object):

def __init__(self, pid, is_client = True, mode = 0666):
super(NamePipe, self).__init__()

name = self._get_pipe_name(pid)
self.in_name = name + ".in"
self.out_name = name + ".out"

try:
os.mkfifo(self.in_name, mode)
os.chmod(self.in_name, mode)
except OSError: pass
try :
os.mkfifo(self.out_name, mode)
os.chmod(self.out_name, mode)
except OSError: pass

self.is_client = is_client

if is_client:
# client
self.in_fd = open(self.in_name, "r")
self.out_fd = open(self.out_name, "w")
else:
# server
self.out_fd = open(self.in_name, "w")
self.in_fd = open(self.out_name, "r")

def write(self, msg):
if self.is_open():
self.out_fd.write("%d\n" % len(msg))
self.out_fd.write(msg)
self.out_fd.flush()
return True
else:
return False

def read(self):
if self.is_open():
sz = self.in_fd.readline()
if sz:
return self.in_fd.read(int(sz))
else:
return ""
else:
return None

def flush(self):
pass

def readline(self):
return self.read()

def is_open(self):
#return True
return not (self.in_fd.closed or self.out_fd.closed)

def close(self):
"""  只尝试unlink掉读端
"""
if self.is_client:
try: os.remove(self.in_name)
except OSError: pass
else:
try: os.remove(self.out_name)
except OSError: pass
self.in_fd.close()
self.out_fd.close()

def _get_pipe_name(self, pid):
return os.path.join(tempfile.gettempdir(), "pipe-%d" % pid)

Debug修饰器

针对要Debug得函数，我们可以使用修饰器进行修饰。

def remote_debug(func):
def wrapper(*args, **kwargs):
import pdb
if _pipe:
try:
pdb.Pdb(stdin = _pipe, stdout = _pipe).runcall(func, *args, **kwargs)
except IOError:
_pipe.close()
_pipe = None
else:
return func(*args, **kwargs)
return wrapper

建立通信

客户端通过SIGUSR2信号来通知服务端建立通信。

# server
_pipe = None

def remote_connect(sig, frame):
if _pipe:
_pipe.close()
_pipe = NamePipe.NamePipe(os.getpid(), False)

def reg_listener():
import signal
signal.signal(signal.SIGUSR1, remote_connect)

-------------------------------------------------------

# clent
#-*- coding:utf-8 -*-

import signal
import os
import sys
import NamePipe
import pdb

Prefix = pdb.Pdb().prompt

pid = int(sys.argv[1])

os.kill(pid, signal.SIGUSR1)

pipe = NamePipe.NamePipe(pid, True)

while True:

while True:
txt = pipe.read()
if txt:
sys.stdout.write("%s" % txt)
sys.stdout.flush()
if txt.startswith(Prefix):
break

txt = raw_input("")
pipe.write(txt)

reference:

Debugging a running python process - http://code.activestate.com/recipes/576515/