远程调试Python进程的小工具

發表於   |   分類於   |  

起因

我们的游戏进程是用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:

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