def register_thread(self, uid):
     """register a thread for redirected IO, registers the current thread"""
     mythread = threading.currentThread()
     mythread.setName(uid)
     input_buffers[uid] = StringBuffer()
     threads[uid] = threading.currentThread()
     debug_msg("registering thread for uid=%s" % uid, 8)

    def acquire_inputstream(self,urlpath):

        streaminfo = None
        # First check mappers, without locking, assuming video stream URL paths won't match mappers
        for mapper in self.mappers:
            streaminfo = mapper.get(urlpath)
            # print >>sys.stderr,"videoserv: get_inputstream: Got streaminfo",`streaminfo`,"from",`mapper`
            if streaminfo is not None and (streaminfo['statuscode'] == 200 or streaminfo['statuscode'] == 301):
                return streaminfo

        if DEBUGLOCK:
            print >>sys.stderr,"vs: acq_input: lock",urlpath,currentThread().getName()
        self.lock.acquire()
        try:
            streaminfo = self.urlpath2streaminfo.get(urlpath,None)
        finally:
            if DEBUGLOCK:
                print >>sys.stderr,"vs: acq_input: unlock",urlpath,currentThread().getName()
            self.lock.release()

        # Grab lock of video stream, such that other threads cannot read from it. Do outside self.lock
        if streaminfo is not None and 'lock' in streaminfo:
            if DEBUGLOCK:
                print >>sys.stderr,"vs: acq_input: stream: lock",urlpath,currentThread().getName()
            streaminfo['lock'].acquire()
        return streaminfo

    def theThread(self, lockType):
        import sys
        if sys.version_info[0] < 3 :
            name = currentThread().getName()
        else :
            name = currentThread().name

        if lockType ==  db.DB_LOCK_WRITE:
            lt = "write"
        else:
            lt = "read"

        anID = self.env.lock_id()
        if verbose:
            print "%s: locker ID: %s" % (name, anID)

        for i in xrange(1000) :
            lock = self.env.lock_get(anID, "some locked thing", lockType)
            if verbose:
                print "%s: Aquired %s lock: %s" % (name, lt, lock)

            self.env.lock_put(lock)
            if verbose:
                print "%s: Released %s lock: %s" % (name, lt, lock)

        self.env.lock_id_free(anID)

    def _getProperList(self):
        """
        Walk providers for propers
        """
        propers = {}

        search_date = datetime.datetime.today() - datetime.timedelta(days=2)

        # for each provider get a list of the
        origThreadName = threading.currentThread().name
        providers = [x for x in sickbeard.providers.sortedProviderList(sickbeard.RANDOMIZE_PROVIDERS) if x.isActive()]
        for curProvider in providers:
            threading.currentThread().name = origThreadName + " :: [" + curProvider.name + "]"

            logger.log(u"Searching for any new PROPER releases from " + curProvider.name)

            try:
                curPropers = curProvider.findPropers(search_date)
            except AuthException, e:
                logger.log(u"Authentication error: " + ex(e), logger.DEBUG)
                continue
            except Exception, e:
                logger.log(u"Error while searching " + curProvider.name + ", skipping: " + ex(e), logger.DEBUG)
                logger.log(traceback.format_exc(), logger.DEBUG)
                continue

    def CCSAssert(self, obj, ExpectationDict, unique_id):
        '''
        CCS断言入口
        '''
        try:
            self.obj = obj
            self.obj.connMy.select_db(self.obj.dbnameMy)   #选择数据库
            self.curMy = self.obj.connMy.cursor()
            ExpDict, BASE64_ExpDict = self.parseExpectationDict(ExpectationDict)
            PrintLog('debug', '[%s] 提取加密字段数据: ExpDict: %s\nBASE64_ExpDict: %s', threading.currentThread().getName(), ExpDict, BASE64_ExpDict)

            #检查base64加密数据
            self.checkBASE64_ExpDict(BASE64_ExpDict, unique_id)

            #检查明文数据
            self.checkExpDict(ExpDict, unique_id)
            return 'PASS',

        except TableNoneError as e:
            PrintLog('info', '[%s] TableNoneError: TableName: %s', threading.currentThread().getName(), unicode(e))
            return 'NONE',unicode(e)
        except AssertionError as e:
            PrintLog('info', '[%s] AssertionError: %s', threading.currentThread().getName(),unicode(e.args[0]))
            return 'FAIL',unicode(e.args[0])
        except Exception as e:
            PrintLog('exception',e)
            return 'ERROR',unicode(e)

    def readerThread(self, d, readerNum):
        import sys
        if sys.version_info[0] < 3 :
            name = currentThread().getName()
        else :
            name = currentThread().name

        finished = False
        while not finished:
            try:
                txn = self.env.txn_begin(None, self.txnFlag)
                c = d.cursor(txn)
                count = 0
                rec = c.first()
                while rec:
                    count += 1
                    key, data = rec
                    self.assertEqual(self.makeData(key), data)
                    rec = c.next()
                if verbose: print "%s: found %d records" % (name, count)
                c.close()
                txn.commit()
                finished = True
            except (db.DBLockDeadlockError, db.DBLockNotGrantedError), val:
                if verbose:
                    print "%s: Aborting transaction (%s)" % (name, val.args[1])
                c.close()
                txn.abort()

    def ClientHandler(self, replyHandler):
        
        t = threading.currentThread()
        log.info("Running %s"  % t.name)
        
        while self.active:
            #blocks on recv, but may timeout
            try:
                rxdata = self.sock.recv(BUFF)
                log.debug("Data Received: %s" %(repr(rxdata)))
            except socket.timeout:
                continue

            try:
                jdata = json.loads(rxdata.strip())
            except:
                log.info("ClientHandler could not parse JSON string: %s" % repr(rxdata))
                continue
            
            log.debug('Client RX jdata: %s'  %(repr(jdata)))
            replyHandler(jdata)

        #cleanup
        self.sock.close()
        log.info("Leaving %s" % threading.currentThread().name)

    def processXmlQuerySync(rspec,url=None):
	    #check if provisioning / monitoring / etc
	if threading.currentThread().callBackURL:
		url = threading.currentThread().callBackURL
        if not rspec.query.provisioning == None :
            status = SyncThread.startMethodAndJoin(ProvisioningDispatcher.processProvisioning, rspec.query.provisioning, url)
            return status

    def __call__(self, parameters):
        DefaultSlave.__call__(self, parameters)

        # Handle the message
        args = self.messageArgs
        msg = "Handling RemoveFromWorkflowManagementLocationList message: %s" % \
                                                                    str(args)
        logging.debug(msg)
        myThread = threading.currentThread()

        # Validate arguments
        if args.has_key("FilesetMatch") and args.has_key("WorkflowId") \
        and args.has_key("Locations"):
            locations = args['Locations'].split(",")
            try:
                myThread.transaction.begin()
                for loc in locations:
                    self.unmarkLocation.execute(workflow = args['WorkflowId'], \
                                                fileset_match = args['FilesetMatch'], \
                                                location = loc, \
                                                conn = myThread.transaction.conn, \
                                                transaction = True)
                myThread.transaction.commit()
            except:
                myThread.transaction.rollback()
                raise
        else:
            logging.error("Received malformed parameters: %s" % str(args))

        # Report as done
        myThread = threading.currentThread()
        myThread.msgService.finish()

    def setUp(self):
        "make a logger instance and create tables"


        self.testInit = TestInit(__file__)
        self.testInit.setLogging()
        self.testInit.setDatabaseConnection()
        self.testInit.setSchema()

        myThread = threading.currentThread()
        if myThread.dialect == 'MySQL':
            myThread.create = """
create table test (bind1 varchar(20), bind2 varchar(20)) ENGINE=InnoDB """
        if myThread.dialect == 'SQLite':
            myThread.create = """
                create table test (bind1 varchar(20), bind2 varchar(20))"""

        myThread.insert = """
insert into test (bind1, bind2) values (:bind1, :bind2) """
        myThread.insert_binds = \
          [ {'bind1':'value1a', 'bind2': 'value2a'},\
            {'bind1':'value1b', 'bind2': 'value2b'},\
            {'bind1':'value1c', 'bind2': 'value2d'} ]
        myThread.select = "select * from test"

        myThread = threading.currentThread()
        myThread.transaction = Transaction(myThread.dbi)
        myThread.transaction.processData(myThread.create)
        myThread.transaction.processData(myThread.insert, myThread.insert_binds)
        myThread.transaction.commit()

        return

    def __call__(self, parameters):
        DefaultSlave.__call__(self, parameters)

        # Handle the message
        args = self.messageArgs
        logging.debug("Handling AddWorkflowToManage message: %s" % str(args))
        myThread = threading.currentThread()

        # Validate arguments
        if "FilesetMatch" in args and "WorkflowId" in args \
        and "SplitAlgo" in args and "Type" in args:
            try:
                myThread.transaction.begin()
                self.addManagedWorkflow.execute(workflow = args['WorkflowId'], \
                                                fileset_match =  args['FilesetMatch'], \
                                                split_algo =  args['SplitAlgo'], \
                                                type =  args['Type'], \
                                                conn = myThread.transaction.conn, \
                                                transaction = True)
                myThread.transaction.commit()
            except:
                myThread.transaction.rollback()
                raise
        else:
            logging.error("Received malformed parameters: %s" % str(args))

        # Report as done
        myThread = threading.currentThread()
        myThread.msgService.finish()

    def main(self, argv, opts_etc=None):
        if threading.currentThread().getName() == "MainThread":
            threading.currentThread().setName("mt")

        err, opts, source, sink = self.opt_parse(argv)
        if err:
            return err

        if opts_etc:
            opts.etc = opts_etc # Used for unit tests, etc.

        logging.info(self.name + "...")
        logging.info(" source : %s", source)
        logging.info(" sink   : %s", sink)
        logging.info(" opts   : %s", opts.safe)

        source_class, sink_class = self.find_handlers(opts, source, sink)
        if not source_class:
            return "error: unknown type of source: " + source
        if not sink_class:
            return "error: unknown type of sink: " + sink

        err = sink_class.check_source(opts, source_class, source, sink_class, sink)
        if err:
            return err

        try:
            return pump.PumpingStation(opts, source_class, source,
                                       sink_class, sink).run()
        except KeyboardInterrupt:
            return "interrupted."

    def UBASAssert(self, obj, response, tablemaxid, ExpectationDict):
        '''
        UBAS断言入口
        '''
        try:
            #检查响应
            response.encoding = response.apparent_encoding
            assert response.status_code == 200, u'HTTP响应码错误'

            responseContent =  unicode(response.content, "utf-8")
            responseContentDict = json.loads(responseContent)

            Expectation_HTTPResponse = ExpectationDict['HTTPResponse']
            Expectation_fieltlist = Expectation_HTTPResponse.keys()
            Expectation_valuelist = Expectation_HTTPResponse.values()

            PrintLog('debug','[%s] 比较响应数据与期望数据各字段: Expectation_HTTPResponse: %s responseContentDict: %s', threading.currentThread().getName(), Expectation_HTTPResponse, responseContentDict)
            for i in xrange(len(Expectation_fieltlist)):
                assert Expectation_valuelist[i] == responseContentDict[Expectation_fieltlist[i]], u'响应%s字段值不正确' % Expectation_fieltlist[i]

            del ExpectationDict['HTTPResponse']
            self._checkdbdata(obj, tablemaxid, ExpectationDict)
            return 'PASS',

        except TableNoneError as e:
            PrintLog('info', '[%s] TableNoneError: TableName: %s', threading.currentThread().getName(), unicode(e))
            return 'NONE',unicode(e)
        except AssertionError as e:
            PrintLog('info', '[%s] AssertionError: %s', threading.currentThread().getName(),unicode(e.args[0]))
            return 'FAIL',unicode(e.args[0])
        except Exception as e:
            PrintLog('exception',e)
            return 'ERROR',unicode(e)

    def RunCCSCase(self, sheet, testid, TestData, TestEnvironment):
        '''
        运行授信接口用例
        '''
        ModCCSO = ModCCS.ModCCS()
        dbinfo = ModCCSO.getRuncaseEnvironment_db(TestEnvironment)

        #读取超时时间
        timeouttask = ModCCSO.getRuncaseEnvironment_Timeouttask(TestEnvironment)
        timeoutdelay = 0

        #测试数据解析
        parseResult = ModCCSO.parseParamsForDriver(TestData, sheet, testid)
        if parseResult is False:
            raise ValueError('parseParamsForDriver is Fail')
        TestData, unique_id = parseResult


        #驱动执行获得返回的唯一userid
        PrintLog('debug', '[%s] 驱动执行:TestData:%s\nunique_id: %s', threading.currentThread().getName(), TestData, unique_id)
        DriverO = Interface_Driver.Interface_DoData(dbinfo)
        DriverResult = DriverO.insert(TestData)    #执行用例
        PrintLog('debug', '[%s] 执行结果:DriverResult:%s', threading.currentThread().getName(), DriverResult)

        #装载任务参数
        if DriverResult is False:
            return False
        taskargs = unique_id
        return timeouttask, timeoutdelay, taskargs

    def writerThread(self, d, keys, readers):
        import sys
        if sys.version_info[0] < 3 :
            name = currentThread().getName()
        else :
            name = currentThread().name
        if verbose:
            print "%s: creating records %d - %d" % (name, start, stop)

        count=len(keys)//len(readers)
        count2=count
        for x in keys :
            key = '%04d' % x
            dbutils.DeadlockWrap(d.put, key, self.makeData(key),
                                 max_retries=12)

            if verbose and x % 100 == 0:
                print "%s: records %d - %d finished" % (name, start, x)

            count2-=1
            if not count2 :
                readers.pop().start()
                count2=count

        if verbose:
            print "%s: thread finished" % name

    def RunUBASCase(self, sheet, testid, TestData, TestEnvironment):
        '''
        运行数据回流接口用例
        '''
        ModUBASO = ModUBAS.ModUBAS()
        dbinfo = ModUBASO.getRuncaseEnvironment_db(TestEnvironment)
        url = ModUBASO.getRuncaseEnvironment_Url(TestEnvironment)
        headers = ModUBASO.getRuncaseEnvironment_Headers(TestEnvironment)

        #读取超时时间
        timeouttask = ModUBASO.getRuncaseEnvironment_Timeouttask(TestEnvironment)
        timeoutdelay = 0

        #获取待插入数据前表的maxid
        Expectation = self.TestCaseO.get_Expectation(sheet, testid)
        TableList = ModUBASO.parseExpForDriver(Expectation)
        if TableList is False:
            return False
        DriverOO = Interface_Driver.Interface_DoData(dbinfo)
        TableMaxid = DriverOO.getTableMaxid(TableList)

        #驱动执行获得response
        PrintLog('debug', '[%s] 驱动执行:headers:%s TestData:%s', threading.currentThread().getName(), headers, TestData)
        DriverO = Interface_Driver.Interface_Http(url)
        DriverResult = DriverO.post(headers, TestData)  #执行用例
        PrintLog('debug', '[%s] 执行结果:DriverResult:%s', threading.currentThread().getName(), DriverResult)

        #装载任务参数
        if DriverResult is False:
            return False
        taskargs = DriverResult,TableMaxid
        return timeouttask, timeoutdelay, taskargs

def searchForNeededEpisodes(show, episodes):
    foundResults = {}

    didSearch = False

    # ask all providers for any episodes it finds
    origThreadName = threading.currentThread().name
    providers = [x for x in sickbeard.providers.sortedProviderList() if x.isActive() and not x.backlog_only]
    for curProviderCount, curProvider in enumerate(providers):
        if curProvider.anime_only and not show.is_anime:
            logger.log(u"" + str(show.name) + " is not an anime skiping ...")
            continue

        threading.currentThread().name = origThreadName + " :: [" + curProvider.name + "]"

        try:
            logger.log(u"Searching RSS cache ...")
            curFoundResults = curProvider.searchRSS(episodes)
        except exceptions.AuthException, e:
            logger.log(u"Authentication error: " + ex(e), logger.ERROR)
            if curProviderCount != len(providers):
                continue
            break
        except Exception, e:
            logger.log(u"Error while searching " + curProvider.name + ", skipping: " + ex(e), logger.ERROR)
            if curProviderCount != len(providers):
                continue
            break

    def RunUPSLabelCase(self, sheet, testid, TestData, TestEnvironment):
        '''
        运行用户画像标签用例
        '''
        ModUPSLabelO = ModUPSLabel.ModUPSLabel()
        dbinfo = ModUPSLabelO.getRuncaseEnvironment_Userdb(TestEnvironment)
        function = ModUPSLabelO.DriverCbFunction

        #读取超时时间
        timeouttask = ModUPSLabelO.getRuncaseEnvironment_Timeouttask_label(TestEnvironment)

        #读取延时时间
        Expectation = self.TestCaseO.get_Expectation(sheet, testid)
        parseRt = ModUPSLabelO.parseExpForDriver(Expectation)
        if parseRt is True:
            timeoutdelay = ModUPSLabelO.getRuncaseEnvironment_Timeoutdelay_label(TestEnvironment)
        else:
            timeoutdelay = 0

        #测试数据解析
        TestData = ModUPSLabelO.parseParamsForDriver(TestData)

        #驱动执行获得返回的唯一userid
        PrintLog('debug', '[%s] 驱动执行:TestData:%s CbFunction:%s', threading.currentThread().getName(), TestData, function.__name__)
        DriverO = Interface_Driver.Interface_DoData(dbinfo)
        DriverResult = DriverO.insert(TestData, function)    #执行用例
        PrintLog('debug', '[%s] 执行结果:DriverResult:%s', threading.currentThread().getName(), DriverResult)

        #装载任务参数
        if DriverResult is False:
            return False
        taskargs = DriverResult
        return timeouttask, timeoutdelay, taskargs

    def _load_data(cr, module_name, idref, mode, kind):
        """

        kind: data, demo, test, init_xml, update_xml, demo_xml.

        noupdate is False, unless it is demo data or it is csv data in
        init mode.

        """
        try:
            if kind in ('demo', 'test'):
                threading.currentThread().testing = True
            for filename in _get_files_of_kind(kind):
                _logger.info("loading %s/%s", module_name, filename)
                noupdate = False
                if kind in ('demo', 'demo_xml') or (filename.endswith('.csv') and kind in ('init', 'init_xml')):
                    noupdate = True
            if tools.config.options.get('noupdate_if_unchanged'):
                cr.execute('select value from ir_values where name=%s and key=%s', (pathname, 'digest'))
                olddigest = (cr.fetchone() or (None,))[0]
                if olddigest is None:
                    cr.execute('insert into ir_values (name, model, key, value) values (%s, %s, %s, NULL)',
                               (pathname, 'ir_module_module', 'digest',))
                digest = md5.md5(fp.read()).hexdigest()
                fp.seek(0)
                if digest == olddigest:
                    noupdate = True
                else:
                    cr.execute('update ir_values set value=%s where name=%s and key=%s', (digest, pathname, 'digest'))        
                tools.convert_file(cr, module_name, filename, idref, mode, noupdate, kind, report)
        finally:
            if kind in ('demo', 'test'):
                threading.currentThread().testing = False

 def algorithm(self, parameters=None):
     """
     Actually runs the code
     """
     logging.debug("Running JSM.JobCreator")
     try:
         self.pollSubscriptions()
     except WMException:
         # self.close()
         myThread = threading.currentThread()
         if getattr(myThread, 'transaction', False) \
                 and getattr(myThread.transaction, 'transaction', False):
             myThread.transaction.rollback()
         raise
     except Exception as ex:
         # self.close()
         myThread = threading.currentThread()
         if getattr(myThread, 'transaction', False) \
                 and getattr(myThread.transaction, 'transaction', False):
             myThread.transaction.rollback()
         # Handle temporary connection problems (Temporary)
         if "(InterfaceError) not connected" in str(ex):
             logging.error(
                 'There was a connection problem during the JobCreator algorithm, I will try again next cycle')
         else:
             msg = "Failed to execute JobCreator. Error: %s" % str(ex)
             logging.exception(msg)
             raise JobCreatorException(msg)