/*
Ensure that handler isn't invoked on IO thread but rather
dispatched to the worker thread pool
351813
protected HttpHandler wrapWithDispatcher(final HttpHandler handler) {
        return new HttpHandler() {
            public void handleRequest(HttpServerExchange exchange) throws Exception {
                if (exchange.isInIoThread()) {
                    exchange.dispatch(handler);
                } else {
                    handler.handleRequest(exchange);
                }
            }
        };
    }
767455
protected boolean dispatch(TCPBaseRequestContext req, IOException ioe) {

        if (req.blockedThread) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                Tr.debug(tc, "dispatcher notifying waiting synch request ");
            }
            if (ioe != null) {
                req.blockingIOError = ioe;
            }
            req.blockWait.simpleNotify();
            return true;
        }
        // dispatch the async work
        return dispatchWorker(new Worker(req, ioe));
    }
1043916
private void handleRunAsync(RunAsync runAsync) {
		final long timeToRun = runAsync.getTimeNanos();
		final long delayNanos;

		if (timeToRun == 0 || (delayNanos = timeToRun - System.nanoTime()) <= 0) {
			// run immediately
			try {
				runAsync.getRunnable().run();
			} catch (Throwable t) {
				log.error("Caught exception while executing runnable in main thread.", t);
				ExceptionUtils.rethrowIfFatalErrorOrOOM(t);
			}
		}
		else {
			// schedule for later. send a new message after the delay, which will then be immediately executed
			FiniteDuration delay = new FiniteDuration(delayNanos, TimeUnit.NANOSECONDS);
			RunAsync message = new RunAsync(runAsync.getRunnable(), timeToRun);

			final Object envelopedSelfMessage = envelopeSelfMessage(message);

			getContext().system().scheduler().scheduleOnce(delay, getSelf(), envelopedSelfMessage,
					getContext().dispatcher(), ActorRef.noSender());
		}
	}
1103396
private void dispatchThread(Runnable runnable)
   {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "dispatchThread");

      try
      {
         // Get a thread from the pool to excute it
         // By only passing the thread we default to wait if the threadpool queue is full
         // We should wait as some callbacks are more important then others but we have no
         // way of distinguishing this.
         threadPool.execute(runnable);
      }
      catch (InterruptedException e)
      {
         // No FFDC code needed
         if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(tc, "Thread was interrupted", e);
      }

      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "dispatchThread");
   }
781824
private int unregisterHandles() {
        int cancelledHandles = 0;
        for (;;) {
            AcceptorOperationFuture future = cancelQueue.poll();
            if (future == null) {
                break;
            }

            // close the channels
            for (SocketAddress a : future.getLocalAddresses()) {
                H handle = boundHandles.remove(a);
                
                if (handle == null) {
                    continue;
                }

                try {
                    close(handle);
                    wakeup(); // wake up again to trigger thread death
                } catch (Throwable e) {
                    ExceptionMonitor.getInstance().exceptionCaught(e);
                } finally {
                    cancelledHandles++;
                }
            }

            future.setDone();
        }

        return cancelledHandles;
    }
989443
void workerRun(TCPBaseRequestContext req, IOException ioe) {
        if (null == req || req.getTCPConnLink().isClosed()) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) {
                Tr.event(tc, "Ignoring IO on closed socket: " + req);
            }
            return;
        }
        try {
            if (ioe == null) {
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                    Tr.debug(tc, "Worker thread processing IO request: " + req);
                }
                attemptIO(req, true);
            } else {
                if (req.isRequestTypeRead()) {
                    TCPReadRequestContextImpl readReq = (TCPReadRequestContextImpl) req;
                    if (readReq.getReadCompletedCallback() != null) {
                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                            Tr.debug(tc, "Worker thread processing read error: " + req.getTCPConnLink().getSocketIOChannel().getChannel());
                        }
                        readReq.getReadCompletedCallback().error(readReq.getTCPConnLink().getVirtualConnection(), readReq, ioe);
                    }
                } else {
                    TCPWriteRequestContextImpl writeReq = (TCPWriteRequestContextImpl) req;
                    if (writeReq.getWriteCompletedCallback() != null) {
                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                            Tr.debug(tc, "Worker thread processing write error: " + req);
                        }
                        writeReq.getWriteCompletedCallback().error(writeReq.getTCPConnLink().getVirtualConnection(), writeReq, ioe);
                    }
                }
            }
        } catch (Throwable t) {
            // Only issue an FFDC if the framework is up/valid..
            if (FrameworkState.isValid()) {
                FFDCFilter.processException(t, getClass().getName(), "workerRun(req)", new Object[] { this, req, ioe });
            }
            if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) {
                Tr.event(tc, "Unexpected error in worker; " + t);
            }
        }
    }
1322784
function(pool) {
    if (idleConnections.isEmpty()) {
      return Promise.reject(null);
    }
    const conn = idleConnections.shift();
    activeConnections[conn.threadId] = conn;
    if (
      opts.minDelayValidation <= 0 ||
      Date.now() - conn.lastUse > opts.minDelayValidation
    ) {
      return pingPromise(conn)
        .then(() => {
          return Promise.resolve(conn);
        })
        .catch(err => {
          delete activeConnections[conn.threadId];
          pool.emit('remove-conn');
          return getIdleValidConnection(pool);
        });
    } else {
      //just check connection state
      if (conn.isValid()) {
        return Promise.resolve(conn);
      } else {
        delete activeConnections[conn.threadId];
        pool.emit('remove-conn');
        return getIdleValidConnection(pool);
      }
    }
  }
163549
def _start_handler_thread(self):
        """"""
        # Create handler thread
        t = Thread(target=self._start_loop)
        t.daemon = True

        # Start handler thread
        t.start()
        self.thread_started = True
749797
function _removeHandler(name, ignore) {
            // Remove handler if needed
            var func = PostMessageUtilities.parseFunction(this[name]);
            if (func) {
                if (!ignore) {
                    func.call(this);
                }

                this[name] = void 0;
                delete this[name];
            }
        }
316027
function runTheHandler(inboundMessage) {
    let handler = convo.handler;
    convo.handler = null;
    
    let result = handler.apply(convo, [convo, inboundMessage]);

    
    if (result === IS_WAIT_FUNCTION) {
      // reset the handler to wait again
      convo.handler = handler;
    }
    // this was a real interaction so reset the timeout and wait count
    else {
      fuse.logger.verbose('Ran handler');
      resetTimeout();
      convo.wait_count = 0;
    }
  }
1086149
protected void notifyToHandlers(ApiZone zone, ApiBuddyImpl buddy) {
        for(ServerHandlerClass handler : handlers) {
            notifyToHandler(handler, zone, buddy);
        }
    }
377815
@Deprecated
    protected synchronized void workerDone(Worker w, boolean taskDied) {

        threads_.remove(w);

        if (taskDied) {
            --activeThreads;
            --poolSize_;
        }

        if (poolSize_ == 0 && shutdown_) {
            maximumPoolSize_ = minimumPoolSize_ = 0; // disable new threads
            notifyAll(); // notify awaitTerminationAfterShutdown
        }

        fireThreadDestroyed(poolSize_);
    }
798637
def UnregisterFlowProcessingHandler(self, timeout=None):
    """"""
    if self.flow_processing_request_handler_thread:
      self.flow_processing_request_handler_stop = True
      self.flow_processing_request_handler_thread.join(timeout)
      if self.flow_processing_request_handler_thread.isAlive():
        raise RuntimeError("Flow processing handler did not join in time.")
      self.flow_processing_request_handler_thread = None
658773
public void setHandler (@Nonnull final EJSEvent eJSEvent, @Nonnull final IHasJSCode aNewHandler)
  {
    ValueEnforcer.notNull (eJSEvent, "JSEvent");
    ValueEnforcer.notNull (aNewHandler, "NewHandler");

    // Set only the new handler and remove any existing handler
    m_aEvents.put (eJSEvent, new CollectingJSCodeProvider ().appendFlattened (aNewHandler));
  }
1212435
protected void enqueueEvent(Object event, EventHandler handler) {
    eventsToDispatch.get().offer(new EventWithHandler(event, handler));
  }
1078513
public boolean flushQueuePointOutputHandler()
    throws SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(
        tc,
        "flushQueuePointOutputHandler");

    boolean done = true;

    for(Iterator nextHandler = _queuePointOutputHandlers.values().iterator();
        nextHandler.hasNext();)
    {
      Object handlerQ = nextHandler.next();
      // Try and drive flush on every output handler.
      // Note that we'll try and drive every output handler
      // even if an earlier output handler required a deferred
      // delete.
      if (handlerQ instanceof PtoPOutputHandler)
        done &= ((PtoPOutputHandler) handlerQ).flushAllForDelete();
    }
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "flushQueuePointOutputHandler", new Boolean(done));
    return done;
  }
69190
def fire(self, *args, **kw):
        
        result = []

        with self._hlock:
            handlers = self.handlers

        if self.threads == 0:  # same-thread execution - synchronized
            for k in handlers:
                # handler, memoize, timeout
                h, m, t = handlers[k]
                try:
                    r = self._memoize(h, m, t, *args, **kw)
                    result.append(tuple(r))
                except:
                    result.append((False, self._error(sys.exc_info()), h))

        elif self.threads > 0:  # multi-thread execution - desynchronized if self.threads > 1
            queue = Queue()

            # result lock just in case [].append() is not  
            # thread-safe in other Python implementations
            rlock = RLock()

            def _execute(*args, **kw):
                """ Executes all handlers stored in the queue """
                while True:
                    try:
                        item = queue.get()
                        if item is None:
                            queue.task_done()
                            break

                        # handler, memoize, timeout
                        h, m, t = handlers[item]  # call under active lock

                        try:
                            r = self._memoize(h, m, t, *args, **kw)
                            if not self.asynch:
                                with rlock:
                                    result.append(tuple(r))
                        except:
                            if not self.asynch:
                                with rlock:
                                    result.append((False, self._error(sys.exc_info()), h))

                        queue.task_done()

                    except Empty:  # never triggered, just to be safe
                        break

            if handlers:
                threads = self._threads(handlers=handlers)

                for _ in range(threads):
                    t = Thread(target=_execute, args=args, kwargs=kw)
                    t.daemon = True
                    t.start()

                for k in handlers:
                    queue.put(k)

                    if self.asynch:  # main thread, no locking required
                        h, _, _ = handlers[k]
                        result.append((None, None, h))

                for _ in range(threads):
                    queue.put(None)  # stop each worker

                if not self.asynch:
                    queue.join()

        return tuple(result) or None