如何在不编写代码的情况下测试思想，体系结构和算法？ 如何制定和验证其性质？ 什么是模型检查器和模型查找器？ 测试能力不足时该怎么办？
你好 我叫Vasil Dyadov，现在我是Yandex.Mail的程序员，在加入英特尔之前，我曾在Verilog / VHDL上为ASIC / FPGA开发过RTL代码（寄存器传输级别）。 我一直很喜欢软件和硬件可靠性，用于开发具有保证的预定义属性的软件和逻辑的数学，工具和方法的主题。
这是该系列的第二篇文章（ 此处是第一篇文章），旨在引起开发人员和管理人员对软件开发的工程方法的关注。 最近，尽管他的方法和支持工具发生了革命性的变化，但他仍然理应受到忽视。
在某些读者看来，第一篇文章太抽象了。 许多人希望看到在接近现实的条件下使用工程方法和正式规格的示例。
在本文中，我们将看一个TLA +解决实际问题的实际应用示例。
我总是乐于讨论与软件开发相关的问题，并且很高兴与读者聊天（交流的坐标在我的个人资料中）。
什么是TLA +？
首先，我想谈谈TLA +和TLC。
TLA +（动作的时间逻辑+数据）是一种基于时间逻辑的形式主义。 由Leslie Lamport设计。
在这种形式主义的框架内，可以描述系统行为变体的空间以及这些行为的属性。
为简单起见，我们可以假设系统的行为由其状态序列表示（例如，无限的珠子，字符串上的球），并且TLA +公式定义了一类链，描述了系统行为的所有可能变体（大量的珠子）。
TLA +非常适合描述相互作用的不确定性有限状态机（例如，系统中服务的相互作用），尽管它的表达能力足以描述许多其他事物（可以用一阶逻辑表示）。
TLC是显式状态模型检查器：根据给定的TLA +系统描述和属性公式，该程序循环访问系统状态并确定系统是否满足指定的属性。
通常，以这种方式构造TLA + / TLC：在TLA +中描述系统，在TLA +中形式化有趣的属性，运行TLC进行验证。
由于在TLA +中直接描述一个或多或少复杂的系统并不容易，因此发明了一种高级语言-PlusCal，可将其翻译为TLA +。 PlusCal以两种方式存在：具有Pascal和类似C的语法。 在我使用类似Pascal的语法的文章中，对我来说似乎更好阅读。 关于TLA +的PlusCal与关于汇编程序的C大致相同。
在这里，我们将不深入理论。 文章末尾提供了浸入TLA + / PlusCal / TLC的文献。
我的主要任务是在一个简单易懂的真实示例中展示TLA + / TLC的应用。
在对前一篇文章的一些评论中，我被指责为我没有画出工具的理论基础，但是本系列文章的目的是展示工具在软件开发中对工程方法的实际应用。
我认为对理论的深入了解对任何人都没有什么兴趣，但是如果您有兴趣，可以随时向项目经理寻求链接和解释，并且据我所知（毕竟，我不是理论数学家，而是软件工程师），我将尽力回答。
问题陈述
首先，我将介绍使用TLA +的任务。
该任务与处理事件流有关。 即，创建一个队列来存储事件并发送有关这些事件的通知。
数据仓库在物理上基于PostgreSQL DBMS进行组织。
您需要了解的主要内容：
- 有事件的来源。 出于我们的目的，我们可以将自己限制在以下事实：每个事件的特征在于计划处理的时间。 这些源将事件写入数据库。 通常，写入数据库的时间与计划处理的时间没有任何关系。
- 有一些协调过程可以从数据库中读取事件，并将即将发生的事件的通知发送到必须响应这些通知的系统组件。
- 基本要求：我们绝不能输掉大事。 在极端情况下，事件的通知可以重复，即必须至少保证一次 。 在分布式系统中，在没有共识机制的情况下仅一次执行保证是非常困难的（甚至可能是不可能的，但是需要证明），并且就延迟和吞吐量而言，它们（至少我所知道的所有）对系统有非常重要的影响。
现在一些细节：
- 源流程有很多；它们可以生成数百万个（在最坏的情况下）进入狭窄时间间隔的事件。
- 事件既可以在将来也可以在过去的时间生成（例如，如果源进程放慢了速度并记录了已经过了片刻的事件）。
- 事件处理的优先级是及时的，即我们必须首先处理最早的事件。
- 对于每个事件，源进程都会生成一个随机数worker_id ，由于该事件会在协调器之间分配事件。
- 有几个协调过程（根据需要根据系统负载进行缩放）。
- 每个协调器进程都为其自己的set worker_id处理事件，即由于worker_id，我们避免了协调器之间的竞争和对锁的需求。
从描述中可以看出，我们只能考虑一个协调过程，而在我们的任务中不考虑worker_id 。
也就是说，为简单起见，我们假定：
- 源流程很多。
- 协调过程是其中之一。
我将分阶段描述解决此问题的思想的演变，以便更清楚地了解该思想如何从简单的实现演变为优化的实现。
前额决定
我们将为事件创建一个标牌，在其中将以时间戳的形式存储事件（我们对该任务中的其他参数不感兴趣）。 让我们在时间戳字段上建立索引。
这似乎是一个完全正常的解决方案。
可伸缩性只有一个问题：事件越多，数据库操作就越慢。
活动可能会持续过去，因此协调员将不得不不断查看整个时间表。
通过按时间将数据库拆分为碎片，可以广泛解决该问题。但这是一种资源密集型的方法。
结果，由于您将不得不读取和合并来自多个数据库的数据，因此协调员的工作将会变慢。
很难在协调器中实现事件缓存，以便不去基地处理每个事件。
更多数据库-更多容错问题。
依此类推。
我们不会详细介绍这种正面解决方案，因为它是琐碎且无趣的。
第一次优化
让我们看看如何改善正面解决方案。
要优化对数据库的访问，可以使索引稍微复杂一点，向在提交数据库中的事务时将生成的事件添加单调递增的标识符。 也就是说，该事件现在以{time，id}对为特征，其中time是安排该事件的时间， id是一个单调递增的计数器。 可以保证每个事件的id都是唯一的，但不能保证id值没有漏洞（也就是说，可以有这样的顺序： 1，2，7，15 ）。
似乎现在我们可以在协调器进程中存储最后一个读取事件的标识符，并在获取时选择标识符大于最后一个处理事件的事件。
但是，这里的问题立即出现：源进程可以记录一个带有将来时间戳的事件。 然后，我们将不得不在协调过程中不断考虑带有小标识符的事件集，而这些事件的处理时间尚未到来。
您会注意到，与当前时间相关的事件分为两类：
- 具有过去时间戳记但标识符较大的事件。 在处理了该时间间隔后，它们最近被写入数据库。 这些是高优先级事件，需要首先进行处理，以使通知（已经晚了）不晚。
- 曾经记录过的事件带有接近当前时刻的时间戳。 这样的事件将具有较低的标识符值。
因此，协调器进程的当前状态以{state.time，state.id}对为特征。
事实证明，高优先级事件在该点的左侧和上方（粉红色区域），而正常事件在右侧（浅蓝色）：
流程图
协调器工作算法如下：
在研究算法时，可能会出现以下问题：
- 如果正常事件的处理开始并且在那一刻过去的新事件到达时（在粉红色的区域）怎么办，它们不会丢失？ 答：它们将在下一个处理高优先级事件的周期中进行处理。 他们不会迷路，因为他们的ID被保证高于state.id。
- 如果在处理了所有正常事件之后-在切换到处理高优先级事件时-具有间隔[state.time，state.time + Delta]的时间戳的新事件到达了，我们会丢失它们吗？ 答案：它们将进入粉红色区域，因为它们的时间 <state.time and id > state.id：它们是最近到达的，并且id单调增加。
算法运算示例
让我们看一下算法的几个步骤：
型号
我们将确保算法不会丢失事件，并且将发送所有通知：我们将组成一个简单的模型并进行验证。
对于模型，我们使用TLA +，更确切地说是PlusCal，它可以转换为TLA +。
---------------- MODULE events ---------------- EXTENDS Naturals, FiniteSets, Sequences, TLC (* --algorithm Events \* \* (by Daniel Jackson) \* small-scope hypothesis, \* , ́ \* \* \* : \* Events - - , \* [time, id], \* \* \* \* Event_Id - \* id \* MAX_TIME - , \* \* TIME_DELTA - Delta, \* \* variables Events = {}, Event_Id = 0, MAX_TIME = 5, TIME_DELTA \in 1..3 define \* \* ZERO_EVT == [time |-> 0, id |-> 0] MAX(S) == CHOOSE x \in S: \A y \in S: y <= x MIN(S) == CHOOSE x \in S: \A y \in S: y >= x \* fold_left/fold_right RECURSIVE SetReduce(_, _, _) SetReduce(Op(_, _), S, value) == IF S = {} THEN value ELSE LET s == CHOOSE s \in S : TRUE IN SetReduce(Op, S \ {s}, Op(s, value)) (* ( ) *) ToSeq(S) == LET op(e, val) == Append(val, e) IN SetReduce(op, S, << >>) (* : *) ToSet(S) == {S[i] : i \in DOMAIN(S)} (* map *) MapSet(Op(_), S) == {Op(x) : x \in S} (* *) \* id GetIds(Evts) == MapSet(LAMBDA x: x.id, Evts) \* time GetTimes(Evts) == MapSet(LAMBDA x: x.time, Evts) (* SQL *) \* \* ORDER BY EventsOrderByTime(e1, e2) == e1.time < e2.time EventsOrderById(e1, e2) == e1.id < e2.id EventsOrder(e1, e2) == \* time, id \/ EventsOrderByTime(e1, e2) \/ /\ e1.time = e2.time /\ EventsOrderById(e1, e2) \* SELECT * FROM events \* WHERE time <= curr_time AND id >= max_id \* ORDER BY time, id SELECT_HIGH_PRIO(state) == LET \* \* time <= curr_time \* AND id >= maxt_id selected == {e \in Events : /\ e.time <= state.time /\ e.id >= state.id } IN selected \* SELECT * FROM events \* WHERE time > current_time AND time - Time <= delta_time \* ORDER BY time, id SELECT_NORMAL(state, delta_time) == LET selected == {e \in Events : /\ e.time <= state.time + delta_time /\ e.time > state.time } IN selected \* Safety predicate \* ALL_EVENTS_PROCESSED(state) == \A e \in Events: \/ e.time >= state.time \/ e.id >= state.id end define; \* - fair process inserter = "Sources" variable n, t; begin forever: while TRUE do \* get_time: \* \* , , \* with evt_time \in 0..MAX_TIME do t := evt_time; end with; \* ; \* : \* 1. . \* 2. , \* Event_Id - , \* commit: \* either - either Events := Events \cup {[time |-> t, id |-> Event_Id]} || Event_Id := Event_Id + 1; or Event_Id := Event_Id + 1; end either; end while; end process fair process coordinator = "Coordinator" variable state = ZERO_EVT, events = {}; begin forever: while TRUE do \* high_prio: events := SELECT_HIGH_PRIO(state); \* process_high_prio: \* , \* Events, \* state.id := MAX({state.id} \union GetIds(events)) || \* , \* Events := Events \ events || \* events , \* events := {}; \* - normal: events := SELECT_NORMAL(state, TIME_DELTA); process_normal: state.time := MAX({state.time} \union GetTimes(events)) || Events := Events \ events || events := {}; end while; end process end algorithm; *) \* BEGIN TRANSLATION \* TLA+, PlusCal \* END TRANSLATION ================================
如您所见，尽管定义（定义）部分相当多，但是描述相对较小，可以将其在单独的模块中取出，然后再使用。
在评论中，我试图解释模型中正在发生的事情。 希望这个
我管理了，不需要更详细地绘制模型。
我只想澄清关于状态和建模特征之间转换的原子性的一点。
通过执行过程的原子步骤进行建模。 在一个过渡中，执行过程的一个原子步骤，在该步骤中可以完成此步骤。 步骤和过程的选择是不确定的：在建模过程中，将对所有过程的原子步骤的所有可能链进行排序。
可能会出现问题：当我们在不同的过程中同时执行几个原子步骤时，如何对真正的并行性建模？
Leslie Lamport在“指定系统和其他著作”一书中早就回答了这个问题。
我不会完整地给出答案，简而言之，要点是：如果没有将每个事件与特定时刻相关联的确切时间范围，那么将并行事件建模为非确定性发生的顺序事件就没有区别，因为我们始终可以假设一个事件比另一个事件更早发生无穷小值。
但是真正重要的是原子步的合理分配。 如果它们太多，则会发生状态空间的组合爆炸。 如果您采取的步骤少于必要的步骤，或者选择了错误的步骤-也就是说，错过无效状态/转换的可能性（即，我们将错过模型上的错误）。
为了正确地将流程分解为原子步骤，您需要根据流程对数据和同步机制的依赖性来很好地了解系统的工作方式。
通常，将过程拆分为原子步骤不会引起大问题。 如果确实如此，那么它表明存在对该问题的缺乏了解，而不是有关编译模型和编写TLA +规范的问题。 这是正式规范的另一个非常有用的功能：它们需要彻底的研究和分析。
一个问题。 通常，如果任务有意义并且被很好地理解，则其形式化就不会有问题。
模型检查
对于建模，我将使用TLA工具箱。 您当然可以从命令行运行所有内容，但是IDE仍然更加方便，尤其是开始学习使用TLA +进行建模时。
该项目的创建在手册，文章和书籍中都有很好的描述，我在文章末尾引用了这些链接，因此，我不再赘述。 我唯一引起您注意的是模拟设置。
TLC是具有显式状态检查的模型检查器 。 显然，状态空间必须受到合理的限制。 一方面，它应该足够大，以便能够验证我们感兴趣的属性，另一方面，应该足够小，以便使用可接受的资源在合理的时间内完成仿真。
这是一个非常微妙的问题，在这里您需要了解系统和模型的属性。 但是它很快就会带来经验。 对于初学者，您可以简单地设置最大可能的限制，这些限制在模拟时间和资源消耗方面仍然可以接受。
还有一种检查模式不是检查整个状态空间，而是检查到一定深度的选择性链。 有时也可能需要使用。
我们返回到模拟设置。
首先，我们定义对系统状态空间的限制。 在“ 高级选项/状态约束模拟设置”部分中设置了限制 。
在那里，我指出了一个TLA +表达式： Cardinality(Events) <= 5 /\ Event_Id <= 5 ，
其中Event_Id是事件标识符的值的上限， Cardinality(Events)是事件记录集的大小（限制了基本模型）
板中的五个记录来存储数据）。
在仿真中，TLC将仅查看该公式为真的状态。
您仍然可以允许有效的状态转换（“ 高级选项” /“操作约束” ），
但我们不需要它。
接下来，我们指示描述系统的TLA +公式： Model Overview / Temporal Formula = Spec ，其中Spec是PlusCal自动生成的TLA +公式的名称（在上面的模型代码中不是：为了节省空间，我没有引用将PlusCal转换为TLA +的结果） 。
值得关注的下一个设置是死锁检查。
（在Model Overview / Deadlock中选中）。 启用此标志后，TLC将检查模型的“挂起”状态，即没有传出过渡的状态。 如果状态空间中存在此类状态，则意味着模型中存在明显的错误。 或在TLC中，它像其他任何不重要的程序一样，都无法避免错误:)在我的实践中（不是那么大），我还没有遇到死锁。
最后，为了开始所有这些测试，“ 模型概述/不变式”中的安全公式为ALL_EVENTS_PROCESSED(state) 。
TLC将在每个状态下验证该公式的有效性，如果该公式为假，
将显示错误消息并显示导致错误的状态顺序。
启动TLC后，工作大约8分钟后，它报告“无错误”。 这意味着该模型已经过测试并符合指定的属性。
TLC还显示了许多有趣的统计数据。 例如，对于该模型，获得了7677824个唯一状态； TLC总共查看了27109029个状态，状态空间的直径为47（这是重复之前状态链的最大长度，
状态和过渡图中非重复状态的最大循环长度）。
如果我们将2700万个州划分为8分钟，则每秒将获得约5.6万个州，这似乎并不快。 但是请记住，我是在节能模式下的笔记本电脑上运行仿真的（因为当时我在火车上开车，我将核心频率强制为800 MHz），并且根本没有针对仿真速度优化模型。
有很多方法可以加快仿真速度：从将TLA +模型代码的一部分移植到Java并即时连接到TLC（这有助于加速各种帮助程序功能），再到在云和集群上运行TLC（Amazon和Azure云支持内置于TLC本身）。
第二次优化
在以前的算法中，一切都很好，除了一些问题：
- 直到我们在间隔
[state.time, state.time + Delta]处理来自蓝色区域的所有事件之前，我们都无法继续处理高优先级事件。 也就是说，迟到的事件将更晚。 通常，延迟是无法预测的。 因此，state.time可能远远落后于当前时间，这是下一个问题的原因。 - 到达正常事件区域的事件可能会延迟（ id > state.id）。 它们已经是高度优先事项，应该被认为是粉红色区域的事件，我们仍然将其视为正常并将其视为正常。
- 组织事件缓存和缓存补充（从数据库中读取）很困难。
如果前两点很明显，那么第三点可能会提出最多的问题。 让我们更详细地讨论它。
假设我们要先将固定数量的事件读入内存，然后对其进行处理。
处理之后，我们希望将使用块查询的数据库中的事件标记为已处理，因为如果您不使用块而是使用单个事件，那么缓存将不会带来太大的收益。
假设我们已经处理了部分块，并希望补充缓存。 然后，如果在处理过程中迟到的高优先级事件到来，我们可以尽早处理它们。
也就是说，非常希望能够读取小块中的事件，以便尽快处理较晚的事件，但是为了提高效率，一次用大块更新数据库中的处理属性。
在这种情况下该怎么办？
尝试在带有蓝色和粉红色区域的小块中使用数据库，并逐步移动状态点。
因此，引入了高速缓存，并从数据库中的数据片中读取了高速缓存，每次读取之后，状态点都会移动，以便不重新读取已读取的事件。
现在，该算法变得更加复杂了，我们开始有限地阅读。
流程图
在该算法中，可以看出由于对可读事件块的限制，从处理低优先级到处理高优先级的过渡中的最大延迟将等于该块的最大处理时间。
也就是说，现在我们既可以将事件以小块的形式读取到缓存中，也可以通过控制读取的最大块大小来控制过渡到处理高优先级事件的最大延迟。
算法运算示例
让我们分步查看工作中的算法。 为方便起见，取LIMIT = 2 。
原来问题解决了吗？ 但是没有 （很明显，如果问题在此阶段已完全解决，
本文不会是:)）
错误吗？
以这种形式，该算法工作了相当长的时间。 测试一切顺利。 生产也没有问题。
但是该算法及其实现的开发人员（我的同事Peter Reznikov）非常有经验，他凭直觉感到这里有些问题。 因此，在主代码旁边进行了检查，该代码在计时器上每隔一段时间检查一次，以查看是否有任何遗漏的事件，并且
如果有的话，我处理了它们。
以这种形式，系统成功运行。 的确，没有人保存检查者选择的事件数量的统计信息。 因此，不幸的是，我们不知道与不及时的事件处理相关联的失败有多少次。
我实现了一个类似的限时对象队列。 在与Peter Reznikov讨论算法的实现和优化时，我们讨论了用于事件处理的算法。 他们怀疑该算法是否正确。 我们决定制作一个小模型来确认或消除怀疑。 结果，我们发现了一个错误。
型号
在使用错误分解跟踪之前，我将提供检测到错误的模型的源代码。
与先前模型的差异非常小，读取块的大小只有一个限制：添加了Limit运算符，并因此更改了事件选择运算符。
为了节省空间，我只对模型的更改部分发表了评论。
---------------- MODULE events ---------------- EXTENDS Naturals, FiniteSets, Sequences, TLC (* --algorithm Events \* LIMIT, \* \* \* variables Events = {}, Event_Id = 0, MAX_TIME = 5, LIMIT \in 1..3, TIME_DELTA \in 1..2 define ZERO_EVT == [time |-> 0, id |-> 0] MAX(S) == CHOOSE x \in S: \A y \in S: y <= x MIN(S) == CHOOSE x \in S: \A y \in S: y >= x RECURSIVE SetReduce(_, _, _) SetReduce(Op(_, _), S, value) == IF S = {} THEN value ELSE LET s == CHOOSE s \in S : TRUE IN SetReduce(Op, S \ {s}, Op(s, value)) ToSeq(S) == LET op(e, val) == Append(val, e) IN SetReduce(op, S, << >>) ToSet(S) == {S[i] : i \in DOMAIN(S)} MapSet(Op(_), S) == {Op(x) : x \in S} GetIds(Evts) == MapSet(LAMBDA x: x.id, Evts) GetTimes(Evts) == MapSet(LAMBDA x: x.time, Evts) EventsOrderByTime(e1, e2) == e1.time < e2.time EventsOrderById(e1, e2) == e1.id < e2.id EventsOrder(e1, e2) == \/ EventsOrderByTime(e1, e2) \/ /\ e1.time = e2.time /\ EventsOrderById(e1, e2) Limit(S, limit) == LET amount == MIN({limit, Len(S)}) result == IF amount > 0 THEN SubSeq(S, 1, amount) ELSE << >> IN result \* SELECT * FROM events \* WHERE time <= curr_time AND id > max_id \* ORDER BY id \* LIMIT limit SELECT_HIGH_PRIO(state, limit) == LET selected == {e \in Events : /\ e.time <= state.time /\ e.id >= state.id } \* Id sorted == SortSeq(ToSeq(selected), EventsOrderById) \* limited == Limit(sorted, limit) IN ToSet(limited) \* SELECT * FROM events \* WHERE time > current_time \* AND time - Time <= delta_time \* ORDER BY time, id \* LIMIT limit SELECT_NORMAL(state, delta_time, limit) == LET selected == {e \in Events: /\ e.time <= state.time + delta_time /\ e.time > state.time } \* sorted == SortSeq(ToSeq(selected), EventsOrder) \* limited == Limit(sorted, limit) IN ToSet(limited) ALL_EVENTS_PROCESSED(state) == \A e \in Events: \/ e.time >= state.time \/ e.id >= state.id end define; fair process inserter = "Sources" variable t; begin forever: while TRUE do get_time: with evt_time \in 0..MAX_TIME do t := evt_time; end with; commit: either Events := Events \union {[time |-> t, id |-> Event_Id]} || Event_Id := Event_Id + 1; or Event_Id := Event_Id + 1; end either; end while; end process fair process event_processor = "Event_processor" variable state = ZERO_EVT, events = {}; begin forever: while TRUE do select: events := LET evts_high_prio == SELECT_HIGH_PRIO(state, LIMIT) new_limit == LIMIT - Cardinality(evts_high_prio) evts_normal == SELECT_NORMAL(state, TIME_DELTA, new_limit) IN evts_high_prio \union evts_normal; proc_evts: Events := Events \ events || state := [ time |-> MAX({state.time} \union GetTimes(events)), id |-> MAX({state.id} \union GetIds(events))] || events := {}; end while; end process end algorithm; *) \* BEGIN TRANSLATION \* TLA+, PlusCal \* END TRANSLATION ===================================================
细心的读者可能会注意到，除了引入Limit之外，event_processor中的标签也已更改。 目标是更好地模拟两个选择在一个事务中执行的真实代码，也就是说，事件的选择可以说是原子执行的。
好吧，如果我们在具有较大原子操作的模型中发现错误，则实际上可以保证在相同模型中发生相同错误，但原子步长较小（相当强的陈述，但我认为这很直观；尽管如果没有得到证明，应该会很好，然后在多种模型上进行检查）。
模型检查
我们以与第一实施例相同的参数开始仿真。
在宽度搜索中，我们在模拟的第19步违反了ALL_EVENTS_PROCESSED属性。
对于给定的初始数据（这是一个很小的状态空间），第19步的错误表明该错误非常罕见且难以检测，因为在此之前，所有长度小于19的状态链均已检查。
因此，该错误很难在测试中发现。 仅当您知道要看的地方时，并特别选择测试和临时小屋。
为了节省空间和时间，我不会整条路线。 这是来自几种状态的段以及错误：
分析与校正
怎么了
如您所见，由于限制在事件{2，1}结束而导致我们错失了事件{2，3}，并随后更改了协调器的状态，因此出现了错误。 只有在某个时刻有多个事件时，才可能发生这种情况。
这就是为什么该错误在测试中难以捉摸。 对于它的表现，非常罕见的事情必须重合：
- 几个事件在同一时间点。
- 事件选择的限制在读取所有这些事件之前就已结束。
如果将协调器的状态扩展一点，则可以相对容易地纠正该错误：如果此事件的时间与下一个state.time值相对应，则将来自正常事件区域的最后一个读取事件的时间和标识符添加到具有最大id的位置。
如果没有这样的事件，那么我们将额外状态（extra_state）设置为无效状态（UNDEF_EVT），并且在工作时不将其考虑在内。
正常区域中在协调程序的上一步中未处理的事件，我们将考虑已经具有较高的优先级，因此将更正高优先级和安全谓词的选择。
可以在高优先级和正常之间引入另一个中间区域，并更改算法。 首先，它将处理高优先级的，然后处理中级的，然后继续处理随后的状态更改。
但是，这样的更改将导致大量重构，并带来明显的好处（该算法将更加清晰；其他优势不会马上显现出来）。
因此，我们决定仅稍微调整当前状态和从数据库中选择事件。
调整后的型号
这是校正后的模型。
------------------- MODULE events ------------------- EXTENDS Naturals, FiniteSets, Sequences, TLC \* CONSTANTS MAX_TIME, LIMIT, TIME_DELTA (* --algorithm Events variables Events = {}, Limit \in LIMIT, Delta \in TIME_DELTA, Event_Id = 0 define \* \* , extra_state UNDEF_EVT == [time |-> MAX_TIME + 1, id |-> 0] ZERO_EVT == [time |-> 0, id |-> 0] MAX(S) == CHOOSE x \in S: \A y \in S: y <= x MIN(S) == CHOOSE x \in S: \A y \in S: y >= x RECURSIVE SetReduce(_, _, _) SetReduce(Op(_, _), S, value) == IF S = {} THEN value ELSE LET s == CHOOSE s \in S : TRUE IN SetReduce(Op, S \ {s}, Op(s, value)) ToSeq(S) == LET op(e, val) == Append(val, e) IN SetReduce(op, S, << >>) ToSet(S) == {S[i] : i \in DOMAIN(S)} MapSet(Op(_), S) == {Op(x) : x \in S} GetIds(Evts) == MapSet(LAMBDA x: x.id, Evts) GetTimes(Evts) == MapSet(LAMBDA x: x.time, Evts) EventsOrderByTime(e1, e2) == e1.time < e2.time EventsOrderById(e1, e2) == e1.id < e2.id EventsOrder(e1, e2) == \/ EventsOrderByTime(e1, e2) \/ /\ e1.time = e2.time /\ EventsOrderById(e1, e2) TakeN(S, limit) == LET amount == MIN({limit, Len(S)}) result == IF amount > 0 THEN SubSeq(S, 1, amount) ELSE << >> IN result (* SELECT * FROM events WHERE time <= curr_time AND id > max_id ORDER BY id Limit limit *) SELECT_HIGH_PRIO(state, limit, extra_state) == LET \* \* time <= curr_time \* AND id > maxt_id selected == {e \in Events : \/ /\ e.time <= state.time /\ e.id >= state.id \/ /\ e.time = extra_state.time /\ e.id > extra_state.id} sorted == \* SortSeq(ToSeq(selected), EventsOrderById) limited == TakeN(sorted, limit) IN ToSet(limited) SELECT_NORMAL(state, delta_time, limit) == LET selected == {e \in Events : /\ e.time <= state.time + delta_time /\ e.time > state.time } sorted == SortSeq(ToSeq(selected), EventsOrder) limited == TakeN(sorted, limit) IN ToSet(limited) \* extra_state UpdateExtraState(events, state, extra_state) == LET exact == {evt \in events : evt.time = state.time} IN IF exact # {} THEN CHOOSE evt \in exact : \A e \in exact: e.id <= evt.id ELSE UNDEF_EVT \* extra_state ALL_EVENTS_PROCESSED(state, extra_state) == \A e \in Events: \/ e.time >= state.time \/ e.id > state.id \/ /\ e.time = extra_state.time /\ e.id > extra_state.id end define; fair process inserter = "Sources" variable t; begin forever: while TRUE do get_time: with evt_time \in 0..MAX_TIME do t := evt_time; end with; commit: either Events := Events \union {[time |-> t, id |-> Event_Id]} || Event_Id := Event_Id + 1; or Event_Id := Event_Id + 1; end either; end while; end process fair process event_processor = "Event_processor" variable events = {}, state = ZERO_EVT, extra_state = UNDEF_EVT; begin forever: while TRUE do select: events := LET evts_high_prio == SELECT_HIGH_PRIO(state, Limit, extra_state) new_limit == Limit - Cardinality(evts_high_prio) evts_normal == SELECT_NORMAL(state, Delta, new_limit) IN evts_high_prio \union evts_normal; proc_evts: Events := Events \ events || state := [ time |-> MAX({state.time} \union GetTimes(events)), id |-> MAX({state.id} \union GetIds(events)) ]; extra_state := UpdateExtraState(events, state, extra_state) || events := {}; end while; end process end algorithm; *) \* BEGIN TRANSLATION \* TLA+, PlusCal \* END TRANSLATION ===================================================
如您所见，更改非常小：
- 向Extra_state状态添加了额外的数据。
- 更改了高优先级事件的选择。
- UpdateExtraState extra_state.
- safety - .
, . , (, , , ).
, , TLA+/TLC . :)
结论
, , ( , , , ).
, , TLA+/TLC, . , .
TLA+/TLC , , ( , ) .
, , , TLA+/TLC .
书本
, , , . .
Michael Jackson Problem Frames: Analysing & Structuring Software Development Problems
( !), . , . , .
Hillel Wayne Practical TLA+: Planning Driven Development
TLA+/PlusCal . , . . : .
MODEL CHECKING.
. , , . , .
Leslie Lamport Specifying Systems: The TLA+ Language and Tools for Hardware and Software Engineers
TLA+. , . : , . , TLA+ , .
Formal Development of a Network-Centric RTOS
TLA+ ( RTOS ) TLC.
, . , TLA+ , , RTOS — Virtuoso. , .
, (, , , , ).
w Amazon Web Services Uses Formal Methods
TLA+ AWS. : http://lamport.azurewebsites.net/tla/amazon-excerpt.html
Hillel Wayne ( "Practical TLA+")
. , . , - .
Ron Pressler
. . , TLA+. TLA+, computer science .
Leslie Lamport
TLA+ computer science . TLA+ .
. . , . . , . . .
, , .
TLA+,
, TLA+. , TLA+.
Hillel Wayne
Hillel Wayne . .
Ron Pressler
, Hillel Wayne, . , . Ron Pressler . ́ , , .
TLA toolbox + TLAPS : TLA+
. Alloy Analyzer .