drools。drools

在六月，我们在博客上发布了一个新的内部状态机，用于管理用户，计时器和引擎线程之间的交互。 现在，我们对该代码进行了另一次大的内部清理，以使其更易于阅读和理解。

如前所述，所有操作（插入，更新，删除等）现在都放入线程安全传播队列中。 执行这些操作时，用户线程再也不会接触引擎，甚至Alpha网络也不会。 这样可以提高线程安全性。 相反，当引擎启动时，它首先排空并评估此队列，这可能会导致Alpha网络评估，然后再进行规则评估和触发。

除了将用户和引擎线程分开之外，状态机的另一个目标是协调Timer线程。 当计时器启动时，发动机可能不起作用或正在运行。 如果引擎处于活动状态，则计时器应只向传播队列提交一个条目，然后让当前执行的线程处理该作业。 如果引擎未处于活动状态并且计时器规则是异步的，则计时器线程应通过executeTask方法来进行评估和触发。 状态机旨在最小化同步和锁，以使争用最小。

引擎现在可以处于5种可能的状态。INACTIVE是启动状态。

引擎评估和规则触发具有三个潜在的入口点fireAllRules，fireUntilHalt和异步计时器规则-后者通过executeTask部分完成。 我们将fireAllRules和fireUntilHalt统一到单个fireLoop方法中，该方法使用作为参数传递的策略类来处理循环的潜在休息状态。 当没有规则触发，没有更多要评估的议程组以及队列为空时，引擎将被视为处于静止状态。

然后，FireAllRules所有规则会将引擎设置为INACTIVE，然后退出循环。 FireUntilHalt将使当前线程等待，直到更多工作进入队列进行处理。 这里已经进行了工作，以确保在这些状态转换期间没有间隙和执行损失。

当线程想要转换为FIRE_ALL_RULES或FIRE_UNTIL_HALT或EXECUTE_TASK时，它必须通过waitAndEnterExecutionState。 如果引擎为非活动状态，则可以立即转换，否则，将进入等待状态，直到当前执行线程完成并将引擎返回至非活动状态：

private void waitAndEnterExecutionState( ExecutionState newState ) {if (currentState != ExecutionState.INACTIVE) {try {stateMachineLock.wait();} catch (InterruptedException e) {throw new RuntimeException( e );}}setCurrentState( newState );
}

让我们看看fireAllRules（）如何使用它。 首先请注意，如果引擎已经在运行，因为先前已调用fireAllRules或fireUntilHalt并仍在运行，则它将退出。 第二个注意事项是，它仅将同步点保持足够长的时间，以退出或进行所需的过渡。 一旦引擎进入FIRE_ALL_RULES状态，它就可以释放同步块，并且状态机将停止任何干扰。

public int fireAllRules(AgendaFilter agendaFilter,int fireLimit) {synchronized (stateMachineLock) {if (currentState.isFiring()) {return 0;}waitAndEnterExecutionState( ExecutionState.FIRING_ALL_RULES );}int fireCount = fireLoop(agendaFilter, fireLimit, RestHandler.FIRE_ALL_RULES);return fireCount;
}

fireLoop现在是通用的，并且由fireAllRules和fireUntilHalt共同使用，并使用RestHandler策略来处理引擎到达静止点时的逻辑。

private int fireLoop(AgendaFilter agendaFilter,int fireLimit,RestHandler restHandler) {// The engine comes to potential rest (inside the loop) when there are no propagations and no rule firings.        // It's potentially at rest, because we cannot guarantee it is at rest.        // This is because external async actions (timer rules) can populate the queue that must be executed immediately.        // A final takeAll within the sync point determines if it can safely come to rest.        // if takeAll returns null, the engine is now safely at rest. If it returns something        // the engine is not at rest and the loop continues.        //        // When FireUntilHalt comes to a safe rest, the thread is put into a wait state,        // when the queue is populated the thread is notified and the loop begins again.        //        // When FireAllRules comes to a safe rest it will put the engine into an INACTIVE state        // and the loop can exit.        //        // When a halt() command is added to the propagation queue and that queue is flushed        // the engine is put into a HALTING state. At this point isFiring returns false and        // no more rules can fire and the loop exits.int fireCount = 0;try {PropagationEntry head = workingMemory.takeAllPropagations();int returnedFireCount = 0;boolean limitReached = fireLimit == 0; // -1 or > 0 will return false. No reason for user to give 0, just handled for completeness.boolean loop = true;while ( isFiring()  )  {if ( head != null ) {// it is possible that there are no action propagations, but there are rules to fire.                this.workingMemory.flushPropagations(head);head = null;}// a halt may have occurred during the flushPropagations,            // which changes the isFiring state. So a second isFiring guard is needed            if (!isFiring()) {break;}evaluateEagerList();InternalAgendaGroup group = getNextFocus();if ( group != null && !limitReached ) {// only fire rules while the limit has not reached.returnedFireCount = fireNextItem( agendaFilter, fireCount, fireLimit, group );fireCount += returnedFireCount;limitReached = ( fireLimit > 0 && fireCount >= fireLimit );head = workingMemory.takeAllPropagations();} else {returnedFireCount = 0; // no rules fired this iteration, so we know this is 0                group = null; // set the group to null in case the fire limit has been reached            }if ( returnedFireCount == 0 && head == null && ( group == null || !group.isAutoDeactivate() ) ) {// if true, the engine is now considered potentially at rest                head = restHandler.handleRest( workingMemory, this );}}if ( this.focusStack.size() == 1 && getMainAgendaGroup().isEmpty() ) {// the root MAIN agenda group is empty, reset active to false, so it can receive more activations.            getMainAgendaGroup().setActive( false );}} finally {// makes sure the engine is inactive, if an exception is thrown.        // if it safely returns, then the engine should already be inactive// it also notifies the state machine, so that another thread can take over        immediateHalt();}return fireCount;
}

fire循环在执行takeAll（）时会经过单个同步点，这是返回当前head实例的简单操作，同时还使成员head字段为空，从而使队列为空。 在此takeAll（）期间，这意味着任何用户或计时器操作都必须等待同步释放后才能添加到队列中。 在执行完其余方法之后，可以执行评估返回的项目列表以及评估网络和触发规则的过程，而无需进行另一个同步或锁定。

其余处理程序都是两个非常简单的代码段：

interface RestHandler {RestHandler FIRE_ALL_RULES = new FireAllRulesRestHandler();RestHandler FIRE_UNTIL_HALT = new FireUntilHaltRestHandler();PropagationEntry handleRest(InternalWorkingMemory wm, DefaultAgenda agenda);class FireAllRulesRestHandler implements RestHandler {@Override        public PropagationEntry handleRest(InternalWorkingMemory wm, DefaultAgenda agenda) {synchronized (agenda.stateMachineLock) {PropagationEntry head = wm.takeAllPropagations();if (head == null) {agenda.halt();}return head;}}}class FireUntilHaltRestHandler  implements RestHandler {@Override        public PropagationEntry handleRest(InternalWorkingMemory wm, DefaultAgenda agenda) {return wm.handleRestOnFireUntilHalt( agenda.currentState );}}
}@Overridepublic PropagationEntry handleRestOnFireUntilHalt(DefaultAgenda.ExecutionState currentState) {// this must use the same sync target as takeAllPropagations, to ensure this entire block is atomic, up to the point of wait    synchronized (propagationList) {PropagationEntry head = takeAllPropagations();// if halt() has called, the thread should not be put into a wait state        // instead this is just a safe way to make sure the queue is flushed before exiting the loop        if (head == null && currentState == DefaultAgenda.ExecutionState.FIRING_UNTIL_HALT) {propagationList.waitOnRest();head = takeAllPropagations();}return head;}
}

请注意，FireAllRulesRestHandler必须在执行最后的takeAll时获取stateMachineLock，然后才能知道它真正安全地返回。 这是由于可能放置在队列中的计时器需要立即触发。 如果要返回引擎，计时器将不会立即触发-这就是我们所说的行为“差距”，现在可以避免。

FireUntilHalt锁定了传播队列，因为除了执行takeAll之外，它还必须原子地执行null检查和等待操作。 再一次，如果空检查不在同步点之内，我们将在行为上产生另一个潜在的差距，现在可以避免这种情况。

难题的最后一部分是executeTask。 这允许以最佳方式发生异步操作（通常是计时器任务）。 如果引擎由于FireAllRules或FireUntilHalt而已在运行，则只需将任务提交到队列中，并让当前运行的线程处理该任务。 如果不是，则进入EXECUTING_TASK状态并在当前线程中执行它。

@Overridepublic void executeTask( ExecutableEntry executable ) {synchronized (stateMachineLock) {// state is never changed outside of a sync block, so this is safe.        if (isFiring()) {executable.enqueue();return;} else if (currentState != ExecutionState.EXECUTING_TASK) {waitAndEnterExecutionState( ExecutionState.EXECUTING_TASK );}}try {executable.execute();} finally {immediateHalt();}
}

我应该补充说，halt（）现在作为命令提交，并作为标准队列消耗的一部分进行评估。 执行时，它将在同步块内将引擎更改为暂停状态。 这将允许外部循环退出：

public void halt() {synchronized (stateMachineLock) {if (currentState.isFiring()) {setCurrentState( ExecutionState.HALTING );}}
}

因此，我们现在有了非常健壮的代码，可以以可理解的方式处理用户，计时器和引擎线程的交互。 我们在清理中付出了很多努力，以便希望每个人都能理解代码和行为。

发动机的最后一部分仍然被认为是不安全的。 在引擎运行时，用户可以在一个线程上调用插入事实的设置方法。 这显然可以流下眼泪。 我们计划允许用户将任务提交到此队列，以便可以使用与正在运行的引擎相同的线程来执行任务。 这将允许用户从引擎外部的另一个线程提交pojo更新作为任务来安全执行。

翻译自: https://www.javacodegeeks.com/2015/12/drools-detailed-description-internal-code-cleanups-fireallrules-fireuntilhalt-timers.html

drools。drools