文档章节

Kqueue 实现非阻塞 Socket 通信

xh4n3
 xh4n3
发布于 2015/11/03 19:00
字数 1231
阅读 1315
收藏 24

如果有误,请大神指出啊!

--

之前留下的坑

之前写过一篇 kqueue 实现文件操作监控,讲了 Kqueue 在文件监控的应用,文章给出的例子只对于一个 test 文件进行监控。

Kqueue 或者 Epoll 更多的是被使用在 Socket 通信的场景中,于是我又写了一个带有 Socket 的版本。代码如下:

# coding=utf-8
import select
from socket import socket
from socket import AF_INET, SOCK_STREAM, SOL_SOCKET, SO_REUSEADDR
from threading import Thread

fd = open('test')
s = socket(AF_INET, SOCK_STREAM)
s.bind(("127.0.0.1", 3000))
s.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
s.listen(1)
kq = select.kqueue()
flags = select.KQ_EV_ADD | select.KQ_EV_ENABLE | select.KQ_EV_CLEAR
fflags = select.KQ_NOTE_DELETE | select.KQ_NOTE_WRITE | select.KQ_NOTE_EXTEND \
         | select.KQ_NOTE_RENAME

# 监测文件事件,如果有新事件在这个 fd 上发生,则返回,监测事件类型由 fflags 规定
file_ev = select.kevent(fd.fileno(), filter=select.KQ_FILTER_VNODE, flags=flags, fflags=fflags)

# 监测 Socket 事件,如果有新数据可读则返回
socket_ev = select.kevent(s.fileno(), filter=select.KQ_FILTER_READ, flags=flags)

# 监测多个对象就只需把很多 kevent 对象塞进 events 列表中,然后传递给 control 函数
events = []
events.append(file_ev)
events.append(socket_ev)


# 处理这个 socket 请求
def socket_handler(cl):
    while True:
        data = cl.recv(100)
        print data
        if not data:
            cl.close()
            print 'socket closed'
            break

while True:
    revents = kq.control(events, 1, None)
    for e in revents:
        # 如果是 socket 触发的事件
        if e.ident == s.fileno():
            print 'Event from socket'
            if e.filter & select.KQ_FILTER_READ:
                cl, _ = s.accept()
                # 如果直接调用 socket_handler 函数,那么这个 eventloop 会被阻塞,所以此处使用线程
                Thread(None, socket_handler, args=(cl,)).start()
            else:
                print e
        # 如果是文件触发的事件
        if e.ident == fd.fileno():
            print 'Event from file'
            if e.fflags & select.KQ_NOTE_EXTEND:
                print 'extend'
            elif e.fflags & select.KQ_NOTE_WRITE:
                print 'write'
            elif e.fflags & select.KQ_NOTE_RENAME:
                print 'rename'
            elif e.fflags & select.KQ_NOTE_DELETE:
                print 'delete'
            else:
                print e

然而随时时间的推移,我发现其实这是有大大的问题的,我自作聪明地为每个新链接产生一个线程应付 Client,然而如果当 Client 源源不断地涌入时,线程数会超标,导致程序发送错误。就算我们维护一个线程的列表,使列表的长度不大于某个标准,每次创建线程的损耗也是存在的。

性能对比

首先我们建立了一个 KqueueEventLoop 的循环类,代码如下:

# 部分代码参考了 SS 源码中的实现
class KqueueEventLoop(object):

    KQ_FILTER_READ = select.KQ_FILTER_READ

    def __init__(self):
        self._fd_map = {}
        self._handler_map = {}
        self._event_map = {}
        self.kq = select.kqueue()
        self.klist = []
        self._stop = False

    # 启动这个事件循环
    def run(self):
        while not self._stop:
            events = self.poll()
            for e in events:
                self._fd_map[e.ident](self._handler_map[e.ident])

    # 从事件池里面取事件出来
    def poll(self):
        events = self.kq.control(self.klist, 1, None)
        return events

    # 注册事件
    def add(self, f, mode, handler):
        fd = f.fileno()
        event = select.kevent(fd, filter=mode, flags=select.KQ_EV_ADD | select.KQ_EV_ENABLE | select.KQ_EV_CLEAR)
        self._handler_map[fd] = f
        self._fd_map[fd] = handler
        self._event_map[fd] = event
        self.klist.append(event)

    # 删除事件
    def remove(self, f):
        fd = f.fileno()
        del self._handler_map[fd]
        del self._fd_map[fd]
        self.klist.remove(self._event_map[fd])

    # 暂停事件循环
    def stop(self):
        self._stop = True

如果是用一开始的代码的想法,为每个 Client 生成一个线程,调用方法如下:

def test():
    loop = KqueueEventLoop()
    s = socket(AF_INET, SOCK_STREAM)
    s.bind(("127.0.0.1", 3000))
    s.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    s.listen(5)

    def callback(f):
        Thread(None, handler, args=(f,)).start()

    def handler(f):
        print 'INFO: New connection established.'
        cl, _ = f.accept()
        while True:
            data = cl.recv(1024)
            if not data:
                print 'INFO: Connection dropped.'
                loop.remove(cl)
                cl.close()
                return
            print 'DATA: %s' % repr(data)

    loop.add(s, KqueueEventLoop.KQ_FILTER_READ, callback)
    loop.run()

if __name__ == '__main__':
    test()

我们模拟了一个 100 个客户端,每隔 0.1 秒向服务器发 1,获取到该服务端内存使用量为 14MB。 之前提到这边线程的产生会造成两个严重的问题,所以我们就不生成线程。那么怎么做到原来的 while 循环来接受这么多客户端发来的数据呢?答案是,不用 while 循环。

继续注册事件

当事件池返回一个新连接建立请求时,接受并建立这个连接,再把这个连接扔回事件池中,代码如下:

def test():
    loop = KqueueEventLoop()
    s = socket(AF_INET, SOCK_STREAM)
    s.bind(("127.0.0.1", 3000))
    s.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    s.listen(5)

    def handler(f):
        print 'INFO: New connection established.'
        cl, _ = f.accept()
        cl.setblocking(False)
        loop.add(cl, KqueueEventLoop.KQ_FILTER_READ, read_data)

    def read_data(cl):
        try:
            data = cl.recv(1024)
            if not data:
                print 'INFO: Connection dropped.'
                loop.remove(cl)
                cl.close()
                return
            print 'DATA: %s' % repr(data)
        except Exception, e:
            print 'ERROR: %s' % repr(e)
            loop.remove(cl)
            cl.close()

    loop.add(s, KqueueEventLoop.KQ_FILTER_READ, handler)
    loop.run()

if __name__ == '__main__':
    test()

我们通过 loop.add(cl, KqueueEventLoop.KQ_FILTER_READ, read_data) 将每个新连接 cl 加入到事件循环 loop 中,并让事件循环类记录回调方法 read_data。此时的状况就是,整个程序只有一个事件循环,每来一个新连接,通过 handler 回调建立该连接。再如果网卡接受到新数据,KqueueEventLoop 找到新数据对应的连接,把数据读入内存并打印出来。

以上实现没有了新建线程带来的损耗,整个程序只保留了一个 while 循环,使得相同情况下内存使用量仅为 5MB。并且并发连接数量有很大提升,不再受限于线程数限制。

其他

在 IO 密集型程序中,其实线程的使用是可以改善程序运行速度的。但线程还是要用在可控的地方,比如用线程去跑事件循环。

© 著作权归作者所有

下一篇: Notes on Generator 2
xh4n3
粉丝 14
博文 28
码字总数 16847
作品 0
杭州
程序员
私信 提问
使用 kqueue 在 FreeBSD 上开发高性能应用服务器

kqueue 是 FreeBSD 上的一种的多路复用机制。它是针对传统的 select/poll 处理大量的文件描述符性能较低效而开发出来的。注册一批描述符到 kqueue 以后,当其中的描述符状态发生变化时,kqu...

红薯
2010/05/21
1K
0
[连载] Socket 深度探索 4 PHP (一)

Socket(套接字)一直是网络层的底层核心内容,也是 TCP/IP 以及 UDP 底层协议的实现通道。随着互联网信息时代的爆炸式发展,当代服务器的性能问题面临越来越大的挑战,著名的 C10K 问题(h...

长平狐
2012/11/19
218
0
高性能web服务器的秘密核武器

最近kangle web服务器已经发布了新版2.3.1,其性能比老版本提升8倍之多,静态文件处理能力达apache的8-10倍。如此高的性能怎么来的 呢?kangle有哪些秘密武器呢?其实作为现代化的其它web服务...

keengo
2011/08/20
674
1
Nginx 的缓存模块--srcache

我们知道,Nginx的核心设计思想是事件驱动的非阻塞I/O。Nginx被设计为可以配置I/O多路复用策略,在Unix系统中传统的多路复用是采用select或poll,但是这两个方法的问题是随着监听socket的增加...

章亦春
2013/04/27
3.7K
0
I/O复用机制概述

接下来我们将介绍几种常见的I/O模型及其区别 blocking I/O nonblocking I/O I/O multiplexing (select and poll) signal driven I/O (SIGIO) asynchronous I/O (the POSIX aio_functions) b......

linuxprobe
2016/09/24
22
0

没有更多内容

加载失败,请刷新页面

加载更多

OSChina 周四乱弹 —— 当你简历注水但还是找到了工作

Osc乱弹歌单(2019)请戳(这里) 【今日歌曲】 @花间小酌 :#今日歌曲推荐# 分享成龙的单曲《男儿当自强》。 《男儿当自强》- 成龙 手机党少年们想听歌,请使劲儿戳(这里) @hxg2016 :刚在...

小小编辑
今天
2.9K
22
靠写代码赚钱的一些门路

作者 @mezod 译者 @josephchang10 如今,通过自己的代码去赚钱变得越来越简单,不过对很多人来说依然还是很难,因为他们不知道有哪些门路。 今天给大家分享一个精彩的 GitHub 库,这个库整理...

高级农民工
昨天
5
0
用好项目管理工具,人人都可以成为项目经理

现在市面上的项目管理工具越来越多了,但是大多数都是一些协同工具或轻量项目管理工具。如果是多团队、跨部门使用或者企业级的项目管理,从管理思想到工具运用,需要适应企业的业务流程体系,...

cs平台
昨天
12
0
只需一步,在Spring Boot中统一Restful API返回值格式与统一处理异常

统一返回值 在前后端分离大行其道的今天,有一个统一的返回值格式不仅能使我们的接口看起来更漂亮,而且还可以使前端可以统一处理很多东西,避免很多问题的产生。 比较通用的返回值格式如下:...

晓月寒丶
昨天
69
0
区块链应用到供应链上的好处和实际案例

区块链可以解决供应链中的很多问题,例如记录以及追踪产品。那么使用区块链应用到各产品供应链上到底有什么好处?猎头悬赏平台解优人才网小编给大家做个简单的分享: 使用区块链的最突出的优...

猎头悬赏平台
昨天
32
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部