jobs.py 文件源码

def create(self, redis):
        """ 
        Create job by create job context in redis, each job will create len(hosts) meta keys, 
            conflict dectection is done by redis key exists check of job meta keys. 

        If no conflict, job context (the `meta_keys`) will created via redis pipeline 
            to avoid operate confilct.
        """
        pipeline = redis.pipeline()
        try:
            pipeline.watch(self.meta_keys)
            for key in self.meta_keys:
                if pipeline.exists(key):
                    raise JobConflictError("operate conflict, job already exists on some host(s)")

            LOG.info("going to create job meta data <{0}>".format(';'.join(self.meta_keys)))
            pipeline.multi()
            for key in self.meta_keys:
                pipeline.hmset(key, dict(startat=self._startat))
            pipeline.execute()
            LOG.info("job meta data create finished, <{0}>".format(';'.join(self.meta_keys)))

        except WatchError:
            LOG.info("conflict detected on job meta data create <{0}>".format(';'.join(self.meta_keys)))
            raise JobConflictError("operate conflict, try again later")
        finally:
            pipeline.reset()