Primary method that will continually check for new IO events
to process.
507823
public void runEventProcessingLoop() {
        final long port = this.completionPort;
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.entry(tc, "runEventProcessingLoop: " + port + " " + Thread.currentThread());
        }

        final IAsyncProvider provider = AsyncLibrary.getInstance();
        final int size = this.batchSize;
        final int jitSize = 8192;
        final int timeout = AsyncProperties.completionTimeout;
        final WsByteBufferPoolManager wsByteBufferManager = ChannelFrameworkFactory.getBufferManager();
        final boolean jitCapable = provider.hasCapability(IAsyncProvider.CAP_JIT_BUFFERS);
        final boolean batchCapable = (1 < size);

        long[] compKeyAddrs = new long[size];
        long[] jitBufferAddressBatch = new long[size];
        WsByteBuffer[] wsBBBatch = new WsByteBuffer[size];

        int keysReady = 0;
        CompletionKey completionKey = null;

        AsyncFuture future;
        int rc = 0;
        long numBytes = 0L;

        CompletionKey[] compKeys = new CompletionKey[size];
        for (int i = 0; i < size; i++) {
            compKeys[i] = (CompletionKey) AsyncLibrary.completionKeyPool.get();
            if (compKeys[i] != null) {
                // Initialize the IOCB obtained from the pool
                compKeys[i].initializePoolEntry(0, 0);

                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                    Tr.debug(tc, "Handler got CompletionKey from Pool:" + compKeys[i]);
                }
            } else {
                compKeys[i] = new CompletionKey();

                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                    Tr.debug(tc, "Handler newed up CompletionKey:" + compKeys[i]);
                }
            }
            compKeyAddrs[i] = compKeys[i].getAddress();
        }

        try {
            // loop until we should not do any more
            while (true) {
                // see if we are shutting down, access directly for better performance
                if (AsyncLibrary.aioInitialized == AsyncLibrary.AIO_SHUTDOWN) {
                    // leave immediately if we are shutting down
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                        Tr.debug(tc, "AIO library is shutdown, ending loop");
                    }
                    break;
                }

                future = null;
                keysReady = 0;

                this.handlersWaiting.getAndIncrement();

                // get the completion keys ready for use
                if (jitCapable) {
                    for (int i = 0; i < size; i++) {
                        // reset the completionKey for next request.  Commenting this out
                        // for performance reasons.  It is driving 4 put calls per IOCD.
                        // The native code will now reset() between calls.
                        //compKeys[i].reset();

                        // see if we need to allocate a jit buffer
                        if (jitBufferAddressBatch[i] == 0) {
                            wsBBBatch[i] = wsByteBufferManager.allocateDirect(jitSize);
                            ByteBuffer bb = wsBBBatch[i].getWrappedByteBufferNonSafe();
                            jitBufferAddressBatch[i] = AbstractAsyncChannel.getBufAddress(bb);

                            compKeys[i].setJITBufferUsed();
                            compKeys[i].setBuffer(jitBufferAddressBatch[i], jitSize, 0);
                        }
                    }
                }

                // Blocks until an async call completes, or the timeout occurs
                try {
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                        Tr.debug(tc, "waiting for native event port=" + port);
                    }

                    if (batchCapable) {
                        keysReady = provider.getCompletionData3(compKeyAddrs, size, timeout, port);

                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                            Tr.debug(tc, "(batch) events: " + keysReady);
                        }
                    } else {
                        boolean gotData = provider.getCompletionData2(compKeyAddrs[0], timeout, port);

                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                            Tr.debug(tc, "(no batch) events: " + gotData);
                        }

                        if (gotData) {
                            keysReady = 1;

                            // Not used below...Commenting out for performance
                            //compKeys[0].setReturnStatus(1);
                        } else {
                            keysReady = 0;

                            // Not used below...Commenting out for performance
                            //compKeys[0].setReturnStatus(0);
                        }
                    }

                } catch (AsyncException exception) {
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                        Tr.debug(tc, "Error getting IO completion event: " + exception);
                    }
                    FFDCFilter.processException(exception, getClass().getName() + ".runEventProcessingLoop", "331", this);
                } finally {
                    this.handlersWaiting.getAndDecrement();
                }

                // if we didn't get data, go back to the top of the loop and try again
                if (keysReady == 0) {
                    // make sure this completion port didn't get closed, and then AIO
                    // restarted while this thread was in the native AIO code
                    if ((AsyncLibrary.aioInitialized != AsyncLibrary.AIO_SHUTDOWN)) {
                        if (!provider.isCompletionPortValid(port)) {
                            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                                Tr.debug(tc, "Completion port not valid");
                            }
                            break;
                        }
                    } else {
                        // AIO shutdown, so break out now
                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                            Tr.debug(tc, "AIO Shutdown detected, break out");
                        }
                        break;
                    }

                    // Timeout path. Just loop back.
                    continue;
                } // end-no-events-received
                else if (this.handlersWaiting.get() == 0) {
                    // if we have work to process and nobody else is currently
                    // calling into the native code, start an extra handler now
                    // (if allowed)
                    startHandler();
                }

                this.numItemsFromNative += keysReady;

                for (int j = keysReady - 1; j >= 0; j--) {
                    completionKey = compKeys[j];

                    // if we get here (no timeout and no exception), we should have a
                    // valid completion key
                    completionKey.postNativePrep();
                    long callid = completionKey.getCallIdentifier();

                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                        Tr.debug(tc, "batch index: " + j + " call id: " + Long.toHexString(callid) + " channel id: " + completionKey.getChannelIdentifier());
                        Tr.debug(tc, "completionKey: " + completionKey);
                    }

                    int channelIndex = (int) (callid & 0x00000000FFFFFFFF);
                    int futureIndex = (int) (callid >> 32);
                    AbstractAsyncChannel theChannel = AbstractAsyncChannel.getChannelFromIndex(channelIndex);

                    if (theChannel == null) {
                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                            Tr.debug(tc, "Could not find channel, possibly closed " + completionKey);
                        }
                        continue;
                    }

                    future = theChannel.getFutureFromIndex(futureIndex);

                    if (future == null) {
                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                            Tr.debug(tc, "Completion event could not find future " + completionKey);
                        }
                        AsyncException exception = new AsyncException("Future not found");
                        FFDCFilter.processException(exception, getClass().getName() + ".runEventProcessingLoop", "142", this);
                        continue;
                    }

                    if (jitCapable && completionKey.wasJITBufferUsed()) {
                        // set the JIT buffer value in the future
                        future.setJITBuffer(wsBBBatch[j]);
                        jitBufferAddressBatch[j] = 0;
                        // reset so new jit buffer will be allocated for next call
                    } else {
                        future.setJITBuffer(null);
                    }
                    rc = completionKey.getReturnCode();
                    numBytes = completionKey.getBytesAffected();

                    // check the event to see if it is for synchronous request
                    if (futureIndex == AbstractAsyncChannel.SYNC_READ_FUTURE_INDEX || futureIndex == AbstractAsyncChannel.SYNC_WRITE_FUTURE_INDEX) {
                        // notify waiting sync request

                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                            Tr.debug(tc, "Processing Sync Request rc: " + rc);
                        }

                        if (rc == 0) {
                            // Defend against erroneous combination of values returned from
                            // native
                            if (numBytes == 0L) {
                                IOException ioe = new IOException("Async IO operation failed, internal error");
                                future.completed(ioe);
                                continue;
                            }
                            // Mark the future as completed
                            future.setCancelInProgress(0);
                            future.completed(numBytes);
                        } else {
                            if (future.getCancelInProgress() == 1) {
                                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                                    Tr.debug(tc, "Cancel is in progress, ignoring");
                                }
                                future.setCancelInProgress(0);
                            } else {
                                future.completed(AsyncLibrary.getIOException("Async IO operation failed (3), reason: ", rc));
                            }
                        }
                        continue; // sync response has been processed
                    }

                    // signal the IO completion
                    // Check to see if the IO operation succeeded
                    if (rc == 0) {
                        future.setCancelInProgress(0);
                        future.completed(numBytes);
                    } else {
                        if (future.getCancelInProgress() == 1) {
                            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                                Tr.debug(tc, "Cancel is in progress, ignoring");
                            }
                            future.setCancelInProgress(0);
                        } else {
                            future.completed(AsyncLibrary.getIOException("Async IO operation failed (2), reason: ", rc));
                        }
                    }

                } // end the batch loop
            } // end of "while (true) {"
        } catch (Throwable t) {
            if (AsyncLibrary.aioInitialized != AsyncLibrary.AIO_SHUTDOWN) {
                FFDCFilter.processException(t, getClass().getName() + ".runEventProcessingLoop", "792", this);
                throw new RuntimeException(t);
            }
        } finally {

            // Decrement the number of in flight handlers, since this one is exiting
            this.numHandlersInFlight.decrementInt();

            // The completion processing thread is being shut down or failed
            // if jit buffer was allocated, release it
            for (int k = 0; k < size; k++) {
                if (jitBufferAddressBatch[k] != 0) {
                    wsBBBatch[k].release();
                }
            }

            // Put the CompletionKeys into the pool for reuse.   They contain a DirectByteBuffer (WsByteBuffer).
            for (int i = 0; i < size; i++) {
                if (compKeys[i] != null) {
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                        Tr.debug(tc, "Handler ending, pooling CompletionKey:\n" + compKeys[i]);
                    }
                    AsyncLibrary.completionKeyPool.put(compKeys[i]);
                }
            }
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.exit(tc, "runEventProcessingLoop: " + port + " " + Thread.currentThread());
        }
    }
139282
def check_events(self):
        ''''''
        while self.event_count() > 0:
            event = self.get_event()
            for callback in self._callbacks:
                callback(event)
460549
private void checkStdinCloses()
   {
      List<OsxProcess> processes = new ArrayList<>();
      // drainTo() is known to be atomic for ArrayBlockingQueue
      closeQueue.drainTo(processes);
      for (OsxProcess process : processes) {
         process.getStdin().close();
      }
   }
1007467
void runInternal() throws IOException, BackoffException {
        LOG.info("Starting to listen for events for {}", eventNames);

        JsonInput jsonInput = openJsonInput();

        int errorCount = 0;

        while (true) {
            try {
                final JsonParser jsonParser = jsonInput.getJsonParser();

                if (Thread.currentThread().isInterrupted()) {
                    throw new InterruptedIOException("Interrupted");
                }

                readBatch(jsonParser);

                errorCount = 0;
            } catch (IOException e) {
                // Remember interrupted flag in case it accidentally gets cleared before the break
                final boolean wasInterrupted = Thread.currentThread().isInterrupted();

                metricsCollector.markErrorWhileConsuming();

                if (errorCount > 0) {
                    LOG.warn("Got [{}] [{}] while reading events for {} after [{}] retries", e.getClass().getSimpleName(), e.getMessage(), eventNames, errorCount, e);
                } else {
                    LOG.info("Got [{}] [{}] while reading events for {}", e.getClass().getSimpleName(), e.getMessage(), eventNames, e);
                }

                jsonInput.close();

                if (wasInterrupted || Thread.currentThread().isInterrupted()) {
                    LOG.warn("Thread was interrupted");
                    break;
                }

                try {
                    LOG.debug("Reconnecting after [{}] errors", errorCount);
                    jsonInput = backoffStrategy.call(errorCount, e, this::openJsonInput);
                    LOG.info("Reconnected after [{}] errors", errorCount);
                    metricsCollector.markReconnection();
                } catch (InterruptedException interruptedException) {
                    LOG.warn("Interrupted during reconnection", interruptedException);

                    Thread.currentThread().interrupt();
                    return;
                }

                errorCount++;
            } catch (Throwable e) {
                LOG.warn("Got [{}] [{}] while reading events for {}", e.getClass().getSimpleName(), e.getMessage(), eventNames, e);

                try {
                    jsonInput.close();
                } catch (Throwable suppressed) {
                    e.addSuppressed(e);
                }
                throw e;
            }
        }
    }
623832
async def run_checks(self):
        

        async for check in self.fsm.health_check():
            yield check

        async for check in self.self_check():
            yield check

        for check in MiddlewareManager.health_check():
            yield check
1316103
protected void triggerFailover() throws IOException {
    clearDataStructures();
    
    // stop sending PREPARE_FAILOVER command
    // we are performing failover now
    prepareFailover = false;
    
    for (DatanodeInfo node : namesystem.datanodeReport(DatanodeReportType.LIVE)) {
      liveDatanodes.add(node);
      outStandingHeartbeats.add(node);
    }
    InjectionHandler
        .processEvent(InjectionEvent.STANDBY_ENTER_SAFE_MODE);
    safeModeState = SafeModeState.FAILOVER_IN_PROGRESS;
    InjectionHandler.processEvent(InjectionEvent.STANDBY_FAILOVER_INPROGRESS);
    safeModeMonitor = new Daemon(new SafeModeMonitor(namesystem, this));
    safeModeMonitor.start();
    try {
      safeModeMonitor.join();
    } catch (InterruptedException ie) {
      throw new IOException("triggerSafeMode() interruped()");
    }
    if (safeModeState != SafeModeState.AFTER_FAILOVER) {
      throw new IOException("safeModeState is : " + safeModeState +
          " which does not indicate a successfull exit of safemode");
    }
  }
16668
private void checkAndNotify(final FileStatusEntry parent, final FileStatusEntry[] previous, final Path[] currentPaths)
      throws IOException {

    int c = 0;
    final FileStatusEntry[] current =
        currentPaths.length > 0 ? new FileStatusEntry[currentPaths.length] : FileStatusEntry.EMPTY_ENTRIES;
    for (final FileStatusEntry previousEntry : previous) {
      while (c < currentPaths.length && comparator.compare(previousEntry.getPath(), currentPaths[c]) > 0) {
        current[c] = createPathEntry(parent, currentPaths[c]);
        doCreate(current[c]);
        c++;
      }
      if (c < currentPaths.length && comparator.compare(previousEntry.getPath(), currentPaths[c]) == 0) {
        doMatch(previousEntry, currentPaths[c]);
        checkAndNotify(previousEntry, previousEntry.getChildren(), listPaths(currentPaths[c]));
        current[c] = previousEntry;
        c++;
      } else {
        checkAndNotify(previousEntry, previousEntry.getChildren(), EMPTY_PATH_ARRAY);
        doDelete(previousEntry);
      }
    }

    for (; c < currentPaths.length; c++) {
      current[c] = createPathEntry(parent, currentPaths[c]);
      doCreate(current[c]);
    }
    parent.setChildren(current);
  }
245701
private void checkContinueWork() throws InterruptedException {
        boolean working = arbitrateEventService.mainStemEvent().check(pipelineId);
        if (!working) {
            logger.warn("mainstem is not run in this node");
            stopup(true);
        }

    }
606820
def listen_for_events():
    
    import_event_modules()
    conn = redis_connection.get_connection()
    pubsub = conn.pubsub()
    pubsub.subscribe("eventlib")
    for message in pubsub.listen():
        if message['type'] != 'message':
            continue
        data = loads(message["data"])
        if 'name' in data:
            event_name = data.pop('name')
            process_external(event_name, data)
1038617
void triggerBasicEvent(Event.EventType type,
                         String message,
                         boolean flushBuffer)
  {
    Event triggeredEvent = new BasicEvent(type, message);

    pushEvent(triggeredEvent, flushBuffer);
  }
918081
def run(self):
        
        listener = EventServer(self.address, EventNotifyHandler)
        log.info("Event listener running on %s", listener.server_address)
        # Listen for events until told to stop
        while not self.stop_flag.is_set():
            listener.handle_request()
335484
protected void process(BufferedReader bufferedReader) throws IOException {
        Matcher eventMatcher;
        Matcher exceptionMatcher;
        String line;
        while ((line = bufferedReader.readLine()) != null) {
            //skip empty line entries
            eventMatcher = regexpPattern.matcher(line);
            if (line.trim().equals("")) {continue;}
            exceptionMatcher = exceptionPattern.matcher(line);
            if (eventMatcher.matches()) {
                //build an event from the previous match (held in current map)
                LoggingEvent event = buildEvent();
                if (event != null) {
                    if (passesExpression(event)) {
                        doPost(event);
                    }
                }
                currentMap.putAll(processEvent(eventMatcher.toMatchResult()));
            } else if (exceptionMatcher.matches()) {
                //an exception line
                additionalLines.add(line);
            } else {
                //neither...either post an event with the line or append as additional lines
                //if this was a logging event with multiple lines, each line will show up as its own event instead of being
                //appended as multiple lines on the same event..
                //choice is to have each non-matching line show up as its own line, or append them all to a previous event
                if (appendNonMatches) {
                    //hold on to the previous time, so we can do our best to preserve time-based ordering if the event is a non-match
                    String lastTime = (String)currentMap.get(TIMESTAMP);
                    //build an event from the previous match (held in current map)
                    if (currentMap.size() > 0) {
                        LoggingEvent event = buildEvent();
                        if (event != null) {
                            if (passesExpression(event)) {
                              doPost(event);
                            }
                        }
                    }
                    if (lastTime != null) {
                        currentMap.put(TIMESTAMP, lastTime);
                    }
                    currentMap.put(MESSAGE, line);
                } else {
                    additionalLines.add(line);
                }
            }
        }

        //process last event if one exists
        LoggingEvent event = buildEvent();
        if (event != null) {
            if (passesExpression(event)) {
                doPost(event);
            }
        }
    }
10115
private synchronized void ensureIsRun() {
        if (!isRun) {
            isRun = true;
            try {
                result = executeProc(args);
                exception = null;
            } catch (ProcError e) {
                throw e.asRuntimeException();
            } catch (Throwable t) {
                exception = t;
                result = null;
            }
        }
    }
517968
def watch(directory=None, auto_clear=False, extensions=[]):
    """"""
    if directory and not os.path.isdir(directory):
        raise ValueError('Directory not found: ' + directory)
    directory = os.path.abspath(directory)

    # Initial run
    event_handler = ChangeHandler(directory, auto_clear, extensions)
    event_handler.run()

    # Setup watchdog
    observer = Observer()
    observer.schedule(event_handler, path=directory, recursive=True)
    observer.start()

    # Watch and run tests until interrupted by user
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()
1153253
def _process_streams(cls, streams):
        
        parameterizeds = defaultdict(set)
        valid, invalid = [], []
        for s in streams:
            if isinstance(s, Stream):
                pass
            elif isinstance(s, param.Parameter):
                s = Params(s.owner, [s.name])
            elif isinstance(s, param.Parameterized):
                s = Params(s)
            elif util.is_param_method(s):
                if not hasattr(s, "_dinfo"):
                    continue
                s = ParamMethod(s)
            else:
                invalid.append(s)
                continue
            if isinstance(s, Params):
                pid = id(s.parameterized)
                overlap = (set(s.parameters) & parameterizeds[pid])
                if overlap:
                    pname = type(s.parameterized).__name__
                    param.main.param.warning(
                        'The %s parameter(s) on the %s object have '
                        'already been supplied in another stream. '
                        'Ensure that the supplied streams only specify '
                        'each parameter once, otherwise multiple '
                        'events will be triggered when the parameter '
                        'changes.' % (sorted(overlap), pname))
                parameterizeds[pid] |= set(s.parameters)
            valid.append(s)
        return valid, invalid
281277
protected void checkNext ()
    {
        _executingNow = !_queue.isEmpty();
        if (_executingNow) {
            // start up a thread to execute the task in question
            ExecutorTask task = _queue.remove(0);
            final ExecutorThread thread = new ExecutorThread(task);
            thread.start();

            // start up a timer that will abort this thread after the specified timeout
            new Interval(Interval.RUN_DIRECT) {
                @Override public void expired () {
                    // this will NOOP if the task has already completed
                    thread.abort();
                }
            }.schedule(task.getTimeout());
        }
    }
974913
def _do_watch_progress(filename, sock, handler):
    """"""
    connection, client_address = sock.accept()
    data = b''
    try:
        while True:
            more_data = connection.recv(16)
            if not more_data:
                break
            data += more_data
            lines = data.split(b'\n')
            for line in lines[:-1]:
                line = line.decode()
                parts = line.split('=')
                key = parts[0] if len(parts) > 0 else None
                value = parts[1] if len(parts) > 1 else None
                handler(key, value)
            data = lines[-1]
    finally:
        connection.close()