/* fslider_ws - Rainy部分源码 */
// data_recv()
// 内部读缓冲区
_buffer.r_queue = Buffer.concat([_buffer.r_queue, data]);
// the "while loop" gets and parses every entire frame remain in buffer
// 循环调用解析器
while (readable_data = wsframe.parse(_buffer.r_queue)) {
FIN = readable_data['frame']['FIN'],
Opcode = readable_data['frame']['Opcode'],
MASK = readable_data['frame']['MASK'],
Payload_len = readable_data['frame']['Payload_len'];
// if recive frame is in fragment
if (!FIN) {
// save the first fragment's Opcode
if (Opcode) _buffer.Opcode = Opcode;
// 处理分片
_buffer.f_payload_data.push(readable_data['frame']['Payload_data']);
} else {
payload_data = readable_data['frame']['Payload_data'];
// don't fragment or the last fragment
// translate raw Payload_data
switch (Opcode) {
// continue frame
// the last fragment
case 0x0:
// 最后一个分片
_buffer.f_payload_data.push(payload_data);
payload_data = Buffer.concat(_buffer.f_payload_data);
// when the whole fragment recived
// get the Opcode from _buffer
Opcode = _buffer.Opcode;
// init the fragment _buffer
_buffer.f_payload_data = [];
// system level binary data
case 0x2:
head_len = payload_data.readUInt8(0);
event = payload_data.slice(1, head_len + 1).toString();
rawdata = payload_data.slice(head_len + 1);
//client.sysEmit(event, rawdata);
break;
}
}
// the rest buffered data
// 每轮解析后余下的数据(每次解析一个frame)
_buffer.r_queue = readable_data.r_queue;
}
/* v0.4.x - RocketEngine */
// transformer
self.r_queue = Buffer.concat([self.r_queue, chunk]);
// the "while loop" gets and parses every entire frame remain in buffer
while (readable_data = wsframe.parse(self.r_queue)) {
FIN = readable_data['frame']['FIN'],
Opcode = readable_data['frame']['Opcode'],
MASK = readable_data['frame']['MASK'],
Payload_len = readable_data['frame']['Payload_len'];
payload_data = readable_data['frame']['Payload_data'];
// if recive frame is in fragment
if (!FIN) {
// save the first fragment's Opcode
if (Opcode) {
switch (Opcode) {
case 0x1:
type = 'text';
break;
case 0x2:
type = 'binary';
}
// 反注释这里就会开启缓存模式
//self.handleFragment(dispatch);
client.sysEmit('firstfragment', { type: type, f_data: payload_data });
}
else
client.sysEmit('fragment', payload_data);
} else {
// don't fragment or the last fragment
// translate raw Payload_data
switch (Opcode) {
// continue frame
// the last fragment
case 0x0:
client.sysEmit('lastfragment', payload_data);
break;
// system level binary data
case 0x2:
subParser.binaryParser(payload_data, dispatch);
break;
}
}
// the rest buffered data
self.r_queue = readable_data.r_queue;
}
高并发和大数据里面挑战无处不在, 花样层出不穷, 总能从中挖掘出新的知识,获得新的技能. 因此也成为架构师和追求程序性能的黑客津津乐道的话题. 再加上"实时"这一前卫特性, 整个程序就更加暗流涌动,险象迭生了.
所谓险象迭生包括内存占用,核心架构,内存泄露,CPU占用和程序耗时. 大数据算法就对内存占用和程序耗时要求特别严格, 而高并发一定会挑战资源占用的极限.
数据遍历问题
今年10月份有一段时间我集中精力处理Bitcoin Address的提取与更新问题. 提取到levledb中的数据集大概有2GB, 全都是BTC地址, 目的是方便后续基于账户地址的数据分析, 比如最简单的"统计Top 100 BTC Address", 我需要对数据集中的每条地址做更新, 这需要调用insight-api的计算函数填充空白地址的balance等等详细信息, 而这个函数内部又需要查询两个数据库, 在经过层层回调和数学计算才能得到一条地址的信息. 其实单看这个问题并不复杂. 有不错的解决方案,比如并行查询更新, 但对当前情况并不适用.
Insight内部使用的是leveldb, 分别为区块和交易各建立一个数据库, key有不同的几种, 包含各种不同的元信息的组合, 每次查询时, 首先要对key的元信息进行检索, 然后截取这个元信息或者获取对应的value. 你可能会问为什么弄得这么繁琐? 主要因为这样做能减小数据库尺寸降低硬盘空间的占用, 但这么做导致的后果就是性能打了折扣. 除此之外, leveldb本身并不支持一个进程多次打开同一实例, Insight为了解决Node.js模块调用导致的多次打开同一实例的问题, 给出了一个即巧妙又蛋疼的方案, 巧妙的是这样的确避免了多次打开,蛋疼的是给功能的扩展带来极大的麻烦. 我从提取地址到更新地址有一般时间花在这上面. 后来找到了多进程访问leveldb的模块, 然后对Insight的设计大换血... 但仍是multi-leveldb模块却带来了性能瓶颈以及更多让人崩溃的问题. 也就是说, 好的方案需要我对Insight原有的代码改头换面才行!
说了这么多, 最核心的问题是什么? 仅仅是性能? 当然不. 在写代码的过程中, 我发现对数据库的遍历会引起内存占用的不断增长, 理论上来说遍历每条数据, 内存占用应该基本不变,并且很低很低, 因为数据库设计上不应该在内存中缓存遍历过的数据. 莫非是内存泄露? 经过反复的检查和测试了代码, 我否定了内存泄露的可能. 那究竟是怎么回事, 想了很久, 推测是因为长时间的循环导致JavaScript主线程一直被占用, 因而垃圾回收机制无法发挥功效. 也就是说, 垃圾回收的速度赶不上内存申请的速度. 这种情况在我换用了MongoDB后效果更加显著...
但是内存调整到16GB后, 新的问题又出现了. leveldb的单个数据库会将数据分散到分片文件中存储, 而数据集过快的遍历导致leveldb需要不断打开新的文件分片. 这些分片有57W个, 打开的这些分片为了其他时候使用而不会关闭. 想想60多W个文件描述符啊. 出现IO Error是迟早的事了.为此我把ulimit的最大文件打开数设置了很大, 大到系统拒绝设置, 但是仍旧抵不过文件描述符疯狂的增长...
高并发情景下大数据的实时传输
这个月初考完编译和算法, 清闲的很, 于是答应帮人做一个Demo, 顺便赚点外快. 其中一个功能是类似微信的chat. 大二曾设计过基于WebSocket的web聊天应用, 其应用逻辑处理的难点无非是身份确认, 消息提醒外加历史记录. 而穿插在这些业务逻辑之间的realTime协议已经有现成的库了. 我原计划直接上Socket.IO, 但碍于Socket.IO没法传输二进制数据, 并且使用的不太熟练, 所以我把自己写的websocket库用上去了, 这样业务层面的逻辑写起来就容易多了.(现在Socket.IO支持二进制数据了, 刚看到最新说明)
不过在大文件大传送过程中出现的bug使我重新思考了这个库的核心设计. 如果所有实时应用都像这个demo一样只是交换一些少量文本和不超过10M的图片, 那么整个世界都"real-time"起来也没问题了, 但产品环境下的realTime免不了巨大而频繁的数据交互, 一个应用底层的库是否经得起考验就要看它在极端环境下的性能(响应速度)以及对系统资源(如内存,CPU,fd,外设I/O)的占用情况.
就拿这个demo来说, 单次传输400MB左右的blob, 内存占用疯狂的上涨, 粗略的内存检测如下:
从应用启动到开始传输:![img1](https://github.com/abbshr/abbshr.github.io/raw/master/source/img/2014-12-24%2022:11:12%20%E7%9A%84%E5%B1%8F%E5%B9%95%E6%88%AA%E5%9B%BE.png)
可以看到Node进程的内存占用从27MB很快蹿升到420MB, 而后:
更是达到了难以想象的1.67GB,
top
的实时检测也显示了这一问题, 作为JavaScript写手我的直觉告诉我一定是内存泄露:从大上一幅图中可以看出, 当400MB的文件全部传完, 进程的内存占用居高不下, 但请注意图中最下方的log信息, 内存占用变成了700MB, 而恰恰在此时我上传了第二个文件, 大小是20MB. 随后的跟踪日志显示如下:
在20MB文件上传完毕之后, 内存占用变成了600MB并保持不变了. 又经过了几个小文件的测试, 内存占用降低到了500多MB, 这就直接否定了"内存泄露"的猜测, 因为内存泄露往往不能回收内存空间.
遇到内存占用不断提升这种情况, 你肯定在最开始想到的是文件缓存到内存中了, 因为这很正常, 如果你把所有数据都放到buffer里必然会导致内存占用量暴增并居高不下. 但注意图中的日志信息: 如果真的是内存中缓存文件导致的, 结果未免太离谱了吧. 一个400MB的文件会让内存占用从27MB增长到1.7GB?
在后续的测试中, 我注释掉了回传那部分代码(因为这部分是面临高度写压力的模块, 为了防止排除影响所以暂时关掉), 但效果仍是一样的, 这就说明了导致内存占用突变的是读模块. 这让我不得不重审源码, 一步步跟踪内存的申请情况.
这个读模块类似
transform stream
, 作为TCP socket的'data'事件回调函数而存在, 每次data
事件事件触发, transformer都会被调用, 然后不断的从底层缓存中抽取完整的websocket frame进行解析.若是单个frame大小超过了分片大小(默认1KB), 则分片发送. 而经过分析发现, 问题也就出现在这里. 下面是v0.3.x版本中数据接收部分出现问题的代码:
问题就出现在我附加中文注释的地方, 下面分析内存的申请和释放情况:
每当'data'事件触发, 假设均能提取出来至少一个frame. 首先data_recv函数会
concat
一个新的buffer,就是原有缓存加上新的数据, 设它为buf_1
.随后进入循环阶段,
wsframe.parse()
函数经过解析会返回结果数据和余下的缓存, 设他们为buf_2
,buf_3
. 从而有buf_1 ≈ buf_2 + buf_3
. 紧接着, 如果buf_2属于分片, 则把他缓存到另一个分片队列f_payload
里,否则经过第二次解析, 生成一个新的rawdata
.我们可以认为rawdata = buf_2
这样下来, 每轮循环会得到: 余下数据的拷贝
buf_3
, 新的framebuf_2
,buf_2
的拷贝rawdata
或新的f_payload
.如果没有分片的话, 那么一次传输的内存占用就是
buf2 * 3
, 如果有分片, 那么每次frame的解析占用的内存将会增加buf_2 * 2 + buf_3 + f_payload
, 假设一帧的大小是120KB, 剩余缓存是360KB, 那么第一次解析下来的增量就是5 * buf_2
, 下一次的增量就是buf_2 * 2 + buf_2 * 2 + buf_2 * 2 = 6 * buf_2
... 你可能会质疑, 每一轮buf_2
不是可以被内存回收的么, 但请注意f_payload
的缓存机制导致在整个帧接受完之前是不会释放的! 也就是每次分片会创建一个新的更大的缓存.我们忽略data
事件触发时新申请的内存, 仅凭f_payload
就足够说明问题了.类比"等差数列的前N项和",内存的占用也是这个道理.这也就解释了为何上传一个400MB的文件内存能增长到1个多GB.
问题明晰之后, 我重构了核心代码, 移除v0.3.x版本中的缓存机制(store mode), 在v0.4.x版本中默认换做outflow mode(流失模式), 也就是仅仅触发分片事件, 并不在内部缓存他们, 这也使整个框架更简洁更灵活:
经过改进后, 测试结果如下:
仍旧是上传400MB的文件, 从开始到上传结束内存占用始终在150MB左右:
之后的多终端同时上传也能让内存控制在200MB左右, 这说明数据接收的内存占用问题得以解决.