iocp 多次投递保证序列吗？

papamms

unread,

Dec 17, 2006, 11:13:47 AM12/17/06

to 高性能网络编程邮件列表

我改了一下微软的例子代码。我现在做的事情是
读一个2M 或者更大的文件到ram.
当有连接的时候，就把所有的数据都发给那个连接。
因为文件太大了。所以要多次投递

我的代码如下

// Module Name: iocmplt.cpp
//
// Description:
//
// This sample illustrates how to develop a simple echo server
Winsock
// application using the completeion port I/O model. This
// sample is implemented as a console-style application and simply
prints
// messages when connections are established and removed from the
server.
// The application listens for TCP connections on port 5150 and
accepts them
// as they arrive. When this application receives data from a
client, it
// simply echos (this is why we call it an echo server) the data
back in
// it's original form until the client closes the connection.
//
// Compile:
//
// cl -o iocmplt iocmplt.cpp ws2_32.lib
//
// Command Line Options:
//
// iocmplt.exe
//
// Note: There are no command line options for this sample.

#include <winsock2.h>
#include <windows.h>
#include <stdio.h>
#include <stdlib.h>

#define PORT 5150
#define DATA_BUFSIZE 8192

typedef struct
{
OVERLAPPED Overlapped; // ÖØµþio
WSABUF DataBuf; // WSABUF
CHAR *Buffer;//
ÕæÊµµÄÊý¾ÝÄÚÈÝ
DWORD BytesSEND; //
ËÍ³öÈ¥µÄÊý¾ÝÊýÁ¿
DWORD BytesRECV; //
ÊÕµ½µÄÊý¾ÝÊýÁ¿
} PER_IO_OPERATION_DATA, * LPPER_IO_OPERATION_DATA;

class usera
{
public:
SOCKET Socket;
unsigned int sendtimes;
};

typedef struct
{
SOCKET Socket;
usera *pusers;

} PER_HANDLE_DATA, * LPPER_HANDLE_DATA;

DWORD WINAPI ServerWorkerThread(LPVOID CompletionPortID);

char *Pdata = NULL;
CRITICAL_SECTION cs ;

void lock()
{
EnterCriticalSection (&cs);
}

void unlock()
{
LeaveCriticalSection (&cs);
}
void main(void)
{
SOCKADDR_IN InternetAddr;
SOCKET Listen;
SOCKET Accept;
HANDLE CompletionPort;
SYSTEM_INFO SystemInfo;
LPPER_HANDLE_DATA PerHandleData;
LPPER_IO_OPERATION_DATA PerIoData;
unsigned int i;
DWORD RecvBytes;
DWORD Flags;
DWORD ThreadID;
WSADATA wsaData;
DWORD Ret;
FILE * datafile;

unsigned int filelen = 0;
InitializeCriticalSection (&cs) ;

datafile = fopen("1.txt", "rb");
if (!datafile) return ;
fseek(datafile, 0, 2);
filelen = ftell(datafile);

Pdata = new char [filelen +1];
Pdata[filelen] = 0;
fseek(datafile, 0, 0);
filelen = 0;

while ( fread(Pdata+filelen, 1, 1, datafile) == 1)
{
filelen+=1;
}

fclose(datafile);
Pdata[filelen] = 0;
if ((Ret = WSAStartup(0x0202, &wsaData)) != 0)
{
printf("WSAStartup failed with error %d\n", Ret);
return;
}

// Setup an I/O completion port.
if ((CompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE,
NULL, 0, 0)) == NULL)
{
printf( "CreateIoCompletionPort failed with error: %d\n",
GetLastError());
return;
}

// Determine how many processors are on the system.

GetSystemInfo(&SystemInfo);

// Create worker threads based on the number of processors available
on the
// system. Create two worker threads for each processor.

for(i = 0; i < SystemInfo.dwNumberOfProcessors * 2; i++)
{
HANDLE ThreadHandle;

// Create a server worker thread and pass the completion port to the
thread.

if ((ThreadHandle = CreateThread(NULL, 0, ServerWorkerThread,
CompletionPort,
0, &ThreadID)) == NULL)
{
printf("CreateThread() failed with error %d\n", GetLastError());
return;
}

// Close the thread handle
CloseHandle(ThreadHandle);
}

// Create a listening socket
½¨Á¢ÕìÌýsocket

if ((Listen = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0,
WSA_FLAG_OVERLAPPED)) == INVALID_SOCKET)
{
printf("WSASocket() failed with error %d\n", WSAGetLastError());
return;
}

InternetAddr.sin_family = AF_INET;
InternetAddr.sin_addr.s_addr = htonl(INADDR_ANY);
InternetAddr.sin_port = htons(PORT);

// °Ñ¶Ë¿Úbind
ÉÏÈ¥

if (bind(Listen, (PSOCKADDR) &InternetAddr, sizeof(InternetAddr)) ==
SOCKET_ERROR)
{
printf("bind() failed with error %d\n", WSAGetLastError());
return;
}

// Prepare socket for listening

if (listen(Listen, 5) == SOCKET_ERROR)
{
printf("listen() failed with error %d\n", WSAGetLastError());
return;
}

// Accept connections and assign to the completion port.

while(TRUE) //
ÓÀÔ¶¿ªÊ¼Ñ»·
{
if ((Accept = WSAAccept(Listen, NULL, NULL, NULL, 0)) ==
SOCKET_ERROR)
{
printf("WSAAccept() failed with error %d\n", WSAGetLastError());
return;
}
// accept Ò»¸öclient

usera *puser = new usera;
puser->sendtimes = 0;
puser->Socket = Accept;

unsigned int send_times = 0;

while ( send_times < filelen) {

if ((PerHandleData = (LPPER_HANDLE_DATA) GlobalAlloc(GPTR,
sizeof(PER_HANDLE_DATA))) == NULL)
{
printf("GlobalAlloc() failed with error %d\n", GetLastError());
return;
}

if (send_times == 0) {

if (CreateIoCompletionPort((HANDLE) Accept, CompletionPort, (DWORD)
PerHandleData,
0) == NULL)
{
printf("CreateIoCompletionPort failed with error %d\n",
GetLastError());
return;
}

}

PerHandleData->pusers = puser;

// Create per I/O socket information structure to associate with the

// WSARecv call below.

lock();

if ((PerIoData = (LPPER_IO_OPERATION_DATA) GlobalAlloc(GPTR,
sizeof(PER_IO_OPERATION_DATA))) == NULL)
{
printf("GlobalAlloc() failed with error %d\n", GetLastError());
return;
}

ZeroMemory(&(PerIoData->Overlapped), sizeof(OVERLAPPED));//
½«io¶Ë¿Ú
ÖÃÎª¿Õ
PerIoData->BytesSEND = 0;// send
×Ö½ÚÎª0
PerIoData->BytesRECV = 0;//
ÊÕµ½×Ö½Ú

if ((filelen -send_times)>= 1000) {

PerIoData->Buffer= Pdata + send_times;
PerIoData->DataBuf.len = 1000;//
BufferµÄ³¤¶È
send_times += 1000;

}
else
{

PerIoData->Buffer= Pdata + send_times;
PerIoData->DataBuf.len = filelen -send_times;//
BufferµÄ³¤¶È
send_times = filelen;

}

printf(" sum send_times is %d \n", send_times);

PerHandleData->pusers->sendtimes += PerIoData->DataBuf.len;

PerIoData->DataBuf.buf = PerIoData->Buffer;//
ÊÕµ½µÄBuffer;

Flags = 0;
//
//
unlock();

if (WSASend(Accept, &(PerIoData->DataBuf), 1, &PerIoData->BytesSEND,
0,
&(PerIoData->Overlapped), NULL) == SOCKET_ERROR)

{
if (WSAGetLastError() != ERROR_IO_PENDING)
{
printf("WSARecv() failed with error %d\n", WSAGetLastError());
return;
}
}

}
send_times = 0;
}
}

DWORD WINAPI ServerWorkerThread(LPVOID CompletionPortID)//
·þÎñ¹¤×÷Ïß³Ì
{
HANDLE CompletionPort = (HANDLE) CompletionPortID;//
´Ó²ÎÊý»ñÈ¡Íê³É¶Ë¿Ú¶ÔÏóhandle
DWORD BytesTransferred;
LPOVERLAPPED Overlapped;
LPPER_HANDLE_DATA PerHandleData;
LPPER_IO_OPERATION_DATA PerIoData;
DWORD SendBytes, RecvBytes;
DWORD Flags;

while(TRUE)// ËÀÑ»·
{

if (GetQueuedCompletionStatus(CompletionPort, &BytesTransferred,//
»ñÈ¡¶ÓÁÐÖÐiocpÖÐµÄ×´Ì¬
(LPDWORD)&PerHandleData, (LPOVERLAPPED *) &PerIoData, INFINITE) ==
0)
{
printf("GetQueuedCompletionStatus failed with error %d\n",
GetLastError());
return 0;
}

// First check to see if an error has occured on the socket and if so
// then close the socket and cleanup the SOCKET_INFORMATION structure
// associated with the socket.

if (BytesTransferred == 0)//
µ±·¢ËÍ½ÓÊÕÊý¾ÝÎª0
{
printf("Closing socket %d\n", PerHandleData->Socket);

if (closesocket(PerHandleData->Socket) == SOCKET_ERROR)
{
printf("closesocket() failed with error %d\n", WSAGetLastError());
return 0;
}

GlobalFree(PerHandleData);//
Çå³ýµ¥ÏòÊý¾Ý
GlobalFree(PerIoData);//
Çå³ýÊý¾Ý
continue;
}

if ( BytesTransferred == PerIoData->BytesSEND) {
lock();
PerHandleData->pusers->sendtimes -=BytesTransferred;
unlock();
printf("BytesTransferred == PerIoData->BytesSEND %d need send\n",
PerHandleData->pusers->sendtimes);
//GlobalFree(PerHandleData);//
Çå³ýµ¥ÏòÊý¾Ý
GlobalFree(PerIoData);//
Çå³ýÊý¾Ý
if (PerHandleData->pusers->sendtimes ==0)
{
closesocket(PerHandleData->pusers->Socket);
//GlobalFree(PerHandleData);

GlobalFree(PerHandleData);
}

continue;
}

if ( BytesTransferred < PerIoData->BytesSEND) { //
·¢ËÍµÄÊý¾ÝÐ¡ÓÚÒªËÍµÄÊý¾Ý
lock();
PerHandleData->pusers->sendtimes -=BytesTransferred;
unlock();
printf(" %d need send\n", PerHandleData->pusers->sendtimes);
printf("send not full buff\n");
ZeroMemory(&(PerIoData->Overlapped), sizeof(OVERLAPPED));
//PerIoData->BytesRECV = ;
PerIoData->BytesSEND = PerIoData->BytesSEND - BytesTransferred;

PerIoData->DataBuf.len= PerIoData->BytesSEND;//
BufferµÄ³¤¶È
PerIoData->Buffer= PerIoData->Buffer + BytesTransferred;
PerIoData->DataBuf.buf= PerIoData->Buffer;//
ÊÕµ½µÄBuffer;

if (WSASend(PerHandleData->Socket, &(PerIoData->DataBuf), 1,
&(PerIoData->BytesSEND), 0,
&(PerIoData->Overlapped), NULL) == SOCKET_ERROR)
{
if (WSAGetLastError() != ERROR_IO_PENDING)
{
printf("WSASend() failed with error %d\n", WSAGetLastError());
return 0;
}
}
}
}

}

sunway

unread,

Dec 17, 2006, 8:30:13 PM12/17/06

to 高性能网络编程邮件列表

大家遇到过吗，在带宽不够的情况下，完成端口没有完成投递给他的-工作就返回了完成包

这个老帖子里有分析的，以后发这类帖子建议先看看老帖子

sunway

unread,

Dec 17, 2006, 8:31:24 PM12/17/06

to 高性能网络编程邮件列表

最好不要贴大量的代码，大多数人不会乐意看大量的代码，然后帮你找BUG.

papamms

unread,

Dec 17, 2006, 8:47:32 PM12/17/06

to 高性能网络编程邮件列表

看过老帖子的。但是老帖子没有代码看不出原因。

这个代码在我比较老的机器上运行有时候正常
有时候错误。

在我的新机器上基本就时不对。

我说一下做法。

主线程拿到accept 过来的socket 以后，就循环使用WSASend
把数据都发出去，然后记数。

在主线程里面使用wsasend
是否是正确的，我不清楚。或者要在工作线程里面send
才对。

工作线程，
就是把每次投递都完成，然后把计数器减小。
当计数器为0 就断开。

我贴代码，只是想让大家跑跑看
看代码在不同机器上是否一样。

按照我的想法。服务器端会显示出
有多少字节需要发送，最后数值变为0, 断开连接。

但实际的情况不是这样的。
有时候能出现我预想的这个情况，大多数情况不行，显示还有内容没有发送
成功，但我的客户端显示已经拿到了足够的内容。（
我没有检测拿到的数据是否乱序，但字符总量是正确的）

papamms

unread,

Dec 17, 2006, 8:52:51 PM12/17/06

to 高性能网络编程邮件列表

typedef struct
{
OVERLAPPED Overlapped;
WSABUF DataBuf;
CHAR Buffer[DATA_BUFSIZE];
DWORD BytesSEND;
DWORD BytesRECV;
} PER_IO_OPERATION_DATA, * LPPER_IO_OPERATION_DATA;

typedef struct
{
SOCKET Socket;
} PER_HANDLE_DATA, * LPPER_HANDLE_DATA;

这2个结构是 m$ 例子里面的。
当一个连接建立起来的时候需要建立一个
PerHandleData
和
PER_IO_OPERATION_DATA 结构

还需要 CreateIoCompletionPort((HANDLE) Accept, CompletionPort,
(DWORD) PerHandleData, 0)
建立一个完成相关完成端口。但是好像不能多次建立这个过程，也就是一个socket
只能建立一次。
是否是这样？

papamms

unread,

Dec 17, 2006, 10:16:30 PM12/17/06

to 高性能网络编程邮件列表

解决了逻辑问题了。

} PER_IO_OPERATION_DATA, * LPPER_IO_OPERATION_DATA;

typedef struct
{
SOCKET Socket;

} PER_HANDLE_DATA, * LPPER_HANDLE_DATA;

只能建立一次。
是否是这样？

这个还要确认一下。

christanxw

unread,

Dec 18, 2006, 8:13:24 PM12/18/06

to 高性能网络编程邮件列表

IOCP是不会保证投递序列按投递顺序返回完成包的。《windows网络编程》说的很清楚。如果你先投递1个10K的操作，在投递一个20K的操作，你GetIoCompletion的时候可能先获取20K那个操作的完成包。

Reply all

Reply to author

Forward

iocp 多次投递 保证序列吗？

papamms

sunway

sunway

papamms

papamms

papamms

christanxw

iocp 多次投递保证序列吗？