summaryrefslogtreecommitdiff
path: root/internal
diff options
context:
space:
mode:
Diffstat (limited to 'internal')
-rw-r--r--internal/client/client.go28
-rw-r--r--internal/config/config.go26
-rw-r--r--internal/config/config_test.go40
-rw-r--r--internal/quorum/quorum.go71
-rw-r--r--internal/quorum/quorum_test.go24
-rw-r--r--internal/run.go5
-rw-r--r--internal/server/tcpserver.go4
-rw-r--r--internal/vote/vote.go4
-rw-r--r--internal/vote/vote_test.go4
9 files changed, 88 insertions, 118 deletions
diff --git a/internal/client/client.go b/internal/client/client.go
index 8f9e851..25d99e0 100644
--- a/internal/client/client.go
+++ b/internal/client/client.go
@@ -3,32 +3,16 @@ package client
import (
"context"
"log"
- "time"
"codeberg.org/snonux/gorum/internal/config"
)
-func Start(ctx context.Context, conf config.Config, winnerCh <-chan string) {
- go func() {
- for {
- log.Println("Starting client")
- run(ctx, conf, winnerCh)
-
- select {
- // Wait a second before restarting the clien
- case <-time.After(time.Second):
- case <-ctx.Done():
- return
- }
- }
- }()
-}
-
-func run(ctx context.Context, conf config.Config, winnerCh <-chan string) {
- fanOut := make([]chan string, len(conf.Participants))
+func Start(ctx context.Context, conf config.Config, liveNodesCh <-chan []string) {
+ log.Println("Starting client")
+ fanOut := make([]chan []string, len(conf.Nodes))
for i := 0; i < len(fanOut); i++ {
- fanOut[i] = make(chan string, 1)
+ fanOut[i] = make(chan []string, 1)
}
go func() {
@@ -40,13 +24,13 @@ func run(ctx context.Context, conf config.Config, winnerCh <-chan string) {
for {
select {
- case winner := <-winnerCh:
+ case liveNodes := <-liveNodesCh:
for _, ch := range fanOut {
select {
case <-ch:
default:
}
- ch <- winner
+ ch <- liveNodes
}
case <-ctx.Done():
return
diff --git a/internal/config/config.go b/internal/config/config.go
index 1e2705c..09302b7 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -12,9 +12,9 @@ import (
)
type Config struct {
- StateDir string
- Address string
- Participants []string
+ StateDir string
+ Address string
+ Nodes []string
}
func New(configFile string) (Config, error) {
@@ -39,21 +39,21 @@ func New(configFile string) (Config, error) {
return c, nil
}
-func (c Config) ParticipantNumber(participant string) (int, error) {
- for i, participant_ := range c.Participants {
- if participant == utils.StripPort(participant_) {
+func (c Config) NodeNumber(node string) (int, error) {
+ for i, node_ := range c.Nodes {
+ if node == utils.StripPort(node_) {
return i, nil
}
}
- return 0, fmt.Errorf("participant %s not found", participant)
+ return 0, fmt.Errorf("node %s not found", node)
}
-func (c Config) IsParticipant(remoteAddr string) bool {
+func (c Config) IsNode(remoteAddr string) bool {
remoteAddr = utils.StripPort(remoteAddr)
- for _, participant := range c.Participants {
- if remoteAddr == utils.StripPort(participant) {
+ for _, node := range c.Nodes {
+ if remoteAddr == utils.StripPort(node) {
return true
}
}
@@ -61,11 +61,11 @@ func (c Config) IsParticipant(remoteAddr string) bool {
return false
}
-func (c Config) IsParticipantWithLookup(remoteAddr string, lookupIP func(string) ([]net.IP, error)) bool {
+func (c Config) IsNodeWithLookup(remoteAddr string, lookupIP func(string) ([]net.IP, error)) bool {
remoteAddr = utils.StripPort(remoteAddr)
- for _, participant := range c.Participants {
- ips, err := lookupIP(utils.StripPort(participant))
+ for _, node := range c.Nodes {
+ ips, err := lookupIP(utils.StripPort(node))
if err != nil {
log.Println(err)
continue
diff --git a/internal/config/config_test.go b/internal/config/config_test.go
index bccfb2e..973797a 100644
--- a/internal/config/config_test.go
+++ b/internal/config/config_test.go
@@ -6,47 +6,47 @@ import (
"testing"
)
-func TestParticipantNumber(t *testing.T) {
- conf := Config{Participants: []string{"localhost:1234", "hamburger:4321"}}
+func TestNodeNumber(t *testing.T) {
+ conf := Config{Nodes: []string{"localhost:1234", "hamburger:4321"}}
- num, err := conf.ParticipantNumber("localhost")
+ num, err := conf.NodeNumber("localhost")
if err != nil {
t.Errorf(err.Error())
}
if num != 0 {
- t.Errorf("localhost should be participant number 0 but is %d", num)
+ t.Errorf("localhost should be node number 0 but is %d", num)
}
- num, err = conf.ParticipantNumber("hamburger")
+ num, err = conf.NodeNumber("hamburger")
if err != nil {
t.Errorf(err.Error())
}
if num != 1 {
- t.Errorf("hamburger should be participant number 1 but is %d", num)
+ t.Errorf("hamburger should be node number 1 but is %d", num)
}
- _, err = conf.ParticipantNumber("doener")
+ _, err = conf.NodeNumber("doener")
if err == nil {
- t.Errorf("doener is not a participant")
+ t.Errorf("doener is not a node")
}
}
-func TestIsParticipant(t *testing.T) {
- conf := Config{Participants: []string{"localhost:1234", "hamburger:4321"}}
+func TestIsNode(t *testing.T) {
+ conf := Config{Nodes: []string{"localhost:1234", "hamburger:4321"}}
remoteAddr := "localhost:323232"
- if !conf.IsParticipant(remoteAddr) {
- t.Errorf("%s should be participant of %v", remoteAddr, conf.Participants)
+ if !conf.IsNode(remoteAddr) {
+ t.Errorf("%s should be node of %v", remoteAddr, conf.Nodes)
}
remoteAddr = "foo.zone:2345"
- if conf.IsParticipant(remoteAddr) {
- t.Errorf("%s should not be participant of %v", remoteAddr, conf.Participants)
+ if conf.IsNode(remoteAddr) {
+ t.Errorf("%s should not be node of %v", remoteAddr, conf.Nodes)
}
}
-func TestIsParticipantWithLookup(t *testing.T) {
- conf := Config{Participants: []string{"localhost:1234", "hamburger:4321"}}
+func TestIsNodeWithLookup(t *testing.T) {
+ conf := Config{Nodes: []string{"localhost:1234", "hamburger:4321"}}
lookupIP := func(addr string) ([]net.IP, error) {
switch addr {
@@ -60,12 +60,12 @@ func TestIsParticipantWithLookup(t *testing.T) {
}
remoteAddr := "127.0.0.1:323232"
- if !conf.IsParticipantWithLookup(remoteAddr, lookupIP) {
- t.Errorf("%s should be participant of %v", remoteAddr, conf.Participants)
+ if !conf.IsNodeWithLookup(remoteAddr, lookupIP) {
+ t.Errorf("%s should be node of %v", remoteAddr, conf.Nodes)
}
remoteAddr = "9.9.9.9:2345"
- if conf.IsParticipantWithLookup(remoteAddr, lookupIP) {
- t.Errorf("%s should not be participant of %v", remoteAddr, conf.Participants)
+ if conf.IsNodeWithLookup(remoteAddr, lookupIP) {
+ t.Errorf("%s should not be node of %v", remoteAddr, conf.Nodes)
}
}
diff --git a/internal/quorum/quorum.go b/internal/quorum/quorum.go
index 01c861a..b299460 100644
--- a/internal/quorum/quorum.go
+++ b/internal/quorum/quorum.go
@@ -2,7 +2,6 @@ package quorum
import (
"context"
- "fmt"
"log"
"sort"
"time"
@@ -12,9 +11,9 @@ import (
)
type Quorum struct {
- conf config.Config
- votes map[string]vote.Vote
- inCh chan vote.Vote
+ conf config.Config
+ votes map[string]vote.Vote
+ voteCh chan vote.Vote
}
type Score struct {
@@ -24,54 +23,37 @@ type Score struct {
func New(conf config.Config) Quorum {
return Quorum{
- conf: conf,
- votes: make(map[string]vote.Vote),
- inCh: make(chan vote.Vote),
+ conf: conf,
+ votes: make(map[string]vote.Vote),
+ voteCh: make(chan vote.Vote),
}
}
-func (quo Quorum) Start(ctx context.Context) chan string {
- ch := make(chan string, 1)
+func (quo Quorum) Start(ctx context.Context) <-chan []string {
+ liveNodesCh := make(chan []string)
go func() {
- defer close(ch)
- var lastWinner string
+ defer close(liveNodesCh)
for {
select {
case <-time.After(vote.Expiry):
- quo.deleteExpiredVotes()
- case vote := <-quo.inCh:
+ liveNodesCh <- quo.deleteExpiredVotes()
+ case vote := <-quo.voteCh:
quo.vote(vote)
- winner, err := quo.winner()
- if err != nil {
- log.Println(err)
- continue
- }
- if lastWinner == winner {
- continue
- }
- lastWinner = winner
-
- // Remove current channel entry, and update with new winner
- select {
- case <-ch:
- default:
- }
- ch <- winner
-
+ liveNodesCh <- quo.deleteExpiredVotes()
case <-ctx.Done():
return
}
}
}()
- return ch
+ return liveNodesCh
}
func (quo Quorum) Vote(v vote.Vote) {
log.Printf("Queing vote %v", v)
- quo.inCh <- v
+ quo.voteCh <- v
}
func (quo Quorum) vote(v vote.Vote) {
@@ -79,14 +61,6 @@ func (quo Quorum) vote(v vote.Vote) {
quo.votes[v.From] = v
}
-func (quo Quorum) winner() (string, error) {
- scores := quo.score()
- if len(scores) == 0 {
- return "", fmt.Errorf("unable to find a winner, empty score list")
- }
- return scores[0].ID, nil
-}
-
func (quo Quorum) score() (scores []Score) {
scoreMap := make(map[string]int)
@@ -109,25 +83,30 @@ func (quo Quorum) score() (scores []Score) {
return scores[i].Value > scores[j].Value
}
- // Score tie, use participant number.
- i_, _ := quo.conf.ParticipantNumber(scores[i].ID)
- j_, _ := quo.conf.ParticipantNumber(scores[j].ID)
+ // Score tie, use node number.
+ i_, _ := quo.conf.NodeNumber(scores[i].ID)
+ j_, _ := quo.conf.NodeNumber(scores[j].ID)
return i_ < j_
})
+ log.Println("Scores are ", scores)
return
}
-func (quo Quorum) deleteExpiredVotes() {
+func (quo Quorum) deleteExpiredVotes() (liveNodes []string) {
var expired []string
- for from, vote := range quo.votes {
+ for fromNode, vote := range quo.votes {
if vote.Expired() {
- expired = append(expired, from)
+ expired = append(expired, fromNode)
+ continue
}
+ liveNodes = append(liveNodes, fromNode)
}
for _, e := range expired {
delete(quo.votes, e)
}
+
+ return
}
diff --git a/internal/quorum/quorum_test.go b/internal/quorum/quorum_test.go
index d0f7ec7..9953c11 100644
--- a/internal/quorum/quorum_test.go
+++ b/internal/quorum/quorum_test.go
@@ -9,7 +9,7 @@ import (
)
func TestScore(t *testing.T) {
- conf := config.Config{Participants: []string{"foo:1234", "bar:4321", "baz:3444"}}
+ conf := config.Config{Nodes: []string{"foo:1234", "bar:4321", "baz:3444"}}
quo := New(conf)
vote1 := vote.New(conf, "foo:334234", "foo bar\n")
@@ -55,7 +55,7 @@ func TestTieScore(t *testing.T) {
t.Run("First tie score test", func(t *testing.T) {
// If it is a tie, the first particpant (here: "foo") will win.
- conf := config.Config{Participants: []string{"foo:1234", "bar:4321", "baz:3444"}}
+ conf := config.Config{Nodes: []string{"foo:1234", "bar:4321", "baz:3444"}}
quo := New(conf)
addVotes(conf, quo)
@@ -68,7 +68,7 @@ func TestTieScore(t *testing.T) {
t.Errorf("Expected score[0] to be {foo,3}: %v", scores[0])
}
- winner, _ := quo.winner()
+ winner := scores[0].ID
if winner != "foo" {
t.Errorf("Expected the winner to be foo but is: %s", winner)
}
@@ -76,7 +76,7 @@ func TestTieScore(t *testing.T) {
t.Run("Second tie score test", func(t *testing.T) {
// If it is a tie, the first particpant (here: "bar") will win.
- conf := config.Config{Participants: []string{"bar:1234", "foo:4321", "baz:3444"}}
+ conf := config.Config{Nodes: []string{"bar:1234", "foo:4321", "baz:3444"}}
quo := New(conf)
addVotes(conf, quo)
@@ -89,7 +89,7 @@ func TestTieScore(t *testing.T) {
t.Errorf("Expected score[0] to be {bar,3}: %v", scores[0])
}
- winner, _ := quo.winner()
+ winner := scores[0].ID
if winner != "bar" {
t.Errorf("Expected the winner to be bar but is: %s", winner)
}
@@ -97,7 +97,7 @@ func TestTieScore(t *testing.T) {
}
func TestExpire(t *testing.T) {
- conf := config.Config{Participants: []string{"foo:1234", "bay:4321"}}
+ conf := config.Config{Nodes: []string{"foo:1234", "bay:4321"}}
quo := New(conf)
vote1 := vote.New(conf, "foo:334234", " foo bar baz bay\n")
@@ -112,8 +112,16 @@ func TestExpire(t *testing.T) {
t.Errorf("Expected to have two votes before expiry: %v", quo)
}
- quo.deleteExpiredVotes()
+ liveNodes := quo.deleteExpiredVotes()
if len(quo.votes) != 1 {
- t.Errorf("Expected to have only one vote after expiry: %v", quo)
+ t.Errorf("Expected to have one vote after expiry: %v", quo)
+ }
+
+ if len(liveNodes) != 1 {
+ t.Errorf("Expected to have one live node after expiry: %v", quo)
+ }
+
+ if liveNodes[0] != "foo" {
+ t.Errorf("Expected 'foo' to be the live node, but got : %v", liveNodes[0])
}
}
diff --git a/internal/run.go b/internal/run.go
index 08fe8dc..9b9df1d 100644
--- a/internal/run.go
+++ b/internal/run.go
@@ -16,10 +16,9 @@ func Run(ctx context.Context, configFile string) error {
}
quo := quorum.New(conf)
- winnerCh := quo.Start(ctx)
-
+ liveNodesCh := quo.Start(ctx)
server.Start(ctx, conf, quo)
- client.Start(ctx, conf, winnerCh)
+ client.Start(ctx, conf, liveNodesCh)
return nil
}
diff --git a/internal/server/tcpserver.go b/internal/server/tcpserver.go
index 4cb5b02..9ebbb60 100644
--- a/internal/server/tcpserver.go
+++ b/internal/server/tcpserver.go
@@ -29,8 +29,8 @@ func tcpServerRun(ctx context.Context, conf config.Config,
continue
}
- if !conf.IsParticipantWithLookup(conn.RemoteAddr().String(), net.LookupIP) {
- log.Printf("Denying connection, peer not a participant: %v\n", conn.RemoteAddr().String())
+ if !conf.IsNodeWithLookup(conn.RemoteAddr().String(), net.LookupIP) {
+ log.Printf("Denying connection, peer not a node: %v\n", conn.RemoteAddr().String())
conn.Close()
continue
}
diff --git a/internal/vote/vote.go b/internal/vote/vote.go
index 079e021..75fa67a 100644
--- a/internal/vote/vote.go
+++ b/internal/vote/vote.go
@@ -20,8 +20,8 @@ type Vote struct {
func New(conf config.Config, from, message string) Vote {
var ids []string
for _, id := range strings.Split(strings.TrimSpace(message), " ") {
- if !conf.IsParticipant(id) {
- log.Printf("%s is not a participant, excluding from the vote", id)
+ if !conf.IsNode(id) {
+ log.Printf("%s is not a node, excluding from the vote", id)
continue
}
ids = append(ids, id)
diff --git a/internal/vote/vote_test.go b/internal/vote/vote_test.go
index 0986e00..af8b741 100644
--- a/internal/vote/vote_test.go
+++ b/internal/vote/vote_test.go
@@ -8,7 +8,7 @@ import (
)
func TestVote(t *testing.T) {
- conf := config.Config{Participants: []string{"foo:1234", "bay:4321"}}
+ conf := config.Config{Nodes: []string{"foo:1234", "bay:4321"}}
v := New(conf, "earth:334234", " foo bar baz bay\n")
if v.From != "earth" {
@@ -29,7 +29,7 @@ func TestVote(t *testing.T) {
}
func TestVoteExpiry(t *testing.T) {
- conf := config.Config{Participants: []string{"foo:1234", "bay:4321"}}
+ conf := config.Config{Nodes: []string{"foo:1234", "bay:4321"}}
v := New(conf, "earth:334234", " foo bar baz bay\n")
// Set expiry 1h into the future
generated by cgit v1.2.3 (git 2.25.1) at 2026-08-19 08:19:15 +0000