slhaf/Partner
1
0
Fork 0
mirror of https://github.com/slhaf/Partner.git synced 2026-05-12 08:43:02 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(action): stabilize scheduler wheel lifecycle

Browse Source
This commit is contained in:
slhafzjw
2026-04-27 16:25:42 +08:00
parent 04ceccc46f
commit 184c396752
1 changed files with 13 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
Partner-Core/src/main/java/work/slhaf/partner/module/action/scheduler/ActionScheduler.kt
View File

@@ -15,12 +15,14 @@ import work.slhaf.partner.core.action.ActionCapability
import work.slhaf.partner.core.action.entity.Action import work.slhaf.partner.core.action.entity.Action
import work.slhaf.partner.core.action.entity.Schedulable import work.slhaf.partner.core.action.entity.Schedulable
import work.slhaf.partner.core.action.entity.SchedulableExecutableAction import work.slhaf.partner.core.action.entity.SchedulableExecutableAction
import work.slhaf.partner.framework.agent.config.ConfigCenter
import work.slhaf.partner.framework.agent.factory.capability.annotation.InjectCapability import work.slhaf.partner.framework.agent.factory.capability.annotation.InjectCapability
import work.slhaf.partner.framework.agent.factory.component.abstracts.AbstractAgentModule import work.slhaf.partner.framework.agent.factory.component.abstracts.AbstractAgentModule
import work.slhaf.partner.framework.agent.factory.component.annotation.Init import work.slhaf.partner.framework.agent.factory.component.annotation.Init
import work.slhaf.partner.framework.agent.factory.component.annotation.InjectModule import work.slhaf.partner.framework.agent.factory.component.annotation.InjectModule
import work.slhaf.partner.module.action.executor.ActionExecutor import work.slhaf.partner.module.action.executor.ActionExecutor
import java.io.Closeable import java.io.Closeable
import java.nio.file.Path
import java.time.Duration import java.time.Duration
import java.time.ZonedDateTime import java.time.ZonedDateTime
import java.time.temporal.ChronoUnit import java.time.temporal.ChronoUnit
@@ -70,8 +72,9 @@ class ActionScheduler : AbstractAgentModule.Standalone() {
val doneCondition: (Schedulable) -> Boolean = { schedulable -> val doneCondition: (Schedulable) -> Boolean = { schedulable ->
if (schedulable is Action) { if (schedulable is Action) {
schedulable.status == Action.Status.FAILED || schedulable.status == Action.Status.SUCCESS schedulable.status == Action.Status.FAILED || schedulable.status == Action.Status.SUCCESS
} else {
true
} }
true
} }
timeWheel = TimeWheel(listScheduledActions, onTrigger, doneCondition) timeWheel = TimeWheel(listScheduledActions, onTrigger, doneCondition)
} }
@@ -126,6 +129,12 @@ class ActionScheduler : AbstractAgentModule.Standalone() {
val onTrigger: (toTrigger: Set<Schedulable>) -> Unit, val onTrigger: (toTrigger: Set<Schedulable>) -> Unit,
val doneCondition: (schedulable: Schedulable) -> Boolean val doneCondition: (schedulable: Schedulable) -> Boolean
) : Closeable { ) : Closeable {
private val tracePath: Path = ConfigCenter.paths.stateDir
.resolve("trace")
.resolve("time-wheel")
.normalize()
.toAbsolutePath()
private val schedulableGroupByHour = Array<MutableSet<Schedulable>>(24) { mutableSetOf() } private val schedulableGroupByHour = Array<MutableSet<Schedulable>>(24) { mutableSetOf() }
private val wheel = Array<MutableSet<Schedulable>>(60 * 60) { mutableSetOf() } private val wheel = Array<MutableSet<Schedulable>>(60 * 60) { mutableSetOf() }
private var recordHour: Int = -1 private var recordHour: Int = -1
@@ -151,15 +160,14 @@ class ActionScheduler : AbstractAgentModule.Standalone() {
logFailedStatus(schedulableData) logFailedStatus(schedulableData)
return@checkThenExecute return@checkThenExecute
} }
log.debug("Action next execution time: {}", parseToZonedDateTime) log.debug("${schedulableData.uuid} next execution time: {}", parseToZonedDateTime)
val hour = parseToZonedDateTime.hour val hour = parseToZonedDateTime.hour
schedulableGroupByHour[hour].add(schedulableData) schedulableGroupByHour[hour].add(schedulableData)
log.debug("Action scheduled at {}", hour)
if (it.hour == hour) { if (it.hour == hour) {
val wheelOffset = parseToZonedDateTime.minute * 60 + parseToZonedDateTime.second val wheelOffset = parseToZonedDateTime.minute * 60 + parseToZonedDateTime.second
wheel[wheelOffset].add(schedulableData) wheel[wheelOffset].add(schedulableData)
state.value = WheelState.ACTIVE state.value = WheelState.ACTIVE
log.debug("Action scheduled at wheel offset {}", wheelOffset) log.debug("${schedulableData.uuid} scheduled at wheel offset {}", wheelOffset)
} }
} }
} }
@@ -293,7 +301,7 @@ class ActionScheduler : AbstractAgentModule.Standalone() {
launchingHour, launchingHour,
includeRemainingHour = tick >= wheel.lastIndex includeRemainingHour = tick >= wheel.lastIndex
) )
if (tick >= wheel.lastIndex || schedulableGroupByHour[launchingHour].isEmpty()) { if (tick >= wheel.lastIndex) {
state.value = WheelState.SLEEPING state.value = WheelState.SLEEPING
shouldBreak = true shouldBreak = true
} }