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./r/n"));   } else {    addReplySds(c,     sdsnew("-NOREPL Not enough reachable nodes "         "for the requested replication level/r/n"));   }   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 count/r/n"));   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);  } }