diff --git a/.golangci.yml b/.golangci.yml index 5ed2951..ddfaa3a 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -34,5 +34,3 @@ issues: run: timeout: 5m - skip-dirs: - - vendor diff --git a/pkg/lib/algorithm/full_sync/sync.go b/pkg/lib/algorithm/full_sync/sync.go index 91ac59c..e6e9df0 100644 --- a/pkg/lib/algorithm/full_sync/sync.go +++ b/pkg/lib/algorithm/full_sync/sync.go @@ -2,6 +2,7 @@ package full_sync import ( "fmt" + "github.com/String-Reconciliation-Ditributed-System/RCDS_GO/pkg/lib/genSync" "github.com/String-Reconciliation-Ditributed-System/RCDS_GO/pkg/set" "github.com/String-Reconciliation-Ditributed-System/RCDS_GO/pkg/util" @@ -120,7 +121,9 @@ func (f *fullSync) SyncClient(ip string, port int) error { return err } f.additionals.InsertKey(d) - f.AddElement(d) + if err = f.AddElement(d); err != nil { + return err + } } return nil } @@ -181,7 +184,9 @@ func (f *fullSync) SyncServer(ip string, port int) error { if !f.FreezeLocal { for elem := range *tempSet.Difference(f.Set) { f.additionals.InsertKey(elem) - f.AddElement(elem) + if err = f.AddElement(elem); err != nil { + return err + } } } else { logrus.Info("Server is freezing local set and skipping set update.") diff --git a/pkg/lib/algorithm/full_sync/sync_test.go b/pkg/lib/algorithm/full_sync/sync_test.go index 2bacb96..bca69f2 100644 --- a/pkg/lib/algorithm/full_sync/sync_test.go +++ b/pkg/lib/algorithm/full_sync/sync_test.go @@ -71,11 +71,11 @@ func TestNewFullSetSync(t *testing.T) { var wg sync.WaitGroup wg.Add(1) go func() { - err := client.SyncServer("", 8080) - assert.NoError(t, err) + syncErr := client.SyncServer("", 9080) + assert.NoError(t, syncErr) wg.Done() }() - err = server.SyncClient("", 8080) + err = server.SyncClient("", 9080) assert.NoError(t, err) wg.Wait() diff --git a/pkg/lib/algorithm/hash_function_test.go b/pkg/lib/algorithm/hash_function_test.go index d4c444f..3e1b1fc 100644 --- a/pkg/lib/algorithm/hash_function_test.go +++ b/pkg/lib/algorithm/hash_function_test.go @@ -2,8 +2,9 @@ package algorithm import ( "crypto" - "github.com/stretchr/testify/assert" "testing" + + "github.com/stretchr/testify/assert" ) func TestHashBytesWithCryptoFunc(t *testing.T) { diff --git a/pkg/lib/algorithm/iblt/sync.go b/pkg/lib/algorithm/iblt/sync.go index 89bf03a..7264135 100644 --- a/pkg/lib/algorithm/iblt/sync.go +++ b/pkg/lib/algorithm/iblt/sync.go @@ -65,7 +65,9 @@ func (i *ibltSync) AddElement(elem interface{}) error { i.Set.Insert(key, elem) for j := range i.resyncIBLTs { - i.resyncIBLTs[j].Insert(key) + if err := i.resyncIBLTs[j].Insert(key); err != nil { + return err + } } return i.Table.Insert(key) } else { @@ -73,7 +75,9 @@ func (i *ibltSync) AddElement(elem interface{}) error { } key := elem.([]byte) for j := range i.resyncIBLTs { - i.resyncIBLTs[j].Insert(key) + if err := i.resyncIBLTs[j].Insert(key); err != nil { + return err + } } return i.Table.Insert(key) } @@ -86,14 +90,18 @@ func (i *ibltSync) DeleteElement(elem interface{}) error { } i.Set.Remove(key) for j := range i.resyncIBLTs { - i.resyncIBLTs[j].Delete(key) + if err := i.resyncIBLTs[j].Delete(key); err != nil { + return err + } } return i.Table.Delete(key) } i.Set.Remove(elem) key := elem.([]byte) for j := range i.resyncIBLTs { - i.resyncIBLTs[j].Delete(key) + if err := i.resyncIBLTs[j].Delete(key); err != nil { + return err + } } return i.Table.Delete(key) } diff --git a/pkg/lib/algorithm/rcds/backtracking.go b/pkg/lib/algorithm/rcds/backtracking.go index e9de4e1..bbe1186 100644 --- a/pkg/lib/algorithm/rcds/backtracking.go +++ b/pkg/lib/algorithm/rcds/backtracking.go @@ -76,7 +76,12 @@ func (s *hashShingleSet) interactiveBacktracking(previousEdge, currentEdge uint6 i = i - 1 } previousEdge = currentEdge - for tail, count := range tails { + + // Sort tail keys to ensure deterministic iteration order + tailKeys := getSortedTailKeys(tails) + + for _, tail := range tailKeys { + count := tails[tail] currentEdge = tail // get the changed shingles from last layer. tailChangeHistory[i] = tailChangeHistory[i-1] @@ -152,7 +157,12 @@ func (s *hashShingleSet) reverseInteractiveBacktracking(hashArray []uint64) (*Cy i = i - 1 } previousEdge = currentEdge - for tail, count := range tails { + + // Sort tail keys to ensure deterministic iteration order + tailKeys := getSortedTailKeys(tails) + + for _, tail := range tailKeys { + count := tails[tail] currentEdge = tail // get the changed shingles from last layer. tailChangeHistory[i] = tailChangeHistory[i-1] @@ -192,6 +202,16 @@ func (s *hashShingleSet) getTailEdges(firstEdge uint64) shingleTailCount { return nil } +// getSortedTailKeys returns tail keys sorted in ascending order for deterministic iteration +func getSortedTailKeys(tails shingleTailCount) []uint64 { + tailKeys := make([]uint64, 0, len(tails)) + for tail := range tails { + tailKeys = append(tailKeys, tail) + } + sort.Slice(tailKeys, func(i, j int) bool { return tailKeys[i] < tailKeys[j] }) + return tailKeys +} + // sort sorts the hash shingle set and its tail counts. func (s *hashShingleSet) sort() error { size := s.Size() diff --git a/pkg/lib/algorithm/rcds/hashShingling.go b/pkg/lib/algorithm/rcds/hashShingling.go index 8af4753..2e9815d 100644 --- a/pkg/lib/algorithm/rcds/hashShingling.go +++ b/pkg/lib/algorithm/rcds/hashShingling.go @@ -95,7 +95,9 @@ func (s *hashShingleSet) addChunksToShingleSet(chunks *[]string) (*algorithm.Dic if err != nil { return nil, err } - s.AddShingle(0, hash, 1) + if err = s.AddShingle(0, hash, 1); err != nil { + return nil, err + } for i := 1; i < len(*chunks); i++ { first, err := dict.AddToDict((*chunks)[i-1]) @@ -107,7 +109,9 @@ func (s *hashShingleSet) addChunksToShingleSet(chunks *[]string) (*algorithm.Dic return nil, err } if _, err = s.getShingleCount(first, second); err != nil { - s.AddShingle(first, second, 1) + if err = s.AddShingle(first, second, 1); err != nil { + return nil, err + } } else { if _, err = s.addShingleCount(first, second, 1); err != nil { return nil, err diff --git a/pkg/lib/genSync/connection.go b/pkg/lib/genSync/connection.go index 6e47b2b..2f586dd 100644 --- a/pkg/lib/genSync/connection.go +++ b/pkg/lib/genSync/connection.go @@ -1,13 +1,17 @@ package genSync import ( + "context" "fmt" - "github.com/String-Reconciliation-Ditributed-System/RCDS_GO/pkg/util" - "github.com/sirupsen/logrus" - "k8s.io/client-go/util/retry" "net" "strconv" "strings" + "syscall" + + "github.com/sirupsen/logrus" + "k8s.io/client-go/util/retry" + + "github.com/String-Reconciliation-Ditributed-System/RCDS_GO/pkg/util" ) type Connection interface { @@ -80,12 +84,29 @@ func (s *socketConnection) Send(data []byte) (int, error) { func (s *socketConnection) Listen() error { var err error - s.listener, err = net.ListenTCP("tcp", s.tcpAddress) - logrus.Infof("listening on: %v", s.tcpAddress) + + // Create ListenConfig with SO_REUSEADDR to allow quick port reuse + lc := &net.ListenConfig{ + Control: func(_, _ string, c syscall.RawConn) error { + var sockOptErr error + if controlErr := c.Control(func(fd uintptr) { + // Set SO_REUSEADDR to allow quick port reuse + sockOptErr = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_REUSEADDR, 1) + }); controlErr != nil { + return controlErr + } + return sockOptErr + }, + } + + listener, err := lc.Listen(context.Background(), "tcp", s.tcpAddress.String()) if err != nil { return fmt.Errorf("failed to listen: %v", err) } + s.listener = listener.(*net.TCPListener) + logrus.Infof("listening on: %v", s.tcpAddress) + s.connection, err = s.listener.AcceptTCP() return err } @@ -201,10 +222,15 @@ func (s *socketConnection) ReceiveSyncStatus() (uint8, error) { } func (s *socketConnection) Close() error { - if err := s.listener.Close(); err != nil { - logrus.Debugf("failed to close listener, %v", err) + if s.listener != nil { + if err := s.listener.Close(); err != nil { + logrus.Debugf("failed to close listener, %v", err) + } + } + if s.connection != nil { + return s.connection.Close() } - return s.connection.Close() + return nil } func (s *socketConnection) GetIp() string { diff --git a/pkg/lib/genSync/conversion_test.go b/pkg/lib/genSync/conversion_test.go index d4c249c..a559c28 100644 --- a/pkg/lib/genSync/conversion_test.go +++ b/pkg/lib/genSync/conversion_test.go @@ -1,8 +1,8 @@ package genSync import ( + "crypto/rand" "math" - "math/rand" "testing" "github.com/stretchr/testify/assert" @@ -43,7 +43,8 @@ func TestConversionBetweenBytesAndBigInt(t *testing.T) { make([]byte, 512), make([]byte, 1024), } { - rand.Read(test) + _, err := rand.Read(test) + assert.NoError(t, err) bytes, err := ToBigInt(test) assert.NoError(t, err) assert.Equal(t, test, bytes.ToBytes()) diff --git a/pkg/set/set.go b/pkg/set/set.go index fc413ef..0ec4a7d 100644 --- a/pkg/set/set.go +++ b/pkg/set/set.go @@ -2,8 +2,9 @@ package set import ( "fmt" - "github.com/String-Reconciliation-Ditributed-System/RCDS_GO/pkg/lib/algorithm" "reflect" + + "github.com/String-Reconciliation-Ditributed-System/RCDS_GO/pkg/lib/algorithm" ) type Set map[interface{}]interface{} diff --git a/pkg/set/set_test.go b/pkg/set/set_test.go index 5eb52b2..de984c6 100644 --- a/pkg/set/set_test.go +++ b/pkg/set/set_test.go @@ -1,11 +1,12 @@ package set import ( - "k8s.io/apimachinery/pkg/util/rand" "testing" + + "k8s.io/apimachinery/pkg/util/rand" ) -func TestSet_Insert(t *testing.T) { +func TestSet_Insert(_ *testing.T) { s := New() s.InsertKey([]byte(rand.String(10))) } diff --git a/pkg/util/auxiliary_test.go b/pkg/util/auxiliary_test.go index 4d6ce56..23470fb 100644 --- a/pkg/util/auxiliary_test.go +++ b/pkg/util/auxiliary_test.go @@ -1,9 +1,10 @@ package util import ( - "github.com/stretchr/testify/assert" "math" "testing" + + "github.com/stretchr/testify/assert" ) func TestBytesAndIntConversion(t *testing.T) {