Determine whether or not the thread pool has been idle long enough to
pause the monitoring task.

@param threadPool a reference to the thread pool
@param intervalCompleted the tasks completed this interval

@return true if the controller has been paused
482262
boolean manageIdlePool(ThreadPoolExecutor threadPool, long intervalCompleted) {

        // Manage the intervalCompleted count
        if (intervalCompleted == 0 && threadPool.getActiveCount() == 0) {
            consecutiveIdleCount++;
        } else {
            consecutiveIdleCount = 0;
        }

        if (consecutiveIdleCount >= IDLE_INTERVALS_BEFORE_PAUSE) {
            pause();
            lastAction = LastAction.PAUSE;
            return true;
        }

        return false;
    }
1369072
public boolean isWaitingWork(final T context) {
	read.lock();
	try {
	    return waitingWork.contains(context);
	} finally {
	    read.unlock();
	}
    }
547657
public void complete(NetworkConnection vc, IOReadRequestContext rctx)
   {	
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "complete", new Object[] {vc, rctx});
      if (thisConnection.isLoggingIOEvents()) thisConnection.getConnectionEventRecorder().logDebug("complete method invoked on read context "+System.identityHashCode(rctx));

      // First update the invocation count. If we are being called back on the same thread as
      // last time, increment the counter. Otherwise, start counting from 1 again.
      synchronized (invocationCountLock)
      {
         if (lastInvokedOnThread == Thread.currentThread())
         {
            invocationCount++;
         }
         else
         {
            invocationCount = 1;
            lastInvokedOnThread = Thread.currentThread();
         }
      }

      try
      {
         synchronized(this)
         {
            boolean done = false;
            do
            {
               done = true;
               WsByteBuffer contextBuffer = rctx.getBuffer();

               contextBuffer.flip();

               // Notify PMI that read has completed.
               if (conversation.getConversationType() == Conversation.CLIENT)
               {
                  xmitParser.setType(Conversation.CLIENT);
               }
               else if (conversation.getConversationType() == Conversation.ME)
               {
                  xmitParser.setType(Conversation.ME);
               }

               if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) JFapUtils.debugTraceWsByteBuffer(this, tc, contextBuffer, 16, "data received");

               xmitParser.parse(contextBuffer);

         		if ((rctx != null) && (!receivePhysicalCloseRequest))
         		{
                  // Calculate the amount of time before the request times out
                  // This is the mechanism by which we implement heartbeat
                  // intervals and time outs.
                  int timeout;
                  if (awaitingHeartbeatResponse.isSet())
                  {
                     // We can only reach this point in the code if we have
                     // made a heartbeat request but this callback was driven
                     // because of a non-heartbeat response.  In previous versions
                     // of the code we were careful to calculate the remaining time and
                     // make another read request.  This caused timeouts ala defect
                     // 363463.  So, now the code is more generous and resets its
                     // timer to the full heartbeat timeout value every time any
                     // non-heartbeat trasmission is received from our peer.  This isn't
                     // unreasonable as the fact we are receiving data indicates that the
                     // peer is probably still healthy.
                     timeout = currentHeartbeatTimeout*1000;
                     if (timeout < 1) timeout = 1;
                  }
                  else
                  {
                     // We are not awaiting a heartbeat response, use the
                     // heartbeat interval as our timeout.
                     timeout = thisConnection.getHeartbeatInterval()*1000;
                  }

                  if (timeout > 0)
                  {
                     if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "setting heartbeat timeout to: "+timeout+" milliseconds");
                  }
                  else
                  {
                     if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "not using a heartbeat timeout");
                     timeout = IOReadRequestContext.NO_TIMEOUT;
                  }

                  boolean closing = false;
                  synchronized(connectionClosingLock)
                  {
                     closing = connectionClosing;
                  }

                  //If the connection is closing/closed our buffers will have been released, so make sure we don't use them again.
                  if (!closing)
                  {
                     // Crude way to ensure we end up with the right sized read buffer.
                     if (isFirstCompleteInvocation)
                     {
                        int readBufferSize =
                           Integer.parseInt(RuntimeInfo.getProperty("com.ibm.ws.sib.jfapchannel.DEFAULT_READ_BUFFER_SIZE", ""+JFapChannelConstants.DEFAULT_READ_BUFFER_SIZE));
                        if (!contextBuffer.isDirect() || (contextBuffer.capacity() < readBufferSize))
                        {
                           //Make sure we release the other buffer so we don't leak memory.
                           contextBuffer.release();

                           contextBuffer = WsByteBufferPool.getInstance().allocateDirect(readBufferSize);
                           rctx.setBuffer(contextBuffer);
                        }
                        isFirstCompleteInvocation = false;
                     }

                     contextBuffer.clear();

                     // Decide whether to explictly request a thread switch. We'll do this if we
                     // have been recursively called more than MAX_INVOCATIONS_BEFORE_THREAD_SWITCH
                     boolean forceQueue = false;
                     synchronized (invocationCountLock)
                     {
                        if (invocationCount > MAX_INVOCATIONS_BEFORE_THREAD_SWITCH)
                        {
                           forceQueue = true;
                           lastInvokedOnThread = null;
                        }
                     }

                     if (thisConnection.isLoggingIOEvents()) thisConnection.getConnectionEventRecorder().logDebug("invoking readCtx.read() on context "+System.identityHashCode(rctx)+" with a timeout of "+timeout);
                     done = (rctx.read(1, this, forceQueue, timeout) == null);
                   
                  }
         		}
            }
            while(!done);
         }
      }
      catch(Error error)
      {
         FFDCFilter.processException
         (error, "com.ibm.ws.sib.jfapchannel.impl.ConnectionReadCompletedCallback", JFapChannelConstants.CONNREADCOMPCALLBACK_COMPLETE_01, thisConnection.getDiagnostics(true));
         if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) SibTr.exception(this, tc, error);

         // It might appear slightly odd for this code to catch Error (especially since the JDK docs say
         // that Error means that something has gone so badly wrong that you should abandon all hope).
         // This code makes one final stab at putting out some diagnostics about what happened (if we
         // propagate the Error up to the TCP Channel, it is sometimes lost) and closing down the
         // connection.  I figured that we might as well try to do something - as we can hardly make
         // things worse... (famous last words)

         thisConnection.invalidate(false, error, "Error caught in ConnectionReadCompletedCallback.complete()");

         // Re-throw the error to ensure that it causes the maximum devastation.
         // The JVM is probably very ill if an Error is thrown so attempt no recovery.
         throw error;
      }
      catch(RuntimeException runtimeException)
      {
         FFDCFilter.processException
         (runtimeException, "com.ibm.ws.sib.jfapchannel.impl.ConnectionReadCompletedCallback", JFapChannelConstants.CONNREADCOMPCALLBACK_COMPLETE_02, thisConnection.getDiagnostics(true));
         if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) SibTr.exception(this, tc, runtimeException);

         // We can reasonably try to recover from a runtime exception by invalidating the associated
         // connection.  This should drive the underlying TCP/IP socket to be closed.
         thisConnection.invalidate(false, runtimeException, "RuntimeException caught in ConnectionReadCompletedCallback.complete");

         // Don't throw the RuntimeException on as we risk blowing away part of the TCP channel.
      }

		if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "complete");
	}
69718
public boolean isFinished()
	{
		if (finished)
			return true;

		try
		{
			final int code = exitCode();

			finished(code);

			return true;
		}
		catch (IllegalThreadStateException e)
		{
			return false;
		}
	}
1043553
private void complete() {
		if (stopped) {
			return;
		}

		log.info("Spidering process is complete. Shutting down...");
		this.stopped = true;
		if (httpSender != null) {
			this.getHttpSender().shutdown();
			httpSender = null;
		}

		// Notify the controller to clean up memory
		controller.reset();

		// Issue the shutdown command on a separate thread, as the current thread is most likely one
		// from the pool
		new Thread(new Runnable() {
			@Override
			public void run() {
				if (threadPool != null) {
					threadPool.shutdown();
				}
				// Notify the listeners -- in the meanwhile
				notifyListenersSpiderComplete(true);
				controller.reset();
				threadPool = null;
			}
		}, "ZAP-SpiderShutdownThread-" + id).start();
	}
186164
final void continuouslyRestart(long interval) throws IOException, InterruptedException {
        LOGGER.entering(interval);

        while (true) {
            if (!isInitialized()) {
                FileDownloader.checkForDownloads(getType(), getLauncherOptions()
                        .isFileDownloadCheckTimeStampOnInvocation(), getLauncherOptions()
                        .isFileDownloadCleanupOnInvocation());
            }

            startProcess(false);

            while (!handler.hasResult()) {
                LOGGER.fine("Child process still running. Going back to sleep.");
                Thread.sleep(interval);
            }

            if (handler.hasResult()) {
                ExecuteException e = handler.getException();
                if (e != null) {
                    LOGGER.log(Level.SEVERE, handler.getException().getMessage(), handler.getException());
                }
            }
            LOGGER.info("Child process quit. Restarting it.");
            setInitialized(false);
        }
    }
1205183
public void tick() throws InterruptedException {
    if (mPreviousTickMs != 0) {
      long executionTimeMs = mClock.millis() - mPreviousTickMs;
      if (executionTimeMs > mIntervalMs) {
        mLogger.warn("{} last execution took {} ms. Longer than the interval {}", mThreadName,
            executionTimeMs, mIntervalMs);
      } else {
        mSleeper.sleep(Duration.ofMillis(mIntervalMs - executionTimeMs));
      }
    }
    mPreviousTickMs = mClock.millis();
  }
385490
private boolean pauseIfRetryNeeded(int retries) {
        if (retries >= connection.getMaxErrorRetry()) {
            return false;
        }
        long delay = (long) (Math.random() * Math.pow(4, retries) * 125);
        try {
            Thread.sleep(delay);
        } catch (Exception e) {
            throw MwsUtl.wrap(e);
        }
        return true;
    }
248362
private boolean handlePausedFlowAfterStep(Execution execution) throws InterruptedException {
        String branchId = execution.getSystemContext().getBranchId();
        PauseReason reason = null;
        ExecutionSummary execSummary = pauseService.readPausedExecution(execution.getExecutionId(), branchId);
        if (execSummary != null && execSummary.getStatus().equals(ExecutionStatus.PENDING_PAUSE)) {
            reason = execSummary.getPauseReason();
        }
        if (reason != null) { // need to pause the execution
            pauseFlow(reason, execution);
            return true;
        }
        return false;
    }
1293911
@Override
    public void finish() {
        // An assertion failure here means this CE
        // did not come from a CE pool and this
        // should never be the case;
        //assert cacheEntryPool != null;
        // if cacheEntryPool is null, the cacheEntry is not allocated from the pool
        if (cacheEntryPool != null) {
            synchronized (cacheEntryPool.cache) {
                decRefCount();
                if (refCount.get() <= 0) {
                    if (refCount.get() < 0)
                        refCount.set(0);
                    if (removeWhenUnpinned && refCount.get() == 0) {
                        returnToPool();
                    }
                }
            }
        } else {
            synchronized (this.id) {
                decRefCount();
                if (refCount.get() <= 0) {
                    reset();
                }
            }
        }
    }
209246
def run(self):
    
    while True:
      if self.pool.name:
        self.idle = True

      try:
        # Wait 60 seconds for a message, otherwise exit. This ensures that the
        # threadpool will be trimmed down when load is light.
        task = self._queue.get(timeout=60)

        if self.pool.name:
          self.idle = False

        try:
          # The pool told us to quit, likely because it is stopping.
          if task == STOP_MESSAGE:
            return

          self.ProcessTask(*task)
        finally:
          self._queue.task_done()

      except queue.Empty:
        if self._RemoveFromPool():
          return

      # Try to trim old threads down when they get too old. This helps the
      # thread pool size to shrink, even when it is not idle (e.g. if it is CPU
      # bound) since threads are forced to exit, but new threads will not be
      # created if the utilization is too high - resulting in a reduction of
      # threadpool size under CPU load.
      if time.time() - self.started > 600 and self._RemoveFromPool():
        return
695131
def monitor(seconds_frozen, test_interval):
    
    current_thread = threading.current_thread()
    hanging_threads = set()
    old_threads = {}  # Threads found on previous iteration.

    while not current_thread.is_stopped():
        new_threads = get_current_frames()

        # Report died threads.
        for thread_id in old_threads.keys():
            if thread_id not in new_threads and thread_id in hanging_threads:
                log_died_thread(thread_id)

        # Process live threads.
        time.sleep(test_interval/1000.)
        now = time.time()
        then = now - seconds_frozen
        for thread_id, thread_data in new_threads.items():
            # Don't report the monitor thread.
            if thread_id == current_thread.ident:
                continue
            frame = thread_data['frame']
            # If thread is new or it's stack is changed then update time.
            if (thread_id not in old_threads or
                    frame != old_threads[thread_id]['frame']):
                thread_data['time'] = now
                # If the thread was hanging then report awaked thread.
                if thread_id in hanging_threads:
                    hanging_threads.remove(thread_id)
                    log_awaked_thread(thread_id)
            else:
                # If stack is not changed then keep old time.
                last_change_time = old_threads[thread_id]['time']
                thread_data['time'] = last_change_time
                # Check if this is a new hanging thread.
                if (thread_id not in hanging_threads and
                        last_change_time < then):
                    # Gotcha!
                    hanging_threads.add(thread_id)
                    # Report the hanged thread.
                    log_hanged_thread(thread_id, frame)
        old_threads = new_threads
442281
public boolean waitForNewTimeslot() throws InterruptedException {
		
		boolean hasWaited = false;
		
		while(true) {
			final long numberOfEventsInTimeSlot = getNumberOfEventsInTimeslot();
			
			if(numberOfEventsInTimeSlot > numberOfEvents) {
				hasWaited = true;
				Thread.sleep(timeslotInMilliseconds / 10);
			} else {
				return hasWaited;
			}
		}
	}
1241627
ReadIteratorResource<Read, ReadGroupSet, Reference> queryNextInterval() {
    Stopwatch w = Stopwatch.createStarted();
    if (!isAtEnd()) {
      intervalIndex++;
    }
    if (isAtEnd()) {
      return null;
    }
    ReadIteratorResource<Read, ReadGroupSet, Reference> result =  
        queryForInterval(currentInterval());
    LOG.info("Interval query took: " + w);
    startTiming();
    return result;
  }
1355902
def is_paused(self):
        
        root_is_paused = False
        if self.root_id != self.id:
            root_is_paused = context.connections.redis.sismember(redis_key("paused_queues"), self.root_id)

        return root_is_paused or context.connections.redis.sismember(redis_key("paused_queues"), self.id)
746273
private int externalAwaitDone() {
    int s = ((this instanceof CountedCompleter) ? // try helping
        ForkJoinPool.common.externalHelpComplete((CountedCompleter<?>) this, 0)
        : ForkJoinPool.common.tryExternalUnpush(this) ? doExec() : 0);
    if (s >= 0 && (s = status) >= 0) {
      boolean interrupted = false;
      do {
        if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
          synchronized (this) {
            if (status >= 0) {
              try {
                wait(0L);
              } catch (InterruptedException ie) {
                interrupted = true;
              }
            } else
              notifyAll();
          }
        }
      } while ((s = status) >= 0);
      if (interrupted)
        Thread.currentThread().interrupt();
    }
    return s;
  }
7438
public static boolean shutdownAndAwaitTermination(ExecutorService pool, long waitSeconds) {
		pool.shutdown(); // Disable new tasks from being submitted
		try {
			// Wait a while for existing tasks to terminate
			if(!pool.awaitTermination(waitSeconds, TimeUnit.SECONDS)) {
				//logger.warn("Thread pool is still running: {}", pool);
				pool.shutdownNow(); // Cancel currently executing tasks
				// Wait a while for tasks to respond to being cancelled
				if(!pool.awaitTermination(waitSeconds, TimeUnit.SECONDS)){
					//logger.warn("Thread pool is not terminating: {}", pool);
					return false;
				}else{
					return true;
				}
			}else{
				return true;
			}
		} catch(InterruptedException ie) {
			// (Re-)Cancel if current thread also interrupted
			pool.shutdownNow();
			// Preserve interrupt status
			Thread.currentThread().interrupt();
			return false;
		}
	}