修复flup中threadpool在twisted中运行

 

修复flup中threadpool在twisted中运行的日记

作者: 日期:

gashero

2009-09-04

目录

• 1   简介
• 2   故障描述
• 3   调试流程
  • 3.1   flup的源码安装
  • 3.2   确定导入的flup模块范围
• 4   FastCGI服务器启动跟踪
  • 4.1   外部入口
  • 4.2   flup.server.fcgi 中的初始化
  • 4.3   server.run() 流程
  • 4.4   threadedserver 模块分析
  • 4.5   fcgi_base 中对thread和threading模块的引用分析
  • 4.6   threadpool 结构分析
• 5   修改尝试
  • 5.1   修改threadpool模块中线程启动方式
  • 5.2   修改threadpool的实现，改为twisted的线程
  • 5.3   修改 threadedserver.ThreadedServer.run ，使得加入reactor的启动
• 6   总结

1   简介

希望在twistd托管模式中运行flup的FastCGI服务器，但是启动以后一直有可以接受连接，但是不接受请求的问题。本文记录修复该问题的过程以及方法。

2   故障描述

twistd托管daemon中的flup可以接受连接，但是不接受请求。

3   调试流程

3.1   flup的源码安装

默认使用egg打包方式放在标准Python模块目录，安装后的看不到源码。所以删除了已经安装的flup，而将解压后的flup源码目录拷贝到Python模块目录了。方便进一步调试。

3.2   确定导入的flup模块范围

只有先确定了导入模块的范围才能更好的定位问题点。而Python的 sys.modules 字典对象展示了所有导入对象。

如下两句在阻塞启动flup服务器之前提供打印所有导入模块的功能:

print 'modules',sys.modules.keys()
print 'flup mod',filter(lambda x:'flup' in x,sys.modules.keys())

如此打印出来的模块还是有诸多干扰，整理一下发现如下flup的模块被导入了:

flup
flup.server
flup.server.flup
flup.server.fcgi
flup.server.fcgi_base
flup.server.threadpool
flup.server.threadedserver

再排除了 flup.server.flup 和顶级没用模块以后目标定位在4个模块中:

flup.server.fcgi
flup.server.fcgi_base
flup.server.threadpool
flup.server.threadedserver

好吧，噩梦来了，以前开发的CDN系统就曾经发现过在twistd的daemon模式中使用threadpool无法启动线程的问题。恐怕这次的主要问题点也在这里。这4个模块总代码量为1635行，还好，还不至于很抓狂。

4   FastCGI服务器启动跟踪

4.1   外部入口

from flup.server import fcgi
server=fcgi.WSGIServer(wsgifunc,bindAddress=('0.0.0.0',8080))
server.run()

4.2   flup.server.fcgi 中的初始化

from flup.server.fcgi_base import BaseFCGIServer,FCGI_RESONDER
from flup.server.threadedserver import ThreadedServer

其中 WSGIServer 继承自 fcgi_base.BaseFCGIServer 和 flup.server.threadedserver.ThreadServer 。其 __init__() 构造函数调用了两个父类的构造函数， BaseFCGIServer 的在前。

4.3   server.run() 流程

调用了 flup.server.fcgi.WSGIServer.run() 函数，其定义中的阻塞部分在:

sock=self._setupSocket()
ret=ThreadedServer.run(self,sock)   #这句阻塞
self._cleanupSocket(sock)
return ret

4.4   threadedserver 模块分析

该模块用于处理多线程服务器的线程部分，而不管FastCGI的处理。也是唯一一处调用了 threadpool 模块的地方。共计175行。

这里对threadpool模块的引用只有3处：

1. 45行，导入模块:

   from flup.server.threadpool import ThreadPool
   

2. 54行，初始化时，定义为服务器的线程池对象:

   self._threadPool=ThreadPool(**kw)
   

3. 97行，生成任务，添加任务到线程池:

   conn=self._jobClass(clientSock,addr,*self._jobArgs)
   if not self._threadPool.addJob(conn,allowQueuing=False):
       clientSock.close()
   

由最后一处调用可见不是几年前郁闷我的那个threadpool模块了，不过相信问题是一样的，就是twistd托管daemon模式时，线程的运行是有问题的。

4.5   fcgi_base 中对thread和threading模块的引用分析

大部分地方是使用了线程锁。而使用了启动线程的地方有一处，MultiplexedConnection类的865行，启动线程:

def _start_request(self,req):
    thread.start_new_thread(req.run,())

实际调试发现如上代码的启动中根本没有创建该类的对象，所以可以排除这里出问题的可能。

至于其他地方使用的锁，就先假设不会出问题吧。

4.6   threadpool 结构分析

该模块一共121行，是flup的作者自己写的，其中只有一个类 ThreadPool ，其包含3个方法:

class ThreadPool(object):
    def __init__(self,minSpare=1,maxSpare=5,maxThreads=sys.maxint)
    def addJob(self,job,allQueuing=True)
    def _worker(self)

虽然引用了thread和threading两个模块，不过任务都比较明晰。启动线程使用 thread.start_new_thread(self._worker,()) ，而threading模块仅用于定义锁 self._lock=threading.Condition() 。

5   修改尝试

5.1   修改threadpool模块中线程启动方式

以前是使用 thread.start_new_thread 方式启动，尝试修改成threading模块的线程对象方式，以及其守护线程模式:

thrd=threading.Thread(target=self._worker)
thrd.setDaemon(True)
thrd.start()

尝试失败。启动的线程仍然在twistd托管时消失了。

5.2   修改threadpool的实现，改为twisted的线程

使用twisted的线程池的相同接口ThreadPool实现:

class ThreadPool(object):
    """Twisted ThreadPool warpper"""

    def __init__(self,minSpare=1,maxSpare=5,maxThreads=sys.maxint):
        #reactor.suggestThreadPoolSize(maxSpare)
        return

    def addJob(self,job,allowQueuing=True):
        print 'call addjob',repr(job)
        reactor.callInThread(job.run)
        return
#threadpool.ThreadPool=ThreadPool
threadedserver.ThreadPool=ThreadPool

重新尝试启动，因为没有阻塞调用 reactor.run() 的地方，所以线程启动失败。

5.3   修改 threadedserver.ThreadedServer.run ，使得加入reactor的启动

这个主要就是利用twisted的思想，将整个函数切块，涉及到信号操作的部分使用 reactor.callFromThread() 调用，其他部分则使用 reactor.callInThread() 放在线程里执行就可以了。

我为了避免污染flup在site-packages目录中的代码，所以是自己在模块里面写代码，然后导入flup以后替换相关对象的方式实现的。这时将原来函数的代码复制过来，做修改，然后替换。同时需要另外导入一些模块，导入与替换:

import select
import errno
from flup.server.threadedserver import setCloseOnExec
threadedserver.ThreadedServer.run=threadedserver_run

通过这样改造以后 server.run() 就可以由twisted的线程进行托管了。启动代码如下:

server=fcgi.WSGIServer(wsgifunc,bindAddress=(addr,port))
reactor.callInThread(server.run)
application=service.Application('httprpc',uid=uid,gid=gid)

6   总结

通过如上一番对flup的改造，就可以将其运行在twisted托管模式了。由twistd负责捕捉错误输出、打印输出，并且使其运行在daemon模式。很方便。

这样修改以后还存在一点不足，就是在普通运行模式下也要用 reactor.callInThread(server.run) 来运行。这种情况下就没法使用Ctrl+C来终止程序了。而必须使用kill -9才行。