我们从Python开源项目中，提取了以下47个代码示例，用于说明如何使用twisted.internet.reactor.callInThread()。
def testWakerOverflow(self): self.failure = None waiter = threading.Event() def threadedFunction(): # Hopefully a hundred thousand queued calls is enough to # trigger the error condition for i in xrange(100000): try: reactor.callFromThread(lambda: None) except: self.failure = failure.Failure() break waiter.set() reactor.callInThread(threadedFunction) waiter.wait(120) if not waiter.isSet(): self.fail("Timed out waiting for event") if self.failure is not None: return defer.fail(self.failure)
def sendResult(self, res, utt_idx, session_id): utt_id = "utt-%s-%d" % (session_id, utt_idx) # Update the preview doc doc = self.db[utt_id] #print 'got utt alignment' doc[res["type"] + "_words"] = res["words"] #del doc[res["type"]] if 'duration' in res: doc['duration'] = res['duration'] # having a duration also implies that the wave file is # ready; import to the database reactor.callInThread(self._put_attachment, doc) else: self.db.onchange(None, {"type": "change", "id": utt_id, "doc": doc}) # make sure "start" time is set on utterances self.ensure_start_times(session_id) self.check_pending_audio_commands()
def onchange(self, sender, change_doc): update = False if change_doc.get("doc", {}).get("type") == "command": # Save kaldi-sequence from the text seq = metasentence.MetaSentence(change_doc["doc"].get("text", ""), vocab).get_kaldi_sequence() change_doc["doc"]["_ks"] = seq self._command_seqs[change_doc["id"]] = seq # Set "sender" to None so that all peers get a change update sender = None update = True elif change_doc["type"] == 'delete' and change_doc["id"] in self._command_seqs: del self._command_seqs[change_doc["id"]] update = True elif change_doc.get("doc", {}).get("type") == "audio-command": print 'got new audio command', change_doc['doc'] self._pending_audio_commands.append(change_doc["doc"]) self.subdir_resources['factory'].check_pending_audio_commands() minidb.DBFactory.onchange(self, sender, change_doc) if update: self.create_language_model() reactor.callInThread( self.subdir_resources['factory'].re_run_everything)
def onupload(self, upl): session_id = upl['path'] # Start a session, if it seems to be a valid media file # XXX: this list is arbitrary; use MIME or smth if not session_id.split('.')[-1] in ['mp3', 'mp4', 'm4v', 'wav', 'aac', 'm4a', 'mkv', 'ogg', 'ogv', 'flac']: return self.db.onchange(None, {"type": "change", "id": session_id, "doc": { "_id": session_id, "type": "session", "peer": upl['filename'], "filename": upl['filename'], "s_time": time.time(), }}) reactor.callInThread(self._process_upload, upl, session_id)
def start_packet_out_stream(self): def packet_generator(): while 1: try: packet = self.packet_out_queue.get(block=True, timeout=1.0) except Empty: if self.stopped: return else: yield packet def stream_packets_out(): generator = packet_generator() try: self.local_stub.StreamPacketsOut(generator) except _Rendezvous, e: if e.code() == StatusCode.UNAVAILABLE: os.system("kill -15 {}".format(os.getpid())) reactor.callInThread(stream_packets_out)
def start_packet_in_stream(self): def receive_packet_in_stream(): streaming_rpc_method = self.local_stub.ReceivePacketsIn iterator = streaming_rpc_method(empty_pb2.Empty()) try: for packet_in in iterator: reactor.callFromThread(self.packet_in_queue.put, packet_in) log.debug('enqued-packet-in', packet_in=packet_in, queue_len=len(self.packet_in_queue.pending)) except _Rendezvous, e: if e.code() == StatusCode.UNAVAILABLE: os.system("kill -15 {}".format(os.getpid())) reactor.callInThread(receive_packet_in_stream)
def start_change_event_in_stream(self): def receive_change_events(): streaming_rpc_method = self.local_stub.ReceiveChangeEvents iterator = streaming_rpc_method(empty_pb2.Empty()) try: for event in iterator: reactor.callFromThread(self.change_event_queue.put, event) log.debug('enqued-change-event', change_event=event, queue_len=len(self.change_event_queue.pending)) except _Rendezvous, e: if e.code() == StatusCode.UNAVAILABLE: os.system("kill -15 {}".format(os.getpid())) reactor.callInThread(receive_change_events)
def start(self): if self.running: return log.debug('starting') self.running = True # Start monitoring the vcore grpc channel reactor.callInThread(self.monitor_vcore_grpc_channel) # Start monitoring logical devices and manage agents accordingly reactor.callLater(0, self.monitor_logical_devices) log.info('started') return self
def test_callInThread(self): """ Test callInThread functionality: set a C{threading.Event}, and check that it's not in the main thread. """ def cb(ign): waiter = threading.Event() result = [] def threadedFunc(): result.append(threadable.isInIOThread()) waiter.set() reactor.callInThread(threadedFunc) waiter.wait(120) if not waiter.isSet(): self.fail("Timed out waiting for event.") else: self.assertEqual(result, [False]) return self._waitForThread().addCallback(cb)
def test_callFromThread(self): """ Test callFromThread functionality: from the main thread, and from another thread. """ def cb(ign): firedByReactorThread = defer.Deferred() firedByOtherThread = defer.Deferred() def threadedFunc(): reactor.callFromThread(firedByOtherThread.callback, None) reactor.callInThread(threadedFunc) reactor.callFromThread(firedByReactorThread.callback, None) return defer.DeferredList( [firedByReactorThread, firedByOtherThread], fireOnOneErrback=True) return self._waitForThread().addCallback(cb)
def test_wakerOverflow(self): """ Try to make an overflow on the reactor waker using callFromThread. """ def cb(ign): self.failure = None waiter = threading.Event() def threadedFunction(): # Hopefully a hundred thousand queued calls is enough to # trigger the error condition for i in xrange(100000): try: reactor.callFromThread(lambda: None) except: self.failure = failure.Failure() break waiter.set() reactor.callInThread(threadedFunction) waiter.wait(120) if not waiter.isSet(): self.fail("Timed out waiting for event") if self.failure is not None: return defer.fail(self.failure) return self._waitForThread().addCallback(cb)
def render_POST(self, request): """ Handle a request from the client. """ script_env = { method: api_method(request, method) for method in request.sdata.api.fns } # Make get do auto-formatting for convenience, even though this # breaks if you try to use literal '{}' named arguments # @@@ reconsider whether this is at all a good idea def get_with_formatting(path, *args): return api_method(request, 'get')(path.format(*args)) script_env['get'] = get_with_formatting script_env['re'] = re script_env['dumps'] = dumps script_env['defaultdict'] = defaultdict script_env['OrderedDict'] = OrderedDict buf = [] def dummy_print(*args): if len(args) == 1 and (isinstance(args[0], list) or isinstance(args[0], dict)): buf.append(dumps(args[0], indent=4)) else: buf.append(' '.join(map(str, args))) script_env['print'] = dummy_print def run_script(script): try: exec script in script_env except: exception_info = sys.exc_info() buf.extend(traceback.format_exception(*exception_info)) request.sdata.log('got reply {}'.format(buf)) request.sdata.add_to_push_queue('script', text=dumps(buf)) script = request.args['script'][0] reactor.callInThread(run_script, script)
def deferToThread(f, *args, **kwargs): """Run function in thread and return result as Deferred.""" d = defer.Deferred() from twisted.internet import reactor reactor.callInThread(_putResultInDeferred, d, f, args, kwargs) return d
def callMultipleInThread(tupleList): """Run a list of functions in the same thread. tupleList should be a list of (function, argsList, kwargsDict) tuples. """ from twisted.internet import reactor reactor.callInThread(_runMultiple, tupleList)
def printResult(self): print print print "callFromThread latency:" sum = 0 for t in self.from_times: sum += t print "%f millisecond" % ((sum / self.numRounds) * 1000) print "callInThread latency:" sum = 0 for t in self.in_times: sum += t print "%f millisecond" % ((sum / self.numRounds) * 1000) print print
def testCallFromThread(self): for i in range(self.numRounds): reactor.callInThread(self.tcmf_2, time.time()) self.wait() assert len(self.in_times) == len(self.from_times) assert len(self.in_times) == self.numRounds self.printResult()
def testCallInThread(self): waiter = threading.Event() result = [] def threadedFunc(): result.append(threadable.isInIOThread()) waiter.set() reactor.callInThread(threadedFunc) waiter.wait(120) if not waiter.isSet(): self.fail("Timed out waiting for event.") else: self.assertEquals(result, [False])
def testWakeUp(self): # Make sure other threads can wake up the reactor d = Deferred() def wake(): time.sleep(0.1) # callFromThread will call wakeUp for us reactor.callFromThread(d.callback, None) reactor.callInThread(wake) return d
def AskForMoreBlocks(self): reactor.callInThread(self.DoAskForMoreBlocks)
def deferToThreadInReactor(reactor, f, *args, **kwargs): """ Run function in thread and return result as Deferred. """ d = defer.Deferred() reactor.callInThread(_putResultInDeferred, reactor, d, f, args, kwargs) return d # uses its own reactor for the threaded calls, unlike Twisted's
def notify_listeners(self, packet): """ Send data to all listeners. :param data: the data to send to all listeners. """ for listener in self._listeners: if listener.use_main_thread: blockingCallFromThread(reactor, self._deliver_later, listener, packet) elif reactor.running: reactor.callInThread(self._deliver_later, listener, packet)
def send(self, socket_address, packet): if not self.is_open(): return if reactor.running and socket_address in internet: reactor.callInThread(internet[socket_address].notify_listeners, (self.wan_address, packet)) else: raise AssertionError("Received data from unregistered address %s" % repr(socket_address))
def render_GET(self, req): reactor.callInThread(self._zip, req) return NOT_DONE_YET
def _process(self): try: while self.consumers != [] and self.queue != []: d = self.consumers.pop(0) obj = self.queue.pop(0) dt = threads.deferToThread(self._process_in_thread, d, obj) #reactor.callInThread(self._process_in_thread, d, obj) except Exception, e: print str(e)
def visitLink(self, uid, url): reactor.callInThread(self.spawnModerator, uid, url)
def adminRespond(self, user): reactor.callInThread(self.spawnAdmin, user)
def gotProtocol(self, p): log.info('gotProtocol, connecting {name}', name=self.name) self.protocol = p #def later(): d = p.connect(self.name, keepalive=0, cleanStart=True) d.addCallback(self.subscribe) #d.addCallback(self.prepareToPublish) #reactor.callLater(random.randint(2, 7), later) #reactor.callInThread(later)
def send_packet_stream(self, stub, interval): queue = Queue() @inlineCallbacks def get_next_from_queue(): packet = yield queue.get() returnValue(packet) def packet_generator(): while 1: packet = queue.get(block=True) yield packet def stream(stub): """This is executed on its own thread""" generator = packet_generator() result = stub.SendPackets(generator) print 'Got this after sending packets:', result, type(result) return result reactor.callInThread(stream, stub) while 1: len = queue.qsize() if len < 100: packet = Packet(source=42, content='beefstew') queue.put(packet) yield asleep(interval)
def __init__(self): self.connector = BroadcastConnector() self.last_status = time.time() self.last_nodes = 0 self.issues = 0 self.notifications = defaultdict(list) reactor.callInThread(self.status_loop) reactor.callInThread(self.feedback_loop) reactor.callLater(1, self.watchdog)
def __init__(self): self._monitor_tx = {} self._monitor_lock = threading.Lock() self.last_status = time.time() self.radar_hosts = 0 self.issues = 0 reactor.callInThread(self.status_loop) reactor.callInThread(self.feedback_loop) reactor.callLater(1, self.watchdog)
def callMultipleInThread(tupleList): """ Run a list of functions in the same thread. tupleList should be a list of (function, argsList, kwargsDict) tuples. """ from twisted.internet import reactor reactor.callInThread(_runMultiple, tupleList)
def _testBlockingCallFromThread(self, reactorFunc): """ Utility method to test L{threads.blockingCallFromThread}. """ waiter = threading.Event() results = [] errors = [] def cb1(ign): def threadedFunc(): try: r = threads.blockingCallFromThread(reactor, reactorFunc) except Exception as e: errors.append(e) else: results.append(r) waiter.set() reactor.callInThread(threadedFunc) return threads.deferToThread(waiter.wait, self.getTimeout()) def cb2(ign): if not waiter.isSet(): self.fail("Timed out waiting for event") return results, errors return self._waitForThread().addCallback(cb1).addBoth(cb2)
def puller(self): reactor.callInThread(self._puller)
def interactive_main(args): shell = create_remote_shell(args.conn_info) response = yield shell.create() intro = '\n'.join(response.stdout) winrs_cmd = WinrsCmd(shell) reactor.callInThread(winrs_cmd.cmdloop, intro)
def do_search_concurrently(search): """ Run the search task concurrently, in another thread of the threadpool. :param search: search task to be run concurrently """ logging.debug("Scheduling search to run concurrently.") reactor.callInThread(search.start) SEARCH_TASKS.append(search)
def startService(self): reactor.callInThread(writeForever) Service.startService(self)
def log(sess_id, host, user, data): reactor.callInThread(Logger.background_log, sess_id, host, user, data)
def main(): parser = argparse.ArgumentParser() parser.add_argument("-m", "--mainnet", action="store_true", default=False, help="Use MainNet instead of the default TestNet") parser.add_argument("-p", "--privnet", action="store_true", default=False, help="Use PrivNet instead of the default TestNet") parser.add_argument("-c", "--config", action="store", help="Use a specific config file") parser.add_argument("-t", "--set-default-theme", dest="theme", choices=["dark", "light"], help="Set the default theme to be loaded from the config file. Default: 'dark'") parser.add_argument('--version', action='version', version='neo-python v{version}'.format(version=__version__)) args = parser.parse_args() if args.config and (args.mainnet or args.privnet): print("Cannot use both --config and --mainnet/--privnet arguments, please use only one.") exit(1) if args.mainnet and args.privnet: print("Cannot use both --mainnet and --privnet arguments") exit(1) # Setup depending on command line arguments. By default, the testnet settings are already loaded. if args.config: settings.setup(args.config) elif args.mainnet: settings.setup_mainnet() elif args.privnet: settings.setup_privnet() if args.theme: preferences.set_theme(args.theme) # Instantiate the blockchain and subscribe to notifications blockchain = LevelDBBlockchain(settings.LEVELDB_PATH) Blockchain.RegisterBlockchain(blockchain) # Start the prompt interface cli = PromptInterface() # Run reactor.suggestThreadPoolSize(15) reactor.callInThread(cli.run) NodeLeader.Instance().Start() reactor.run()
def reconcile(self, device): self.log.info('reconciling-asfvolt16-starts',device=device) if not device.host_and_port: device.oper_status = OperStatus.FAILED device.reason = 'No host_and_port field provided' self.adapter_agent.update_device(device) return try: # Establishing connection towards OLT self.bal.connect_olt(device.host_and_port, self.device_id,is_init=False) device.connect_status = ConnectStatus.REACHABLE device.oper_status = OperStatus.ACTIVE self.adapter_agent.update_device(device) reactor.callInThread(self.bal.get_indication_info, self.device_id) except Exception as e: self.log.exception('device-unreachable', error=e) device.connect_status = ConnectStatus.UNREACHABLE device.oper_status = OperStatus.UNKNOWN self.adapter_agent.update_device(device) return if self.is_heartbeat_started == 0: self.log.info('heart-beat-is-not-yet-started-starting-now') self.start_heartbeat() # Now set the initial PM configuration for this device self.pm_metrics=Asfvolt16OltPmMetrics(device) pm_config = self.pm_metrics.make_proto() self.log.info("initial-pm-config", pm_config=pm_config) self.adapter_agent.update_device_pm_config(pm_config,init=True) # Apply the PM configuration self.update_pm_config(device, pm_config) # Request PM counters from OLT device. self._handle_pm_counter_req_towards_device(device) # Set the logical device id device = self.adapter_agent.get_device(device.id) if device.parent_id: self.logical_device_id = device.parent_id self.log.info("reconcile-logical-device") self.adapter_agent.reconcile_logical_device(device.parent_id) else: self.log.info('no-logical-device-set') # Reconcile child devices self.log.info("reconcile-all-child-devices") self.adapter_agent.reconcile_child_devices(device.id) self.log.info('reconciling-asfvolt16-device-ends',device=device)
