Windows 开发日记
背景： 把一个多线程网络软件移植到Windows上。
基本简介：
- 软件类似代理。转发上层应用UDP数据包，仅添加一些字段在数据头。
- 每一条服务端发过来的都是消息，有边界，不是字节流。
- 线程A负责接收数据(libpcap抓包），线程B负责发送数据和处理数据。线程A收到的数据要传给线程B处理。数据流向：线程A->线程B->上层应用。
问题：
- Mac/Linux上速度很快，Windows上面速度极其慢。非常诡异的是，通过流量监控，发现软件下载速度非常快，但上层应用收到的数据的速度极其慢。
过程：
1. UDP
在linux上，线程A和B数据收发是通过unix域套接字。
在Windows，由于消息是有边界的，所以自然想到了同步方式是UDP，自然走loopback比较合适。。没有选择命名管道、完成端口的原因是：软件的消息队列用的是libuv，对这两种SOCK_DGRAM方式的通信不支持。
使用UDP后发现在Windows上，软件速度极其慢。为了比较，在linux上把通信方式也切换为UDP，发现linux上速度仍旧很快。
2. Winpcap不支持某些无线网卡抓包
考虑到有些无线网卡不支持抓包，所以造成没有速度。改用有线连接的时候，发现果真能抓数据了。但速度仍然很慢。通过网络监控发现，软件的网络速度非常快，但应用层基本没速度。
3. Windows loopback上的套接字缓冲很小
线程A发送一个数据，我就把cnt1增1，线程B收到一个数据我就把cnt2增1. 运行一会儿就发现，cnt2 只有cnt1的10%不到。意思是丢包率超过90%。
由于以前从来没有在Windows上开发过，实在想不通为什么丢包率这么高。
折腾了几天找不出原因后，网上问了一下，有人说Windows loopback队列很小。问题简单了，把两个udp socket的buf都增大，直接设置成了10M。情况好多了，但运行久了，软件接受数据变快以后，丢包又开始上升。这说明通过简单的扩大buf size行不通。
4. 线程A和线程B使用可靠传输
既然在loopback下udp丢包率这么高，自然而然的想到了使用TCP。问题是TCP是面向流的，是没有消息边界的。
TCP消息分界有如下几种方式，解决这个问题：
- 发送固定大小的长度N。消息的长度在数据的最前端。接收端先读取长度len（比如4个字节的int)，然后读取N - 4个字节。后一次读取的len字节就是数据。数据以后的字节全部丢弃。
- 用分界符。比如用x12341234来区分消息。如果数据中本身包含分界符，那么再用一个分界符去转移。这种在以太网中也有应用。
- 按
数据长度+数据组成来发送，大小不固定。
第一种方式，比较容易，但容易浪费带宽。
第二种方式，比较费时间，每一次都要遍历数据，不可取。
第三种方式，逻辑比较复杂，比如读取消息可能不完整，读取了几个消息。任何一次出错会造成后面所有的数据都出错。要求正确率100%。
最终选择了第三种方式。然后再也没有出现软件速度飞快，上层APP没速度的情况了。
5. 实现
下面是部分代码。
发送端：
int PacketStream::Send(int nread, const rbuf_t &rbuf) {
if (nread > 0) {
char buf[MAX_BUF_SIZE];
*(COUNT_TYPE *)buf = nread;
char *p = buf + COUNT_SIZE;
memcpy(p, rbuf.base, nread);
assert(nread + COUNT_SIZE <= MAX_BUF_SIZE);
return doSend(buf, COUNT_SIZE + nread);
}
return nread;
}
int PacketStream::doSend(const char *buf, int nread) {
if (nread > 0) {
// LOGD << "nread: " << nread;
assert(nread < OM_MAX_PKT_SIZE);
int nret = send(mWriteSock, buf, nread, 0);
LOGE_IF(nret < 0) << "send error: " << strerror(errno) << ", nret: " << nret;
return nret;
}
return nread;
}
接收端：
int PacketStream::rawInput(int nread, const char *buf) {
if (nread > 0) {
assert(mNextLen <= MAX_BUF_SIZE);
if (nread > (RSOCK_UV_MAX_BUF + MAX_BUF_SIZE)) {
LOGE << "nread too large: " << nread;
assert(0);
}
const int totLen = nread + mBufLen; // data in buf and in mTempBuf, not including length(4) of mBufLen
//LOGD << "mNextLen: " << mNextLen << ", mBufLen: " << mBufLen << ", nread: " << nread << ", totLen: " << totLen;
if (totLen > mNextLen) { // totLen > mNextLen. copy two buf into one
char *pbuf = mLargeBuf; // mLargeBuf is large enough to store both mTempBuf and buf
int left = totLen;
if (mNextLen > 0) { // if nextLen valid. prepend it to buf.
*(COUNT_TYPE*)pbuf = mNextLen;
pbuf += COUNT_SIZE;
left += COUNT_SIZE; // DON'T FORGET THIS !!!!
}
memcpy(pbuf, mTempBuf, mBufLen);
pbuf += mBufLen;
memcpy(pbuf, buf, nread);
const char *p = mLargeBuf;
int nret = 0;
std::vector<int> head;
// now mLargeBuf always starts with len information
while (left >= 0) {
if (left < COUNT_SIZE) { // left is only part of int, prevent invalid memory access.
markTempBuf(0, p, left);
// LOGD << "mNextLen: " << mNextLen << ", left: " << left << ", mBufLen: " << mBufLen;
break;
}
COUNT_TYPE nextLen = *(COUNT_TYPE *)p;
//LOGD << "nextLen: " << nextLen << ", left: " << left;
head.push_back(nextLen);
assert(nextLen > 0 && nextLen < MAX_BUF_SIZE);
p += COUNT_SIZE;
left -= COUNT_SIZE;
if (nextLen <= left) {
nret = Input(nextLen, new_buf(nextLen, p, nullptr));
left -= nextLen;
p += nextLen;
//LOGD << "nret: " << nret << ", nextLen: " << nextLen << ", left: " << left << ", p - mLargeBuf: " << (p - mLargeBuf);
if (left >= COUNT_SIZE) {
COUNT_TYPE temp = *(COUNT_TYPE*)p;
if (temp <= 0 || temp > OM_MAX_PKT_SIZE) {
LOGE << "temp error: " << temp;
}
}
} else {
markTempBuf(nextLen, p, left);
//LOGD << "nextLen: " << nextLen << ", left: " << left;
break;
}
}
return nret;
} else if (totLen < mNextLen) { // if not enough data. just copy
memcpy(mTempBuf + mBufLen, buf, nread);
mBufLen += nread;
//LOGD << "less than: mNextLen: " << mNextLen << ", mBufLen: " << mBufLen << ", nread: " << nread;
return 0;
} else { // equal: (totLen == mNextLen)
memcpy(mTempBuf + mBufLen, buf, nread);
// LOGD << "input: " << totLen;
int n = Input(totLen, new_buf(totLen, mTempBuf, nullptr));
markTempBuf(0, nullptr, 0);
//LOGD << "euqal, after input: mNextLen: " << mNextLen << ", mBufLen: " << mBufLen << ", nread: " << nread;
return n;
}
} else if (nread == UV_EOF) {
markTempBuf(0, nullptr, 0); // todo: when error occurs, should close this synconn and create a new one
LOGE << "nread = EOF";
} else if (nread < 0) {
LOGE << "error: " << uv_strerror(nread);
assert(0); // todo: when error occurs, should close this synconn and create a new one
}
return nread;
}
