FTP文件管理项目（本地云）项目日报（五）

//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);
    }
}

//对头包的修改

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;   
}