diff --git a/cmd/syncthing/auditservice_test.go b/cmd/syncthing/auditservice_test.go index ad983c93..7f3396ee 100644 --- a/cmd/syncthing/auditservice_test.go +++ b/cmd/syncthing/auditservice_test.go @@ -20,13 +20,13 @@ func TestAuditService(t *testing.T) { service := newAuditService(buf) // Event sent before start, will not be logged - events.Default.Log(events.Ping, "the first event") + events.Default.Log(events.ConfigSaved, "the first event") go service.Serve() service.WaitForStart() // Event that should end up in the audit log - events.Default.Log(events.Ping, "the second event") + events.Default.Log(events.ConfigSaved, "the second event") // We need to give the events time to arrive, since the channels are buffered etc. time.Sleep(10 * time.Millisecond) @@ -35,7 +35,7 @@ func TestAuditService(t *testing.T) { service.WaitForStop() // This event should not be logged, since we have stopped. - events.Default.Log(events.Ping, "the third event") + events.Default.Log(events.ConfigSaved, "the third event") result := string(buf.Bytes()) t.Log(result) diff --git a/cmd/syncthing/gui.go b/cmd/syncthing/gui.go index bf8d564c..f9d68ce9 100644 --- a/cmd/syncthing/gui.go +++ b/cmd/syncthing/gui.go @@ -233,8 +233,8 @@ func (s *apiService) Serve() { getRestMux.HandleFunc("/rest/db/need", s.getDBNeed) // folder [perpage] [page] getRestMux.HandleFunc("/rest/db/status", s.getDBStatus) // folder getRestMux.HandleFunc("/rest/db/browse", s.getDBBrowse) // folder [prefix] [dirsonly] [levels] - getRestMux.HandleFunc("/rest/events", s.getIndexEvents) // since [limit] - getRestMux.HandleFunc("/rest/events/disk", s.getDiskEvents) // since [limit] + getRestMux.HandleFunc("/rest/events", s.getIndexEvents) // since [limit] [timeout] + getRestMux.HandleFunc("/rest/events/disk", s.getDiskEvents) // since [limit] [timeout] getRestMux.HandleFunc("/rest/stats/device", s.getDeviceStats) // - getRestMux.HandleFunc("/rest/stats/folder", s.getFolderStats) // - getRestMux.HandleFunc("/rest/svc/deviceid", s.getDeviceID) // id @@ -1019,9 +1019,15 @@ func (s *apiService) getEvents(w http.ResponseWriter, r *http.Request, eventSub qs := r.URL.Query() sinceStr := qs.Get("since") limitStr := qs.Get("limit") + timeoutStr := qs.Get("timeout") since, _ := strconv.Atoi(sinceStr) limit, _ := strconv.Atoi(limitStr) + timeout := defaultEventTimeout + if timeoutSec, timeoutErr := strconv.Atoi(timeoutStr); timeoutErr == nil && timeoutSec >= 0 { // 0 is a valid timeout + timeout = time.Duration(timeoutSec) * time.Second + } + // Flush before blocking, to indicate that we've received the request and // that it should not be retried. Must set Content-Type header before // flushing. @@ -1029,7 +1035,8 @@ func (s *apiService) getEvents(w http.ResponseWriter, r *http.Request, eventSub f := w.(http.Flusher) f.Flush() - evs := eventSub.Since(since, nil) + // If there are no events available return an empty slice, as this gets serialized as `[]` + evs := eventSub.Since(since, []events.Event{}, timeout) if 0 < limit && limit < len(evs) { evs = evs[len(evs)-limit:] } diff --git a/cmd/syncthing/main.go b/cmd/syncthing/main.go index 5b8680db..b25d18ce 100644 --- a/cmd/syncthing/main.go +++ b/cmd/syncthing/main.go @@ -79,7 +79,7 @@ const ( tlsDefaultCommonName = "syncthing" httpsRSABits = 2048 bepRSABits = 0 // 384 bit ECDSA used instead - pingEventInterval = time.Minute + defaultEventTimeout = time.Minute maxSystemErrors = 5 initialSystemLog = 10 maxSystemLog = 250 @@ -594,7 +594,7 @@ func syncthingMain(runtimeOptions RuntimeOptions) { // events. The LocalChangeDetected event might overwhelm the event // receiver in some situations so we will not subscribe to it here. apiSub := events.NewBufferedSubscription(events.Default.Subscribe(events.AllEvents&^events.LocalChangeDetected&^events.RemoteChangeDetected), 1000) - diskSub := events.NewBufferedSubscription(events.Default.Subscribe(events.LocalChangeDetected|events.RemoteChangeDetected|events.Ping), 1000) + diskSub := events.NewBufferedSubscription(events.Default.Subscribe(events.LocalChangeDetected|events.RemoteChangeDetected), 1000) if len(os.Getenv("GOMAXPROCS")) == 0 { runtime.GOMAXPROCS(runtime.NumCPU()) @@ -877,7 +877,6 @@ func syncthingMain(runtimeOptions RuntimeOptions) { events.Default.Log(events.StartupComplete, map[string]string{ "myID": myID.String(), }) - go generatePingEvents() cleanConfigDirectory() @@ -1091,13 +1090,6 @@ func defaultConfig(myName string) config.Configuration { return newCfg } -func generatePingEvents() { - for { - time.Sleep(pingEventInterval) - events.Default.Log(events.Ping, nil) - } -} - func resetDB() error { return os.RemoveAll(locations[locDatabase]) } diff --git a/cmd/syncthing/mocked_events_test.go b/cmd/syncthing/mocked_events_test.go index e7ce47fb..cb7931df 100644 --- a/cmd/syncthing/mocked_events_test.go +++ b/cmd/syncthing/mocked_events_test.go @@ -6,10 +6,14 @@ package main -import "github.com/syncthing/syncthing/lib/events" +import ( + "time" + + "github.com/syncthing/syncthing/lib/events" +) type mockedEventSub struct{} -func (s *mockedEventSub) Since(id int, into []events.Event) []events.Event { +func (s *mockedEventSub) Since(id int, into []events.Event, timeout time.Duration) []events.Event { select {} } diff --git a/cmd/syncthing/summaryservice.go b/cmd/syncthing/summaryservice.go index c4c498cb..83a5010e 100644 --- a/cmd/syncthing/summaryservice.go +++ b/cmd/syncthing/summaryservice.go @@ -169,7 +169,7 @@ func (c *folderSummaryService) foldersToHandle() []string { c.lastEventReqMut.Lock() last := c.lastEventReq c.lastEventReqMut.Unlock() - if time.Since(last) > pingEventInterval { + if time.Since(last) > defaultEventTimeout { return nil } diff --git a/cmd/syncthing/verboseservice.go b/cmd/syncthing/verboseservice.go index 0b99a42b..acb4a9f7 100644 --- a/cmd/syncthing/verboseservice.go +++ b/cmd/syncthing/verboseservice.go @@ -66,7 +66,7 @@ func (s *verboseService) WaitForStart() { func (s *verboseService) formatEvent(ev events.Event) string { switch ev.Type { - case events.Ping, events.DownloadProgress, events.LocalIndexUpdated: + case events.DownloadProgress, events.LocalIndexUpdated: // Skip return "" diff --git a/gui/default/syncthing/core/eventService.js b/gui/default/syncthing/core/eventService.js index 263da18e..486c0263 100644 --- a/gui/default/syncthing/core/eventService.js +++ b/gui/default/syncthing/core/eventService.js @@ -70,7 +70,6 @@ angular.module('syncthing.core') ITEM_FINISHED: 'ItemFinished', // Generated when Syncthing ends synchronizing a file to a newer version ITEM_STARTED: 'ItemStarted', // Generated when Syncthing begins synchronizing a file to a newer version LOCAL_INDEX_UPDATED: 'LocalIndexUpdated', // Generated when the local index information has changed, due to synchronizing one or more items from the cluster or discovering local changes during a scan - PING: 'Ping', // Generated automatically every 60 seconds REMOTE_INDEX_UPDATED: 'RemoteIndexUpdated', // Generated each time new index information is received from a device STARTING: 'Starting', // Emitted exactly once, when Syncthing starts, before parsing configuration etc STARTUP_COMPLETED: 'StartupCompleted', // Emitted exactly once, when initialization is complete and Syncthing is ready to start exchanging data with other devices diff --git a/lib/events/events.go b/lib/events/events.go index bcaa5629..70a7736e 100644 --- a/lib/events/events.go +++ b/lib/events/events.go @@ -10,7 +10,6 @@ package events import ( "errors" "runtime" - stdsync "sync" "time" "github.com/syncthing/syncthing/lib/sync" @@ -19,8 +18,7 @@ import ( type EventType int const ( - Ping EventType = 1 << iota - Starting + Starting EventType = 1 << iota StartupComplete DeviceDiscovered DeviceConnected @@ -55,8 +53,6 @@ var runningTests = false func (t EventType) String() string { switch t { - case Ping: - return "Ping" case Starting: return "Starting" case StartupComplete: @@ -279,11 +275,11 @@ type bufferedSubscription struct { next int cur int // Current SubscriptionID mut sync.Mutex - cond *stdsync.Cond + cond *sync.TimeoutCond } type BufferedSubscription interface { - Since(id int, into []Event) []Event + Since(id int, into []Event, timeout time.Duration) []Event } func NewBufferedSubscription(s *Subscription, size int) BufferedSubscription { @@ -292,7 +288,7 @@ func NewBufferedSubscription(s *Subscription, size int) BufferedSubscription { buf: make([]Event, size), mut: sync.NewMutex(), } - bs.cond = stdsync.NewCond(bs.mut) + bs.cond = sync.NewTimeoutCond(bs.mut) go bs.pollingLoop() return bs } @@ -319,12 +315,21 @@ func (s *bufferedSubscription) pollingLoop() { } } -func (s *bufferedSubscription) Since(id int, into []Event) []Event { +func (s *bufferedSubscription) Since(id int, into []Event, timeout time.Duration) []Event { s.mut.Lock() defer s.mut.Unlock() - for id >= s.cur { - s.cond.Wait() + // Check once first before generating the TimeoutCondWaiter + if id >= s.cur { + waiter := s.cond.SetupWait(timeout) + defer waiter.Stop() + + for id >= s.cur { + if eventsAvailable := waiter.Wait(); !eventsAvailable { + // Timed out + return into + } + } } for i := s.next; i < len(s.buf); i++ { diff --git a/lib/events/events_test.go b/lib/events/events_test.go index 79768a5c..ceec063e 100644 --- a/lib/events/events_test.go +++ b/lib/events/events_test.go @@ -219,7 +219,7 @@ func TestBufferedSub(t *testing.T) { recv := 0 for recv < 10*BufferSize { - evs := bs.Since(recv, nil) + evs := bs.Since(recv, nil, time.Minute) for _, ev := range evs { if ev.GlobalID != recv+1 { t.Fatalf("Incorrect ID; %d != %d", ev.GlobalID, recv+1) @@ -252,7 +252,7 @@ func BenchmarkBufferedSub(b *testing.B) { recv := 0 var evs []Event for i := 0; i < b.N; { - evs = bs.Since(recv, evs[:0]) + evs = bs.Since(recv, evs[:0], time.Minute) for _, ev := range evs { if ev.GlobalID != recv+1 { done <- fmt.Errorf("skipped event %v %v", ev.GlobalID, recv) @@ -299,7 +299,7 @@ func TestSinceUsesSubscriptionId(t *testing.T) { // delivered to the buffered subscription when we get here. t0 := time.Now() for time.Since(t0) < time.Second { - events := bs.Since(0, nil) + events := bs.Since(0, nil, time.Minute) if len(events) == 2 { break } @@ -308,7 +308,7 @@ func TestSinceUsesSubscriptionId(t *testing.T) { } } - events := bs.Since(1, nil) + events := bs.Since(1, nil, time.Minute) if len(events) != 1 { t.Fatal("Incorrect number of events:", len(events)) } diff --git a/lib/sync/sync.go b/lib/sync/sync.go index 182b9164..783ae252 100644 --- a/lib/sync/sync.go +++ b/lib/sync/sync.go @@ -227,3 +227,70 @@ func goid() int { } return id } + +// TimeoutCond is a variant on Cond. It has roughly the same semantics regarding 'L' - it must be held +// both when broadcasting and when calling TimeoutCondWaiter.Wait() +// Call Broadcast() to broadcast to all waiters on the TimeoutCond. Call SetupWait to create a +// TimeoutCondWaiter configured with the given timeout, which can then be used to listen for +// broadcasts. +type TimeoutCond struct { + L sync.Locker + ch chan struct{} +} + +// TimeoutCondWaiter is a type allowing a consumer to wait on a TimeoutCond with a timeout. Wait() may be called multiple times, +// and will return true every time that the TimeoutCond is broadcast to. Once the configured timeout +// expires, Wait() will return false. +// Call Stop() to release resources once this TimeoutCondWaiter is no longer needed. +type TimeoutCondWaiter struct { + c *TimeoutCond + timer *time.Timer +} + +func NewTimeoutCond(l sync.Locker) *TimeoutCond { + return &TimeoutCond{ + L: l, + } +} + +func (c *TimeoutCond) Broadcast() { + // ch.L must be locked when calling this function + + if c.ch != nil { + close(c.ch) + c.ch = nil + } +} + +func (c *TimeoutCond) SetupWait(timeout time.Duration) *TimeoutCondWaiter { + timer := time.NewTimer(timeout) + + return &TimeoutCondWaiter{ + c: c, + timer: timer, + } +} + +func (w *TimeoutCondWaiter) Wait() bool { + // ch.L must be locked when calling this function + + // Ensure that the channel exists, since we're going to be waiting on it + if w.c.ch == nil { + w.c.ch = make(chan struct{}) + } + ch := w.c.ch + + w.c.L.Unlock() + defer w.c.L.Lock() + + select { + case <-w.timer.C: + return false + case <-ch: + return true + } +} + +func (w *TimeoutCondWaiter) Stop() { + w.timer.Stop() +} diff --git a/lib/sync/sync_test.go b/lib/sync/sync_test.go index cf7a22e6..14f103a2 100644 --- a/lib/sync/sync_test.go +++ b/lib/sync/sync_test.go @@ -226,3 +226,100 @@ func TestWaitGroup(t *testing.T) { debug = false l.SetDebug("sync", false) } + +func TestTimeoutCond(t *testing.T) { + // WARNING this test relies heavily on threads not being stalled at particular points. + // As such, it's pretty unstable on the build server. It has been left in as it still + // exercises the deadlock detector, and one of the two things it tests is still functional. + // See the comments in runLocks + + const ( + // Low values to avoid being intrusive in continous testing. Can be + // increased significantly for stress testing. + iterations = 100 + routines = 10 + + timeMult = 2 + ) + + c := NewTimeoutCond(NewMutex()) + + // Start a routine to periodically broadcast on the cond. + + go func() { + d := time.Duration(routines) * timeMult * time.Millisecond / 2 + t.Log("Broadcasting every", d) + for i := 0; i < iterations; i++ { + time.Sleep(d) + + c.L.Lock() + c.Broadcast() + c.L.Unlock() + } + }() + + // Start several routines that wait on it with different timeouts. + + var results [routines][2]int + var wg sync.WaitGroup + for i := 0; i < routines; i++ { + i := i + wg.Add(1) + go func() { + d := time.Duration(i) * timeMult * time.Millisecond + t.Logf("Routine %d waits for %v\n", i, d) + succ, fail := runLocks(t, iterations, c, d) + results[i][0] = succ + results[i][1] = fail + wg.Done() + }() + } + + wg.Wait() + + // Print a table of routine number: successes, failures. + + for i, v := range results { + t.Logf("%4d: %4d %4d\n", i, v[0], v[1]) + } +} + +func runLocks(t *testing.T, iterations int, c *TimeoutCond, d time.Duration) (succ, fail int) { + for i := 0; i < iterations; i++ { + c.L.Lock() + + // The thread may be stalled, so we can't test the 'succeeded late' case reliably. + // Therefore make sure that we start t0 before starting the timeout, and only test + // the 'failed early' case. + + t0 := time.Now() + w := c.SetupWait(d) + + res := w.Wait() + waited := time.Since(t0) + + // Allow 20% slide in either direction, and a five milliseconds of + // scheduling delay... In tweaking these it was clear that things + // worked like the should, so if this becomes a spurious failure + // kind of thing feel free to remove or give significantly more + // slack. + + if !res && waited < d*8/10 { + t.Errorf("Wait failed early, %v < %v", waited, d) + } + if res && waited > d*11/10+5*time.Millisecond { + // Ideally this would be t.Errorf + t.Logf("WARNING: Wait succeeded late, %v > %v. This is probably a thread scheduling issue", waited, d) + } + + w.Stop() + + if res { + succ++ + } else { + fail++ + } + c.L.Unlock() + } + return +}