felixfoertsch/syncthing
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 10 Wiki Activity

all: Remove lib/util package (#9049)

Browse Source 
Grab-bag packages are nasty, this cleans it up a little by splitting it
into topical packages sempahore, netutil, stringutil, structutil.
This commit is contained in:
Jakob Borg
2023-08-21 19:44:33 +02:00
committed by GitHub
parent 40b3b9ad15
commit acd767b30b
36 changed files with 414 additions and 385 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/model/folder.go
+5 -4
View File
@@ -24,10 +24,11 @@ import (
"github.com/syncthing/syncthing/lib/osutil"
"github.com/syncthing/syncthing/lib/protocol"
"github.com/syncthing/syncthing/lib/scanner"
"github.com/syncthing/syncthing/lib/semaphore"
"github.com/syncthing/syncthing/lib/stats"
"github.com/syncthing/syncthing/lib/stringutil"
"github.com/syncthing/syncthing/lib/svcutil"
"github.com/syncthing/syncthing/lib/sync"
"github.com/syncthing/syncthing/lib/util"
"github.com/syncthing/syncthing/lib/versioner"
"github.com/syncthing/syncthing/lib/watchaggregator"
)
@@ -39,7 +40,7 @@ type folder struct {
stateTracker
config.FolderConfiguration
*stats.FolderStatisticsReference
ioLimiter *util.Semaphore
ioLimiter *semaphore.Semaphore
localFlags uint32
@@ -95,7 +96,7 @@ type puller interface {
pull() (bool, error) // true when successful and should not be retried
}
func newFolder(model *model, fset *db.FileSet, ignores *ignore.Matcher, cfg config.FolderConfiguration, evLogger events.Logger, ioLimiter *util.Semaphore, ver versioner.Versioner) folder {
func newFolder(model *model, fset *db.FileSet, ignores *ignore.Matcher, cfg config.FolderConfiguration, evLogger events.Logger, ioLimiter *semaphore.Semaphore, ver versioner.Versioner) folder {
f := folder{
stateTracker: newStateTracker(cfg.ID, evLogger),
FolderConfiguration: cfg,
@@ -426,7 +427,7 @@ func (f *folder) pull() (success bool, err error) {
// Pulling failed, try again later.
delay := f.pullPause + time.Since(startTime)
l.Infof("Folder %v isn't making sync progress - retrying in %v.", f.Description(), util.NiceDurationString(delay))
l.Infof("Folder %v isn't making sync progress - retrying in %v.", f.Description(), stringutil.NiceDurationString(delay))
f.pullFailTimer.Reset(delay)
return false, err
lib/model/folder_recvenc.go
+2 -2
View File
@@ -16,7 +16,7 @@ import (
"github.com/syncthing/syncthing/lib/fs"
"github.com/syncthing/syncthing/lib/ignore"
"github.com/syncthing/syncthing/lib/protocol"
"github.com/syncthing/syncthing/lib/util"
"github.com/syncthing/syncthing/lib/semaphore"
"github.com/syncthing/syncthing/lib/versioner"
)
@@ -28,7 +28,7 @@ type receiveEncryptedFolder struct {
*sendReceiveFolder
}
func newReceiveEncryptedFolder(model *model, fset *db.FileSet, ignores *ignore.Matcher, cfg config.FolderConfiguration, ver versioner.Versioner, evLogger events.Logger, ioLimiter *util.Semaphore) service {
func newReceiveEncryptedFolder(model *model, fset *db.FileSet, ignores *ignore.Matcher, cfg config.FolderConfiguration, ver versioner.Versioner, evLogger events.Logger, ioLimiter *semaphore.Semaphore) service {
f := &receiveEncryptedFolder{newSendReceiveFolder(model, fset, ignores, cfg, ver, evLogger, ioLimiter).(*sendReceiveFolder)}
f.localFlags = protocol.FlagLocalReceiveOnly // gets propagated to the scanner, and set on locally changed files
return f
lib/model/folder_recvonly.go
+2 -2
View File
@@ -15,7 +15,7 @@ import (
"github.com/syncthing/syncthing/lib/events"
"github.com/syncthing/syncthing/lib/ignore"
"github.com/syncthing/syncthing/lib/protocol"
"github.com/syncthing/syncthing/lib/util"
"github.com/syncthing/syncthing/lib/semaphore"
"github.com/syncthing/syncthing/lib/versioner"
)
@@ -57,7 +57,7 @@ type receiveOnlyFolder struct {
*sendReceiveFolder
}
func newReceiveOnlyFolder(model *model, fset *db.FileSet, ignores *ignore.Matcher, cfg config.FolderConfiguration, ver versioner.Versioner, evLogger events.Logger, ioLimiter *util.Semaphore) service {
func newReceiveOnlyFolder(model *model, fset *db.FileSet, ignores *ignore.Matcher, cfg config.FolderConfiguration, ver versioner.Versioner, evLogger events.Logger, ioLimiter *semaphore.Semaphore) service {
sr := newSendReceiveFolder(model, fset, ignores, cfg, ver, evLogger, ioLimiter).(*sendReceiveFolder)
sr.localFlags = protocol.FlagLocalReceiveOnly // gets propagated to the scanner, and set on locally changed files
return &receiveOnlyFolder{sr}
lib/model/folder_sendonly.go
+2 -2
View File
@@ -12,7 +12,7 @@ import (
"github.com/syncthing/syncthing/lib/events"
"github.com/syncthing/syncthing/lib/ignore"
"github.com/syncthing/syncthing/lib/protocol"
"github.com/syncthing/syncthing/lib/util"
"github.com/syncthing/syncthing/lib/semaphore"
"github.com/syncthing/syncthing/lib/versioner"
)
@@ -24,7 +24,7 @@ type sendOnlyFolder struct {
folder
}
func newSendOnlyFolder(model *model, fset *db.FileSet, ignores *ignore.Matcher, cfg config.FolderConfiguration, _ versioner.Versioner, evLogger events.Logger, ioLimiter *util.Semaphore) service {
func newSendOnlyFolder(model *model, fset *db.FileSet, ignores *ignore.Matcher, cfg config.FolderConfiguration, _ versioner.Versioner, evLogger events.Logger, ioLimiter *semaphore.Semaphore) service {
f := &sendOnlyFolder{
folder: newFolder(model, fset, ignores, cfg, evLogger, ioLimiter, nil),
}
lib/model/folder_sendrecv.go
+5 -5
View File
@@ -27,9 +27,9 @@ import (
"github.com/syncthing/syncthing/lib/osutil"
"github.com/syncthing/syncthing/lib/protocol"
"github.com/syncthing/syncthing/lib/scanner"
"github.com/syncthing/syncthing/lib/semaphore"
"github.com/syncthing/syncthing/lib/sha256"
"github.com/syncthing/syncthing/lib/sync"
"github.com/syncthing/syncthing/lib/util"
"github.com/syncthing/syncthing/lib/versioner"
"github.com/syncthing/syncthing/lib/weakhash"
)
@@ -125,17 +125,17 @@ type sendReceiveFolder struct {
queue *jobQueue
blockPullReorderer blockPullReorderer
writeLimiter *util.Semaphore
writeLimiter *semaphore.Semaphore
tempPullErrors map[string]string // pull errors that might be just transient
}
func newSendReceiveFolder(model *model, fset *db.FileSet, ignores *ignore.Matcher, cfg config.FolderConfiguration, ver versioner.Versioner, evLogger events.Logger, ioLimiter *util.Semaphore) service {
func newSendReceiveFolder(model *model, fset *db.FileSet, ignores *ignore.Matcher, cfg config.FolderConfiguration, ver versioner.Versioner, evLogger events.Logger, ioLimiter *semaphore.Semaphore) service {
f := &sendReceiveFolder{
folder: newFolder(model, fset, ignores, cfg, evLogger, ioLimiter, ver),
queue: newJobQueue(),
blockPullReorderer: newBlockPullReorderer(cfg.BlockPullOrder, model.id, cfg.DeviceIDs()),
writeLimiter: util.NewSemaphore(cfg.MaxConcurrentWrites),
writeLimiter: semaphore.New(cfg.MaxConcurrentWrites),
}
f.folder.puller = f
@@ -1492,7 +1492,7 @@ func (*sendReceiveFolder) verifyBuffer(buf []byte, block protocol.BlockInfo) err
}
func (f *sendReceiveFolder) pullerRoutine(snap *db.Snapshot, in <-chan pullBlockState, out chan<- *sharedPullerState) {
requestLimiter := util.NewSemaphore(f.PullerMaxPendingKiB * 1024)
requestLimiter := semaphore.New(f.PullerMaxPendingKiB * 1024)
wg := sync.NewWaitGroup()
for state := range in {
lib/model/model.go
+12 -12
View File
@@ -38,11 +38,11 @@ import (
"github.com/syncthing/syncthing/lib/osutil"
"github.com/syncthing/syncthing/lib/protocol"
"github.com/syncthing/syncthing/lib/scanner"
"github.com/syncthing/syncthing/lib/semaphore"
"github.com/syncthing/syncthing/lib/stats"
"github.com/syncthing/syncthing/lib/svcutil"
"github.com/syncthing/syncthing/lib/sync"
"github.com/syncthing/syncthing/lib/ur/contract"
"github.com/syncthing/syncthing/lib/util"
"github.com/syncthing/syncthing/lib/versioner"
)
@@ -136,10 +136,10 @@ type model struct {
shortID protocol.ShortID
// globalRequestLimiter limits the amount of data in concurrent incoming
// requests
globalRequestLimiter *util.Semaphore
globalRequestLimiter *semaphore.Semaphore
// folderIOLimiter limits the number of concurrent I/O heavy operations,
// such as scans and pulls.
folderIOLimiter *util.Semaphore
folderIOLimiter *semaphore.Semaphore
fatalChan chan error
started chan struct{}
keyGen *protocol.KeyGenerator
@@ -160,7 +160,7 @@ type model struct {
// fields protected by pmut
pmut sync.RWMutex
conn map[protocol.DeviceID]protocol.Connection
connRequestLimiters map[protocol.DeviceID]*util.Semaphore
connRequestLimiters map[protocol.DeviceID]*semaphore.Semaphore
closed map[protocol.DeviceID]chan struct{}
helloMessages map[protocol.DeviceID]protocol.Hello
deviceDownloads map[protocol.DeviceID]*deviceDownloadState
@@ -173,7 +173,7 @@ type model struct {
var _ config.Verifier = &model{}
type folderFactory func(*model, *db.FileSet, *ignore.Matcher, config.FolderConfiguration, versioner.Versioner, events.Logger, *util.Semaphore) service
type folderFactory func(*model, *db.FileSet, *ignore.Matcher, config.FolderConfiguration, versioner.Versioner, events.Logger, *semaphore.Semaphore) service
var folderFactories = make(map[config.FolderType]folderFactory)
@@ -222,8 +222,8 @@ func NewModel(cfg config.Wrapper, id protocol.DeviceID, clientName, clientVersio
finder: db.NewBlockFinder(ldb),
progressEmitter: NewProgressEmitter(cfg, evLogger),
shortID: id.Short(),
globalRequestLimiter: util.NewSemaphore(1024 * cfg.Options().MaxConcurrentIncomingRequestKiB()),
folderIOLimiter: util.NewSemaphore(cfg.Options().MaxFolderConcurrency()),
globalRequestLimiter: semaphore.New(1024 * cfg.Options().MaxConcurrentIncomingRequestKiB()),
folderIOLimiter: semaphore.New(cfg.Options().MaxFolderConcurrency()),
fatalChan: make(chan error),
started: make(chan struct{}),
keyGen: keyGen,
@@ -243,7 +243,7 @@ func NewModel(cfg config.Wrapper, id protocol.DeviceID, clientName, clientVersio
// fields protected by pmut
pmut: sync.NewRWMutex(),
conn: make(map[protocol.DeviceID]protocol.Connection),
connRequestLimiters: make(map[protocol.DeviceID]*util.Semaphore),
connRequestLimiters: make(map[protocol.DeviceID]*semaphore.Semaphore),
closed: make(map[protocol.DeviceID]chan struct{}),
helloMessages: make(map[protocol.DeviceID]protocol.Hello),
deviceDownloads: make(map[protocol.DeviceID]*deviceDownloadState),
@@ -1966,8 +1966,8 @@ func (m *model) Request(conn protocol.Connection, folder, name string, _, size i
// skipping nil limiters, then returns a requestResponse of the given size.
// When the requestResponse is closed the limiters are given back the bytes,
// in reverse order.
func newLimitedRequestResponse(size int, limiters ...*util.Semaphore) *requestResponse {
multi := util.MultiSemaphore(limiters)
func newLimitedRequestResponse(size int, limiters ...*semaphore.Semaphore) *requestResponse {
multi := semaphore.MultiSemaphore(limiters)
multi.Take(size)
res := newRequestResponse(size)
@@ -2261,9 +2261,9 @@ func (m *model) AddConnection(conn protocol.Connection, hello protocol.Hello) {
// 0: default, <0: no limiting
switch {
case device.MaxRequestKiB > 0:
m.connRequestLimiters[deviceID] = util.NewSemaphore(1024 * device.MaxRequestKiB)
m.connRequestLimiters[deviceID] = semaphore.New(1024 * device.MaxRequestKiB)
case device.MaxRequestKiB == 0:
m.connRequestLimiters[deviceID] = util.NewSemaphore(1024 * defaultPullerPendingKiB)
m.connRequestLimiters[deviceID] = semaphore.New(1024 * defaultPullerPendingKiB)
}
m.helloMessages[deviceID] = hello
lib/model/model_test.go
+8 -8
View File
@@ -35,8 +35,8 @@ import (
"github.com/syncthing/syncthing/lib/protocol"
protocolmocks "github.com/syncthing/syncthing/lib/protocol/mocks"
srand "github.com/syncthing/syncthing/lib/rand"
"github.com/syncthing/syncthing/lib/testutils"
"github.com/syncthing/syncthing/lib/util"
"github.com/syncthing/syncthing/lib/semaphore"
"github.com/syncthing/syncthing/lib/testutil"
"github.com/syncthing/syncthing/lib/versioner"
)
@@ -2968,10 +2968,10 @@ func TestConnCloseOnRestart(t *testing.T) {
m := setupModel(t, w)
defer cleanupModelAndRemoveDir(m, fcfg.Filesystem(nil).URI())
br := &testutils.BlockingRW{}
nw := &testutils.NoopRW{}
br := &testutil.BlockingRW{}
nw := &testutil.NoopRW{}
ci := &protocolmocks.ConnectionInfo{}
m.AddConnection(protocol.NewConnection(device1, br, nw, testutils.NoopCloser{}, m, ci, protocol.CompressionNever, nil, m.keyGen), protocol.Hello{})
m.AddConnection(protocol.NewConnection(device1, br, nw, testutil.NoopCloser{}, m, ci, protocol.CompressionNever, nil, m.keyGen), protocol.Hello{})
m.pmut.RLock()
if len(m.closed) != 1 {
t.Fatalf("Expected just one conn (len(m.closed) == %v)", len(m.closed))
@@ -3113,9 +3113,9 @@ func TestDeviceWasSeen(t *testing.T) {
}
func TestNewLimitedRequestResponse(t *testing.T) {
l0 := util.NewSemaphore(0)
l1 := util.NewSemaphore(1024)
l2 := (*util.Semaphore)(nil)
l0 := semaphore.New(0)
l1 := semaphore.New(1024)
l2 := (*semaphore.Semaphore)(nil)
// Should take 500 bytes from any non-unlimited non-nil limiters.
res := newLimitedRequestResponse(500, l0, l1, l2)
lib/model/queue_test.go
+13 -25
View File
@@ -13,6 +13,7 @@ import (
"time"
"github.com/d4l3k/messagediff"
"golang.org/x/exp/slices"
)
func TestJobQueue(t *testing.T) {
@@ -282,7 +283,6 @@ func BenchmarkJobQueuePushPopDone10k(b *testing.B) {
q.Done(n)
}
}
}
func TestQueuePagination(t *testing.T) {
@@ -302,21 +302,21 @@ func TestQueuePagination(t *testing.T) {
progress, queued, skip = q.Jobs(1, 5)
if len(progress) != 0 || len(queued) != 5 || skip != 0 {
t.Error("Wrong length", len(progress), len(queued), 0)
} else if !equalStrings(queued, names[:5]) {
} else if !slices.Equal(queued, names[:5]) {
t.Errorf("Wrong elements in queued, got %v, expected %v", queued, names[:5])
}
progress, queued, skip = q.Jobs(2, 5)
if len(progress) != 0 || len(queued) != 5 || skip != 5 {
t.Error("Wrong length", len(progress), len(queued), 0)
} else if !equalStrings(queued, names[5:]) {
} else if !slices.Equal(queued, names[5:]) {
t.Errorf("Wrong elements in queued, got %v, expected %v", queued, names[5:])
}
progress, queued, skip = q.Jobs(2, 7)
if len(progress) != 0 || len(queued) != 3 || skip != 7 {
t.Error("Wrong length", len(progress), len(queued), 0)
} else if !equalStrings(queued, names[7:]) {
} else if !slices.Equal(queued, names[7:]) {
t.Errorf("Wrong elements in queued, got %v, expected %v", queued, names[7:])
}
@@ -338,23 +338,23 @@ func TestQueuePagination(t *testing.T) {
progress, queued, skip = q.Jobs(1, 5)
if len(progress) != 1 || len(queued) != 4 || skip != 0 {
t.Error("Wrong length", len(progress), len(queued), 0)
} else if !equalStrings(progress, names[:1]) {
} else if !slices.Equal(progress, names[:1]) {
t.Errorf("Wrong elements in progress, got %v, expected %v", progress, names[:1])
} else if !equalStrings(queued, names[1:5]) {
} else if !slices.Equal(queued, names[1:5]) {
t.Errorf("Wrong elements in queued, got %v, expected %v", queued, names[1:5])
}
progress, queued, skip = q.Jobs(2, 5)
if len(progress) != 0 || len(queued) != 5 || skip != 5 {
t.Error("Wrong length", len(progress), len(queued), 0)
} else if !equalStrings(queued, names[5:]) {
} else if !slices.Equal(queued, names[5:]) {
t.Errorf("Wrong elements in queued, got %v, expected %v", queued, names[5:])
}
progress, queued, skip = q.Jobs(2, 7)
if len(progress) != 0 || len(queued) != 3 || skip != 7 {
t.Error("Wrong length", len(progress), len(queued), 0)
} else if !equalStrings(queued, names[7:]) {
} else if !slices.Equal(queued, names[7:]) {
t.Errorf("Wrong elements in queued, got %v, expected %v", queued, names[7:])
}
@@ -378,25 +378,25 @@ func TestQueuePagination(t *testing.T) {
progress, queued, skip = q.Jobs(1, 5)
if len(progress) != 5 || len(queued) != 0 || skip != 0 {
t.Error("Wrong length", len(progress), len(queued), 0)
} else if !equalStrings(progress, names[:5]) {
} else if !slices.Equal(progress, names[:5]) {
t.Errorf("Wrong elements in progress, got %v, expected %v", progress, names[:5])
}
progress, queued, skip = q.Jobs(2, 5)
if len(progress) != 3 || len(queued) != 2 || skip != 5 {
t.Error("Wrong length", len(progress), len(queued), 0)
} else if !equalStrings(progress, names[5:8]) {
} else if !slices.Equal(progress, names[5:8]) {
t.Errorf("Wrong elements in progress, got %v, expected %v", progress, names[5:8])
} else if !equalStrings(queued, names[8:]) {
} else if !slices.Equal(queued, names[8:]) {
t.Errorf("Wrong elements in queued, got %v, expected %v", queued, names[8:])
}
progress, queued, skip = q.Jobs(2, 7)
if len(progress) != 1 || len(queued) != 2 || skip != 7 {
t.Error("Wrong length", len(progress), len(queued), 0)
} else if !equalStrings(progress, names[7:8]) {
} else if !slices.Equal(progress, names[7:8]) {
t.Errorf("Wrong elements in progress, got %v, expected %v", progress, names[7:8])
} else if !equalStrings(queued, names[8:]) {
} else if !slices.Equal(queued, names[8:]) {
t.Errorf("Wrong elements in queued, got %v, expected %v", queued, names[8:])
}
@@ -405,15 +405,3 @@ func TestQueuePagination(t *testing.T) {
t.Error("Wrong length", len(progress), len(queued), 0)
}
}
func equalStrings(first, second []string) bool {
if len(first) != len(second) {
return false
}
for i := range first {
if first[i] != second[i] {
return false
}
}
return true
}