Distributed AFS

An AFS-like distributed file system with Raft-based replication, whole-file caching, and leader-failover. See the slides here and PDFs at docs/AFS_STAGE1.pdf & docs/AFS_Complete.pdf.

AFS Demo

afs10.mov

afs11.mov

Distributed Prime-finding Application Demo

coordinator-worker.mov

servers.mov

Client API → Server RPCs

AFS File Operations

Client API	RPC Sent	Server Handler	Description
`Open(path)`	`TestAuth`	`handleTestAuthRPC`	Cache hit: validate via SHA-256 content hash
`Open(path)`	`FetchFile`	`handleFetchFileRPC`	Cache miss or stale: download file + hash
`Create(path)`	`Create`	`handleCreateRPC`	Create new file (`O_EXCL`, fails if exists)
`Read(buf, offset)`	—	—	Local read from cached file
`ReadAll()`	—	—	Read entire cached file into memory
`Write(data, offset)`	—	—	Local write to cached file
`Append(data)`	—	—	Append data to end of cached file
`Close()`	`StoreFile`	`handleStoreFileRPC`	Flush dirty file to server (whole-file upload)
`DirectoryInfo()`	`ListDir`	`handleListDirRPC`	List input + output directory contents

Raft Internal RPCs

RPC	Server Handler	Description
`RequestVote`	`handleRequestVoteRPC`	Leader election vote request
`AppendEntries`	`handleAppendEntriesRPC`	Log replication + heartbeat
`SyncFiles`	`handleSyncFilesRPC`	Full file sync for new/recovering followers

Repository Layout

distributed-afs/
├── Makefile
├── go.mod
├── cmd/
│   ├── client/main.go            CLI client entrypoint
│   ├── server/main.go            server entrypoint
│   ├── coordinator/main.go       Stage 2: coordinator entrypoint
│   ├── worker/main.go            Stage 2: worker entrypoint
│   └── test/main.go              single-process baseline prime finder
├── internal/
│   ├── client/
│   │   ├── cache.go              cache entry store
│   │   └── client.go             AFS client API (Open/Create/Read/Write/Append/Close/ReadAll)
│   ├── prime/
│   │   ├── types.go              PrimeTask, CompletedTask, RPC payloads
│   │   ├── primality.go          deterministic Miller-Rabin for uint64
│   │   ├── primality_test.go     unit tests for primality
│   │   ├── coordinator.go        task partitioner, lease manager, shard recovery, merge
│   │   └── worker.go             task executor, heartbeat, shard writer
│   ├── replication/
│   │   ├── raft.go               leader election + log replication
│   │   └── raft_test.go
│   ├── rpcfs/
│   │   ├── client.go             TCP RPC client (dial, send, receive)
│   │   ├── codec.go              len + crc32 + gob framing
│   │   ├── server.go             TCP RPC server (accept, dispatch)
│   │   └── types.go              RPC methods, status codes, request/response types
│   └── server/
│       ├── dedup.go              at-most-once dedup table (client_id, seq_no)
│       ├── server.go             AFS + Raft RPC handlers
│       └── storage.go            file I/O, path resolution, SHA-256 hashing
├── scripts/
│   └── gen_data.py               generates input datasets for all servers
├── tests/
│   ├── crash_recovery_integration_test.go
│   ├── system_integration_test.go
│   ├── prime_integration_test.go Stage 2 coordinator-worker tests
│   └── manual_tests.txt          manual test plan
└── docs/
    ├── DESIGN.md                 full design document
    ├── RPC_IMPL.md               RPC implementation details
    └── STAGE1_IMPL.md            Stage 1 implementation summary

Quick Start

1. Build

make build

2. Generate Data

make clean-data   # wipe old data dirs
make gen           # runs scripts/gen_data.py

This creates data/s{1,2,3}/input_dir/ each with 5 input files (input_dataset_001.txt .. 005.txt, 100k numbers each).

3. Start a 3-Node Cluster (3 terminals)

make server1   # server-1 on :50051
make server2   # server-2 on :50052
make server3   # server-3 on :50053

4. Client Commands

SERVERS=127.0.0.1:50051,127.0.0.1:50052,127.0.0.1:50053

# interactive mode (no -cmd flag) - this will list the server directories and prompt for commands
./bin/client -servers $SERVERS

# read an input file
./bin/client -servers $SERVERS -cmd cat -remote input_dir/input_dataset_001.txt

# download to local path
./bin/client -servers $SERVERS -cmd get -remote input_dir/input_dataset_001.txt -local ./data/tmp/afs/copy.txt

# upload a local file
./bin/client -servers $SERVERS -cmd put -remote output_dir/primes.txt -local ./data/tmp/afs/primes.txt

Data Directories

Each server has two directories:

Directory	Purpose	Path resolution
`input-dir`	Pre-loaded read-only datasets	Reads check here first
`output-dir`	Created/written files (replicated)	Creates and stores go here; reads fall back here

Default layout after make gen:

data/
  s1/input_dir/   s1/output_dir/
  s2/input_dir/   s2/output_dir/
  s3/input_dir/   s3/output_dir/

Client cache lives in ./data/tmp/afs/ by default (override with -cache-dir).

Tests

make test              # all tests
make test-unit         # internal package tests
make test-integration  # multi-node integration tests

Manual test plan: tests/manual_tests.txt

Server Flags

Flag	Default	Description
`--id`	`server-1`	Server identifier
`--host`	`127.0.0.1`	Bind host
`--port`	`50051`	Bind port
`--input-dir`	—	Input dataset directory
`--output-dir`	—	Output/result directory
`--peers`	—	Comma-separated peer addresses

Client Flags

Flag	Default	Description
`-servers`	—	Comma-separated server addresses
`-cache-dir`	`./data/tmp/afs`	Local cache directory
`-client-id`	(auto)	Explicit client identity
`-cmd`	—	`get`, `put`, or `cat`
`-remote`	—	Remote file path
`-local`	—	Local file path

Stage 2 — Prime Finding (Coordinator-Worker)

Stage 2 runs a distributed prime-finding application on top of the AFS cluster. A coordinator partitions input files into line-range tasks, and workers pull tasks, test each number for primality, and report results back.

1. Prepare input data & start the AFS cluster

make clean-data && make gen   # generate input datasets
make server1                  # terminal 1
make server2                  # terminal 2
make server3                  # terminal 3

2. Start the coordinator (terminal 4)

make coordinator

The coordinator discovers all input_dataset_*.txt files via AFS, splits them into chunks (--chunk-lines, default 1000), and waits for workers.

3. Start workers (terminals 5, 6, …)

make worker N=1   # worker-1 on :9100
make worker N=2   # worker-2 on :9101
make worker N=3   # worker-3 on :9102
... # add more workers as needed
# or use `make worker N=<id> PORT=<port>` to customize

Each worker registers with the coordinator (receiving AFS server addresses), pulls tasks, tests numbers for primality (deterministic Miller-Rabin), and writes results to per-worker shard files in DFS. When all tasks are done the coordinator reads shards and writes output_dir/primes.txt (sorted, deduplicated).

4. Read the result

./bin/client -servers 127.0.0.1:50051,127.0.0.1:50052,127.0.0.1:50053 \
  -cmd cat -remote output_dir/primes.txt

5. Clean up Stage 2 output

make clean-out   # removes primes.txt, shards, snapshots, and caches

Coordinator Flags

Flag	Default	Description
`-addr`	`127.0.0.1:9000`	Coordinator bind address
`-servers`	`127.0.0.1:50051`	Comma-separated AFS servers
`-cache-dir`	`./data/tmp/coordinator`	AFS cache directory
`-client-id`	`coordinator`	AFS client ID
`-chunk-lines`	`50000`	Lines per task chunk

Worker Flags

Flag	Default	Description
`-id`	`worker-1`	Worker identifier
`-addr`	`127.0.0.1:9100`	Worker bind address
`-coordinator`	`127.0.0.1:9000`	Coordinator address
`-cache-dir`	`./data/tmp/{id}`	AFS cache directory (auto-derived)

Workers obtain AFS server addresses from the coordinator during registration — no --servers flag needed.

Programmatic API

All components can be used as Go libraries — no CLI required.

RPC Transport (`internal/rpcfs`)

import "github.com/guntas-13/afs-distributed/internal/rpcfs"

// ── Server ──
srv := rpcfs.NewTCPServer("127.0.0.1:9000")
srv.Register("MyMethod", func(req rpcfs.RequestEnvelope) rpcfs.ResponseEnvelope {
    // handle request ...
    return rpcfs.ResponseEnvelope{Code: rpcfs.StatusOK, Body: payload}
})

stopCh := make(chan struct{})
go srv.ListenAndServe(stopCh) // or bind yourself: net.Listen + srv.Serve(ln, stopCh)
// ...
close(stopCh) // graceful shutdown

// ── Client ──
client := rpcfs.NewTCPClient(30 * time.Second)
resp, err := client.Call("127.0.0.1:9000", rpcfs.RequestEnvelope{
    Method: "MyMethod",
    Body:   payload,
})

AFS Server (`internal/server`)

import (
    "github.com/guntas-13/afs-distributed/internal/server"
    "github.com/guntas-13/afs-distributed/internal/rpcfs"
)

// Create a server node with its storage directories and Raft peers.
svc, _ := server.New(
    "server-1",            // id
    "127.0.0.1:50051",     // bind address
    "./data/s1/input_dir", // input directory (read-only datasets)
    "./data/s1/output_dir",// output directory (replicated writes)
    []string{"127.0.0.1:50052", "127.0.0.1:50053"}, // Raft peers
)

// Wire RPC handlers and start serving.
rpcSrv := rpcfs.NewTCPServer("127.0.0.1:50051")
svc.RegisterHandlers(rpcSrv)

stopCh := make(chan struct{})
go rpcSrv.ListenAndServe(stopCh)
defer func() { svc.Stop(); close(stopCh) }()

AFS Client (`internal/client`)

import "github.com/guntas-13/afs-distributed/internal/client"

servers := []string{"127.0.0.1:50051", "127.0.0.1:50052", "127.0.0.1:50053"}
afs, _ := client.NewAFSClient(servers, "./cache", "my-client")

// List files on the cluster.
dir := afs.DirectoryInfo()
fmt.Println(dir.InputFiles, dir.OutputFiles)

// Read a file (whole-file cached locally, validated via SHA-256).
fh, _ := afs.Open("input_dir/input_dataset_001.txt")
buf := make([]byte, 4096)
n, _ := fh.Read(buf, 0)   // read from offset 0
fmt.Println(string(buf[:n]))
fh.Close()                 // no-op for clean files (not dirty)

// Create and write a new file, then flush to the cluster.
fh, _ = afs.Create("output_dir/result.txt")
fh.Write([]byte("hello world\n"), 0)
fh.Close() // flushes via StoreFile RPC → replicated to quorum

Prime Coordinator (`internal/prime`)

import (
    "github.com/guntas-13/afs-distributed/internal/client"
    "github.com/guntas-13/afs-distributed/internal/prime"
    "github.com/guntas-13/afs-distributed/internal/rpcfs"
)

afs, _ := client.NewAFSClient(servers, "./cache/coord", "coordinator")

coord := prime.NewCoordinator(
    afs,
    "127.0.0.1:9000", // coordinator bind address
    1000,              // chunk size (lines per task)
)

rpcSrv := rpcfs.NewTCPServer("127.0.0.1:9000")
coord.RegisterHandlers(rpcSrv)

stopCh := make(chan struct{})
go rpcSrv.ListenAndServe(stopCh)

// Run blocks until all tasks are done and workers have drained.
// It discovers input files, partitions them into tasks, waits for
// workers to pull/execute/report, then writes output/primes.txt.
err := coord.Run()

coord.Stop()
close(stopCh)

Prime Worker (`internal/prime`)

w := prime.NewWorker(
    "worker-1",        // id
    "127.0.0.1:9100",  // worker bind address
    "127.0.0.1:9000",  // coordinator address
    "./cache/w1",      // AFS cache directory
)

// Run blocks: registers with coordinator (receives AFS server addresses),
// creates AFS client, pulls tasks, tests primality, reports results.
// Exits when coordinator signals AllDone.
err := w.Run()

w.Stop()

Fault Tolerance

Retries with exponential backoff + jitter
Leader redirect: non-leader returns leader hint, client retries automatically
Dead-leader avoidance: client tracks servers that fail at the TCP level (deadAddrs); redirects to a dead leader are skipped and the next live server is tried instead, preventing redirect loops after a leader crash
Leader address resolution: resolveLeader() maps Raft-internal leader addresses (which may differ from the client's server list, e.g. :50052 vs 10.x.x.x:50052) back to the client's known server entries for accurate dead-leader detection
At-most-once writes: server dedup table keyed by (client_id, seq_no); client IDs include a nanosecond timestamp to avoid collisions across process restarts
Quorum writes: success only after majority replication
Cache validation: SHA-256 content hash via TestAuth RPC on reopen
Follower sync retry: when a recovering follower's syncFromLeader fails (e.g. during a leader transition), it retries up to 3 times with 200 ms delays
Shard verification: before final merge, the coordinator cross-checks in-memory task completions against actual shard files on DFS; missing tasks are re-queued

Cluster size is set by --peers. Quorum = floor(N/2) + 1 (3 nodes → quorum 2, 5 nodes → quorum 3).

On Apple M1 ARM

Navigation Menu

Repository files navigation

Distributed AFS

AFS Demo

Distributed Prime-finding Application Demo

Client API → Server RPCs

AFS File Operations

Raft Internal RPCs

Repository Layout

Quick Start

1. Build

2. Generate Data

3. Start a 3-Node Cluster (3 terminals)

4. Client Commands

Data Directories

Tests

Server Flags

Client Flags

Stage 2 — Prime Finding (Coordinator-Worker)

1. Prepare input data & start the AFS cluster

2. Start the coordinator (terminal 4)

3. Start workers (terminals 5, 6, …)

4. Read the result

5. Clean up Stage 2 output

Coordinator Flags

Worker Flags

Programmatic API

RPC Transport (`internal/rpcfs`)

AFS Server (`internal/server`)

AFS Client (`internal/client`)

Prime Coordinator (`internal/prime`)

Prime Worker (`internal/prime`)

Fault Tolerance

About

Releases

Packages

Contributors

Languages

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
cmd		cmd
data		data
docs		docs
internal		internal
scripts		scripts
tests		tests
.gitignore		.gitignore
LICENSE		LICENSE
Makefile		Makefile
README.md		README.md
go.mod		go.mod

Latest commit

History

Folders and files

Repository files navigation

Distributed AFS

AFS Demo

Distributed Prime-finding Application Demo

Client API → Server RPCs

AFS File Operations

Raft Internal RPCs

Repository Layout

Quick Start

1. Build

2. Generate Data

3. Start a 3-Node Cluster (3 terminals)

4. Client Commands

Data Directories

Tests

Server Flags

Client Flags

Stage 2 — Prime Finding (Coordinator-Worker)

1. Prepare input data & start the AFS cluster

2. Start the coordinator (terminal 4)

3. Start workers (terminals 5, 6, …)

4. Read the result

5. Clean up Stage 2 output

Coordinator Flags

Worker Flags

Programmatic API

RPC Transport (internal/rpcfs)

AFS Server (internal/server)

AFS Client (internal/client)

Prime Coordinator (internal/prime)

Prime Worker (internal/prime)

Fault Tolerance

About

Topics

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages

RPC Transport (`internal/rpcfs`)

AFS Server (`internal/server`)

AFS Client (`internal/client`)

Prime Coordinator (`internal/prime`)

Prime Worker (`internal/prime`)