#define REDIS_BIO_CLOSE_FILE    0 /*关闭一个系统调用*/

#define REDIS_BIO_AOF_FSYNC     1 /*文件数据刷新到磁盘*/

#define REDIS_BIO_NUM_OPS       2/*支持任务类型数*/

static pthread_t bio_threads[REDIS_BIO_NUM_OPS];/*多线程情况下线程的个数*/

static pthread_mutex_t bio_mutex[REDIS_BIO_NUM_OPS];/*互斥锁*/

static pthread_cond_t bio_condvar[REDIS_BIO_NUM_OPS];/*条件变量*/

static list *bio_jobs[REDIS_BIO_NUM_OPS];/*后台任务链表，每个线程一个，根据下标区分*/

static unsigned long long bio_pending[REDIS_BIO_NUM_OPS];/*记录每个线程剩余的任务数*/

struct bio_job

{/*后台IO节点，任务是通过链表来维护的，这个算是链表内的节点数据，第一个是创建时间，第二个是指定参数。*/

    time_t time; /* Time at which the job was created. */

    /* Job specific arguments pointers. If we need to pass more than three

     * arguments we can just pass a pointer to a structure or alike. */

    void *arg1, *arg2, *arg3;

};

void bioInit(void); /初始化变量数据和线程数据/

void bioCreateBackgroundJob(int type, void *arg1, void *arg2, void *arg3);/*添加一个任务*/void *bioProcessBackgroundJobs(void *arg);/*处理指定的任务数组*/

unsigned long long bioPendingJobsOfType(int type);/*获取剩余的任务数*/

void bioKillThreads(void);/*关闭线程*/

void bioInit(void)

{

    pthread_attr_t attr;

    pthread_t thread;

    size_t stacksize;

    int j;

    /*初始化条件变量和互斥锁*/

    for (j = ; j < REDIS_BIO_NUM_OPS; j++)

    {

        pthread_mutex_init(&bio_mutex[j],NULL);

        pthread_cond_init(&bio_condvar[j],NULL);

        bio_jobs[j] = listCreate();

        bio_pending[j] = ;

    }

    /*初始化线程属性，自动增加线程栈的大小*/

    pthread_attr_init(&attr);

    pthread_attr_getstacksize(&attr,&stacksize);

    if (!stacksize) stacksize = ; /* The world is full of Solaris Fixes */

    while (stacksize < REDIS_THREAD_STACK_SIZE) stacksize *= ;

    pthread_attr_setstacksize(&attr, stacksize);

    /* Ready to spawn our threads. We use the single argument the thread

     * function accepts in order to pass the job ID the thread is

     * responsible of. */

    for (j = ; j < REDIS_BIO_NUM_OPS; j++)

    {

        void *arg = (void*)(unsigned long) j;

        if (pthread_create(&thread,&attr,bioProcessBackgroundJobs,arg) != )

        {

            redisLog(REDIS_WARNING,"Fatal: Can't initialize Background Jobs.");

            exit();

        }

        bio_threads[j] = thread;

    }

}//这个没啥可说的，无非是初始化同步的数据和线程数据

 /*创建一个后台任务，创建任务支持多线程

 添加事件和处理事件就像是消费者和生产者的问题一样，采用互斥和条件变量来控制

 参数1是添加到哪个队列中，剩余是三个参数*/

 void bioCreateBackgroundJob(int type, void *arg1, void *arg2, void *arg3)

 {

     struct bio_job *job = zmalloc(sizeof(*job));

     job->time = time(NULL);

     job->arg1 = arg1;

     job->arg2 = arg2;

     job->arg3 = arg3;

     /*添加的时候要注意并发问题，添加完了发个信号*/

     pthread_mutex_lock(&bio_mutex[type]);

     listAddNodeTail(bio_jobs[type],job);

     bio_pending[type]++;

     pthread_cond_signal(&bio_condvar[type]);

     pthread_mutex_unlock(&bio_mutex[type]);

 }

/*事件消费函数*/

void *bioProcessBackgroundJobs(void *arg)

{

    struct bio_job *job;

    unsigned long type = (unsigned long) arg;

    sigset_t sigset;

    /* Make the thread killable at any time, so that bioKillThreads()

     * can work reliably. */

    pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);/*设置线程为收到cancle信号马上退出*/

    pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);

    pthread_mutex_lock(&bio_mutex[type]);

    /* Block SIGALRM so we are sure that only the main thread will

     * receive the watchdog signal. */

    sigemptyset(&sigset);

    sigaddset(&sigset, SIGALRM);

    if (pthread_sigmask(SIG_BLOCK, &sigset, NULL))

        redisLog(REDIS_WARNING,

            "Warning: can't mask SIGALRM in bio.c thread: %s", strerror(errno));

    while()

    {

        listNode *ln;

        /* The loop always starts with the lock hold. */

        if (listLength(bio_jobs[type]) == )

        {

            pthread_cond_wait(&bio_condvar[type],&bio_mutex[type]);

            continue;

        }

        /* Pop the job from the queue. */

        ln = listFirst(bio_jobs[type]);

        job = ln->value;

        /* It is now possible to unlock the background system as we know have

         * a stand alone job structure to process.*/

        pthread_mutex_unlock(&bio_mutex[type]);

　　　　 /*处理任务的时候解除互斥锁，提高效率*/

        /* 区分任务类型，具体执行 */

        if (type == REDIS_BIO_CLOSE_FILE)

        {

            close((long)job->arg1);

        }

        else if (type == REDIS_BIO_AOF_FSYNC)

        {

            aof_fsync((long)job->arg1);

        }

        else

        {

            redisPanic("Wrong job type in bioProcessBackgroundJobs().");

        }

        zfree(job);

        /* Lock again before reiterating the loop, if there are no longer

         * jobs to process we'll block again in pthread_cond_wait(). */

        pthread_mutex_lock(&bio_mutex[type]);/*修改公共资源的的是要加锁发，防止并发问题*/

        listDelNode(bio_jobs[type],ln);/*删除节点，并减掉任务数*/

        bio_pending[type]--;

    }

}

 void bioKillThreads(void)

 {

     int err, j;

     for (j = ; j < REDIS_BIO_NUM_OPS; j++)

     {

         if (pthread_cancel(bio_threads[j]) == ) /*给线程发送结束信号，发送成功之后等待其结束*/

         {

             if ((err = pthread_join(bio_threads[j],NULL)) != )

             {

                 redisLog(REDIS_WARNING,"Bio thread for job type #%d can be joined: %s",j, strerror(err));

             }

             else

             {

                 redisLog(REDIS_WARNING,"Bio thread for job type #%d terminated",j);

             }

         }

     }

 }