redisbio pass valgrind test

redisbio/Makefrag

 OBJDIRS	+= redisbio
 
-REDISBIO_SRCS := token.cc
+REDISBIO_SRCS := token.cc adlist.cc bio.cc zmalloc.cc
 
 all: $(OBJDIR)/libredisbio.so
 

redisbio/adlist.cc

+#include <stdlib.h>
+#include "adlist.h"
+#include "zmalloc.h"
+
+
+list *listCreate(void){
+    struct list *list;
+    // 分配内存
+    if ((list = (struct list*)zmalloc(sizeof(*list))) == NULL)
+        return NULL;
+    // 初始化属性
+    list->head = list->tail = NULL;
+    list->len = 0;
+    list->dup = NULL;
+    list->free = NULL;
+    list->match = NULL;
+    return list;
+}
+
+
+list *listAddNodeTail(list *list, void *value){
+    listNode *node;
+    // 为新节点分配内存
+    if ((node = (listNode*)zmalloc(sizeof(*node))) == NULL)
+        return NULL;
+    // 保存值指针
+    node->value = value;
+    // 目标链表为空
+    if (list->len == 0) {
+        list->head = list->tail = node;
+        node->prev = node->next = NULL;
+    // 目标链表非空
+    } else {
+        node->prev = list->tail;
+        node->next = NULL;
+        list->tail->next = node;
+        list->tail = node;
+    }
+    // 更新链表节点数
+    list->len++;
+    return list;
+}
+
+void listDelNode(list *list, listNode *node){
+    // 调整前置节点的指针
+    if (node->prev)
+        node->prev->next = node->next;
+    else
+        list->head = node->next;
+    // 调整后置节点的指针
+    if (node->next)
+        node->next->prev = node->prev;
+    else
+        list->tail = node->prev;
+    // 释放值
+    if (list->free) list->free(node->value);
+    // 释放节点
+    zfree(node);
+    // 链表数减一
+    list->len--;
+}
+
+void listRelease(list *list){
+    unsigned long len;
+    listNode *current, *next;
+    // 指向头指针
+    current = list->head;
+    // 遍历整个链表
+    len = list->len;
+    while(len--) {
+        next = current->next;
+        // 如果有设置值释放函数，那么调用它
+        if (list->free) list->free(current->value);
+        // 释放节点结构
+        zfree(current);
+        current = next;
+    }
+    // 释放链表结构
+    zfree(list);
+}
+
+

redisbio/adlist.h

+#pragma once
+
+typedef struct listNode {
+    // 前置节点
+    struct listNode *prev;
+    // 后置节点
+    struct listNode *next;
+    // 节点的值
+    void *value;
+} listNode;
+
+typedef struct listIter {
+    // 当前迭代到的节点
+    listNode *next;
+    // 迭代的方向
+    int direction;
+} listIter;
+
+
+typedef struct list {
+    // 表头节点
+    listNode *head;
+    // 表尾节点
+    listNode *tail;
+    // 节点值复制函数
+    void *(*dup)(void *ptr);
+    // 节点值释放函数
+    void (*free)(void *ptr);
+    // 节点值对比函数
+    int (*match)(void *ptr, void *key);
+    // 链表所包含的节点数量
+    unsigned long len;
+} list;
+
+
+
+#define listLength(l) ((l)->len)
+// 返回给定链表的表头节点
+// T = O(1)
+#define listFirst(l) ((l)->head)
+// 返回给定链表的表尾节点
+// T = O(1)
+#define listLast(l) ((l)->tail)
+// 返回给定节点的前置节点
+// T = O(1)
+#define listPrevNode(n) ((n)->prev)
+// 返回给定节点的后置节点
+// T = O(1)
+#define listNextNode(n) ((n)->next)
+// 返回给定节点的值
+// T = O(1)
+#define listNodeValue(n) ((n)->value)
+
+// 将链表 l 的值复制函数设置为 m
+// T = O(1)
+#define listSetDupMethod(l,m) ((l)->dup = (m))
+// 将链表 l 的值释放函数设置为 m
+// T = O(1)
+#define listSetFreeMethod(l,m) ((l)->free = (m))
+// 将链表的对比函数设置为 m
+// T = O(1)
+#define listSetMatchMethod(l,m) ((l)->match = (m))
+
+// 返回给定链表的值复制函数
+// T = O(1)
+#define listGetDupMethod(l) ((l)->dup)
+// 返回给定链表的值释放函数
+// T = O(1)
+#define listGetFree(l) ((l)->free)
+// 返回给定链表的值对比函数
+// T = O(1)
+#define listGetMatchMethod(l) ((l)->match)
+
+/* Prototypes */
+list *listCreate(void);
+void listRelease(list *list);
+list *listAddNodeHead(list *list, void *value);
+list *listAddNodeTail(list *list, void *value);
+list *listInsertNode(list *list, listNode *old_node, void *value, int after);
+void listDelNode(list *list, listNode *node);
+listIter *listGetIterator(list *list, int direction);
+listNode *listNext(listIter *iter);
+void listReleaseIterator(listIter *iter);
+list *listDup(list *orig);
+listNode *listSearchKey(list *list, void *key);
+listNode *listIndex(list *list, long index);
+void listRewind(list *list, listIter *li);
+void listRewindTail(list *list, listIter *li);
+void listRotate(list *list);
+
+/* Directions for iterators 
+ *
+ * 迭代器进行迭代的方向
+ */
+// 从表头向表尾进行迭代
+#define AL_START_HEAD 0
+// 从表尾到表头进行迭代
+#define AL_START_TAIL 1
+
+
+

redisbio/bio.cc

+#include <string.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <signal.h>
+#include "zmalloc.h"
+#include "bio.h"
+#include "adlist.h"
+
+int(*userFunction)(unsigned long,void*) = NULL;
+
+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];
+
+/*function for the working thread*/
+void *
+bioProcessBackgroundJobs(void *arg);
+
+#define REDIS_THREAD_STACK_SIZE (1024*1024*4)
+
+void bioInit(void) {
+    pthread_attr_t attr;
+    pthread_t thread;
+    size_t stacksize;
+    unsigned int j;
+    /* Initialization of state vars and objects 
+     *
+     */
+    for (j = 0u; 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] = 0;
+    }
+    /* Set the stack size as by default it may be small in some system 
+     *
+     */
+    pthread_attr_init(&attr);
+    pthread_attr_getstacksize(&attr,&stacksize);
+    if (!stacksize) stacksize = 1; /* The world is full of Solaris Fixes */
+    while (stacksize < REDIS_THREAD_STACK_SIZE) stacksize *= 2;
+    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 = 0u; j < REDIS_BIO_NUM_OPS; j++) {
+        void *arg = (void*)(unsigned long) j;
+        if (pthread_create(&thread,&attr,bioProcessBackgroundJobs,arg) != 0) {
+            printf("Fatal: Can't initialize Background Jobs.\n");
+            exit(1);
+        }
+        bio_threads[j] = thread;
+    }
+}
+
+void bioCreateBackgroundJob(int type,const void *arg1, void *arg2, int stop) {
+    struct bio_job *job = (struct bio_job*)zmalloc(sizeof(*job));
+    job->time = time(NULL);
+    job->arg1 = const_cast<void*>(arg1);
+    job->arg2 = arg2;
+    job->stop = stop;
+    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]);
+}
+
+//each thread is responsible for the list indexed by 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);
+    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))
+            printf("Warning: can't mask SIGALRM in bio.c thread:\n");
+    while(1) {
+        listNode *ln;
+        /* The loop always starts with the lock hold. */
+        if (listLength(bio_jobs[type]) == 0) {
+            pthread_cond_wait(&bio_condvar[type],&bio_mutex[type]);
+            continue;
+        }
+
+        /* Pop the job from the queue. 
+         *
+         * 取出（但不删除）队列中的首个任务
+         */
+        ln = listFirst(bio_jobs[type]);
+        job = (struct bio_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]);
+
+        int stop = userFunction(type,(void*)job);
+        zfree(job);
+        if(stop == 1){
+            break;
+        }else{
+        }
+        /* 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]--;
+    }
+    return NULL;
+}
+
+unsigned long long bioPendingJobsOfType(int type) {
+    unsigned long long val;
+    pthread_mutex_lock(&bio_mutex[type]);
+    val = bio_pending[type];
+    pthread_mutex_unlock(&bio_mutex[type]);
+    return val;
+}
+
+void bioKillThreads(void) {
+    unsigned int err, j;
+    for (j = 0; j < REDIS_BIO_NUM_OPS; j++) {
+        if ((err = pthread_join(bio_threads[j],NULL)) != 0) {
+//            printf("Bio thread for job type can be joined\n");
+        } else {
+//            printf("Bio thread for job type terminated\n");
+        }
+    }
+}

redisbio/bio.h

+#pragma once
+#include <time.h>
+void bioInit(void);
+void bioCreateBackgroundJob(int type,const void *arg1, void *arg2, int stop);
+unsigned long long bioPendingJobsOfType(int type);
+void bioWaitPendingJobsLE(int type, unsigned long long num);
+time_t bioOlderJobOfType(int type);
+void bioKillThreads(void);
+struct bio_job {
+    time_t time;
+    void *arg1, *arg2;
+    int stop;
+};
+extern 
+int(*userFunction)(unsigned long,void*);
+
+const unsigned int REDIS_BIO_NUM_OPS = 10;

test_redisbio/biotest.cc

+#include <string.h>
+#include <stdio.h>
+#include <unistd.h>
+#include <string>
+#include <vector>
+#include "redisbio/zmalloc.h"
+#include "redisbio/adlist.h"
+#include "redisbio/bio.h"
+
+//return 0 to continue, return 1 to exit.
+static
+int
+work(unsigned long id, void *input){
+    struct bio_job *task = (struct bio_job *)input;
+    if(task->stop == 1) return 1;
+    char * query = (char*)(task->arg1);
+    (void)query;
+//    printf("id=%lu %s\n",id,query);
+    return 0;
+}
+
+int
+main(){
+    userFunction = work;
+    bioInit();
+    std::vector<std::string> input{
+        "a",
+        "b",
+        "c",
+        "d",
+        "e",
+        "f",
+        "g",
+        "h",
+        "i",
+        "j"
+    };
+    int type = 0;    
+    for(int i=0;i<10000;i++) {
+        bioCreateBackgroundJob(type,(input[i%10].c_str()),NULL,0);
+        type+=1;
+        type%=10;
+    }
+    for(int i=0;i<10;i++) {
+        bioCreateBackgroundJob(type,NULL,NULL,1);
+        type+=1;
+        type%=10;
+    }
+    printf("tobe joined\n");
+    bioKillThreads();
+    return 0;
+}