我们从Python开源项目中，提取了以下47个代码示例，用于说明如何使用gevent.Greenlet()。
def test_greenlet(self): """??????Greenlet????""" class MyGreenlet(gevent.Greenlet): def __init__(self, message, n): super(MyGreenlet, self).__init__() self.message = message self.n = n def _run(self): print(self.message) gevent.sleep(self.n) g1 = MyGreenlet("Hi there111!", 1) g1.start() g2 = MyGreenlet("Hi there222!", 2) g2.start() gevent.joinall([g1, g2]) # def test_shutdown(self): # def run_forever(): # _log.info('run_forever start..') # gevent.sleep(1000) # gevent.signal(signal.SIGQUIT, gevent.kill) # thread = gevent.spawn(run_forever) # thread.join()
def __init__(self, *args, **kwargs): # get the current Context if available current_g = gevent.getcurrent() ctx = getattr(current_g, CONTEXT_ATTR, None) # create the Greenlet as usual super(TracedGreenlet, self).__init__(*args, **kwargs) # the context is always available made exception of the main greenlet if ctx: # create a new context that inherits the current active span # TODO: a better API for Context, should get the tuple at once new_ctx = Context( trace_id=ctx._parent_trace_id, span_id=ctx._parent_span_id, sampled=ctx._sampled, ) new_ctx._current_span = ctx._current_span setattr(self, CONTEXT_ATTR, new_ctx)
def wrapper(o_fn): if timeout: f = functools.partial(gevent.with_timeout, timeout, o_fn, timeout_value=sync_ret_val) else: f = o_fn @functools.wraps(o_fn) def wrapped(*args, **kwargs): g = gevent.Greenlet(f, *args, **kwargs) g.link_exception(self._on_error) g.link(lambda v: self._running_greenlets.discard(g)) self._running_greenlets.add(g) g.start() return sync_ret_val return wrapped
def start(self, graph_id, done_callback): """ Execute a graph. """ self.logger.debug('Graph {}: Starting execution'.format(graph_id)) graph = self.get_graph(graph_id) network = Network(graph) executor = gevent.Greenlet(network.go) # FIXME: should we delete the executor from self._executors on finish? # this has an impact on the result returned from get_status(). Leaving # it means that after completion it will be started:True, running:False # until stop() is triggered, at which point it will be started:False, # running:False executor.link(lambda g: done_callback()) self._executors[graph_id] = (executor, network) executor.start() # if executor.is_running(): # raise ValueError('Graph {} is already started'.format(graph_id))
def _interface_poll_loop(self): """Greenlet: Polls host endpoints for changes to their IP addresses. Sends updates to the EndpointManager via the _on_iface_ips_update() message. If polling is disabled, then it reads the interfaces once and then stops. """ known_interfaces = {} while True: known_interfaces = self._poll_interfaces(known_interfaces) if self.config.HOST_IF_POLL_INTERVAL_SECS <= 0: _log.info("Host interface polling disabled, stopping after " "initial read. Further changes to host endpoint " "IPs will be ignored.") break gevent.sleep(self.config.HOST_IF_POLL_INTERVAL_SECS)
def cleanup_tasks(): tasks = [ running_task for running_task in gc.get_objects() if isinstance(running_task, gevent.Greenlet) ] gevent.killall(tasks) gevent.hub.reinit()
def __init__(self): self.inbox = Queue() gevent.Greenlet.__init__(self)
def _start_watcher(self): """ Start watcher coroutine for watch status of etcd. :return: :rtype: gevent.Greenlet """ co = gevent.spawn(self._watcher_handler) log.info('watcher_handler(%s) started.', co) return co
def _start_heartbeat(self): """ Start heartbeat coroutine for watch status of etcd. :return: :rtype: gevent.Greenlet """ co = gevent.spawn(self._heartbeat_handler) log.info('watcher_handler(%s) started.', co) return co #### coroutine handler ####
def patch(): """ Patch the gevent module so that all references to the internal ``Greenlet`` class points to the ``DatadogGreenlet`` class. This action ensures that if a user extends the ``Greenlet`` class, the ``TracedGreenlet`` is used as a parent class. """ _replace(TracedGreenlet) ddtrace.tracer.configure(context_provider=GeventContextProvider())
def unpatch(): """ Restore the original ``Greenlet``. This function must be invoked before executing application code, otherwise the ``DatadogGreenlet`` class may be used during initialization. """ _replace(__Greenlet) ddtrace.tracer.configure(context_provider=DefaultContextProvider())
def _replace(g_class): """ Utility function that replace the gevent Greenlet class with the given one. """ # replace the original Greenlet class with the new one gevent.greenlet.Greenlet = g_class # replace gevent shortcuts gevent.Greenlet = gevent.greenlet.Greenlet gevent.spawn = gevent.greenlet.Greenlet.spawn gevent.spawn_later = gevent.greenlet.Greenlet.spawn_later
def _on_child_hook(): # This is called in the hub greenlet. To let the function # do more useful work, like use blocking functions, # we run it in a new greenlet; see gevent.hub.signal if callable(_child_handler): # None is a valid value for the frame argument from gevent import Greenlet greenlet = Greenlet(_child_handler, _signal.SIGCHLD, None) greenlet.switch()
def _spawn_greenlet(fn, *args, **kwargs): from gevent import Greenlet g = Greenlet(fn, *args, **kwargs) g.start() return g
def start(self): g = gevent.Greenlet(self._start) g.start()
def activate_inputs(self): """Main activation method: launches a greenlet waiting for messages forever""" inputs_listener = gevent.Greenlet(self.receive_loop) inputs_listener.start() return inputs_listener
def start(self): # print('*** Registering signal handlers ***') # self._register_signal_handlers() self.console.print_('*** Starting note scheduler ***') self.note_scheduler.start() self._target_time = self._initial_time = time.time() g = gevent.Greenlet(self._gevent_loop) g.start() g.join()
def __init__(self, url, protocols=None, extensions=None, ssl_options=None, headers=None): """ WebSocket client that executes the :meth:`run() <ws4py.websocket.WebSocket.run>` into a gevent greenlet. .. code-block:: python ws = WebSocketClient('ws://localhost:9000/echo', protocols=['http-only', 'chat']) ws.connect() ws.send("Hello world") def incoming(): while True: m = ws.receive() if m is not None: print str(m) else: break def outgoing(): for i in range(0, 40, 5): ws.send("*" * i) greenlets = [ gevent.spawn(incoming), gevent.spawn(outgoing), ] gevent.joinall(greenlets) """ WebSocketBaseClient.__init__(self, url, protocols, extensions, ssl_options=ssl_options, headers=headers) self._th = Greenlet(self.run) self.messages = Queue() """ Queue that will hold received messages. """
def __init__(self, url, protocols=None, extensions=None, heartbeat_freq=None, ssl_options=None, headers=None): """ WebSocket client that executes the :meth:`run() <ws4py.websocket.WebSocket.run>` into a gevent greenlet. .. code-block:: python ws = WebSocketClient('ws://localhost:9000/echo', protocols=['http-only', 'chat']) ws.connect() ws.send("Hello world") def incoming(): while True: m = ws.receive() if m is not None: print str(m) else: break def outgoing(): for i in range(0, 40, 5): ws.send("*" * i) greenlets = [ gevent.spawn(incoming), gevent.spawn(outgoing), ] gevent.joinall(greenlets) """ WebSocketBaseClient.__init__(self, url, protocols, extensions, heartbeat_freq, ssl_options=ssl_options, headers=headers) self._th = Greenlet(self.run) self.messages = Queue() """ Queue that will hold received messages. """
def start_publish_loop(self, publish_channel, publish_message): self.publish_channel = publish_channel self.publish_message = publish_message self.loop_greenlet = gevent.Greenlet(self._publish_loop) self.greenlets.append(self.loop_greenlet) self.loop_greenlet.start()
def spawn(f): gevent.Greenlet(f).start()
def tee_and_handle(f, msgs): queue = Queue() # unbounded buffer def _run(): for msg in msgs: print(COLOR_RECV, 'Received:', COLOR_ENDC, msg.command) if msg.command == b'ping': send(f, msg_pong(nonce=msg.nonce)) queue.put(msg) t = gevent.Greenlet(_run) t.start() while True: yield(queue.get())
def __init__( self, cloudDest, cbReceiveMessage, orgId, installerId, platform, architecture, sensorId = None, enrollmentToken = None, cbDebugLog = None, cbEnrollment = None ): gevent.Greenlet.__init__( self ) self._cbDebugLog = cbDebugLog self._cbReceiveMessage = cbReceiveMessage self._cbEnrollment = cbEnrollment try: self._destServer, self._destPort = cloudDest.split( ':' ) except: self._destServer = cloudDest self._destPort = 443 self._oid = uuid.UUID( str( orgId ) ) self._iid = uuid.UUID( str( installerId ) ) self._sid = sensorId self._arch = architecture self._plat = platform if self._sid is not None: self._sid = uuid.UUID( str( self._sid ) ) self._enrollmentToken = enrollmentToken self._socket = None self._threads = gevent.pool.Group() self._stopEvent = gevent.event.Event() self._lock = Semaphore( 1 ) self._connectedEvent = gevent.event.Event() self._r = rpcm( isHumanReadable = True, isDebug = self._log ) self._r.loadSymbols( Symbols.lookups ) self._hcpModules = [] self._hbsProfileHash = ( "\x00" * 32 )
def __init__(self): self.inbox = Queue() Greenlet.__init__(self)
def _on_error(self, e): """ Called when an error happens by something this module called. """ if isinstance(e, gevent.Greenlet): e = e.exception self.client.on_module_error(self, e) logger.exception("Exception raised %r", e)
def spawn(self, f, *args, **kwargs): """ Spawns a greenlet and does some book-keeping to make sure the greenlet is killed when the module is unloaded. """ g = gevent.Greenlet(f, *args, **kwargs) g.link_exception(self._on_error) g.link(lambda v: self._running_greenlets.discard(g)) self._running_greenlets.add(g) g.start() return g
def spawn_after(self, delay, f, *args, **kwargs): """ Spawns a greenlet that will start after delay seconds. Otherwise, same as Module.spawn """ g = gevent.Greenlet(f, *args, **kwargs) g.link_exception(self._on_error) g.link(lambda v: self._running_greenlets.discard(g)) self._running_greenlets.add(g) g.start_later(delay) return g
def start(self, right_away=True): if self._greenlet: raise RuntimeError("Periodic already started.") self._greenlet = Greenlet(self._run) self._greenlet.link(self._discard_greenlet) if right_away: self._greenlet.start() else: self._greenlet.start_later(self.interval)
def __init__(self): self.logger = logging.getLogger('%s.%s' % (self.__class__.__module__, self.__class__.__name__)) self._component_types = {} # Component metadata, keyed by component name # type: Dict[str, Graph] self._graphs = {} # Graph instances, keyed by graph ID # type: Dict[str, Tuple[Greenlet, Network]] self._executors = {} # GraphExecutor instances, keyed by graph ID self.logger.debug('Initialized runtime!')
def _on_worker_died(self, watch_greenlet): """ Greenlet: spawned by the gevent Hub if our worker thread dies. """ _log.critical("Worker greenlet died: %s; exiting.", watch_greenlet) sys.exit(1)
def __init__(self, qualifier=None): self._event_queue = collections.deque() # Set to True when the main loop is actively processing the input # queue or has been scheduled to do so. Set to False when the loop # runs out of work and switches to the Hub to wait for more. self._scheduled = True # (Monotonic time) timestamp of last schedule. self._last_scheduled = None # Cache the gevent Hub and main loop. self._gevent_hub = gevent.get_hub() self._gevent_loop = self._gevent_hub.loop self.greenlet = gevent.Greenlet(self._loop) self._op_count = 0 self._current_msg = None self.started = False # Message being processed; purely for logging. self.msg_id = None # Logging parameters self.qualifier = qualifier if qualifier: self.name = "%s(%s)" % (self.__class__.__name__, qualifier) else: self.name = self.__class__.__name__ # Can't use str(self) yet, it might not be ready until subclass # constructed. _log.info("%s created.", self.name)
def __init__(self, config, hosts_ipset): super(EtcdAPI, self).__init__(config.ETCD_ADDRS, etcd_scheme=config.ETCD_SCHEME, etcd_key=config.ETCD_KEY_FILE, etcd_cert=config.ETCD_CERT_FILE, etcd_ca=config.ETCD_CA_FILE) self._config = config # Timestamp storing when the EtcdAPI started. This info is needed # in order to report uptime to etcd. self._start_time = monotonic_time() # Create an Actor to report per-endpoint status into etcd. We defer # startup of this and our other workers until we get started. self.status_reporter = EtcdStatusReporter(config) # Create the main etcd-watching greenlet. self._watcher = _FelixEtcdWatcher(config, self, self.status_reporter, hosts_ipset) self._watcher.link(self._on_worker_died) # Create a greenlet to trigger periodic resyncs. self._resync_greenlet = gevent.Greenlet(self._periodically_resync) self._resync_greenlet.link_exception(self._on_worker_died) # Create a greenlet to report felix's liveness into etcd. self.done_first_status_report = False self._status_reporting_greenlet = gevent.Greenlet( self._periodically_report_status ) self._status_reporting_greenlet.link_exception(self._on_worker_died) self.status_reporter.greenlet.link(self._on_worker_died)
def _periodically_report_status(self): """ Greenlet: periodically writes Felix's status into etcd. :return: Does not return, unless reporting disabled. """ _log.info("Started status reporting thread. Waiting for config.") self._watcher.configured.wait() ttl = self._config.REPORTING_TTL_SECS interval = self._config.REPORTING_INTERVAL_SECS _log.debug("Reporting interval: %s, TTL: %s", interval, ttl) if interval == 0: _log.info("Interval is 0, status reporting disabled.") return while True: try: self._update_felix_status(ttl) except EtcdException as e: _log.warning("Error when trying to check into etcd (%r), " "retrying after %s seconds.", e, RETRY_DELAY) self.reconnect() gevent.sleep(RETRY_DELAY) else: # Jitter by 10% of interval. jitter = random.random() * 0.1 * interval sleep_time = interval + jitter gevent.sleep(sleep_time)
def _on_worker_died(self, watch_greenlet): """ Greenlet: spawned by the gevent Hub if the etcd watch loop ever stops, kills the process. """ _log.critical("Worker greenlet died: %s; exiting.", watch_greenlet) sys.exit(1)
def __init__(self, config, etcd_api, status_reporter, hosts_ipset): super(_FelixEtcdWatcher, self).__init__() self._config = config self._etcd_api = etcd_api self._status_reporter = status_reporter self.hosts_ipset = hosts_ipset # Whether we've been in sync with etcd at some point. self._been_in_sync = False # Keep track of the config loaded from etcd so we can spot if it # changes. self.last_global_config = None self.last_host_config = None self.my_config_dir = dir_for_per_host_config(self._config.HOSTNAME) # Events triggered by the EtcdAPI Actor to tell us to load the config # and start polling. These are one-way flags. self.load_config = Event() self.begin_polling = Event() # Event that we trigger once the config is loaded. self.configured = Event() # Polling state initialized at poll start time. self.splitter = None # Next-hop IP addresses of our hosts, if populated in etcd. self.ipv4_by_hostname = {} # Forces a resync after the current poll if set. Safe to set from # another thread. Automatically reset to False after the resync is # triggered. self.resync_requested = False self.dispatcher = PathDispatcher() # The Popen object for the driver. self._driver_process = None # Stats. self.read_count = 0 self.msgs_processed = 0 self.last_rate_log_time = monotonic_time() # Register for events when values change. self._register_paths() self._usage_report_greenlet = gevent.Greenlet( self._periodically_usage_report )
def _periodically_usage_report(self): """ Greenlet: periodically report the cluster existence to projectcalico.org. Period is about once per day. :return: Does not return, unless USAGE_REPORT disabled. """ interval = 86400 # Once every 24 hours minus 12 minute jitter jitter = random.random() * 0.01 * interval try: calico_version = str(pkg_resources.require("calico")[0].version) except ResolutionError: calico_version = "NA" _log.info("Started usage report thread. Usage report interval: %s, pre-jitter: %s", interval, jitter) # Pre-Jitter the reporting thread start by 1% of interval (about 12 minutes) # Jitter prevents thundering herd for large clusters when the cluster first starts # Do pre-jitter only for clusters greater than 25. felix_count = self.estimated_host_count() if (felix_count >= 25): gevent.sleep(jitter) while True: config = self._config felix_count = self.estimated_host_count() cluster_type = "NA" if self._config.USAGE_REPORT: _log.info("usage report is enabled") report_usage_and_get_warnings(calico_version, config.HOSTNAME, config.CLUSTER_GUID, felix_count, cluster_type) # Jitter by 10% of interval (about 120 minutes) jitter = random.random() * 0.1 * interval sleep_time = interval - jitter _log.info("Usage report interval: %s, sleep-time: %s", interval, sleep_time) gevent.sleep(sleep_time)
def __init__(self, *args, **kwargs): self._inbox = Queue() self._running = True Greenlet.__init__(self) self.start() self._args = args self._kwargs = kwargs
def run_parallel(self, hosts, cmd): codes = {"total": 0, "error": 0, "success": 0} def worker(host, cmd): p = Popen(self.get_parallel_ssh_options(host, cmd), stdout=PIPE, stderr=PIPE) while True: outs, _, _ = select([p.stdout, p.stderr], [], []) if p.stdout in outs: outline = p.stdout.readline() else: outline = "" if p.stderr in outs: errline = p.stderr.readline() else: errline = "" if outline == "" and errline == "" and p.poll() is not None: break if outline != "": print("%s: %s" % (colored(host, "blue", attrs=["bold"]), outline.strip())) if errline != "": print("%s: %s" % (colored(host, "blue", attrs=["bold"]), colored(errline.strip(), "red"))) if p.poll() == 0: codes["success"] += 1 else: codes["error"] += 1 codes["total"] += 1 pool = Pool(self.ssh_threads) for host in hosts: pool.start(Greenlet(worker, host, cmd)) pool.join() self.print_exec_results(codes)
def ping_parallel(self, hosts, pc): """ping:\n pings host (using shell cmd)""" codes = {"total": 0, "error": 0, "success": 0} def worker(host): if pc == 0: args = ["ping", host] else: args = ["ping", "-c", str(pc), host] p = Popen(args, stdout=PIPE, stderr=PIPE) while True: outs, _, _ = select([p.stdout, p.stderr], [], []) if p.stdout in outs: outline = p.stdout.readline() else: outline = "" if p.stderr in outs: errline = p.stderr.readline() else: errline = "" if outline == "" and errline == "" and p.poll() is not None: break if outline != "": print("%s: %s" % (colored(host, "blue", attrs=["bold"]), outline.strip())) if errline != "": print("%s: %s" % (colored(host, "blue", attrs=["bold"]), colored(errline.strip(), "red"))) if p.poll() == 0: codes["success"] += 1 else: codes["error"] += 1 codes["total"] += 1 pool = Pool(self.ssh_threads) for host in hosts: pool.start(Greenlet(worker, host)) pool.join() self.print_exec_results(codes)
def defer(f=None, predicate=None): """ Schedule a function to run in a cpu_bound thread, returns a AsyncFuture Optional predicate parameter to determine if the function should be dispatched """ if f is None: def p_wrap(f): return defer(f, predicate) return p_wrap else: def f_wrap(f, *args, **kwargs): if CPUThread._thread is None: CPUThread._thread = CPUThread() return CPUThread._thread.apply(f, args, kwargs) def wrapper(*args, **kwargs): a = AsyncFuture(None, None) # TODO: unit test this if (predicate is not None and not predicate) or utils.in_cpubound_thread(): v = f(*args, **kwargs) a._value = v else: g = Greenlet(f_wrap, f, *args, **kwargs) g.start() a._future = g return a return wrapper
def __init__(self, config, ip_type, iptables_updater, workload_disp_chains, host_disp_chains, rules_manager, fip_manager, status_reporter): super(EndpointManager, self).__init__(qualifier=ip_type) # Configuration and version to use self.config = config self.ip_type = ip_type self.ip_version = futils.IP_TYPE_TO_VERSION[ip_type] # Peers/utility classes. self.iptables_updater = iptables_updater self.workload_disp_chains = workload_disp_chains self.host_disp_chains = host_disp_chains self.rules_mgr = rules_manager self.status_reporter = status_reporter self.fip_manager = fip_manager # All endpoint dicts that are on this host. self.endpoints_by_id = {} # Dict that maps from interface name ("tap1234") to endpoint ID. self.endpoint_id_by_iface_name = {} # Cache of IPs applied to host endpoints. (I.e. any interfaces that # aren't workload interfaces.) self.host_ep_ips_by_iface = {} # Host interface dicts by ID. We'll resolve these with the IPs above # and inject the (resolved) ones as endpoints. self.host_eps_by_id = {} # Cache of interfaces that we've resolved and injected as endpoints. self.resolved_host_eps = {} # Set of endpoints that are live on this host. I.e. ones that we've # increffed. self.local_endpoint_ids = set() # Index tracking what policy applies to what endpoints. self.policy_index = LabelValueIndex() self.policy_index.on_match_started = self.on_policy_match_started self.policy_index.on_match_stopped = self.on_policy_match_stopped self._label_inherit_idx = LabelInheritanceIndex(self.policy_index) # Tier orders by tier ID. We use this to look up the order when we're # sorting the tiers. self.tier_orders = {} # Cache of the current ordering of tier IDs. self.tier_sequence = [] # And their associated orders. self.profile_orders = {} # Set of profile IDs to apply to each endpoint ID. self.pol_ids_by_ep_id = MultiDict() self.endpoints_with_dirty_policy = set() self._data_model_in_sync = False self._iface_poll_greenlet = gevent.Greenlet(self._interface_poll_loop) self._iface_poll_greenlet.link_exception(self._on_worker_died)
def run_collapse(self, hosts, cmd): progress = None if self.progressbar: from progressbar import ProgressBar, Percentage, Bar, ETA, FileTransferSpeed progress = ProgressBar( widgets=["Running: ", Percentage(), ' ', Bar(marker='.'), ' ', ETA(), ' ', FileTransferSpeed()], maxval=len(hosts)) codes = {"total": 0, "error": 0, "success": 0} outputs = defaultdict(list) def worker(host, cmd): p = Popen(self.get_parallel_ssh_options(host, cmd), stdout=PIPE, stderr=PIPE) o = "" while True: outs, _, _ = select([p.stdout, p.stderr], [], []) outline = errline = "" if p.stdout in outs: outline = p.stdout.readline() if p.stderr in outs: errline = p.stderr.readline() o += outline + errline if outline == "" and errline == "" and p.poll() is not None: break if o == "": o = colored("[ No Output ]\n", "yellow") outputs[o].append(host) if p.poll() == 0: codes["success"] += 1 else: codes["error"] += 1 codes["total"] += 1 if self.progressbar: progress.update(codes["total"]) pool = Pool(self.ssh_threads) if self.progressbar: progress.start() for host in hosts: pool.start(Greenlet(worker, host, cmd)) try: pool.join() except KeyboardInterrupt: pass if self.progressbar: progress.finish() self.print_exec_results(codes) print() for output, hosts in outputs.items(): msg = " %s " % ','.join(hosts) table_width = min([len(msg) + 2, terminal_size()[0]]) cprint("=" * table_width, "blue", attrs=["bold"]) cprint(msg, "blue", attrs=["bold"]) cprint("=" * table_width, "blue", attrs=["bold"]) print(output)
