testthread.cpp

模仿wiondws写的linux/freeBSD系统的IOCP

#include "utility/lockable_queue.h"
#include "iocp/iohandler.h"
#include "iocp/eventtrigger.h"
#include "utility/socketwrapper.h"
#include "utility/sysinfo.h"
#include "testthread.h"
#include "utility/fixobjpool.h"
#include "socket.h"
//extern fixObjPool_lockbase<userContext> context_pool;

bool cpuWorker::run()
{
	for(;;)
	{
		cpuUseage();
		Thread::sleep(1);
	}
}


bool listener::run()
{
		struct sockaddr_in servaddr;
		if(0 > (listenfd = Tcp_Listen(ip.c_str(),port,servaddr,backlog)))
			return false;
		int i = 1;
		for( ; ;)
		{
			if(terminated)
				break;

			struct sockaddr_in cliaddr;
			socklen_t len;
			int connfd = Accept(listenfd,(struct sockaddr*)NULL,NULL);
			if(connfd > 0)
			{

				char address[128];
				unsigned short port;
				getRemoteAddrPort(connfd,address,port);
				SockHandle tmph =AttachIocp(iocp,connfd); 

				ksocket *ks = new ksocket(tmph);
				printf("a user comming,%d\n",i++);

				ks->recv(1024);

				/*userContext *context = context_pool.Alloc();
				context->ioop.buf = context->buf;
				context->ioop.handler = tmph;
				context->ioop.rsize = 1024;
				context->ioop.type = IOREAD;
				context->ioop.userdata = context;
				Recv(tmph,&context->ioop);
				*/

			}
	
		}

		close(listenfd);

		return true;
}

bool completeWorker::run()
{
	for( ; ;)
	{
		IO_CONTEXT *op; 
		int ret = WaitForComplete(iocp,op);
		//printf("waitforcomplete threadid %u\n",pthread_self());
		if(0 == ret)
		{
			//userContext *context = (userContext*)op->userdata;
			ksocket *ks = (ksocket*)op->userdata;
			if(op->type == READCLOSE || op->type == WRITECLOSE)
			{
				hellocmd *cmd; 
				while(cmd = (hellocmd*)ks->unpack())
				{
					//printf("%s\n",cmd->buf);
					delete cmd;
					cmd = NULL;
				}
				DetachIocp(iocp,op->handler);
				delete ks;
				//context_pool.Release(context);
			}
			/*else if(op->type == IOREAD)
			{
				printf("%s\n",op->buf);
				op->type = IOWRITE;
				Send(op->handler,op);
			}*/
			else if(op->type == IOWRITE)
			{
				//memset(context->buf,0,sizeof(context->buf));
				//op->type = IOREAD;
				//op->rsize = 1024;
				//Recv(op->handler,op);
				ks->setCanSend();
			}
			else if(op->type == IOREAD)
			{
				//printf("size %d\n",op->rsize);
				//memset(context->buf,0,sizeof(context->buf));
				//op->type = IOREAD;
				//op->rsize = 1024;
				//Recv(op->handler,op);
				ks->setCanRecv();
				
				ks->put(op->buf,op->rsize);
				ks->recv(1024);
				hellocmd *cmd; 
				while(cmd = (hellocmd*)ks->unpack())
				{

					//printf("%s\n",cmd->buf);
					delete cmd;
					cmd = NULL;
				}

				
			}
		}

	}
}