Disque 源码分析

struct disqueServer {

    // 其他属性……

    /* Jobs & Queues */

    // jobs 是一个字典，它包含了服务器储存的所有任务
    // 其中字典的键为任务 ID ，而键对应的值则是一个 job.h/job 结构
    dict *jobs;                 /* Main jobs hash table, by job ID. */

    // queues 也是一个字典，它包含了服务器储存的所有队列
    // 其中字典的键为队列的名字，而键对应的则是一个 queue.h/queue 结构
    dict *queues;               /* Main queues hash table, by queue name. */

    // 其他属性……
};

typedef struct queue {

    // 队列的名字
    robj *name;      /* Queue name as a string object. */

    // 一个跳跃表，储存了所有被放进队列里面的任务
    // 各个任务按照任务 ID 从小到大有序地进行排列
    skiplist *sl;    /* The skiplist with the queued jobs. */

    // 其他属性……

} queue;

typedef struct job {

    // 任务 ID
    char id[JOB_ID_LEN];    /* Job ID. */

    // 其他属性……

    // 所属的队列
    robj *queue;            /* Job queue name. */

    // 任务的内容
    sds body;               /* Body, or NULL if job is just an ACK. */

    // 其他属性……

} job;

// 传播任务及其内容
/* This data section is used in different message types where we need to
 * transmit one or multiple full jobs.
 *
 * Used by: ADDJOB, YOURJOBS. */
typedef struct {
    uint32_t numjobs;   /* Number of jobs stored here. */
    uint32_t datasize;  /* Number of bytes following to describe jobs. */
    /* The variable data section here is composed of 4 bytes little endian
     * prefixed length + serialized job data for each job:
     * [4 bytes len] + [serialized job] + [4 bytes len] + [serialized job] ...
     * For a total of exactly 'datasize' bytes. */
     unsigned char jobs_data[8]; /* Defined as 8 bytes just for alignment. */
} clusterMsgDataJob;

// 传播任务 ID
/* This data section is used when we need to send just a job ID.
 *
 * Used by: GOTJOB, SETACK, and many more. */
typedef struct {
    char id[JOB_ID_LEN];
    uint32_t aux; /* Optional field:
                     For SETACK: Number of nodes that may have this message.
                     For QUEUEJOB: Delay starting from msg reception. */
} clusterMsgDataJobID;

// 向其他节点请求传播某个队列的任务
/* This data section is used by NEEDJOBS to specify in what queue we need
 * a job, and how many jobs we request. */
typedef struct {
    uint32_t count;     /* How many jobs we request. */
    uint32_t qnamelen;  /* Queue name total length. */
    char qname[8];      /* Defined as 8 bytes just for alignment. */
} clusterMsgDataNeedJobs;

void addjobCommand(client *c) {

    // 变量声明
    long long replicate = server.cluster->size > 3 ? 3 : server.cluster->size;
    long long ttl = 3600*24;
    long long retry = -1;
    long long delay = 0;
    long long maxlen = 0; /* Max queue length for job to be accepted. */
    mstime_t timeout;
    int j, retval;
    int async = 0;  /* Asynchronous request? */
    int extrepl = getMemoryWarningLevel() > 0; /* Replicate externally? */
    static uint64_t prev_ctime = 0;

    // 参数分析
    /* Parse args. */
    for (j = 4; j < c->argc; j++) {
        char *opt = c->argv[j]->ptr;
        int lastarg = j == c->argc-1;
        if (!strcasecmp(opt,"replicate") && !lastarg) {
            retval = getLongLongFromObject(c->argv[j+1],&replicate);
            if (retval != DISQUE_OK || replicate <= 0 || replicate > 65535) {
                addReplyError(c,"REPLICATE must be between 1 and 65535");
                return;
            }
            j++;
        } else if (!strcasecmp(opt,"ttl") && !lastarg) {
            retval = getLongLongFromObject(c->argv[j+1],&ttl);
            if (retval != DISQUE_OK || ttl <= 0) {
                addReplyError(c,"TTL must be a number > 0");
                return;
            }
            j++;
        } else if (!strcasecmp(opt,"retry") && !lastarg) {
            retval = getLongLongFromObject(c->argv[j+1],&retry);
            if (retval != DISQUE_OK || retry < 0) {
                addReplyError(c,"RETRY time must be a non negative number");
                return;
            }
            j++;
        } else if (!strcasecmp(opt,"delay") && !lastarg) {
            retval = getLongLongFromObject(c->argv[j+1],&delay);
            if (retval != DISQUE_OK || delay < 0) {
                addReplyError(c,"DELAY time must be a non negative number");
                return;
            }
            j++;
        } else if (!strcasecmp(opt,"maxlen") && !lastarg) {
            retval = getLongLongFromObject(c->argv[j+1],&maxlen);
            if (retval != DISQUE_OK || maxlen <= 0) {
                addReplyError(c,"MAXLEN must be a positive number");
                return;
            }
            j++;
        } else if (!strcasecmp(opt,"async")) {
            async = 1;
        } else {
            addReply(c,shared.syntaxerr);
            return;
        }
    }

    /* Parse the timeout argument. */
    if (getTimeoutFromObjectOrReply(c,c->argv[3],&timeout,UNIT_MILLISECONDS)
        != DISQUE_OK) return;

    /* REPLICATE > 1 and RETRY set to 0 does not make sense, why to replicate
     * the job if it will never try to be re-queued if case the job processing
     * is not acknowledged? */
    if (replicate > 1 && retry == 0) {
        addReplyError(c,"With RETRY set to 0 please explicitly set  "
                        "REPLICATE to 1 (at-most-once delivery)");
        return;
    }

    /* DELAY greater or equal to TTL is silly. */
    if (delay >= ttl) {
        addReplyError(c,"The specified DELAY is greater than TTL. Job refused "
                        "since would never be delivered");
        return;
    }

    /* When retry is not specified, it defaults to 1/10 of the TTL. */
    if (retry == -1) {
        retry = ttl/10;
        if (retry == 0) retry = 1;
    }

    /* Check if REPLICATE can't be honoured at all. */
    int additional_nodes = extrepl ? replicate : replicate-1;

    if (additional_nodes > server.cluster->reachable_nodes_count) {
        if (extrepl &&
            additional_nodes-1 == server.cluster->reachable_nodes_count)
        {
            addReplySds(c,
                sdsnew("-NOREPL Not enough reachable nodes "
                       "for the requested replication level, since I'm unable "
                       "to hold a copy of the message for memory usage "
                       "problems.rn"));
        } else {
            addReplySds(c,
                sdsnew("-NOREPL Not enough reachable nodes "
                       "for the requested replication levelrn"));
        }
        return;
    }

    // 检查队列是否已经达到最大长度
    /* If maxlen was specified, check that the local queue len is
     * within the requested limits. */
    if (maxlen && queueNameLength(c->argv[1]) > (unsigned long) maxlen) {
        addReplySds(c,
            sdsnew("-MAXLEN Queue is already longer than "
                   "the specified MAXLEN countrn"));
        return;
    }

    // 创建一个新任务
    /* Create a new job. */
    job *job = createJob(NULL,JOB_STATE_WAIT_REPL,ttl); // 创建任务
    job->queue = c->argv[1];                            // 记录任务所在队列的名字
    incrRefCount(c->argv[1]);
    job->repl = replicate;

    /* If no external replication is used, add myself to the list of nodes
     * that have a copy of the job. */
    if (!extrepl)
        dictAdd(job->nodes_delivered,myself->name,myself);

    // 设置任务的各项属性
    /* Job ctime is milliseconds * 1000000. Jobs created in the same
     * millisecond gets an incremental ctime. The ctime is used to sort
     * queues, so we have some weak sorting semantics for jobs: non-requeued
     * jobs are delivered roughly in the order they are added into a given
     * node. */
    job->ctime = mstime()*1000000;
    if (job->ctime <= prev_ctime) job->ctime = prev_ctime+1;
    prev_ctime = job->ctime;

    job->etime = server.unixtime + ttl;
    job->delay = delay;
    job->retry = retry;
    job->body = sdsdup(c->argv[2]->ptr);

    /* Set the next time the job will be queued. Note that once we call
     * enqueueJob() the first time, this will be set to 0 (never queue
     * again) for jobs that have a zero retry value (at most once jobs). */
    if (delay) {
        job->qtime = server.mstime + delay*1000;
    } else {
        /* This will be updated anyway by enqueueJob(). */
        job->qtime = server.mstime + retry*1000;
    }

    /* Register the job locally in all the cases but when the job
     * is externally replicated and asynchronous replicated at the same
     * time: in this case we don't want to take a local copy at all. */
    if (!(async && extrepl) && registerJob(job) == DISQUE_ERR) {
        /* A job ID with the same name? Practically impossible but
         * let's handle it to trap possible bugs in a cleaner way. */
        serverLog(DISQUE_WARNING,"ID already existing in ADDJOB command!");
        freeJob(job);
        addReplyError(c,"Internal error creating the job, check server logs");
        return;
    }

    /* For replicated messages where ASYNC option was not asked, block
     * the client, and wait for acks. Otherwise if no synchronous replication
     * is used, or ASYNC option was enabled, we just queue the job and
     * return to the client ASAP.
     *
     * Note that for REPLICATE > 1 and ASYNC the replication process is
     * best effort. */
    if (replicate > 1 && !async) {

        // 决定以同步方式向其他节点传播这个任务

        c->bpop.timeout = timeout;
        c->bpop.job = job;
        c->bpop.added_node_time = server.mstime;
        blockClient(c,DISQUE_BLOCKED_JOB_REPL);
        setJobAssociatedValue(job,c);
        /* Create the nodes_confirmed dictionary only if we actually need
         * it for synchronous replication. It will be released later
         * when we move away from JOB_STATE_WAIT_REPL. */
        job->nodes_confirmed = dictCreate(&clusterNodesDictType,NULL);
        /* Confirm itself as an acknowledged receiver if this node will
         * retain a copy of the job. */
        if (!extrepl) dictAdd(job->nodes_confirmed,myself->name,myself);
    } else {

        // 决定以异步方式传递这个任务

        if (job->delay == 0) {
            if (!extrepl) enqueueJob(job); /* Will change the job state. */
        } else {
            /* Delayed jobs that don't wait for replication can move
             * forward to ACTIVE state ASAP, and get scheduled for
             * queueing. */
            job->state = JOB_STATE_ACTIVE;
            updateJobAwakeTime(job,0);
        }
        addReplyJobID(c,job);
        AOFLoadJob(job);
    }

    // 向其他节点传播这个任务……

    /* If the replication factor is > 1, send REPLJOB messages to REPLICATE-1
     * nodes. */
    if (additional_nodes > 0)
        clusterReplicateJob(job, additional_nodes, async);

    /* If the job is asynchronously and externally replicated at the same time,
     * send a QUEUE message ASAP to one random node, and delete the job from
     * this node right now. */
    if (async && extrepl) {
        dictEntry *de = dictGetRandomKey(job->nodes_delivered);
        if (de) {
            clusterNode *n = dictGetVal(de);
            clusterSendEnqueue(n,job,job->delay);
        }
        /* We don't have to unregister the job since we did not registered
         * it if it's async + extrepl. */
        freeJob(job);
    }
}