1707 lines
96 KiB
C
1707 lines
96 KiB
C
/*
|
||
* 事件式召唤共用状态机。Slave负责通信,应用通过BEGIN/NEED_DATA/FINISHED交互。
|
||
* 请求任务仅保存原身份及10字节应答副本;业务事件为纯值,可以移动和复制。
|
||
* 单目标预留FINISHED槽,多目标使用每站终结记录,断线/STOP/取消/撤销/超时不丢清理通知。
|
||
* 确认/终止沿原身份走HIGH,数据尝试直接提交,避免跨连接重放和确认/数据/终止乱序。
|
||
*
|
||
* 正常阶段:START生成BEGIN并等待应用接受 -> CONFIRM提交激活确认 -> DATA反复索取并提交批次
|
||
* -> TERMINATE提交激活终止 -> 交付FINISHED。多目标按全部确认、逐站数据、全部终止推进。
|
||
* 只执行动作的种类在CONFIRM成功后直接FINISHED;种类差异仅由创建时的只读规则表决定。
|
||
* FINISHING保留多目标状态区至终结事件读完,不阻塞网络或其他短应答。
|
||
* REPLY表示当前站只需提交确认或错误应答;广播继续下一站,否则释放任务。
|
||
* Handle_Event更新会话和接收请求,条件允许时立即提交短应答;Process推进保存的任务。
|
||
* NextEvent把业务事件交给应用;一个实例至多一个未终结召唤,同种类忙时否定新激活。
|
||
* Reply_Begin/Complete_Data接收决定;Submit_Batch当场尝试不排队提交,WAIT后应用重新组包。
|
||
* 总召可重取当前值,计数量读取须保持本轮冻结快照;引擎不保存业务点值。
|
||
* 事件来源和会话可用性统一由Slave接口校验,本模块维护各实例自己的阶段和业务期限。
|
||
* 单帧应答与召唤任务分开选择,双方可推进时轮转;召唤等待应用不阻塞单帧应答。
|
||
* 请求解析后按CA选站,合法多目标广播持有节点集合;错误广播仍用共享快照遍历。
|
||
*/
|
||
#include "IEC104_Interrogation_Internal.h"
|
||
#include "IEC104_TcpServer_Config.h"
|
||
#include "IEC104_Net.h"
|
||
#include <string.h>
|
||
|
||
/* 库内三类存储均为静态有界容量:实例池、每实例协议任务池、每实例业务事件FIFO。
|
||
* 多目标广播另外借用调用方提供的轻量状态数组,不分配数据快照或每站发送队列。
|
||
* 单目标路径需同时容纳BEGIN及预留的FINISHED,因此业务事件容量至少为两个。 */
|
||
#ifndef IEC104_INTERROGATION_CAPACITY
|
||
#define IEC104_INTERROGATION_CAPACITY (2U * IEC104_TCP_SERVER_CAPACITY)
|
||
#endif
|
||
#ifndef IEC104_INTERROGATION_TASK_CAPACITY
|
||
#define IEC104_INTERROGATION_TASK_CAPACITY 4U
|
||
#endif
|
||
#ifndef IEC104_INTERROGATION_EVENT_CAPACITY
|
||
#define IEC104_INTERROGATION_EVENT_CAPACITY 4U
|
||
#endif
|
||
_Static_assert(IEC104_INTERROGATION_CAPACITY > 0U, "Interrogation needs instance storage");
|
||
_Static_assert(IEC104_INTERROGATION_TASK_CAPACITY > 0U, "Interrogation needs request storage");
|
||
_Static_assert(IEC104_INTERROGATION_EVENT_CAPACITY >= 2U, "BEGIN needs a reserved FINISHED slot");
|
||
|
||
/* 支持的104召唤命令固定为6字节公共头、3字节IOA和1字节限定词;入口完整校验后才保存。
|
||
* 独立保存原身份和必要原文,不借用SlaveEvent、ASDU视图或发送接口中的自引用。
|
||
* 不保存接收APCI;发送序号、确认号及链路期限由底层连接管理。 */
|
||
typedef struct {
|
||
IEC104_SlaveEventIdentity_t identity;
|
||
uint8_t bytes[IEC104_ASDU_HEADER_LEN + 3U + 1U];
|
||
} InterrogationResponse;
|
||
|
||
/* 任务使用固定槽及顺序索引;压缩存储不改变任务数量、调度顺序和背压策略。 */
|
||
typedef struct {
|
||
InterrogationResponse request; /* 原服务/代次和独立10字节请求,保存时已设置应答CA。 */
|
||
IEC104_InterrogationRequest_t info; /* 唯一业务元数据;供事件交付及批次校验,终结时仍描述原激活。 */
|
||
IEC104_Station_t *station; /* 当前站额外引用,独立于广播数组持有的引用和CA路由资格。 */
|
||
IEC104_StationSnapshot_t stations; /* 单目标广播/错误广播的遍历快照;正常多目标广播使用每站状态数组。 */
|
||
uint64_t request_id; /* 激活任务编号;未建立总召事务的单帧应答为0。 */
|
||
uint64_t started_ms, application_since_ms; /* 单调毫秒:BEGIN生成、当前工作事件入队。 */
|
||
size_t batch_index, objects_submitted; /* 已被发送层接手的批次数、对象总数。 */
|
||
IEC104_InterrogationPhase_t phase; /* 当前站下一步动作;整轮广播阶段另由broadcast.stage管理。 */
|
||
uint8_t reply_cause; /* 非零为单帧确认/错误应答,肯定撤销确认也使用此路径。 */
|
||
bool reply_negative; /* REPLY阶段发送的P/N标志;肯定撤销确认为false。 */
|
||
bool data_send_waiting; /* 本批已及时提供有效数据但发送WAIT;不缓存原文,不再计本批取数期限。 */
|
||
bool send_waiting; /* 当前帧提交曾返回WAIT;无新的本地工作时等网络推进后再重试。 */
|
||
bool event_outstanding, event_delivered; /* 当前交互待应答、该工作事件已被应用取出。 */
|
||
bool begin_delivered, finish_reserved; /* 应用已取过当前BEGIN;finish_reserved仅用于单目标终结预留。 */
|
||
bool occupied; /* 固定任务槽正在使用;任务可非队首完成,释放后立即复用此槽。 */
|
||
} InterrogationTask;
|
||
|
||
/* 整轮依次完成全部站确认、逐站取数、全部站终止;FINISH只等待终结事件交付和资源回收。 */
|
||
enum { BROADCAST_ACCEPT, BROADCAST_DATA, BROADCAST_TERMINATE, BROADCAST_FINISH };
|
||
/* ACCEPTED仅表示应用接受;CONFIRMED表示肯定ACT_CON已被发送层接手,尚不代表主站ACK。
|
||
* DATA_DONE等其他站取数完成后再发ACT_TERM;ENDED保留终结原因,不能立即回收状态项。 */
|
||
enum { STATION_PENDING, STATION_ACCEPTED, STATION_CONFIRMED, STATION_DATA_DONE, STATION_ENDED };
|
||
/* BEGIN_READ决定是否欠应用一次FINISHED;QUEUED防止重复入队,READ才表示已经交付。
|
||
* DEACTIVATE单独记录待发送COT=9,站点ENDED或FINISHED已读均不能代替撤销确认提交。 */
|
||
enum { STATION_BEGIN_READ = 1U, STATION_FINISH_QUEUED = 2U, STATION_FINISH_READ = 4U, STATION_DEACTIVATE = 8U };
|
||
|
||
/* 一个实例至多一轮广播,三阶段共用同一协议任务,应用可逐站复用取数游标。 */
|
||
typedef struct {
|
||
InterrogationTask *task; /* 三阶段复用同一固定槽,FINISHING期间仍占用。 */
|
||
size_t count, index; /* 已捕获目标数、当前站在调用方状态数组中的位置。 */
|
||
uint8_t stage; /* 整轮阶段;当前站的等待/发送动作保存在task->phase。 */
|
||
bool cancelling, cancel_all; /* 有一组撤销确认待提交;cancel_all区分整轮撤销与定向撤销。 */
|
||
} InterrogationBroadcast;
|
||
|
||
struct IEC104_Interrogation {
|
||
IEC104_Slave_t *slave; /* NULL为空闲槽;有效期间Slave必须存活。 */
|
||
const IEC104_InterrogationRules_t *rules; /* 创建时借用的只读种类规则,寿命覆盖实例。 */
|
||
IEC104_InterrogationConfig_t config; /* 配置对象按值复制,broadcast_states指向的数组仍借用调用方存储。 */
|
||
IEC104_InterrogationStatus_t status; /* 会话状态和累计统计;即时进度由Get_Status补齐。 */
|
||
IEC104_SlaveEventIdentity_t session; /* Slave校验后交付的服务身份和已处理的连接代次。 */
|
||
uint64_t instance_id, next_request_id;
|
||
InterrogationBroadcast broadcast;
|
||
/* instance_id区分总召静态槽的历次使用;next_request_id在本实例内递增,
|
||
* 两者共同标识一次总召,独立于Slave的服务实例和连接身份。 */
|
||
unsigned count, event_count, reserved_finishes;
|
||
/* count包含等待清理的广播任务;event_count仅统计已入FIFO的事件。
|
||
* reserved_finishes仅为单目标预留,二者之和不超过FIFO容量;广播延迟FINISHED另存每站状态。 */
|
||
unsigned order[IEC104_INTERROGATION_TASK_CAPACITY]; /* 有效任务槽下标,保持应答形成的先后顺序。 */
|
||
bool prefer_interrogation; /* 单帧提交后给可推进总召一次机会,跨Process返回保留。 */
|
||
InterrogationTask tasks[IEC104_INTERROGATION_TASK_CAPACITY]; /* 固定地址的协议任务池。 */
|
||
IEC104_InterrogationEvent_t events[IEC104_INTERROGATION_EVENT_CAPACITY]; /* 可移动的纯值事件FIFO。 */
|
||
};
|
||
static IEC104_Interrogation_t interrogation_pool[IEC104_INTERROGATION_CAPACITY];
|
||
/* 不随槽释放清零,避免旧事件误命中新实例;达到上限拒绝创建,不回绕。 */
|
||
static uint64_t instance_serial;
|
||
static bool Advance(IEC104_Interrogation_t *self, InterrogationTask *task);
|
||
static bool Next_Broadcast_Station(IEC104_Interrogation_t *self);
|
||
static void Flush_Broadcast_Finishes(IEC104_Interrogation_t *self);
|
||
|
||
/* 仅识别使用状态数组的正常多目标广播,单目标广播和错误广播不走此分支。 */
|
||
static bool Is_Broadcast(const IEC104_Interrogation_t *self, const InterrogationTask *task)
|
||
{
|
||
return task != NULL && self->broadcast.task == task;
|
||
}
|
||
|
||
static IEC104_InterrogationStationState_t *Broadcast_Current(IEC104_Interrogation_t *self)
|
||
{
|
||
return &self->config.broadcast_states[self->broadcast.index];
|
||
}
|
||
|
||
/* 切站或终结前保存共享任务的当前进度;这里只存统计,不保存应用游标或点值快照。 */
|
||
static void Save_Broadcast_Progress(IEC104_Interrogation_t *self)
|
||
{
|
||
IEC104_InterrogationStationState_t *entry = Broadcast_Current(self);
|
||
entry->_batch_index = self->broadcast.task->batch_index;
|
||
entry->_objects_submitted = self->broadcast.task->objects_submitted;
|
||
}
|
||
|
||
/* 检查句柄是否指向仍被占用的实例槽;先比较地址再解引用,NULL和池外指针返回false。
|
||
* 这里只验证实例占用,Slave的寿命仍由调用方保证。 */
|
||
static bool Owned(const IEC104_Interrogation_t *self)
|
||
{
|
||
size_t i;
|
||
for (i = 0U; i < IEC104_INTERROGATION_CAPACITY; ++i) {
|
||
if (self == &interrogation_pool[i]) { return self->slave != NULL; }
|
||
}
|
||
return false;
|
||
}
|
||
|
||
/* 取得固定任务槽并追加顺序索引;调用方完成请求复制,失败时须立即Remove_Task。
|
||
* 不移动已有任务,保持快照的_self校验及broadcast.task指针有效。 */
|
||
static InterrogationTask *Allocate_Task(IEC104_Interrogation_t *self)
|
||
{
|
||
unsigned slot;
|
||
if (self->count == IEC104_INTERROGATION_TASK_CAPACITY) { return NULL; }
|
||
for (slot = 0U; slot < IEC104_INTERROGATION_TASK_CAPACITY; ++slot) {
|
||
InterrogationTask *task = &self->tasks[slot];
|
||
if (!task->occupied) {
|
||
memset(task, 0, sizeof(*task));
|
||
task->occupied = true;
|
||
self->order[self->count++] = slot;
|
||
return task;
|
||
}
|
||
}
|
||
return NULL;
|
||
}
|
||
|
||
/* 分别归还当前站引用、遍历快照及每个广播状态项的引用,随后清空本轮状态。
|
||
* 当前站与对应状态项各持有一次引用,因此两次Release均须保留。 */
|
||
static void Release_Stations(IEC104_Interrogation_t *self, InterrogationTask *task)
|
||
{
|
||
size_t i;
|
||
if (task->station != NULL) {
|
||
(void)IEC104_Slave_Release_Station(self->slave, task->station);
|
||
task->station = NULL;
|
||
}
|
||
if (task->stations._owner != NULL) { (void)IEC104_Slave_StationSnapshot_End(&task->stations); }
|
||
if (Is_Broadcast(self, task)) {
|
||
for (i = 0U; i < self->broadcast.count; ++i) {
|
||
IEC104_InterrogationStationState_t *entry = &self->config.broadcast_states[i];
|
||
(void)IEC104_Slave_Release_Station(self->slave, entry->_station);
|
||
memset(entry, 0, sizeof(*entry));
|
||
}
|
||
memset(&self->broadcast, 0, sizeof(self->broadcast));
|
||
}
|
||
}
|
||
|
||
/* 仅压紧顺序索引,不移动其他任务;调用前须完成终结通知安排。
|
||
* 多目标广播须经Reap_Broadcast确认FINISHED已读完,才能释放任务及借用状态项。 */
|
||
static void Remove_Task(IEC104_Interrogation_t *self, InterrogationTask *task)
|
||
{
|
||
unsigned i;
|
||
for (i = 0U; i < self->count; ++i) {
|
||
if (&self->tasks[self->order[i]] == task) {
|
||
Release_Stations(self, task);
|
||
memset(task, 0, sizeof(*task));
|
||
memmove(self->order + i, self->order + i + 1U,
|
||
(self->count - i - 1U) * sizeof(self->order[0]));
|
||
self->order[--self->count] = 0U;
|
||
return;
|
||
}
|
||
}
|
||
}
|
||
|
||
/* 原总召转为拒绝/撤销确认时,复用原槽并按应答形成时刻放到顺序末尾。
|
||
* 满池也能转换;不让这条新应答越过已经等待的其他单帧应答。 */
|
||
static void Move_Reply_To_Tail(IEC104_Interrogation_t *self, const InterrogationTask *task)
|
||
{
|
||
unsigned i;
|
||
for (i = 0U; i < self->count; ++i) {
|
||
unsigned slot = self->order[i];
|
||
if (&self->tasks[slot] == task) {
|
||
memmove(self->order + i, self->order + i + 1U,
|
||
(self->count - i - 1U) * sizeof(self->order[0]));
|
||
self->order[self->count - 1U] = slot;
|
||
self->prefer_interrogation = false;
|
||
return;
|
||
}
|
||
}
|
||
}
|
||
|
||
/* 返回唯一未终结总召的固定槽下标;request_id为0查当前总召,非0核对指定请求。
|
||
* 无匹配返回TASK_CAPACITY;广播在当前站否定确认等待期间仍占总召资格。
|
||
* FINISHING不参与业务推进,接纳新请求时另检查broadcast.task,防止复用未清理的状态区。 */
|
||
static unsigned Find_Interrogation(const IEC104_Interrogation_t *self, uint64_t request_id)
|
||
{
|
||
unsigned i;
|
||
for (i = 0U; i < self->count; ++i) {
|
||
unsigned slot = self->order[i];
|
||
const InterrogationTask *task = &self->tasks[slot];
|
||
if (task->request_id != 0U && task->phase != IEC104_INTERROGATION_FINISHING &&
|
||
(task->phase != IEC104_INTERROGATION_REPLY || task->stations._owner != NULL || Is_Broadcast(self, task)) &&
|
||
(request_id == 0U || task->request_id == request_id)) { return slot; }
|
||
}
|
||
return IEC104_INTERROGATION_TASK_CAPACITY;
|
||
}
|
||
|
||
/* 最早形成的单帧应答;立即提交也必须遵守此顺序,不能越过旧REPLY。
|
||
* 广播撤销借用原任务发送一组确认,即使当前站phase未转REPLY,也按短应答参与排序。 */
|
||
static unsigned Find_Reply(const IEC104_Interrogation_t *self)
|
||
{
|
||
unsigned i;
|
||
for (i = 0U; i < self->count; ++i) {
|
||
unsigned slot = self->order[i];
|
||
if (self->tasks[slot].phase == IEC104_INTERROGATION_REPLY ||
|
||
(Is_Broadcast(self, &self->tasks[slot]) && self->broadcast.cancelling)) { return slot; }
|
||
}
|
||
return IEC104_INTERROGATION_TASK_CAPACITY;
|
||
}
|
||
|
||
/* 业务推进无需等待应用或事件空位时为true;发送容量仍由实际提交结果决定。 */
|
||
static bool Interrogation_Can_Progress(const IEC104_Interrogation_t *self, const InterrogationTask *task)
|
||
{
|
||
if (task->phase == IEC104_INTERROGATION_START || task->phase == IEC104_INTERROGATION_DATA) {
|
||
unsigned needed = task->phase == IEC104_INTERROGATION_DATA || Is_Broadcast(self, task) ? 1U : 2U;
|
||
return !task->event_outstanding &&
|
||
self->event_count + self->reserved_finishes + needed <= IEC104_INTERROGATION_EVENT_CAPACITY;
|
||
}
|
||
return task->phase != IEC104_INTERROGATION_FINISHING;
|
||
}
|
||
|
||
/* 单帧保持FIFO;双方可推进时先单帧再总召,轮转位置跨调用保存。
|
||
* 总召等应用或业务FIFO空位时只选单帧;均不能推进时返回容量哨兵。
|
||
* Process与状态查询共用选择规则;send_waiting不阻止网络推进后的显式重试。 */
|
||
static unsigned Select_Task(const IEC104_Interrogation_t *self)
|
||
{
|
||
unsigned slot = Find_Interrogation(self, 0U), reply_slot = Find_Reply(self);
|
||
if (slot != IEC104_INTERROGATION_TASK_CAPACITY &&
|
||
(self->tasks[slot].phase == IEC104_INTERROGATION_REPLY ||
|
||
(Is_Broadcast(self, &self->tasks[slot]) && self->broadcast.cancelling) ||
|
||
!Interrogation_Can_Progress(self, &self->tasks[slot]))) { slot = IEC104_INTERROGATION_TASK_CAPACITY; }
|
||
if (reply_slot != IEC104_INTERROGATION_TASK_CAPACITY &&
|
||
(!self->prefer_interrogation || slot == IEC104_INTERROGATION_TASK_CAPACITY)) { return reply_slot; }
|
||
return slot != IEC104_INTERROGATION_TASK_CAPACITY ? slot : reply_slot;
|
||
}
|
||
|
||
/* 将任务身份、请求元数据和当前进度复制成一条业务事件;本函数不执行应用代码。
|
||
* type决定应用需要接受、交付数据还是清理;reason仅在FINISHED时有业务意义。
|
||
* 调用方须已检查容量,或已将本任务的终结预留释放为可写位置。 */
|
||
static void Push_Event(IEC104_Interrogation_t *self, const InterrogationTask *task,
|
||
IEC104_InterrogationEventType_t type, IEC104_InterrogationFinishReason_t reason)
|
||
{
|
||
IEC104_InterrogationEvent_t *event = &self->events[self->event_count++];
|
||
memset(event, 0, sizeof(*event));
|
||
event->type = type;
|
||
event->instance_id = self->instance_id;
|
||
event->request_id = task->request_id;
|
||
event->request = task->info;
|
||
event->batch_index = task->batch_index;
|
||
event->objects_submitted = task->objects_submitted;
|
||
event->reason = reason;
|
||
}
|
||
|
||
/* 终结记录留在每站状态区,FIFO有空位才入队;2槽FIFO也可交付全部已配置目标的终结通知。
|
||
* 仅为应用已读BEGIN的站点生成一次FINISHED,未入队的记录保留到后续调用继续交付。
|
||
* 共享任务此时可能已切到其他站,故公共请求字段沿用task,身份/CA/进度须从原站状态覆盖。 */
|
||
static void Flush_Broadcast_Finishes(IEC104_Interrogation_t *self)
|
||
{
|
||
size_t i;
|
||
InterrogationTask *task = self->broadcast.task;
|
||
if (task == NULL) { return; }
|
||
for (i = 0U; i < self->broadcast.count &&
|
||
self->event_count + self->reserved_finishes < IEC104_INTERROGATION_EVENT_CAPACITY; ++i) {
|
||
IEC104_InterrogationStationState_t *entry = &self->config.broadcast_states[i];
|
||
IEC104_InterrogationEvent_t *event;
|
||
if (entry->_reason == IEC104_INTERROGATION_FINISH_NONE ||
|
||
(entry->_flags & (STATION_BEGIN_READ | STATION_FINISH_QUEUED)) != STATION_BEGIN_READ) { continue; }
|
||
Push_Event(self, task, IEC104_INTERROGATION_FINISHED, (IEC104_InterrogationFinishReason_t)entry->_reason);
|
||
event = &self->events[self->event_count - 1U];
|
||
event->request_id = entry->_request_id;
|
||
event->request.station_id = IEC104_Station_Get_Id(entry->_station);
|
||
event->request.common_address = IEC104_Station_Get_Common_Address(entry->_station);
|
||
event->batch_index = entry->_batch_index;
|
||
event->objects_submitted = entry->_objects_submitted;
|
||
entry->_flags |= STATION_FINISH_QUEUED;
|
||
}
|
||
}
|
||
|
||
/* 只计尚未入FIFO的终结记录;状态查询再加event_count,避免重复统计已排队的FINISHED。 */
|
||
static unsigned Pending_Broadcast_Finishes(const IEC104_Interrogation_t *self)
|
||
{
|
||
size_t i;
|
||
unsigned pending = 0U;
|
||
if (self->broadcast.task == NULL) { return 0U; }
|
||
for (i = 0U; i < self->broadcast.count; ++i) {
|
||
const IEC104_InterrogationStationState_t *entry = &self->config.broadcast_states[i];
|
||
if (entry->_reason != IEC104_INTERROGATION_FINISH_NONE &&
|
||
(entry->_flags & (STATION_BEGIN_READ | STATION_FINISH_QUEUED)) == STATION_BEGIN_READ) { ++pending; }
|
||
}
|
||
return pending;
|
||
}
|
||
|
||
/* 整轮结束、撤销确认提交完且每个已读BEGIN的FINISHED也已读出后,才归还状态区和引用。
|
||
* 入FIFO不等于交付;此函数可能清空broadcast.task,调用方不能假定原任务仍有效。 */
|
||
static void Reap_Broadcast(IEC104_Interrogation_t *self)
|
||
{
|
||
size_t i;
|
||
if (self->broadcast.task == NULL || self->broadcast.stage != BROADCAST_FINISH || self->broadcast.cancelling) { return; }
|
||
for (i = 0U; i < self->broadcast.count; ++i) {
|
||
uint8_t flags = self->config.broadcast_states[i]._flags;
|
||
if ((flags & (STATION_BEGIN_READ | STATION_FINISH_READ)) == STATION_BEGIN_READ) { return; }
|
||
}
|
||
Remove_Task(self, self->broadcast.task);
|
||
}
|
||
|
||
/* 取消当前请求尚未消费的工作事件,保留所有FINISHED;已交给应用的旧应答由身份/状态拒绝。 */
|
||
static void Remove_Work_Events(IEC104_Interrogation_t *self, InterrogationTask *task)
|
||
{
|
||
unsigned read, write = 0U;
|
||
for (read = 0U; read < self->event_count; ++read) {
|
||
const IEC104_InterrogationEvent_t *event = &self->events[read];
|
||
if (event->request_id == task->request_id && event->type != IEC104_INTERROGATION_FINISHED) { continue; }
|
||
self->events[write++] = *event;
|
||
}
|
||
self->event_count = write;
|
||
task->event_outstanding = task->event_delivered = false;
|
||
}
|
||
|
||
/* 可终结当前站或其他已确认站;仅当前站需要保存共享进度、撤销未完成的工作交互。
|
||
* 首次终结原因不被后续整轮取消覆盖;节点引用及待发撤销标志留到各自清理边界。 */
|
||
static void End_Broadcast_Station(IEC104_Interrogation_t *self, size_t index,
|
||
IEC104_InterrogationFinishReason_t reason)
|
||
{
|
||
IEC104_InterrogationStationState_t *entry = &self->config.broadcast_states[index];
|
||
if (entry->_state == STATION_ENDED) { return; }
|
||
if (index == self->broadcast.index) {
|
||
Save_Broadcast_Progress(self);
|
||
Remove_Work_Events(self, self->broadcast.task);
|
||
}
|
||
entry->_reason = (uint8_t)reason;
|
||
entry->_state = STATION_ENDED;
|
||
self->status.last_finish_reason = reason;
|
||
}
|
||
|
||
/* 结束一个总召的应用交互,reason保存为该请求的终结原因;不自行释放协议任务。
|
||
* 多目标广播把终结记录写入调用方状态区,FIFO有空位时才交付。
|
||
* 移除本请求尚未取出的BEGIN/NEED_DATA,保留其他事件和所有历史FINISHED。
|
||
* 单目标BEGIN已取出时,用预留位置交付一次FINISHED;BEGIN未取出时只撤销事件和预留,
|
||
* 因为应用尚未获得请求,无需通知它清理资源。finish_reserved同时防止重复终结。 */
|
||
static void Finish(IEC104_Interrogation_t *self, InterrogationTask *task,
|
||
IEC104_InterrogationFinishReason_t reason)
|
||
{
|
||
if (Is_Broadcast(self, task)) {
|
||
End_Broadcast_Station(self, self->broadcast.index, reason);
|
||
Flush_Broadcast_Finishes(self);
|
||
return;
|
||
}
|
||
if (task->request_id != 0U && task->phase != IEC104_INTERROGATION_REPLY) {
|
||
self->status.last_finish_reason = reason;
|
||
}
|
||
if (!task->finish_reserved) { return; }
|
||
Remove_Work_Events(self, task);
|
||
--self->reserved_finishes;
|
||
task->finish_reserved = false;
|
||
task->event_outstanding = task->event_delivered = false;
|
||
if (task->begin_delivered) {
|
||
Push_Event(self, task, IEC104_INTERROGATION_FINISHED, reason);
|
||
task->begin_delivered = false;
|
||
}
|
||
}
|
||
|
||
/* 肯定ACT_CON尚未被发送层接手时,结束本次业务并复用原槽保存否定确认。
|
||
* 应用主动拒绝、BEGIN失败以及START/CONFIRM超时共用此路径,保留各自终结原因。
|
||
* Finish清除未读工作事件并兑现终结预留;不取消其他任务、不额外占槽、不关闭连接。
|
||
* 新形成的否定确认排在已有REPLY之后;后续Process提交,WAIT时保留重试。 */
|
||
static void Reject_Request(IEC104_Interrogation_t *self, InterrogationTask *task,
|
||
IEC104_InterrogationFinishReason_t reason)
|
||
{
|
||
++self->status.rejected;
|
||
Finish(self, task, reason);
|
||
task->reply_cause = IEC60870_ASDU_COT_ACTIVATION_CON;
|
||
task->reply_negative = true;
|
||
task->phase = IEC104_INTERROGATION_REPLY;
|
||
task->send_waiting = false;
|
||
self->status.waiting = false;
|
||
Move_Reply_To_Tail(self, task);
|
||
}
|
||
|
||
/* 按reason结束本实例全部任务,不关闭Slave或连接;Slave已接手的帧不撤销。
|
||
* 单目标任务可立即释放,多目标留在FINISHING,待各站FINISHED读完再归还状态区。
|
||
* cancelled按取消的协议任务数累计,包含单帧应答;整轮广播只计一个,不按站点或事件计数。 */
|
||
static void Cancel(IEC104_Interrogation_t *self, IEC104_InterrogationFinishReason_t reason)
|
||
{
|
||
unsigned i = 0U;
|
||
while (i < self->count) {
|
||
InterrogationTask *task = &self->tasks[self->order[i]];
|
||
if (Is_Broadcast(self, task)) {
|
||
size_t station;
|
||
if (task->phase != IEC104_INTERROGATION_FINISHING) { ++self->status.cancelled; }
|
||
for (station = 0U; station < self->broadcast.count; ++station) {
|
||
End_Broadcast_Station(self, station, reason);
|
||
self->config.broadcast_states[station]._flags &= (uint8_t)~STATION_DEACTIVATE;
|
||
}
|
||
self->broadcast.stage = BROADCAST_FINISH;
|
||
self->broadcast.cancelling = false;
|
||
task->phase = IEC104_INTERROGATION_FINISHING;
|
||
task->send_waiting = task->data_send_waiting = false;
|
||
++i;
|
||
} else {
|
||
Finish(self, task, reason);
|
||
Remove_Task(self, task);
|
||
++self->status.cancelled;
|
||
}
|
||
}
|
||
Flush_Broadcast_Finishes(self);
|
||
Reap_Broadcast(self);
|
||
self->prefer_interrogation = false;
|
||
self->status.waiting = false;
|
||
}
|
||
|
||
/* 内部服务故障或肯定确认提交后的执行失败,关闭identity所属连接并按reason取消全部任务。
|
||
* identity可能位于任务中,必须先关闭原连接再清任务;旧代次不会关闭新连接。
|
||
* true包括成功关闭、原连接已断开和代次已过期;false表示底层关闭失败,任务仍已取消。 */
|
||
static bool Reject_Connection(IEC104_Interrogation_t *self, const IEC104_SlaveEventIdentity_t *identity,
|
||
IEC104_InterrogationFinishReason_t reason)
|
||
{
|
||
IEC104_TcpResult_t result = IEC104_Slave_Close_Connection_By_Identity(self->slave, identity, NULL);
|
||
Cancel(self, reason);
|
||
self->status.active = false;
|
||
return result == IEC104_TCP_OK || result == IEC104_TCP_NOT_CONNECTED ||
|
||
result == IEC104_TCP_STALE_GENERATION;
|
||
}
|
||
|
||
/* 按肯定确认的提交边界处理业务失败:START/CONFIRM仍可否定激活并保留连接;
|
||
* 肯定ACT_CON被发送层接手即越过边界:单目标转DATA,多目标标记CONFIRMED并继续确认其他站。
|
||
* DATA/TERMINATE失败沿用关闭原连接的策略,不补发成功ACT_TERM。
|
||
* 调用方已校验会话与任务;true表示失败已处理,并不表示否定确认已经提交。 */
|
||
static bool Fail_Interrogation(IEC104_Interrogation_t *self, InterrogationTask *task,
|
||
IEC104_InterrogationFinishReason_t reason)
|
||
{
|
||
if (task->phase == IEC104_INTERROGATION_START || task->phase == IEC104_INTERROGATION_CONFIRM) {
|
||
Reject_Request(self, task, reason);
|
||
return true;
|
||
}
|
||
return Reject_Connection(self, &task->request.identity, reason);
|
||
}
|
||
|
||
/* DEFER表示暂不能尝试提交,WAIT表示已尝试但发送层未接手;均保存原请求,或在满槽时丢弃并计数。
|
||
* 只使用现有HIGH接口,不越过Slave通知或旧REPLY,不改变总召数据的发送顺序。 */
|
||
typedef enum { REPLY_DEFER, REPLY_WAIT, REPLY_SENT, REPLY_FAILED } ImmediateReplyResult;
|
||
|
||
/* 为请求校验选择首个实际CA;广播任务随后建立快照逐站推进,未知CA保留原值。 */
|
||
static uint16_t Resolve_Common_Address(const IEC104_Interrogation_t *self, uint16_t address)
|
||
{
|
||
if (self->config.use_station_registry) {
|
||
IEC104_Station_t *station = address == IEC104_CA_BROADCAST ?
|
||
IEC104_Slave_Next_Station(self->slave, NULL) : IEC104_Slave_Get_Station(self->slave, address);
|
||
return station != NULL ? IEC104_Station_Get_Common_Address(station) : address;
|
||
}
|
||
return address == IEC104_CA_BROADCAST ? self->config.common_address : address;
|
||
}
|
||
|
||
/* 保存可明确识别的单对象Type100及原身份;IOA/QOI必须完整,零对象/多对象不在此构造应答。
|
||
* 应答固定为10字节、VSQ=1,保留唯一对象、OA和TEST;CA按原规则处理广播。
|
||
* 调用方须记住原SQ/尾字节错误并仅作否定应答,不能把规范化副本当作合法激活或撤销执行。 */
|
||
static bool Copy_Response_Request(IEC104_Interrogation_t *self, InterrogationResponse *destination,
|
||
const IEC104_SlaveEventIdentity_t *identity, const uint8_t *data, uint16_t length, uint16_t response_ca)
|
||
{
|
||
IEC60870_Asdu_t asdu;
|
||
if (length < sizeof(destination->bytes) || data[0] != self->rules->type_id ||
|
||
(data[1] & 0x7fU) != 1U) { return false; } /* VSQ低7位为对象数。 */
|
||
destination->identity = *identity;
|
||
memcpy(destination->bytes, data, sizeof(destination->bytes));
|
||
destination->bytes[1] = 1U;
|
||
return IEC60870_Asdu_Create_From_Buffer(&asdu, IEC104_Slave_Get_Application_Layer_Parameters(self->slave),
|
||
destination->bytes, sizeof(destination->bytes)) == IEC60870_ASDU_OK &&
|
||
IEC60870_Asdu_Set_CA(&asdu, response_ca) == IEC60870_ASDU_OK;
|
||
}
|
||
|
||
/* 选中节点后持有引用;删除只退出路由,当前事务仍使用这个注册编号及context。 */
|
||
/* 当前任务持有原节点,并把请求副本CA改为实际目标站;连接身份、OA、QOI等保持原请求语义。 */
|
||
static bool Bind_Station(IEC104_Interrogation_t *self, InterrogationTask *task, IEC104_Station_t *station)
|
||
{
|
||
IEC60870_Asdu_t asdu;
|
||
if (station == NULL || IEC104_Slave_Acquire_Station(self->slave, station) != IEC104_STATION_OK) { return false; }
|
||
task->station = station;
|
||
task->info.common_address = IEC104_Station_Get_Common_Address(station);
|
||
task->info.station_id = IEC104_Station_Get_Id(station);
|
||
return IEC60870_Asdu_Create_From_Buffer(&asdu, IEC104_Slave_Get_Application_Layer_Parameters(self->slave),
|
||
task->request.bytes, sizeof(task->request.bytes)) == IEC60870_ASDU_OK &&
|
||
IEC60870_Asdu_Set_CA(&asdu, task->info.common_address) == IEC60870_ASDU_OK;
|
||
}
|
||
|
||
/* 在同一任务槽内切站:归还当前额外引用,再从状态项恢复本站身份、起始时间和提交进度。
|
||
* 状态项本身仍持有原节点,故删除或同CA替换不会使切换误绑定新站;此处不重新按CA查找。
|
||
* 每站只恢复协议状态,BEGIN/NEED_DATA的交互标志重新开始,应用取数游标由应用另行维护。 */
|
||
static bool Select_Broadcast_Station(IEC104_Interrogation_t *self, size_t index)
|
||
{
|
||
InterrogationTask *task = self->broadcast.task;
|
||
IEC104_InterrogationStationState_t *entry = &self->config.broadcast_states[index];
|
||
if (task->station != NULL) {
|
||
(void)IEC104_Slave_Release_Station(self->slave, task->station);
|
||
task->station = NULL;
|
||
}
|
||
self->broadcast.index = index;
|
||
if (!Bind_Station(self, task, entry->_station)) { return false; }
|
||
task->request_id = entry->_request_id;
|
||
task->started_ms = entry->_started_ms;
|
||
task->application_since_ms = 0U;
|
||
task->batch_index = entry->_batch_index;
|
||
task->objects_submitted = entry->_objects_submitted;
|
||
task->phase = self->broadcast.stage == BROADCAST_ACCEPT ? IEC104_INTERROGATION_START :
|
||
self->broadcast.stage == BROADCAST_DATA ? IEC104_INTERROGATION_DATA : IEC104_INTERROGATION_TERMINATE;
|
||
task->reply_cause = 0U;
|
||
task->reply_negative = task->send_waiting = task->data_send_waiting = false;
|
||
task->event_outstanding = task->event_delivered = task->finish_reserved = false;
|
||
task->begin_delivered = (entry->_flags & STATION_BEGIN_READ) != 0U;
|
||
return true;
|
||
}
|
||
|
||
/* 接收时一次取得所有节点引用;后续确认、数据、终止使用同一集合,不混入新增/替代节点。
|
||
* 入口已检查整轮容量和请求编号余量;本段串行执行且不调用应用,捕获期间注册集合不变。
|
||
* 每成功持有一项便计入count,部分失败时由整轮取消/回收路径归还已取得的引用。 */
|
||
static bool Capture_Broadcast(IEC104_Interrogation_t *self, InterrogationTask *task)
|
||
{
|
||
IEC104_Station_t *station = NULL;
|
||
self->broadcast.task = task;
|
||
while ((station = IEC104_Slave_Next_Station(self->slave, station)) != NULL) {
|
||
IEC104_InterrogationStationState_t *entry;
|
||
if (self->broadcast.count == self->config.broadcast_capacity ||
|
||
IEC104_Slave_Acquire_Station(self->slave, station) != IEC104_STATION_OK) { return false; }
|
||
entry = &self->config.broadcast_states[self->broadcast.count];
|
||
memset(entry, 0, sizeof(*entry));
|
||
entry->_station = station;
|
||
entry->_request_id = self->broadcast.count == 0U ? task->request_id : ++self->next_request_id;
|
||
++self->broadcast.count;
|
||
}
|
||
return self->broadcast.count != 0U && Select_Broadcast_Station(self, 0U);
|
||
}
|
||
|
||
/* 先完成当前阶段的余下站点,再从数组首项进入下一阶段,保证全部确认先于数据、全部数据先于终止。
|
||
* 仅跳过尚未选中便被删除的未来站;已确认站即使退役,也按原引用继续完成或显式终结。
|
||
* 最后一遍完成后进入FINISHING;若没有待交付终结事件,本次调用即可回收任务。 */
|
||
static bool Next_Broadcast_Station(IEC104_Interrogation_t *self)
|
||
{
|
||
size_t index = self->broadcast.index + 1U;
|
||
while (self->broadcast.stage < BROADCAST_FINISH) {
|
||
for (; index < self->broadcast.count; ++index) {
|
||
IEC104_InterrogationStationState_t *entry = &self->config.broadcast_states[index];
|
||
if (self->broadcast.stage == BROADCAST_ACCEPT && entry->_state == STATION_PENDING &&
|
||
!IEC104_Station_Is_Registered(entry->_station)) {
|
||
entry->_state = STATION_ENDED; /* 未选择/未确认的退役站不启动,不发成功终止。 */
|
||
}
|
||
if ((self->broadcast.stage == BROADCAST_ACCEPT && entry->_state == STATION_PENDING) ||
|
||
(self->broadcast.stage == BROADCAST_DATA && entry->_state == STATION_CONFIRMED) ||
|
||
(self->broadcast.stage == BROADCAST_TERMINATE && entry->_state == STATION_DATA_DONE)) {
|
||
return Select_Broadcast_Station(self, index);
|
||
}
|
||
}
|
||
++self->broadcast.stage;
|
||
index = 0U;
|
||
}
|
||
self->broadcast.task->phase = IEC104_INTERROGATION_FINISHING;
|
||
self->broadcast.task->send_waiting = false;
|
||
Flush_Broadcast_Finishes(self);
|
||
Reap_Broadcast(self); /* 无待读终结记录时立即回收,不依赖下一轮网络唤醒。 */
|
||
return true;
|
||
}
|
||
|
||
/* 已确认但尚未轮到取数、或等最后终止的站也仍是活动事务;尚未选中的PENDING站不算活动。 */
|
||
static bool Broadcast_Station_Active(const IEC104_Interrogation_t *self, size_t index)
|
||
{
|
||
const IEC104_InterrogationStationState_t *entry = &self->config.broadcast_states[index];
|
||
return entry->_state != STATION_ENDED &&
|
||
(entry->_state != STATION_PENDING || index == self->broadcast.index);
|
||
}
|
||
|
||
/* 撤销匹配本轮原目标,允许找到已退役但仍活动的站;广播地址匹配任一活动站,count表示无匹配。 */
|
||
static size_t Find_Broadcast_Activation(const IEC104_Interrogation_t *self, uint16_t common_address)
|
||
{
|
||
size_t i;
|
||
for (i = 0U; i < self->broadcast.count; ++i) {
|
||
if (Broadcast_Station_Active(self, i) &&
|
||
(common_address == IEC104_CA_BROADCAST ||
|
||
IEC104_Station_Get_Common_Address(self->config.broadcast_states[i]._station) == common_address)) { return i; }
|
||
}
|
||
return self->broadcast.count;
|
||
}
|
||
|
||
static size_t Pending_Broadcast_Deactivation(const IEC104_Interrogation_t *self)
|
||
{
|
||
size_t i;
|
||
for (i = 0U; i < self->broadcast.count; ++i) {
|
||
if ((self->config.broadcast_states[i]._flags & STATION_DEACTIVATE) != 0U) { return i; }
|
||
}
|
||
return self->broadcast.count;
|
||
}
|
||
|
||
/* 在原任务内保存撤销应答,不要求额外队列槽;第二条重叠撤销在此应答完成前按未匹配拒绝。
|
||
* 先记录各站原活动状态再终结,只有活动站欠一次肯定COT=9;全局撤销也结束尚未启动的未来站。
|
||
* 定向撤销保留其他站进度,待这组撤销确认提交后继续原阶段;不覆盖其他站已有终结原因。 */
|
||
static void Deactivate_Broadcast(IEC104_Interrogation_t *self, uint16_t common_address)
|
||
{
|
||
size_t i;
|
||
self->broadcast.cancel_all = common_address == IEC104_CA_BROADCAST;
|
||
for (i = 0U; i < self->broadcast.count; ++i) {
|
||
IEC104_InterrogationStationState_t *entry = &self->config.broadcast_states[i];
|
||
bool active = Broadcast_Station_Active(self, i);
|
||
if (self->broadcast.cancel_all ||
|
||
(active && IEC104_Station_Get_Common_Address(entry->_station) == common_address)) {
|
||
if (active) { entry->_flags |= STATION_DEACTIVATE; }
|
||
End_Broadcast_Station(self, i, IEC104_INTERROGATION_FINISH_DEACTIVATED);
|
||
}
|
||
}
|
||
self->broadcast.cancelling = true;
|
||
self->broadcast.task->send_waiting = false;
|
||
Move_Reply_To_Tail(self, self->broadcast.task);
|
||
Flush_Broadcast_Finishes(self);
|
||
++self->status.deactivated;
|
||
++self->status.cancelled;
|
||
}
|
||
|
||
/* 只在对应COT=9被发送层接手后清标志;最后一条完成后收尾整轮,或恢复未被定向撤销的工作。 */
|
||
static bool Advance_Broadcast_Deactivation(IEC104_Interrogation_t *self, size_t index)
|
||
{
|
||
self->config.broadcast_states[index]._flags &= (uint8_t)~STATION_DEACTIVATE;
|
||
self->broadcast.task->send_waiting = false;
|
||
if (Pending_Broadcast_Deactivation(self) != self->broadcast.count) { return true; }
|
||
self->broadcast.cancelling = false;
|
||
if (self->broadcast.cancel_all) {
|
||
self->broadcast.stage = BROADCAST_FINISH;
|
||
self->broadcast.task->phase = IEC104_INTERROGATION_FINISHING;
|
||
Reap_Broadcast(self);
|
||
return true;
|
||
}
|
||
return Broadcast_Current(self)->_state == STATION_ENDED ? Next_Broadcast_Station(self) : true;
|
||
}
|
||
|
||
/* 每次发送只建立临时ASDU视图;WAIT保留紧凑原文和身份,输入Slave事件始终不变。 */
|
||
static bool Prepare_Response(IEC104_Interrogation_t *self, InterrogationResponse *request,
|
||
uint8_t cause, bool negative, IEC60870_Asdu_t *asdu)
|
||
{
|
||
return IEC60870_Asdu_Create_From_Buffer(asdu, IEC104_Slave_Get_Application_Layer_Parameters(self->slave),
|
||
request->bytes, sizeof(request->bytes)) == IEC60870_ASDU_OK &&
|
||
IEC60870_Asdu_Set_COT(asdu, cause) == IEC60870_ASDU_OK &&
|
||
IEC60870_Asdu_Set_Negative(asdu, negative) == IEC60870_ASDU_OK;
|
||
}
|
||
|
||
/* 仅对ERROR复核正常STOP拒绝,避免在消费DEACTIVATED前误记SEND_ERROR。
|
||
* 须同时确认原代次仍连接、会话/队列无故障、传输已停用且有未读通知;普通新通知不足以放行。
|
||
* OK/WAIT沿原路径处理,FATAL可能已发送字节,绝不降级为等待;本函数不消费通知或重置期限。 */
|
||
static bool Defer_For_Stop(IEC104_Interrogation_t *self,
|
||
const IEC104_SlaveEventIdentity_t *identity, IEC104_TcpSubmitResult_t submitted)
|
||
{
|
||
IEC104_TcpServerStatus_t status;
|
||
if (submitted != IEC104_TCP_SUBMIT_ERROR) { return false; }
|
||
if (identity->slave != self->slave || identity->instance_id != self->session.instance_id ||
|
||
identity->generation != self->session.generation ||
|
||
IEC104_Slave_Get_Status(self->slave, &status, NULL) != IEC104_TCP_OK ||
|
||
status.generation != identity->generation || !status.connected || !status.connection_status_valid ||
|
||
!status.asdu_queue_status_valid || status.connection.phase != IEC104_TCP_SESSION_ACTIVE ||
|
||
status.connection.fault != IEC104_TCP_FAULT_NONE || status.connection.data_transfer_started ||
|
||
status.pending_event_count == 0U) { return false; }
|
||
self->status.events_pending = true;
|
||
return true;
|
||
}
|
||
|
||
static ImmediateReplyResult Try_Immediate_Reply(IEC104_Interrogation_t *self,
|
||
InterrogationResponse *request,
|
||
uint8_t cause, bool negative, const InterrogationTask *queued)
|
||
{
|
||
IEC60870_Asdu_t asdu;
|
||
IEC104_TcpSubmitResult_t submitted;
|
||
IEC104_SlaveRequestState_t state;
|
||
unsigned first = Find_Reply(self);
|
||
if (first != IEC104_INTERROGATION_TASK_CAPACITY && &self->tasks[first] != queued) { return REPLY_DEFER; }
|
||
state = IEC104_Slave_Check_Request(self->slave, &self->session);
|
||
self->status.events_pending = state == IEC104_SLAVE_REQUEST_WAIT_EVENTS;
|
||
if (self->status.events_pending) { return REPLY_DEFER; }
|
||
if (state == IEC104_SLAVE_REQUEST_ERROR || !self->status.active || state != IEC104_SLAVE_REQUEST_READY) {
|
||
Cancel(self, state == IEC104_SLAVE_REQUEST_ERROR ? IEC104_INTERROGATION_FINISH_SERVICE_ERROR :
|
||
IEC104_INTERROGATION_FINISH_CONNECTION_CHANGED);
|
||
self->status.active = false;
|
||
return REPLY_FAILED;
|
||
}
|
||
if (!Prepare_Response(self, request, cause, negative, &asdu)) { return REPLY_FAILED; }
|
||
submitted = IEC104_Slave_Send_Response(self->slave, &request->identity, &asdu);
|
||
if (submitted == IEC104_TCP_SUBMIT_OK) {
|
||
++self->status.submitted;
|
||
if (negative) { ++self->status.negative_submitted; }
|
||
self->prefer_interrogation = true;
|
||
return REPLY_SENT;
|
||
}
|
||
if (submitted == IEC104_TCP_SUBMIT_WAIT) {
|
||
++self->status.waits;
|
||
self->status.waiting = true;
|
||
return REPLY_WAIT;
|
||
}
|
||
if (Defer_For_Stop(self, &request->identity, submitted)) { return REPLY_DEFER; }
|
||
Cancel(self, IEC104_INTERROGATION_FINISH_SEND_ERROR);
|
||
self->status.active = false;
|
||
return REPLY_FAILED;
|
||
}
|
||
|
||
/* 单次只读期限计算的结果,不缓存到任务;已确认广播站总超时决定整轮必须断连。 */
|
||
typedef struct {
|
||
IEC104_InterrogationFinishReason_t reason;
|
||
uint32_t remaining_ms;
|
||
bool confirmed_total_expired;
|
||
} InterrogationTimeout;
|
||
|
||
/* 将一项期限并入当前最近期限:now/since为单调毫秒,duration为允许时长。
|
||
* timeout是累积输入输出;reason为NONE表示此前还没有有效候选。
|
||
* 用经过时间比较,避免绝对期限相加溢出;0时长禁用,到期剩余值为0。
|
||
* 相同剩余时间保留先检查的原因,因此Task_Timeout先检查总期限以确定并发超时的归因。 */
|
||
static void Limit_Timeout(uint64_t now, uint64_t since, uint32_t duration,
|
||
IEC104_InterrogationFinishReason_t candidate, InterrogationTimeout *timeout)
|
||
{
|
||
uint64_t elapsed;
|
||
uint32_t left;
|
||
if (duration == 0U) { return; }
|
||
elapsed = now - since;
|
||
left = elapsed >= duration ? 0U : duration - (uint32_t)elapsed;
|
||
if (timeout->reason == IEC104_INTERROGATION_FINISH_NONE || left < timeout->remaining_ms) {
|
||
timeout->remaining_ms = left;
|
||
timeout->reason = candidate;
|
||
}
|
||
}
|
||
|
||
/* 只读计算task在同一now时刻的最近期限及已确认站超时标志;reason为NONE表示没有启用的期限。
|
||
* 只有尚未终结的激活业务受定时器约束,单帧应答不参加业务超时计算。
|
||
* START尚未生成BEGIN时不计业务期限;读取任务的总期限从BEGIN生成起覆盖取数及发送等待。
|
||
* 动作命令接受后已执行副作用,CONFIRM不再计业务期限;发送层仍处理连接期限。
|
||
* BEGIN/NEED_DATA期限从工作事件入队起计算;首次有效批次发送WAIT后不再计本批取数期限。
|
||
* 本函数不触发超时动作,供Expire执行处理和Get_Status计算唤醒时间共用。 */
|
||
static InterrogationTimeout Task_Timeout(const IEC104_Interrogation_t *self,
|
||
const InterrogationTask *task, uint64_t now)
|
||
{
|
||
InterrogationTimeout timeout = {IEC104_INTERROGATION_FINISH_NONE, 0U, false};
|
||
if (Is_Broadcast(self, task)) {
|
||
size_t i;
|
||
/* 当前任务可能已切到后续站的BEGIN,先前已确认站的总期限仍独立有效,不能只查当前站。 */
|
||
for (i = 0U; i < self->broadcast.count; ++i) {
|
||
const IEC104_InterrogationStationState_t *entry = &self->config.broadcast_states[i];
|
||
if (entry->_state == STATION_CONFIRMED || entry->_state == STATION_DATA_DONE) {
|
||
Limit_Timeout(now, entry->_started_ms, self->config.total_timeout_ms,
|
||
IEC104_INTERROGATION_FINISH_TOTAL_TIMEOUT, &timeout);
|
||
}
|
||
}
|
||
/* 此时只汇入了已确认站的总期限;先保存到期事实,再合入当前站的期限。 */
|
||
timeout.confirmed_total_expired = timeout.reason == IEC104_INTERROGATION_FINISH_TOTAL_TIMEOUT &&
|
||
timeout.remaining_ms == 0U;
|
||
if (self->config.broadcast_states[self->broadcast.index]._state == STATION_ENDED) { return timeout; }
|
||
}
|
||
if (task->request_id == 0U || task->phase == IEC104_INTERROGATION_REPLY ||
|
||
task->phase == IEC104_INTERROGATION_FINISHING) { return timeout; }
|
||
if (task->phase == IEC104_INTERROGATION_START && !task->event_outstanding) { return timeout; }
|
||
/* 已接受的动作可能已经改变计数器,不能因确认排队超时再改发否定确认。
|
||
* 这里只停止当前动作站的业务期限;传输期限、STOP和断线仍有效。 */
|
||
if (task->phase == IEC104_INTERROGATION_CONFIRM && !self->rules->has_data(task->info.freeze)) { return timeout; }
|
||
Limit_Timeout(now, task->started_ms, self->config.total_timeout_ms,
|
||
IEC104_INTERROGATION_FINISH_TOTAL_TIMEOUT, &timeout);
|
||
if (task->event_outstanding && !task->data_send_waiting) {
|
||
bool begin = task->phase == IEC104_INTERROGATION_START;
|
||
Limit_Timeout(now, task->application_since_ms,
|
||
begin ? self->config.begin_timeout_ms : self->config.data_timeout_ms,
|
||
begin ? IEC104_INTERROGATION_FINISH_BEGIN_TIMEOUT : IEC104_INTERROGATION_FINISH_DATA_TIMEOUT,
|
||
&timeout);
|
||
}
|
||
return timeout;
|
||
}
|
||
|
||
/* 只检查唯一未终结总召;确认前超时转为否定REPLY,确认后超时关闭原连接并取消。
|
||
* 已交付BEGIN的FINISHED有预留槽或每站记录,迟到应答随即失效;已终结站不再计算期限。
|
||
* 调用方须先确认Slave通知已读空,使先到的STOP/断线/撤销先于超时处理。
|
||
* 返回0为未超时,1为已处理超时,-1为任务已取消但底层关闭失败。 */
|
||
static int Expire(IEC104_Interrogation_t *self)
|
||
{
|
||
unsigned slot = Find_Interrogation(self, 0U);
|
||
InterrogationTask *task;
|
||
InterrogationTimeout timeout;
|
||
if (slot == IEC104_INTERROGATION_TASK_CAPACITY) { return 0; }
|
||
task = &self->tasks[slot];
|
||
timeout = Task_Timeout(self, task, IEC104_Net_Now_Ms());
|
||
if (timeout.reason != IEC104_INTERROGATION_FINISH_NONE && timeout.remaining_ms == 0U) {
|
||
++self->status.timeouts;
|
||
if (timeout.confirmed_total_expired) {
|
||
/* 若超时的是先前已确认站,即使当前站尚在START,也必须按确认后失败结束原连接。 */
|
||
return Reject_Connection(self, &task->request.identity, timeout.reason) ? 1 : -1;
|
||
}
|
||
return Fail_Interrogation(self, task, timeout.reason) ? 1 : -1;
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
/* 查找唯一未终结总召;它可能等待BEGIN或排在错误应答后面,不能只看任务队首。
|
||
* event/info/qualifier来自已校验的撤销请求,response_ca已解析为实际站CA;
|
||
* 原连接代次、实际站CA、OA、TEST和QOI须全部相同。
|
||
* 广播与定向地址解析到同一本站,允许广播激活后定向撤销,也允许广播撤销本站任务。
|
||
* 返回内部任务指针供Request原位改为撤销确认;NULL表示没有匹配的未终结激活任务。 */
|
||
static InterrogationTask *Find_Activation(IEC104_Interrogation_t *self,
|
||
const IEC104_SlaveEvent_t *event, const IEC60870_AsduInfo_t *info,
|
||
uint16_t response_ca, uint8_t qualifier)
|
||
{
|
||
unsigned i;
|
||
for (i = 0U; i < self->count; ++i) {
|
||
InterrogationTask *task = &self->tasks[self->order[i]];
|
||
if (Is_Broadcast(self, task)) {
|
||
if (!self->broadcast.cancelling && task->phase != IEC104_INTERROGATION_FINISHING &&
|
||
task->info.generation == event->transport.generation &&
|
||
task->info.originator_address == info->originator_address &&
|
||
task->info.is_test == (info->is_test != 0U) && task->info.qualifier == qualifier &&
|
||
Find_Broadcast_Activation(self, info->common_address) != self->broadcast.count) { return task; }
|
||
continue;
|
||
}
|
||
if (task->request_id != 0U &&
|
||
(task->phase != IEC104_INTERROGATION_REPLY ||
|
||
(task->stations._owner != NULL && info->common_address == IEC104_CA_BROADCAST)) &&
|
||
task->info.generation == event->transport.generation &&
|
||
(task->info.common_address == response_ca || info->common_address == IEC104_CA_BROADCAST) &&
|
||
task->info.originator_address == info->originator_address &&
|
||
task->info.is_test == (info->is_test != 0U) && task->info.qualifier == qualifier) { return task; }
|
||
}
|
||
return NULL;
|
||
}
|
||
|
||
/* 解析一个I帧通知中的总召请求,完成布局、地址、QOI和事务状态校验。
|
||
* 未启用时或有效的其他Type返回IGNORED;接手、拒绝或满槽丢弃请求均返回HANDLED。
|
||
* 公共头不可用交公共入口处理;总召对象不明确则丢弃,单对象完整但布局非法则仅否定。
|
||
* 请求错误和容量耗尽均不取消已有任务;只有原会话或内部服务故障沿用原失败策略。
|
||
* 单帧应答先尝试安全提交,未接手才保存;匹配的撤销复用原任务槽。
|
||
* 只有空闲时才建立START任务;此时已占用总召资格,Process有事件空位后生成BEGIN。 */
|
||
static IEC104_InterrogationEventResult_t Request(IEC104_Interrogation_t *self,
|
||
const IEC104_SlaveEvent_t *event, const IEC104_SlaveEventIdentity_t *identity)
|
||
{
|
||
IEC60870_AsduInfo_t info;
|
||
uint32_t ioa;
|
||
uint8_t qualifier, freeze;
|
||
InterrogationResponse response;
|
||
const uint8_t *data;
|
||
uint16_t length;
|
||
uint16_t response_ca;
|
||
uint8_t reply_cause = 0U;
|
||
bool deactivation;
|
||
bool overload_counted = false;
|
||
bool send_waiting = false;
|
||
bool broadcast;
|
||
bool known_address;
|
||
bool invalid_layout;
|
||
size_t station_count;
|
||
InterrogationTask *activation;
|
||
InterrogationTask *task;
|
||
if (!self->status.active) { return IEC104_INTERROGATION_EVENT_IGNORED; }
|
||
/* 来源/身份已由Handle_Event验证;先保护长度再读取类型,不替其他业务处理ASDU错误。 */
|
||
if (event->transport.data.i_frame.apdu_length <= IEC104_APCI_LEN ||
|
||
event->transport.data.i_frame.apdu_length > sizeof(event->transport.data.i_frame.apdu) ||
|
||
event->transport.data.i_frame.apdu[IEC104_APCI_LEN] != self->rules->type_id) {
|
||
return IEC104_INTERROGATION_EVENT_IGNORED;
|
||
}
|
||
if (event->asdu_result != IEC60870_ASDU_OK ||
|
||
IEC60870_Asdu_Get_Data(&event->asdu, &data, &length) != IEC60870_ASDU_OK ||
|
||
!IEC60870_Parse_ASDU_Header(event->asdu.parameters, data, length, &info)) {
|
||
return IEC104_INTERROGATION_EVENT_IGNORED;
|
||
}
|
||
if (info.object_count != 1U || info.information_objects_len < 3U + 1U) {
|
||
/* 无唯一完整IOA/QOI,不能猜测或选取多个对象之一应答;HANDLED防止进入未知类型兜底。 */
|
||
++self->status.rejected;
|
||
return IEC104_INTERROGATION_EVENT_HANDLED;
|
||
}
|
||
invalid_layout = info.is_sequence != 0U || length != sizeof(response.bytes);
|
||
if (invalid_layout) {
|
||
/* 只为拒绝应答读取唯一对象;不修改原文。原布局错误必须在匹配撤销/建立激活前拦截。 */
|
||
info.is_sequence = 0U;
|
||
info.information_objects_len = 3U + 1U;
|
||
}
|
||
if ((!invalid_layout && IEC60870_Asdu_Parse(event->asdu.parameters, data, length, &info) != IEC60870_ASDU_OK) ||
|
||
!self->rules->decode(&info, &ioa, &qualifier, &freeze)) {
|
||
++self->status.rejected;
|
||
return IEC104_INTERROGATION_EVENT_HANDLED;
|
||
}
|
||
/* 只决定应答地址,不改变原请求;后续仍按COT优先级决定是否接受及如何否定。 */
|
||
response_ca = Resolve_Common_Address(self, info.common_address);
|
||
deactivation = self->rules->has_deactivation && info.cause_of_transmission == IEC60870_ASDU_COT_DEACTIVATION;
|
||
activation = deactivation && !invalid_layout ? Find_Activation(self, event, &info, response_ca, qualifier) : NULL;
|
||
if (activation != NULL) {
|
||
response_ca = Is_Broadcast(self, activation) && info.common_address != IEC104_CA_BROADCAST ?
|
||
info.common_address : activation->info.common_address;
|
||
}
|
||
station_count = IEC104_Slave_Get_Station_Count(self->slave);
|
||
broadcast = self->config.use_station_registry && info.common_address == IEC104_CA_BROADCAST &&
|
||
station_count != 0U;
|
||
known_address = activation != NULL || (self->config.use_station_registry ?
|
||
IEC104_Slave_Get_Station(self->slave, response_ca) != NULL :
|
||
response_ca == self->config.common_address &&
|
||
(self->config.station == NULL || IEC104_Station_Is_Registered(self->config.station)));
|
||
if (!Copy_Response_Request(self, &response, identity, data, length, response_ca)) {
|
||
return IEC104_INTERROGATION_EVENT_ERROR;
|
||
}
|
||
/* 无歧义对象才可回应;错误原因优先级COT -> CA -> IOA -> 布局/QOI/P/N/事务状态。 */
|
||
if (info.cause_of_transmission != IEC60870_ASDU_COT_ACTIVATION && !deactivation) {
|
||
reply_cause = IEC60870_ASDU_COT_UNKNOWN_CAUSE;
|
||
} else if (!known_address) {
|
||
reply_cause = IEC60870_ASDU_COT_UNKNOWN_COMMON_ADDRESS;
|
||
} else if (ioa != 0U) {
|
||
reply_cause = IEC60870_ASDU_COT_UNKNOWN_IOA;
|
||
} else if (invalid_layout || !self->rules->qualifier_valid(qualifier) || info.is_negative_confirm != 0U) {
|
||
reply_cause = deactivation ? IEC60870_ASDU_COT_DEACTIVATION_CON : IEC60870_ASDU_COT_ACTIVATION_CON;
|
||
} else if (deactivation) {
|
||
task = activation;
|
||
if (task != NULL) {
|
||
ImmediateReplyResult sent;
|
||
if (Is_Broadcast(self, task)) {
|
||
Deactivate_Broadcast(self, info.common_address);
|
||
return IEC104_INTERROGATION_EVENT_HANDLED;
|
||
}
|
||
/* 原位替换为撤销确认,不要求额外任务槽。
|
||
* 先保存撤销请求,Finish仍使用原激活的info/id/进度通知应用清理。
|
||
* 已提交HIGH的帧按原顺序在确认前发出;本地未提交批次就此丢弃。 */
|
||
task->request = response;
|
||
Finish(self, task, IEC104_INTERROGATION_FINISH_DEACTIVATED);
|
||
task->phase = IEC104_INTERROGATION_REPLY;
|
||
task->reply_cause = IEC60870_ASDU_COT_DEACTIVATION_CON;
|
||
task->reply_negative = false;
|
||
task->data_send_waiting = false;
|
||
task->send_waiting = false;
|
||
/* 广播撤销终结余下遍历;定向撤销只终结当前站,确认后继续下一站。 */
|
||
if (info.common_address == IEC104_CA_BROADCAST && task->stations._owner != NULL) {
|
||
(void)IEC104_Slave_StationSnapshot_End(&task->stations);
|
||
}
|
||
if (task->stations._owner == NULL) { task->request_id = 0U; }
|
||
Move_Reply_To_Tail(self, task);
|
||
self->status.waiting = false;
|
||
++self->status.deactivated;
|
||
++self->status.cancelled;
|
||
sent = Try_Immediate_Reply(self, &task->request, task->reply_cause, false, task);
|
||
if (sent == REPLY_SENT && !Advance(self, task)) { return IEC104_INTERROGATION_EVENT_ERROR; }
|
||
if (sent == REPLY_FAILED) { return IEC104_INTERROGATION_EVENT_ERROR; }
|
||
if (sent == REPLY_WAIT) { task->send_waiting = true; }
|
||
return IEC104_INTERROGATION_EVENT_HANDLED;
|
||
}
|
||
reply_cause = IEC60870_ASDU_COT_DEACTIVATION_CON;
|
||
} else if (Find_Interrogation(self, 0U) != IEC104_INTERROGATION_TASK_CAPACITY || self->broadcast.task != NULL) {
|
||
/* 从保存激活到终结全程占用总召资格;相同/不同QOI都否定,不替换原请求。
|
||
* FINISHING虽不再有取数任务,广播状态区仍被未读FINISHED占用,必须继续拒绝新激活。
|
||
* 此判断仅作用于Type100激活,其他业务及匹配的COT=8撤销不受忙状态拦截。 */
|
||
reply_cause = IEC60870_ASDU_COT_ACTIVATION_CON;
|
||
}
|
||
/* 发出任何肯定确认前检查整轮状态容量;不足时整轮拒绝,不启动只能覆盖部分站点的广播。 */
|
||
if (reply_cause == 0U && broadcast && station_count > 1U &&
|
||
station_count > self->config.broadcast_capacity) {
|
||
++self->status.overloads;
|
||
overload_counted = true;
|
||
reply_cause = IEC60870_ASDU_COT_ACTIVATION_CON;
|
||
}
|
||
/* 合法的新总召没有空槽时改为否定ACT_CON,仍可尝试提交,不影响当前任务。 */
|
||
if (reply_cause == 0U && self->count == IEC104_INTERROGATION_TASK_CAPACITY) {
|
||
++self->status.overloads;
|
||
overload_counted = true;
|
||
reply_cause = IEC60870_ASDU_COT_ACTIVATION_CON;
|
||
}
|
||
if (reply_cause != 0U) {
|
||
ImmediateReplyResult sent;
|
||
++self->status.rejected;
|
||
/* 多站应答先占一个遍历任务,不能发出部分站应答后才发现无处保存余下游标。 */
|
||
sent = broadcast ? REPLY_DEFER : Try_Immediate_Reply(self, &response, reply_cause, true, NULL);
|
||
if (sent == REPLY_SENT) { return IEC104_INTERROGATION_EVENT_HANDLED; }
|
||
if (sent == REPLY_FAILED) { return IEC104_INTERROGATION_EVENT_ERROR; }
|
||
send_waiting = sent == REPLY_WAIT;
|
||
if (self->count == IEC104_INTERROGATION_TASK_CAPACITY) {
|
||
if (!overload_counted) { ++self->status.overloads; }
|
||
++self->status.dropped_requests;
|
||
return IEC104_INTERROGATION_EVENT_HANDLED; /* 无法保存/提交的新请求已丢弃;不能重投或断链。 */
|
||
}
|
||
}
|
||
if (reply_cause == 0U && (self->next_request_id == UINT64_MAX ||
|
||
(broadcast && station_count > UINT64_MAX - self->next_request_id))) {
|
||
return Reject_Connection(self, identity, IEC104_INTERROGATION_FINISH_SERVICE_ERROR) ?
|
||
IEC104_INTERROGATION_EVENT_HANDLED : IEC104_INTERROGATION_EVENT_ERROR;
|
||
}
|
||
task = Allocate_Task(self);
|
||
if (task == NULL) {
|
||
(void)Reject_Connection(self, identity, IEC104_INTERROGATION_FINISH_SERVICE_ERROR);
|
||
return IEC104_INTERROGATION_EVENT_ERROR;
|
||
}
|
||
task->request = response;
|
||
task->reply_cause = reply_cause;
|
||
task->reply_negative = reply_cause != 0U;
|
||
task->phase = reply_cause != 0U ? IEC104_INTERROGATION_REPLY : IEC104_INTERROGATION_START;
|
||
task->send_waiting = send_waiting;
|
||
task->info.generation = event->transport.generation;
|
||
task->info.receive_serial = event->transport.data.i_frame.receive_serial;
|
||
task->info.common_address = response_ca; /* 合法总召均为本站CA;错误应答保留原错误CA且不交付BEGIN。 */
|
||
task->info.originator_address = info.originator_address;
|
||
task->info.qualifier = qualifier;
|
||
task->info.freeze = freeze;
|
||
task->info.data_cause = self->rules->data_cause(qualifier);
|
||
task->info.is_test = info.is_test != 0U;
|
||
task->info.max_objects_per_asdu = self->config.max_objects_per_asdu;
|
||
if (reply_cause == 0U) {
|
||
task->request_id = ++self->next_request_id;
|
||
}
|
||
if (broadcast && reply_cause == 0U && station_count > 1U) {
|
||
if (!Capture_Broadcast(self, task)) {
|
||
(void)Reject_Connection(self, identity, IEC104_INTERROGATION_FINISH_SERVICE_ERROR);
|
||
return IEC104_INTERROGATION_EVENT_ERROR;
|
||
}
|
||
} else if (broadcast) {
|
||
/* 单目标无须三阶段状态数组;错误广播仅逐站应答,使用共用快照保存范围和游标即可。 */
|
||
if (IEC104_Slave_StationSnapshot_Begin(self->slave, &task->stations) != IEC104_STATION_OK ||
|
||
!Bind_Station(self, task, IEC104_Slave_StationSnapshot_Next(&task->stations))) {
|
||
(void)Reject_Connection(self, identity, IEC104_INTERROGATION_FINISH_SERVICE_ERROR);
|
||
return IEC104_INTERROGATION_EVENT_ERROR;
|
||
}
|
||
} else if (reply_cause == 0U && (self->config.use_station_registry || self->config.station != NULL)) {
|
||
IEC104_Station_t *station = self->config.use_station_registry ?
|
||
IEC104_Slave_Get_Station(self->slave, response_ca) : self->config.station;
|
||
if (!Bind_Station(self, task, station)) {
|
||
(void)Reject_Connection(self, identity, IEC104_INTERROGATION_FINISH_SERVICE_ERROR);
|
||
return IEC104_INTERROGATION_EVENT_ERROR;
|
||
}
|
||
}
|
||
return IEC104_INTERROGATION_EVENT_HANDLED;
|
||
}
|
||
|
||
/* 为总召任务生成一次BEGIN或NEED_DATA,并标记等待应用应答;task须处于START或DATA。
|
||
* event_outstanding为true时复用现有交互,不重复入队或重置等待起点。
|
||
* 单目标BEGIN需要自身和未来FINISHED两个位置;多目标终结另存状态项,BEGIN仅占一个位置。
|
||
* 优先将已有终结记录入队;新工作事件不能占用单目标预留位置或挤掉尚未交付的清理通知。
|
||
* BEGIN前等FIFO空位不计业务期限,但已占用总召资格;已启用的整体期限从BEGIN生成起算。
|
||
* NEED_DATA等FIFO空位仍受已启用的整体期限约束;每次工作事件入队才启动本次应用应答期限。 */
|
||
static void Request_Application(IEC104_Interrogation_t *self, InterrogationTask *task)
|
||
{
|
||
self->status.waiting = true;
|
||
if (task->event_outstanding) { return; }
|
||
Flush_Broadcast_Finishes(self);
|
||
if (task->phase == IEC104_INTERROGATION_START) {
|
||
unsigned needed = Is_Broadcast(self, task) ? 1U : 2U;
|
||
if (self->event_count + self->reserved_finishes + needed > IEC104_INTERROGATION_EVENT_CAPACITY) { return; }
|
||
if (!Is_Broadcast(self, task)) {
|
||
task->finish_reserved = true;
|
||
++self->reserved_finishes;
|
||
}
|
||
Push_Event(self, task, IEC104_INTERROGATION_BEGIN, IEC104_INTERROGATION_FINISH_NONE);
|
||
} else {
|
||
if (self->event_count + self->reserved_finishes >= IEC104_INTERROGATION_EVENT_CAPACITY) { return; }
|
||
Push_Event(self, task, IEC104_INTERROGATION_NEED_DATA, IEC104_INTERROGATION_FINISH_NONE);
|
||
}
|
||
task->event_outstanding = true;
|
||
task->event_delivered = false;
|
||
task->data_send_waiting = false;
|
||
task->application_since_ms = IEC104_Net_Now_Ms();
|
||
if (task->phase == IEC104_INTERROGATION_START) {
|
||
task->started_ms = task->application_since_ms;
|
||
if (Is_Broadcast(self, task)) { Broadcast_Current(self)->_started_ms = task->started_ms; }
|
||
}
|
||
}
|
||
|
||
/* 应用应答的共同入口检查;expected指定应答BEGIN还是NEED_DATA,不在这里消费交互。
|
||
* 核对实例、请求、批次、类型、阶段和已取出标记,旧事件或重复应答返回STALE。
|
||
* Slave仍有通知时返回WAIT,调用方先按FIFO处理它们,防止跳过STOP/撤销/断线。
|
||
* 当前会话失效时取消;业务超时按确认边界终结,两者均返回STALE;参数/状态查询失败返回ERROR。
|
||
* OK只表示本次可以继续校验和应用应答;实际请求元数据始终取内部副本,不信任回传字段。 */
|
||
static IEC104_InterrogationReplyResult_t Check_Reply(IEC104_Interrogation_t *self,
|
||
const IEC104_InterrogationEvent_t *event, IEC104_InterrogationEventType_t expected,
|
||
InterrogationTask **matched_task)
|
||
{
|
||
InterrogationTask *task;
|
||
IEC104_SlaveRequestState_t state;
|
||
unsigned slot;
|
||
if (!Owned(self) || event == NULL) { return IEC104_INTERROGATION_REPLY_ERROR; }
|
||
slot = Find_Interrogation(self, event->request_id);
|
||
if (slot == IEC104_INTERROGATION_TASK_CAPACITY) { return IEC104_INTERROGATION_REPLY_STALE; }
|
||
task = &self->tasks[slot];
|
||
if (event->instance_id != self->instance_id || event->request_id != task->request_id ||
|
||
event->type != expected || event->batch_index != task->batch_index ||
|
||
!task->event_outstanding || !task->event_delivered ||
|
||
(expected == IEC104_INTERROGATION_BEGIN && task->phase != IEC104_INTERROGATION_START) ||
|
||
(expected == IEC104_INTERROGATION_NEED_DATA && task->phase != IEC104_INTERROGATION_DATA)) {
|
||
return IEC104_INTERROGATION_REPLY_STALE;
|
||
}
|
||
state = IEC104_Slave_Check_Request(self->slave, &self->session);
|
||
if (state == IEC104_SLAVE_REQUEST_ERROR) {
|
||
return IEC104_INTERROGATION_REPLY_ERROR;
|
||
}
|
||
self->status.events_pending = state == IEC104_SLAVE_REQUEST_WAIT_EVENTS;
|
||
if (self->status.events_pending) { return IEC104_INTERROGATION_REPLY_WAIT; }
|
||
if (!self->status.active || state != IEC104_SLAVE_REQUEST_READY) {
|
||
Cancel(self, IEC104_INTERROGATION_FINISH_CONNECTION_CHANGED);
|
||
self->status.active = false;
|
||
return IEC104_INTERROGATION_REPLY_STALE;
|
||
}
|
||
if (Expire(self) != 0) { return IEC104_INTERROGATION_REPLY_STALE; }
|
||
*matched_task = task; /* 仅OK交付内部任务地址,调用方不再假定它位于队首。 */
|
||
return IEC104_INTERROGATION_REPLY_OK;
|
||
}
|
||
|
||
/* 标记当前工作事件已经应答,解除应用等待;只在应答内容校验并接手后调用。
|
||
* 关闭event_outstanding也停止本次BEGIN/NEED_DATA期限,后续事件会设置新的计时起点。 */
|
||
static void Consume_Reply(IEC104_Interrogation_t *self, InterrogationTask *task)
|
||
{
|
||
task->event_outstanding = task->event_delivered = false;
|
||
task->data_send_waiting = false;
|
||
task->send_waiting = false;
|
||
self->status.waiting = false;
|
||
}
|
||
|
||
/* Process或Submit_Batch确认发送层已接手ASDU后推进所选任务;发送WAIT/失败时不得调用。
|
||
* 单目标CONFIRM转入DATA;多目标依次遍历确认、数据、终止三阶段。
|
||
* DATA由Submit_Batch累计进度;TERMINATE生成FINISHED(COMPLETED),REPLY结束当前站。
|
||
* 释放时只移动顺序索引,保持其他任务的槽位和调度顺序;此处不等待主站ACK。 */
|
||
static bool Advance(IEC104_Interrogation_t *self, InterrogationTask *task)
|
||
{
|
||
bool action_complete = task->phase == IEC104_INTERROGATION_CONFIRM && !self->rules->has_data(task->info.freeze);
|
||
task->send_waiting = false;
|
||
if (Is_Broadcast(self, task)) {
|
||
if (task->phase == IEC104_INTERROGATION_DATA) {
|
||
++task->batch_index;
|
||
Save_Broadcast_Progress(self);
|
||
return true;
|
||
}
|
||
if (task->phase == IEC104_INTERROGATION_REPLY && self->broadcast.cancelling) {
|
||
task->reply_cause = 0U; /* 当前站的否定确认已接手,接着兑现其他站的撤销确认。 */
|
||
return true;
|
||
}
|
||
if (task->phase == IEC104_INTERROGATION_CONFIRM && !action_complete) {
|
||
Broadcast_Current(self)->_state = STATION_CONFIRMED;
|
||
} else if (task->phase == IEC104_INTERROGATION_TERMINATE || action_complete) {
|
||
++self->status.gi_submitted;
|
||
Finish(self, task, IEC104_INTERROGATION_FINISH_COMPLETED);
|
||
}
|
||
return Next_Broadcast_Station(self);
|
||
}
|
||
if (task->phase == IEC104_INTERROGATION_CONFIRM && !action_complete) {
|
||
task->phase = IEC104_INTERROGATION_DATA;
|
||
} else if (task->phase == IEC104_INTERROGATION_DATA) {
|
||
++task->batch_index;
|
||
} else {
|
||
if (task->phase == IEC104_INTERROGATION_TERMINATE || action_complete) {
|
||
++self->status.gi_submitted;
|
||
Finish(self, task, IEC104_INTERROGATION_FINISH_COMPLETED);
|
||
}
|
||
if (task->stations._owner != NULL) {
|
||
IEC104_Station_t *next;
|
||
if (task->station != NULL) {
|
||
(void)IEC104_Slave_Release_Station(self->slave, task->station);
|
||
task->station = NULL;
|
||
}
|
||
next = IEC104_Slave_StationSnapshot_Next(&task->stations);
|
||
if (next != NULL) {
|
||
if (!Bind_Station(self, task, next) ||
|
||
(task->request_id != 0U && self->next_request_id == UINT64_MAX)) {
|
||
return false;
|
||
}
|
||
if (task->request_id != 0U) {
|
||
task->request_id = ++self->next_request_id;
|
||
task->phase = IEC104_INTERROGATION_START;
|
||
task->reply_cause = 0U;
|
||
task->reply_negative = false;
|
||
}
|
||
task->started_ms = task->application_since_ms = 0U;
|
||
task->batch_index = task->objects_submitted = 0U;
|
||
task->data_send_waiting = false;
|
||
return true;
|
||
}
|
||
}
|
||
Remove_Task(self, task);
|
||
}
|
||
return true;
|
||
}
|
||
|
||
/**
|
||
* 功能:初始化调用方提供的配置,设置站地址、每批对象数以及应用应答和执行期限。
|
||
* 参数:config为可写配置对象;NULL时不执行任何操作。
|
||
* 说明:先清零再设置默认值,不创建实例或占用资源;应用可在Create前修改配置。
|
||
*/
|
||
void IEC104_Interrogation_Config_Default(IEC104_InterrogationConfig_t *config)
|
||
{
|
||
if (config == NULL) { return; }
|
||
memset(config, 0, sizeof(*config));
|
||
config->common_address = 1U;
|
||
config->max_objects_per_asdu = 127U;
|
||
config->begin_timeout_ms = 15000U;
|
||
config->data_timeout_ms = 15000U;
|
||
config->total_timeout_ms = 0U;
|
||
}
|
||
|
||
/**
|
||
* 功能:从静态池创建总召实例,复制配置并借用一个有效Slave。
|
||
* 参数:slave须在本实例整个寿命内保持有效;config可在成功返回后复用。
|
||
* 返回:新实例句柄;配置非法、Slave无效、重复绑定、池满或编号耗尽时返回NULL。
|
||
* 说明:扫描完整静态池拒绝重复绑定,失败不占槽;实例编号在整个进程内不重复。
|
||
* 创建时尚无活动会话,后续依靠Handle_Event接收CONNECTED/ACTIVATED建立状态。
|
||
*/
|
||
IEC104_Interrogation_t *IEC104_Interrogation_Create_With_Rules(IEC104_Slave_t *slave,
|
||
const IEC104_InterrogationConfig_t *config, const IEC104_InterrogationRules_t *rules)
|
||
{
|
||
size_t i;
|
||
uintptr_t storage_begin, storage_end;
|
||
IEC104_Interrogation_t *free_slot = NULL;
|
||
if (rules == NULL || rules->decode == NULL || rules->qualifier_valid == NULL ||
|
||
rules->data_cause == NULL || rules->data_allowed == NULL || rules->has_data == NULL ||
|
||
config == NULL || (config->use_station_registry && config->station != NULL) ||
|
||
(!config->use_station_registry && config->station == NULL &&
|
||
(config->common_address == 0U || config->common_address == IEC104_CA_BROADCAST)) ||
|
||
config->max_objects_per_asdu == 0U || config->max_objects_per_asdu > 127U ||
|
||
IEC104_Slave_Get_Application_Layer_Parameters(slave) == NULL || instance_serial == UINT64_MAX) { return NULL; }
|
||
if ((config->broadcast_states == NULL) != (config->broadcast_capacity == 0U) ||
|
||
config->broadcast_capacity > SIZE_MAX / sizeof(*config->broadcast_states)) { return NULL; }
|
||
storage_begin = (uintptr_t)config->broadcast_states;
|
||
if (storage_begin > UINTPTR_MAX - config->broadcast_capacity * sizeof(*config->broadcast_states)) { return NULL; }
|
||
storage_end = storage_begin + config->broadcast_capacity * sizeof(*config->broadcast_states);
|
||
for (i = 0U; i < IEC104_INTERROGATION_CAPACITY; ++i) {
|
||
if (interrogation_pool[i].slave == slave && interrogation_pool[i].rules == rules) { return NULL; }
|
||
if (config->broadcast_states != NULL && interrogation_pool[i].slave != NULL) {
|
||
const IEC104_InterrogationConfig_t *other = &interrogation_pool[i].config;
|
||
uintptr_t other_begin = (uintptr_t)other->broadcast_states;
|
||
uintptr_t other_end = other_begin + other->broadcast_capacity * sizeof(*other->broadcast_states);
|
||
if (storage_begin < other_end && other_begin < storage_end) { return NULL; }
|
||
}
|
||
if (free_slot == NULL && interrogation_pool[i].slave == NULL) { free_slot = &interrogation_pool[i]; }
|
||
}
|
||
if (free_slot != NULL) {
|
||
if (config->station != NULL && !IEC104_Station_Is_Registered(config->station)) { return NULL; }
|
||
if (config->station != NULL &&
|
||
IEC104_Slave_Acquire_Station(slave, config->station) != IEC104_STATION_OK) { return NULL; }
|
||
memset(free_slot, 0, sizeof(*free_slot));
|
||
free_slot->slave = slave;
|
||
free_slot->rules = rules;
|
||
free_slot->config = *config;
|
||
if (config->station != NULL) {
|
||
free_slot->config.common_address = IEC104_Station_Get_Common_Address(config->station);
|
||
}
|
||
free_slot->instance_id = ++instance_serial;
|
||
}
|
||
return free_slot;
|
||
}
|
||
|
||
/**
|
||
* 功能:释放已经没有协议任务和业务事件的实例槽,不销毁借用的Slave。
|
||
* 参数:self为Create返回的有效实例。
|
||
* 返回:true已释放,句柄失效;false为非法句柄或仍有任务/事件,原状态不变。
|
||
* 说明:不会隐式Cancel或丢弃通知;调用方须先取消任务、读完FINISHED并清理应用资源。
|
||
*/
|
||
bool IEC104_Interrogation_Destroy(IEC104_Interrogation_t *self)
|
||
{
|
||
if (!Owned(self) || self->count != 0U || self->event_count != 0U) { return false; }
|
||
if (self->config.station != NULL &&
|
||
IEC104_Slave_Release_Station(self->slave, self->config.station) != IEC104_STATION_OK) { return false; }
|
||
memset(self, 0, sizeof(*self));
|
||
return true;
|
||
}
|
||
|
||
/**
|
||
* 功能:以CANCELLED原因放弃本实例尚未提交的工作,保留终结事件供应用清理。
|
||
* 参数:self为总召实例;非法句柄无操作。
|
||
* 说明:不关闭连接、不发送确认或终止,也不撤回发送层已经接手的ASDU。
|
||
* 该入口是应用本地取消;主站撤销请求由Handle_Event执行匹配和确认流程。
|
||
*/
|
||
void IEC104_Interrogation_Cancel(IEC104_Interrogation_t *self)
|
||
{
|
||
if (Owned(self)) { Cancel(self, IEC104_INTERROGATION_FINISH_CANCELLED); }
|
||
}
|
||
|
||
/**
|
||
* 功能:处理应用转交的一个Slave历史通知,更新会话状态或保存总召请求。
|
||
* 参数:self为有效实例;event须为同一Slave按FIFO取出的原事件或Copy_Event副本。
|
||
* 返回:HANDLED已处理请求(含拒绝/满槽丢弃)或会话变化;IGNORED可交给其他业务;ERROR为参数/处理失败。
|
||
* 说明:校验事件所属服务及代次,不自行读取Slave FIFO,不保存调用方事件指针。
|
||
* 不调用应用代码;短应答在Slave通知已读空、会话有效且没有旧REPLY时尝试立即提交。
|
||
* 布局错误或满槽只拒绝/丢弃新请求;取消时保留FINISHED,内部服务故障沿用失败策略。
|
||
*/
|
||
IEC104_InterrogationEventResult_t IEC104_Interrogation_Handle_Event(
|
||
IEC104_Interrogation_t *self, const IEC104_SlaveEvent_t *event)
|
||
{
|
||
IEC104_SlaveEventIdentity_t identity;
|
||
/* Slave负责来源和视图校验;本模块额外保持首次绑定的服务实例,不接管复用后的另一服务。 */
|
||
if (!Owned(self) || !IEC104_Slave_Get_Event_Identity(self->slave, event, &identity) ||
|
||
(self->session.instance_id != 0U && self->session.instance_id != identity.instance_id)) {
|
||
return IEC104_INTERROGATION_EVENT_ERROR;
|
||
}
|
||
self->session.slave = identity.slave;
|
||
self->session.instance_id = identity.instance_id;
|
||
if (event->transport.type == IEC104_TCP_EVENT_CONNECTED) {
|
||
/* TCP连通仅建立新代次;必须再收到ACTIVATED,才能接受本代次总召。 */
|
||
Cancel(self, IEC104_INTERROGATION_FINISH_CONNECTION_CHANGED);
|
||
self->session.generation = identity.generation;
|
||
self->status.active = false;
|
||
return IEC104_INTERROGATION_EVENT_HANDLED;
|
||
}
|
||
if (identity.generation != self->session.generation) {
|
||
/* 旧连接遗留的断开、回执或I帧不能改变新连接的业务状态。 */
|
||
return IEC104_INTERROGATION_EVENT_IGNORED;
|
||
}
|
||
switch (event->transport.type) {
|
||
case IEC104_TCP_EVENT_ACTIVATED:
|
||
self->status.active = true;
|
||
return IEC104_INTERROGATION_EVENT_HANDLED;
|
||
case IEC104_TCP_EVENT_DEACTIVATED:
|
||
case IEC104_TCP_EVENT_DISCONNECTED:
|
||
Cancel(self, event->transport.type == IEC104_TCP_EVENT_DEACTIVATED ?
|
||
IEC104_INTERROGATION_FINISH_STOPPED : IEC104_INTERROGATION_FINISH_DISCONNECTED);
|
||
self->status.active = false;
|
||
if (event->transport.type == IEC104_TCP_EVENT_DISCONNECTED) { self->session.generation = 0U; }
|
||
return IEC104_INTERROGATION_EVENT_HANDLED;
|
||
case IEC104_TCP_EVENT_I_FRAME: return Request(self, event, &identity);
|
||
default: return IEC104_INTERROGATION_EVENT_IGNORED; /* ACK回执不触发业务重发。 */
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 功能:自动推进当前可处理的单帧应答和总召任务,直到空闲或需要等待。
|
||
* 参数:self为有效实例;调用方不需要提供处理步数。
|
||
* 返回:true为本轮正常结束,包括等待、取消或成功处理超时;false为参数、状态查询、组包或关闭失败。
|
||
* 说明:Slave通知未读空时仅设置events_pending,应用先转交网络通知再继续调用。
|
||
* START有事件空位时生成BEGIN并开始执行计时;DATA生成NEED_DATA,应用直接提交批次。
|
||
* 双方可推进时轮转;总召等应用或业务FIFO空位时仍处理单帧应答。
|
||
* 每步重查Slave通知,发送过程中新交付的通知也须先处理;只有接手ASDU后才Advance。
|
||
* ERROR若确认为原连接正常STOP拒绝,则先让出交付通知;其他ERROR/FATAL仍为SEND_ERROR。
|
||
* 发送WAIT保留任务并返回,下轮优先检查单帧应答。
|
||
* 本函数不接收新请求、不执行应用代码;广播逐站复用任务,每轮至多2倍任务容量步。
|
||
* 网络推进并读空Slave通知后须调用本函数重试;本地变化后按need_process及业务期限安排唤醒。
|
||
*/
|
||
bool IEC104_Interrogation_Process(IEC104_Interrogation_t *self)
|
||
{
|
||
IEC104_SlaveRequestState_t state;
|
||
unsigned steps;
|
||
if (!Owned(self)) { return false; }
|
||
for (steps = 0U; steps < 2U * IEC104_INTERROGATION_TASK_CAPACITY; ++steps) {
|
||
InterrogationTask *task;
|
||
IEC60870_Asdu_t asdu;
|
||
InterrogationResponse cancellation;
|
||
InterrogationResponse *response;
|
||
IEC104_TcpSubmitResult_t submitted;
|
||
uint8_t cause;
|
||
bool negative = false;
|
||
unsigned slot;
|
||
size_t cancel_index = 0U;
|
||
bool cancelling;
|
||
int expired;
|
||
Flush_Broadcast_Finishes(self);
|
||
Reap_Broadcast(self);
|
||
state = IEC104_Slave_Check_Request(self->slave, &self->session);
|
||
if (state == IEC104_SLAVE_REQUEST_ERROR) {
|
||
Cancel(self, IEC104_INTERROGATION_FINISH_SERVICE_ERROR);
|
||
self->status.active = false;
|
||
return false;
|
||
}
|
||
self->status.events_pending = state == IEC104_SLAVE_REQUEST_WAIT_EVENTS;
|
||
if (self->status.events_pending) { return true; }
|
||
self->status.waiting = false;
|
||
if (self->count == 0U) { return true; }
|
||
if (!self->status.active || state != IEC104_SLAVE_REQUEST_READY) {
|
||
Cancel(self, IEC104_INTERROGATION_FINISH_CONNECTION_CHANGED);
|
||
self->status.active = false;
|
||
return true;
|
||
}
|
||
expired = Expire(self);
|
||
if (expired != 0) {
|
||
if (expired < 0 || !self->status.active) { return expired > 0; }
|
||
continue; /* 确认前超时仅转为REPLY;重查通知后,本轮即可尝试提交否定确认。 */
|
||
}
|
||
slot = Select_Task(self);
|
||
if (slot == IEC104_INTERROGATION_TASK_CAPACITY) { self->status.waiting = true; return true; }
|
||
task = &self->tasks[slot];
|
||
/* 当前站的否定ACT_CON可能与另一站撤销并存,先提交原否定确认,再继续待发COT=9。 */
|
||
cancelling = Is_Broadcast(self, task) && self->broadcast.cancelling &&
|
||
!(task->phase == IEC104_INTERROGATION_REPLY && task->reply_cause != 0U);
|
||
response = &task->request;
|
||
if (!cancelling && (task->phase == IEC104_INTERROGATION_START || task->phase == IEC104_INTERROGATION_DATA)) {
|
||
Request_Application(self, task);
|
||
self->prefer_interrogation = false;
|
||
continue; /* 应用稍后取事件;继续处理当前已有的单帧应答。 */
|
||
}
|
||
/* 普通应答沿用当前站CA;撤销可能针对其他站,须用临时副本改CA,不能改坏当前站请求。
|
||
* WAIT后原任务和撤销标志仍保留,下次按同一目标重建应答。 */
|
||
if (cancelling) {
|
||
cancel_index = Pending_Broadcast_Deactivation(self);
|
||
if (cancel_index == self->broadcast.count ||
|
||
!Copy_Response_Request(self, &cancellation, &task->request.identity, task->request.bytes,
|
||
sizeof(task->request.bytes),
|
||
IEC104_Station_Get_Common_Address(self->config.broadcast_states[cancel_index]._station))) {
|
||
Cancel(self, IEC104_INTERROGATION_FINISH_SERVICE_ERROR);
|
||
return false;
|
||
}
|
||
response = &cancellation;
|
||
cause = IEC60870_ASDU_COT_DEACTIVATION_CON;
|
||
} else if (task->phase == IEC104_INTERROGATION_CONFIRM) {
|
||
cause = IEC60870_ASDU_COT_ACTIVATION_CON;
|
||
} else if (task->phase == IEC104_INTERROGATION_TERMINATE) {
|
||
cause = IEC60870_ASDU_COT_ACTIVATION_TERMINATION;
|
||
} else if (task->phase == IEC104_INTERROGATION_REPLY) {
|
||
cause = task->reply_cause;
|
||
negative = task->reply_negative;
|
||
} else {
|
||
Cancel(self, IEC104_INTERROGATION_FINISH_SERVICE_ERROR);
|
||
return false;
|
||
}
|
||
if (!Prepare_Response(self, response, cause, negative, &asdu)) {
|
||
(void)Reject_Connection(self, &task->request.identity, IEC104_INTERROGATION_FINISH_SERVICE_ERROR);
|
||
return false;
|
||
}
|
||
submitted = IEC104_Slave_Send_Response(self->slave, &task->request.identity, &asdu);
|
||
if (submitted != IEC104_TCP_SUBMIT_OK) {
|
||
if (Defer_For_Stop(self, &task->request.identity, submitted)) { return true; }
|
||
if (submitted == IEC104_TCP_SUBMIT_WAIT) {
|
||
/* 确认/终止/短应答保留原文重试;总召数据仅由Submit_Batch直接提交。 */
|
||
++self->status.waits;
|
||
self->status.waiting = true;
|
||
task->send_waiting = true;
|
||
self->prefer_interrogation = false;
|
||
} else {
|
||
Cancel(self, IEC104_INTERROGATION_FINISH_SEND_ERROR);
|
||
self->status.active = false;
|
||
}
|
||
return true;
|
||
}
|
||
self->prefer_interrogation = cancelling || task->phase == IEC104_INTERROGATION_REPLY;
|
||
++self->status.submitted;
|
||
if (negative) { ++self->status.negative_submitted; }
|
||
if (!(cancelling ? Advance_Broadcast_Deactivation(self, cancel_index) : Advance(self, task))) {
|
||
(void)Reject_Connection(self, &task->request.identity, IEC104_INTERROGATION_FINISH_SERVICE_ERROR);
|
||
return false;
|
||
}
|
||
}
|
||
return true;
|
||
}
|
||
|
||
/**
|
||
* 功能:将最早一条业务事件复制到调用方并从FIFO移除,不执行网络操作。
|
||
* 参数:self为有效实例;event为非NULL输出对象,接收可独立保存的纯值副本。
|
||
* 返回:READY写出事件;EMPTY表示暂时无事件;ERROR表示参数非法,后两者不改输出。
|
||
* 说明:工作事件出队后才允许应答;BEGIN出队同时标记应用已持有本请求,后续须配对FINISHED。
|
||
* 业务事件不含自引用指针,因此可以memmove压紧FIFO;协议任务仍通过order维护槽位顺序。
|
||
* 每次先补入延迟FINISHED,因此仅反复NextEvent也能读完全部终结通知,无须靠Process补齐。
|
||
* 读出最后一条时可能释放广播持有的站点引用;应用若仍需节点/context,须事先独立Acquire。
|
||
*/
|
||
IEC104_InterrogationNextResult_t IEC104_Interrogation_NextEvent(
|
||
IEC104_Interrogation_t *self, IEC104_InterrogationEvent_t *event)
|
||
{
|
||
InterrogationTask *task = NULL;
|
||
if (!Owned(self) || event == NULL) { return IEC104_INTERROGATION_NEXT_ERROR; }
|
||
Flush_Broadcast_Finishes(self);
|
||
Reap_Broadcast(self);
|
||
if (self->event_count == 0U) { return IEC104_INTERROGATION_NEXT_EMPTY; }
|
||
if (self->events[0].type != IEC104_INTERROGATION_FINISHED) {
|
||
unsigned slot = Find_Interrogation(self, self->events[0].request_id);
|
||
if (slot == IEC104_INTERROGATION_TASK_CAPACITY) { return IEC104_INTERROGATION_NEXT_ERROR; }
|
||
task = &self->tasks[slot];
|
||
}
|
||
*event = self->events[0];
|
||
--self->event_count;
|
||
memmove(self->events, self->events + 1, self->event_count * sizeof(self->events[0]));
|
||
if (task != NULL) {
|
||
task->event_delivered = true;
|
||
if (event->type == IEC104_INTERROGATION_BEGIN) { task->begin_delivered = true; }
|
||
if (event->type == IEC104_INTERROGATION_BEGIN && Is_Broadcast(self, task)) {
|
||
Broadcast_Current(self)->_flags |= STATION_BEGIN_READ;
|
||
}
|
||
} else if (self->broadcast.task != NULL && event->type == IEC104_INTERROGATION_FINISHED) {
|
||
size_t i;
|
||
for (i = 0U; i < self->broadcast.count; ++i) {
|
||
IEC104_InterrogationStationState_t *entry = &self->config.broadcast_states[i];
|
||
if (entry->_request_id == event->request_id) { entry->_flags |= STATION_FINISH_READ; break; }
|
||
}
|
||
Reap_Broadcast(self);
|
||
}
|
||
return IEC104_INTERROGATION_NEXT_READY;
|
||
}
|
||
|
||
/**
|
||
* 功能:接收应用对已取出BEGIN事件的接受或拒绝决定。
|
||
* 参数:event为本实例当前BEGIN的原样副本;accept为true接受、false拒绝。
|
||
* 返回:沿用Check_Reply的WAIT/STALE/ERROR;OK表示决定已消费,不可用同一事件重复应答。
|
||
* 说明:接受后转为CONFIRM,等待Process提交肯定ACT_CON;拒绝先生成FINISHED(REJECTED),
|
||
* 再将任务转为否定ACT_CON应答。资源由应用读取FINISHED后释放,本函数不执行清理回调。
|
||
*/
|
||
IEC104_InterrogationReplyResult_t IEC104_Interrogation_Reply_Begin(
|
||
IEC104_Interrogation_t *self, const IEC104_InterrogationEvent_t *event, bool accept)
|
||
{
|
||
InterrogationTask *task;
|
||
IEC104_InterrogationReplyResult_t result = Check_Reply(self, event, IEC104_INTERROGATION_BEGIN, &task);
|
||
if (result != IEC104_INTERROGATION_REPLY_OK) { return result; }
|
||
Consume_Reply(self, task);
|
||
if (task->info.is_test && !self->rules->has_data(task->info.freeze)) { accept = false; }
|
||
if (accept) {
|
||
task->phase = IEC104_INTERROGATION_CONFIRM;
|
||
if (Is_Broadcast(self, task)) { Broadcast_Current(self)->_state = STATION_ACCEPTED; }
|
||
}
|
||
else { Reject_Request(self, task, IEC104_INTERROGATION_FINISH_REJECTED); }
|
||
return IEC104_INTERROGATION_REPLY_OK;
|
||
}
|
||
|
||
/* 校验种类、事件、会话及期限后,先消费BEGIN再执行同步动作,重试同一事件不会再次执行。
|
||
* action只接收引擎保存的请求元数据,不信任调用方可改写的event.request;禁止回调重入。
|
||
* TEST直接拒绝;action失败须保证业务数据未变,成功后保持肯定确认,不在回调之后重新判超时。 */
|
||
IEC104_InterrogationReplyResult_t IEC104_Interrogation_Execute_Action(
|
||
IEC104_Interrogation_t *self, const IEC104_InterrogationEvent_t *event,
|
||
const IEC104_InterrogationRules_t *rules,
|
||
bool (*action)(void *context, const IEC104_InterrogationRequest_t *request), void *context)
|
||
{
|
||
InterrogationTask *task;
|
||
IEC104_InterrogationReplyResult_t result;
|
||
bool accepted;
|
||
if (!Owned(self) || self->rules != rules || action == NULL) { return IEC104_INTERROGATION_REPLY_ERROR; }
|
||
result = Check_Reply(self, event, IEC104_INTERROGATION_BEGIN, &task);
|
||
if (result != IEC104_INTERROGATION_REPLY_OK) { return result; }
|
||
if (self->rules->has_data(task->info.freeze)) { return IEC104_INTERROGATION_REPLY_ERROR; }
|
||
Consume_Reply(self, task);
|
||
accepted = !task->info.is_test && action(context, &task->info);
|
||
if (accepted) {
|
||
task->phase = IEC104_INTERROGATION_CONFIRM;
|
||
if (Is_Broadcast(self, task)) { Broadcast_Current(self)->_state = STATION_ACCEPTED; }
|
||
} else { Reject_Request(self, task, IEC104_INTERROGATION_FINISH_REJECTED); }
|
||
return IEC104_INTERROGATION_REPLY_OK;
|
||
}
|
||
|
||
/**
|
||
* 功能:验证一次NEED_DATA对应的非空批次,当场尝试不排队提交。
|
||
* 参数:event为当前已取出的NEED_DATA;asdu为应用构造的完整监视信息ASDU。
|
||
* 返回:OK发送层已接手;WAIT通知未读空或发送背压;STALE交互失效/发送失败已取消;ERROR批次非法。
|
||
* 说明:按当前目标站验证CA,按原请求验证COT、OA、TEST,按配置检查对象数及计数溢出。
|
||
* 发送OK才消费事件并累计进度。发送WAIT不缓存、不推进游标,应用下次重新组包;
|
||
* 总召可重取当前值,计数量仍须使用本轮冻结快照。
|
||
* 首次有效批次发生发送WAIT后结束本批取数计时;同一交互的重试由应用在网络推进后驱动。
|
||
* 非法批次不结束取数计时。成功后原文由发送层持有,不借用应用缓冲;空数据调用Complete_Data。
|
||
*/
|
||
IEC104_InterrogationReplyResult_t IEC104_Interrogation_Submit_Batch(
|
||
IEC104_Interrogation_t *self, const IEC104_InterrogationEvent_t *event,
|
||
const IEC60870_Asdu_t *asdu)
|
||
{
|
||
InterrogationTask *task;
|
||
IEC60870_AsduInfo_t info;
|
||
const uint8_t *data;
|
||
uint16_t length;
|
||
IEC104_TcpSubmitResult_t submitted;
|
||
const IEC60870_AppLayerParameters_t *parameters;
|
||
IEC104_InterrogationReplyResult_t result = Check_Reply(self, event, IEC104_INTERROGATION_NEED_DATA, &task);
|
||
if (result != IEC104_INTERROGATION_REPLY_OK) { return result; }
|
||
parameters = IEC104_Slave_Get_Application_Layer_Parameters(self->slave);
|
||
if (asdu == NULL || asdu->parameters == NULL ||
|
||
asdu->parameters->sizeOfCOT != 2 || asdu->parameters->sizeOfCA != 2 ||
|
||
asdu->parameters->sizeOfIOA != 3 ||
|
||
IEC60870_Asdu_Get_Data(asdu, &data, &length) != IEC60870_ASDU_OK ||
|
||
length > IEC104_MAX_ASDU_LEN ||
|
||
IEC60870_Asdu_Parse(parameters, data, length, &info) != IEC60870_ASDU_OK ||
|
||
!self->rules->data_allowed(&info, task->info.freeze) || info.object_count == 0U ||
|
||
info.object_count > self->config.max_objects_per_asdu ||
|
||
info.cause_of_transmission != task->info.data_cause ||
|
||
info.common_address != task->info.common_address ||
|
||
info.originator_address != task->info.originator_address ||
|
||
info.is_test != (uint8_t)task->info.is_test || info.is_negative_confirm != 0U ||
|
||
task->batch_index == SIZE_MAX || task->objects_submitted > SIZE_MAX - info.object_count) {
|
||
return IEC104_INTERROGATION_REPLY_ERROR;
|
||
}
|
||
/* 直接提交须等待既有发送工作先行;WAIT不缓存本批旧值,应用重读实时值后再试。
|
||
* 点值更新/突发入队与本次读值/提交须由应用串行协调,才能保持同一点的新旧发送顺序。 */
|
||
submitted = IEC104_Slave_Try_Send_Response(self->slave, &task->request.identity, asdu);
|
||
if (submitted == IEC104_TCP_SUBMIT_WAIT) {
|
||
task->data_send_waiting = true;
|
||
task->send_waiting = true;
|
||
self->status.waiting = true;
|
||
++self->status.waits;
|
||
return IEC104_INTERROGATION_REPLY_WAIT;
|
||
}
|
||
if (submitted != IEC104_TCP_SUBMIT_OK) {
|
||
if (Defer_For_Stop(self, &task->request.identity, submitted)) {
|
||
return IEC104_INTERROGATION_REPLY_WAIT; /* 仅等待STOP通知,不算发送背压,不停止取数计时。 */
|
||
}
|
||
Cancel(self, IEC104_INTERROGATION_FINISH_SEND_ERROR);
|
||
self->status.active = false;
|
||
return IEC104_INTERROGATION_REPLY_STALE;
|
||
}
|
||
task->objects_submitted += info.object_count;
|
||
++self->status.submitted;
|
||
Consume_Reply(self, task);
|
||
(void)Advance(self, task); /* DATA仅递增批次,不切换站点或释放资源。 */
|
||
self->prefer_interrogation = false;
|
||
return IEC104_INTERROGATION_REPLY_OK;
|
||
}
|
||
|
||
/**
|
||
* 功能:应答当前NEED_DATA,声明当前站已没有更多总召数据。
|
||
* 参数:event须为当前已取出的NEED_DATA,不能使用BEGIN或已经交付过批次的旧事件。
|
||
* 返回:OK已消费交互;单目标转TERMINATE,多目标继续下一站数据或进入终止遍历。
|
||
* 说明:此处不发送ACT_TERM;后续Process提交成功才产生FINISHED(COMPLETED),发送等待时保留任务。
|
||
* 应用可在OK后释放本站取数游标供下一站使用;迟到的FINISHED须按请求身份清理,避免误清新站。
|
||
*/
|
||
IEC104_InterrogationReplyResult_t IEC104_Interrogation_Complete_Data(
|
||
IEC104_Interrogation_t *self, const IEC104_InterrogationEvent_t *event)
|
||
{
|
||
InterrogationTask *task;
|
||
IEC104_InterrogationReplyResult_t result = Check_Reply(self, event, IEC104_INTERROGATION_NEED_DATA, &task);
|
||
if (result != IEC104_INTERROGATION_REPLY_OK) { return result; }
|
||
Consume_Reply(self, task);
|
||
if (Is_Broadcast(self, task)) {
|
||
Save_Broadcast_Progress(self);
|
||
Broadcast_Current(self)->_state = STATION_DATA_DONE;
|
||
if (!Next_Broadcast_Station(self)) {
|
||
(void)Reject_Connection(self, &task->request.identity, IEC104_INTERROGATION_FINISH_SERVICE_ERROR);
|
||
return IEC104_INTERROGATION_REPLY_ERROR;
|
||
}
|
||
return IEC104_INTERROGATION_REPLY_OK;
|
||
}
|
||
task->phase = IEC104_INTERROGATION_TERMINATE;
|
||
return IEC104_INTERROGATION_REPLY_OK;
|
||
}
|
||
|
||
/**
|
||
* 功能:应用在准备数据源或取得数据失败时,按肯定确认的提交边界终结本次总召。
|
||
* 参数:event为当前已取出且尚未应答的BEGIN或NEED_DATA。
|
||
* 返回:WAIT要求先处理Slave通知;STALE表示交互已失效;ERROR为检查或关闭失败。
|
||
* 通过检查后,BEGIN失败转为否定REPLY并保留连接;NEED_DATA失败关闭原连接。
|
||
* OK表示失败已处理,否定确认由Process提交;若关闭失败返回ERROR,任务仍已取消,不能重试。
|
||
* 说明:两种路径均保留SOURCE_ERROR并交付FINISHED供应用清理,不发送成功ACT_TERM。
|
||
*/
|
||
IEC104_InterrogationReplyResult_t IEC104_Interrogation_Fail_Request(
|
||
IEC104_Interrogation_t *self, const IEC104_InterrogationEvent_t *event)
|
||
{
|
||
IEC104_InterrogationReplyResult_t result;
|
||
InterrogationTask *task;
|
||
if (event == NULL) { return IEC104_INTERROGATION_REPLY_ERROR; }
|
||
if (event->type != IEC104_INTERROGATION_BEGIN && event->type != IEC104_INTERROGATION_NEED_DATA) {
|
||
return Owned(self) ? IEC104_INTERROGATION_REPLY_STALE : IEC104_INTERROGATION_REPLY_ERROR;
|
||
}
|
||
result = Check_Reply(self, event, event->type, &task);
|
||
if (result != IEC104_INTERROGATION_REPLY_OK) { return result; }
|
||
return Fail_Interrogation(self, task, IEC104_INTERROGATION_FINISH_SOURCE_ERROR) ?
|
||
IEC104_INTERROGATION_REPLY_OK : IEC104_INTERROGATION_REPLY_ERROR;
|
||
}
|
||
|
||
/**
|
||
* 功能:组合会话状态、累计统计、活动总召进度及最近业务期限,生成独立状态副本。
|
||
* 参数:self为有效实例;status为非NULL输出对象。
|
||
* 返回:true已写出快照;false为非法句柄/NULL,输出保持不变。
|
||
* 说明:phase和进度优先描述当前站;收尾为FINISHING,仅有应答时REPLY,无任务时IDLE。
|
||
* 尚未生成BEGIN不计业务期限;timer_active为false时剩余值无意义。
|
||
* 实时查询Slave通知和本地可推进任务;need_process不等于!waiting,不保证发送接手。
|
||
* 只读取当前单调时钟及服务状态,不触发超时、发送或事件入队。
|
||
*/
|
||
bool IEC104_Interrogation_Get_Status(const IEC104_Interrogation_t *self,
|
||
IEC104_InterrogationStatus_t *status)
|
||
{
|
||
const InterrogationTask *task;
|
||
IEC104_SlaveRequestState_t state;
|
||
unsigned slot;
|
||
InterrogationTimeout timeout;
|
||
if (!Owned(self) || status == NULL) { return false; }
|
||
slot = Find_Interrogation(self, 0U);
|
||
task = slot != IEC104_INTERROGATION_TASK_CAPACITY ? &self->tasks[slot] :
|
||
(self->count != 0U ? &self->tasks[self->order[0]] : NULL);
|
||
*status = self->status;
|
||
status->gi_pending = slot != IEC104_INTERROGATION_TASK_CAPACITY;
|
||
status->generation = self->session.generation;
|
||
status->pending_requests = self->count;
|
||
status->pending_events = self->event_count + Pending_Broadcast_Finishes(self);
|
||
status->phase = task != NULL ? task->phase : IEC104_INTERROGATION_IDLE;
|
||
status->batch_index = task != NULL ? task->batch_index : 0U;
|
||
status->objects_submitted = task != NULL ? task->objects_submitted : 0U;
|
||
status->data_send_waiting = task != NULL && task->data_send_waiting;
|
||
status->application_waiting = task != NULL && task->event_outstanding && !task->data_send_waiting;
|
||
if (task != NULL && task->phase != IEC104_INTERROGATION_REPLY &&
|
||
!Interrogation_Can_Progress(self, task)) { status->waiting = true; }
|
||
status->timer_active = false;
|
||
status->timeout_remaining_ms = 0U;
|
||
if (task != NULL) {
|
||
timeout = Task_Timeout(self, task, IEC104_Net_Now_Ms());
|
||
if (timeout.reason != IEC104_INTERROGATION_FINISH_NONE) {
|
||
status->timer_active = true;
|
||
status->timeout_remaining_ms = timeout.remaining_ms;
|
||
}
|
||
}
|
||
/* 只查询Slave通知,不消费或发送;不能跨过未读STOP/撤销推进旧业务。
|
||
* 选择规则与Process一致,避免把等应用的总召误当成阻塞所有短应答的队首。
|
||
* WAIT只抑制本地空转;网络推进后调用方仍须显式Process重试,不以本标志跳过。 */
|
||
state = IEC104_Slave_Check_Request(self->slave, &self->session);
|
||
status->events_pending = state == IEC104_SLAVE_REQUEST_WAIT_EVENTS;
|
||
slot = Select_Task(self);
|
||
status->need_process = self->count != 0U && !status->events_pending &&
|
||
(state != IEC104_SLAVE_REQUEST_READY || !status->active ||
|
||
(status->timer_active && status->timeout_remaining_ms == 0U) ||
|
||
(slot != IEC104_INTERROGATION_TASK_CAPACITY && !self->tasks[slot].send_waiting));
|
||
/* 仅剩终结事件交付时由NextEvent推进;pending_events负责唤醒,重复Process不能代替应用读取。 */
|
||
if (self->broadcast.task != NULL && self->broadcast.stage == BROADCAST_FINISH &&
|
||
!self->broadcast.cancelling && self->count == 1U) { status->need_process = false; }
|
||
return true;
|
||
}
|