自学教程：C++ CreateIoCompletionPort函数代码示例

int IOCP::Create(void){  int iHandle;  HANDLE hnd;  hnd = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);  if ( hnd == NULL )  {    return -1;  }  iHandle = UniqueCount :: GetUniqueCount();  iocp.insert( MapWordToPtr::value_type( iHandle, hnd ) );  return iHandle;}

static int uv_set_pipe_handle(uv_pipe_t* handle, HANDLE pipeHandle) {  DWORD mode = PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT;  if (!SetNamedPipeHandleState(pipeHandle, &mode, NULL, NULL)) {    return -1;  }  if (CreateIoCompletionPort(pipeHandle,                             LOOP->iocp,                             (ULONG_PTR)handle,                             0) == NULL) {    return -1;  }  return 0;}

bool IOCP_Manager::Initialize(){    SYSTEM_INFO SystemInfo;    GetSystemInfo(&SystemInfo);    mIoThreadCount = SystemInfo.dwNumberOfProcessors * 2;    WSADATA wsaData;    if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)    {        printf_s("WSAStartup Error/n");        return false;    }    mCompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, NULL, mIoThreadCount);    if (mCompletionPort == NULL)    {        printf_s("Create CompletionPort error/n");        return false;    }    mListenSocket = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, NULL, WSA_FLAG_OVERLAPPED);    if (mListenSocket == INVALID_SOCKET)    {        printf_s("Create ListenSocket Error/n");        return false;    }    int opt = 1;    setsockopt(mListenSocket, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(int));    SOCKADDR_IN ServerAddr;    int SizeServerAddr = sizeof(SOCKADDR_IN);    memset(&ServerAddr, 0, SizeServerAddr);    if (WSAStringToAddress(L"127.0.0.1:9001", AF_INET, NULL, (SOCKADDR*)&ServerAddr, &SizeServerAddr) == SOCKET_ERROR)    {        printf_s("ServerAddress Setting Error/n");        return false;    }    if (bind(mListenSocket, (SOCKADDR*)&ServerAddr, SizeServerAddr) == SOCKET_ERROR)    {        printf_s("Bind Error/n");        return false;    }    return true;}

int SP_IocpLFServer :: run(){	int ret = 0;	int listenFD = -1;	ret = SP_IOUtils::tcpListen( mBindIP, mPort, &listenFD, 0 );	if( 0 == ret ) {		mCompletionHandler = mAcceptArg->mHandlerFactory->createCompletionHandler();		SP_IocpMsgQueue * msgQueue = new SP_IocpMsgQueue( mEventArg->getCompletionPort(),			SP_IocpEventCallback::eKeyMsgQueue, SP_IocpEventCallback::onResponse, mEventArg );		mEventArg->setResponseQueue( msgQueue );		mEventArg->loadDisconnectEx( listenFD );		mAcceptArg->mAcceptEvent = CreateEvent( NULL, TRUE, FALSE, NULL );		mAcceptArg->mListenSocket = (HANDLE)listenFD;		if( NULL == CreateIoCompletionPort( mAcceptArg->mListenSocket,				mEventArg->getCompletionPort(), SP_IocpEventCallback::eKeyAccept, 0 ) ) {			sp_syslog( LOG_ERR, "CreateIoCompletionPort fail, errno %d", WSAGetLastError() );			return -1;				}		if( NULL == mAcceptArg->mIOChannelFactory ) {			mAcceptArg->mIOChannelFactory = new SP_DefaultIOChannelFactory();		}		sp_thread_t thread;		ret = sp_thread_create( &thread, NULL, acceptThread, mAcceptArg );		if( 0 == ret ) {			sp_syslog( LOG_NOTICE, "Thread #%ld has been created to accept socket", thread );		} else {			sp_syslog( LOG_WARNING, "Unable to create a thread to accept socket, %s", strerror( errno ) );			return -1;		}		mIsRunning = 1;		mThreadPool = new SP_ThreadPool( mMaxThreads );		for( int i = 0; i < mMaxThreads; i++ ) {			mThreadPool->dispatch( lfHandler, this );		}	}	return ret;}

static int uv_tcp_set_socket(uv_loop_t* loop, uv_tcp_t* handle,    SOCKET socket, int imported) {  DWORD yes = 1;  assert(handle->socket == INVALID_SOCKET);  /* Set the socket to nonblocking mode */  if (ioctlsocket(socket, FIONBIO, &yes) == SOCKET_ERROR) {    uv__set_sys_error(loop, WSAGetLastError());    return -1;  }  /* Make the socket non-inheritable */  if (!SetHandleInformation((HANDLE)socket, HANDLE_FLAG_INHERIT, 0)) {    uv__set_sys_error(loop, GetLastError());    return -1;  }  /* Associate it with the I/O completion port. */  /* Use uv_handle_t pointer as completion key. */  if (CreateIoCompletionPort((HANDLE)socket,                             loop->iocp,                             (ULONG_PTR)socket,                             0) == NULL) {    if (imported) {      handle->flags |= UV_HANDLE_EMULATE_IOCP;    } else {      uv__set_sys_error(loop, GetLastError());      return -1;    }  }  if (pSetFileCompletionNotificationModes) {    if (pSetFileCompletionNotificationModes((HANDLE) socket,        FILE_SKIP_SET_EVENT_ON_HANDLE |        FILE_SKIP_COMPLETION_PORT_ON_SUCCESS)) {      handle->flags |= UV_HANDLE_SYNC_BYPASS_IOCP;    } else if (GetLastError() != ERROR_INVALID_FUNCTION) {      uv__set_sys_error(loop, GetLastError());      return -1;    }  }  handle->socket = socket;  return 0;}

EVQ_API intevq_add (struct event_queue *evq, struct event *ev){    const unsigned int ev_flags = ev->flags;    struct win32thr *wth = &evq->head;    ev->wth = wth;    if (ev_flags & EVENT_SIGNAL)	return signal_add(evq, ev);    if (ev_flags & EVENT_WINMSG) {	evq->win_msg = ev;	evq->nevents++;	return 0;    }    if ((ev_flags & (EVENT_SOCKET | EVENT_SOCKET_ACC_CONN)) == EVENT_SOCKET     && evq->iocp.h) {	if (!CreateIoCompletionPort((HANDLE) ev->fd, evq->iocp.h, 0, 0)	 && GetLastError() != ERROR_INVALID_PARAMETER)  /* already assosiated */	    return -1;	ev->flags |= EVENT_AIO;	evq->iocp.n++;	evq->nevents++;	if (pSetFileCompletionNotificationModes	 && pSetFileCompletionNotificationModes((HANDLE) ev->fd, 3))	    ev->flags |= EVENT_AIO_SKIP;	win32iocp_set(ev, ev_flags);	return 0;    }    while (wth->n >= NEVENT - 1)	if (!(wth = wth->next)) break;    if (wth)	win32thr_sleep(wth);    else {	wth = win32thr_init(evq);	if (!wth) return -1;    }    return win32thr_add(wth, ev);}

CIOCPManager::CIOCPManager(int IOCPThreadCount /*= 0*/){	SOCKET TmpSock;	GUID guidAcceptEx = WSAID_ACCEPTEX;	GUID guidGetAcceptExSockaddrs = WSAID_GETACCEPTEXSOCKADDRS;	DWORD dwBytes;	INT i;	OutputDebugStr(_T("IOCPManager::IOCPManager/n"));	// 使用 2.2版的WS2_32.DLL	if (WSAStartup(0x0202, &mwsaData) != 0) {		throw exception("WSAStartup Fails");	}	// 获取AcceptEx和GetAcceptExSockaddrs的函数指针	TmpSock = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);	if (TmpSock == INVALID_SOCKET) {		throw exception("WSASocket Fails");	}	if (SOCKET_ERROR == WSAIoctl(TmpSock, SIO_GET_EXTENSION_FUNCTION_POINTER,		&guidAcceptEx, sizeof(guidAcceptEx), &g_AcceptEx, sizeof(g_AcceptEx), &dwBytes, NULL, NULL)) {		throw exception("WSAIoctl WSAID_ACCEPTEX Fails");	}	if (SOCKET_ERROR == WSAIoctl(TmpSock, SIO_GET_EXTENSION_FUNCTION_POINTER,		&guidGetAcceptExSockaddrs, sizeof(guidGetAcceptExSockaddrs), &g_GetAcceptExSockaddrs, sizeof(g_GetAcceptExSockaddrs), &dwBytes, NULL, NULL)) {		throw exception("WSAIoctl WSAID_GETACCEPTEXSOCKADDRS Fails");	}	closesocket(TmpSock);	// 初始化临界区	InitializeCriticalSection(&mSockListCS);	InitializeCriticalSection(&mOverLappedListCS);	// 初始化IOCP完成端口	mCompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0, 0);	if (IOCPThreadCount <= 0) {		SYSTEM_INFO SysInfo;		GetSystemInfo(&SysInfo);		IOCPThreadCount = SysInfo.dwNumberOfProcessors + 2;	}	mIocpWorkThreadCount = IOCPThreadCount;	mIocpWorkThreads = new HANDLE[mIocpWorkThreadCount];	// 创建IOCP工作线程	for (i = 0; i < mIocpWorkThreadCount; i++) {		mIocpWorkThreads[i] = (HANDLE)_beginthreadex(NULL, 0, IocpWorkThread, (PVOID)mCompletionPort, 0, NULL);		if (mIocpWorkThreads[i] == NULL) {			throw exception("CreateThread FIocpWorkThreads Fails");		}	}}

int iocp::iocp_open(int thread_count){	m_iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, NULL, 0);	if(!m_iocp)	{		m_error = GetLastError();		printf("iocp initialize failed!/n");		return 0;	}	m_thread_count = thread_count;	for(int i = 0; i < m_thread_count; ++i)	{		QueueUserWorkItem(iocp_threaad_pro, this, WT_EXECUTELONGFUNCTION);	}	CreateSemaphore(NULL, 0, m_thread_count, NULL);	return 1;}

void mpk_import_add_named_pipe_instance(common_vars *common_vars) {  HANDLE named_pipe_handle = CreateNamedPipeA("////.//pipe//" m_mpk_import_named_pipe_name, PIPE_ACCESS_INBOUND | FILE_FLAG_OVERLAPPED, PIPE_TYPE_BYTE, PIPE_UNLIMITED_INSTANCES, 4096, 4096, 0, nullptr);  if (!named_pipe_handle) {    m_last_error_str(err);    m_die("cannot create mpk import named pipe instance, call to /"CreateNamedPipeA/" failed/nerr: %s", err);  }  HANDLE io_completion_port = CreateIoCompletionPort(named_pipe_handle, common_vars->io_completion_port, mpk_import_named_pipe_event, 0);  if (!io_completion_port) {    m_last_error_str(err_str);    m_die("cannot connect mpk import named pipe to io completion port, call to /"CreateIoCompletionPort/" failed/nerr: %s", err_str);  }  mpk_import_named_pipe_instance *new_named_pipe_instance = (mpk_import_named_pipe_instance *)memory_pool_allocate(&common_vars->mpk_import_named_pipe_instance_memory_pool);  *new_named_pipe_instance = {};  new_named_pipe_instance->handle = named_pipe_handle;  sllist_push(&common_vars->mpk_import_named_pipe_instance_list, new_named_pipe_instance);  ConnectNamedPipe(named_pipe_handle, &new_named_pipe_instance->overlapped);}

bool ClientSession::PostConnect(){	CRASH_ASSERT( LThreadType == THREAD_MAIN );	HANDLE handle = CreateIoCompletionPort( (HANDLE)mSocket, GIocpManager->GetComletionPort(), ( ULONG_PTR )this, 0 );	if ( handle != GIocpManager->GetComletionPort() )	{		printf_s( "[DEBUG] CreateIoCompletionPort error: %d/n", GetLastError() );		return false;	}	OverlappedConnectContext* connectContext = new OverlappedConnectContext( this );	DWORD bytes = 0;	connectContext->mWsaBuf.len = 0;	connectContext->mWsaBuf.buf = nullptr;		if ( SERVER_PORT <= 4000 || SERVER_PORT > 10000 )	{		SERVER_PORT = 9001;	}	if ( HOST_NAME.length() == 0 )	{		HOST_NAME = "127.0.0.1";	}	sockaddr_in addr;	ZeroMemory( &addr, sizeof( addr ) );	addr.sin_family = AF_INET;	addr.sin_addr.s_addr = inet_addr( HOST_NAME.c_str() );	addr.sin_port = htons( SERVER_PORT );	if ( FALSE == IocpManager::ConnectEx( mSocket, (SOCKADDR*)&addr, sizeof( addr ), 		GIocpManager->mConnectBuf, 0, &bytes, (LPOVERLAPPED)connectContext ) )	{		if ( WSAGetLastError() != WSA_IO_PENDING )		{			DeleteIoContext( connectContext );			printf_s( "ConnectEx Error : %d/n", GetLastError() );			return false;		}	}		return true;}

VOID Acceptor::Init( IoHandler *pIoHandler, DWORD dwNumberOfThreads ){	assert( dwNumberOfThreads <= MAX_ACCEPT_THREAD );	m_pIoHandler	= pIoHandler;	m_numThreads	= dwNumberOfThreads;	// Accept 创建IOCP 	m_hIOCP		= CreateIoCompletionPort( INVALID_HANDLE_VALUE, NULL, 0, 0 );	// 遍历, 创建线程	unsigned threadID;	for( DWORD i = 0; i < m_numThreads; ++i )	{		m_hThread[i] = (HANDLE)_beginthreadex( NULL, 0, accept_thread, (LPVOID)this, 0/*CREATE_SUSPENDED*/, &threadID );	}}

bool ClientSession::OnConnect(SOCKADDR_IN* addr){	//TODO: 
C++ CreateList函数代码示例
C++ CreateImage函数代码示例