This commit is contained in:
Saturneic 2019-01-16 18:08:59 +08:00
parent 10546558c2
commit be9940f570
5 changed files with 390 additions and 18 deletions

28
client.cpp Normal file
View File

@ -0,0 +1,28 @@
//
// main.cpp
// Netc
//
// Created by 胡一兵 on 2019/1/13.
// Copyright © 2019年 Bakantu. All rights reserved.
//
#include "net.h"
#include "server.h"
int main(int argc, char *argv[])
{
try {
Socket client("127.0.0.1",9048,false,false);
while (1) {
client.PacketSend("Hello");
usleep(50000);
}
} catch (char const *str) {
printf("%s\n",str);
return 0;
}
}

44
net.cpp
View File

@ -12,7 +12,7 @@
#include "cthread.h" #include "cthread.h"
extern list<CThread *> daemon_list; extern list<CThread *> daemon_list;
extern list<Server *> server_list; extern list<server_clock> server_list;
static struct itimerval oitrl, itrl; static struct itimerval oitrl, itrl;
void init(void){ void init(void){
@ -30,14 +30,52 @@ void setThreadsClock(void){
} }
//时钟滴答调用函数 //时钟滴答调用函数
void threadsClock(int n){ void threadsClock(int n){
//printf("Clock click.\n");
for(auto i = daemon_list.begin(); i != daemon_list.end(); i++){ for(auto i = daemon_list.begin(); i != daemon_list.end(); i++){
(*i)->Daemon(); (*i)->Daemon();
} }
// 服务器守护程序
for(auto i = server_list.begin(); i != server_list.end(); i++){
if(--(*i).click == 0){
(*i).psvr->Deamon();
}
}
for(auto i = server_list.begin(); i != server_list.end();){
if((*i).click == 0){
i = server_list.erase(i);
}
else i++;
}
daemon_list.clear(); daemon_list.clear();
} }
int main(void){ int main(void){
init(); init();
Server srvr("127.0.0.1");
//srvr.Deamon();
vector<int> fargs = {1,0,0,1};
vector<void *>args;
CPart::addArg<double>(&args, 12.63);
CPart::addArg<int>(&args, 10);
CPart::addArg<int>(&args, 6);
CPart::addArg<double>(&args, 8.2);
// 输入过程
struct compute_result cpur = {"Test",&args,&args,&fargs,&fargs};
packet pkt = srvr.CPURS2Packet(cpur);
raw_data rwd = srvr.Packet2Rawdata(pkt);
// 输出过程
srvr.Rawdata2Packet(rwd);
while(1){
sleep(100);
}
return 0;
}
void CPMT(void){
CMap map("./PCS"); CMap map("./PCS");
CThread thread(&map); CThread thread(&map);
thread.AddArgs<int>("B", 4); thread.AddArgs<int>("B", 4);
@ -48,10 +86,6 @@ int main(void){
thread.DoLine(); thread.DoLine();
thread.SetDaemon(); thread.SetDaemon();
thread.CancelChildPCS(0); thread.CancelChildPCS(0);
while(1){
sleep(100);
}
return 0;
} }

108
net.h
View File

@ -28,6 +28,8 @@
#include <netdb.h> #include <netdb.h>
#include <sys/time.h> #include <sys/time.h>
#include <signal.h> #include <signal.h>
#include <fcntl.h>
#include <errno.h>
#include "cpart.h" #include "cpart.h"
@ -40,6 +42,7 @@ using std::ifstream;
using std::cout; using std::cout;
using std::endl; using std::endl;
class Addr{ class Addr{
public: public:
struct sockaddr_in address; struct sockaddr_in address;
@ -84,7 +87,7 @@ public:
struct sockaddr_in c_addr; struct sockaddr_in c_addr;
Addr addr; Addr addr;
int nsfd,sfd,port; int nsfd,sfd,port;
bool server,tcp,ipv4; bool server,tcp,ipv4,set_fcntl = false;
void (*func)(class Socket &,int ,Addr); void (*func)(class Socket &,int ,Addr);
Socket(string ip_addr, int port, bool server = false, bool tcp = true, bool ipv4 = true){ Socket(string ip_addr, int port, bool server = false, bool tcp = true, bool ipv4 = true){
if(ipv4) if(ipv4)
@ -151,20 +154,109 @@ public:
sendto(sfd, buff.data(), buff.size(), 0, addr.obj(), addr.size()); sendto(sfd, buff.data(), buff.size(), 0, addr.obj(), addr.size());
} }
string PacketRecv(Addr t_addr){ // 发送一段二进制信息
void PacketSendRAW(char *buff, unsigned long size){
if(!tcp)
sendto(sfd, buff, size, 0, addr.obj(), addr.size());
}
// 接受储存字符串信息的UDP包
int PacketRecv(Addr &t_addr, string &str){
if(!tcp){ if(!tcp){
char buff[BUFSIZ]; char buff[BUFSIZ];
ssize_t tlen = recvfrom(sfd, (void *)buff, BUFSIZ, 0, t_addr.obj(), t_addr.sizep()); ssize_t tlen;
if(tlen > 0){ // 非阻塞输入
buff[tlen] = '\0'; if(set_fcntl){
string str = buff; tlen = recvfrom(sfd, (void *)buff, BUFSIZ, 0, t_addr.obj(), t_addr.sizep());
return str; // 读取错误
if(tlen == -1 && errno != EAGAIN){
str = "";
return -1;
}
// 缓冲区没有信息
else if(tlen == 0 || (tlen == -1 && errno == EAGAIN)){
str = "";
return 0;
}
// 成功读取信息
else{
str = buff;
buff[tlen] = '\0';
return 1;
}
}
else{
tlen = recvfrom(sfd, (void *)buff, BUFSIZ, 0, t_addr.obj(), t_addr.sizep());
if(~tlen){
str = buff;
buff[tlen] = '\0';
return 1;
}
else{
str = "";
return -1;
}
} }
else throw "packet receive fail";
} }
throw "connection is tcp"; throw "connection is tcp";
} }
// 接受储存二进制信息的UDP包
ssize_t PacketRecvRAW(Addr &t_addr, char *p_rdt){
if(!tcp){
char buff[BUFSIZ];
ssize_t tlen;
// 非阻塞输入
if(set_fcntl){
tlen = recvfrom(sfd, (void *)buff, BUFSIZ, 0, t_addr.obj(), t_addr.sizep());
// 读取错误
if(tlen == -1 && errno != EAGAIN){
p_rdt = nullptr;
return -1;
}
// 缓冲区没有信息
else if(tlen == 0 || (tlen == -1 && errno == EAGAIN)){
p_rdt = nullptr;
return 0;
}
// 成功读取信息
else{
p_rdt = (char *)malloc(tlen);
memcpy(p_rdt, buff, tlen);
return tlen;
}
}
else{
tlen = recvfrom(sfd, (void *)buff, BUFSIZ, 0, t_addr.obj(), t_addr.sizep());
if(~tlen){
p_rdt = (char *)malloc(tlen);
memcpy(p_rdt, buff, tlen);
return tlen;
}
else{
p_rdt = nullptr;
return -1;
}
}
}
throw "connection is tcp";
}
unsigned long IfHasPacket(void){
return 0;
}
void UDPSetFCNTL(void){
if(!tcp){
int flags = fcntl(sfd, F_GETFL, 0);
fcntl(sfd, F_SETFL, flags | O_NONBLOCK);
set_fcntl = true;
}
}
string Recv(int t_nsfd){ string Recv(int t_nsfd){
if(tcp){ if(tcp){
char buff[BUFSIZ]; char buff[BUFSIZ];

View File

@ -7,3 +7,9 @@
// //
#include "server.h" #include "server.h"
list<server_clock> server_list;
void setServerClock(Server *psvr, int clicks){
server_list.push_back({psvr,clicks});
}

222
server.h
View File

@ -13,13 +13,18 @@
class Server; class Server;
static list<Server *> server_list;
//外来数据包解析结构 //外来数据包解析结构
struct compute_result{ struct compute_result{
string name; string name;
vector<void *> *args_in; vector<void *> *args_in;
vector<void *> *args_out; vector<void *> *args_out;
vector<int> *fargs_in;
vector<int> *fargs_out;
};
struct server_clock{
Server *psvr;
int click;
}; };
//原始数据包 //原始数据包
@ -28,15 +33,222 @@ struct packet{
vector<pair<unsigned int, void *>> buffs; vector<pair<unsigned int, void *>> buffs;
}; };
//二进制原始串及其信息标签
struct raw_data{
char *data = NULL;
unsigned long size = 0;
uint32_t head, tail;
uint32_t info;
char *msg = NULL;
unsigned long msg_size = 0;
};
//设置服务器守护程序的时钟
void setServerClock(Server *psvr, int clicks);
class Server{ class Server{
vector<compute_result> cpurs; public:
vector<packet> packets; vector<compute_result> cpurs_in;
vector<packet> packets_in;
vector<string> rawstr_in;
Socket socket; Socket socket;
int packet_max = 30;
Server(string ip_addr):socket(ip_addr,9048,true,false){ Server(string ip_addr):socket(ip_addr,9048,true,false){
socket.UDPSetFCNTL();
} }
void Deamon(void){ void Deamon(void){
socket.PacketRecv(<#Addr t_addr#>) //cout<<"Server Deamon Checked."<<endl;
Addr f_addr;
int rtn, prm = packet_max;
string str;
printf("Checking Packet.\n");
do{
rtn = socket.PacketRecv(f_addr,str);
if(rtn == 1){
cout<<str<<endl;
rawstr_in.push_back(str);
}
else{
}
}while (rtn && prm-- > 0);
setServerClock(this, 2);
} }
// 将计算结果包转化为结构数据包
packet CPURS2Packet(compute_result tcpur){
packet rawpkt;
rawpkt.type = 0;
int count = 0;
// 写入计算模块名字
rawpkt.buffs.push_back({tcpur.name.size(),(void *)tcpur.name.data()});
// 写入输入参数个数
int *p_value = (int *)malloc(sizeof(uint32_t));
*p_value = (int)tcpur.args_in->size();
rawpkt.buffs.push_back({sizeof(uint32_t),p_value});
// 写入输入参数
vector<int> &fargs_in = *(tcpur.fargs_in);
for(auto i = tcpur.args_in->begin(); i != tcpur.args_in->end(); i++,count++){
if(fargs_in[count] == INT){
int *p_value = (int *)malloc(sizeof(int));
*p_value = *((int *)(*i));
rawpkt.buffs.push_back({sizeof(int),p_value});
}
else if(fargs_in[count] == DOUBLE){
double *p_value = (double *)malloc(sizeof(double));
*p_value = *((double *)(*i));
rawpkt.buffs.push_back({sizeof(double),p_value});
}
}
// 写入输入参数个数
p_value = (int *)malloc(sizeof(uint32_t));
*p_value = (int)tcpur.args_out->size();
rawpkt.buffs.push_back({sizeof(uint32_t),p_value});
// 写入输出参数
count = 0;
vector<int> &fargs_out = *(tcpur.fargs_out);
for(auto i = tcpur.args_out->begin(); i != tcpur.args_out->end(); i++,count++){
if(fargs_out[count] == INT){
int *p_value = (int *)malloc(sizeof(int));
*p_value = *((int *)(*i));
rawpkt.buffs.push_back({sizeof(int),p_value});
}
else if(fargs_out[count] == DOUBLE){
double *p_value = (double *)malloc(sizeof(double));
*p_value = *((double *)(*i));
rawpkt.buffs.push_back({sizeof(double),p_value});
}
}
return rawpkt;
}
// 将结构数据包转换成原始二进制串
raw_data Packet2Rawdata(packet tpkt){
raw_data rdta;
char *data = (char *)malloc(BUFSIZ);
memset(data, 0, BUFSIZ);
rdta.data = data;
char *idx = data;
string fdata;
// 写入包ID信息
memcpy(idx, &tpkt.type, sizeof(uint32_t));
idx += sizeof(uint32_t);
for(auto i = tpkt.buffs.begin(); i != tpkt.buffs.end(); i++){
// 写入数据块大小信息
memcpy(idx, &(*i).first, sizeof(uint32_t));
idx += sizeof(uint32_t);
// 写入数据块信息
memcpy(idx, &(*i).second, (*i).first);
idx += (*i).first;
}
rdta.size = idx - data;
return rdta;
}
// 将二进制串信息转换为结构数据包
packet Rawdata2Packet(raw_data trdta){
packet pkt;
char *idx = trdta.data;
// 数据包ID
uint32_t uint;
memcpy(&pkt.type, idx, sizeof(uint32_t));
idx += sizeof(uint32_t);
// 数据包主体
while(idx - trdta.data < trdta.size){
memcpy(&uint, idx, sizeof(uint32_t));
idx += sizeof(uint32_t);
void *data = malloc(uint);
memcpy(data, idx, uint);
idx += uint;
pkt.buffs.push_back({uint,data});
}
return pkt;
}
// 将结构数据包转化为计算结果包
compute_result Packet2CPUR(packet *tpkt){
compute_result tcpur;
tcpur.args_in = new vector<void *>();
tcpur.args_out = new vector<void *>();
if(tpkt->type == 0){
int nargs_in = *(int *)(tpkt->buffs[0].second);
int nargs_out = *(int *)(tpkt->buffs[nargs_in+1].second);
// 转化输入参数
for(int i = 0; i < nargs_in; i++){
(*tcpur.args_in)[i] = malloc(tpkt->buffs[i+1].first);
memcpy((*tcpur.args_in)[i], tpkt->buffs[i+1].second, tpkt->buffs[i+1].first);
}
for(int i = nargs_in+1; i < nargs_in+nargs_out+2; i++){
(*tcpur.args_out)[i] = malloc(tpkt->buffs[i+1].first);
memcpy((*tcpur.args_out)[i], tpkt->buffs[i+1].second, tpkt->buffs[i+1].first);
}
}
return tcpur;
}
void freeRawdataServer(struct raw_data trdt){
free(trdt.data);
if(trdt.msg != NULL) free(trdt.msg);
}
void freePcaketServer(struct packet tpkt){
for(auto i = tpkt.buffs.begin(); i != tpkt.buffs.end(); i++)
free(i->second);
delete &tpkt.buffs;
}
void freeCPURServer(struct compute_result tcpur){
// 释放输入参数容器所占用的所有内存
for(auto i = tcpur.args_in->begin(); i != tcpur.args_in->end(); i++)
free(*i);
delete tcpur.args_in;
// 释放输出参数容器所占用的所有内存
for(auto i = tcpur.args_out->begin(); i != tcpur.args_out->end(); i++)
free(*i);
delete tcpur.args_out;
}
// 为原始二进制串打上信息标签
void RawdataAddInfo(struct raw_data *trdt,char info[]){
// 填充标签信息
memcpy(&trdt->head, "NETC", sizeof(uint32_t));
memcpy(&trdt->tail, "CTEN", sizeof(uint32_t));
memcpy(&trdt->head, info, sizeof(uint32_t));
// 整合信息
char *msg = (char *)malloc(sizeof(uint32_t) * 3 + trdt->size);
trdt->msg_size = sizeof(uint32_t) * 3 + trdt->size;
char *idx = msg;
memcpy(idx, &trdt->head, sizeof(uint32_t));
idx += sizeof(uint32_t);
memcpy(idx, &trdt->info, sizeof(uint32_t));
idx += sizeof(uint32_t);
memcpy(idx, &trdt->data, trdt->size);
idx += trdt->size;
memcpy(idx, &trdt->tail, sizeof(uint32_t));
}
// 发送已经打上标签的原始二进制串
int SentRawdata(struct raw_data *trdt){
socket.PacketSendRAW(trdt->msg, trdt->msg_size);
return 0;
}
// 检查二进制信息是否为一个打上标签的原始二进制串
bool CheckRawMsg(char *p_rdt, ssize_t size){
uint32_t head, tail;
char *idx = p_rdt;
memcpy(&head, "NETC", sizeof(uint32_t));
memcpy(&tail, "CTEN", sizeof(uint32_t));
if(memcmp(idx, &head, sizeof(uint32_t))){
idx += size-sizeof(uint32_t);
if(memcmp(idx, &tail, sizeof(uint32_t))) return true;
else return false;
}
else return false;
}
}; };
#endif /* server_h */ #endif /* server_h */