Hello there, It is been a while.
I came across Fly.io Distributed Systems Challenge and decided to solve the problems to get my hands dirty with distributed systems concepts. Since I was familiar with Go and challenge provided a Go Library, I decided to complete the challenge with go. The code is available at Github. I have only added the relevant code snippets here and rest of the code can be viewed on github. Whether you're new to distributed systems or looking to solidify concepts like consensus, replication, and fault tolerance through hands-on code, this walkthrough should give you a practical, Go-based reference to learn from or compare against your own solutions.
Challenge 1 is easy, it is just to get the hang of the maelstrom package and the functions it provides.
func echo(m maelstrom.Message) any {
msg := body{}
if err := json.Unmarshal(m.Body, &msg); err != nil {
return fmt.Errorf("ECHO: Error while decoding json: %s", err)
}
return map[string]any{"type": "echo_ok", "echo": msg.Echo}
}
Generate
ID generation is based on the ideas behind Twitter's Snowflake algorithm, which Twitter uses to generate 64-bit unique identifiers across all the nodes without any contention. The ID is made up of
- A single bit,always set to 0, to denote a positive number.
- 41 bits which represent the number of milliseconds passed since a custom epoch.
- 10 bits to represent a machine ID.
- 12 bits to represent the per-machine sequence number of the ID generated.
A single server can create up to 4096 IDs per millisecond and all the IDs are sortable by time.
Based on the above principles, the ID generator for Maelstrom consists of a counter to keep track of the sequence and the timestamp.
const (
epoch = int64(1672531200000)
seqBits = int64(12)
machineIdBits = int64(10)
maxSeqBits = int64(-1) ^ (int64(-1) << seqBits)
maxMachineBits = int64(-1) ^ (int64(-1) << machineIdBits)
machineIdShift = seqBits
timestampShift = seqBits + machineIdBits
)
type generator struct {
mu sync.Mutex
seq int64
timestamp int64
}
func (g *generator) GetID(nodeId string) int64 {
g.mu.Lock()
defer g.mu.Unlock()
machineId := getMachineId(nodeId)
currTimestamp := time.Now().UnixMilli()
currTimestamp = max(currTimestamp, g.timestamp)
if currTimestamp == g.timestamp {
g.seq = (g.seq + 1) & maxSeqBits
if g.seq == 0 {
for currTimestamp <= g.timestamp {
currTimestamp = time.Now().UnixMilli()
}
}
} else {
g.seq = 0
}
g.timestamp = currTimestamp
id := ((currTimestamp - epoch) << timestampShift) | (machineId << machineIdShift) | g.seq
return id
}
This ensures that a new unique ID is generated across various nodes. Machine ID must be unique, or it would lead to collisions.
Broadcast
Broadcast starts off with defining a few handlers for the basic operations such as
broadcast,
read,
and
topology
.
Broadcast
saves the input integer,
read
returns the list of saved integers and
topology
provides a node topology which will be used in the later exercises.
type config struct {
node *maelstrom.Node
topology map[string][]string
retryHandler retry.RetryHandler
store *store
}
func (c *config) setTopology(msg maelstrom.Message) error {
body := topologyMsg{}
if err := json.Unmarshal(msg.Body, &body); err != nil {
return fmt.Errorf("TOPOLOGY: Error while decoding json: %s", err)
}
c.topology = body.Topology
return c.node.Reply(msg, reply{Type: "topology_ok"})
}
func (c *config) read(msg maelstrom.Message) error {
return c.node.Reply(msg, reply{Type: "read_ok", Messages: c.store.Get()})
}
func (c *config) broadcast(msg maelstrom.Message) error {
body := broadcastMsg{}
if err := json.Unmarshal(msg.Body, &body); err != nil {
return fmt.Errorf("BROADCAST: Error while decoding json: %s", err)
}
c.store.Store(body.Message)
return c.node.Reply(msg, reply{Type: "broadcast_ok"})
}
Store
is a data structure with an array which stores the values, a cache which stores whether a number is stored or not and a mutex to enable atomic operations.
import "sync"
type store struct {
mu sync.RWMutex
cache map[int]bool
store []int
}
func NewStore() *store {
return &store{
mu: sync.RWMutex{},
store: make([]int, 0),
cache: make(map[int]bool),
}
}
func (s *store) Store(value int) {
s.mu.Lock()
defer s.mu.Unlock()
if isPresent := s.cache[value]; isPresent {
return
}
s.store = append(s.store, value)
s.cache[value] = true
}
func (s *store) StoreMultiple(values []int) []int {
s.mu.Lock()
defer s.mu.Unlock()
newValues := make([]int, 0)
for _, value := range values {
if isPresent := s.cache[value]; isPresent {
continue
}
s.store = append(s.store, value)
s.cache[value] = true
newValues = append(newValues, value)
}
return newValues
}
func (s *store) Get() []int {
s.mu.RLock()
defer s.mu.RUnlock()
storeCopy := make([]int, len(s.store))
copy(storeCopy, s.store)
return storeCopy
}
The exercises 3B, 3C are multi node exercises and hence the code needs to be modified.
While 3B requires us to send the received integer to the rest of the connected nodes, 3C introduces network partitioning. Essentially for a certain period of time, nodes might not be able to send messages to each other.
To solve this problem, we can introduce a retry mechanism which works asynchronously. Once a node receives a message, it can store the message and broadcast to other nodes. In case of network timeout, we can retry until the network partition is healed.
package retry
import (
"context"
"errors"
"math"
"sync"
"time"
maelstrom "github.com/jepsen-io/maelstrom/demo/go"
)
type RetryHandler interface {
Send(dst string, msgType string, values ...any)
}
type RetryCfg struct {
mu *sync.RWMutex
dstConnMap map[string][]any
isWorking map[string]bool
node *maelstrom.Node
}
func NewRetryHandler(node *maelstrom.Node) RetryHandler {
return &RetryCfg{
mu: &sync.RWMutex{},
dstConnMap: make(map[string][]any),
isWorking: make(map[string]bool),
node: node,
}
}
func (b *RetryCfg) Send(dst string, msgType string, values ...any) {
b.mu.Lock()
defer b.mu.Unlock()
if _, ok := b.dstConnMap[dst]; !ok {
b.dstConnMap[dst] = make([]any, 0)
}
b.dstConnMap[dst] = append(b.dstConnMap[dst], values...)
isWorking := b.isWorking[dst]
if !isWorking {
b.isWorking[dst] = true
go func() {
count := 0
waitTime := 100
contextTime := 500
for {
time.Sleep(time.Millisecond * time.Duration(waitTime*pow(2, count)))
b.mu.RLock()
values := b.dstConnMap[dst]
b.mu.RUnlock()
ctx, cancel := context.WithTimeout(context.TODO(), time.Millisecond*time.Duration(contextTime))
_, err := b.node.SyncRPC(ctx, dst, getMessage(msgType, values))
cancel()
if maelstrom.ErrorCode(err) == maelstrom.Timeout || errors.Is(err, context.DeadlineExceeded) {
count++
continue
}
b.mu.Lock()
b.dstConnMap[dst] = b.dstConnMap[dst][len(values):]
remaining := len(b.dstConnMap[dst])
if remaining == 0 {
b.isWorking[dst] = false
}
b.mu.Unlock()
if remaining > 0 {
count = 0
continue
}
return
}
}()
}
}
func getMessage(msgType string, values []any) map[string]any {
if msgType == "txn-update" {
return map[string]any{"type": msgType, "txn": values}
}
return map[string]any{"type": msgType, "message": values}
}
func pow(x, y int) int {
return int(math.Pow(float64(x), float64(y)))
}
The above is the implementation of the retry mechanism, which ensures messages are delivered to the destination nodes by retrying until it succeeds. If
node.SyncRPC()
fails due to a timeout error, the action is retried again. Each time
node.SyncRPC()
fails, count is incremented by 1. The code
time.Sleep(time.Millisecond * time.Duration(waitTime*pow(2, count)))
is part of the exponential back-off which ensures each node does not flood the network during a network partition.
This satisfies 3C, which requires working with network partitioning.
A keen viewer will notice that it is not an exact representation of a retry logic. This is part of 3D and 3E which require us to minimize the messages-per-operation, median latency and maximum latency.
dstConnMap
holds a message array for each destination node. When source node wants to send a message, it is first stored in this map. The initial sleep is used as a batch operation so messages can be sent in batches rather than individually. During each retry, the latest message array is fetched so all the data is sent as a single batch. This reduces the msgs-per-op but increases latency.
Another way to do this is to move the batch processing only during retries. This increases the msgs-per-op but decreases latency. A goroutine is spawned for each destination node and once all the data is sent, the goroutine exits. Once a new set of data is required to be sent, a new go thread is spawned. So our broadcast works in the event of a network partition and still within the limits present in 3E.
The updated code for broadcast is below.
func (c *config) broadcast(msg maelstrom.Message) error {
body := broadcastMsg{}
if err := json.Unmarshal(msg.Body, &body); err != nil {
return fmt.Errorf("BROADCAST: Error while decoding json: %s", err)
}
c.store.Store(body.Message)
nodeId := c.node.ID()
for _, connectedNode := range c.topology[nodeId] {
if connectedNode == msg.Src || connectedNode == nodeId {
continue
}
c.retryHandler.Send(connectedNode, "broadcast-group", body.Message)
}
return c.node.Reply(msg, reply{Type: "broadcast_ok"})
}
func (c *config) broadcastGroup(msg maelstrom.Message) error {
body := broadcastGrpMsg{}
if err := json.Unmarshal(msg.Body, &body); err != nil {
return fmt.Errorf("BROADCAST: Error while decoding json: %s", err)
}
body.Message = c.store.StoreMultiple(body.Message)
messages := make([]any, len(body.Message))
for i, msg := range body.Message {
messages[i] = msg
}
if len(messages) == 0 {
return c.node.Reply(msg, reply{Type: "broadcast_ok"})
}
nodeId := c.node.ID()
for _, connectedNode := range c.topology[nodeId] {
if connectedNode == msg.Src {
continue
}
c.retryHandler.Send(connectedNode, "broadcast-group", messages...)
}
return c.node.Reply(msg, reply{Type: "broadcast_ok"})
}
Grow-Only Counter
This challenge introduces us to build a counter distributed across multiple nodes and handle a network partition as well. The challenge introduces us to a sequentially consistent key/value store service which can be used to store data.
The simplest approach to solving this problem is to increment the counter with the incoming delta and store it.
func (c *config) add(msg maelstrom.Message) error {
body := input{}
if err := json.Unmarshal(msg.Body, &body); err != nil {
return fmt.Errorf("Add: Error while decoding json: %s", err)
}
for {
oldVal, err := c.kv.ReadInt(context.TODO(), c.node.ID())
newVal := oldVal + int(body.Delta)
isNewKey := maelstrom.ErrorCode(err) == maelstrom.KeyDoesNotExist
if err := c.kv.CompareAndSwap(context.TODO(), c.node.ID(), oldVal, newVal, isNewKey); err != nil {
continue
}
break
}
return c.node.Reply(msg, simpleOutput{Type: "add_ok"})
}
When the counter value is requested, the node handling the request reads its own counter value from the store, fetches the other nodes' counter values, and returns the sum. This way, we can handle a network partition by getting values from nodes which are available. During network partition, we only get the partial sum. Once the network partition is healed, we can retrieve the final counter value.
This is a concept called Conflict free-Replicated Data Type(CRDT). CRDTs allow multiple nodes to update data without a central coordinator or synchronization system. State based CRDTs require 3 functions
- Function to produce initial state
- Function to merge states of all the nodes
- Function to update the state of a node.
The merge function must be commutative, associative, and idempotent.
func (c *config) read(msg maelstrom.Message) error {
sum, err := c.kv.ReadInt(context.TODO(), c.node.ID())
if maelstrom.ErrorCode(err) == maelstrom.KeyDoesNotExist {
sum = 0
}
for _, id := range c.node.NodeIDs() {
if id == c.node.ID() {
continue
}
ctx, cancel := context.WithTimeout(context.TODO(), time.Millisecond*200)
recvMsg, err := c.node.SyncRPC(ctx, id, simpleOutput{Type: "get_counter"})
cancel()
if err != nil {
continue
}
body := output{}
if err := json.Unmarshal(recvMsg.Body, &body); err != nil {
return fmt.Errorf("Add: Error while decoding json: %s", err)
}
sum += int(body.Value)
}
return c.node.Reply(msg, output{Type: "read_ok", Value: sum})
}
func (c *config) getCounter(msg maelstrom.Message) error {
val, err := c.kv.ReadInt(context.TODO(), c.node.ID())
if maelstrom.ErrorCode(err) == maelstrom.KeyDoesNotExist {
val = 0
}
return c.node.Reply(msg, output{Type: "get_counter_ok", Value: val})
}
There is a trade-off between read and write operations. Read operations are instantaneous since they are stored locally. But write operation is costly since each node has to be queried. If done asynchronously, the execution time would be the maximum latency to get the data from a node.
Nodes using CRDTs usually share their state with other nodes. This ensures that even if a node is not available, cached data of the missing nodes can be used temporarily. This is not present in the above code.
Kafka
This challenge expects us to implement a replicated log service like Kafka.
The nodes need to store an append only log to handle the Kafka workload.
Challenge 5a introduces us to the required operations which are
send,
poll,
commit_offsets,
and
list_committed_offsets
Each
send
request contains a key and a value. The node must reply with an offset.
{
"type": "send",
"key": "k1",
"msg": 123
}
A
poll
request requests the node to return messages from a set of logs starting from a given offset.
{
"type": "poll",
"offsets": {
"k1": 1000,
"k2": 2000
}
}
A
commit_offsets
request informs the node that messages have been successfully processed up to and including the given offset.
{
"type": "commit_offsets",
"offsets": {
"k1": 1000,
"k2": 2000
}
}
Client uses a
list_committed_offsets
request to fetch a map of committed offsets for a given set of keys. It is used by the clients to figure out where to start consuming from for a given key.
{
"type": "list_committed_offsets",
"keys": ["k1", "k2"]
}
Challenge 5a expects a single node system. All the above operations can be done by using a
map[int][]int
to store key along with array of values with offset. Another map
map[int]int
can be used to store key and the corresponding committed offset.
Challenge 5b expects the node to handle the above requests in a multi-node setup while handling network partitions. The challenge introduces us to a linearizable version of the KV store similar to the sequential KV introduced in the Counter section.
Sequential Consistency
Sequential consistency is a consistency model for distributed systems formalized by Lamport. It states that
The Result of any execution is the same as if the operations of all the processors were executed in some sequential order, and the operations of each individual processor occur in this sequence in the order specified by its program
Each process's own operations appear in the global order in the same sequence it issued them. All operations from all processes can be arranged into a single sequence that every process agrees on, even though operations from different processes can be interleaved within it.
// Disqualified
P1 : w(x,1)
P1 : w(x,2)
P2 : r(x) : 2
P2 : r(x) : 1
The above is not sequentially consistent because write(x,1) executes before write(x,2), hence other processes should not read 1 after reading 2
P1 : w(x,1)
P1 : w(x,2)
P2 : w(y,1)
P3 : r(x) : 2
P3 : r(y) : 0 (stale)
P4 : r(x) : 1
P4 : r(y) : 1
The above is allowed since P3 might execute after P1's operations and before P2, while P4 might execute after P1 and P2's operations. This is possible since operations of different processes can be ordered while ensuring operations of the same process are in sequence as issued.
Linearizable Consistency
Linearizable consistency is a commonly used consistency model and considered as the strongest single-object consistency model. It was formalized by Herlihy and Wing in 1990.
Each operation applied by concurrent processes takes effect instantaneously at some point between its invocation and its response.
Linearizability is built on top of sequential consistency by ensuring sequential consistency along with real time precedence.
If an operation X completes before operation Y begins, X must appear before Y in the total order.
Linearizability = Sequential Consistency + real-time ordering
Linearizability is achieved either using consensus protocols like Raft, Paxos where a single leader serializes all operations, or using timestamp based approach which utilizes clocks to synchronize between the nodes.
By utilizing both KV stores, we are able to implement the Kafka workload.
type record struct {
offset int
msg int
}
func (c *config) send(msg maelstrom.Message) error {
body := sendInput{}
if err := json.Unmarshal(msg.Body, &body); err != nil {
return fmt.Errorf("Send: Error while decoding json: %s", err)
}
offset := 0
retry := 0
for retry < 10 {
offset = 0
response := make(map[int]int)
err := c.linKv.ReadInto(context.TODO(), body.Key, &response)
if maelstrom.ErrorCode(err) == maelstrom.KeyDoesNotExist {
response[offset] = body.Msg
if err := c.linKv.CompareAndSwap(context.TODO(), body.Key, response, response, true); err != nil {
retry++
continue
}
} else {
offset = offset + len(response)
newResponse := maps.Clone(response)
newResponse[offset] = body.Msg
if err := c.linKv.CompareAndSwap(context.TODO(), body.Key, response, newResponse, false); err != nil {
retry++
continue
}
}
break
}
if retry >= 10 {
return fmt.Errorf("Send: could not set value")
}
return c.node.Reply(msg, sendOutput{Type: "send_ok", Offset: offset})
}
func (c *config) poll(msg maelstrom.Message) error {
input := pollInput{}
if err := json.Unmarshal(msg.Body, &input); err != nil {
return fmt.Errorf("Poll: Error while decoding json: %s", err)
}
msgs := make(map[string][][2]int)
for key, startOffset := range input.Offsets {
body := make(map[int]int)
if err := c.linKv.ReadInto(context.TODO(), key, &body); maelstrom.ErrorCode(err) == maelstrom.KeyDoesNotExist {
continue
}
for offset, val := range body {
if offset >= startOffset {
msgs[key] = append(msgs[key], [2]int{offset, int(val)})
}
}
}
for _, val := range msgs {
slices.SortFunc(val, func(i, j [2]int) int {
return i[0] - j[0]
})
}
return c.node.Reply(msg, pollOutput{Type: "poll_ok", Msgs: msgs})
}
func (c *config) commitOffsets(msg maelstrom.Message) error {
body := commitOffsetsInput{}
if err := json.Unmarshal(msg.Body, &body); err != nil {
return fmt.Errorf("COMMIT_OFFSETS: Error while decoding json: %s", err)
}
for key, offset := range body.Offsets {
for {
val, err := c.seqKv.ReadInt(context.TODO(), key+"Commit")
if maelstrom.ErrorCode(err) == maelstrom.KeyDoesNotExist {
if err := c.seqKv.CompareAndSwap(context.TODO(), key+"Commit", val, offset, true); err != nil {
continue
}
} else if err != nil {
return fmt.Errorf("commit_offsets: Could not fetch key %s details: %s", key, err)
} else {
offset = max(val, offset)
if err := c.seqKv.CompareAndSwap(context.TODO(), key+"Commit", val, offset, false); err != nil {
continue
}
}
break
}
}
return c.node.Reply(msg, commitOffsetsOutput{Type: "commit_offsets_ok"})
}
func (c *config) listCommitOffsets(msg maelstrom.Message) error {
body := listCommitOffsetsInput{}
if err := json.Unmarshal(msg.Body, &body); err != nil {
return fmt.Errorf("LIST_COMMITED_OFFSETS: Error while decoding json: %s", err)
}
result := make(map[string]int)
for _, key := range body.Keys {
val, err := c.seqKv.ReadInt(context.TODO(), key+"Commit")
if err != nil {
continue
}
result[key] = val
}
return c.node.Reply(msg, listCommitOffsetsOutput{Type: "list_committed_offsets_ok", Offsets: result})
}
I have utilized linearizable KV store for saving value, offset pair. This ensures all operations executed after
X -> write(k1,v1)
will see the value v1. If two nodes n1 and n2 try to write at the same time, the atomicity of compare-and-swap ensures only one write can succeed for a given expected value. Linearizability additionally guarantees that this success/failure is observed consistently by all nodes, with no ambiguity about which write actually took effect. Stronger consistency and high contention for the same key lead to higher CAS failures in a high throughput system which has to be handled by the node, using a simple retry mechanism to handle CAS failures.
Sequential KV store is used to store the commit offsets. On each write,
offset = max(val, offset)
ensures that max offset is saved. Since Kafka follows an at-least-once-delivery, it does not hurt the clients when they have to retry/replay certain offsets.
Totally-Available Transactions
This challenge expects us to execute a set of transactions. Each message can contain multiple operations which must be executed as a single transaction.
{
"type": "txn",
"msg_id": 3,
"txn": [
["r", 1, null],
["w", 1, 6],
["w", 2, 9]
]
}
For each read
r
operation, we need to populate the third index of the array and send. For each write
w
operation, we must persist the value and send the array back to the client.
I have used a simple map to store the key value pair and broadcast the values to other nodes.
Challenge 6a requires a single node which is totally available. This challenge is as simple as writing the data to the store and returning the resulting map.
Challenge 6b requires a multi-node setup which is totally available and guarantees read uncommitted consistency. It prohibits dirty write which is defined as below
- G0 (dirty write): a cycle of transactions linked by write-write dependencies. For instance, transaction T1 appends 1 to key x, transaction T2 appends 2 to x, and T1 appends 3 to x again, producing the value[1, 2, 3].
So for this challenge, we have to ensure that when two transactions execute concurrently, the first transaction must fully complete its state changes before the second transaction can execute.
In case T1 rolls back, there is a chance of T2 reading the dirty write before it's rolled back. While the above code does not differentiate between read and write operations and locks the store until it is unlocked. Read-uncommitted allows T2 to read T1's uncommitted data as well.
Challenge 6c requires a multi-node setup which is totally available and guarantees read committed consistency. There are 3 more anomalies which are prohibited along with G0
- G1a (aborted read): An aborted transaction T1 writes key x, and a committed transaction T2 reads that write to x.
- G1b (intermediate read): A committed transaction T2 reads a value of key x that was generated by transaction T1 other than T1’s final write of x.
- G1c (circular information flow): A cycle of transactions linked by either write-read or read-write dependencies. For instance, transaction T1 writes x = 1 and reads y = 2, and transaction T2 writes y = 2 and reads x = 1.
type config struct {
node *maelstrom.Node
retryHandler retry.RetryHandler
store map[int]int
inputChan chan inputChanMsg
}
func newConfig(node *maelstrom.Node, retryHandler retry.RetryHandler) *config {
inputChan := make(chan inputChanMsg)
config := &config{
node: node,
retryHandler: retryHandler,
store: make(map[int]int),
inputChan: inputChan,
}
go config.processTxn()
return config
}
func Handle(node *maelstrom.Node, retryHandler retry.RetryHandler) {
config := newConfig(node, retryHandler)
node.Handle("txn", config.txn)
node.Handle("txn-update", config.txnUpdate)
}
func (c *config) txn(msg maelstrom.Message) error {
body := input{}
if err := json.Unmarshal(msg.Body, &body); err != nil {
return fmt.Errorf("TXN: Error while decoding json: %s", err)
}
outputChan := make(chan outputChanMsg)
defer close(outputChan)
c.inputChan <- inputChanMsg{input: body, src: msg.Src, ch: outputChan, shouldBroadcast: true}
processedMsg := <-outputChan
if processedMsg.err != nil {
return c.node.Reply(msg, output{Type: "error", Code: maelstrom.TxnConflict, Text: "txn abort"})
}
return c.node.Reply(msg, output{Type: "txn_ok", Txn: processedMsg.txns})
}
func (c *config) txnUpdate(msg maelstrom.Message) error {
body := input{}
if err := json.Unmarshal(msg.Body, &body); err != nil {
return fmt.Errorf("TXN: Error while decoding json: %s", err)
}
outputChan := make(chan outputChanMsg)
defer close(outputChan)
c.inputChan <- inputChanMsg{input: body, src: msg.Src, ch: outputChan, shouldBroadcast: false}
processedMsg := <-outputChan
if processedMsg.err != nil {
return c.node.Reply(msg, output{Type: "error", Code: maelstrom.TxnConflict, Text: "txn abort"})
}
return c.node.Reply(msg, output{Type: "txn_ok", Txn: processedMsg.txns})
}
func (c *config) processTxn() {
for {
msg := <-c.inputChan
pendingWrites := make(map[int]int)
writtenTxns := make([]any, 0)
aborted := false
for i, t := range msg.Txns {
txn := t.([]any)
op := txn[0].(string)
key := int(txn[1].(float64))
if op == "r" {
if v, ok := pendingWrites[key]; ok {
txn[2] = v
} else {
txn[2] = c.store[key]
}
} else {
val := txn[2].(float64)
pendingWrites[key] = int(val)
writeTxn := make([]any, len(txn))
copy(writeTxn, txn)
writtenTxns = append(writtenTxns, writeTxn)
}
msg.Txns[i] = txn
if isAborted(key) {
aborted = true
break
}
}
if aborted == true {
msg.ch <- outputChanMsg{err: fmt.Errorf("Transaction abort")}
continue
}
maps.Copy(c.store, pendingWrites)
if msg.shouldBroadcast && len(writtenTxns) > 0 {
for _, id := range c.node.NodeIDs() {
if id == c.node.ID() || id == msg.src {
continue
}
c.retryHandler.Send(id, "txn-update", writtenTxns...)
}
}
msg.ch <- outputChanMsg{txns: msg.Txns}
}
}
func isAborted(key int) bool {
// Dummy function to simulate transaction rollback
return false
}
Detailed code is available at link.
The above code is written based on Communicating Sequential Processes(CSP).
Communicating Sequential Processes
Communicating Sequential Processes(CSP) is a formal language for concurrency which was introduced by Tony Hoare in 1978. It is built on the idea that concurrent processes interact only by sending messages over channels. There is no shared memory or locks. Go's goroutines and channels are directly inspired by CSP. In CSP, each process is independent sequential computations. They execute their own steps and communicate via channels, and this communication is synchronous.
The above code creates a single goroutine called
processTxn
which handles data mutation. The calling functions
txn
and
txnUpdate
pass the message along with an output channel. This output channel ensures the data is returned to the correct calling function.
processTxn
can process a single txn at a time. By using an
inputChan
and
outputChan
we ensure that each send is blocked until a corresponding output is received. Rollbacks are taken care of by ensuring all writes are written to a temporary map and then copied to the main store. Since retry is asynchronous, this makes the above code totally available even during a network partition. It is to note that while committing, it does not guarantee the txn execution in other nodes immediately due to asynchronous retry logic. But this works since read-committed consistency model is expected.
Conclusion
Overall, this challenge provided me with an opportunity to implement many theories and see them applied to a real-world problem. Maelstrom has even more challenges on different topics which I will solve in the future. I would highly recommend you try this challenge to get your hands dirty with distributed systems.