feat: add deterministic regional scheduler
This commit is contained in:
Rijad Zuzo
@@ -0,0 +1,59 @@
|
||||
class_name AnalyticalRegionalUpdates
|
||||
extends RefCounted
|
||||
|
||||
|
||||
static func linear_value(
|
||||
current_value: float,
|
||||
rate_per_tick: float,
|
||||
elapsed_ticks: int,
|
||||
minimum: float = -INF,
|
||||
maximum: float = INF
|
||||
) -> float:
|
||||
if elapsed_ticks <= 0 or not is_finite(current_value) or not is_finite(rate_per_tick):
|
||||
return clampf(current_value, minimum, maximum)
|
||||
return clampf(current_value + rate_per_tick * elapsed_ticks, minimum, maximum)
|
||||
|
||||
|
||||
static func exponential_value(
|
||||
current_value: float,
|
||||
multiplier_per_tick: float,
|
||||
elapsed_ticks: int,
|
||||
minimum: float = -INF,
|
||||
maximum: float = INF
|
||||
) -> float:
|
||||
if (
|
||||
elapsed_ticks <= 0
|
||||
or not is_finite(current_value)
|
||||
or not is_finite(multiplier_per_tick)
|
||||
or multiplier_per_tick < 0.0
|
||||
):
|
||||
return clampf(current_value, minimum, maximum)
|
||||
return clampf(current_value * pow(multiplier_per_tick, elapsed_ticks), minimum, maximum)
|
||||
|
||||
|
||||
static func periodic_occurrences(
|
||||
first_due_tick: int, interval_ticks: int, through_tick: int
|
||||
) -> int:
|
||||
if first_due_tick < 0 or interval_ticks <= 0 or through_tick < first_due_tick:
|
||||
return 0
|
||||
return (through_tick - first_due_tick) / interval_ticks + 1
|
||||
|
||||
|
||||
static func transfer_amount(
|
||||
available: float,
|
||||
demand: float,
|
||||
rate_per_tick: float,
|
||||
elapsed_ticks: int,
|
||||
loss_fraction: float = 0.0
|
||||
) -> Dictionary:
|
||||
var sent := minf(
|
||||
maxf(available, 0.0),
|
||||
minf(maxf(demand, 0.0), maxf(rate_per_tick, 0.0) * maxi(elapsed_ticks, 0))
|
||||
)
|
||||
var received := sent * (1.0 - clampf(loss_fraction, 0.0, 1.0))
|
||||
return {
|
||||
"sent": sent,
|
||||
"received": received,
|
||||
"remaining_available": maxf(available - sent, 0.0),
|
||||
"remaining_demand": maxf(demand - received, 0.0),
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
uid://bw56tsrd8qtgr
|
||||
@@ -0,0 +1,69 @@
|
||||
class_name KeyedRandom
|
||||
extends RefCounted
|
||||
|
||||
const MASK_32 := 0xFFFFFFFF
|
||||
const UINT32_RANGE := 4294967296.0
|
||||
|
||||
|
||||
static func value_u32(
|
||||
world_seed: int, system_id: StringName, entity_id: StringName, occurrence: int
|
||||
) -> int:
|
||||
var hash_value := int(world_seed) & MASK_32
|
||||
hash_value = _mix_u32(hash_value ^ 0x9E3779B9)
|
||||
for byte in String(system_id).to_utf8_buffer():
|
||||
hash_value = _mix_u32(hash_value ^ int(byte))
|
||||
hash_value = _mix_u32(hash_value ^ 0x85EBCA6B)
|
||||
for byte in String(entity_id).to_utf8_buffer():
|
||||
hash_value = _mix_u32(hash_value ^ int(byte))
|
||||
hash_value = _mix_u32(hash_value ^ (occurrence & MASK_32))
|
||||
return hash_value
|
||||
|
||||
|
||||
static func value_float(
|
||||
world_seed: int, system_id: StringName, entity_id: StringName, occurrence: int
|
||||
) -> float:
|
||||
return float(value_u32(world_seed, system_id, entity_id, occurrence)) / UINT32_RANGE
|
||||
|
||||
|
||||
static func range_int(
|
||||
world_seed: int,
|
||||
system_id: StringName,
|
||||
entity_id: StringName,
|
||||
occurrence: int,
|
||||
minimum: int,
|
||||
maximum: int
|
||||
) -> int:
|
||||
if maximum < minimum:
|
||||
return minimum
|
||||
var width := maximum - minimum + 1
|
||||
return minimum + value_u32(world_seed, system_id, entity_id, occurrence) % width
|
||||
|
||||
|
||||
static func range_float(
|
||||
world_seed: int,
|
||||
system_id: StringName,
|
||||
entity_id: StringName,
|
||||
occurrence: int,
|
||||
minimum: float,
|
||||
maximum: float
|
||||
) -> float:
|
||||
if maximum <= minimum:
|
||||
return minimum
|
||||
return lerpf(minimum, maximum, value_float(world_seed, system_id, entity_id, occurrence))
|
||||
|
||||
|
||||
static func chance(
|
||||
world_seed: int,
|
||||
system_id: StringName,
|
||||
entity_id: StringName,
|
||||
occurrence: int,
|
||||
probability: float
|
||||
) -> bool:
|
||||
return value_float(world_seed, system_id, entity_id, occurrence) < clampf(probability, 0.0, 1.0)
|
||||
|
||||
|
||||
static func _mix_u32(value: int) -> int:
|
||||
var mixed := value & MASK_32
|
||||
mixed = ((mixed ^ (mixed >> 16)) * 0x7FEB352D) & MASK_32
|
||||
mixed = ((mixed ^ (mixed >> 15)) * 0x846CA68B) & MASK_32
|
||||
return (mixed ^ (mixed >> 16)) & MASK_32
|
||||
@@ -0,0 +1 @@
|
||||
uid://c00kbf23ey8na
|
||||
@@ -0,0 +1,349 @@
|
||||
class_name RegionalJobScheduler
|
||||
extends RefCounted
|
||||
|
||||
const SCHEMA_VERSION := 1
|
||||
const UNLIMITED_BUDGET := -1
|
||||
|
||||
var _jobs_by_id: Dictionary = {}
|
||||
var _job_heap: Array[ScheduledJobRecord] = []
|
||||
var _heap_index_by_id: Dictionary = {}
|
||||
var _dedupe_job_ids: Dictionary = {}
|
||||
var _sequence_job_ids: Dictionary = {}
|
||||
var _next_stable_sequence := 0
|
||||
var _last_drain_tick := -1
|
||||
var _executed_count := 0
|
||||
var _last_executed_job: ScheduledJobRecord
|
||||
|
||||
|
||||
static func from_dictionary(record_data: Dictionary) -> RegionalJobScheduler:
|
||||
var scheduler := RegionalJobScheduler.new()
|
||||
return scheduler if scheduler.restore_from_dictionary(record_data) else null
|
||||
|
||||
|
||||
func size() -> int:
|
||||
return _jobs_by_id.size()
|
||||
|
||||
|
||||
func is_empty() -> bool:
|
||||
return _jobs_by_id.is_empty()
|
||||
|
||||
|
||||
func has_job(job_id: StringName) -> bool:
|
||||
return _jobs_by_id.has(job_id)
|
||||
|
||||
|
||||
func get_by_id(job_id: StringName) -> ScheduledJobRecord:
|
||||
return _jobs_by_id.get(job_id) as ScheduledJobRecord
|
||||
|
||||
|
||||
func schedule(job: ScheduledJobRecord) -> bool:
|
||||
if (
|
||||
job == null
|
||||
or not job.is_valid()
|
||||
or has_job(job.get_job_id())
|
||||
or _sequence_job_ids.has(job.get_stable_sequence())
|
||||
or (_last_executed_job != null and not ScheduledJobRecord.compare(_last_executed_job, job))
|
||||
):
|
||||
return false
|
||||
var dedupe_key := job.get_dedupe_key()
|
||||
if not dedupe_key.is_empty() and _dedupe_job_ids.has(dedupe_key):
|
||||
return false
|
||||
var stored := ScheduledJobRecord.from_dictionary(job.to_dictionary())
|
||||
if stored == null:
|
||||
return false
|
||||
_jobs_by_id[stored.get_job_id()] = stored
|
||||
_sequence_job_ids[stored.get_stable_sequence()] = stored.get_job_id()
|
||||
if not dedupe_key.is_empty():
|
||||
_dedupe_job_ids[dedupe_key] = stored.get_job_id()
|
||||
_heap_push(stored)
|
||||
_next_stable_sequence = maxi(_next_stable_sequence, stored.get_stable_sequence() + 1)
|
||||
return true
|
||||
|
||||
|
||||
func schedule_new(
|
||||
job_id: StringName,
|
||||
job_type: StringName,
|
||||
due_tick: int,
|
||||
phase: int,
|
||||
entity_id: StringName,
|
||||
payload: Dictionary = {},
|
||||
dedupe_key: StringName = &"",
|
||||
repeat_interval: int = ScheduledJobRecord.NO_REPEAT
|
||||
) -> ScheduledJobRecord:
|
||||
var job := ScheduledJobRecord.create(
|
||||
job_id,
|
||||
job_type,
|
||||
due_tick,
|
||||
phase,
|
||||
entity_id,
|
||||
_next_stable_sequence,
|
||||
payload,
|
||||
dedupe_key,
|
||||
repeat_interval
|
||||
)
|
||||
return job if schedule(job) else null
|
||||
|
||||
|
||||
func cancel(job_id: StringName) -> ScheduledJobRecord:
|
||||
var job := get_by_id(job_id)
|
||||
if job == null:
|
||||
return null
|
||||
var remove_index := int(_heap_index_by_id[job_id])
|
||||
var last_index := _job_heap.size() - 1
|
||||
if remove_index != last_index:
|
||||
_heap_swap(remove_index, last_index)
|
||||
_job_heap.pop_back()
|
||||
_heap_index_by_id.erase(job_id)
|
||||
_jobs_by_id.erase(job_id)
|
||||
_sequence_job_ids.erase(job.get_stable_sequence())
|
||||
var dedupe_key := job.get_dedupe_key()
|
||||
if not dedupe_key.is_empty():
|
||||
_dedupe_job_ids.erase(dedupe_key)
|
||||
if remove_index < _job_heap.size():
|
||||
var parent := (remove_index - 1) / 2
|
||||
if (
|
||||
remove_index > 0
|
||||
and ScheduledJobRecord.compare(_job_heap[remove_index], _job_heap[parent])
|
||||
):
|
||||
_heap_sift_up(remove_index)
|
||||
else:
|
||||
_heap_sift_down(remove_index)
|
||||
return job
|
||||
|
||||
|
||||
func peek_next() -> ScheduledJobRecord:
|
||||
return _job_heap[0] if not _job_heap.is_empty() else null
|
||||
|
||||
|
||||
func get_all_sorted() -> Array[ScheduledJobRecord]:
|
||||
var jobs := _job_heap.duplicate()
|
||||
jobs.sort_custom(ScheduledJobRecord.compare)
|
||||
return jobs
|
||||
|
||||
|
||||
func get_due_count(current_tick: int) -> int:
|
||||
var count := 0
|
||||
for job in _jobs_by_id.values():
|
||||
if (job as ScheduledJobRecord).get_due_tick() <= current_tick:
|
||||
count += 1
|
||||
return count
|
||||
|
||||
|
||||
func take_due(
|
||||
current_tick: int, execution_budget: int = UNLIMITED_BUDGET
|
||||
) -> Array[ScheduledJobRecord]:
|
||||
var due: Array[ScheduledJobRecord] = []
|
||||
if current_tick < 0 or current_tick < _last_drain_tick or execution_budget < UNLIMITED_BUDGET:
|
||||
return due
|
||||
_last_drain_tick = current_tick
|
||||
if execution_budget == 0:
|
||||
return due
|
||||
var remaining := execution_budget
|
||||
while not _job_heap.is_empty():
|
||||
var next := _job_heap[0]
|
||||
if next.get_due_tick() > current_tick:
|
||||
break
|
||||
if remaining == 0:
|
||||
break
|
||||
var job := cancel(next.get_job_id())
|
||||
due.append(job)
|
||||
_executed_count += 1
|
||||
_last_executed_job = job
|
||||
if job.is_repeating():
|
||||
var next_occurrence := job.create_next_occurrence(_next_stable_sequence)
|
||||
if not schedule(next_occurrence):
|
||||
push_error("Could not schedule next occurrence for '%s'" % job.get_job_id())
|
||||
if remaining > 0:
|
||||
remaining -= 1
|
||||
return due
|
||||
|
||||
|
||||
func execute_due(
|
||||
current_tick: int, execution_budget: int, executor: Callable
|
||||
) -> Array[ScheduledJobRecord]:
|
||||
var executed: Array[ScheduledJobRecord] = []
|
||||
if not executor.is_valid() or execution_budget < UNLIMITED_BUDGET:
|
||||
return executed
|
||||
var remaining := execution_budget
|
||||
while remaining != 0:
|
||||
var batch := take_due(current_tick, 1)
|
||||
if batch.is_empty():
|
||||
break
|
||||
var job := batch[0]
|
||||
executed.append(job)
|
||||
executor.call(job)
|
||||
if remaining > 0:
|
||||
remaining -= 1
|
||||
return executed
|
||||
|
||||
|
||||
func restore(records: Array[ScheduledJobRecord], cursor: Dictionary = {}) -> bool:
|
||||
var candidate := RegionalJobScheduler.new()
|
||||
for record in records:
|
||||
if not candidate.schedule(record):
|
||||
return false
|
||||
var next_sequence := int(cursor.get("next_stable_sequence", candidate._next_stable_sequence))
|
||||
var last_drain_tick := int(cursor.get("last_drain_tick", -1))
|
||||
var executed_count := int(cursor.get("executed_count", 0))
|
||||
var saved_last_order: Variant = cursor.get("last_executed_order")
|
||||
var last_executed_job := _parse_cursor_order(saved_last_order)
|
||||
if (
|
||||
next_sequence < candidate._next_stable_sequence
|
||||
or last_drain_tick < -1
|
||||
or executed_count < 0
|
||||
or (executed_count == 0 and last_executed_job != null)
|
||||
or (executed_count > 0 and last_executed_job == null)
|
||||
or (saved_last_order != null and last_executed_job == null)
|
||||
or (
|
||||
last_executed_job != null
|
||||
and (
|
||||
last_executed_job.get_due_tick() > last_drain_tick
|
||||
or next_sequence <= last_executed_job.get_stable_sequence()
|
||||
)
|
||||
)
|
||||
):
|
||||
return false
|
||||
if (
|
||||
last_executed_job != null
|
||||
and not candidate._job_heap.is_empty()
|
||||
and not ScheduledJobRecord.compare(last_executed_job, candidate._job_heap[0])
|
||||
):
|
||||
return false
|
||||
_adopt(candidate)
|
||||
_next_stable_sequence = next_sequence
|
||||
_last_drain_tick = last_drain_tick
|
||||
_executed_count = executed_count
|
||||
_last_executed_job = last_executed_job
|
||||
return true
|
||||
|
||||
|
||||
func restore_from_dictionary(record_data: Dictionary) -> bool:
|
||||
if int(record_data.get("schema_version", -1)) != SCHEMA_VERSION:
|
||||
return false
|
||||
if not record_data.has_all(["jobs", "cursor"]) or not record_data["jobs"] is Array:
|
||||
return false
|
||||
if not record_data["cursor"] is Dictionary:
|
||||
return false
|
||||
if not (
|
||||
record_data["cursor"]
|
||||
. has_all(
|
||||
[
|
||||
"next_stable_sequence",
|
||||
"last_drain_tick",
|
||||
"executed_count",
|
||||
"last_executed_order",
|
||||
]
|
||||
)
|
||||
):
|
||||
return false
|
||||
var records: Array[ScheduledJobRecord] = []
|
||||
for saved_job in record_data["jobs"]:
|
||||
if not saved_job is Dictionary:
|
||||
return false
|
||||
var job := ScheduledJobRecord.from_dictionary(saved_job)
|
||||
if job == null:
|
||||
return false
|
||||
records.append(job)
|
||||
return restore(records, record_data["cursor"])
|
||||
|
||||
|
||||
func get_cursor() -> Dictionary:
|
||||
return {
|
||||
"next_stable_sequence": _next_stable_sequence,
|
||||
"last_drain_tick": _last_drain_tick,
|
||||
"executed_count": _executed_count,
|
||||
"last_executed_order": _serialize_cursor_order(_last_executed_job),
|
||||
}
|
||||
|
||||
|
||||
func to_dictionary() -> Dictionary:
|
||||
var jobs: Array[Dictionary] = []
|
||||
for job in get_all_sorted():
|
||||
jobs.append(job.to_dictionary())
|
||||
return {
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"jobs": jobs,
|
||||
"cursor": get_cursor(),
|
||||
}
|
||||
|
||||
|
||||
func _heap_push(job: ScheduledJobRecord) -> void:
|
||||
var index := _job_heap.size()
|
||||
_job_heap.append(job)
|
||||
_heap_index_by_id[job.get_job_id()] = index
|
||||
_heap_sift_up(index)
|
||||
|
||||
|
||||
func _heap_sift_up(start_index: int) -> void:
|
||||
var index := start_index
|
||||
while index > 0:
|
||||
var parent := (index - 1) / 2
|
||||
if not ScheduledJobRecord.compare(_job_heap[index], _job_heap[parent]):
|
||||
break
|
||||
_heap_swap(index, parent)
|
||||
index = parent
|
||||
|
||||
|
||||
func _heap_sift_down(start_index: int) -> void:
|
||||
var index := start_index
|
||||
while true:
|
||||
var left := index * 2 + 1
|
||||
if left >= _job_heap.size():
|
||||
return
|
||||
var right := left + 1
|
||||
var smallest := left
|
||||
if (
|
||||
right < _job_heap.size()
|
||||
and ScheduledJobRecord.compare(_job_heap[right], _job_heap[left])
|
||||
):
|
||||
smallest = right
|
||||
if not ScheduledJobRecord.compare(_job_heap[smallest], _job_heap[index]):
|
||||
return
|
||||
_heap_swap(index, smallest)
|
||||
index = smallest
|
||||
|
||||
|
||||
func _heap_swap(first_index: int, second_index: int) -> void:
|
||||
var first := _job_heap[first_index]
|
||||
_job_heap[first_index] = _job_heap[second_index]
|
||||
_job_heap[second_index] = first
|
||||
_heap_index_by_id[_job_heap[first_index].get_job_id()] = first_index
|
||||
_heap_index_by_id[_job_heap[second_index].get_job_id()] = second_index
|
||||
|
||||
|
||||
func _adopt(source: RegionalJobScheduler) -> void:
|
||||
_jobs_by_id = source._jobs_by_id
|
||||
_job_heap = source._job_heap
|
||||
_heap_index_by_id = source._heap_index_by_id
|
||||
_dedupe_job_ids = source._dedupe_job_ids
|
||||
_sequence_job_ids = source._sequence_job_ids
|
||||
_next_stable_sequence = source._next_stable_sequence
|
||||
|
||||
|
||||
static func _serialize_cursor_order(job: ScheduledJobRecord) -> Variant:
|
||||
if job == null:
|
||||
return null
|
||||
return {
|
||||
"due_tick": job.get_due_tick(),
|
||||
"phase": job.get_phase(),
|
||||
"entity_id": String(job.get_entity_id()),
|
||||
"stable_sequence": job.get_stable_sequence(),
|
||||
}
|
||||
|
||||
|
||||
static func _parse_cursor_order(value: Variant) -> ScheduledJobRecord:
|
||||
if value == null:
|
||||
return null
|
||||
if (
|
||||
not value is Dictionary
|
||||
or not value.has_all(["due_tick", "phase", "entity_id", "stable_sequence"])
|
||||
):
|
||||
return null
|
||||
return ScheduledJobRecord.create(
|
||||
&"__cursor__",
|
||||
&"__cursor__",
|
||||
int(value["due_tick"]),
|
||||
int(value["phase"]),
|
||||
StringName(value["entity_id"]),
|
||||
int(value["stable_sequence"])
|
||||
)
|
||||
@@ -0,0 +1 @@
|
||||
uid://d03xv4blyvbr6
|
||||
Reference in New Issue
Block a user