2016-08-3116:44:09
server 端
/*******************************************************************
author:xuqing
date:2016-5-20
about the io thread
the init part must be less than 1000,just the real_part_count
THREAD_DEAL_COUNT is the count one thread can deal with
if the real_part_count is 10;
the THREAD_DEAL_COUNT is 5
we just need 2 thread to deal the all io write
success return 0
error return 负数
sucess 有意义 正数
***********************************************************************/
#include "stdafx.h"
#include "rts_server.h"
#include "memory_part.h"
#pragma comment(lib,"ws2_32.lib") #define THREAD_DEAL_COUNT 5
HANDLE h_complete_point;
extern memory_part *temp_memory[]; int socket_init()
{
RTS_DIARY_RET_ERROR_VAR;
WSAData wsaData;
if (WSAStartup(MAKEWORD(, ), &wsaData) != )
{
printf("WSAStartup失败!\n");
RTS_DIARY_ERROR_GOTO;
}
if (LOBYTE(wsaData.wVersion) != || HIBYTE(wsaData.wVersion) != )
{
printf("版本不对！\n");
RTS_DIARY_ERROR_GOTO;
}
return ;
ret_error:
if (i_ret) return i_ret;
return ; } int write_memory(char *record)
{
RTS_DIARY_RET_ERROR_VAR;
char str_cust_id[];
get_custid2data(record, str_cust_id);
int int_cust_id = atoi(str_cust_id);
int part_index = int_cust_id % MAX_PART;
memory_part *save_part = temp_memory[part_index];
if (save_part == NULL) RTS_DIARY_ERROR_GOTO;
SqQueue *save_queue = save_part->queue_offset;
if (save_queue == NULL) RTS_DIARY_ERROR_GOTO; if ((save_queue->rear + ) % MAX_RECORD_OF_PART == save_queue->front)
{
//write memory wait!!!
Sleep();
}
i_ret=save_data2memory(part_index, record);
if (i_ret != ) RTS_DIARY_ERROR_GOTO; return ; ret_error:
if (i_ret) return i_ret;
return ; } DWORD WINAPI ServerWorkerThread(LPVOID CompletionPort)
{
RTS_DIARY_RET_ERROR_VAR;
HANDLE ComPort = (HANDLE)CompletionPort;
DWORD BytesTransferred;
//LPOVERLAPPED Overlapped;
LPPER_HANDLE_DATA PerHandleData;
LPPER_IO_OPERATION_DATA PerIoData;
DWORD SendBytes, RecvBytes;
DWORD Flags;
int recv_int;
char str[] = "";
while (TRUE)
{
//等待完成端口上SOCKET的完成 阻塞
printf( "等待完成端口上SOCKET的完成!\n");
if ((GetQueuedCompletionStatus(ComPort,&BytesTransferred, (PULONG_PTR)&PerHandleData,(LPOVERLAPPED *)&PerIoData, INFINITE)) == NULL)
{
continue;
}
recv_int = BytesTransferred;
printf("BytesTransferred work:%d\n", BytesTransferred);
//检查是否有错误产生
if (BytesTransferred == &&(PerIoData->OperationType == RECV_POSTED ||PerIoData->OperationType == SEND_POSTED))
{
//关闭SOCKET
printf("%d SOCKET关闭\n", PerHandleData->sock);
closesocket(PerHandleData->sock);
free(PerHandleData);
free(PerIoData);
continue;
}
//为请求服务
if (PerIoData->OperationType == RECV_POSTED)
{
printf( "接收处理\n");
memcpy(str, PerIoData->Buff, recv_int);
str[recv_int] = '\0';
printf("%d SOCKET :%s\n", PerHandleData->sock, str);
//although it is multithread but they don't visit the same memory,we have use the queue structure to avoid it
write_memory(str);
ZeroMemory(PerIoData->Buff, );
strcpy(PerIoData->Buff, str);
Flags = ;
ZeroMemory((LPVOID)&(PerIoData->Overlapped), sizeof(OVERLAPPED));
PerIoData->DataBuff[].len = ;
PerIoData->DataBuff[].buf = PerIoData->Buff;
PerIoData->OperationType = SEND_POSTED;
WSASend(PerHandleData->sock, PerIoData->DataBuff,, &SendBytes, , &(PerIoData->Overlapped), NULL);
printf("the send byte is %d\n", SendBytes);
}
else //if(PerIoData->OperationType == SEND_POSTED)
{
printf( "发送处理!\n");
Flags = ;
ZeroMemory((LPVOID)&(PerIoData->Overlapped), sizeof(OVERLAPPED));
ZeroMemory(PerIoData->Buff, );
PerIoData->DataBuff[].len = ;
PerIoData->DataBuff[].buf = PerIoData->Buff;
PerIoData->OperationType = RECV_POSTED;
WSARecv(PerHandleData->sock, PerIoData->DataBuff,, &RecvBytes, &Flags, &(PerIoData->Overlapped), NULL);
//recv_int = RecvBytes;
printf("the recv byte is %d\n", RecvBytes);
}
}
return ;
ret_error:
if (i_ret) return i_ret;
return ; } DWORD WINAPI write_disk(LPVOID p)
{
RTS_DIARY_RET_ERROR_VAR;
Sleep();
//循环处理不同的part 一个线程可以负责处理多个不同的part
memory_part *write_part;
SqQueue *queue;
int i = (int)p;
while (true)
{
write_part = temp_memory[i];
queue = write_part->queue_offset;
if (write_part == NULL || queue == NULL) continue;
printf("rear : %d, front :%d", queue->rear, queue->front);
while (true)
{
if (queue->front != (queue->rear))
{
printf("while front %d\n", queue->front);
printf("hello\n");
save_data2disk(+i);
}
else
{
//if the read thread pointer==write thread pointer
//the read thread must sleep 500
Sleep();
break;
}
}
i = i + ;
i = i % THREAD_DEAL_COUNT + (int)p;
}
return ;
ret_error:
if (i_ret) return i_ret;
return ; } int start_rts_server()
{
RTS_DIARY_RET_ERROR_VAR;
LPPER_HANDLE_DATA perHandleData;
LPPER_IO_OPERATION_DATA ioperdata;
SYSTEM_INFO siSys;
SOCKET sockListen;
struct sockaddr_in addrLocal;
int nRet = ;
DWORD nThreadID;
SOCKET sockAccept;
DWORD dwFlags;
DWORD dwRecvBytes;
int nReuseAddr = ;
int ret_code = ;
int real_part_count = ;
printf("初始环境...!\n");
if (socket_init() != ) RTS_DIARY_ERROR_GOTO;
init_memory_part(real_part_count);
//创建一个IO完成端口
printf("创建一个IO完成端口!\n");
h_complete_point = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, , );
if (h_complete_point == INVALID_HANDLE_VALUE)
{
printf("创建IO完成端口失败!\n");
RTS_DIARY_ERROR_GOTO;
}
//获取CPU数目
GetSystemInfo(&siSys);
for (int i = ; i<(int)siSys.dwNumberOfProcessors * ; i++)//NumberOfProcessors
{
HANDLE hThread;
hThread = CreateThread(NULL, , ServerWorkerThread, (LPVOID)h_complete_point, , &nThreadID);
printf("创建工作者线程 ---->%d!\n", i);
CloseHandle(hThread);
}
//创建监听SOCKET
printf("创建监听SOCKET!\n");
sockListen = WSASocket(AF_INET, SOCK_STREAM, , NULL, , WSA_FLAG_OVERLAPPED);
if (sockListen == SOCKET_ERROR)
{
printf("WSASocket错误!\n");
RTS_DIARY_ERROR_GOTO;
} if (setsockopt(sockListen, SOL_SOCKET, SO_REUSEADDR, (const char *)&nReuseAddr, sizeof(int)) != )
{
printf("setsockopt错误!\n");
RTS_DIARY_ERROR_GOTO;
}
addrLocal.sin_family = AF_INET;
addrLocal.sin_addr.s_addr = htonl(INADDR_ANY);
addrLocal.sin_port = htons();
if (bind(sockListen, (struct sockaddr *)&addrLocal, sizeof(sockaddr_in)) != )
{
printf("bind错误\n");
i_ret = WSAGetLastError();
RTS_DIARY_ERROR_GOTO;
}
//准备监听
printf("准备监听\n");
if (listen(sockListen, ) != )
{
printf("listen错误\n");
RTS_DIARY_ERROR_GOTO;
}
int client_addr = sizeof(SOCKADDR_IN);
int index_part = ;
// the thread to save data into the disk
for (int i = ; i < real_part_count / THREAD_DEAL_COUNT;i++)
{
index_part = THREAD_DEAL_COUNT*i;
HANDLE write_handle = CreateThread(NULL, , write_disk,(LPVOID)index_part, , NULL);
} //main thread to get client accept
while (true)
{
//阻塞
sockAccept = WSAAccept(sockListen, NULL, NULL, NULL, );
perHandleData = (LPPER_HANDLE_DATA)malloc(sizeof(PER_HANDLE_DATA));
if (perHandleData == NULL) continue;
printf("socket number ---> %d 接入！\n", sockAccept);
perHandleData->sock = sockAccept;
if (CreateIoCompletionPort((HANDLE)sockAccept, h_complete_point, (ULONG_PTR)perHandleData, ) == NULL)
{
free(perHandleData);
printf("iocp client!\n");
continue;
}
ioperdata = (LPPER_IO_OPERATION_DATA)malloc(sizeof(PER_IO_OPERATION_DATA));
memset(&(ioperdata->Overlapped), , sizeof(OVERLAPPED));
(ioperdata->DataBuff[]).len = ;
(ioperdata->DataBuff[]).buf = ioperdata->Buff;
ioperdata->OperationType = RECV_POSTED;
if (ioperdata == NULL)
{
free(perHandleData);
continue;
}
printf("投递接收操作!\n");
dwFlags = ;
int b=WSARecv(perHandleData->sock, &ioperdata->DataBuff[], , &dwRecvBytes, &dwFlags, &(ioperdata->Overlapped), NULL);
printf("the recv byte main():%d \n", dwRecvBytes);
if (b == SOCKET_ERROR)
{
ret_code = WSAGetLastError();
if (ret_code= WSA_IO_PENDING)
{
printf("was recv sucess %d\n", WSAGetLastError()); }
else
{
printf("was recv error!\n");
RTS_DIARY_ERROR_GOTO;
}
}
}
ret_error:
if (i_ret) return i_ret;
closesocket(sockListen);
return ;
}
client 端
#include "stdafx.h"
#include "rts_client.h"
#pragma comment(lib,"ws2_32.lib")
SOCKET sockClient;
struct sockaddr_in addrServer;
char buf[1024];
int n = 0;
int socket_init()
{
WSAData wsaData;
if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)
{
printf("WSAStartup失败!\n");
return -1;
} if (LOBYTE(wsaData.wVersion) != 2 || HIBYTE(wsaData.wVersion) != 2)
{
printf("版本不对！\n");
WSACleanup();
return -1;
}
return 0;
} DWORD WINAPI recv_thread(LPVOID par)
{
while (true)
{
if (recv(sockClient, buf, 1024, 0) <= 0)
{
printf("recv失败,可能连接断开!\n");
//break;
goto theend;
}
printf("服务器应答：%s", buf);
memset(buf, 0, 1024); Sleep(200);
} theend:
WSACleanup();
getchar();
return 0;
} int client_start()
{
int i_ret_error = 0;
ff2028_gset_ctx_t gset_st;
ff2028_gset_ctx_t *p_gset_st=&gset_st;
unsigned char tag[4];
i_ret_error = ff2028_gset_ctx_init(p_gset_st);
if (i_ret_error < 0)
{
printf("error-> %s,%s,%d\n", __FILE__, __FUNCTION__, __LINE__);
return i_ret_error - __LINE__;
}
if (socket_init() != 0)
goto theend; sockClient = socket(AF_INET, SOCK_STREAM, 0);
if (sockClient == INVALID_SOCKET)
{
printf("socket 失败!\n");
WSACleanup();
goto theend;
}
memset(&addrServer, 0, sizeof(sockaddr_in));
addrServer.sin_family = AF_INET;
addrServer.sin_addr.s_addr = inet_addr("127.0.0.1");
addrServer.sin_port = htons(9090);
printf("连接服务器...\n");
if (connect(sockClient, (const struct sockaddr *)&addrServer, sizeof(sockaddr)) != 0)
{
printf("connect 失败\n");
WSACleanup();
goto theend;
}
memset(buf, 0, 1024);
sprintf(buf, "1005,5443543546546,,6566,4564645646546,t435435345353453534535 --->%d\r\n", n);
tag[0] = 100;
i_ret_error = ff2028_gset_add_app_data(p_gset_st, tag, 1, (unsigned char *)buf, strlen(buf));
if (i_ret_error < 0)
{
printf("error-> %s,%s,%d\n", __FILE__, __FUNCTION__, __LINE__);
return i_ret_error - __LINE__;
}
ff2028_gset_add_app_data_end(p_gset_st);
//HANDLE receive = CreateThread(NULL, 0, recv_thread, NULL, 0, NULL);
while (true)
{
sprintf(buf, "1005,5443543546546,,6566,4564645646546,t435435345353453534535 --->%d\r\n", n);
if (send(sockClient, buf, strlen(buf), 0) <= 0)
{
printf( "send失败,可能连接断开!\n");
//break;
goto theend;
}
memset(buf, 0, 1024); //接收服务端应答
if (recv(sockClient, buf, 1024, 0) <= 0)
{
printf("recv失败,可能连接断开!\n");
//break;
goto theend;
}
printf("服务器应答：%s", buf);
memset(buf, 0, 1024);
Sleep(200);
n++;
if (n == 20)
break;
}
theend:
WSACleanup();
return 0; }