should be invoked from reactor thread
777559
T unsafeGetIfOpened() {

        if (innerObject != null && innerObject.getState() == IOObject.IOObjectState.OPENED) {
            return innerObject;
        }
        return null;
    }
942703
def check_timer
            @reactor.update_time

            existing = @next
            schedule = @scheduled.first
            @next = schedule.nil? ? nil : schedule.next_scheduled

            if existing != @next
                # lazy load the timer
                if @timer.nil?
                    new_timer
                else
                    @timer.stop
                end

                if not @next.nil?
                    in_time = @next - @reactor.now

                    # Ensure there are never negative start times
                    if in_time > 3
                        @timer.start(in_time)
                    else
                        # Effectively next tick
                        @timer.start(0)
                    end
                end
            end
        end
133061
def callMultipleInGreenlet(tupleList):
    """"""
    from twisted.internet import reactor
    assert reactor.greenlet == getcurrent(), "must invoke this in the reactor greenlet"
    reactor.callInGreenlet(_runMultiple, tupleList)
1325119
def run_in_reactor(self, function):
        
        result = self._run_in_reactor(function)
        # Backwards compatibility; use __wrapped__ instead.
        try:
            result.wrapped_function = function
        except AttributeError:
            pass
        return result
693744
def _startReapingProcesses(self):
        
        lc = LoopingCall(self._reapAllProcesses)
        lc.clock = self._reactor
        lc.start(0.1, False)
721028
def main(reactor):
    """"""
    parser = argparse.ArgumentParser(
        description='Fetch a URI or series of URIs and print a title '
                    'or summary for each.',
        epilog='If no URIs are passed on the command line, they are '
               'read from standard input, one per line.')
    parser.add_argument(
        'uris', metavar='URI', nargs='*', help='URI to fetch')
    parser.add_argument(
        '-H', '--hostname-tag', action='store_true',
        help='prefix titles with a hostname tag')
    args = parser.parse_args()
    uris = args.uris or imap(lambda x: x.strip(), sys.stdin)
    finished = DeferredList([
        fetch_title(uri.decode(locale.getpreferredencoding()),
                    hostname_tag=args.hostname_tag, friendly_errors=True)
        for uri in uris])
    finished.addCallback(print_and_exit)
    return finished
158488
def threadsafe_call(self, fn, *args, **kwargs):
        """"""
        def handler():
            try:
                fn(*args, **kwargs)
            except Exception:
                warn("error caught while excecuting async callback\n%s\n",
                     format_exc())

        def greenlet_wrapper():
            gr = greenlet.greenlet(handler)
            gr.switch()

        self._async_session.threadsafe_call(greenlet_wrapper)
26511
def process_forever(self, timeout=0.2):
        
        # This loop should specifically *not* be mutex-locked.
        # Otherwise no other thread would ever be able to change
        # the shared state of a Reactor object running this function.
        log.debug("process_forever(timeout=%s)", timeout)
        one = functools.partial(self.process_once, timeout=timeout)
        consume(infinite_call(one))
224170
def run(self):
        

        task = None
        monitor_task = MonitorTask(
            notification_endpoint=self._handle_message)

        while True:
            if self._terminate:
                self._logger.info("scsgate.Reactor exiting")
                self._connection.close()
                break
            try:
                task = self._request_queue.get_nowait()
                self._logger.debug("scsgate.Reactor: got task {}".format(task))
            except queue.Empty:
                task = monitor_task

            try:
                task.execute(connection=self._connection)
            except ExecutionError as err:
                self._logger.error(err)
991474
def run(self):
        ""
        self.addServices()
        self.catalogServers(self.hendrix)
        action = self.action
        fd = self.options['fd']

        if action.startswith('start'):
            chalk.blue(self._listening_message())
            getattr(self, action)(fd)

            ###########################
            # annnnd run the reactor! #
            ###########################
            try:
                self.reactor.run()
            finally:
                shutil.rmtree(PID_DIR, ignore_errors=True)  # cleanup tmp PID dir

        elif action == 'restart':
            getattr(self, action)(fd=fd)
        else:
            getattr(self, action)()
431465
def run_twisted(self, args):
        
        server = self

        from twisted.internet import reactor
        from twisted.web import resource
        from twisted.web.static import File
        from twisted.web.server import Site
        from autobahn.twisted.websocket import (WebSocketServerFactory,
                                                WebSocketServerProtocol)
        from autobahn.twisted.resource import WebSocketResource

        class DevWebSocketHandler(WebSocketServerProtocol):
            def onConnect(self, request):
                super(DevWebSocketHandler, self).onConnect(request)
                server.on_open(self)

            def onMessage(self, payload, isBinary):
                server.on_message(self, payload)

            def onClose(self, wasClean, code, reason):
                super(DevWebSocketHandler,self).onClose(wasClean, code, reason)
                server.on_close(self)

            def write_message(self, message, binary=False):
                self.sendMessage(message, binary)

        #: Set the call later method
        server.call_later = reactor.callLater
        server.add_callback = reactor.callFromThread

        factory = WebSocketServerFactory(u"ws://0.0.0.0:{}".format(self.port))
        factory.protocol = DevWebSocketHandler

        class MainHandler(resource.Resource):
            def render_GET(self, req):
                return str(server.index_page)

        root = resource.Resource()
        root.putChild("", MainHandler())
        root.putChild("dev", WebSocketResource(factory))
        reactor.listenTCP(self.port, Site(root))
        print("Twisted Dev server started on {}".format(self.port))
        reactor.run()
80019
@Ready
    public void onReady(/* In you don't want injection AtmosphereResource r */) {
        logger.info("Browser {} connected", r.uuid());
        logger.info("BroadcasterFactory used {}", factory.getClass().getName());
        logger.info("Broadcaster injected {}", broadcaster.getID());

    }
1112105
def create_threads(self):
        """"""

        # Create thread for neutron notifications.
        neutron_thrd = utils.EventProcessingThread('Neutron_Event',
                                                   self.neutron_event,
                                                   'event_handler',
                                                   self._excpq)
        self.dfa_threads.append(neutron_thrd)

        # Create thread for processing notification events.
        qp_thrd = utils.EventProcessingThread('Event_Queue', self,
                                              'process_queue', self._excpq)
        self.dfa_threads.append(qp_thrd)

        # Create thread for keystone notifications.
        keys_thrd = utils.EventProcessingThread('Keystone_Event',
                                                self.keystone_event,
                                                'event_handler', self._excpq)
        self.dfa_threads.append(keys_thrd)

        # Create thread to process RPC calls.
        hb_thrd = utils.EventProcessingThread('RPC_Server', self, 'start_rpc',
                                              self._excpq)
        self.dfa_threads.append(hb_thrd)

        # Create thread to listen to dcnm network events.
        if self.dcnm_event is not None:
            dcnmL_thrd = utils.EventProcessingThread('DcnmListener',
                                                     self.dcnm_event,
                                                     'process_amqp_msgs',
                                                     self._excpq)
            self.dfa_threads.append(dcnmL_thrd)

        # Create periodic task to process failure cases in create/delete
        # networks and projects.
        fr_thrd = utils.PeriodicTask(interval=constants.FAIL_REC_INTERVAL,
                                     func=self.add_events,
                                     event_queue=self.pqueue,
                                     priority=self.PRI_LOW_START + 10,
                                     excq=self._excpq)

        # Start all the threads.
        for t in self.dfa_threads:
            t.start()

        # Run the periodic tasks.
        fr_thrd.run()
1257328
void invokeAndWait(final Runnable r) {
        final CountDownLatch latch = new CountDownLatch(1);
        queue.postRunnable(() -> {
            try {
                r.run();
            } finally {
                latch.countDown();
            }
        });
        try {
            latch.await();
        } catch (InterruptedException e) { }
    }
493296
def run(self, low):
        '''
        
        '''
        self.populate_client_cache(low)
        try:
            l_fun = getattr(self, low['state'])
        except AttributeError:
            log.error(
                'ReactWrap is missing a wrapper function for \'%s\'',
                low['state']
            )

        try:
            wrap_call = salt.utils.args.format_call(l_fun, low)
            args = wrap_call.get('args', ())
            kwargs = wrap_call.get('kwargs', {})
            # TODO: Setting user doesn't seem to work for actual remote pubs
            if low['state'] in ('runner', 'wheel'):
                # Update called function's low data with event user to
                # segregate events fired by reactor and avoid reaction loops
                kwargs['__user__'] = self.event_user
                # Replace ``state`` kwarg which comes from high data compiler.
                # It breaks some runner functions and seems unnecessary.
                kwargs['__state__'] = kwargs.pop('state')
                # NOTE: if any additional keys are added here, they will also
                # need to be added to filter_kwargs()

            if 'args' in kwargs:
                # New configuration
                reactor_args = kwargs.pop('args')
                for item in ('arg', 'kwarg'):
                    if item in low:
                        log.warning(
                            'Reactor \'%s\' is ignoring \'%s\' param %s due to '
                            'presence of \'args\' param. Check the Reactor System '
                            'documentation for the correct argument format.',
                            low['__id__'], item, low[item]
                        )
                if low['state'] == 'caller' \
                        and isinstance(reactor_args, list) \
                        and not salt.utils.data.is_dictlist(reactor_args):
                    # Legacy 'caller' reactors were already using the 'args'
                    # param, but only supported a list of positional arguments.
                    # If low['args'] is a list but is *not* a dictlist, then
                    # this is actually using the legacy configuration. So, put
                    # the reactor args into kwarg['arg'] so that the wrapper
                    # interprets them as positional args.
                    kwargs['arg'] = reactor_args
                    kwargs['kwarg'] = {}
                else:
                    kwargs['arg'] = ()
                    kwargs['kwarg'] = reactor_args
                if not isinstance(kwargs['kwarg'], dict):
                    kwargs['kwarg'] = salt.utils.data.repack_dictlist(kwargs['kwarg'])
                    if not kwargs['kwarg']:
                        log.error(
                            'Reactor \'%s\' failed to execute %s \'%s\': '
                            'Incorrect argument format, check the Reactor System '
                            'documentation for the correct format.',
                            low['__id__'], low['state'], low['fun']
                        )
                        return
            else:
                # Legacy configuration
                react_call = {}
                if low['state'] in ('runner', 'wheel'):
                    if 'arg' not in kwargs or 'kwarg' not in kwargs:
                        # Runner/wheel execute on the master, so we can use
                        # format_call to get the functions args/kwargs
                        react_fun = self.client_cache[low['state']].functions.get(low['fun'])
                        if react_fun is None:
                            log.error(
                                'Reactor \'%s\' failed to execute %s \'%s\': '
                                'function not available',
                                low['__id__'], low['state'], low['fun']
                            )
                            return

                        react_call = salt.utils.args.format_call(
                            react_fun,
                            low,
                            expected_extra_kws=REACTOR_INTERNAL_KEYWORDS
                        )

                if 'arg' not in kwargs:
                    kwargs['arg'] = react_call.get('args', ())
                if 'kwarg' not in kwargs:
                    kwargs['kwarg'] = react_call.get('kwargs', {})

            # Execute the wrapper with the proper args/kwargs. kwargs['arg']
            # and kwargs['kwarg'] contain the positional and keyword arguments
            # that will be passed to the client interface to execute the
            # desired runner/wheel/remote-exec/etc. function.
            ret = l_fun(*args, **kwargs)

            if ret is False:
                log.error('Reactor \'%s\' failed  to execute %s \'%s\': '
                            'TaskPool queue is full!'
                            ' Consider tuning reactor_worker_threads and/or'
                            ' reactor_worker_hwm',
                            low['__id__'], low['state'], low['fun']
                )

        except SystemExit:
            log.warning(
                'Reactor \'%s\' attempted to exit. Ignored.', low['__id__']
            )
        except Exception:
            log.error(
                'Reactor \'%s\' failed to execute %s \'%s\'',
                low['__id__'], low['state'], low['fun'], exc_info=True
            )
1169449
def run(self):
        
        while not self.stopEvent.isSet():
            try:
                # no need to monitor if no observers
                if 0 < self.observable.countObservers():
                    currentReaders = self.readerProc()
                    addedReaders = []
                    removedReaders = []

                    if currentReaders != self.readers:
                        for reader in currentReaders:
                            if reader not in self.readers:
                                addedReaders.append(reader)
                        for reader in self.readers:
                            if reader not in currentReaders:
                                removedReaders.append(reader)

                        if addedReaders or removedReaders:
                            # Notify observers
                            self.readers = []
                            for r in currentReaders:
                                self.readers.append(r)
                            self.observable.setChanged()
                            self.observable.notifyObservers((addedReaders,
                                                            removedReaders))

                # wait every second on stopEvent
                self.stopEvent.wait(self.period)

            except Exception:
                # FIXME Tighten the exceptions caught by this block
                traceback.print_exc()
                # Most likely raised during interpreter shutdown due
                # to unclean exit which failed to remove all observers.
                # To solve this, we set the stop event and pass the
                # exception to let the thread finish gracefully.
                self.stopEvent.set()
711924
def populate_client_cache(self, low):
        '''
        
        '''
        reaction_type = low['state']
        if reaction_type not in self.client_cache:
            log.debug('Reactor is populating %s client cache', reaction_type)
            if reaction_type in ('runner', 'wheel'):
                # Reaction types that run locally on the master want the full
                # opts passed.
                self.client_cache[reaction_type] = \
                    self.reaction_class[reaction_type](self.opts)
                # The len() function will cause the module functions to load if
                # they aren't already loaded. We want to load them so that the
                # spawned threads don't need to load them. Loading in the
                # spawned threads creates race conditions such as sometimes not
                # finding the required function because another thread is in
                # the middle of loading the functions.
                len(self.client_cache[reaction_type].functions)
            else:
                # Reactions which use remote pubs only need the conf file when
                # instantiating a client instance.
                self.client_cache[reaction_type] = \
                    self.reaction_class[reaction_type](self.opts['conf_file'])