def parse_result(self, res):
        retcode = res.status
        result = RestResult(retcode, res.reason)
        content_type = res.getheader('Content-Type')
        if content_type and content_type.find(";"):
            types = content_type.split(";")
            for t in types:
                t = t.strip()
                if t.startswith('charset'):
                    result.charset = t
                else: result.content_type = t
        else: result.content_type = content_type
        content_length = res.getheader('Content_Length')
        if content_length:
            result.content_length = int(content_length)
            result.content = res.read()
            while len(result.content) < result.content_length:
                result.content += res.read()
        else:
            result.content = res.read()
            result.content_length = len(result.content)

        last_modified = res.getheader('Last-Modified')
        if last_modified:
            logger.info('HTTP response from %s, Last-Modified header is: %s', self.host, last_modified)
            result.last_modified = last_modified

        session_id = res.getheader('X-Schemus-Session')
        if session_id:
            logger.info('HTTP response from %s, X-Schemus-Session header is: %s', self.host, session_id)
            result.session_id = session_id

        return result

 def run(self):
     logger.info('dispatch thread is running...')
     while not self.__stop_event.isSet():
         try:
             self.__event.wait(timeout=5)
         except Exception:
             pass
         else:
             if self.__cur_tasks:
                 if self._run_tasks():
                     self.__event.set()
                 else:
                     self.__event.clear()
             else:
                 self.__cur_tasks = g_task_dict.get()
                 if not self.__cur_tasks:
                     self.__task_handler_pool.kill_idle_handlers()
                     self.__event.clear()
                     continue
                 if self._run_tasks():
                     self.__event.set()
                 else:
                     self.__event.clear()
         finally:
             pass
     logger.info('dispatch thread %s stopped', self.name)

def main():
    """
    The function gets all the user callback objects whose notify times are between the
    current time and 1 minute from now. For each callback object it calls the function notification_callback,
    each call is spawned as separate thread. This way, all users get concurrent notifications and if there
    are a large number of notifications we won't be taking up more than 1 minute to complete the execution.
    All the spawned threads are joined with this function, so that it waits until they complete their
    execution.
    @params: None
    @output: None
    """
    now = datetime.datetime.now()
    now_time = datetime.time(now.hour, now.minute, now.second)
    
    xmins_frm_now = now + datetime.timedelta(seconds = 60)
    xmins_frm_now = datetime.time(xmins_frm_now.hour, xmins_frm_now.minute, xmins_frm_now.second)

    today = datetime.date.today()
    days = {0:'mon', 1:'tue', 2:'wed', 3:'thu', 4:'fri', 5:'sat', 6:'sun'}
    today = days[today.weekday()]
        
    user_callbacks = models.UserCallback.objects.filter(notify_time__gte=now_time, notify_time__lt=xmins_frm_now)
    logger.info("number of user callbacks to process: %d" % user_callbacks.count())

    threads = []
    for user_callback in user_callbacks:
        thread = threading.Thread(target = notification_callback, args = (user_callback, today,))
        threads.append(thread)
        try:
            thread.start()
        except Exception, e:
            logger.error("error while running user callback: %s" % e)

    def run(self):
        logger.info('Devices\' information sync with airwatch started at %s under account %d!' % (str(time.time()), self.aw_account))
        try:
            logger.info('Step 1: Get device information from RS.')
            rs_retval = self._get_devinfo_from_rs()
            logger.debug('Account %d includes %d device needed sync with airwatch!' % (self.aw_account, rs_retval))
            if rs_retval <= 0:
                # report
                return  # exit, no work to do

            logger.info('Step 2: Get device information from airwatch.')
            aw_retval = self._get_devinfo_from_aw(self.devinfo_original_queue, self.devinfo_refresh_queue)
            logger.debug('Account %d, get %d device information from airwatch!' % (self.aw_account, aw_retval))
            if (aw_retval != rs_retval):
                #The devices do not exist in airwatch needed to be updated as unenroll status.
                logger.warning('Account %d, Original device number (%d) and refresh device number (%d) NOT match!' % (self.aw_account, rs_retval, aw_retval))

            logger.info('Step 3: Sync device information to RS.')
            sync_retval = self._sync_devinfo()
            logger.debug('Account %d, sync %d device information with RS finished!' % (self.aw_account, sync_retval))
            if (aw_retval != sync_retval):
                #causes of updating RS fail
                logger.warning('Account %d, Refresh device number (%d) and Sync device number (%d) NOT match!' % (self.aw_account, aw_retval, sync_retval))

            # Step 4: Update the defect devices to be unenroll.
            if self.m_defect_devs:
                defect_cnt = len(self.m_defect_devs)
                logger.info('Step 4: Set %d devices to be "unenroll" status!' % defect_cnt)
                ok_cnt = self._unenroll_dev_status(self.m_defect_devs)
                if defect_cnt != ok_cnt:
                    logger.warning('Account %d, Set %d devices to be "unenroll status failed!"' % (self.aw_account, (defect_cnt - ok_cnt)))
            # Step 5: Report
        except Exception, e:
            logger.error(repr(e))

    def _compute_detailed_scores(self, item_ids, query_item_ids=None, max_terms=20):
        # if set to None we assume previously queried items
        if query_item_ids is None:
            query_item_ids = self.item_ids

        # if the query vector is different than previously computed
        # or not computed at all, we need to recompute it.
        if not hasattr(self, 'q') or query_item_ids != self.item_ids:
            if not self.is_valid_query(query_item_ids):
                return []
            else:
                logger.info('Computing the query vector ...')
                self._make_query_vector()

        # computing the score for each item
        scores = []
        for id in item_ids:
            if id not in self.item_id_to_index:
                scores.append(utils._O(total_score=0, scores=[]))
                continue

            xi = self.X[self.item_id_to_index[id]]
            xi_ind = xi.indices
            
            feat = (self.index_to_feat[i] for i in xi_ind)
            
            qi = self.q.transpose()[xi_ind]
            qi = scipy.asarray(qi).flatten()

            sc = sorted(zip(feat, qi), key=lambda x: (x[1], x[0]), reverse=True)
            total_score = qi.sum()
            scores.append(utils._O(total_score=total_score, scores=sc[0:max_terms]))

        return scores

 def _reset_notify(self):
     from service_mgr.lib.service.service_main import ServiceMgr
     services = ServiceMgr().get_run_services(US_DEVICE_END)
     [device_type_notify(service.control_rpc, self.get_device_type()) for service in services]
     logger.info("DeviceTypeMgr _reset_notify services:%s, device_type:%s"
                 % ([service.id for service in services],
                    self.get_device_type()))

 def top_phrases(self, msg):
     logger.info("%s: sending top phrases to user %s" % (msg.text, user_log_string(msg)))
     lines = ["Top phrases (use '/say ID' to choose phrase):"]
     for audio_id, count in self.dataset.top_phrases.most_common(5):
         lines.append(
             "*[%2d]*  _%s_ (%d reproductions)" % (audio_id, self.dataset.get_audio_data(audio_id)['title'], count))
     self.reply_to(msg, "\n".join(lines), parse_mode="Markdown")

 def send_phrase(self, msg, audio_id):
     if 0 <= audio_id < len(self.dataset.data_keys):
         audio, duration = self.dataset.get_phrase(audio_id)
         logger.info("%s: sending phrase #%d to user %s" % (msg.text, audio_id, user_log_string(msg)))
         self.send_audio(msg.chat.id, audio, duration)
     else:
         bot.send_random_phrase(msg)

def main():
    rel_info = ReleaseInfo()
    start_directory = os.getcwd()
    parse_args(rel_info)
    try:
        setup_test_branch(rel_info)
        merge_branches(rel_info)
        test_tag_doc(rel_info)

        if rel_info.args.merge_only:
            return

        build_platform(rel_info)
    except KeyboardInterrupt:
        pass
    except:
        logger.exception("[FAILED] Please check your changes. "\
                         "You can not pass this checker unless your branch "\
                         "can be merged without conflicts.")
        logger.info("Note: all repos are in test branch %s" % rel_info.test_branch)
    else:
        logger.info(
            "[PASS] Your changes can be successfully merged and build.")
    finally:
        cleanup(rel_info)
        os.chdir(start_directory)

    def do_access(self, resource, method_name='POST', data=None, headers=None):
        """request resource in specified HTTP method
        Parameters:
            resource: an absolute path which supplied by server.
            method_name: HTTP METHOD, can be 'GET', 'POST', 'PUT' or 'DELETE', the default is 'POST'
            data: the data should transfer to server.
            headers: HTTP headers, should in dict form.
        """
        boundary = templates.generate_boundary()
        http_headers = self._generate_headers(boundary)
        if headers:
            http_headers.update(headers)
        if data:
            body = self._generate_post_body(data, boundary)
        else:
            body = ""
            http_headers.pop('Content-Type')

        logger.info('metanate request body: %s', body)
        r = super(self.__class__, self).do_access(resource, method_name='POST', data=body, headers=http_headers)

        if r.code == 200:
            if not self.session_id:
                self.session_id = r.session_id
                logger.info('HTTP response header X-Schemus-Session is: %s', self.session_id)
            if r.content_type == "text/ldif":
                r.content = templates.parse_ldif(r.content)
            elif r.content == "application/json":
                r.content = json.loads(r.content)
        else:
            logger.error('metanate request: %s error occurred: %s: %s', resource, r.code, r.content)
            r.content = {'ErrorCode': r.code, 'Message': r.reason, 'Detail': r.content}

        return r

    def process_result(self, result, live_server, token):
        """
        Munge the result from a client and add the data into our repository.
        """
        # Record the data from our transaction in the Tokens
        # Ensures that duplicated work is not accepted after the
        # token has been consumed (to prevent previous generation's
        # evals from affecting the current gen)
        if not result.has_key('FAILED'):
            if token in self.Tokens.keys():
                self.data_lock.acquire()
                if token[0] == 'map':
                    (shard, shuffle_keys) = result.values()[0]
                    self.shuffle_keys.update(shuffle_keys)
                    logger.info('Have %d shuffle keys (only alpha)' % (len(self.shuffle_keys)))
                    self.map_result_shards.append(shard)
                elif token[0] == 'shuffle':
                    self.shuffle_result_shards.append(result.values()[0])
                self.data_lock.release()

                # Consume the token if we've performed enough Evals
                self.print_complete(token, self.Tokens, live_server, live_servers, idle_servers)
                del self.Tokens[token]
            else: # Delete extraneous data
                if token[0] == 'map':
                    (shard, shuffle_keys) = result.values()[0]
                    os.remove(shard)

    def start_net(self, name):
        logger.info("Start network " + name)

        if name in self.networks:
            conn = libvirt.open(self.urls[self.networks[name].htype])
        else:
            conn = libvirt.open(self.def_connection)

        try:
            net = conn.networkLookupByName(name)
            if not net.isActive():
                logger.debug("Network registered in libvirt - start it")
                net.create()
            else:
                logger.debug("Network already active")

        except libvirt.libvirtError:
            try:
                logger.debug("No such network in libvirt")
                net = self.networks[name]
            except KeyError:
                msg = "Can't found network {0!r}".format(name)
                logger.error(msg)
                raise CloudError(msg)

            xml = xmlbuilder.XMLBuilder('network')
            xml.name(name)
            xml.bridge(name=net.bridge)
            with xml.ip(address=net.ip, netmask=net.netmask):
                xml.dhcp.range(start=net.ip1, end=net.ip2)

            logger.debug("Create network")
            conn.networkCreateXML(str(xml))

    def finish(self, link):
        """
        The stop point of test case. At least one stop point must be exist
        in each test case.
        
        This keyword represents the end of the test case. Test tool will send
        stop signal to dispatcher then the dispatcher will notify all other
        tools to stop. After all tools stopped, the dispatcher will send test
        reports to the verification module to be verified.
        """
        
        if self.msg_buff is None:
            self._init_msg_buff()
        
        src, dst = link.split('--')
        
        ne_link = self._topology.link(src, dst)

        if ne_link is None:
            logger.warn('No link between NEs to send message. '
                        '%s--%s: Stop' % (src, dst))
            return

        if ((not (ne_link.simulator == src and ne_link.dut == dst)) and
            (not (ne_link.simulator == dst and ne_link.dut == src))):
            logger.warn('Finish on unassigned link. '
                        '%s--%s: Stop' % (src, dst))
            return

        msg_str = self._compose_finish()
        msg_buff = self._get_msg_buff(ne_link.names[0])
        msg_buff.append(msg_str)
        
        logger.info('"Finish" added to buffer of %s' %
                    ne_link.names[0])

    def user_insert(self, user_name, password, jid, jid_pwd, type, device_type="", des=""):
        """
        用户插入
        :param user_name: 用户名
        :param password: 用户密码
        :param jid: 用户jid
        :param jid_pwd: 用户jid密码
        :param type: 用户类型
        :param device_type: 设备类型
        :param des: 描述
        :return: {"result"}
        """
        if not user_name\
                or not password\
                or not jid\
                or not jid_pwd\
                or type is None:
            return {'result': error_code.ERROR_PARAMS_ERROR}

        user_info = {"user_name": user_name,
                     "password": password,
                     "des": des,
                     "jid": jid,
                     "jid_pwd": jid_pwd,
                     "device_type": device_type,
                     "type": type,
                     "create_time": datetime.datetime.now()}

        g_RedisCoordinator.set(user_name, user_info)

        logger.info("RpcHandler::user_insert, user_name:%s, password:%s, jid:%s, jid_pwd:%s, device_type:%s, type:%s, des:%s"
                    % (user_name, password, jid, jid_pwd, device_type, type, des))
        return {'result': error_code.ERROR_SUCCESS}

    def __create_distro(self, compose, distro_file):
        rhel_version = self.get_rhel_version(compose)
        if rhel_version == 5:
            compose_url = "http://download.englab.nay.redhat.com/pub/rhel/rel-eng/%s/tree-x86_64/" % compose
        elif rhel_version == 6:
#           compose_url = "http://download.englab.nay.redhat.com/pub/rhel/rel-eng/%s/%s/Server/x86_64/os/" % (compose, self.get_rhel_version(compose))
    	    compose_url = "http://download.englab.nay.redhat.com/pub/rhel/rel-eng/%s/compose/Server/x86_64/os/" % compose
    	elif rhel_version == 7:
            compose_url = "http://download.englab.nay.redhat.com/pub/rhel/rel-eng/%s/compose/Server/x86_64/os/" % compose
        if not self.__check_file_exist(distro_file):
            cmd = ('cat <<EOF > %s\n'
                '[General]\n'
                'arch : x86_64\n'
                'breed : redhat\n'
                'comment :\n'
                '\n'
                'kernel : %simages/pxeboot/vmlinuz\n'
                'initrd : %simages/pxeboot/initrd.img\n'
                'kernel_options : biosdevname=0 reboot=pci\n'
                'kernel_options_post :\n'
                'ks_meta :\n'
                'mgmt_classes :\n'
                '\n'
                'os_version : rhel%s\n'
                'redhat_management_key :\n'
                'redhat_management_server :\n'
                'template_files :\n'
                'EOF' % (distro_file, compose_url, compose_url, rhel_version)
                )
            logger.info("Created distro file: %s" % distro_file)
            self.run(cmd)
        else:
            logger.info("Distro file: %s already existed ..." % distro_file)

def init_platform():
    config_file = cfg_from_file('config.yml')
    default_file = cfg_from_file('default.yml')
    logger.info(pprint.pformat(default_file))
    logger.info(pprint.pformat(config_file))

    merge_a_into_b(config_file, config)
    merge_a_into_b(default_file, default)
    default.best_model_path = ''

    if default.gpu == '':
        default.gpu = None
    if default.gpu is not None:
        os.environ["CUDA_VISIBLE_DEVICES"] = default.gpu

    default.distributed = default.world_size > 1
    if default.distributed:
        dist.init_process_group(backend=default.dist_backend, init_method=default.dist_url,
                                world_size=default.world_size)

    default.lr_epoch = [int(ep) for ep in default.lr_step.split(',')]

    if default.seed is not None:
        seed = default.seed
        np.random.seed(seed)
        random.seed(seed)
        torch.manual_seed(seed)
        cudnn.deterministic = True

 def _create_indexes(self, ids, fts):
     logger.info("Creating indices ...")
     self.item_id_to_index = dict(ids)
     self.index_to_item_id = dict((i, id) for id, i in ids.iteritems())
     self.index_to_feat = dict((i, ft) for ft, i in fts.iteritems())
     self.no_items = len(ids)
     self.no_features = len(fts)

    def wait(self, link, tm):
        """
        Wait for certain seconds specified by 'tm'.
        
        'tm' is the time to be waited in millisecond.
        """
        
        if self.msg_buff is None:
            self._init_msg_buff()
        
        src, dst = link.split('--')
        ne_link = self._topology.link(src, dst)
        
        if ne_link is None:
            logger.warn('No link between NEs to send message. '
                        '%s->%s: Wait' % (src, dst))
            return

        if ((not (ne_link.simulator == src and ne_link.dut == dst)) and
            (not (ne_link.simulator == dst and ne_link.dut == src))):
            logger.warn('Stop on unassigned link. '
                        '%s-%s: Stop' % (src, dst))
            return

        msg_str = self._compose_wait(tm)
        msg_buff = self._get_msg_buff(ne_link.names[0])
        msg_buff.append(msg_str)

        logger.info('"wait %s" added to buffer %s' %
                    (tm, ne_link.names[0]))

    def retrieve_data(self, link,
                      result, operation, paras='',
                      start_message='', end_message=''):
        
        if self.msg_buff is None:
            self._init_msg_buff()
        
        src, dst = link.split('--')
        
        ne_link = self._topology.link(src, dst)

        if ne_link is None:
            logger.warn('No link between NEs to send message. '
                        '%s--%s: Retrieve %s' % (src, dst, operation))
            return

        if ((not (ne_link.simulator == src and ne_link.dut == dst)) and
            (not (ne_link.simulator == dst and ne_link.dut == src))):
            logger.warn('Retrieve data on unassigned link. '
                        '%s--%s: Retrieve %s' % (src, dst, operation))
            return

        msg_str = self._compose_retrieve_data(result, operation, paras,
                                              start_message, end_message)
        msg_buff = self._get_msg_buff(ne_link.names[0])
        msg_buff.append(msg_str)
        
        logger.info('"Retrieve" added to buffer of %s' %
                    ne_link.names[0])

 def check(self, expr, info=''):
     
     if self.msg_buff is None:
         self._init_msg_buff()
     
     self.chk_lst.check(expr, info)
     logger.info('"Check" added to check list')