mormot.core.threads.pas unit

Purpose: Framework Core Multi-Threading Support
- this unit is a part of the Open Source Synopse mORMot framework 2, licensed under a MPL/GPL/LGPL three license - see LICENSE.md

目的：框架核心多线程支持

• 本单元是开源Synopse mORMot框架2的一部分，根据MPL/GPL/LGPL三重许可进行许可-参见LICENSE.md

1.1. Units used in the mormot.core.threads unit （在mormot.core.threads单元中使用的单元）

1.2. mormot.core.threads class hierarchy

TSynThread
TSynThreadPoolWorkThread
TNotifiedThreadTLoggedThread
TLoggedWorkThreadTThreadAbstract
TSynBackgroundThreadAbstract
TSynBackgroundThreadProcess
TSynBackgroundTimer
TSynBackgroundThreadMethodAbstract
TSynParallelProcessThread
TSynBackgroundThreadProcedure
TSynBackgroundThreadMethod
TSynBackgroundThreadEventTThread
TSynPersistentStoreTSynQueue
TSynPersistentLockTSynParallelProcess
TSynPersistent
TBlockingProcessPool
TBlockingProcess
TBlockingProcessPoolItem
TSynEventTObject
TSynThreadPool
TPendingTaskList
TInterfacedObjectWithCustomCreate
TLockedDocVariantIInterface
ILockedDocVariant
ESynException
ESynThread

mormot.core.threads class hierarchy

1.3. Objects implemented in the mormot.core.threads unit （mormot.core.threads 单元中实现的对象）

1.3.1. ESynThread

ESynThread = class(ESynException)

1.3.2. TSynQueue

TSynQueue = class(TSynPersistentStore)

• 内部使用TDynArray存储，具有滑动算法，比FPC或Delphi的TQueue或简单的TDynArray.Add/Delete更高效
• 如果需要，支持TSynPersistentStore二进制持久化
• 该结构在设计上也是线程安全的*

constructor Create(aTypeInfo: PRttiInfo; const aName: RawUtf8 = ''); reintroduce; virtual;

*初始化队列存储

• aTypeInfo应是一个动态数组的TypeInfo() RTTI指针，它将在这个TSynQueue实例中存储值
• 可以为这个实例可选地分配一个名称*

destructor Destroy; override;

*销毁存储

• 将释放所有内部存储的值，并调用WaitPopFinalize*

function Capacity: integer;

*返回当前在内存中预留的槽位数

• 队列具有优化的自动调整大小算法，可以使用此方法返回其当前capacity
• 此方法不是线程安全的，因此返回的值仅具有指示性*

function Count: integer;

*返回当前存储在此队列中的项数

• 此方法不是线程安全的，因此返回的值要么具有指示性，要么应使用显式的安全锁定/解锁
• 如果要检查队列是否为空，请调用Pending*

function Peek(out aValue): boolean;

*从队列中按FIFO（先入先出）方式查找一个项

• 如果aValue已被填充了一个pending项，而不从队列中删除它（如Pop方法所做的），则返回true
• 如果队列为空，则返回false
• 此方法是线程安全的，因为它将锁定实例*

*如果队列中有一些当前pending的项，则返回true

• 比检查Count=0更快，并且比Pop或Peek快得多
• 此方法不是线程安全的，因此返回的值仅具有指示性*

function Pop(out aValue): boolean;

*从队列中按FIFO（先入先出）方式提取一个项

• 如果aValue已被填充了一个pending项，并且该项已从队列中删除（如果不想删除它，请使用Peek），则返回true
• 如果队列为空，则返回false
• 此方法是线程安全的，因为它将锁定实例*

function PopEquals(aAnother: pointer; aCompare: TDynArraySortCompare; out aValue): boolean;

*从队列中按FIFO（先入先出）方式提取一个匹配的项

• 当前pending项与aAnother值进行比较*

function WaitPeekLocked(aTimeoutMS: integer; const aWhenIdle: TThreadMethod): pointer;

*等待从队列中按FIFO（先入先出）方式查找一个项

• 在aTimeoutMS时间内返回一个指向pending项的指针
• 保持Safe.ReadWriteLock，因此调用者可以检查其内容，然后如果它是预期的项，则调用Pop()，并最终调用Safe.ReadWriteUnlock
• 如果在时间内没有将任何内容推入队列，则返回nil
• 此方法是线程安全的，但仅在需要时锁定实例*

• 如果在指定的aTimeoutMS时间内aValue已被填充了一个pending项，则返回true
• 如果在时间内没有将任何内容推入队列，或者已调用WaitPopFinalize，则返回false
• aWhenIdle可以被分配，例如给VCL/LCL Application.ProcessMessages
• 可以在返回之前可选地比较pending项（例如，当多个线程将项放入队列时可以使用）
• 此方法是线程安全的，但仅在需要时锁定实例*

procedure Clear;

*删除当前存储在此队列中的所有项，并清空其capacity

• 此方法是线程安全的，因为它将锁定实例*

procedure Push(const aValue);

*将一个项存储到队列中

• 此方法是线程安全的，因为它将锁定实例*

procedure Save(out aDynArrayValues; aDynArray: PDynArray = nil); overload;

*使用存储的队列项初始化一个动态数组

• aDynArrayValues应是一个在Create中定义的aTypeInfo变量
• 可以检索一个可选的TDynArray包装器，例如用于二进制或JSON持久化
• 此方法是线程安全的，并将复制队列数据*

*确保任何pending或未来的WaitPop()方法立即返回false

• 总是由Destroy析构函数调用
• 也可以从UI的OnClose事件中调用，以避免任何锁定
• 此方法是线程安全的，但仅在需要时锁定实例*

1.3.3. TPendingTaskListItem

TPendingTaskListItem = packed record

内部项目定义，由TPendingTaskList存储使用

相关联的task，以原始二进制形式存储

Timestamp: Int64;

当TPendingTaskList.GetTimestamp达到此值时，应执行task

TPendingTaskList = class(TObject)

*线程安全的任务列表，以RawByteString形式存储，并带有时间戳

• 你可以向内部列表添加任务，以在给定延迟后执行，使用类似于post/peek的算法
• 执行延迟预计不准确，但根据每次NextPendingTask调用和GetTimestamp解析度进行最佳猜测*

constructor Create; reintroduce;

默认返回GetTickCount64来初始化列表内存和资源

function NextPendingTask: RawByteString; virtual;

*检索下一个待处理task

• 如果当前没有计划中的task，则返回''
• 返回与指定延迟相对应的下一个堆栈*

procedure AddTask(aMilliSecondsDelayFromNow: integer; const aTask: RawByteString); virtual;

从当前时间开始，指定毫秒延迟后Append一个task

procedure AddTasks(const aMilliSecondsDelays: array of integer; const aTasks: array of RawByteString);

*指定任务之间的毫秒延迟后Append多个任务

• 第一个提供的延迟将从当前时间开始计算，然后它将指定等待下一个提供task需要多长时间——也就是说，aMilliSecondsDelays不是绝对延迟*

procedure Clear; virtual;

清除所有待处理任务

property Count: integer read GetCount;

当前定义了多少个待处理任务

property Task: TPendingTaskListItemDynArray read fTask;

*直接低级访问内部task列表

• 警告：此动态数组的长度是列表容量：请使用Count属性来检索存储的确切项数
• 使用try ... finally Safe.Unlock块中的Safe.Lock/TryLock进行线程安全访问此数组
• 项目按递增的Timestamp存储，即第一个项目是NextPendingTask方法将返回的下一个项目*

property Timestamp: Int64 read GetTimestamp;

*访问内部TPendingTaskListItem.Timestamp存储的值

• 对应当前时间
• 默认实现是返回GetTickCount64，在Windows下典型分辨率为16毫秒*

ILockedDocVariant = interface(IInterface)

*用于线程安全访问TDocVariant文档的引用计数接口

• 例如，由TLockedDocVariant实现，用于IoC/DI解析
• 快速且安全地存储任何类似JSON的对象，作为属性/值对，或类似JSON的数组，作为值*

function AddExistingProp(const Name: RawUtf8; var Obj: variant): boolean;

*将现有属性value添加到给定的TDocVariant文档对象

• 如果Name存在，则返回TRUE并将Name/Value对添加到Obj
• 如果存储的文档中不存在Name，则返回FALSE
• 此方法在查找Name期间会使用lock，但无论是否返回FALSE，总是会释放lock（参见AddExistingPropOrLock）*

function AddExistingPropOrLock(const Name: RawUtf8; var Obj: variant): boolean;

*将现有属性value添加到给定的TDocVariant文档对象

• 如果Name存在，则返回TRUE并将Name/Value对添加到Obj，使用内部lock确保线程安全
• 如果存储的文档中不存在Name，则返回FALSE并锁定内部存储：调用者最终应通过AddNewPropAndUnlock()释放lock
• 可以这样使用，以实现线程安全的缓存：

if not cache.AddExistingPropOrLock('Articles',Scope) then
  cache.AddNewPropAndUnlock('Articles',GetArticlesFromDB,Scope);

这里GetArticlesFromDB会在主要lock内部执行

function Copy: variant;

对内部TDocVariant文档对象或数组进行线程安全的复制

function Exists(const Name: RawUtf8; out Value: Variant): boolean;

*按名称检查和返回一个给定的属性

• 如果找到Name，则返回TRUE并将与提供的Name相关联的value填充到Value中，使用内部lock确保线程安全
• 如果未找到Name，则返回FALSE并释放内部lock：如果你想要添加缺失的value，请使用ExistsOrLock()*

function ExistsOrLock(const Name: RawUtf8; out Value: Variant): boolean;

*按名称检查和返回一个给定的属性

• 如果找到Name，则返回TRUE并将与提供的Name相关联的value填充到Value中，使用内部lock确保线程安全
• 如果Name不存在，则返回FALSE并设置内部lock：调用者随后应通过ReplaceAndUnlock()释放lock*

function ToJson(HumanReadable: boolean = false): RawUtf8;

procedure AddItem(const Value: variant);

*将value添加到内部TDocVariant文档数组

• 不应与其他基于文档的替代方案（如Exists/AddExistingPropOrLock或AddExistingProp）结合使用*

procedure AddNewProp(const Name: RawUtf8; const Value: variant; var Obj: variant);

*将属性value添加到给定的TDocVariant文档对象

• 此方法在调用时不会期望资源被锁定，与AddNewPropAndUnlock不同
• 将使用内部lock确保线程安全
• 如果Name已存在，将更新/更改现有的value
• 可以这样使用，以实现线程安全的缓存：

if not cache.AddExistingProp('Articles',Scope) then
  cache.AddNewProp('Articles',GetArticlesFromDB,Scope);

这里GetArticlesFromDB会在主lock之外执行

procedure AddNewPropAndUnlock(const Name: RawUtf8; const Value: variant; var Obj: variant);

*将属性value添加到给定的TDocVariant文档对象以及内部存储的文档，然后释放之前的lock

• 调用此方法之前应由AddExistingPropOrLock()返回false，即在一个已锁定的实例上执行*

procedure Clear;

删除所有存储的属性

procedure ReplaceAndUnlock(const Name: RawUtf8; const Value: Variant; out LocalValue: Variant);

*通过属性名称设置value，并设置本地副本

• 可以这样使用，以实现线程安全的缓存：

if not cache.ExistsOrLock('prop',local) then
  cache.ReplaceAndUnlock('prop',newValue,local);

• 调用此方法之前应由ExistsOrLock()返回false，即在一个已锁定的实例上执行

• 这是此存储的主要入口点
• 如果在读取时Name不存在，将引发EDocVariant异常
• 实现类将对variant value进行线程安全的副本*

TLockedDocVariant = class(TInterfacedObjectWithCustomCreate)

*允许线程安全地访问TDocVariant文档

• 此类从TInterfacedObjectWithCustomCreate继承，因此您可以定义一个mormot.core.interfaces.pas TInjectableObject的已发布属性为ILockedDocVariant，以便此类可以自动注入*

constructor Create(options: TDocVariantOptions); reintroduce; overload;

使用相应的选项初始化线程安全的文档存储

constructor Create(options: TDocVariantModel); reintroduce; overload;

从给定的模板初始化线程安全的文档存储

constructor Create; overload; override;

*使用快速的TDocVariant初始化线程安全的文档

• 即调用Create(true)或Create(JSON_FAST)
• 这将是TInterfacedObjectWithCustomCreate的默认构造函数，例如在IoC/DI解析期间调用*

destructor Destroy; override;

销毁存储

function AddExistingProp(const Name: RawUtf8; var Obj: variant): boolean;

*向给定的TDocVariant文档对象添加一个已存在的属性value

• 如果Name存在，则返回TRUE并将Name/Value对添加到Obj
• 如果存储的文档中不存在Name，则返回FALSE
• 此方法在查找Name期间会使用lock，但无论是否返回FALSE，总是会释放lock（参见AddExistingPropOrLock）*

function AddExistingPropOrLock(const Name: RawUtf8; var Obj: variant): boolean;

*向给定的TDocVariant文档对象添加一个已存在的属性value

• 如果Name存在，则返回TRUE并将Name/Value对添加到Obj
• 如果存储的文档中不存在Name，则返回FALSE，并期望最终调用Lock.Leave或AddNewPropAndUnlock()

function Copy: variant;

对内部TDocVariant文档对象或数组进行线程安全的复制

function Exists(const Name: RawUtf8; out Value: Variant): boolean;

按名称检查和返回一个给定的属性

function ExistsOrLock(const Name: RawUtf8; out Value: Variant): boolean;

*按名称检查和返回一个给定的属性

• 如果找到，则返回TRUE并返回现有Name的value
• 如果未找到，则返回FALSE，并期望最终调用Lock.Leave或ReplaceAndUnlock()

function Lock: TAutoLocker;

对关联的线程安全互斥锁进行低级访问

function ToJson(HumanReadable: boolean = false): RawUtf8;

*将存储的value保存为UTF-8编码的JSON对象

• 仅仅是VariantSaveJson()的包装*

procedure AddItem(const Value: variant);

将value添加到内部TDocVariant文档数组

procedure AddNewProp(const Name: RawUtf8; const Value: variant; var Obj: variant);

*向给定的TDocVariant文档对象添加一个属性value

• 此方法在调用时不会期望资源被锁定，与AddNewPropAndUnlock不同
• 将使用内部lock确保线程安全
• 如果Name已存在，将更新/更改现有的value*

向给定的TDocVariant文档对象和内部存储的文档添加一个属性value

procedure Clear;

删除所有存储的属性

通过属性名称设置value，并设置本地副本

• 如果Name不存在，将引发EDocVariant异常
• 返回的variant结果是副本，不是varByRef，因为副本将更具线程安全性*

1.3.7. TThreadAbstract

TThreadAbstract = class(TThread)

*调用此方法以start线程

• Resume在最新的RTL中已被弃用，因为一些操作系统（例如Linux）没有实现此暂停/恢复功能；我们在此为Delphi的旧版本定义此方法*

procedure Terminate; reintroduce;

重新引入以调用TerminatedSet

*正确terminate线程

• 由重新引入的Terminate调用*

property Terminated;

定义为public，因为可能用于terminate处理方法

TSynBackgroundThreadAbstract = class(TThreadAbstract)

• 不应直接使用此类，而应使用TSynBackgroundThreadMethodAbstract / TSynBackgroundThreadEvent / TSynBackgroundThreadMethod，并提供更方便的回调*

constructor Create(const aThreadName: RawUtf8; const OnBeforeExecute: TOnNotifyThread = nil; const OnAfterExecute: TOnNotifyThread = nil; CreateSuspended: boolean = false); reintroduce;

*初始化线程

• 可以定义一些回调来嵌套线程执行，例如分配给TRestServer.BeginCurrentThread/EndCurrentThread，或者至少将OnAfterExecute设置为TSynLogFamily.OnThreadEnded*

destructor Destroy; override;

*释放使用的资源

• 调用WaitForNotExecuting(100)以确保正确终结*

function SleepOrTerminated(MS: cardinal): boolean;

*Sleep()的安全版本，不会中断线程进程

• 如果线程已终止，则返回TRUE
• 如果成功等待了MS毫秒，则返回FALSE*

procedure TerminatedSet; override;

*正确终止线程

• 由重新引入的Terminate调用*

procedure WaitForNotExecuting(maxMS: integer = 500);

*等待Execute/ExecuteLoop结束（即fExecute<>exRun）

• 在循环中调用Sleep()，直到达到超时
• 例如在Destroy中使用，以避免任何GPF并确保干净地终结*

property Pause: boolean read fExecuteLoopPause write SetExecuteLoopPause;

*临时停止ExecuteLoop的执行，直到设置为false

• 例如可由TSynBackgroundTimer使用，以延迟后台任务的进程*

property ProcessEvent: TSynEvent read fProcessEvent;

*访问与低级别相关的事件，用于通知后台线程任务执行

• 可以调用ProcessEvent.SetEvent来触发内部处理循环*

TSynBackgroundThreadMethodAbstract = class(TSynBackgroundThreadAbstract)

• 典型用途是用于处理数据或远程访问的后台线程，同时UI将保持响应，通过在循环中运行OnIdle事件：例如，查看mormot.rest.client.pas单元中TRestClientUri.OnIdle如何处理这种情况
• 不应直接使用此类，而应从中继承并重写Process方法，或使用TSynBackgroundThreadEvent / TSynBackgroundThreadMethod并提供更方便的回调*

constructor Create(const aOnIdle: TOnIdleSynBackgroundThread; const aThreadName: RawUtf8; const OnBeforeExecute: TOnNotifyThread = nil; const OnAfterExecute: TOnNotifyThread = nil); reintroduce;

*初始化线程

• 如果aOnIdle未设置（即等于nil），它将简单地等待后台进程完成，直到RunAndWait()返回
• 可以定义一些回调来嵌套线程执行，例如分配给TRestServer.BeginCurrentThread/EndCurrentThread*

destructor Destroy; override;

终结线程

function RunAndWait(OpaqueParam: pointer): boolean;

*在后台线程中异步启动Process抽象方法

• 等待进程完成，同时调用OnIdle()回调
• 在后台线程中引发的任何异常都将在调用者线程中转换
• 如果self未设置，或者从当前正在处理的同一线程中调用（以避免从OnIdle()回调引发的竞态条件），则返回false
• 当后台进程完成时返回true
• OpaqueParam将用于指定后台进程的线程安全内容
• 此方法是线程安全的，即它将等待由另一个线程启动的任何进程：可以从任何线程调用此方法，即使其主要目的是从主UI线程调用*

property OnAfterProcess: TOnNotifyThread read fOnAfterProcess write fOnAfterProcess;

在每个Process之后在Execute中触发的可选回调事件

property OnBeforeProcess: TOnNotifyThread read fOnBeforeProcess write fOnBeforeProcess;

在每个Process之前在Execute中触发的可选回调事件

property OnIdle: TOnIdleSynBackgroundThread read fOnIdle write fOnIdle;

*在远程阻塞进程期间循环执行的回调事件，例如，在请求时间较长时刷新UI

• 可以为此属性分配一个回调，例如调用Application.ProcessMessages，在后台线程中执行远程请求，但让UI保持响应：mORMotUILogin.pas中的TLoginForm.OnIdleProcess和OnIdleProcessForm方法将满足此属性的预期
• 如果OnIdle未设置（即等于nil），它将简单地等待后台进程完成，直到RunAndWait()返回*

property OnIdleBackgroundThreadActive: boolean read GetOnIdleBackgroundThreadActive;

*如果后台线程处于活动状态，并且在处理过程中调用了OnIdle事件，则为TRUE

• 例如，用于确保用户界面消息不会重新进入*

TSynBackgroundThreadEvent = class(TSynBackgroundThreadMethodAbstract)

constructor Create(const aOnProcess: TOnProcessSynBackgroundThread; const aOnIdle: TOnIdleSynBackgroundThread; const aThreadName: RawUtf8); reintroduce;

property OnProcess: TOnProcessSynBackgroundThread read fOnProcess write fOnProcess;

TSynBackgroundThreadMethod = class(TSynBackgroundThreadMethodAbstract)

procedure RunAndWait(Method: TThreadMethod); reintroduce;

• 使用此方法，而不是继承的RunAndWait()*

TSynBackgroundThreadProcedure = class(TSynBackgroundThreadMethodAbstract)

constructor Create(aOnProcess: TOnProcessSynBackgroundThreadProc; const aOnIdle: TOnIdleSynBackgroundThread; const aThreadName: RawUtf8); reintroduce;

property OnProcess: TOnProcessSynBackgroundThreadProc read fOnProcess write fOnProcess;

TSynBackgroundThreadProcess = class(TSynBackgroundThreadAbstract)

constructor Create(const aThreadName: RawUtf8; const aOnProcess: TOnSynBackgroundThreadProcess; aOnProcessMS: cardinal; const aOnBeforeExecute: TOnNotifyThread = nil; const aOnAfterExecute: TOnNotifyThread = nil; aStats: TSynMonitorClass = nil; CreateSuspended: boolean = false); reintroduce; virtual;

destructor Destroy; override;

property OnException: TNotifyEvent read fOnException write fOnException;

*当OnProcess引发异常时执行的事件回调

• 提供的Sender参数是引发的异常实例*

property OnProcess: TOnSynBackgroundThreadProcess read fOnProcess;

访问周期性任务的实现事件

property OnProcessMS: cardinal read fOnProcessMS write fOnProcessMS;

访问周期性任务处理的延迟（以毫秒为单位）

property Stats: TSynMonitor read fStats;

*处理统计信息

• 如果在类构造函数中aStats为nil，则可能为nil*

TSynBackgroundTimerTask = record

由TSynBackgroundTimer内部注册列表使用

TSynBackgroundTimer = class(TSynBackgroundThreadProcess)

• 例如，通过继承的TRestBackgroundTimer，由TRest.TimerEnable/TimerDisable方法使用
• 每个进程可以有自己的text消息FIFO队列
• 如果你expect更新某些GUI，你应该使用TTimer组件（例如，周期为200ms），因为TSynBackgroundTimer将使用自己的独立线程*

constructor Create(const aThreadName: RawUtf8; const aOnBeforeExecute: TOnNotifyThread = nil; aOnAfterExecute: TOnNotifyThread = nil; aStats: TSynMonitorClass = nil; aLogClass: TSynLogClass = nil); reintroduce; virtual;

*初始化线程以进行周期性taskprocessing

• 你可以定义一些回调来嵌套线程执行，例如分配给TRestServer.BeginCurrentThread/EndCurrentThread，如TRestBackgroundTimer.Create所做的那样*

destructor Destroy; override;

终结线程

function DeQueue(const aOnProcess: TOnSynBackgroundTimerProcess; const aMsg: RawUtf8): boolean;

*从processing列表中移除一条消息

• 提供的消息将在与aOnProcess相关联的内部FIFO列表中搜索，如果找到则从列表中移除
• aOnProcess应通过之前的调用到Enable()方法进行注册
• 如果成功则返回true，如果提供的消息未注册则返回false*

function Disable(const aOnProcess: TOnSynBackgroundTimerProcess): boolean;

*取消定义一个在周期秒数上运行的task

• aOnProcess应通过之前的调用到Enable()方法进行注册
• 如果成功则返回true，如果提供的task未注册则返回false
• 对于mORMot服务上的后台进程，考虑使用TRestServer的TimerEnable/TimerDisable方法及其TSynBackgroundTimer线程*

function EnQueue(const aOnProcess: TOnSynBackgroundTimerProcess; const aMsgFmt: RawUtf8; const Args: array of const; aExecuteNow: boolean = false): boolean; overload;

*在task的下次执行期间添加一条格式化消息进行处理

• 提供的消息将被添加到与aOnProcess相关联的内部FIFO列表中，然后每次调用时作为aMsg参数提供
• 如果aExecuteNow为true，则不会等待下一个aOnProcessSecs的发生
• aOnProcess应通过之前的调用到Enable()方法进行注册
• 如果成功则返回true，如果提供的task未注册则返回false*

function EnQueue(const aOnProcess: TOnSynBackgroundTimerProcess; const aMsg: RawUtf8; aExecuteNow: boolean = false): boolean; overload;

*在task的下次执行期间添加一条消息进行处理

• 提供的消息将被添加到与aOnProcess相关联的内部FIFO列表中，然后每次调用时作为aMsg参数提供
• 如果aExecuteNow为true，则不会等待下一个aOnProcessSecs的发生
• aOnProcess应通过之前的调用到Enable()方法进行注册
• 如果成功则返回true，如果提供的task未注册则返回false*

function ExecuteNow(const aOnProcess: TOnSynBackgroundTimerProcess): boolean;

*立即执行一个task，不等待下一个aOnProcessSecs的发生

• aOnProcess应通过之前的调用到Enable()方法进行注册
• 如果成功则返回true，如果提供的task未注册则返回false*

function ExecuteOnce(const aOnProcess: TOnSynBackgroundTimerProcess): boolean;

*立即执行一个task，不等待下一个aOnProcessSecs的发生

• aOnProcess不应通过之前的调用到Enable()方法进行注册*

procedure Enable(const aOnProcess: TOnSynBackgroundTimerProcess; aOnProcessSecs: cardinal);

*定义一个在周期秒数上运行的task的处理方法

• 对于mORMot服务上的后台进程，考虑使用TRest的TimerEnable/TimerDisable方法及其关联的BackgroundTimer线程*

procedure WaitUntilNotProcessing(timeoutsecs: integer = 10);

等待直到没有后台task正在处理

property Processing: boolean read fProcessing;

如果当前正在处理某些tasks，则返回TRUE

对内部task列表的低级访问

property Tasks: TDynArrayLocked read fTasks;

对内部task列表的线程安全包装的低级访问 Low-level access to the internal task list wrapper and safe

TBlockingProcess = class(TSynEvent)

• 例如，在mormot.rest.server.pas中的TBlockingCallback中使用
• 一旦创建，进程将通过WaitFor调用进行阻塞，当进程后台线程调用NotifyFinished时，阻塞将被释放*

constructor Create(aTimeOutMs: integer); reintroduce; overload; virtual;

*初始化信号量实例

• 指定阻塞执行应被视为失败的超时毫秒数周期（如果设置为0，则默认使用3000）
• 创建一个与之关联的互斥锁，并由该实例拥有*

constructor Create(aTimeOutMs: integer; aSafe: PSynLocker); reintroduce; overload; virtual;

*重写以reset相关参数并初始化信号量实例

• 指定阻塞执行应被视为失败的超时毫秒数周期（如果设置为0，则默认使用3000）
• 应提供一个关联的互斥锁*

destructor Destroy; override;

终结实例

• 调用者随后将允许其WaitFor方法返回
• 如果成功则返回TRUE（即状态不是evRaised或evTimeout）
• 如果实例已经被锁定（例如，从TBlockingProcessPool.FromCallLocked检索时），你可以设置alreadyLocked=TRUE*

function Reset: boolean; virtual;

*只是一个包装器，用于将内部Event状态重置为evNone

• 在成功完成WaitFor/NotifyFinished过程后，可以使用它来重用同一个TBlockingProcess实例
• 如果成功则返回TRUE（即状态不是evWaiting），将当前状态设置为evNone，并将Call属性设置为0
• 如果当前正在进行WaitFor过程，则返回FALSE*

function WaitFor: TBlockingEvent; reintroduce; overload; virtual;

*调用以等待NotifyFinished()被调用或触发超时

• 返回进程的最终状态，即evRaised或evTimeOut*

function WaitFor(TimeOutMS: integer): TBlockingEvent; reintroduce; overload;

*调用以等待NotifyFinished()被调用或触发超时

• 返回进程的最终状态，即evRaised或evTimeOut*

只是fSafe^.Lock的一个包装器

只是fSafe^.Unlock的一个包装器

property Event: TBlockingEvent read fEvent;

*进程的当前状态

• 使用Reset方法可以在WaitFor过程后重用此实例*

property TimeOutMs: integer read fTimeOutMS;

在构造函数级别定义的超时周期，以毫秒为单位

TBlockingProcessPoolItem = class(TBlockingProcess)

*在TBlockingProcessPool中使用的信号量

• 此类信号量具有一个Call字段来标识每次执行*

property Call: TBlockingProcessPoolCall read fCall;

*一个唯一标识符，当被TBlockingProcessPool拥有时，用于标识每次执行

• Reset方法会将此字段恢复为其默认值0，以便重用或重新开始新的执行*

这样，TBlockingProcessPoolItem类作为 TBlockingProcess的子类，在继承其信号量功能的基础上，增加了 Call属性来支持在 TBlockingProcessPool中的唯一标识，使得在池化管理信号量时能够区分不同的执行实例。通过 Reset方法，可以重置信号量的状态以及 Call属性，以便循环利用。

TBlockingProcessPool = class(TSynPersistent)

*管理一组TBlockingProcessPoolItem实例

• 每次调用将通过唯一的TBlockingProcessPoolCall值进行标识
• 用于模拟例如从异步事件驱动的DDD进程中的阻塞执行
• 它还允许重用TEvent系统资源*

constructor Create(aClass: TBlockingProcessPoolItemClass = nil); reintroduce;

为给定的实现类设置TBlockingProcessPoolItem.Call并初始化池

destructor Destroy; override;

*终结池

• 也会强制所有挂起的WaitFor触发evTimeOut*

function FromCall(call: TBlockingProcessPoolCall; locked: boolean = false): TBlockingProcessPoolItem; virtual;

*根据其调用标识符检索TBlockingProcess

• 可以从异步过程的回调中使用，例如设置继承自TBlockingProcess的一些附加参数，然后调用NotifyFinished来释放调用者的WaitFor
• 如果leavelocked为TRUE，则返回的实例将被锁定：调用者应在使用后执行result.Unlock或NotifyFinished(true)*

function NewProcess(aTimeOutMs: integer): TBlockingProcessPoolItem; virtual;

*从内部池中预订一个TBlockingProcess

• 出错时返回nil（例如，实例正在销毁）
• 或者返回与此调用对应的阻塞进程实例；其Call属性将标识异步回调的调用，然后在WaitFor之后，应运行Reset方法来释放池的互斥锁*

TSynParallelProcessThread = class(TSynBackgroundThreadMethodAbstract)

执行TSynParallelProcess过程的线程

TSynParallelProcess = class(TSynPersistentLock)

constructor Create(ThreadPoolCount: integer; const ThreadName: RawUtf8; const OnBeforeExecute: TOnNotifyThread = nil; const OnAfterExecute: TOnNotifyThread = nil; MaxThreadPoolCount: integer = 32); reintroduce; virtual;

• 可以定义一些回调来嵌套线程执行，例如分配给TRestServer.BeginCurrentThread/EndCurrentThread
• 最多可以设置MaxThreadPoolCount=32个线程（你可以允许更大的值，但此线程池的意义在于使其进程饱和每个CPU核心）
• 如果ThreadPoolCount为0，则不会创建线程，进程将在当前线程中执行*

destructor Destroy; override;

终结线程池

procedure ParallelRunAndWait(const Method: TOnSynParallelProcess; MethodCount: integer; const OnMainThreadIdle: TNotifyEvent = nil);

*并行运行一个方法，并等待执行完成

• 将在线程上split执行Method[0..MethodCount-1]
• 如果在进程中发生任何异常，此方法将引发ESynParallel异常
• 如果设置了OnMainThreadIdle，当前线程（例如，预期为主UI线程）将不会处理任何事情，但在等待后台线程时调用此事件*

property ParallelRunCount: integer read fParallelRunCount;

已激活的线程数

property ThreadName: RawUtf8 read fThreadName;

一些text标识符，用于区分每个拥有的线程

property ThreadPoolCount: integer read fThreadPoolCount;

当前在此实例线程池中的线程数

TSynThread = class(TThreadAbstract)

constructor Create(CreateSuspended: boolean); reintroduce; virtual;

function SleepOrTerminated(MS: cardinal): boolean;

property StartNotified: TObject read fStartNotified write fStartNotified;

TNotifiedThread = class(TSynThread)

• 例如，服务器线程
• 不要直接使用此类，而应使用THttpServer、THttpApiServer或TWebSocketServer（如mormot.net.websock中定义）*

constructor Create(CreateSuspended: boolean; const OnStart, OnStop: TOnNotifyThread; const ProcessName: RawUtf8); reintroduce; virtual;

初始化服务器实例，处于非挂起状态

procedure SetServerThreadsAffinityPerCpu( const log: ISynLog; const threads: TThreadDynArray);

*将每个线程分配给单个逻辑CPU核心

• 例如，对于HTTP服务器，它可能确保具有短生命周期请求和高线程数的更好可扩展性*

procedure SetServerThreadsAffinityPerSocket( const log: ISynLog; const threads: TThreadDynArray);

*将每个线程分配给单个硬件CPU套接字

• 例如，对于HTTP服务器，它可能确保在具有多个物理CPU包的复杂硬件上更好的可扩展性 - 但它非常挑剔，因此仅应在实际硬件上进行适当测试后启用*

1.3.23. TLoggedThread

TLoggedThread = class(TSynThread)

抽象类，用于实现具有日志通知的线程

constructor Create(CreateSuspended: boolean; Logger: TSynLogClass; const ProcName: RawUtf8); reintroduce; virtual;

初始化服务器实例，处于非挂起状态

procedure TerminateAndWaitFinished(TimeOutMs: integer = 5000); virtual;

通知线程终止，并等待DoExecute完成

property LogClass: TSynLogClass read fLogClass;

关联的日志类

property ProcessName: RawUtf8 read fProcessName;

此线程的名称，如提供给SetCurrentThreadName()的

TLoggedWorkThread = class(TLoggedThread)

constructor Create(Logger: TSynLogClass; const ProcessName: RawUtf8; const NameValuePairs: array of const; const OnExecute: TOnLoggedWorkProcess; const OnExecuted: TNotifyEvent = nil); reintroduce; overload;

• 上下文作为具有名称/值对的TDocVariantData对象*

constructor Create(Logger: TSynLogClass; const ProcessName: RawUtf8; Sender: TObject; const OnExecute: TNotifyEvent; const OnExecuted: TNotifyEvent = nil); reintroduce; overload;

*此构造函数将直接在后台启动线程

• 上下文作为常规的TNotifyEvent*

TSynThreadPoolWorkThread = class(TSynThread)

定义了由TSynThreadPool使用的工作线程

constructor Create(Owner: TSynThreadPool); reintroduce;

异常安全地调用fOwner.Task()来初始化线程

destructor Destroy; override;

终结线程

procedure Execute; override;

将循环等待任何待处理的任务，并执行fOwner.Task()

property Owner: TSynThreadPool read fOwner;

关联的线程池

TSynThreadPool = class(TObject)

*一个简单的线程池，例如用于快速处理HTTP/1.0请求

• 在Windows上通过I/O完成端口实现，或在Linux/POSIX上通过经典的事件驱动方法实现*

constructor Create(NumberOfThreads: integer = 32; aOverlapHandle: THandle = INVALID_HANDLE_VALUE; const aName: RawUtf8 = '');

*使用提供的线程数初始化线程池

• 抽象Task()虚方法将由其中一个线程调用
• 一个线程池最多可以关联256个线程
• 在Windows上，可以可选地接受一个aOverlapHandle - 一个先前使用Windows重叠I/O（IOCP）打开的句柄
• 在POSIX上，如果aQueuePendingContext=true，则会将待处理上下文存储到内部队列中，以便在可能时处理，直到队列满为止Push()返回true*

destructor Destroy; override;

关闭线程池，释放所有关联的线程

function Push(aContext: pointer; aWaitOnContention: boolean = false): boolean;

*让线程池处理一个任务（指定为指针）

• 如果线程池中没有空闲线程可用，并且使用了Create(aQueuePendingContext=false)（调用者应稍后重试），则返回false；如果在Create中aQueuePendingContext为true，或者使用了IOCP，则提供的上下文将被添加到内部列表，并在可能时处理
• 如果aWaitOnContention默认为false，则当队列满时立即返回；设置aWaitOnContention=true以等待最多ContentionAbortDelay毫秒并重试将任务排队*

property ContentionAbortCount: cardinal read fContentionAbortCount;

*由于线程池争用而被拒绝的任务数量

• 如果此数量较高，请考虑设置更高的线程数，或对Task方法进行分析和调优*

property ContentionAbortDelay: integer read fContentionAbortDelay write fContentionAbortDelay;

*由于争用而拒绝连接的毫秒延迟

• 默认为5000，即在IOCP或aQueuePendingContext内部列表中有可用空间之前等待5秒
• 在此延迟期间，没有新的连接可用（即不调用Accept），以便负载均衡器可以检测到争用并切换到池中的另一个实例，或者直接客户端最终可能拒绝连接，因此不会开始发送数据*

property ContentionCount: cardinal read fContentionCount;

*线程池等待队列中可用槽位的次数

• 争用不会立即失败，但会重试直到ContentionAbortDelay
• 这里的任何高数值可能需要增加线程数
• 使用此属性和ContentionTime来计算平均争用时间*

property ContentionTime: Int64 read fContentionTime;

*等待队列中可用槽位的总毫秒数

• 争用不会立即失败，但会重试直到ContentionAbortDelay
• 这里的任何高数值需要对Task方法进行代码重构*

property RunningThreads: integer read fRunningThreads;

*当前在此线程池中处理任务的线程数

• 范围在0..WorkThreadCount之间*

property WorkThread: TSynThreadPoolWorkThreads read fWorkThread;

对此线程池中定义的线程的低级访问

property WorkThreadCount: integer read fWorkThreadCount;

*线程池中可用的线程数

• 映射Create()参数，即默认为32*

1.4. 在mormot.core.threads单元中实现的类型

TBlockingEvent = ( evNone, evWaiting, evTimeOut, evRaised );

TBlockingProcess实例的当前状态

TBlockingProcessPoolCall = type integer;

*用于标识每个TBlockingProcessPool调用

• 允许匹配给定的TBlockingProcessPoolItem信号量*

TBlockingProcessPoolItemClass = class of TBlockingProcessPoolItem;

TBlockingProcess的类引用类型（元类）

TEvent = syncobjs.TEvent;

• 请注意，您可能更希望使用来自mormot.core.os.pas的TSynEvent*

TOnIdleSynBackgroundThread = procedure(Sender: TSynBackgroundThreadAbstract; ElapsedMS: integer) of object;

*在后台线程进行远程阻塞处理期间，由TSynBackgroundThreadAbstract在调用者线程中调用的空闲方法

• 典型用途是运行Application.ProcessMessages，例如对于TRestClientUri.Uri()，即使在远程访问缓慢阻塞的情况下也能提供响应的UI
• 提供从请求开始经过的时间（以毫秒为单位）（例如，可用于弹出临时等待消息）
• 在请求开始时调用一次，此时ElapsedMS=0
• 在请求结束时调用一次，此时ElapsedMS=-1
• 请参见mORMotUILogin.pas中的TLoginForm.OnIdleProcess和OnIdleProcessForm*

TOnLoggedWorkProcess = procedure(const Context: TDocVariantData) of object;

由TLoggedWorkThread.Create在后台线程中调用的事件

TOnNotifyThread = procedure(Sender: TThread) of object;

事件原型，例如由TSynBackgroundThreadAbstract和TSynThread回调使用

TOnProcessSynBackgroundThread = procedure(Sender: TSynBackgroundThreadEvent; ProcessOpaqueParam: pointer) of object;

*由TSynBackgroundThreadEvent调用的后台处理方法

• 当执行Process虚拟方法时，将提供提供给RunAndWait()方法的OpaqueParam参数*

TOnProcessSynBackgroundThreadProc = procedure(ProcessOpaqueParam: pointer);

*由TSynBackgroundThreadProcedure调用的后台处理过程

• 当执行Process虚拟方法时，将提供提供给RunAndWait()方法的OpaqueParam参数*

TOnSynBackgroundThreadProcess = procedure(Sender: TSynBackgroundThreadProcess) of object;

由TSynBackgroundThreadProcess定期执行的事件回调

TOnSynBackgroundTimerProcess = procedure(Sender: TSynBackgroundTimer; const Msg: RawUtf8) of object;

*由TSynBackgroundTimer定期执行的事件回调

• 如果此任务的FIFO中没有待处理消息，则Msg为''
• 对于此任务的FIFO中的每个待处理消息，都会设置Msg
• 在mORMot 1中，有一个现在已被移除的TWaitEvent参数*

TOnSynParallelProcess = procedure(IndexStart, IndexStop: integer) of object;

*为TSynParallelProcess实现并行处理过程的回调

• 如果0<=IndexStart<=IndexStop，则应执行一些处理过程*

TPendingTaskListItemDynArray = array of TPendingTaskListItem;

内部列表定义，用于TPendingTaskList存储

TSynBackgroundThreadProcessStep = ( flagIdle, flagStarted, flagFinished, flagDestroying );

TSynBackgroundThreadAbstract进程的状态机状态

TSynBackgroundThreadProcessSteps = set of TSynBackgroundThreadProcessStep;

TSynBackgroundThreadAbstract进程的状态机状态集合

TSynBackgroundTimerTaskDynArray = array of TSynBackgroundTimerTask;

保护Msg[]列表存储TSynBackgroundTimer的内部注册列表

TThreadDynArray = array of TThread;

TWaitResult = syncobjs.TWaitResult;

• 请注意，您可能更希望使用来自mormot.core.os.pas的TSynEvent*

1.5. 在mormot.core.threads单元中实现的常量

THREADPOOL_MAXTHREADS = 256;

允许为TSynThreadPoolWorkThread堆栈分配最多256 * 2MB = 512MB的内存

wrSignaled = syncobjs.wrSignaled;

建议使用mormot.core.os中的TSynEvent，而不是此处定义的TEvent，以避免在uses子句中显式链接到syncobjs