This method will execute {@code runnable.run()} method on behalf of the specified tenant.

@param tenant   the tenant
@param runnable an instance on which {@code run()} method will be executed
762916
static void execute(Tenant tenant, Runnable runnable) {
        startTenantFlow();
        try {
            getThreadLocalInstance().setTenant(tenant);
            runnable.run();
        } finally {
            endTenantFlow();
        }
    }
1040101
@Override
    public void execute(Runnable task) {
        // count a new task in process
        tasksInProcess.incrementAndGet();
        try {
            super.execute(task);
        } catch (RuntimeException e) { // specifically handle RejectedExecutionException
            tasksInProcess.decrementAndGet();
            throw e;
        } catch (Error e) {
            tasksInProcess.decrementAndGet();
            throw e;
        }
    }
112454
private Runnable runStepAndResetIsRunning(Runnable runnable) {
        return () -> {
            try {
                runnable.run();
            } catch (Throwable t) {
                //we're in control of the runnable and it should catch it's own execeptions, but just in case it doesn't
                log.error("Exeception while running a step", t);
            } finally {
                isRunningAStep.getAndSet(false);
            }
        };
    }
297563
public C run(Runnable runnable) throws SniffyAssertionError {
        checkOpened();
        try {
            runnable.run();
        } catch (Throwable e) {
            throw verifyAndAddToException(e);
        }

        verify();
        return self();
    }
977140
@Override
    public void run(Context context) throws IOException, InterruptedException {
        outer = context;
        int numberOfThreads = getThreadCount(context);
        mapClass = getMapperClass(context);
        if (LOG.isDebugEnabled()) {
            LOG.debug("Running with " + numberOfThreads + " threads");
        }        
        // current mapper takes 1 thread
        numberOfThreads--;
        
        InputSplit split = context.getInputSplit();

        // submit runners
        try {
	        List<Future<?>> taskList = null;
	        if (threadPool != null) {
            	taskList = new ArrayList<Future<?>>();
                synchronized (threadPool) {
                    for (int i = 0; i < numberOfThreads; ++i) {
                        MapRunner runner = new MapRunner();
                        BaseMapper<K1,V1,K2,V2> mapper = runner.getMapper();
                        if (!threadPool.isShutdown()) {
                            Collection<Future<Object>> tasks = 
                                mapper.submitTasks(threadPool, split);
                            taskList.addAll(tasks);
                            numberOfThreads -= tasks.size();
                            Future<?> future = threadPool.submit(runner);
                            taskList.add(future);	                        
                        } else {
                            throw new InterruptedException(
                                "Thread Pool has been shut down");
                        }
                    }
                    threadPool.notify();
                }
	            
                // MapRunner that runs in current thread
                MapRunner r = new MapRunner();
                r.run();
    
                for (Future<?> f : taskList) {
                    f.get();
                }
	        } else {
	            runners = new ArrayList<MapRunner>(numberOfThreads);
                for (int i = 0; i < numberOfThreads; ++i) {
                    MapRunner thread;
                    thread = new MapRunner();
                    thread.start();
                    runners.add(i, thread);
                }
                // MapRunner runs in current thread
                MapRunner r = new MapRunner();
                r.run();
                
                for (int i = 0; i < numberOfThreads; ++i) {
                    MapRunner thread = runners.get(i);
                    thread.join();
                    Throwable th = thread.throwable;
                    if (th != null) {
                        if (th instanceof IOException) {
                            throw (IOException) th;
                        } else if (th instanceof InterruptedException) {
                            throw (InterruptedException) th;
                        } else {
                            throw new RuntimeException(th);
                        }
                    }
                }
            }       
        } catch (ClassNotFoundException e) {
            LOG.error("MapRunner class not found", e);
        } catch (ExecutionException e) {
        	LOG.error("Error waiting for MapRunner threads to complete", e);
		}        
    }
749594
@Override
    public Thread newThread(Runnable runnable) {
        Thread thread = new Thread(runnable);
        thread.setPriority(threadPriority);
        return thread;
    }
1358637
synchronized void execute(File root, Runnable task) {
    if (executors == null) {
      throw new RuntimeException("AsyncDiskService is already shutdown");
    }
    ThreadPoolExecutor executor = executors.get(root);
    if (executor == null) {
      throw new RuntimeException("Cannot find root " + root
          + " for execution of task " + task);
    } else {
      executor.execute(task);
    }
  }
298427
private function runRunnable($runnable)
    {
        if (!$runnable) return;

        $runnable->run();
        if ($runnable->getException()) {
            self::$console->setExitStatus(1);
            if (self::$console->options['stop']) {
                $this->reporter->afterRun();
                exit(1);
            }
        }
    }
904467
@Override
    public void run() {
        final boolean trace = TraceComponent.isAnyTracingEnabled();
        if (trace && tc.isEntryEnabled())
            Tr.entry(this, tc, "run[" + taskId + ']', persistentExecutor);

        Config config = persistentExecutor.configRef.get();
        if (persistentExecutor.deactivated || !config.enableTaskExecution) {
            if (trace && tc.isEntryEnabled())
                Tr.exit(this, tc, "run[" + taskId + ']', persistentExecutor.deactivated ? "deactivated" : ("enableTaskExecution? " + config.enableTaskExecution));
            return;
        }

        // Work around for when the scheduled executor fires too early
        long execTime = new Date().getTime();
        if (execTime < expectedExecTime) {
            long delay = expectedExecTime - execTime;
            persistentExecutor.scheduledExecutor.schedule(this, delay, TimeUnit.MILLISECONDS);
            if (trace && tc.isEntryEnabled())
                Tr.exit(this, tc, "run[" + taskId + ']', "attempted to run " + delay + " ms too early. Rescheduled.");
            return;
        }

        // If a Configuration update is in progress, then we will defer the execution of this Task until after.
        if (persistentExecutor.deferExecutionForConfigUpdate(this) == true) {
            if (trace && tc.isEntryEnabled())
                Tr.exit(this, tc, "run[" + taskId + ']', "attempted to run during a configuration update.");
            return;
        }

        String taskName = null;
        String taskIdForPropTable = null;
        TaskLocker ejbSingletonLockCollaborator = null;
        String ownerForDeferredTask = null;
        ClassLoader loader = null;
        Throwable failure = null;
        Short prevFailureCount = null, nextFailureCount = null;
        Long nextExecTime = null;
        TaskStore taskStore = persistentExecutor.taskStore;
        ApplicationTracker appTracker = persistentExecutor.appTrackerRef.getServiceWithException();
        TransactionManager tranMgr = persistentExecutor.tranMgrRef.getServiceWithException();
        taskIdsOfRunningTasks.set(taskId);
        try {
            int timeout = txTimeout == 0 && (binaryFlags & TaskRecord.Flags.SUSPEND_TRAN_OF_EXECUTOR_THREAD.bit) != 0 ? DEFAULT_TIMEOUT_FOR_SUSPENDED_TRAN : txTimeout;
            tranMgr.setTransactionTimeout(timeout);

            TaskRecord ejbSingletonRecord = null;
            if ((binaryFlags & TaskRecord.Flags.EJB_SINGLETON.bit) != 0 && (binaryFlags & TaskRecord.Flags.SUSPEND_TRAN_OF_EXECUTOR_THREAD.bit) == 0) {
                tranMgr.begin();
                ejbSingletonRecord = taskStore.getTrigger(taskId);
                tranMgr.commit();
            }

            tranMgr.begin();
            TaskRecord taskRecord;

            // Execution property TRANSACTION=SUSPEND indicates the task should not run in the persistent executor transaction.
            // Lock an entry in a different table to prevent concurrent execution, and run with that transaction suspended.
            if ((binaryFlags & TaskRecord.Flags.SUSPEND_TRAN_OF_EXECUTOR_THREAD.bit) != 0) {
                if (!taskStore.createProperty(taskIdForPropTable = "{" + taskId + "}", " "))
                    throw new IllegalStateException(taskIdForPropTable); // Internal error if this path is ever reached
                Transaction suspendedTran = tranMgr.suspend();
                try {
                    // We still need the task information, but get it in a new transaction that we can commit right away.
                    Throwable failed = null;
                    tranMgr.begin();
                    try {
                        taskRecord = taskStore.find(taskId,
                                                    persistentExecutor.getPartitionId(),
                                                    new Date().getTime(),
                                                    false);
                    } catch (Throwable x) {
                        throw failed = x;
                    } finally {
                        if (failed == null)
                            tranMgr.commit();
                        else
                            tranMgr.rollback();
                    }
                } finally {
                    tranMgr.resume(suspendedTran); // will be suspended again by application of transaction context
                }
            } else {
                if (ejbSingletonRecord != null) {
                    String owner = ejbSingletonRecord.getIdentifierOfOwner();
                    if (!appTracker.isStarted(owner)) {
                        ownerForDeferredTask = owner;
                        if (trace && tc.isEntryEnabled())
                            Tr.exit(this, tc, "run[" + taskId + ']', "unavailable - deferred");
                        return; // Ignore, we are deferring the task because the application or module is unavailable
                    }

                    byte[] bytes = ejbSingletonRecord.getTrigger();
                    ejbSingletonLockCollaborator = (TaskLocker) persistentExecutor.deserialize(bytes, priv.getSystemClassLoader());
                    if (trace && tc.isDebugEnabled())
                        Tr.debug(this, tc, "notify EJB container to lock singleton");
                    ejbSingletonLockCollaborator.lock();
                }

                taskRecord = taskStore.find(taskId,
                                            persistentExecutor.getPartitionId(),
                                            new Date().getTime(),
                                            true);
            }

            if (taskRecord == null || (taskRecord.getState() & TaskState.ENDED.bit) != 0) {
                if (trace && tc.isEntryEnabled())
                    Tr.exit(this, tc, "run[" + taskId + ']', "not appropriate to run task at this time");
                return; // Ignore, because the task was canceled or someone else already ran it
            }

            taskName = taskRecord.getName();
            prevFailureCount = taskRecord.getConsecutiveFailureCount();

            String classLoaderIdentifier = taskRecord.getIdentifierOfClassLoader();
            if (trace && tc.isDebugEnabled())
                Tr.debug(this, tc, "classloader identifier", classLoaderIdentifier);
            loader = ejbSingletonRecord == null ? persistentExecutor.classloaderIdSvc.getClassLoader(classLoaderIdentifier) : priv.getSystemClassLoader();
            if (trace && tc.isDebugEnabled())
                Tr.debug(this, tc, "classloader", loader);

            String owner = taskRecord.getIdentifierOfOwner();
            if (loader == null || !appTracker.isStarted(owner)) {
                ownerForDeferredTask = owner;
                if (trace && tc.isEntryEnabled())
                    Tr.exit(this, tc, "run[" + taskId + ']', "unavailable - deferred");
                return; // Ignore, we are deferring the task because the application or module is unavailable
            }

            TaskInfo info = (TaskInfo) persistentExecutor.deserialize(taskRecord.getTaskInformation(), null);

            byte[] triggerBytes = taskRecord.getTrigger();
            Trigger trigger = triggerBytes == null ? null : ejbSingletonRecord != null
                                                            && Arrays.equals(triggerBytes,
                                                                             ejbSingletonRecord.getTrigger()) ? ejbSingletonLockCollaborator : (Trigger) persistentExecutor.deserialize(triggerBytes,
                                                                                                                                                                                        loader);
            if (trigger == null) {
                String triggerClassName = info.getClassNameForNonSerializableTrigger();
                if (triggerClassName != null)
                    trigger = (Trigger) loader.loadClass(triggerClassName).newInstance();
            }

            byte[] taskBytes = taskRecord.getTask();
            Object task = taskBytes == null ? null : persistentExecutor.deserialize(taskBytes, loader);
            if (task == null) {
                String taskClassName = info.getClassNameForNonSerializableTask();
                if (taskClassName == null)
                    task = trigger; // optimization to share single instance for task and trigger
                else
                    task = loader.loadClass(taskClassName).newInstance();
            }

            byte[] resultBytes = taskRecord.getResult();

            boolean skipped = false;
            Throwable skippedX = null;
            LastExecution lastExecution = null;
            long startTime = 0, stopTime = 0;
            Object result = null;

            Map<String, String> execProps = persistentExecutor.getExecutionProperties(task);
            ThreadContextDescriptor threadContext = info.deserializeThreadContext(execProps);
            ArrayList<ThreadContext> contextAppliedToThread = threadContext == null ? null : threadContext.taskStarting();
            try {
                if (trigger != null) {
                    Long prevScheduledStart = taskRecord.getPreviousScheduledStartTime();
                    if (prevScheduledStart != null)
                        lastExecution = new LastExecutionImpl(persistentExecutor, taskId, taskName, resultBytes, taskRecord.getPreviousStopTime(), taskRecord.getPreviousStartTime(), prevScheduledStart, loader);
                    try {
                        skipped = trigger.skipRun(lastExecution, new Date(taskRecord.getNextExecutionTime()));
                    } catch (RuntimeException x) {
                        skipped = true;
                        skippedX = x;
                    }
                }

                if (skipped) {
                    if (trace && tc.isDebugEnabled())
                        Tr.debug(this, tc, "skipping task", skippedX);
                    Date nextExecDate = trigger.getNextRunTime(lastExecution, new Date(taskRecord.getOriginalSubmitTime()));
                    nextExecTime = nextExecDate == null ? null : nextExecDate.getTime();
                } else {
                    // Fixed result for one-shot runnable
                    if (!info.isSubmittedAsCallable() && info.getInterval() == -1 && trigger == null && resultBytes != null)
                        result = persistentExecutor.deserialize(resultBytes, loader);

                    if (trace && tc.isDebugEnabled())
                        Tr.debug(this, tc, "task about to start " + task);

                    startTime = new Date().getTime();
                    try {
                        if (info.isSubmittedAsCallable())
                            result = ((Callable<?>) task).call();
                        else
                            ((Runnable) task).run();

                        if (trace && tc.isDebugEnabled())
                            Tr.debug(this, tc, "task result " + result);
                        nextFailureCount = (short) 0;
                    } catch (Throwable x) {
                        if (trace && tc.isDebugEnabled())
                            Tr.debug(this, tc, "task failed", x);
                        failure = x;
                        nextFailureCount = (short) ((prevFailureCount < Short.MAX_VALUE) ? (prevFailureCount + 1) : Short.MAX_VALUE);
                        // If we will retry, immediately roll back, then can update persistent store with failure count
                        config = persistentExecutor.configRef.get();
                        if (config.retryLimit == -1 || nextFailureCount <= config.retryLimit)
                            throw failure;
                    } finally {
                        stopTime = new Date().getTime();
                    }

                    // Compute next execution time if the task did not fail.
                    if (failure == null) {
                        long interval = info.getInterval();
                        if (interval == -1) {
                            if (trigger == null) {
                                nextExecTime = null; // one-shot task
                            } else {
                                lastExecution = new LastExecutionImpl(persistentExecutor, taskId, taskName, result, stopTime, startTime, taskRecord.getNextExecutionTime());
                                Date nextExecDate = trigger.getNextRunTime(lastExecution, new Date(taskRecord.getOriginalSubmitTime()));
                                nextExecTime = nextExecDate == null ? null : nextExecDate.getTime();
                            }
                        } else if (info.isFixedRate()) {
                            long originalScheduledStartTime = taskRecord.getOriginalSubmitTime() + info.getInitialDelay();
                            long elapsed = stopTime - originalScheduledStartTime;
                            nextExecTime = (elapsed / interval + 1) * interval + originalScheduledStartTime;
                        } else
                            nextExecTime = stopTime + interval; // fixed-delay
                    }
                }
            } finally {
                if (contextAppliedToThread != null)
                    threadContext.taskStopping(contextAppliedToThread);
            }

            short autoPurgeBit = failure == null ? TaskRecord.Flags.AUTO_PURGE_ON_SUCCESS.bit : TaskRecord.Flags.AUTO_PURGE_ALWAYS.bit;

            if ((nextExecTime == null || !skipped && nextFailureCount > 0) && (binaryFlags & autoPurgeBit) != 0) {
                // Autopurge the completed task if it hasn't already been ended (removal/cancellation by self or other)
                taskStore.remove(taskId, null, false);
            } else {
                // Update state
                TaskRecord updates = new TaskRecord(false);
                if (nextExecTime != null)
                    updates.setNextExecutionTime(nextExecTime);
                short state = nextExecTime == null ? (short) (TaskState.ENDED.bit | TaskState.SUCCESSFUL.bit) : TaskState.SCHEDULED.bit;
                if (skipped) {
                    state |= TaskState.SKIPPED.bit;
                    if (skippedX != null) {
                        state |= TaskState.SKIPRUN_FAILED.bit;
                        byte[] previousResultBytes = persistentExecutor.serialize(lastExecution == null ? null : lastExecution.getResult());
                        updates.setResult(persistentExecutor.serialize(new TaskSkipped(previousResultBytes, skippedX, loader, persistentExecutor)));
                    }
                } else {
                    updates.setConsecutiveFailureCount(nextFailureCount);
                    updates.setPreviousScheduledStartTime(taskRecord.getNextExecutionTime());
                    updates.setPreviousStartTime(startTime);
                    updates.setPreviousStopTime(stopTime);
                    if (failure == null) {
                        // Only update result blob if it changed
                        byte[] updatedResultBytes = result == null ? null : serializeResult(result, loader);
                        if (updatedResultBytes == null || !Arrays.equals(resultBytes, updatedResultBytes))
                            updates.setResult(updatedResultBytes);
                    } else {
                        updates.setResult(persistentExecutor.serialize(new TaskFailure(failure, loader, persistentExecutor, TaskFailure.FAILURE_LIMIT_REACHED, Short.toString(nextFailureCount))));
                        state = (short) (TaskState.ENDED.bit | TaskState.FAILURE_LIMIT_REACHED.bit);
                    }
                }
                updates.setState(state);
                // Only update blobs if they have changed
                if (taskBytes != null) {
                    byte[] updatedTaskBytes = persistentExecutor.serialize(task);
                    if (!Arrays.equals(taskBytes, updatedTaskBytes))
                        updates.setTask(updatedTaskBytes);
                }
                if (triggerBytes != null) {
                    byte[] updatedTriggerBytes = persistentExecutor.serialize(trigger);
                    if (!Arrays.equals(triggerBytes, updatedTriggerBytes))
                        updates.setTrigger(updatedTriggerBytes);
                }
                TaskRecord expected = new TaskRecord(false);
                expected.setId(taskId);
                expected.setVersion(taskRecord.getVersion());
                boolean updatesPersisted = taskStore.persist(updates, expected);

                if (!updatesPersisted) {
                    // Optimistic update unsuccessful. Need to take into account changes made by the task to itself.
                    TaskRecord taskRecordRefresh = taskStore.findById(taskId, null, false);
                    if (taskRecordRefresh != null) {
                        short refreshedState = taskRecordRefresh.getState();
                        if ((refreshedState & TaskState.CANCELED.bit) != 0
                            && (binaryFlags & TaskRecord.Flags.SUSPEND_TRAN_OF_EXECUTOR_THREAD.bit) == 0) {
                            // task canceled itself, combine the canceled state with other updates
                            updates.setState(state = refreshedState);
                            expected.setVersion(taskRecordRefresh.getVersion());
                            taskStore.persist(updates, expected);
                        } else if ((refreshedState & TaskState.SUSPENDED.bit) != 0) {
                            // task suspended itself, merge the suspended state with new state and combine with other updates
                            state |= TaskState.SUSPENDED.bit;
                            state &= ~TaskState.SCHEDULED.bit;
                            expected.setVersion(taskRecordRefresh.getVersion());
                            taskStore.persist(updates, expected);
                        }
                    } // else the task removed itself
                }
            }
        } catch (Throwable x) {
            if (trace && tc.isDebugEnabled())
                Tr.debug(this, tc, "marking transaction to roll back in response to error", x);
            try {
                tranMgr.setRollbackOnly();
            } catch (Throwable t) {
            }
            if (failure == null)
                failure = x;
        } finally {
            if (ejbSingletonLockCollaborator != null)
                ejbSingletonLockCollaborator.unlock();

            taskIdsOfRunningTasks.remove();

            try {
                tranMgr.setTransactionTimeout(0); // clear the value so we don't impact subsequent transactions on this thread

                if (tranMgr.getStatus() == Status.STATUS_MARKED_ROLLBACK) {
                    if (trace && tc.isEventEnabled())
                        Tr.event(this, tc, "rolling back task execution attempt");
                    if (nextFailureCount == null || nextFailureCount == 0)
                        nextFailureCount = (short) (prevFailureCount == null ? 1 : prevFailureCount < Short.MAX_VALUE ? (prevFailureCount + 1) : Short.MAX_VALUE);
                    tranMgr.rollback();
                    if (config == null)
                        config = persistentExecutor.configRef.get();
                    processRetryableTaskFailure(failure, loader, nextFailureCount, config, taskName);
                } else {
                    if (taskIdForPropTable != null)
                        taskStore.removeProperty(taskIdForPropTable);

                    tranMgr.commit();

                    // Immediately reschedule tasks that should run in the near future if the transaction commits
                    config = persistentExecutor.configRef.get();
                    if (config.enableTaskExecution && nextExecTime != null
                        && (config.pollInterval < 0 || nextExecTime <= new Date().getTime() + config.pollInterval)) {

                        expectedExecTime = nextExecTime;
                        long delay = nextExecTime - new Date().getTime();

                        ScheduledExecutorService executor = persistentExecutor.scheduledExecutor;
                        if (executor == null) {
                            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                                Tr.debug(this, tc, "deactivated - reschedule skipped");
                        } else
                            executor.schedule(this, delay, TimeUnit.MILLISECONDS);
                    } else if (ownerForDeferredTask != null)
                        appTracker.deferTask(this, ownerForDeferredTask, persistentExecutor);
                    else {
                        persistentExecutor.inMemoryTaskIds.remove(taskId);
                        if (failure != null) {
                            taskName = taskName == null || taskName.length() == 0 || taskName.length() == 1 && taskName.charAt(0) == ' ' ? String.valueOf(taskId) // empty task name
                                            : taskId + " (" + taskName + ")";
                            Tr.warning(tc, "CWWKC1511.retry.limit.reached.failed", persistentExecutor.name, taskName, nextFailureCount, failure);
                        }
                    }
                }
            } catch (Throwable x) {
                if (failure != null)
                    failure = x;

                // Retry the task if an error occurred
                processRetryableTaskFailure(failure, loader, nextFailureCount, config, taskName);
            }
        }

        if (trace && tc.isEntryEnabled())
            Tr.exit(this, tc, "run[" + taskId + ']', failure);
    }
1141605
public static void runIntoJAT(final JRebirthRunnable runnable) {
        if (Platform.isFxApplicationThread()) {
            // We are into a JAT so just run it synchronously
            runnable.run();
        } else {
            // The runnable will be run into the JAT during the next round
            Platform.runLater(runnable);
        }
    }
818670
public static CompletableFuture<Void> runAsync(Runnable runnable,
            Executor executor) {
        return runAsync(runnable,
                handleExceptionally("runAsync", runnable, executor), executor);
    }
189264
public static void run(final @Nonnull ThrowingRunnable runnable) {
    Objects.requireNonNull(runnable, "runnable must not be null");
    try {
      runnable.run();
    } catch (final RuntimeException exc) {
      throw exc;
    } catch (final Exception exc) {
      throw new RuntimeException(exc);
    }
  }
77653
public final <T> T executeInTransaction(final Runnable<T> runnable) throws Exception {
        return executeInTransaction(runnable, true);
    }
371167
public void doBackgroundOp( final Runnable run, final boolean showWaitCursor )
  {
    final Component[] key = new Component[1];
    ExecutorService jobRunner = getJobRunner();
    if( jobRunner != null )
    {
      jobRunner.submit( () -> performBackgroundOp( run, key, showWaitCursor ) );
    }
    else
    {
      run.run();
    }
  }
1109701
public void execute(Runnable statements)
	{
		try (HibernateTransaction tx = start().withAutoRollback())
		{
			statements.run();

			// Success, perform a TX commit
			tx.commit();
		}
	}
921145
public <T extends Runnable> T executeAsync(T runnable) {
		synchronized (this.tasks) {
			this.tasks.add(runnable);
			// Notify that a task has been added
			this.tasks.notifyAll();

			if (this.tasks.size() == 1) {
				this.onTaskBecameNotEmpty();
			}

			this.onTaskAdded();
		}
		return runnable;
	}
334478
public void excute(Runnable runnable) {
        ExecutorService executorService = getExecutorService();
        if (executorService != null) {
            // 优先使用线程池，提高效率
            executorService.execute(runnable);
        } else {
            // 线程池获取失败，则直接使用线程
            new Thread(runnable).start();
        }
    }