def testIsInIOThread(self):
     foreignResult = []
     t = threading.Thread(target=lambda: foreignResult.append(threadable.isInIOThread()))
     t.start()
     t.join()
     self.failIf(foreignResult[0], "Non-IO thread reported as IO thread")
     self.failUnless(threadable.isInIOThread(), "IO thread reported as not IO thread")

 def test_isInIOThread(self):
     """
     L{threadable.isInIOThread} returns C{True} if and only if it is called
     in the same thread as L{threadable.registerAsIOThread}.
     """
     threadable.registerAsIOThread()
     foreignResult = []
     t = threading.Thread(
         target=lambda: foreignResult.append(threadable.isInIOThread()))
     t.start()
     t.join()
     self.assertFalse(
         foreignResult[0], "Non-IO thread reported as IO thread")
     self.assertTrue(
         threadable.isInIOThread(), "IO thread reported as not IO thread")

    def commit_now(self, vacuum=False, exiting=False):
        if self._should_commit and isInIOThread():
            try:
                self._logger.info(u"Start committing...")
                self.execute(u"COMMIT;")
            except:
                self._logger.exception(u"COMMIT FAILED")
                raise
            self._should_commit = False

            if vacuum:
                self._logger.info(u"Start vacuuming...")
                self.execute(u"VACUUM;")

            if not exiting:
                try:
                    self._logger.info(u"Beginning another transaction...")
                    self.execute(u"BEGIN;")
                except:
                    self._logger.exception(u"Failed to execute BEGIN")
                    raise
            else:
                self._logger.info(u"Exiting, not beginning another transaction")

        elif vacuum:
            self.execute(u"VACUUM;")

	def sendMessage(self, sendType, dataDict, callback=None):
		self.updateComplete = False
		if callback != None:
			self.factory.dataQueue.append(EmulatorResponse(dataDict, callback, self.factory))
		self.factory.sendDataToServer(sendType, dataDict)
		if not threadable.isInIOThread() and callback != None:
			return self.waitForUpdate()

    def in_thread_wrapper(*args, **kw):

        if isInIOThread():

            return func(*args, **kw)

        f = Future()

        def twisted_wrapper():
            try:
                d = func(*args, **kw)
                if isinstance(d, Deferred):

                    def _done(result):
                        f.set_result(result)
                        f.done()

                    def _error(e):
                        f.set_exception(e)
                        f.done()

                    d.addCallback(_done)
                    d.addErrback(_error)

                else:
                    f.set_result(d)
                    f.done()

            except Exception, e:
                f.set_exception(e)
                f.done()

 def start(self):
     if self._started:
         raise RuntimeError("worker has already been started")
     if not threadable.isInIOThread():
         raise RuntimeError("worker can only be started in the IO thread")
     self._started = True
     callInGreenThread(self.__run__)

    def shutdown(self):
        """
        Checkpoints the session and closes it, stopping the download engine.
        This method has to be called from the reactor thread.
        """
        assert isInIOThread()

        @inlineCallbacks
        def on_early_shutdown_complete(_):
            """
            Callback that gets called when the early shutdown has been completed.
            Continues the shutdown procedure that is dependant on the early shutdown.
            :param _: ignored parameter of the Deferred
            """
            self.config.write()
            yield self.checkpoint_downloads()
            self.lm.shutdown_downloads()
            self.lm.network_shutdown()
            if self.lm.mds:
                self.lm.mds.shutdown()

            if self.sqlite_db:
                self.sqlite_db.close()
            self.sqlite_db = None

        return self.lm.early_shutdown().addCallback(on_early_shutdown_complete)

    def get_id(self, model, unique, fields):
        ''' Get an ID from the cache or from the database.
        If doesn't exist - create an item.
        All database operations are done from
        the separate thread
        '''
        assert isInIOThread()

        fval = fields[unique]

        try:
            result = self.cache[model][fval]
            self.counters['hit'][model] += 1
            returnValue(result)
        except KeyError:
            self.counters['miss'][model] += 1

        selectors = {unique: fval}

        result, created = yield deferToThreadPool(
            self.reactor, self.read_pool,
            get_or_create,
            model, fields, **selectors)

        result = result.id

        if created:
            self.counters['db_create'][model] += 1
        else:
            self.counters['db_hit'][model] += 1

        self.cache[model][fval] = result
        returnValue(result)

def call_in_twisted_thread(func, *args, **kwargs):
    if threadable.isInIOThread():
        func(*args, **kwargs)
    else:
        from twisted.internet import reactor

        reactor.callFromThread(func, *args, **kwargs)

    def _do_save(self):
        assert not isInIOThread()

        while not self.write_queue.empty():
            items = []

            try:
                self.writelock = True
                try:
                    while True:
                        items.append(self.write_queue.get_nowait())
                except Empty:
                    pass

                session = Session()

                try:
                    session.add_all(items)
                    session.commit()
                except:
                    session.rollback()
                    raise
                finally:
                    session.close()
            finally:
                self.writelock = False

    def __call__(self, environ, start_response):
        """
        This function have to be called in a worker thread, not the IO thread.
        """
        rargs = environ['wsgiorg.routing_args'][1]
        controller = rargs['controller']

        # Media Transport
        if controller == 'mt':
            name = rargs['name']
            if name in self.mts:
                return self.mts[name](environ, start_response)
            else:
                return not_found(environ, start_response)

        if controller != 'upnp':
            return not_found(environ, start_response)

        try:
            udn = rargs['udn']
            if isInIOThread():
                # TODO: read request body
                return self.devices[udn](environ, start_response)
            else:
                # read request body
                input = environ['wsgi.input']
                environ['upnp.body'] = input.read(self.SOAP_BODY_MAX)
                # call the app in IO thread
                args = [udn, environ, start_response]
                blockingCallFromThread(self.reactor, self._call_handler, args)
                return args[3]
        except Exception, e:
            #print e
            #print 'Unknown access: ' + environ['PATH_INFO']
            return not_found(environ, start_response)

    def wait(self, timeout=None):
        """
        Return the result, or throw the exception if result is a failure.

        It may take an unknown amount of time to return the result, so a
        timeout option is provided. If the given number of seconds pass with
        no result, a TimeoutError will be thrown.

        If a previous call timed out, additional calls to this function will
        still wait for a result and return it if available. If a result was
        returned or raised on one call, additional calls will return/raise the
        same result.
        """
        if threadable.isInIOThread():
            raise RuntimeError("EventualResult.wait() must not be run in the reactor thread.")

        if imp.lock_held():
            # If EventualResult.wait() is run during module import, if the
            # Twisted code that is being run also imports something the result
            # will be a deadlock. Even if that is not an issue it would
            # prevent importing in other threads until the call returns.
            raise RuntimeError("EventualResult.wait() must not be run at module import time.")

        result = self._result(timeout)
        if isinstance(result, Failure):
            result.raiseException()
        return result

 def __init__(self, crawler, update_vars=None, code=None):
     self.crawler = crawler
     self.update_vars = update_vars or (lambda x: None)
     self.item_class = load_object(crawler.settings['DEFAULT_ITEM_CLASS'])
     self.spider = None
     self.inthread = not threadable.isInIOThread()
     self.code = code
     self.vars = {}

 def _queue_action(self, action, event=None):
     if isinstance(action, Action):
         if isInIOThread():
             self.bot.route_response(action, event)
         else:
             reactor.callFromThread(self.bot.route_response, action, event)
     else:
         self.log.error('tried to queue invalid action: {0!r}'.format(action))

 def write(self, *args):
     """Ensure that all writes are serialized regardless if the command is executing in another thread.
     """
     if self.write_buffer is not None and hasattr(self.write_buffer, 'append'):
         self.write_buffer.append(' '.join(map(str, args)))
     deferredMethod = (reactor.callFromThread if not isInIOThread() else defer.maybeDeferred)
     # XXX: HACK: force str for the first param to avoid UnicodeDecodeError happening in Conch
     return deferredMethod(self.terminal.write, str(args[0]), *args[1:])

def get_connection(accept_main_thread=False):
    if not accept_main_thread and isInIOThread() and not _testing:
        raise Exception('The ZODB should not be accessed from the main thread')

    global _connection
    if not hasattr(_connection, 'x'):
        _connection.x = get_db().open()
    return _connection.x

	def sendDataToServer(self, sendType, dataDict):
		dictToSend = {'pktType': sendType}
		dictToSend.update(dataDict)
		pickledData = cPickle.dumps(dictToSend)
		if threadable.isInIOThread():
			self.connProtocol.sendString(pickledData)
		else:
			reactor.callFromThread(self.connProtocol.sendString, pickledData)

        def trans(txn):
            self.assertFalse(threadable.isInIOThread(), "Transactions must not run in main thread")

            yield Transaction(name="TEST1").save()
            yield Transaction(name="TEST2").save()

            barrier.wait()  # wait here to delay commit
            returnValue("return value")

 def _sendMessage(self, msg):
     """ Internally used method to send messages via RCERobotProtocol.
         
         @param msg:         Message which should be sent.
     """
     if isInIOThread():
         self._sendMessageSynced(msg)
     else:
         self._reactor.callFromThread(self._sendMessageSynced, msg)

 def callFromThread(self, f, *args, **kw):
     """See twisted.internet.interfaces.IReactorThreads.callFromThread.
     """
     assert callable(f), "%s is not callable" % f
     if threadable.isInIOThread():
         self.callLater(0, f, *args, **kw)
     else:
         # lists are thread-safe in CPython, but not in Jython
         # this is probably a bug in Jython, but until fixed this code
         # won't work in Jython.
         self.threadCallQueue.append((f, args, kw))
         self.wakeUp()