



阅读gunicorn代码 v0.0.1 文档

### Navigation

* [阅读gunicorn代码 v0.0.1 文档](index.html#document-index) »

阅读gunicorn代码文档[¶](#gunicorn "Permalink to this headline")

=========================================================

说明[¶](#id1 "Permalink to this headline")

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

因为在玩heroku app时，heroku给出的教程中，采用了gunicorn来部署应用；在南昌echo提出用freebsd的kqueue改造flup时，我建议可以对比一下gunicorn。但自己仅仅是在heroku上小用了一把而已，并没有深入研究，为了此次南昌 pythoner 2012年第一次聚会，特意浏览了一下gunicorn的代码，感觉上gunicorn代码写得比较小巧精炼，整理出来，算是抛砖引玉。

文档内容[¶](#id2 "Permalink to this headline")

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

### 介绍[¶](#id1 "Permalink to this headline")

#### 什么是gunicorn[¶](#gunicorn "Permalink to this headline")

gunicorn，是“Green Unicorn”，脱胎于ruby社区的Unicorn，移植到python上，成为一个WSGI HTTP Server，WSGI是”Web Server Gateway Interface”，是python的web接口规范。

#### gunicorn的特性[¶](#id2 "Permalink to this headline")

* 支持 Django，paster，wsgi程序

* 非常容易配置（相比较而言）

* 自动管理多个worker进程

* 可以采用不同的后台扩展接口（sync, gevent, tornado等）

### 设计[¶](#id1 "Permalink to this headline")

这部分是采用gunicorn的英文文档，同时加入自己的理解，源代码，以及参考资料等等

#### 服务器模型[¶](#id2 "Permalink to this headline")

gunicorn是基于”pre-fork worker”模型，这就意味着有一个中心主控master进程，由它来管理一组worker进程。这个主控进程并不知晓任何客户端，所有的请求和响应都完全是由多个worker进程来处理

##### 解释pre-fork[¶](#pre-fork "Permalink to this headline")

pre-fork服务器和fork服务器相似，通过一个单独的进程来处理每条请求。

不同的是，pre-fork服务器会通过预先开启大量的进程，等待并处理接到的请求。

由于采用了这种方式来开启进程，服务器并不需要等待新的进程启动而消耗时间，因而能够以更快的速度应付多用户请求。

另外，pre-fork服务器在遇到极大的高峰负载时仍能保持良好的性能状态。这是因为不管什么时候，只要预先设定的所有进程都已被用来处理请求时，服务器仍可追加额外的进程。

缺点是，当遇到高峰负载时，由于要启动新的服务器进程，不可避免地会带来响应的延迟。

#### 主控master进程[¶](#master "Permalink to this headline")

主控master进程，就是一个简单的循环，用来不断侦听不同进程信号并作出不同的动作，仅此而已。它通过一些信号，诸如TTIN, TTOU, 和CHLD等等， 管理着那些正在运行的worker进程。

TTIN 和 TTOU信号是告诉主控master进程增加或减少正在运行的worker数量。

CHLD信号是在一个子进程已经中止之后，由主控master进程重启这个失效的worker进程。

我们看几段代码，在gunicorn/arbiter.py中

|  |  |

| --- | --- |

| 

```

 1

 2

 3

 4

 5

 6

 7

 8

 9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

```

 | 

```

class Arbiter(object):

    """

 Arbiter maintain the workers processes alive. It launches or

 kills them if needed. It also manages application reloading

 via SIGHUP/USR2.

 """

    # A flag indicating if a worker failed to

    # to boot. If a worker process exist with

    # this error code, the arbiter will terminate.

    WORKER\_BOOT\_ERROR = 3

    START\_CTX = {}

    LISTENER = None

    WORKERS = {}

    PIPE = []

    # I love dynamic languages

    SIG\_QUEUE = []

    SIGNALS = map(

        lambda x: getattr(signal, "SIG%s" % x),

        "HUP QUIT INT TERM TTIN TTOU USR1 USR2 WINCH".split()

    )

    SIG\_NAMES = dict(

        (getattr(signal, name), name[3:].lower()) for name in dir(signal)

        if name[:3] == "SIG" and name[3] != "\_"

    )

    def \_\_init\_\_(self, app):

        os.environ["SERVER\_SOFTWARE"] = SERVER\_SOFTWARE

        self.setup(app)

        self.pidfile = None

        self.worker\_age = 0

        self.reexec\_pid = 0

        self.master\_name = "Master"

```

 |

在上面代码中的第23行中，列举了相应的多个信号：

* HUP，重启所有的配置和所有的worker进程

* QUIT，正常关闭，它会等待所有worker进程处理完各自的东西后关闭

* INT/TERM，立即关闭，强行中止所有的处理

* TTIN，增加一个worker进程

* TTOU，减少一个worker进程

* USR1，重新打开由master和worker所有的日志处理

* USR2，重新运行master和worker

* WINCH，正常关闭所有worker进程，保持主控master进程的运行

下面是针对不同信号的代码:

|  |  |

| --- | --- |

| 

```

 1

 2

 3

 4

 5

 6

 7

 8

 9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

```

 | 

```

def handle\_chld(self, sig, frame):

    "SIGCHLD handling"

    self.wakeup()

def handle\_hup(self):

    """\

 HUP handling.

 - Reload configuration

 - Start the new worker processes with a new configuration

 - Gracefully shutdown the old worker processes

 """

    self.log.info("Hang up: %s", self.master\_name)

    self.reload()

def handle\_quit(self):

    "SIGQUIT handling"

    raise StopIteration

def handle\_int(self):

    "SIGINT handling"

    self.stop(False)

    raise StopIteration

def handle\_term(self):

    "SIGTERM handling"

    self.stop(False)

    raise StopIteration

def handle\_ttin(self):

    """\

 SIGTTIN handling.

 Increases the number of workers by one.

 """

    self.num\_workers += 1

    self.manage\_workers()

def handle\_ttou(self):

    """\

 SIGTTOU handling.

 Decreases the number of workers by one.

 """

    if self.num\_workers <= 1:

        return

    self.num\_workers -= 1

    self.manage\_workers()

def handle\_usr1(self):

    """\

 SIGUSR1 handling.

 Kill all workers by sending them a SIGUSR1

 """

    self.kill\_workers(signal.SIGUSR1)

    self.log.reopen\_files()

def handle\_usr2(self):

    """\

 SIGUSR2 handling.

 Creates a new master/worker set as a slave of the current

 master without affecting old workers. Use this to do live

 deployment with the ability to backout a change.

 """

    self.reexec()

def handle\_winch(self):

    "SIGWINCH handling"

    if os.getppid() == 1 or os.getpgrp() != os.getpid():

        self.log.info("graceful stop of workers")

        self.num\_workers = 0

        self.kill\_workers(signal.SIGQUIT)

    else:

        self.log.info("SIGWINCH ignored. Not daemonized")

```

 |

#### 同步workers[¶](#workers "Permalink to this headline")

大多数情况下，采用的worker类型是同步方式，也就是说一次仅处理一个请求。这种模型方式是最简单的，因为期间发生的任何错误最多只影响到一个请求。

尽管下面我们描述的内容，也是一次处理一个请求，但实际上在编写应用的时候，是加一些条件的。（？？）

#### 异步workers[¶](#id3 "Permalink to this headline")

可用的异步workers，主要是基于greenlets软件包（通过eventlet和gevent）。greenlet是用python来实现的协程方式（cooperative multi-threading）。通常情况下，我们编写的应用代码不需要作出什么改变，就能使用上这些异步workers的特性的。

#### tornado workers[¶](#tornado-workers "Permalink to this headline")

还有一个可以用上的worker，是tornado worker，它是用在那些采用tornado框架的程序上。尽管tornado worker也可以用于wsgi程序上，但是这不是一个推荐的做法。

#### 选择worker进程类型[¶](#worker "Permalink to this headline")

默认的同步worker在cpu和带宽方面会消耗资源的。这就意味着你的应用不可能无节制地做任何事。例如，互联网上的一个请求，必须要遵守这个准则。

这个资源绑定的条件，也就是为什么我们会在gunicorn默认配置前做一个缓冲的代理。如果你把同步worker，直接暴露在internet上，一个dos(Denial of Service)攻击就会给服务器不停地制造大流量无用数据，而是服务器无法正常提供服务。Slowloris，就是这样一个有趣的例子，专门用来做这事的。

有些情况可以考虑采用异步worker：

* 需要长时间阻塞调用的应用，比如外部的web service

* 直接给internet提供服务

* 流请求和响应（是类似flv流么？）

* 长轮询

* Web sockets（WebSocket是HTML5规格中的一个非常重要的新特性，它的存在可以允许用户在浏览器中实现双向通信，实现数据的及时推送）

* Comet（基于 HTTP 长连接的“服务器推”技术，是一种新的 Web 应用架构。基于这种架构开发的应用中，服务器端会主动以异步的方式向客户端程序推送数据，而不需要客户端显式的发出请求。Comet 架构非常适合事件驱动的 Web 应用，以及对交互性和实时性要求很强的应用）

#### 启动多少个workers?[¶](#id4 "Permalink to this headline")

不要试图做这样的事，你预期多少个客户端就启用多少个worker。gunicorn只需要启用4–12个workers，就足以每秒钟处理几百甚至上千个请求了。

在处理请求时，gunicorn依靠操作系统来提供负载均衡。通常我们推荐的worker数量是：(2 x $num\_cores) + 1，这个公式很简单，它是基于给定的核心处理器数量，在其他worker处理请求时，每个worker将从socket那进行读写操作。

很显然，你的硬件环境和应用将影响到worker数量。我们推荐先采用上述公式来安排，在应用启动之后，然后再通过TTIN和TTOU这两个信号来调整worker数量。

记住：太多的worker，肯定会在某一个时刻，让你的整个系统急剧降低性能。（只能意译了）

### 快速入门[¶](#id1 "Permalink to this headline")

#### 安装gunicorn[¶](#gunicorn "Permalink to this headline")

下载源代码:

```

git clone https://github.com/benoitc/gunicorn.git

```

使用开发模式来安装，便于今后调试:

```

python setup.py develop

```

这样安装后，会在pythone的site-packages中仅仅增加一个链接，在自己目录下的源代码可以随时进行调整测试

#### 简单的wsgi应用[¶](#wsgi "Permalink to this headline")

随便编写一个myapp.py

|  |  |

| --- | --- |

| 

```

1

2

3

4

5

6

7

8

9

```

 | 

```

# -\*- coding: utf8 -\*-

def app(environ, start\_response):

    data = "Hello, World!\n"

    start\_response("200 OK", [

        ("Content-Type", "text/plain"),

        ("Content-Length", str(len(data)))

    ])

    return iter([data])

```

 |

运行之:

```

$ gunicorn --workers=2 myapp:app

```

运行的结果:

```

2012-02-13 10:48:02 [2481] [INFO] Starting gunicorn 0.13.4

2012-02-13 10:48:02 [2481] [INFO] Listening at: http://127.0.0.1:8000 (2481)

2012-02-13 10:48:02 [2481] [INFO] Using worker: sync

2012-02-13 10:48:02 [2484] [INFO] Booting worker with pid: 2484

2012-02-13 10:48:02 [2485] [INFO] Booting worker with pid: 2485

```

#### 简单的django应用[¶](#django "Permalink to this headline")

快速体验:

```

$ django-admin.py startproject hello

$ cd hello

$ gunicorn_django --workers=2

```

运行结果:

```

2012-02-13 11:11:15 [2565] [INFO] Starting gunicorn 0.13.4

2012-02-13 11:11:15 [2565] [INFO] Listening at: http://127.0.0.1:8000 (2565)

2012-02-13 11:11:15 [2565] [INFO] Using worker: sync

2012-02-13 11:11:15 [2568] [INFO] Booting worker with pid: 2568

2012-02-13 11:11:15 [2569] [INFO] Booting worker with pid: 2569

```

#### 命令行参数[¶](#id2 "Permalink to this headline")

执行gunicorn:

```

(mypy)hzg@gofast:~/gunicorn$ gunicorn -h

Usage: gunicorn [OPTIONS] APP_MODULE

Options:

  --version             show program's version number and exit

  -h, --help            show this help message and exit

  -c FILE, --config=FILE

                        The path to a Gunicorn config file. [None]

  --debug               Turn on debugging in the server. [False]

  --spew                Install a trace function that spews every line

                        executed by the server. [False]

  --access-logfile=FILE

                        The Access log file to write to. [None]

  --access-logformat=STRING

                        The Access log format . [%(h)s %(l)s %(u)s %(t)s

                        "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s"]

  --error-logfile=FILE, --log-file=FILE

                        The Error log file to write to. [-]

  --log-level=LEVEL     The granularity of Error log outputs. [info]

  --logger-class=STRING

                        The logger you want to use to log events in gunicorn.

                        [simple]

  -n STRING, --name=STRING

                        A base to use with setproctitle for process naming.

                        [None]

  --preload             Load application code before the worker processes are

                        forked. [False]

  -D, --daemon          Daemonize the Gunicorn process. [False]

  -p FILE, --pid=FILE   A filename to use for the PID file. [None]

  -u USER, --user=USER  Switch worker processes to run as this user. [1000]

  -g GROUP, --group=GROUP

                        Switch worker process to run as this group. [1000]

  -m INT, --umask=INT   A bit mask for the file mode on files written by

                        Gunicorn. [0]

  -b ADDRESS, --bind=ADDRESS

                        The socket to bind. [127.0.0.1:8000]

  --backlog=INT         The maximum number of pending connections.     [2048]

  -w INT, --workers=INT

                        The number of worker process for handling requests.

                        [1]

  -k STRING, --worker-class=STRING

                        The type of workers to use. [sync]

  --worker-connections=INT

                        The maximum number of simultaneous clients. [1000]

  --max-requests=INT    The maximum number of requests a worker will process

                        before restarting. [0]

  -t INT, --timeout=INT

                        Workers silent for more than this many seconds are

                        killed and restarted. [30]

  --keep-alive=INT      The number of seconds to wait for requests on a Keep-

                        Alive connection. [2]

```

### 从哪开始阅读源代码[¶](#id1 "Permalink to this headline")

#### 代码树[¶](#id2 "Permalink to this headline")

执行命令:

```

tree -I '\*.pyc'

```

先看看gunicorn的目录树:

```

gunicorn

├── app

│   ├── base.py

│   ├── djangoapp.py

│   ├── __init__.py

│   ├── pasterapp.py

│   └── wsgiapp.py

├── arbiter.py

├── config.py

├── debug.py

├── errors.py

├── glogging.py

├── http

│   ├── body.py

│   ├── errors.py

│   ├── __init__.py

│   ├── message.py

│   ├── parser.py

│   ├── _sendfile.py

│   ├── unreader.py

│   └── wsgi.py

├── __init__.py

├── logging_config.py

├── management

│   ├── commands

│   │   ├── __init__.py

│   │   └── run_gunicorn.py

│   └── __init__.py

├── pidfile.py

├── sock.py

├── util.py

└── workers

    ├── async.py

    ├── base.py

    ├── geventlet.py

    ├── ggevent.py

    ├── ggevent_wsgi.py

    ├── gtornado.py

    ├── __init__.py

    ├── sync.py

    └── workertmp.py

```

#### 从脚本命令开始[¶](#id3 "Permalink to this headline")

就从脚本命令gunicorn和gunicorn\_django开始

gunicorn

```

#!/home/hzg/mypy/bin/python

# EASY-INSTALL-ENTRY-SCRIPT: 'gunicorn==0.13.4','console\_scripts','gunicorn'

\_\_requires\_\_ = 'gunicorn==0.13.4'

import sys

from pkg\_resources import load\_entry\_point

sys.exit(

   load\_entry\_point('gunicorn==0.13.4', 'console\_scripts', 'gunicorn')()

)

```

gunicorn\_django

```

#!/home/hzg/mypy/bin/python

# EASY-INSTALL-ENTRY-SCRIPT: 'gunicorn==0.13.4','console\_scripts','gunicorn\_django'

\_\_requires\_\_ = 'gunicorn==0.13.4'

import sys

from pkg\_resources import load\_entry\_point

sys.exit(

   load\_entry\_point('gunicorn==0.13.4', 'console\_scripts', 'gunicorn\_django')()

)

```

这是eggs文件的规范，实际上调用入口，可以从entry\_points.txt看出来：

entry\_points.txt:

```

[console\_scripts]

gunicorn=gunicorn.app.wsgiapp:run

gunicorn\_django=gunicorn.app.djangoapp:run

gunicorn\_paster=gunicorn.app.pasterapp:run

```

注意在[console\_scripts]部分，有

* gunicorn <–> gunicorn.app.wsgiapp:run

* gunicorn\_django <–> gunicorn.app.djangoapp:run

那么，我们就从gunicorn/app目录下开始，从wsgiapp.py, djangoapp.py开始。

##### wsgiapp.py[¶](#wsgiapp-py "Permalink to this headline")

|  |  |

| --- | --- |

| 

```

 1

 2

 3

 4

 5

 6

 7

 8

 9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

```

 | 

```

import os

import sys

from gunicorn import util

from gunicorn.app.base import Application

class WSGIApplication(Application):

    def init(self, parser, opts, args):

        if len(args) != 1:

            parser.error("No application module specified.")

        self.cfg.set("default\_proc\_name", args[0])

        self.app\_uri = args[0]

        sys.path.insert(0, os.getcwd())

    def load(self):

        return util.import\_app(self.app\_uri)

def run():

    """\

 The ``gunicorn`` command line runner for launcing Gunicorn with

 generic WSGI applications.

 """

    from gunicorn.app.wsgiapp import WSGIApplication

    WSGIApplication("%prog [OPTIONS] APP\_MODULE").run()

```

 |

gunicorn.app.wsgiapp:run最终是调用其中的一个函数run()，焦点集中在上述代码的最后两行，创建WSGIApplication后，立刻执行run()。

##### djangoapp.py[¶](#djangoapp-py "Permalink to this headline")

|  |  |

| --- | --- |

| 

```

 1

 2

 3

 4

 5

 6

 7

 8

 9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

```

 | 

```

ENVIRONMENT\_VARIABLE = 'DJANGO\_SETTINGS\_MODULE'

class DjangoApplication(Application):

    def init(self, parser, opts, args):

        self.global\_settings\_path = None

        self.project\_path = None

        if args:

            self.global\_settings\_path = args[0]

            if not os.path.exists(os.path.abspath(args[0])):

                self.no\_settings(args[0])

    def get\_settings\_modname(self):

        from django.conf import ENVIRONMENT\_VARIABLE

        # get settings module

        settings\_modname = None

        if not self.global\_settings\_path:

            project\_path = os.getcwd()

        # ......

    def setup\_environ(self, settings\_modname):

        from django.core.management import setup\_environ

        # setup environ

        # ......

    def no\_settings(self, path, import\_error=False):

        if import\_error:

            error = "Error: Can't find '%s' in your PYTHONPATH.\n" % path

        # .......

    def activate\_translation(self):

        from django.conf import settings

        from django.utils import translation

        translation.activate(settings.LANGUAGE\_CODE)

    def validate(self):

        """ Validate models. This also ensures that all models are

 imported in case of import-time side effects."""

        from django.core.management.base import CommandError

        from django.core.management.validation import get\_validation\_errors

        # ......

    def load(self):

        from django.core.handlers.wsgi import WSGIHandler

        self.setup\_environ(self.get\_settings\_modname())

        self.validate()

        self.activate\_translation()

        return WSGIHandler()

# ......

def run():

  """\

 The ``gunicorn\_django`` command line runner for launching Django

 applications.

 """

  from gunicorn.app.djangoapp import DjangoApplication

  DjangoApplication("%prog [OPTIONS] [SETTINGS\_PATH]").run()

```

 |

gunicorn.app.djangoapp:run最终是调用其中的一个函数run()，焦点集中在上述代码的最后两行，创建DjangoApplication后，立刻执行run()。

尽管DjangoApplication定义了不少方法，但注意与WSGIAppliction的骨架一样，实际上就定义了两个方法:

* init()

* load()

最关键的方法run()，估计就是从父类Application里继承而来。

### 深入Application[¶](#application "Permalink to this headline")

回想一下上节关注的WSGIApplication和DjangoApplication，都有一个共同的父类Application，并且都调用了run()方法。

#### 关注内部[¶](#id1 "Permalink to this headline")

Application的代码不长，下面来看看整个情况:

|  |  |

| --- | --- |

| 

```

  1

  2

  3

  4

  5

  6

  7

  8

  9

 10

 11

 12

 13

 14

 15

 16

 17

 18

 19

 20

 21

 22

 23

 24

 25

 26

 27

 28

 29

 30

 31

 32

 33

 34

 35

 36

 37

 38

 39

 40

 41

 42

 43

 44

 45

 46

 47

 48

 49

 50

 51

 52

 53

 54

 55

 56

 57

 58

 59

 60

 61

 62

 63

 64

 65

 66

 67

 68

 69

 70

 71

 72

 73

 74

 75

 76

 77

 78

 79

 80

 81

 82

 83

 84

 85

 86

 87

 88

 89

 90

 91

 92

 93

 94

 95

 96

 97

 98

 99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

```

 | 

```

import errno

import os

import sys

import traceback

from gunicorn.glogging import Logger

from gunicorn import util

from gunicorn.arbiter import Arbiter

from gunicorn.config import Config

from gunicorn import debug

class Application(object):

    """\

 An application interface for configuring and loading

 the various necessities for any given web framework.

 """

    def \_\_init\_\_(self, usage=None):

        self.usage = usage

        self.cfg = None

        self.callable = None

        self.logger = None

        self.do\_load\_config()

    def do\_load\_config(self):

        try:

            self.load\_config()

        except Exception, e:

            sys.stderr.write("\nError: %s\n" % str(e))

            sys.stderr.flush()

            sys.exit(1)

    def load\_config(self):

        # init configuration

        self.cfg = Config(self.usage)

        # parse console args

        parser = self.cfg.parser()

        opts, args = parser.parse\_args()

        # optional settings from apps

        cfg = self.init(parser, opts, args)

        # Load up the any app specific configuration

        if cfg and cfg is not None:

            for k, v in cfg.items():

                self.cfg.set(k.lower(), v)

        # Load up the config file if its found.

        if opts.config and os.path.exists(opts.config):

            cfg = {

                "\_\_builtins\_\_": \_\_builtins\_\_,

                "\_\_name\_\_": "\_\_config\_\_",

                "\_\_file\_\_": opts.config,

                "\_\_doc\_\_": None,

                "\_\_package\_\_": None

            }

            try:

                execfile(opts.config, cfg, cfg)

            except Exception:

                print "Failed to read config file: %s" % opts.config

                traceback.print\_exc()

                sys.exit(1)

            for k, v in cfg.items():

                # Ignore unknown names

                if k not in self.cfg.settings:

                    continue

                try:

                    self.cfg.set(k.lower(), v)

                except:

                    sys.stderr.write("Invalid value for %s: %s\n\n" % (k, v))

                    raise

        # Lastly, update the configuration with any command line

        # settings.

        for k, v in opts.\_\_dict\_\_.items():

            if v is None:

                continue

            self.cfg.set(k.lower(), v)

    def init(self, parser, opts, args):

        raise NotImplementedError

    def load(self):

        raise NotImplementedError

    def reload(self):

        self.do\_load\_config()

        if self.cfg.spew:

            debug.spew()

    def wsgi(self):

        if self.callable is None:

            self.callable = self.load()

        return self.callable

    def run(self):

        if self.cfg.spew:

            debug.spew()

        if self.cfg.daemon:

            util.daemonize()

        else:

            try:

                os.setpgrp()

            except OSError, e:

                if e[0] != errno.EPERM:

                    raise

        try:

            Arbiter(self).run()

        except RuntimeError, e:

            sys.stderr.write("\nError: %s\n\n" % e)

            sys.stderr.flush()

            sys.exit(1)

```

 |

注意其中几个地方:

1. init()和load()的定义方式

2. run()的核心内容

从WSGIApplication与DjangoApplication继承自Application，并且从init()和load()的定义看，这里采用了一个最常见的设计模式: Template Method:

```

在父类一级定义好一系列的方法，作为算法的骨架结构，由不同的子类来实现不同的具体功能。

```

落实在Application中，我们可以看到算法的骨架结构就在\_\_init\_\_()和run()中，由它们来调用init()和load()两个空的方法，一旦有具体实例WSGIApplication和DjangoApplication中有实现，就调用具体实例的init()和load()。

#### 代码流程[¶](#id2 "Permalink to this headline")

1. 初始化流程

> 

> 

> 

> `__init__`()[¶](#__init__ "Permalink to this definition")

> 

> 

> 

> `do_load_config`()[¶](#do_load_config "Permalink to this definition")

> 

> 

> 

> `load_config`()[¶](#load_config "Permalink to this definition")

> 

> 

> 

> `init`(*parser*, *opts*, *args*)[¶](#init "Permalink to this definition")

> 

> 

2. run运行流程

> 

> 

> 

> `run`()[¶](#run "Permalink to this definition")

> 

> 

> 

> `Arbiter`(*self).run(*)[¶](#Arbiter "Permalink to this definition")

> 

> 

> 

> `wsgi`()[¶](#wsgi "Permalink to this definition")

> 

> 

> 

> `load`()[¶](#load "Permalink to this definition")

> 

> 

### 配置参数[¶](#id1 "Permalink to this headline")

初始化流程主要解决的是加载配置参数或文件，重点部分在load\_config()方法中。

配置参数的处理主要是在config.py文件中，有三个主要的类:

* Config

* Setting

* SettingMeta

先挑有趣的代码开始阅读。

```

class Setting(object):

  \_\_metaclass\_\_ = SettingMeta

class SettingMeta(type):

  def \_\_new\_\_(cls, name, bases, attrs):

      super\_new = super(SettingMeta, cls).\_\_new\_\_

```

#### 元类编程[¶](#id2 "Permalink to this headline")

SettingMeta和Setting继承的方式，是完全不同的，尽管他们都是用class来定义。

\_\_metaclass\_\_ = SettingMeta

metaclass是元类，它是一个类的类（型）。相对于类可以在运行时动态构造对象而言，元类也可以在运行时动态生成类。

##### 解释type[¶](#type "Permalink to this headline")

示例:

```

>>> type('abc')

<type 'str'>

>>> type('abc')('def')

'def'

>>> type(type(1))

<type 'type'>

>>> type(type(1))(1)

<type 'int'>

>>> Calculator = type('Calculator', (), {'add': lambda self, x, y: x+y, 'sub': lambda sef, x, y: x-y })

>>> calc = Calculator()

>>> type(calc)

<class '\_\_main\_\_.Calculator'>

>>>

>>> print calc.add(1, 2)

3

>>> print calc.sub(1, 2)

-1

```

type实际上是接收3个参数，第一个参数是类名，第二个是父类（由于允许多重继承，所以是个元组，空元组表示父类为object），第三个参数为类的成员字典。它会返回一个新风格的type对象，这个对象实际上就是一个动态生成的类。

##### 跟踪子类[¶](#id3 "Permalink to this headline")

下面的内容是摘自pro django一书中的内容，是自己在2009-03-31翻译的，放在博客上，从中可以很清楚的知道元类的使用方式。

考虑一个应用，在任何时候，访问一个特定类的所有子类列表。metaclass是一个非常好的处理手段，但是存在一个问题。记住，每一个带有\_\_metaclass\_\_属性的类都要处理，包括新的基类，他们是不需要被注册的（只有它的子类要被注册）。

要处理好这个问题，就要作些额外的处理，但这样作也是很直接了当的，同时也是很有益处的。

示例:

```

>>> class SubclassTracker(type):

...     def \_\_init\_\_(cls, name, bases, attrs):

...         try:

...             if TrackedClass not in bases:

...                 return

...         except NameError:

...             return

...         TrackedClass.\_registry.append(cls)

...

>>> class TrackedClass(object):

...     \_\_metaclass\_\_ = SubclassTracker

...     \_registry = []

...

>>> class ClassOne(TrackedClass):

...     pass

...

>>> TrackedClass.\_registry

[<class '\_\_main\_\_.ClassOne'>]

>>> class ClassTwo(TrackedClass):

...     pass

...

>>> TrackedClass.\_registry

[<class '\_\_main\_\_.ClassOne'>, <class '\_\_main\_\_.ClassTwo'>]

```

这个metaclass执行了两个功能。首先，try块确保父类，TrackedClass，已经定义好了。如果没有的话，就抛出NameError异常， 这个过程就表明metaclass当前正处理TrackedClass。TrackedClass那还能处理更多的东西，但是这个例子为了简单，忽略掉了，只要通过注册就行了。

...

所有TrackedClass的子类能在任何时候从注册表中提取。 TrackedClass的任何子类都将出现在这个注册表中，不管子类在哪里定义的。执行这个类定义的过程就开始注册它，应用程序能导入任何有这些类和 metaclass的模块。

#### gunicorn配置怎么处理[¶](#gunicorn "Permalink to this headline")

Setting、SettingMeta与TrackedClass、SubclassTracker主体上是一致的

|  |  |

| --- | --- |

| 

```

 1

 2

 3

 4

 5

 6

 7

 8

 9

10

11

12

13

14

15

16

17

18

19

```

 | 

```

class SettingMeta(type):

    def \_\_new\_\_(cls, name, bases, attrs):

        super\_new = super(SettingMeta, cls).\_\_new\_\_

        parents = [b for b in bases if isinstance(b, SettingMeta)]

        if not parents:

            return super\_new(cls, name, bases, attrs)

        attrs["order"] = len(KNOWN\_SETTINGS)

        attrs["validator"] = wrap\_method(attrs["validator"])

        new\_class = super\_new(cls, name, bases, attrs)

        new\_class.fmt\_desc(attrs.get("desc", ""))

        KNOWN\_SETTINGS.append(new\_class)

        return new\_class

    def fmt\_desc(cls, desc):

        desc = textwrap.dedent(desc).strip()

        setattr(cls, "desc", desc)

        setattr(cls, "short", desc.splitlines()[0])

```

 |

|  |  |

| --- | --- |

| 

```

 1

 2

 3

 4

 5

 6

 7

 8

 9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

```

 | 

```

class Setting(object):

    \_\_metaclass\_\_ = SettingMeta

    name = None

    value = None

    section = None

    cli = None

    validator = None

    type = None

    meta = None

    action = None

    default = None

    short = None

    desc = None

    def \_\_init\_\_(self):

        if self.default is not None:

            self.set(self.default)

    def add\_option(self, parser):

        if not self.cli:

            return

        args = tuple(self.cli)

        kwargs = {

            "dest": self.name,

            "metavar": self.meta or None,

            "action": self.action or "store",

            "type": self.type or "string",

            "default": None,

            "help": "%s [%s]" % (self.short, self.default)

        }

        if kwargs["action"] != "store":

            kwargs.pop("type")

        parser.add\_option(\*args, \*\*kwargs)

    def copy(self):

        return copy.copy(self)

    def get(self):

        return self.value

    def set(self, val):

        assert callable(self.validator), "Invalid validator: %s" % self.name

        self.value = self.validator(val)

```

 |

|  |  |

| --- | --- |

| 

```

 1

 2

 3

 4

 5

 6

 7

 8

 9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

```

 | 

```

class Bind(Setting):

    name = "bind"

    section = "Server Socket"

    cli = ["-b", "--bind"]

    meta = "ADDRESS"

    validator = validate\_string

    default = "127.0.0.1:8000"

    desc = """\

 The socket to bind.

 A string of the form: 'HOST', 'HOST:PORT', 'unix:PATH'. An IP is a valid

 HOST.

 """

class Workers(Setting):

    name = "workers"

    section = "Worker Processes"

    cli = ["-w", "--workers"]

    meta = "INT"

    validator = validate\_pos\_int

    type = "int"

    default = 1

    desc = """\

 The number of worker process for handling requests.

 A positive integer generally in the 2-4 x $(NUM\_CORES) range. You'll

 want to vary this a bit to find the best for your particular

 application's work load.

 """

```

 |

注意在SettingMeta中的KNOWN\_SETTINGS.append(new\_class)，和TrackedClass.\_registry，Bind、Workers在声明时，就已经执行SettingMeta的\_\_new\_\_()，留意\_\_new\_\_()与\_\_init\_\_()的区别

```

>>> from gunicorn import config

>>> b = config.Bind()

>>> b.default

'127.0.0.1:8000'

>>> b.cli

['-b', '--bind']

>>> b.desc

"The socket to bind.\n\nA string of the form: 'HOST', 'HOST:PORT', 'unix:PATH'. An IP is a valid\nHOST."

>>> b.name

'bind'

>>> b.value

'127.0.0.1:8000'

>>> b.validator

<bound method Bind.\_wrapped of <gunicorn.config.Bind object at 0x8eb4eec>>

>>> b.order

1

```

SettingMeta中第8-9行，动态生成的有：validator()方法与order属性，在导入config时就已完成。

|  |  |

| --- | --- |

| 

```

1

2

3

4

```

 | 

```

def wrap\_method(func):

    def \_wrapped(instance, \*args, \*\*kwargs):

        return func(\*args, \*\*kwargs)

    return \_wrapped

```

 |

这里使用了python decorator的特性，wrap\_method(‘validate\_string’)通过这样处理后，把validate\_string替换成新的名称validator。

```

>>> from gunicorn import config

>>> c = config.Config()

>>> c.address

('127.0.0.1', 8000)

>>> c.workers

1

>>> c.worker\_class

<class 'gunicorn.workers.sync.SyncWorker'>

```

由Config中的parser方法，来解析命令行的参数

### 真正的主宰者[¶](#id1 "Permalink to this headline")

#### 循环主体[¶](#id2 "Permalink to this headline")

通过上面的抽丝拨茧，进入到Arbiter类，这个才是真正的主宰者。在前面涉及gunicorn设计一节中，已经讲到了部分Arbiter的信号处理，这里将详细介绍run()方法。

|  |  |

| --- | --- |

| 

```

  1

  2

  3

  4

  5

  6

  7

  8

  9

 10

 11

 12

 13

 14

 15

 16

 17

 18

 19

 20

 21

 22

 23

 24

 25

 26

 27

 28

 29

 30

 31

 32

 33

 34

 35

 36

 37

 38

 39

 40

 41

 42

 43

 44

 45

 46

 47

 48

 49

 50

 51

 52

 53

 54

 55

 56

 57

 58

 59

 60

 61

 62

 63

 64

 65

 66

 67

 68

 69

 70

 71

 72

 73

 74

 75

 76

 77

 78

 79

 80

 81

 82

 83

 84

 85

 86

 87

 88

 89

 90

 91

 92

 93

 94

 95

 96

 97

 98

 99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

```

 | 

```

class Arbiter(object):

    """

 Arbiter maintain the workers processes alive. It launches or

 kills them if needed. It also manages application reloading

 via SIGHUP/USR2.

 """

    # A flag indicating if a worker failed to

    # to boot. If a worker process exist with

    # this error code, the arbiter will terminate.

    WORKER\_BOOT\_ERROR = 3

    START\_CTX = {}

    LISTENER = None

    WORKERS = {}

    PIPE = []

    # I love dynamic languages

    SIG\_QUEUE = []

    SIGNALS = map(

        lambda x: getattr(signal, "SIG%s" % x),

        "HUP QUIT INT TERM TTIN TTOU USR1 USR2 WINCH".split()

    )

    SIG\_NAMES = dict(

        (getattr(signal, name), name[3:].lower()) for name in dir(signal)

        if name[:3] == "SIG" and name[3] != "\_"

    )

    def \_\_init\_\_(self, app):

        os.environ["SERVER\_SOFTWARE"] = SERVER\_SOFTWARE

        self.setup(app)

        self.pidfile = None

        self.worker\_age = 0

        self.reexec\_pid = 0

        self.master\_name = "Master"

        # get current path, try to use PWD env first

        try:

            a = os.stat(os.environ['PWD'])

            b = os.stat(os.getcwd())

            if a.ino == b.ino and a.dev == b.dev:

                cwd = os.environ['PWD']

            else:

                cwd = os.getcwd()

        except:

            cwd = os.getcwd()

        args = sys.argv[:]

        args.insert(0, sys.executable)

        # init start context

        self.START\_CTX = {

            "args": args,

            "cwd": cwd,

            0: sys.executable

        }

    def setup(self, app):

        self.app = app

        self.cfg = app.cfg

        self.log = self.cfg.logger\_class(app.cfg)

        if 'GUNICORN\_FD' in os.environ:

            self.log.reopen\_files()

        self.address = self.cfg.address

        self.num\_workers = self.cfg.workers

        self.debug = self.cfg.debug

        self.timeout = self.cfg.timeout

        self.proc\_name = self.cfg.proc\_name

        self.worker\_class = self.cfg.worker\_class

        if self.cfg.debug:

            self.log.debug("Current configuration:")

            for config, value in sorted(self.cfg.settings.iteritems()):

                self.log.debug(" %s: %s", config, value.value)

        if self.cfg.preload\_app:

            if not self.cfg.debug:

                self.app.wsgi()

            else:

                self.log.warning("debug mode: app isn't preloaded.")

    def start(self):

        """\

 Initialize the arbiter. Start listening and set pidfile if needed.

 """

        self.log.info("Starting gunicorn %s", \_\_version\_\_)

        self.cfg.on\_starting(self)

        self.pid = os.getpid()

        self.init\_signals()

        if not self.LISTENER:

            self.LISTENER = create\_socket(self.cfg, self.log)

        if self.cfg.pidfile is not None:

            self.pidfile = Pidfile(self.cfg.pidfile)

            self.pidfile.create(self.pid)

        self.log.debug("Arbiter booted")

        self.log.info("Listening at: %s (%s)", self.LISTENER,

            self.pid)

        self.log.info("Using worker: %s",

                self.cfg.settings['worker\_class'].get())

        self.cfg.when\_ready(self)

    def init\_signals(self):

        """\

 Initialize master signal handling. Most of the signals

 are queued. Child signals only wake up the master.

 """

        if self.PIPE:

            map(os.close, self.PIPE)

        self.PIPE = pair = os.pipe()

        map(util.set\_non\_blocking, pair)

        map(util.close\_on\_exec, pair)

        self.log.close\_on\_exec()

        map(lambda s: signal.signal(s, self.signal), self.SIGNALS)

        signal.signal(signal.SIGCHLD, self.handle\_chld)

    def signal(self, sig, frame):

        if len(self.SIG\_QUEUE) < 5:

            self.SIG\_QUEUE.append(sig)

            self.wakeup()

    def run(self):

        "Main master loop."

        self.start()

        util.\_setproctitle("master [%s]" % self.proc\_name)

        self.manage\_workers()

        while True:

            try:

                self.reap\_workers()

                sig = self.SIG\_QUEUE.pop(0) if len(self.SIG\_QUEUE) else None

                if sig is None:

                    self.sleep()

                    self.murder\_workers()

                    self.manage\_workers()

                    continue

                if sig not in self.SIG\_NAMES:

                    self.log.info("Ignoring unknown signal: %s", sig)

                    continue

                signame = self.SIG\_NAMES.get(sig)

                handler = getattr(self, "handle\_%s" % signame, None)

                if not handler:

                    self.log.error("Unhandled signal: %s", signame)

                    continue

                self.log.info("Handling signal: %s", signame)

                handler()

                self.wakeup()

            except StopIteration:

                self.halt()

            except KeyboardInterrupt:

                self.halt()

            except HaltServer, inst:

                self.halt(reason=inst.reason, exit\_status=inst.exit\_status)

            except SystemExit:

                raise

            except Exception:

                self.log.info("Unhandled exception in main loop:\n%s",

                            traceback.format\_exc())

                self.stop(False)

                if self.pidfile is not None:

                    self.pidfile.unlink()

                sys.exit(-1)

```

 |

正如在设计一节中说到的，主控master进程，就是一个简单的循环，用来不断侦听各种信号，然后作出不同的动作。

* 130行，start()，用来创建一个侦听用的socket，并创建一个pidfile（就是进程号文件），放在/tmp下

* 133行，manage\_workers()

* 134行，正式进入循环体

#### 核心语句[¶](#id3 "Permalink to this headline")

那下面我们感兴趣的地方，自然是落在那些信号的初始化上，仔细进入start()里面看看，start()中主要的语句在:

```

self.pid = os.getpid()

self.init\_signals()

if not self.LISTENER:

    self.LISTENER = create\_socket(self.cfg, self.log)

if self.cfg.pidfile is not None:

    self.pidfile = Pidfile(self.cfg.pidfile)

    self.pidfile.create(self.pid)

```

self.init\_signals():

```

#对预设的各种信号进行map处理

map(lambda s: signal.signal(s, self.signal), self.SIGNALS)

signal.signal(signal.SIGCHLD, self.handle\_chld)

```

各种信号就是Arbiter的类变量:

```

SIGNALS = map(

      lambda x: getattr(signal, "SIG%s" % x),

      "HUP QUIT INT TERM TTIN TTOU USR1 USR2 WINCH".split()

  )

```

这里运用了python有趣的动态语言特性，lambda，map，那这些信号map之后，到哪去了呢？注意看self.signal()

```

def signal(self, sig, frame):

    if len(self.SIG\_QUEUE) < 5:

        self.SIG\_QUEUE.append(sig)

        self.wakeup()

```

把所有的信号放进一个信号列表，然后执行wakeup()。

进入run()中的循环体之后，真正处理信号的就是handle()，而这个对应的是各种信号的处理handle\_xxx()。真正管理用户请求和响应的，并不是在arbiter中处理，而是交给了worker。所以注意看manage\_worker()和spawn\_worker()

manage\_worker()代码

|  |  |

| --- | --- |

| 

```

 1

 2

 3

 4

 5

 6

 7

 8

 9

10

11

12

13

```

 | 

```

def manage\_workers(self):

        """\

 Maintain the number of workers by spawning or killing

 as required.

 """

        if len(self.WORKERS.keys()) < self.num\_workers:

            self.spawn\_workers()

        workers = self.WORKERS.items()

        workers.sort(key=lambda w: w[1].age)

        while len(workers) > self.num\_workers:

            (pid, \_) = workers.pop(0)

            self.kill\_worker(pid, signal.SIGQUIT)

```

 |

根据worker数量，如果小于配置中的数量，就生成一个worker进程，否则中断worker进程

spawn\_worker()代码

|  |  |

| --- | --- |

| 

```

 1

 2

 3

 4

 5

 6

 7

 8

 9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

```

 | 

```

def spawn\_worker(self):

    self.worker\_age += 1

    worker = self.worker\_class(self.worker\_age, self.pid, self.LISTENER,

                                self.app, self.timeout/2.0,

                                self.cfg, self.log)

    self.cfg.pre\_fork(self, worker)

    pid = os.fork()

    if pid != 0:

        self.WORKERS[pid] = worker

        return pid

    # Process Child

    worker\_pid = os.getpid()

    try:

        util.\_setproctitle("worker [%s]" % self.proc\_name)

        self.log.info("Booting worker with pid: %s", worker\_pid)

        self.cfg.post\_fork(self, worker)

        worker.init\_process()

        sys.exit(0)

    except SystemExit:

        raise

    except:

        self.log.exception("Exception in worker process:")

        if not worker.booted:

            sys.exit(self.WORKER\_BOOT\_ERROR)

        sys.exit(-1)

    finally:

        self.log.info("Worker exiting (pid: %s)", worker\_pid)

        try:

            worker.tmp.close()

            self.cfg.worker\_exit(self, worker)

        except:

            pass

```

 |

其中:

```

worker = self.worker\_class(self.worker\_age, self.pid, self.LISTENER,

                                self.app, self.timeout/2.0,

                                self.cfg, self.log)

......

worker.init\_process()

```

由此进入worker枢纽环节，将在后续介绍之。

self.cfg.pre\_fork(self, worker)是server hook之一。代码在哪呢？在config.py中

```

class Prefork(Setting):

    name = "pre\_fork"

    section = "Server Hooks"

    validator = validate\_callable(2)

    type = "callable"

    def pre\_fork(server, worker):

        pass

    default = staticmethod(pre\_fork)

    desc = """\

 Called just before a worker is forked.

 The callable needs to accept two instance variables for the Arbiter and

 new Worker.

 """

```

可以在自定义的配置文件中定义。之后的self.cfg.post\_fork(self, worker)，self.cfg.worker\_exit(self, worker)都是在config.py中作为server hook:

```

pid = os.fork()

worker\_pid = os.getpid()

```

这是我们熟知的进程处理方式。

### worker积木[¶](#worker "Permalink to this headline")

之所以称之为\*积木\*，意思就是它可以随时进行替换，也是gunicorn让人称道的地方。

#### 同步worker[¶](#id1 "Permalink to this headline")

默认的worker class，是同步的方式，也就是一次只能处理一个请求，且必须等待它完成后返回。对应的class SyncWorker(base.Worker)，还是进入到源代码看看吧

|  |  |

| --- | --- |

| 

```

  1

  2

  3

  4

  5

  6

  7

  8

  9

 10

 11

 12

 13

 14

 15

 16

 17

 18

 19

 20

 21

 22

 23

 24

 25

 26

 27

 28

 29

 30

 31

 32

 33

 34

 35

 36

 37

 38

 39

 40

 41

 42

 43

 44

 45

 46

 47

 48

 49

 50

 51

 52

 53

 54

 55

 56

 57

 58

 59

 60

 61

 62

 63

 64

 65

 66

 67

 68

 69

 70

 71

 72

 73

 74

 75

 76

 77

 78

 79

 80

 81

 82

 83

 84

 85

 86

 87

 88

 89

 90

 91

 92

 93

 94

 95

 96

 97

 98

 99

100

101

102

103

104

105

106

107

108

109

110

```

 | 

```

import gunicorn.http as http

import gunicorn.http.wsgi as wsgi

import gunicorn.util as util

import gunicorn.workers.base as base

class SyncWorker(base.Worker):

    def run(self):

        # self.socket appears to lose its blocking status after

        # we fork in the arbiter. Reset it here.

        self.socket.setblocking(0)

        while self.alive:

            self.notify()

            # Accept a connection. If we get an error telling us

            # that no connection is waiting we fall down to the

            # select which is where we'll wait for a bit for new

            # workers to come give us some love.

            try:

                client, addr = self.socket.accept()

                client.setblocking(1)

                util.close\_on\_exec(client)

                self.handle(client, addr)

                # Keep processing clients until no one is waiting. This

                # prevents the need to select() for every client that we

                # process.

                continue

            except socket.error, e:

                if e[0] not in (errno.EAGAIN, errno.ECONNABORTED):

                    raise

            # If our parent changed then we shut down.

            if self.ppid != os.getppid():

                self.log.info("Parent changed, shutting down: %s", self)

                return

            try:

                self.notify()

                ret = select.select([self.socket], [], self.PIPE, self.timeout)

                if ret[0]:

                    continue

            except select.error, e:

                if e[0] == errno.EINTR:

                    continue

                if e[0] == errno.EBADF:

                    if self.nr < 0:

                        continue

                    else:

                        return

                raise

    def handle(self, client, addr):

        try:

            parser = http.RequestParser(client)

            req = parser.next()

            self.handle\_request(req, client, addr)

        except StopIteration, e:

            self.log.debug("Closing connection. %s", e)

        except socket.error, e:

            if e[0] != errno.EPIPE:

                self.log.exception("Error processing request.")

            else:

                self.log.debug("Ignoring EPIPE")

        except Exception, e:

            self.handle\_error(client, e)

        finally:

            util.close(client)

    def handle\_request(self, req, client, addr):

        environ = {}

        try:

            self.cfg.pre\_request(self, req)

            request\_start = datetime.now()

            resp, environ = wsgi.create(req, client, addr,

                    self.address, self.cfg)

            # Force the connection closed until someone shows

            # a buffering proxy that supports Keep-Alive to

            # the backend.

            resp.force\_close()

            self.nr += 1

            if self.nr >= self.max\_requests:

                self.log.info("Autorestarting worker after current request.")

                self.alive = False

            respiter = self.wsgi(environ, resp.start\_response)

            try:

                if isinstance(respiter, environ['wsgi.file\_wrapper']):

                    resp.write\_file(respiter)

                else:

                    for item in respiter:

                        resp.write(item)

                resp.close()

                request\_time = datetime.now() - request\_start

                self.log.access(resp, environ, request\_time)

            finally:

                if hasattr(respiter, "close"):

                    respiter.close()

        except socket.error:

            raise

        except Exception, e:

            # Only send back traceback in HTTP in debug mode.

            self.handle\_error(client, e)

            return

        finally:

            try:

                self.cfg.post\_request(self, req, environ)

            except:

                pass

```

 |

SyncWorker总共才定义了三个方法，run()，handle()，handle\_request()，真正的处理就在run()内，它是从父类base.Worker继承而来。很自然，我们就会回想起app的处理，wsgiapp/djangoapp的设计模式，从WSGIApplication与DjangoApplication继承自Application。

观察了源代码的结构后，发现的确这种设计模式是一脉相承的。在base.Worker中，run()方法是作为一个空架子放在那里。

|  |  |

| --- | --- |

| 

```

 1

 2

 3

 4

 5

 6

 7

 8

 9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

```

 | 

```

class Worker(object):

    SIGNALS = map(

        lambda x: getattr(signal, "SIG%s" % x),

        "HUP QUIT INT TERM USR1 USR2 WINCH CHLD".split()

    )

    PIPE = []

    def \_\_init\_\_(self, age, ppid, socket, app, timeout, cfg, log):

        """\

 This is called pre-fork so it shouldn't do anything to the

 current process. If there's a need to make process wide

 changes you'll want to do that in ``self.init\_process()``.

 """

        self.age = age

        self.ppid = ppid

        self.socket = socket

        self.app = app

        self.timeout = timeout

        self.cfg = cfg

        self.booted = False

        self.nr = 0

        self.max\_requests = cfg.max\_requests or sys.maxint

        self.alive = True

        self.log = log

        self.debug = cfg.debug

        self.address = self.socket.getsockname()

        self.tmp = WorkerTmp(cfg)

    def \_\_str\_\_(self):

        return "<Worker %s>" % self.pid

    @property

    def pid(self):

        return os.getpid()

    def notify(self):

        """\

 Your worker subclass must arrange to have this method called

 once every ``self.timeout`` seconds. If you fail in accomplishing

 this task, the master process will murder your workers.

 """

        self.tmp.notify()

    def run(self):

        """\

 This is the mainloop of a worker process. You should override

 this method in a subclass to provide the intended behaviour

 for your particular evil schemes.

 """

        raise NotImplementedError()

```

 |

在主控master那里启动worker.init\_process()，恰恰就是定义在base.Worker中.

|  |  |

| --- | --- |

| 

```

 1

 2

 3

 4

 5

 6

 7

 8

 9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

```

 | 

```

class Worker(object):

    # ......

    def init\_process(self):

        """\

 If you override this method in a subclass, the last statement

 in the function should be to call this method with

 super(MyWorkerClass, self).init\_process() so that the ``run()``

 loop is initiated.

 """

        util.set\_owner\_process(self.cfg.uid, self.cfg.gid)

        # Reseed the random number generator

        util.seed()

        # For waking ourselves up

        self.PIPE = os.pipe()

        map(util.set\_non\_blocking, self.PIPE)

        map(util.close\_on\_exec, self.PIPE)

        # Prevent fd inherientence

        util.close\_on\_exec(self.socket)

        util.close\_on\_exec(self.tmp.fileno())

        self.log.close\_on\_exec()

        self.init\_signals()

        self.wsgi = self.app.wsgi()

        # Enter main run loop

        self.booted = True

        self.run()

    def init\_signals(self):

        map(lambda s: signal.signal(s, signal.SIG\_DFL), self.SIGNALS)

        signal.signal(signal.SIGQUIT, self.handle\_quit)

        signal.signal(signal.SIGTERM, self.handle\_exit)

        signal.signal(signal.SIGINT, self.handle\_exit)

        signal.signal(signal.SIGWINCH, self.handle\_winch)

        signal.signal(signal.SIGUSR1, self.handle\_usr1)

        # Don't let SIGQUIT and SIGUSR1 disturb active requests

        # by interrupting system calls

        if hasattr(signal, 'siginterrupt'):  # python >= 2.6

            signal.siginterrupt(signal.SIGQUIT, False)

            signal.siginterrupt(signal.SIGUSR1, False)

```

 |

在init\_process()执行了子类包括SyncWorker的run()方法。

#### 异步worker[¶](#id2 "Permalink to this headline")

异步worker分两种，一个是Eventlet，另一个就是大名鼎鼎的gevent。整体的设计模式，与同步worker类似，都是采用的template pattern。

##### EventletWorker[¶](#eventletworker "Permalink to this headline")

EventletWorker继承路线上是分成了三级:

`class EventletWorker(AsyncWorker)`

`class AsyncWorker(base.Worker)`

先看EventletWorker

|  |  |

| --- | --- |

| 

```

 1

 2

 3

 4

 5

 6

 7

 8

 9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

```

 | 

```

from \_\_future\_\_ import with\_statement

import os

try:

    import eventlet

except ImportError:

    raise RuntimeError("You need eventlet installed to use this worker.")

from eventlet import hubs

from eventlet.greenio import GreenSocket

from gunicorn.workers.async import AsyncWorker

class EventletWorker(AsyncWorker):

    @classmethod

    def setup(cls):

        import eventlet

        if eventlet.version\_info < (0,9,7):

            raise RuntimeError("You need eventlet >= 0.9.7")

        eventlet.monkey\_patch(os=False)

    def init\_process(self):

        hubs.use\_hub()

        super(EventletWorker, self).init\_process()

    def timeout\_ctx(self):

        return eventlet.Timeout(self.cfg.keepalive, False)

    def run(self):

        self.socket = GreenSocket(family\_or\_realsock=self.socket.sock)

        self.socket.setblocking(1)

        self.acceptor = eventlet.spawn(eventlet.serve, self.socket,

                self.handle, self.worker\_connections)

        while self.alive:

            self.notify()

            if self.ppid != os.getppid():

                self.log.info("Parent changed, shutting down: %s", self)

                break

            eventlet.sleep(1.0)

        self.notify()

        with eventlet.Timeout(self.timeout, False):

            eventlet.kill(self.acceptor, eventlet.StopServe)

```

 |

run()方法中的语句:

```

self.acceptor = eventlet.spawn(eventlet.serve, self.socket,

                self.handle, self.worker\_connections)

```

调用的是:

`eventlet.``serve`(*sock*, *handle*, *concurrency=1000*)[¶](#eventlet.serve "Permalink to this definition")

这里包含了self.handle，这个方法在AsyncWorker中声明定义

|  |  |

| --- | --- |

| 

```

 1

 2

 3

 4

 5

 6

 7

 8

 9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

```

 | 

```

class AsyncWorker(base.Worker):

    def \_\_init\_\_(self, \*args, \*\*kwargs):

        super(AsyncWorker, self).\_\_init\_\_(\*args, \*\*kwargs)

        self.worker\_connections = self.cfg.worker\_connections

    def timeout\_ctx(self):

        raise NotImplementedError()

    def handle(self, client, addr):

        try:

            parser = http.RequestParser(client)

            try:

                while True:

                    req = None

                    with self.timeout\_ctx():

                        req = parser.next()

                    if not req:

                        break

                    self.handle\_request(req, client, addr)

            except StopIteration, e:

                self.log.debug("Closing connection. %s", e)

        except socket.error, e:

            if e[0] not in (errno.EPIPE, errno.ECONNRESET):

                self.log.exception("Socket error processing request.")

            else:

                if e[0] == errno.ECONNRESET:

                    self.log.debug("Ignoring connection reset")

                else:

                    self.log.debug("Ignoring EPIPE")

        except Exception, e:

            self.handle\_error(client, e)

        finally:

            util.close(client)

    def handle\_request(self, req, sock, addr):

        try:

            self.cfg.pre\_request(self, req)

            request\_start = datetime.now()

            resp, environ = wsgi.create(req, sock, addr, self.address, self.cfg)

            self.nr += 1

            if self.alive and self.nr >= self.max\_requests:

                self.log.info("Autorestarting worker after current request.")

                resp.force\_close()

                self.alive = False

            respiter = self.wsgi(environ, resp.start\_response)

            if respiter == ALREADY\_HANDLED:

                return False

            try:

                for item in respiter:

                    resp.write(item)

                resp.close()

                request\_time = datetime.now() - request\_start

                self.log.access(resp, environ, request\_time)

            finally:

                if hasattr(respiter, "close"):

                  respiter.close()

            if resp.should\_close():

                raise StopIteration()

        finally:

            try:

                self.cfg.post\_request(self, req, environ)

            except:

                pass

        return True

```

 |

##### GeventWorker[¶](#geventworker "Permalink to this headline")

这里的代码关系要复杂一些

|  |  |

| --- | --- |

| 

```

 1

 2

 3

 4

 5

 6

 7

 8

 9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

```

 | 

```

class GeventWorker(AsyncWorker):

    server\_class = None

    wsgi\_handler = None

    @classmethod

    def setup(cls):

        from gevent import monkey

        monkey.noisy = False

        monkey.patch\_all()

    def timeout\_ctx(self):

        return gevent.Timeout(self.cfg.keepalive, False)

    def run(self):

        self.socket.setblocking(1)

        pool = Pool(self.worker\_connections)

        if self.server\_class is not None:

            server = self.server\_class(

                self.socket, application=self.wsgi, spawn=pool, log=self.log,

                handler\_class=self.wsgi\_handler)

        else:

            server = StreamServer(self.socket, handle=self.handle, spawn=pool)

        server.start()

        try:

            while self.alive:

                self.notify()

                if self.ppid != os.getppid():

                    self.log.info("Parent changed, shutting down: %s", self)

                    break

                gevent.sleep(1.0)

        except KeyboardInterrupt:

            pass

        try:

            # Try to stop connections until timeout

            self.notify()

            server.stop(timeout=self.timeout)

        except:

            pass

    def handle\_request(self, \*args):

        try:

            super(GeventWorker, self).handle\_request(\*args)

        except gevent.GreenletExit:

            pass

    if hasattr(gevent.core, 'dns\_shutdown'):

        def init\_process(self):

            #gevent 0.13 and older doesn't reinitialize dns for us after forking

            #here's the workaround

            gevent.core.dns\_shutdown(fail\_requests=1)

            gevent.core.dns\_init()

            super(GeventWorker, self).init\_process()

```

 |

主要的焦点在:

```

server = self.server\_class(

                self.socket, application=self.wsgi, spawn=pool, log=self.log,

                handler\_class=self.wsgi\_handler)

```

而self.server\_class和self.wsgi\_handler都是空值，怎么能使用呢？

再继续往下看，原来它派生出了一个子类

```

class GeventPyWSGIWorker(GeventWorker):

    "The Gevent StreamServer based workers."

    server\_class = PyWSGIServer

    wsgi\_handler = PyWSGIHandler

```

在gunicorn的代码树中，唯一没有介绍到的只有http目录了，下一步进入http

### 自带的httpserver[¶](#httpserver "Permalink to this headline")

gunicorn自带了一个python wsgi http server，是为了默认的syncworker而设计的。

```

class SyncWorker(base.Worker):

    def run(self):

        # self.socket appears to lose its blocking status after

        # we fork in the arbiter. Reset it here.

        self.socket.setblocking(0)

        while self.alive:

            self.notify()

            # Accept a connection. If we get an error telling us

            # that no connection is waiting we fall down to the

            # select which is where we'll wait for a bit for new

            # workers to come give us some love.

            try:

                client, addr = self.socket.accept()

                client.setblocking(1)

                util.close\_on\_exec(client)

                self.handle(client, addr)

                # Keep processing clients until no one is waiting. This

                # prevents the need to select() for every client that we

                # process.

                continue

            except socket.error, e:

                if e[0] not in (errno.EAGAIN, errno.ECONNABORTED):

                    raise

```

终于在这里看到了久违的self.socket.accept()，接收一个访问连接，由self.handle(client, addr)来处理客户端来的请求命令。

```

class SyncWorker(base.Worker):

    #......

    def handle(self, client, addr):

        try:

            parser = http.RequestParser(client)

            req = parser.next()

            self.handle\_request(req, client, addr)

        except StopIteration, e:

            self.log.debug("Closing connection. %s", e)

        except socket.error, e:

            if e[0] != errno.EPIPE:

                self.log.exception("Error processing request.")

            else:

                self.log.debug("Ignoring EPIPE")

        except Exception, e:

            self.handle\_error(client, e)

        finally:

            util.close(client)

```

从http.RequestParser(client)进入http模块中，在self.handle\_request(req, client, addr)中调用wsgi模块

```

class SyncWorker(base.Worker):

    #......

    def handle\_request(self, req, client, addr):

        environ = {}

        try:

            self.cfg.pre\_request(self, req)

            request\_start = datetime.now()

            resp, environ = wsgi.create(req, client, addr,

                    self.address, self.cfg)

            # Force the connection closed until someone shows

            # a buffering proxy that supports Keep-Alive to

            # the backend.

            resp.force\_close()

            self.nr += 1

            if self.nr >= self.max\_requests:

                self.log.info("Autorestarting worker after current request.")

                self.alive = False

            respiter = self.wsgi(environ, resp.start\_response)

            try:

                if isinstance(respiter, environ['wsgi.file\_wrapper']):

                    resp.write\_file(respiter)

                else:

                    for item in respiter:

                        resp.write(item)

                resp.close()

                request\_time = datetime.now() - request\_start

                self.log.access(resp, environ, request\_time)

            finally:

                if hasattr(respiter, "close"):

                    respiter.close()

        except socket.error:

            raise

        except Exception, e:

            # Only send back traceback in HTTP in debug mode.

            self.handle\_error(client, e)

            return

        finally:

            try:

                self.cfg.post\_request(self, req, environ)

            except:

                pass

```

其中涉及到wsgi模块的有:

```

resp, environ = wsgi.create(req, client, addr,

                    self.address, self.cfg)

```

获得最终的响应结果:

```

respiter = self.wsgi(environ, resp.start\_response)

```

这里的self.wsgi的踪迹在哪里呢？

1. 由base.Worker中的方法init\_process()中调用

```

self.init\_signals()

self.wsgi = self.app.wsgi()

# Enter main run loop

self.booted = True

self.run()

```

2. 在app，也就是wsgiapp或djangoapp中WSGIApplication -> Application中，有

```

def wsgi(self):

    if self.callable is None:

        self.callable = self.load()

    return self.callable

```

3. 直接由self.load()执行，回转到WSGIApplication中的load()

```

def load(self):

    return util.import\_app(self.app\_uri)

```

4. 调用util模块中的import\_app(self.app\_uri)

```

def import\_app(module):

    parts = module.split(":", 1)

    if len(parts) == 1:

        module, obj = module, "application"

    else:

        module, obj = parts[0], parts[1]

    try:

        \_\_import\_\_(module)

    except ImportError:

        if module.endswith(".py") and os.path.exists(module):

            raise ImportError("Failed to find application, did "

                "you mean '%s:%s'?" % (module.rsplit(".",1)[0], obj))

        else:

            raise

    mod = sys.modules[module]

    app = eval(obj, mod.\_\_dict\_\_)

    if app is None:

        raise ImportError("Failed to find application object: %r" % obj)

    if not callable(app):

        raise TypeError("Application object must be callable.")

    return app

```

这才是最终wsgi应用程序的发起。还记得我们在快速入门中编写的简单wsgi程序么?

那么我们就在这里加上一句print，看看是不是:

```

print u'waiting for you ...(%s, %s)' % (module, obj)

```

再执行命令:

```

$ gunicorn --workers=2 myapp:app

```

结果如下:

```

(mypy)hzg@xubuntu:~/sample$ gunicorn --workers=2 myapp:app

2012-02-16 08:45:09 [2007] [INFO] Starting gunicorn 0.13.4

2012-02-16 08:45:09 [2007] [INFO] Listening at: http://127.0.0.1:8000 (2007)

2012-02-16 08:45:09 [2007] [INFO] Using worker: sync

2012-02-16 08:45:09 [2010] [INFO] Booting worker with pid: 2010

waiting for you ...(myapp, app)

2012-02-16 08:45:09 [2011] [INFO] Booting worker with pid: 2011

waiting for you ...(myapp, app)

```

### 自己动手[¶](#id1 "Permalink to this headline")

我们再来看看用自定义配置文件

```

#import multiprocessing

bind = "127.0.0.1:8000"

#workers = multiprocessing.cpu\_count() \* 2 + 1

worker\_class = "sync"

#worker\_class = "gevent\_pywsgi"

#worker\_class = "egg:gunicorn#tornado"

#worker\_class = "gunicorn.workers.ggevent.GeventWorker"

worker\_connections = 100

timeout = 50

loglevel = 'debug'

proc\_name = 'justforfun'

def post\_fork(server, worker):

    #server.log.info("Worker spawned (pid: %s)", worker.pid)

    pass

def pre\_fork(server, worker):

    #server.log.info("Pre fork ...")

    pass

```

执行很简单:

```

gunicorn -c myconf.py myapp:app

```

输出结果:

```

(mypy)hzg@gofast:~/work/rdgunicorn/sample$ gunicorn -c myconf.py myapp:app

2012-02-19 20:05:43 [2139] [INFO] Starting gunicorn 0.13.4

2012-02-19 20:05:43 [2139] [DEBUG] Arbiter booted

2012-02-19 20:05:43 [2139] [INFO] Listening at: http://127.0.0.1:8000 (2139)

2012-02-19 20:05:43 [2139] [INFO] Using worker: sync

2012-02-19 20:05:43 [2142] [INFO] Booting worker with pid: 2142

2012-02-19 20:05:50 [2142] [DEBUG] GET /

2012-02-19 20:06:16 [2139] [INFO] Handling signal: winch

2012-02-19 20:06:16 [2139] [INFO] SIGWINCH ignored. Not daemonized

2012-02-19 20:07:37 [2139] [INFO] Handling signal: ttin

2012-02-19 20:07:37 [2207] [INFO] Booting worker with pid: 2207

2012-02-19 20:07:55 [2139] [INFO] Handling signal: ttou

2012-02-19 20:07:55 [2142] [INFO] Worker exiting (pid: 2142)

```

看到其中的ttin, ttou，就是用来增加和中止worker进程，怎么做呢？:

```

$ kill -TTIN 2139

$ kill -TTOU 2139

```

### 部署应用[¶](#id1 "Permalink to this headline")

gunicorn官方主站强烈推荐nginx作为web server proxy，同时给出建议：如果选择其他的web server，在使用gunicorn的sync worker时，就需要确保它能缓冲慢速客户端的请求。没有这个缓冲的功能，gunicorn很容易遭受dos的攻击。可以用slowloris测试一下，看看你的web server proxy的响应如何。

slowloris是什么? 瞧瞧去:

```

http://ha.ckers.org/slowloris/

```

slowloris，采用perl编写的一个脚本，有趣有趣。

转回正题，看看下面是nginx的配置示例，在nginx.conf中http部分加入:

```

upstream app\_server {

    server http://localhost:8000 fail\_timeout=0;

}

```

在sites-available目录下的default中:

```

server {

    listen 80 default;

    client_max_body_size 4G;

    server_name _;

    keepalive_timeout 5;

    location / {

        # checks for static file, if not found proxy to app

        try_files $uri @proxy_to_app;

    }

    location @proxy_to_app {

        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

        proxy_set_header Host $http_host;

        proxy_redirect off;

        proxy_pass   http://app_server;

    }

}

```

在浏览器中输入:

```

http://localhost

```

即可得到输出结果

索引表[¶](#id3 "Permalink to this headline")

=========================================

* [Index](genindex.html)

* [Module Index](py-modindex.html)

* [Search Page](search.html)

### [Table Of Contents](index.html#document-index)

* [介绍](index.html#document-intro)

	+ [什么是gunicorn](index.html#gunicorn)

	+ [gunicorn的特性](index.html#id2)

* [设计](index.html#document-design)

	+ [服务器模型](index.html#id2)

		- [解释pre-fork](index.html#pre-fork)

	+ [主控master进程](index.html#master)

	+ [同步workers](index.html#workers)

	+ [异步workers](index.html#id3)

	+ [tornado workers](index.html#tornado-workers)

	+ [选择worker进程类型](index.html#worker)

	+ [启动多少个workers?](index.html#id4)

* [快速入门](index.html#document-getstart)

	+ [安装gunicorn](index.html#gunicorn)

	+ [简单的wsgi应用](index.html#wsgi)

	+ [简单的django应用](index.html#django)

	+ [命令行参数](index.html#id2)

* [从哪开始阅读源代码](index.html#document-readstart)

	+ [代码树](index.html#id2)

	+ [从脚本命令开始](index.html#id3)

		- [wsgiapp.py](index.html#wsgiapp-py)

		- [djangoapp.py](index.html#djangoapp-py)

* [深入Application](index.html#document-app_base)

	+ [关注内部](index.html#id1)

	+ [代码流程](index.html#id2)

* [配置参数](index.html#document-config)

	+ [元类编程](index.html#id2)

		- [解释type](index.html#type)

		- [跟踪子类](index.html#id3)

	+ [gunicorn配置怎么处理](index.html#gunicorn)

* [真正的主宰者](index.html#document-arbiter)

	+ [循环主体](index.html#id2)

	+ [核心语句](index.html#id3)

* [worker积木](index.html#document-route)

	+ [同步worker](index.html#id1)

	+ [异步worker](index.html#id2)

		- [EventletWorker](index.html#eventletworker)

		- [GeventWorker](index.html#geventworker)

* [自带的httpserver](index.html#document-http)

* [自己动手](index.html#document-play)

* [部署应用](index.html#document-deploy)

### Quick search

### Navigation

* [阅读gunicorn代码 v0.0.1 文档](index.html#document-index) »

 © Copyright 2012, 江西 hzg.

 Created using [Sphinx](http://sphinx-doc.org/) 1.5.3.

 

