///|
/// Minimal switched-fabric model for K1 partition/heal scenario hooks.
///|
pub(all) struct FabricCapabilities {
single_initiator : Bool
bounded_delay : Bool
total_order_delivery : Bool
partitionable : Bool
clock_sync_guarantee : Bool
background_load : Bool
} derive(Eq, Debug)
///|
pub fn FabricCapabilities::minimal_partitionable() -> FabricCapabilities {
{
single_initiator: false,
bounded_delay: false,
total_order_delivery: false,
partitionable: true,
clock_sync_guarantee: false,
background_load: false,
}
}
///|
pub fn FabricCapabilities::unlocks_partition_faults(
self : FabricCapabilities,
) -> Bool {
self.partitionable
}
///|
pub(all) struct FabricNode {
id : @core.EndpointId
name : String
} derive(Eq, Debug)
///|
pub fn FabricNode::new(id~ : @core.EndpointId, name~ : String) -> FabricNode {
{ id, name }
}
///|
pub(all) struct PortGroup {
name : String
members : Array[@core.EndpointId]
} derive(Eq, Debug)
///|
pub fn PortGroup::new(
name~ : String,
members~ : Array[@core.EndpointId],
) -> PortGroup {
{ name, members }
}
///|
pub fn PortGroup::contains(
self : PortGroup,
endpoint : @core.EndpointId,
) -> Bool {
self.members.any(fn(port) { port == endpoint })
}
///|
pub(all) struct MinimalFabric {
nodes : Array[FabricNode]
base_delay : @core.Duration
mut partitioned : Bool
mut groups : Array[PortGroup]
} derive(Debug)
///|
pub fn MinimalFabric::new(
nodes~ : Array[FabricNode],
base_delay~ : @core.Duration,
) -> MinimalFabric {
{ nodes, base_delay, partitioned: false, groups: [] }
}
///|
pub fn MinimalFabric::isolate(
self : MinimalFabric,
groups : Array[PortGroup],
) -> Unit {
self.partitioned = true
self.groups = copy_port_groups(groups)
}
///|
pub fn MinimalFabric::heal(self : MinimalFabric) -> Unit {
self.partitioned = false
self.groups = []
}
///|
pub fn MinimalFabric::is_partitioned(self : MinimalFabric) -> Bool {
self.partitioned
}
///|
pub fn MinimalFabric::can_deliver(
self : MinimalFabric,
source : @core.EndpointId,
target : @core.EndpointId,
) -> Bool {
if !self.has_node(source) || !self.has_node(target) {
return false
}
if !self.partitioned {
return true
}
let source_group = self.group_index(source)
let target_group = self.group_index(target)
source_group is Some(_) && source_group == target_group
}
///|
fn MinimalFabric::group_index(
self : MinimalFabric,
endpoint : @core.EndpointId,
) -> Int? {
for index, group in self.groups {
if group.contains(endpoint) {
return Some(index)
}
}
None
}
///|
fn MinimalFabric::has_node(
self : MinimalFabric,
endpoint : @core.EndpointId,
) -> Bool {
self.nodes.any(fn(node) { node.id == endpoint })
}
///|
pub impl @core.Medium for MinimalFabric with fn submit(self, ev) {
if !self.has_node(ev.source) {
raise @core.TransportError::UnboundEndpoint(ev.source)
}
if !self.has_node(ev.target) {
raise @core.TransportError::UnboundEndpoint(ev.target)
}
if !self.can_deliver(ev.source, ev.target) {
[]
} else {
[
{
id: ev.id + 1000,
parent_id: ev.id,
vtime: ev.vtime.add(self.base_delay),
channel_id: ev.channel_id,
seq: ev.seq,
source: ev.source,
target: ev.target,
frame: ev.frame.sending().sent().receiving().received(),
},
]
}
}
///|
pub impl @core.Medium for MinimalFabric with fn lookahead(self) {
self.base_delay
}
///|
pub(all) struct FabricDelivery {
tx_id : Int
source : @core.EndpointId
target : @core.EndpointId
submitted_at : @core.VTime
delivered_at : @core.VTime?
label : String
dropped : Bool
delayed_by : @core.Duration
} derive(Eq, Debug)
///|
pub(all) struct FabricPartitionScript {
seed : Int
partition_at : @core.VTime
heal_at : @core.VTime
groups : Array[PortGroup]
} derive(Eq, Debug)
///|
pub fn FabricPartitionScript::minimal(seed~ : Int) -> FabricPartitionScript {
{
seed,
partition_at: @core.VTime::from_ns(10L),
heal_at: @core.VTime::from_ns(80L),
groups: [
PortGroup::new(name="left", members=[
@core.EndpointId(1),
@core.EndpointId(2),
]),
PortGroup::new(name="right", members=[
@core.EndpointId(3),
@core.EndpointId(4),
]),
],
}
}
///|
pub(all) struct FabricRun {
seed : Int
capabilities : FabricCapabilities
script : FabricPartitionScript
deliveries : Array[FabricDelivery]
trace : @trace.TraceLog
digest : @core.SimDigest
partition_count : Int
heal_count : Int
drop_count : Int
late_count : Int
reorder_count : Int
} derive(Debug)
///|
pub fn run_fabric_partition_heal(seed~ : Int) -> FabricRun {
let capabilities = FabricCapabilities::minimal_partitionable()
let script = FabricPartitionScript::minimal(seed~)
let fabric = MinimalFabric::new(
nodes=[
FabricNode::new(id=@core.EndpointId(1), name="controller-a"),
FabricNode::new(id=@core.EndpointId(2), name="drive-a"),
FabricNode::new(id=@core.EndpointId(3), name="controller-b"),
FabricNode::new(id=@core.EndpointId(4), name="drive-b"),
],
base_delay=@core.Duration::from_ns(5L),
)
let env = @scenario.SimEnv::new(seed~)
let partition_timer = @scenario.timeout(
env,
@core.Duration::from_ns(script.partition_at.ns()),
)
let heal_timer = @scenario.timeout(
env,
@core.Duration::from_ns(script.heal_at.ns()),
)
let log = @trace.TraceLog::new()
let deliveries : Array[FabricDelivery] = []
let mut event_id = 1
let mut partition_count = 0
let mut heal_count = 0
let mut drop_count = 0
let mut late_count = 0
let mut reorder_count = 0
let pre = deliver(
fabric,
tx_id=1,
source=@core.EndpointId(1),
target=@core.EndpointId(3),
submitted_at=@core.VTime::from_ns(0L),
extra_delay=@core.Duration::from_ns(0L),
label="fabric.rx.delivered",
)
deliveries.push(pre)
event_id = append_delivery_trace(log, pre, seed~, event_id~)
ignore(env.run_until_event(partition_timer))
fabric.isolate(script.groups)
let partition_hit = @fault.TopologyFaultHit::make(
seed~,
step=event_id,
kind=@fault.Partition,
)
log.append(
partition_hit.to_trace_event(
event_id~,
vtime=env.now(),
clock_domain="sim",
node_id="fabric",
medium_id="fabric-min",
backend=@core.SimNative,
),
)
event_id += 1
partition_count += 1
let dropped = deliver(
fabric,
tx_id=2,
source=@core.EndpointId(1),
target=@core.EndpointId(4),
submitted_at=env.now(),
extra_delay=@core.Duration::from_ns(0L),
label="fabric.drop.partition",
)
deliveries.push(dropped)
event_id = append_delivery_trace(log, dropped, seed~, event_id~)
drop_count += 1
let jitter = @core.Duration::from_ns(
3L + Int64::from_int(normalized_seed(seed) % 5),
)
let late = deliver(
fabric,
tx_id=3,
source=@core.EndpointId(1),
target=@core.EndpointId(2),
submitted_at=env.now(),
extra_delay=jitter,
label="fabric.rx.late",
)
deliveries.push(late)
event_id = append_delivery_trace(log, late, seed~, event_id~)
late_count += 1
ignore(env.run_until_event(heal_timer))
fabric.heal()
let heal_hit = @fault.TopologyFaultHit::make(
seed~,
step=event_id,
kind=@fault.Heal,
)
log.append(
heal_hit.to_trace_event(
event_id~,
vtime=env.now(),
clock_domain="sim",
node_id="fabric",
medium_id="fabric-min",
backend=@core.SimNative,
),
)
event_id += 1
heal_count += 1
let slow = deliver(
fabric,
tx_id=4,
source=@core.EndpointId(3),
target=@core.EndpointId(1),
submitted_at=env.now(),
extra_delay=@core.Duration::from_ns(8L),
label="fabric.rx.reordered",
)
let fast = deliver(
fabric,
tx_id=5,
source=@core.EndpointId(3),
target=@core.EndpointId(1),
submitted_at=env.now(),
extra_delay=@core.Duration::from_ns(1L),
label="fabric.rx.reordered",
)
let ordered = if seed % 2 == 0 { [fast, slow] } else { [slow, fast] }
for delivery in ordered {
deliveries.push(delivery)
event_id = append_delivery_trace(log, delivery, seed~, event_id~)
}
reorder_count += 1
{
seed,
capabilities,
script,
deliveries,
trace: log,
digest: log.portable_digest(seed~),
partition_count,
heal_count,
drop_count,
late_count,
reorder_count,
}
}
///|
pub fn FabricRun::timeline_text(self : FabricRun) -> String {
let buf = StringBuilder::new()
buf.write_string("fabric=fabric-min|seed=")
buf.write_string(self.seed.to_string())
buf.write_string("|partitionable=")
buf.write_string(self.capabilities.partitionable.to_string())
buf.write_string("|events=")
buf.write_string(self.trace.len().to_string())
for event in self.trace.events {
buf.write_char('\n')
buf.write_string("event=")
buf.write_string(event.event_id.to_string())
buf.write_string("|label=")
buf.write_string(event.label)
buf.write_string("|time_ns=")
buf.write_string(event.vtime.ns().to_string())
}
buf.to_string()
}
///|
fn deliver(
fabric : MinimalFabric,
tx_id~ : Int,
source~ : @core.EndpointId,
target~ : @core.EndpointId,
submitted_at~ : @core.VTime,
extra_delay~ : @core.Duration,
label~ : String,
) -> FabricDelivery {
let delayed_by = @core.Duration::from_ns(
fabric.base_delay.ns() + extra_delay.ns(),
)
if !fabric.can_deliver(source, target) {
{
tx_id,
source,
target,
submitted_at,
delivered_at: None,
label,
dropped: true,
delayed_by: @core.Duration::from_ns(0L),
}
} else {
{
tx_id,
source,
target,
submitted_at,
delivered_at: Some(submitted_at.add(delayed_by)),
label,
dropped: false,
delayed_by,
}
}
}
///|
fn append_delivery_trace(
log : @trace.TraceLog,
delivery : FabricDelivery,
seed~ : Int,
event_id~ : Int,
) -> Int {
let delivered_at = match delivery.delivered_at {
Some(time) => time
None => delivery.submitted_at
}
log.append(
@trace.TraceEvent::make(
event_id~,
parent_id=Some(delivery.tx_id),
vtime=delivered_at,
clock_domain="sim",
raw_ns=delivered_at.ns(),
node_id="fabric",
medium_id="fabric-min",
channel_id=Some(@core.ChannelId(delivery.source.value())),
direction=if delivery.dropped { @trace.Fault } else { @trace.Rx },
payload_digest=Some(delivery.tx_id),
rng_step=event_id,
seed~,
backend=@core.SimNative,
label=delivery.label,
),
)
event_id + 1
}
///|
fn copy_port_groups(groups : Array[PortGroup]) -> Array[PortGroup] {
let out : Array[PortGroup] = []
for group in groups {
out.push(group)
}
out
}
///|
fn normalized_seed(seed : Int) -> Int {
if seed < 0 {
-seed
} else {
seed
}
}