协程和并发课程【部分三~四】

2017-04-30

第三部分：协程和事件分发

协程还能用于处理事件流，很强大吧！这里作者举的例子是处理 XML 文件，采用的方法是 SAX（Simple API for XML）。下边先简介下 SAX 及其用法。

SAX 简介

SAX（Simple API for XML）跟 DOM 一样也是用于访问 XML 文档的接口，原先只是 Java 上的 API，后来也在其他语言中有实现版本，如 Python 版的 SAX。详细介绍可参考 IBM 的一篇博文使用 SAX 处理 XML 文档。

SAX 是一个事件驱动的接口，这使得它能够增量处理大容量 XML 文件而不用担心内存不足，其模型如下：

一个简单的例子为：

# basicsax.py
#
# A very simple example illustrating the SAX XML parsing interface

import xml.sax


class MyHandler(xml.sax.ContentHandler):
    def startElement(self, name, attrs):
        print "startElement", name

    def endElement(self, name):
        print "endElement", name

    def characters(self, text):
        print "characters", repr(text)[:40]


xml.sax.parse("allroutes.xml", MyHandler())

这里我把它改进了下，让输出更友好点：

# basicsax2.py
#
# A very simple example illustrating the SAX XML parsing interface

import xml.sax


class MyHandler(xml.sax.ContentHandler):
    def __init__(self):
        self.current_tag = ''
        self.id = ''
        self.route = ''
        self.color = ''
        self.revenue = ''
        self.direction = ''
        self.latitude = ''
        self.longitude = ''
        self.pattern = ''
        self.patternDirection = ''
        self.run = ''
        self.finalStop = ''
        self.operator = ''

    def startElement(self, tag, attrs):
        self.current_tag = tag
        if tag == 'bus':
            print '\n\n\n'
            print '***** Bus *****'

    def endElement(self, tag):
        if self.current_tag == 'id':
            print 'id:', self.id
            self.id = ''
        elif self.current_tag == 'route':
            print 'route:', self.route
            self.route = ''
        elif self.current_tag == 'color':
            print 'color:', self.color
            self.color = ''
        elif self.current_tag == 'revenue':
            print 'revenue:', self.revenue
            self.revenue = ''
        elif self.current_tag == 'direction':
            print 'direction:', self.direction
            self.direction = ''
        elif self.current_tag == 'latitude':
            print 'latitude:', self.latitude
            self.latitude = ''
        elif self.current_tag == 'longitude':
            print 'longitude:', self.longitude
            self.longitude = ''
        elif self.current_tag == 'pattern':
            print 'pattern:', self.pattern
            self.pattern = ''
        elif self.current_tag == 'patternDirection':
            print 'patternDirection:', self.patternDirection
            self.patternDirection = ''
        elif self.current_tag == 'run':
            print 'run:', self.run
            self.run = ''
        elif self.current_tag == 'finalStop':
            print 'finalStop:', self.finalStop
            self.finalStop = ''
        elif self.current_tag == 'operator':
            print 'operator:', self.operator
            self.operator = ''
        self.current_tag = ''

    def characters(self, content):
        if self.current_tag == 'id':
            self.id += content	# += -> 以防有该标签下内容有换行导致的数据读取不全
        elif self.current_tag == 'route':
            self.route += content
        elif self.current_tag == 'color':
            self.color += content
        elif self.current_tag == 'revenue':
            self.revenue += content
        elif self.current_tag == 'direction':
            self.direction += content
        elif self.current_tag == 'latitude':
            self.latitude += content
        elif self.current_tag == 'longitude':
            self.longitude += content
        elif self.current_tag == 'pattern':
            self.pattern += content
        elif self.current_tag == 'patternDirection':
            self.patternDirection += content
        elif self.current_tag == 'run':
            self.run += content
        elif self.current_tag == 'finalStop':
            self.finalStop += content
        elif self.current_tag == 'operator':
            self.operator += content


xml.sax.parse("allroutes.xml", MyHandler())
# Results
# ***** Bus *****
# id: 7574
# route: 147
# color: #3300ff
# revenue: true
# direction: North Bound
# latitude: 41.925682067871094
# longitude: -87.63092803955078
# pattern: 2499
# patternDirection: North Bound
# run: P675
# finalStop: Paulina & Howard Terminal
# operator: 42493
# ......
# 
# ***** Bus *****
# id: 6842
# ......

但作者认为这种底层事件处理有点过于底层和笨拙，所以就引入协程。下边我们来看看怎么用协程来处理 XML 文件。

协程 + SAX

方案

协程加 SAX 的方案如下：

代码如下：

# cosax.py
#
# An example showing how to push SAX events into a coroutine target

import xml.sax


class EventHandler(xml.sax.ContentHandler):
    def __init__(self, target):
        self.target = target

    def startElement(self, name, attrs):
        self.target.send(('start', (name, attrs._attrs)))

    def characters(self, text):
        self.target.send(('text', text))

    def endElement(self, name):
        self.target.send(('end', name))


# example use
if __name__ == '__main__':
    from coroutine import *

    @coroutine
    def printer():
        while True:
            event = (yield)
            print event

    xml.sax.parse("allroutes.xml",
                  EventHandler(printer()))

过滤版本

过滤版本的太强大了：

# buses.py
#
# An example of setting up an event handling pipeline with coroutines
# and XML parsing.

from coroutine import *


@coroutine
def buses_to_dicts(target):
    while True:
        event, value = (yield)
        # Look for the start of a <bus> element
        if event == 'start' and value[0] == 'bus':
            busdict = {}
            fragments = []
            # Capture text of inner elements in a dict
            while True:
                event, value = (yield)
                if event == 'start':
                    fragments = []
                elif event == 'text':
                    fragments.append(value)
                elif event == 'end':
                    if value != 'bus':
                        busdict[value] = "".join(fragments)
                    else:
                        target.send(busdict)
                        break


@coroutine
def filter_on_field(fieldname, value, target):
    while True:
        d = (yield)
        if d.get(fieldname) == value:
            target.send(d)


@coroutine
def bus_locations():
    while True:
        bus = (yield)
        print "%(route)s,%(id)s,\"%(direction)s\"," \
              "%(latitude)s,%(longitude)s" % bus

        # Example


if __name__ == '__main__':
    import xml.sax
    from cosax import EventHandler

	# Filter on route "22" and direction "North Bound"
    xml.sax.parse(
        "allroutes.xml",
        EventHandler(
            buses_to_dicts(
                filter_on_field(
                    "route", "22",
                    filter_on_field(
                        "direction", "North Bound",
                        bus_locations()
                    )
                )
            )
        )
    )

# Results:
# 22,1485,"North Bound",41.880481123924255,-87.62948191165924
# 22,1629,"North Bound",42.01851969751819,-87.6730209876751
# 22,1489,"North Bound",41.962393500588156,-87.66610128229314
# 22,1533,"North Bound",41.92381583870231,-87.6395345910803
# 22,1779,"North Bound",41.989253234863284,-87.66976165771484
# 22,1595,"North Bound",41.892801920572914,-87.62985568576389
# 22,1567,"North Bound",41.91437446296989,-87.63357444862267
# 22,1795,"North Bound",41.98753767747145,-87.66956552358774
# 22,1543,"North Bound",41.92852973937988,-87.64240264892578
# 22,1315,"North Bound",41.96697834559849,-87.66706085205078
# 22,6069,"North Bound",41.98728592755043,-87.66953517966074
# 22,1891,"North Bound",41.92987823486328,-87.64342498779297
# 22,1569,"North Bound",42.003393713033425,-87.6723536365437
# 22,1617,"North Bound",41.90174682617187,-87.63128570556641
# 22,1821,"North Bound",41.976410124037,-87.66838073730469
# 22,1499,"North Bound",41.96970504369491,-87.66764088166066

用 Expat 解析 XML

Expat 相当于 sax.xml.parse 版本，但速度更快，而且用法一样：

# coexpat.py
#
# An example of pushing XML events generated by the low-level expat
# XML library into coroutines.

import xml.parsers.expat


def expat_parse(f, target):
    parser = xml.parsers.expat.ParserCreate()
    parser.buffer_size = 65536
    parser.buffer_text = True
    parser.returns_unicode = False
    parser.StartElementHandler = \
        lambda name, attrs: target.send(('start', (name, attrs)))
    parser.EndElementHandler = \
        lambda name: target.send(('end', name))
    parser.CharacterDataHandler = \
        lambda data: target.send(('text', data))
    parser.ParseFile(f)


# Example.  This uses the bus processing code from earlier with no changes.

if __name__ == '__main__':
    from buses import *

    expat_parse(
        open("allroutes.xml"),
        buses_to_dicts(
            filter_on_field(
                "route", "22",
                filter_on_field(
                    "direction", "North Bound",
                    bus_locations()
                )
            )
        )
    )

作者测试了当输入的 XML 文件大小为 30MB 时，SAX 版本的所需时间是 8.37s，而 Expat 版本的则只需要 4.51s，节省了接近一半。

自定义 XML 解析器

作者很有心，还自己定制了一个 C 语言版 XML 解析器，源代码 cxml/cxmlparse.c：

#include "Python.h"
#include "expat.h"

#define TEXT_BUF_SIZE  65536

static PyObject *send_meth;
static char     *textbuf[TEXT_BUF_SIZE];
static int       textindex = 0;

static PyObject *START_EVENT;
static PyObject *END_EVENT;
static PyObject *TEXT_EVENT;

void send_text() {
	if (textindex) {
		PyObject *result = 0;
		PyObject *text;
		PyObject *args;
		PyObject *event;

		text = PyString_FromStringAndSize(textbuf, textindex);
		event = PyTuple_Pack(2, TEXT_EVENT, text);
		args = PyTuple_Pack(1, event);
		result = PyEval_CallObject(send_meth, args);
		Py_DECREF(args);
		Py_DECREF(event);
		Py_DECREF(text);
		Py_XDECREF(result);
	}
	textindex = 0;
}

void start_hndl(void *data, const char *el, const char **attrs) {
	PyObject *event, *evtvalue;
	PyObject *result = 0;
	PyObject *args;
	PyObject *name;
	PyObject *attr;

	send_text();

	name = PyString_FromString(el);
	attr = PyDict_New();
	evtvalue = PyTuple_Pack(2, name, attr);
	event = PyTuple_Pack(2, START_EVENT, evtvalue);
	args = PyTuple_Pack(1, event);

	result = PyEval_CallObject(send_meth, args);

	Py_DECREF(args);
	Py_DECREF(event);
	Py_DECREF(evtvalue);
	Py_DECREF(attr);
	Py_DECREF(name);
	Py_XDECREF(result);
}

void end_hndl(void *data, const char *el) {
	PyObject *event;
	PyObject *result = 0;
	PyObject *name;
	PyObject *args;

	send_text();
	name = PyString_FromString(el);
	event = PyTuple_Pack(2, END_EVENT, name);
	args = PyTuple_Pack(1, event);

	result = PyEval_CallObject(send_meth, args);

	Py_DECREF(args);
	Py_DECREF(event);
	Py_DECREF(name);
	Py_XDECREF(result);
}

void char_hndl(void *data, const char *text, int txtlen) {
	int remaining;
	remaining = TEXT_BUF_SIZE - textindex;
	if (remaining > txtlen) {
		memmove(textbuf + textindex, text, txtlen);
		textindex += txtlen;
		return;
	}
	memmove(textbuf + textindex, text, remaining);
	textindex += remaining;
	send_text();
	memmove(textbuf, text + remaining, txtlen - remaining);
	textindex = txtlen - remaining;
}

#define BUFSIZE 65536

PyObject *parse(PyObject *self, PyObject *args) {
	char     *filename;
	PyObject *target;

	XML_Parser p;
	if (!PyArg_ParseTuple(args, "sO", &filename, &target)) {
		return NULL;
	}
	send_meth = PyObject_GetAttrString(target, "send");
	p = XML_ParserCreate(NULL);
	XML_SetElementHandler(p, start_hndl, end_hndl);
	XML_SetCharacterDataHandler(p, char_hndl);

	/* Run the parser */
	{
		FILE *f = fopen(filename, "r");
		for (;;) {
			int bytes_read;
			void *buff = XML_GetBuffer(p, BUFSIZE);
			if (buff == NULL) {
				break;
			}
			bytes_read = fread(buff, 1, BUFSIZE, f);
			XML_ParseBuffer(p, bytes_read, bytes_read == 0);
			if (bytes_read == 0) {
				break;
			}
		}
		fclose(f);
	}
	Py_XDECREF(send_meth);
	return Py_BuildValue("i", 0);
}

static PyMethodDef cxmlparsemethods[] = {
	{ "parse", parse, METH_VARARGS },
	{ NULL, NULL }
};

void initcxmlparse() {
	Py_InitModule("cxmlparse", cxmlparsemethods);
	START_EVENT = PyString_FromString("start");
	END_EVENT = PyString_FromString("end");
	TEXT_EVENT = PyString_FromString("text");
}

要让该模块可用，需要编译并安装，具体步骤如下（cxml 目录下）：

如果没有 expat.h 文件，安装 Expat：yum install expat-devel.x86_64
编译并安装：python setup.py build_ext –inplace

运行测试文件 test：python test.py

import cxmlparse


def coroutine(func):
    def start(*args, **kwargs):
        cr = func(*args, **kwargs)
        cr.next()
        return cr

    return start


@coroutine
def buses_to_dicts(target):
    while True:
        event, value = (yield)
        if event == 'start' and value[0] == 'bus':
            busdict = {}
            fragments = []
            # Capture inner elements of <bus> elements
            while True:
                event, value = (yield)
                if event == 'start':
                    fragments = []
                elif event == 'text':
                    fragments.append(value)
                elif event == 'end':
                    if value != 'bus':
                        busdict[value] = "".join(fragments)
                    else:
                        target.send(busdict)
                        break


@coroutine
def filter_on_field(fieldname, value, target):
    while True:
        d = (yield)
        if d.get(fieldname) == value:
            target.send(d)


@coroutine
def bus_locations():
    while True:
        bus = (yield)
        print "%(route)s,%(id)s,\"%(direction)s\"," \
              "%(latitude)s,%(longitude)s" % bus


@coroutine
def printer():
    while True:
        item = (yield)
        print item


cxmlparse.parse(
    "../allroutes.xml",
    buses_to_dicts(
        filter_on_field(
            "route", "22",
            filter_on_field(
                "direction", "North Bound",
                bus_locations()
            )
        )
    )
)

用作者定制版的，处理 30MB 的 XML 文件就只需 2.95s 左右，更加省时。

插曲：用 ElementTree 解析 XML

用 ElementTree 来解析 XML 相对要简单一点（缺点在于处理大容量 XML 文件时容易导致内存不足）：

# iterbus.py
#
# An example of incremental XML parsing with the ElementTree library

from xml.etree.cElementTree import iterparse

for event, elem in iterparse("allroutes.xml", ('start', 'end')):
    if event == 'start' and elem.tag == 'buses':
        buses = elem
    elif event == 'end' and elem.tag == 'bus':
        busdict = dict((child.tag, child.text) for child in elem)
        if busdict['route'] == '22' and busdict['direction'] == 'North Bound':
            print "%(id)s,%(route)s,\"%(direction)s\"," "%(latitude)s,%(longitude)s" % busdict
        buses.remove(elem)

第四部分：数据处理及并发编程

我们可以把协程包装进线程和子进程中，如下图：

多线程版本

# cothread.py
#
# A thread object that runs a coroutine inside it.  Messages get sent
# via a Queue object

from threading import Thread
from Queue import Queue
from coroutine import *


@coroutine
def threaded(target):
    messages = Queue()

    def run_target():
        while True:
            item = messages.get()
            if item is GeneratorExit:
                target.close()
                return
            else:
                target.send(item)

    Thread(target=run_target).start()
    try:
        while True:
            item = (yield)
            messages.put(item)
    except GeneratorExit:
        messages.put(GeneratorExit)


# Example use

if __name__ == '__main__':
    import xml.sax
    from cosax import EventHandler
    from buses import *

    xml.sax.parse(
        "allroutes.xml",
        EventHandler(
            buses_to_dicts(
                threaded(
                    filter_on_field(
                        "route", "22",
                        filter_on_field(
                            "direction", "North Bound",
                            bus_locations()
                        )
                    )
                )
            )
        )
    )

该例子概览图如下：

作者测试，用多线程后，速度慢了 50%。

子进程版本

# coprocess.py
#
# An example of running a coroutine in a subprocess connected by a pipe

import cPickle as pickle
from coroutine import *


@coroutine
def sendto(f):
    try:
        while True:
            item = (yield)
            pickle.dump(item, f)
            f.flush()
    except StopIteration:
        f.close()


def recvfrom(f, target):
    try:
        while True:
            item = pickle.load(f)
            target.send(item)
    except EOFError:
        target.close()


# Example use
if __name__ == '__main__':
    import xml.sax
    from cosax import EventHandler
    from buses import *

    import subprocess

    p = subprocess.Popen(['python', 'busproc.py'], stdin=subprocess.PIPE)

    xml.sax.parse(
        "allroutes.xml",
        EventHandler(
            buses_to_dicts(
                sendto(p.stdin)
            )
        )
    )

# busproc.py
#
# Bus processor.  This runs as a subprocess from the coprocess.py example

import sys
from coprocess import recvfrom
from buses import *

recvfrom(
    sys.stdin,
    filter_on_field(
        "route", "22",
        filter_on_field(
            "direction", "North Bound",
            bus_locations()
        )
    )
)

该例子概览图如下：

这里需要注意的是，用多线程和多进程不一定就能够加快速度，需要仔细测试。

隐藏的威胁

向正在执行的协程发送数据

调用 send() 向正在执行的协程发送数据会导致程序崩溃，如用多线程同时向一个协程发送数据：

# cocrash.py
#
# An example of hooking coroutines up in a way that might cause a potential
# crash.   Basically, there are two threads feeding data into the
# printer() coroutine.    

from cobroadcast import *
from cothread import threaded

p = printer()
target = broadcast([threaded(grep('foo', p)), threaded(grep('bar', p))])

# Adjust the count if this doesn't cause a crash
for i in xrange(10):
    target.send("foo is nice\n")
    target.send("bar is bad\n")

del target
del p

协程之间形成环

如下图，在协程之间形成环也会不行的：