def do_establish_taskqueue(self, tqname=None, namespace='mobile', account=1, version=1, description=None):
        '''
        Establish one taskqueue. If the task queue exists, the establishment would been cancelled.

        Params:
            tqname: Name of task queue.
            namespace: Namespace of the task queue.
            account: Account ID.
            version: Version of the resource, default value 1.
            description: Description for the task queue
        Return:
            True: Establish success.
            False: Establish failed.
        '''
        assert(tqname)
        resource = TQRES_TEMPLATE_STR % (version, account, namespace, tqname)
        logger.debug('Establish one taskqueue - name:%s, resource:%s, description:%s' % (tqname, resource, description))
        result = self.do_get_taskqueue(tqname, namespace, account, version)
        if result == True:
            logger.debug('Task Queue exists, canceling the establishment!')
            return True
        else:
            if isinstance(description, basestring):
                description = {'description' : description}
            elif isinstance(description, dict):
                pass
            else:
                logger.error('Establish Task Queue failed! Unknow description type!')
                return False
            data = json.dumps(description)
            result = self.m_wrest.do_access(resource, 'PUT', data=data, headers=None)
            if result.code >= 200 and result.code < 300:
                logger.debug('Establish Task Queue success!')
                return True

    def start_service(self):
        """
        启动服务
        :return:
        """
        args = ArgumentParser().args
        try:
            Jobs().add_interval_job(UPDATE_INTERVAL, self.update)

            if not self.is_sm:
                port = {"tcp": args.tcp_port}
                port.update({"https": args.http_port} if args.is_https else {"http": args.http_port})
                self.adv = ServiceAdvertiser(self.service_type, port, self.get_jid(), self.service_version)
                self.adv.advertise()

            checker_ls = []
            self.add_port_checker(args, checker_ls)
            PortChecker(checker_ls).start()

            self.services(args, self.thread_ls)
            self.__sys_services(args, self.thread_ls)

            logger.warn("start services for %s, args:%s" % (self.service_type, args))
            gevent.joinall([thread.start() for thread in self.thread_ls])
        except:
            logger.error(traceback.format_exc())
            sys.exit(0)

    def run(self):
        while self.processor.shared.paused():
            time.sleep(1)

        self.ircname = self.host + ' ' + self.getname()
        logger.info("joining IRC")

        while not self.processor.shared.stopped():
            client = irc.client.IRC()
            try:
                c = client.server().connect('irc.freenode.net', 6667, self.nick, self.password, ircname=self.ircname)
            except irc.client.ServerConnectionError:
                logger.error('irc', exc_info=True)
                time.sleep(10)
                continue

            c.add_global_handler("welcome", self.on_connect)
            c.add_global_handler("join", self.on_join)
            c.add_global_handler("quit", self.on_quit)
            c.add_global_handler("kick", self.on_kick)
            c.add_global_handler("whoreply", self.on_who)
            c.add_global_handler("namreply", self.on_name)
            c.add_global_handler("disconnect", self.on_disconnect)
            c.set_keepalive(60)

            self.connection = c
            try:
                client.process_forever()
            except BaseException as e:
                logger.error('irc', exc_info=True)
                time.sleep(10)
                continue

        logger.info("quitting IRC")

    def find_services(self, service_type, rdm_type=RT_CPU_USAGE_RDM, rdm_param=1):
        """
        查找服务列表
        :param service_type:服务类型
        :param rdm_type:随机类型，0选择cpu使用率最低的；1一致性hash选择
        :param rdm_param:如果随机类型是0,参数整形,表示随机个数
                         如果随机类型是1,list形式,hash key 列表
        :return:
        """
        if rdm_type == RT_HASH_RING and not isinstance(rdm_param, list):
            rdm_param = [rdm_param]

        if not ServiceGrpMgr().get_service_grp(service_type):
            logger.error("TcpHandler::find_service Failed!!!, ERROR_PARAMS_ERROR, service_type:%s" % service_type)
            return {}

        service_obj_ls = ServiceMgr().get_run_services(service_type, rdm_type, rdm_param)
        if not service_obj_ls:
            return {}

        result = {}
        for idx, service_obj in enumerate(service_obj_ls):
            result_key = idx if rdm_type == RT_CPU_USAGE_RDM else rdm_param[idx]
            result[result_key] = {"ip": service_obj.ip,
                                  "port": service_obj.port,
                                  "jid": service_obj.jid}
        return result

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 format_tq_payload(self, source=None, encode=False, **keyval):
        '''
        Format payload members for task queue operation commands.

        Params:
            source : Input payload dict. If this param is empty, the interface would create a new one.
            encode : Enable/Disable encoding. Default value is false. It would remove '\n' when enable encoding.
                    Converts data to json str at first, then with base64 encoding.
            keyval : Key name and value of the specific payload.

        Return:
            dict of the payload
            Str with base64 encoded.
            Exception message
        '''
        if source:
            assert(isinstance(source, dict))
            retval = source
        else:
            retval = {}

        retval.update(keyval)  # append key
        if encode:
            # convert into json
            try:
                retval = json.dumps(retval)
            except Exception, e:
                logger.error('Encoding payload error: %s!' % e)
                raise Exception(e)

            # base64
            retval = base64.encodingstring(retval).replace("\n", '')

    def do_get_task_total(self, tqname=None, namespace='mobile', account=1, version=1, tags=None):
        '''
        Get total number of the specific task queue.

        Params:
            tqname: Name of task queue.
            namespace: Namespace of the task queue.
            account: Account ID.
            version: Version of the resource, default value 1.
            tags: Search conditions.

        Return:
            Total number of tasks on this taskqueue
            Exception message
        '''
        assert(tqname)
        resource = TQRES_TEMPLATE_STR % (version, account, namespace, tqname)
        resource = '/'.join([resource, 'stats'])
        if tags:
            data = json.dumps({'tags' : tags}, encoding="utf-8")
        else:
            data = tags
        logger.debug('Get total tasks number - resource:%s, data:%s' % (resource, data))
        result = self.m_wrest.do_access(resource, 'POST', data=data, headers=None)
        if result.code >= 200 and result.code < 300:
            logger.debug('Get Task Queue stats success!')
            try: #parse stat data
                tqs = json.loads(result.content)
                total_tasks = tqs.get('total_tasks')
                return total_tasks
            except Exception, e:
                logger.error('Get Task Queue stats Failed! %s' % repr(e))

def _post(url, data=None, json=None, etag=None, **kwargs):
    try:
        res = requests.post(url, data=data, json=json, **kwargs)
    except requests.exceptions.RequestException as e:
        logger.error(">> [POST] {0} failed with exception: {1}".format(url, e))
        return None, e
    return _rest_check_response(res)

def _delete(url, **kwargs):
    try:
        res = requests.delete(url, **kwargs)
    except requests.exceptions.RequestException as e:
        logger.error(">> [DELETE] {0} failed with exception: {1}".format(url, e))
        return None, e
    return _rest_check_response(res)

def third_user_login(request):

    """
    The method is third user login
    @param request: third_key, third_type
    @return:
    """
    if request.method == 'GET':
        return response_fail_to_mobile(500, REQUEST_METHOD_ERROR)
    try:
        username = request.POST['third_key']
        password = request.POST['third_type']
        user = authenticate(username=username, password=password, type=3)
        auth_login(request, user)
        logic = UserProfileLogic.get_user_profile_by_user_id(user_id=request.user.id)

        last_login = datetime_convert_current_timezone(logic.last_login_time)
        if last_login is not None:
            last_login = datetime_to_string(last_login)
        if logic.user_nick is None:
            user_nick = ''
        else:
            user_nick = logic.user_nick

        rec = dict()
        rec['session_id'] = request.session._session_key
        rec['user_id'] = request.user.id
        rec['user_name'] = user_nick
        rec['last_login'] = last_login
        UserProfileLogic.update_last_login_time(request.user.id)
        return response_success_to_mobile(rec)
    except Exception as e:
        error(e)
        auth_logout(request)
        return response_fail_to_mobile(500, 'Third part account login fail!')

    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 send_sms(cls, msg, level):
        if level not in ('error', 'fatal'):
            raise ValueError("level MUST be error or fatal")
        to_ls = cls.error_s_ls if level == 'error' else cls.fatal_s_ls
        result = []
        for mobile in to_ls:
            http_client = HttpRpcClient()
            headers = {
                'Accept': 'application/json',
                'Content-Type': 'application/x-www-form-urlencoded;charset=utf-8'
            }

            msg = {
                'apiKey': API_KEY,
                'appId': APP_ID,
                'templateId': TEMPLATE['LOGGER'],
                'mobile': mobile,
                'param': msg
            }
            body = ujson.dumps(msg)
            response = ujson.loads(http_client.fetch_async(URL, headers, body))
            if response['result'] != 'SUCCESS':
                logger.error('SMSService error_code: %s at %s' % (response['reason'], time.ctime()))
            result.append(response)
        return result

        def wechat_recv_param_wapper(self, *args, **kwargs):
            if not signature_checker(token,
                                     kwargs['timestamp'],
                                     kwargs['nonce'],
                                     kwargs['signature']):
                logger.error("wechat_recv_wapper, failed, access_token:%s args:%s, kwargs:%s" %
                             (token, args, kwargs))
                return "-1"

            # 如果没有内容尽是验证，则即可返回
            if not self.request.body:
                return kwargs.get("echostr", "0")

            # 内容解析
            crypt = WXBizMsgCrypt(token, encodingAESKey, appid)
            ret, xml_body = crypt.DecryptMsg(self.request.body,
                                               kwargs['msg_signature'],
                                               kwargs['timestamp'],
                                               kwargs['nonce'])
            assert ret == WXBizMsgCrypt_OK

            order_dic_body = xmltodict.parse(xml_body)
            dic_body = dict(order_dic_body["xml"])
            kwargs['body'] = dic_body

            logger.info("%s:wechat_recv_wapper args:%s kwargs:%s" % (fun.__name__, args, kwargs))
            fun(self, *args, **kwargs)
            return kwargs.get("echostr", "0")

    def getfullblock(self, block_hash):
        block = self.bitcoind('getblock', [block_hash])

        rawtxreq = []
        i = 0
        for txid in block['tx']:
            rawtxreq.append({
                "method": "getrawtransaction",
                "params": [txid],
                "id": i,
            })
            i += 1

        postdata = dumps(rawtxreq)
        try:
            respdata = urllib.urlopen(self.bitcoind_url, postdata).read()
        except:
            logger.error("groestlcoind error (getfullblock)",exc_info=True)
            self.shared.stop()

        r = loads(respdata)
        rawtxdata = []
        for ir in r:
            if ir['error'] is not None:
                self.shared.stop()
                print_log("Error: make sure you run groestlcoind with txindex=1; use -reindex if needed.")
                raise BaseException(ir['error'])
            rawtxdata.append(ir['result'])
        block['tx'] = rawtxdata
        return block

def _get(url, params=None, etag=None, **kwargs):
    try:
        res = requests.get(url, params=params, **kwargs)
    except requests.exceptions.RequestException as e:
        logger.error(">> [GET] {0} failed with exception: {1}".format(url, e))
        return None, e
    return _rest_check_response(res)

    def get_block(self, block_hash):
        block = self.bitcoind('getblock', (block_hash,))

        rawtxreq = []
        i = 0
        for txid in block['tx']:
            rawtxreq.append({
                "method": "getrawtransaction",
                "params": (txid,),
                "id": i,
            })
            i += 1
        postdata = dumps(rawtxreq)

        while True:
            try:
                response = urllib.urlopen(self.bitcoind_url, postdata)
                r = load(response)
                response.close()
            except:
                logger.error("bitcoind error (getfullblock)")
                self.wait_on_bitcoind()
                continue
            try:
                rawtxdata = []
                for ir in r:
                    assert ir['error'] is None, "Error: make sure you run bitcoind with txindex=1; use -reindex if needed."
                    rawtxdata.append(ir['result'])
            except BaseException as e:
                logger.error(str(e))
                self.wait_on_bitcoind()
                continue

            block['tx'] = rawtxdata
            return block