由于上次说了改工期,两天算一个工作日,所以昨天就没有日报了。
统计一下这两天团队成员们都干了啥,纪录一下我都干了啥。
上一波日报
1号 | https://blog.csdn.net/qq_43762191/article/details/106870906 |
---|---|
2号 | https://blog.csdn.net/qq_45792305/article/details/106881268 |
有点尴尬啊,可能是其他人都等着开完会才写日报,或者是写完还没过审,然后今天大家都太忙。。。
那就直接纪录我自己吧。
我都进度
预期进度
测试完文件方面代码
结果:没碰。
实际进度:epoll模块、进程间通信模块、线程池模块进行中,解压包模块依旧没去测试。
为什么?这也不是理由,没做就是没做,我不喜欢找理由。
昨天下午参加了一个评审,我是评审嘉宾之一很开心。
今天突然收到噩耗明天要考毛概,期末考,所以我就抽出了今晚预留给解压包模块测试的时间去复习了。
进程间通信模块设计(“分布式服务器”设计(初体验))
不好搞,是真的不好搞,还好我学过点皮毛的设计模式,还知道要用“中介者”。
中介者服务器(中控)采用accept,边缘服务器全部采用connect,在连接成功后向中控中心汇报自己的情况(fd对应的服务器名是啥,方便通信),奈何我天资愚钝,这个图我想了一晚上。
中控服务器设计
代码写到一半就收到要考试的噩耗,所以任务队列模块还没写,就先不贴出来了。
线程池模块(代码测试完毕)
这个模块也算是重新理解了一遍,顺便整理了一篇关于线程池的博客,代码注释率达到百分八十,非常适合新手看
:勉强还能看的线程池详解。
写过博客呢,代码思路就很清晰,两个小时就连图带代码全写好了。
//Mutex.h
#pragma once
#include <stdio.h>
#include <pthread.h>
#include <errno.h>
class Mutex
{
public:
Mutex();
virtual ~Mutex();
bool lock();
bool unlock();
bool trylock();
protected:
pthread_mutex_t mutex;
};
//Mutex.cpp
#include "Mutex.h"
Mutex::Mutex()
{
// mutex = PTHREAD_MUREX_INITALIZER; //ÕâÑù¾²Ì¬³õʼ»¯¶àºÃ
if (pthread_mutex_init(&mutex, NULL) != 0)
{
perror("pthread mutex init error");
}
}
Mutex::~Mutex()
{
if (pthread_mutex_destroy(&mutex) != 0)
{
perror("pthread mutex destroy error");
}
}
bool Mutex::lock()
{
int ret = pthread_mutex_lock(&mutex);
if (ret != 0)
{
perror("pthread mutex lock error");
}
return ret == 0 ? true : false;
}
bool Mutex::unlock()
{
int ret = pthread_mutex_unlock(&mutex);
if (ret != 0)
{
perror("pthread mutex unlock error");
}
return ret == 0 ? true : false;
}
bool Mutex::trylock()
{
int ret = pthread_mutex_trylock(&mutex);
if (ret != 0)
{
perror("pthread mutex trylock error");
}
return ret == 0 ? true : false;
}
//Cond.h
#pragma once
#include "Mutex.h"
class Cond : public Mutex
{
public:
Cond();
virtual ~Cond();
bool wait();
bool timewait(unsigned int sec);
bool signal();
bool broadcast();
bool isTimeout();
private:
pthread_cond_t cond;
bool timeout;
};
//Cond.cpp
#include "Cond.h"
#include <sys/time.h>
Cond::Cond() :Mutex()
{
timeout = false;
if (pthread_cond_init(&cond, NULL) != 0)
{
perror("pthread cond init error");
}
}
Cond::~Cond()
{
if (pthread_cond_destroy(&cond) != 0)
{
perror("pthread cond destroy error");
}
}
bool Cond::wait()
{
int ret = pthread_cond_wait(&cond, &mutex);
if (ret != 0)
{
perror("pthread cond wait error");
}
return ret == 0 ? true : false;
}
bool Cond::timewait(unsigned int sec)
{
struct timespec abstime;
clock_gettime(CLOCK_REALTIME, &abstime);
abstime.tv_sec += sec;
timeout = false;
int ret = pthread_cond_timedwait(&cond, &mutex, &abstime);
if (ret == ETIMEDOUT)
{
timeout = true;
return false;
}
else if (ret != 0)
{
perror("pthread cond timedwait error");
}
return ret == 0 ? true : false;
}
bool Cond::signal()
{
int ret = pthread_cond_signal(&cond);
if (ret != 0)
{
perror("pthread cond signal error");
}
return ret == 0 ? true : false;
}
//¹ã²¥ÉÙÓÃ
bool Cond::broadcast()
{
int ret = pthread_cond_broadcast(&cond);
if (ret != 0)
{
perror("pthread cond broadcast error");
}
return ret == 0 ? true : false;
}
bool Cond::isTimeout()
{
return timeout;
}
//Pthread_Pool.h
#pragma once
#include <pthread.h>
#include <unistd.h>
#include <list> //据说list不安全,不安全就不安全吧,更不安全的都忍了
#include "Cond.h" //封装过的条件变量类,继承自封装的mutex锁类,所以具有锁和条件变量的双重属性
using namespace std;
class Task //任务接口,每个任务必须实现的接口,以供工作线程调度任务的执行
{
public:
Task() {}
virtual ~Task() {}
virtual int run() = 0; //留给子类实现
};
typedef list<Task*> list_task; //任务队列,用于暂存等待处理的任务,等待线程唤醒时处理,提供一种缓冲机制。
class Pthread_Pool //线程池类
{
public:
Pthread_Pool(unsigned int max = 100, unsigned int min = 10, unsigned int wait = 60);
~Pthread_Pool();
void addTask(Task* task); // 往任务队列中添加新线程
private:
static void* taskThread(void* arg);// 工作线程
void createThread(); // 新建一个线程
void destroyThread(); // 销毁一个线程池
unsigned int maxcount; // 最大线程数
unsigned int mincount; // 最小线程数
unsigned int count; // 当前线程池中线程数
unsigned int waitcount; // 等待线程数
unsigned int waitsec; // 等待时间
list_task taskList; //任务队列
Cond taskCond; //任务锁,线程接任务时使用
Cond cond; //线程锁,创建线程时使用
bool Stop; //线程池是否被允许运作,初始化线程池对象时置0,线程池销毁时置为1
};
//Pthread_Pool.cpp
#include "Pthread_Pool.h"
//开放接口1
Pthread_Pool::Pthread_Pool(unsigned int max, unsigned int min, unsigned int wait)
{
//配置基本参数
count = 0; //当前线程池为空
waitcount = 0; //没有等待线程
mincount = min; //核心线程数(出厂配置)
maxcount = max; //最大线程数(能承受的最高配置)
waitsec = wait; //线程保活时长(过了时长还没接到任务,那就裁掉)
Stop = false; //允许运作
//上锁,创建一定数量的线程作为初始线程池
cond.lock();
for (unsigned i = 0; i < mincount; i++)
{
createThread(); //跳转到这个函数的实现->->->->->
}
cond.unlock();
}
Pthread_Pool::~Pthread_Pool()
{
destroyThread(); //销毁线程池
}
void Pthread_Pool::createThread()
{
pthread_t tid;
int ret = pthread_create(&tid, NULL, taskThread, (void*)this);
//以执行taskThread()为目的创建线程,跳转到taskThread()函数的实现 ->->->->->
if (ret < 0)
perror("pthread create error");
else
count++;
}
// 工作线程
void* Pthread_Pool::taskThread(void* arg)
{
pthread_detach(pthread_self()); //设置线程自分离属性
Pthread_Pool* pool = (Pthread_Pool*)arg;
while (1)
{
pool->cond.lock();
//如果没有工作线程在等待
if (pool->taskList.empty())
{
if (pool->Stop) //当收到线程池停止运行的消息时
{
pool->count--; //线程数减一
pool->cond.unlock();
pthread_exit(NULL); //本线程强制退出
}
pool->waitcount++; //等待任务的线程数加一
bool bSignal = pool->cond.timewait(pool->waitsec); //新任务等待被唤醒
pool->waitcount--; //没等到,没事干,喝西北风了
// 删除无用线程
if (!bSignal && pool->count > pool->mincount) //如果没事干 && 有多余线程
{
pool->count--; //先裁员一个,不要一次做绝了,反正是在while循环里面,没事干裁员机会多得是
pool->cond.unlock();
pthread_exit(NULL);
}
}
pool->cond.unlock(); //记得要释放锁
//如果有工作线程在等待
if (!pool->taskList.empty())
{
pool->taskCond.lock(); //上任务锁
Task* t = pool->taskList.front(); //获取任务队列中最前端的任务并执行
pool->taskList.pop_front(); //移除被领取的任务
pool->taskCond.unlock();//记得解锁
t->run(); //任务开始
delete t; //弄完就删了
}
}
pthread_exit(NULL);
}
//开放接口2,向任务队列中添加任务
void Pthread_Pool::addTask(Task* task)
{
if (Stop) //线程池是否停止工作
return;
//向任务队列中添加新任务
taskCond.lock(); //上任务锁
taskList.push_back(task); //添加任务
taskCond.unlock(); //记得解锁
cond.lock(); //上线程锁
if (waitcount) //如果有空闲线程
{
cond.signal(); //唤醒一个线程
}
else if (count < maxcount) //如果没有空闲线程,一般来说,走到这里面来,那这个线程池的设计是有点失败了
{
createThread(); //那就创建一个
cond.signal(); //然后唤醒
}
cond.unlock();
}
void Pthread_Pool::destroyThread()
{
printf("destroy?\n");
#if 0 //强行清理
list_task::iterator it = taskList.begin();
for (; it!= taskList.end(); it++)
{
Task* t = *it;
delete t;
t = NULL;
}
taskList.clear();
#endif
// 等待所有线程执行完毕
Stop = true;
while (count > 0)
{
cond.lock();
cond.broadcast(); //广播
cond.unlock();
sleep(1);
}
}
epoll连接模块
这个模块呢,因为进程间通信未搭建成功,所以还没完全竣工,epoll模块是有现成了,就是配件未到位。
要看epoll的文章,这篇不错,适合新手:epoll,求知者离我近点。
对解压包模块对修订
解压包不采用同一个包对象,这是之前没想到的,因为解包时初始化的大小不一定适用于封包时的大小哦。
此外,协议中头包的预留空间拨出一半给服务器名,用于边缘服务器与中控通信。
//对头包的修改
typedef struct packet_header_st
{
int fd;//用于前后端通信即目标客户端fd(服务器用到)
int funcId; // 功能号
//登录包0x01,注册包0x02,找回密码0x03,修改密码0x04
//客户端获取文件列表0x11,上传文件0x12,下载文件0x13,共享文件0x14,除获取列表外各种文件业务应答0x15
//心跳0x21
//中介服务器信息填充0x30
int optid; // 操作码:请求0x00 和 应答0x01
int usrlenth;// 包体的长度
int packet_seq; //包序号
int packet_sum; //包总数
char to_fd[6]; //目标服务器名称(填信息时为本服务器名称)
char dstAddr[6]; //预留
int syn; // 判断包头是否正确 0x04
}packet_header_t;
//解压包处的修改
bool PacketBase::pack()
{
Body = new char[Body_Size];
memcpy(m_Data, &this->Head, sizeof(packet_header_t));
memcpy(m_Data + sizeof(packet_header_t), this->Body, sizeof(Body)); //Õâ¸öBody³¤¶ÈÔÚ·â°üµÄʱºò¶¨
memcpy(m_Data + sizeof(packet_header_t) + sizeof(Body), &this->Tail, sizeof(packet_tali_t));
return true;
}
bool PacketBase::unpack()
{
if (Body_Size <= 0) { //Èç¹ûÊý¾Ý²»×ã
std::cout << "Êý¾Ý°üÆÆËð" << std::endl;
return false;
}
Body = new char[Body_Size]; //·ÀÖ¹Ô½½ç¿ÉÒÔÔÚÕâÀïÏÂÊÖ
memcpy(&this->Head, m_Data, sizeof(packet_header_t)); //ÏȽ«°üÍ·¶Á³ö
memcpy(Body, m_Data + sizeof(packet_header_t), sizeof(Body));
memcpy(&this->Tail, m_Data + sizeof(packet_header_t) + sizeof(Body), sizeof(packet_tali_t));
return true;
}
今天的分享就到这里。