1 changed files with 81 additions and 0 deletions
@ -0,0 +1,81 @@ |
|||||||
|
package simulation |
||||||
|
|
||||||
|
type DatabaseNode struct { |
||||||
|
ID string |
||||||
|
Label string |
||||||
|
Replication int |
||||||
|
CurrentLoad int |
||||||
|
Queue []*Request |
||||||
|
Replicas []*DatabaseNode |
||||||
|
Alive bool |
||||||
|
Targets []string |
||||||
|
output []*Request |
||||||
|
replicationQueue []*Request |
||||||
|
} |
||||||
|
|
||||||
|
func (n *DatabaseNode) GetID() string { return n.ID } |
||||||
|
func (n *DatabaseNode) Type() string { return "database" } |
||||||
|
func (n *DatabaseNode) IsAlive() bool { return n.Alive } |
||||||
|
func (n *DatabaseNode) GetQueue() []*Request { return n.Queue } |
||||||
|
|
||||||
|
func (n *DatabaseNode) Tick(tick int) { |
||||||
|
if len(n.Queue) == 0 { |
||||||
|
return |
||||||
|
} |
||||||
|
|
||||||
|
maxProcessPerTick := 3 |
||||||
|
processCount := min(len(n.Queue), maxProcessPerTick) |
||||||
|
|
||||||
|
queue := n.Queue |
||||||
|
n.Queue = n.Queue[:0] |
||||||
|
|
||||||
|
for i := 0; i < processCount; i++ { |
||||||
|
req := queue[i] |
||||||
|
|
||||||
|
if req.Type == "READ" { |
||||||
|
req.LatencyMS += 20 |
||||||
|
req.Path = append(req.Path, n.ID) |
||||||
|
n.output = append(n.output, req) |
||||||
|
} else { |
||||||
|
req.LatencyMS += 50 |
||||||
|
req.Path = append(req.Path, n.ID) |
||||||
|
|
||||||
|
// Replicate to replicas
|
||||||
|
for _, replica := range n.Replicas { |
||||||
|
replicationReq := &Request{ |
||||||
|
ID: req.ID + "-repl", |
||||||
|
Timestamp: req.Timestamp, |
||||||
|
LatencyMS: req.LatencyMS + 5, |
||||||
|
Origin: req.Origin, |
||||||
|
Type: "REPLICATION", |
||||||
|
Path: append(append([]string{}, req.Path...), "->"+replica.ID), |
||||||
|
} |
||||||
|
n.replicationQueue = append(n.replicationQueue, replicationReq) |
||||||
|
} |
||||||
|
|
||||||
|
n.output = append(n.output, req) |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
// Apply queuing penalty if overloaded
|
||||||
|
if len(n.Queue) > 10 { |
||||||
|
for _, req := range n.Queue { |
||||||
|
req.LatencyMS += 10 |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
func (n *DatabaseNode) Receive(req *Request) { |
||||||
|
if req == nil { |
||||||
|
return |
||||||
|
} |
||||||
|
req.LatencyMS += 2 // DB connection overhead
|
||||||
|
n.Queue = append(n.Queue, req) |
||||||
|
} |
||||||
|
|
||||||
|
func (n *DatabaseNode) Emit() []*Request { |
||||||
|
out := append(n.output, n.replicationQueue...) |
||||||
|
n.output = n.output[:0] |
||||||
|
n.replicationQueue = n.replicationQueue[:0] |
||||||
|
return out |
||||||
|
} |
||||||
Loading…
Reference in new issue