Returns a snapshot of the Jobs that are currently being processed by the
    ThreadPool. These jobs can not be found in the #pending_jobs() list.
969539
def current_jobs(self):
    

    jobs = []
    with synchronized(self.__queue):
      for worker in self.__threads:
        with synchronized(worker):
          if worker.current:
            jobs.append(worker.current)
    return jobs
1017518
def results( self ):
        """"""
        rs = []
        for k in self._results.keys():
            # filter out pending job ids, which can be anything except dicts
            ars = [ res for res in self._results[k] if isinstance(res, dict) ]
            rs.extend(ars)
        return rs
436067
def ListThreads(self):
    
    if self.inferior.is_running:
      return self.inferior.threads
    logging.error('Not attached to any process.')
    return []
1095790
async def run_check(self) -> int:
        
        await self.async_run()
        if self.jobs_failed:
            failed_job_results = [r for r in await self.pool.all_job_results() if not r.success]
            raise FailedJobs(self.jobs_failed, failed_job_results)
        else:
            return self.jobs_complete
444501
def get_running_jobs(opts):
    '''
    
    '''

    ret = []
    proc_dir = os.path.join(opts['cachedir'], 'proc')
    if not os.path.isdir(proc_dir):
        return ret
    for fn_ in os.listdir(proc_dir):
        path = os.path.join(proc_dir, fn_)
        data = read_proc_file(path, opts)
        if not data:
            continue
        if not is_pid_healthy(data['pid']):
            continue
        ret.append(data)
    return ret
1080418
def results
      return self.get_jobs.map do |task|
        next if task == nil
        next if task.server == nil
        {
          :job_id => task.job_id,
          :name   => task.info[0],
          :host   => task.server.name,
          :status => task.status,
          :error  => task.error,
          :total  => self.get_total_execution_time(task.status, task.time_start, task.time_end),
          :start  => task.time_start,
          :end    => task.time_end,
          :worker => task
        }
      end
    end
946229
def _main_loop(self):
        
        fetch_priorities_delay = timedelta(seconds=self.fetch_priorities_delay)
        fetch_delayed_delay = timedelta(seconds=self.fetch_delayed_delay)

        while not self.must_stop():
            self.set_status('waiting')

            if self.last_update_keys + fetch_priorities_delay < datetime.utcnow():
                self.update_keys()

            if self.last_requeue_delayed + fetch_delayed_delay < datetime.utcnow():
                self.requeue_delayed_jobs()

            try:
                queue_and_job = self.wait_for_job()
                if queue_and_job is None:
                    # timeout for blpop
                    continue
                queue, job = queue_and_job
            except Exception as e:
                self.log('Unable to get job: %s\n%s'
                    % (str(e), traceback.format_exc()), level='error')
            else:
                self.num_loops += 1
                try:
                    identifier = 'pk:%s' % job.pk.get()
                except Exception as e:
                    identifier = '??'
                try:
                    self.set_status('running')
                    identifier, status = job.hmget('identifier', 'status')
                     # some cache, don't count on it on subclasses
                    job._cached_identifier = identifier
                    job._cached_status = status
                    queue._cached_name = queue.name.hget()

                    if status == STATUSES.DELAYED:
                        self.job_delayed(job, queue)
                    elif status != STATUSES.WAITING:
                        self.job_skipped(job, queue)
                    else:
                        try:
                            self.job_started(job, queue)
                            job_result = self.callback(job, queue)
                        except Exception as e:
                            trace = None
                            if self.save_tracebacks:
                                trace = traceback.format_exc()
                            self.job_error(job, queue, e, trace)
                        else:
                            job._cached_status = job.status.hget()
                            if job._cached_status == STATUSES.DELAYED:
                                self.job_delayed(job, queue)
                            elif job._cached_status == STATUSES.CANCELED:
                                self.job_skipped(job, queue)
                            else:
                                self.job_success(job, queue, job_result)
                except Exception as e:
                    self.log('[%s] unexpected error: %s\n%s'
                        % (identifier, str(e), traceback.format_exc()), level='error')
                    try:
                        queue.errors.rpush(job.ident)
                    except Exception as e:
                        self.log('[%s] unable to add the error in the queue: %s\n%s'
                            % (identifier, str(e), traceback.format_exc()), level='error')
449962
public Map<String, ThreadPoolStatus> getStatusOfThreadPools(){
		Map<String, ThreadPoolStatus> result = new TreeMap<String, ThreadPoolStatus>();
		for (Map.Entry<String, ? extends ThreadPoolExecutor> entry: threadPools.entrySet()){
			ThreadPoolStatus status = new ThreadPoolStatus();
			status.setName(entry.getKey());
			ThreadPoolExecutor pool = entry.getValue();
			status.setActive(pool.getActiveCount());
			status.setKeepAliveSeconds(pool.getKeepAliveTime(TimeUnit.SECONDS));
			status.setLargestSize(pool.getLargestPoolSize());
			status.setMaxSize(pool.getMaximumPoolSize());
			status.setSize(pool.getPoolSize());
			status.setQueueLength(pool.getQueue().size());
			result.put(entry.getKey(), status);
		}
		return result;
	}
334615
def find_jobs(self, job_ids):
        """"""
        matched_jobs = []
        if self.skip:
            return matched_jobs

        json_data = self.download_queue(job_ids)
        if not json_data:
            return matched_jobs

        jobs = json_data["jobs"]
        for job in jobs:
            if (
                job.get("id") in job_ids
                and job.get("status", "").lower() not in _NOT_FINISHED_STATUSES
            ):
                matched_jobs.append(job)

        return matched_jobs
212350
def get_jobs_from_queue(self, queue: str, max_jobs: int) -> List[Job]:
        """"""
        jobs_json_string = self._run_script(
            self._get_jobs_from_queue,
            self._to_namespaced(queue),
            self._to_namespaced(RUNNING_JOBS_KEY.format(self._id)),
            JobStatus.RUNNING.value,
            max_jobs
        )

        jobs = json.loads(jobs_json_string.decode())
        jobs = [Job.deserialize(job) for job in jobs]

        return jobs
395653
public int getRunningThreadCount() {
        int rc = 0;
        WorkQueue[] ws; WorkQueue w;
        if ((ws = workQueues) != null) {
            for (int i = 1; i < ws.length; i += 2) {
                if ((w = ws[i]) != null && w.isApparentlyUnblocked())
                    ++rc;
            }
        }
        return rc;
    }
949907
def requeue_delayed_jobs(self):
        
        failures = []
        for queue in self.queue_model.get_all_by_priority(self.queues):
            failures.extend(queue.requeue_delayed_jobs())

        self.last_requeue_delayed = datetime.utcnow()

        for failure in failures:
            self.log('Unable to requeue %s: %s' % failure)
684631
public List<Future<JobReport>> submitAll(List<Job> jobs) {
        try {
            return executorService.invokeAll(jobs);
        } catch (InterruptedException e) {
            throw new RuntimeException("Unable to execute jobs", e);
        }
    }
1125060
def clear(self):
    

    with synchronized(self.__queue):
      jobs = self.__queue.snapshot()
      self.__queue.clear()
    return jobs
185185
def _obtain_queue(num_jobs):
    
    if _should_use_multiprocessing(num_jobs):
        return ReprQueue(multiprocessing.Manager().Queue())

    return ReprQueue(Queue())
105818
def get_jobs():
    """"""

    jobs = current_app.apscheduler.get_jobs()

    job_states = []

    for job in jobs:
        job_states.append(job)

    return jsonify(job_states)
158306
def jobs(self):
        ''''''
        # If we should resume work, then we should hand those out first,
        # assuming we can still heartbeat them
        for job in self.resume:
            try:
                if job.heartbeat():
                    yield job
            except exceptions.LostLockException:
                logger.exception('Cannot resume %s' % job.jid)
        while True:
            seen = False
            for queue in self.queues:
                job = queue.pop()
                if job:
                    seen = True
                    yield job
            if not seen:
                yield None