syncthing/lib/events/events.go

562 lines
12 KiB
Go
Raw Normal View History

2014-11-16 20:13:20 +00:00
// Copyright (C) 2014 The Syncthing Authors.
2014-09-29 19:43:32 +00:00
//
2015-03-07 20:36:35 +00:00
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this file,
// You can obtain one at https://mozilla.org/MPL/2.0/.
2014-07-25 12:50:14 +00:00
2014-07-13 19:07:24 +00:00
// Package events provides event subscription and polling functionality.
package events
import (
"context"
"encoding/json"
2014-07-13 19:07:24 +00:00
"errors"
"fmt"
"runtime"
2014-07-13 19:07:24 +00:00
"time"
2015-04-22 22:54:31 +00:00
2020-11-17 12:19:04 +00:00
"github.com/thejerf/suture/v4"
2015-08-06 09:29:25 +00:00
"github.com/syncthing/syncthing/lib/sync"
2014-07-13 19:07:24 +00:00
)
type EventType int64
2014-07-13 19:07:24 +00:00
const (
Starting EventType = 1 << iota
2014-07-13 19:07:24 +00:00
StartupComplete
DeviceDiscovered
DeviceConnected
DeviceDisconnected
DeviceRejected
2015-08-23 19:56:10 +00:00
DevicePaused
DeviceResumed
LocalChangeDetected
RemoteChangeDetected
2014-07-13 19:07:24 +00:00
LocalIndexUpdated
RemoteIndexUpdated
ItemStarted
2015-02-01 17:31:19 +00:00
ItemFinished
2014-07-17 11:38:36 +00:00
StateChanged
FolderRejected
2014-09-06 15:31:23 +00:00
ConfigSaved
DownloadProgress
RemoteDownloadProgress
FolderSummary
FolderCompletion
FolderErrors
2015-08-26 22:49:06 +00:00
FolderScanProgress
FolderPaused
FolderResumed
FolderWatchStateChanged
ListenAddressesChanged
LoginAttempt
Failure
2014-07-13 19:07:24 +00:00
AllEvents = (1 << iota) - 1
2014-07-13 19:07:24 +00:00
)
var (
runningTests = false
errNoop = errors.New("method of a noop object called")
)
const eventLogTimeout = 15 * time.Millisecond
2014-07-13 19:07:24 +00:00
func (t EventType) String() string {
switch t {
2014-07-17 11:38:36 +00:00
case Starting:
return "Starting"
2014-07-13 19:07:24 +00:00
case StartupComplete:
return "StartupComplete"
case DeviceDiscovered:
return "DeviceDiscovered"
case DeviceConnected:
return "DeviceConnected"
case DeviceDisconnected:
return "DeviceDisconnected"
case DeviceRejected:
return "DeviceRejected"
case LocalChangeDetected:
return "LocalChangeDetected"
case RemoteChangeDetected:
return "RemoteChangeDetected"
2014-07-13 19:07:24 +00:00
case LocalIndexUpdated:
return "LocalIndexUpdated"
case RemoteIndexUpdated:
return "RemoteIndexUpdated"
case ItemStarted:
return "ItemStarted"
2015-02-01 17:31:19 +00:00
case ItemFinished:
return "ItemFinished"
2014-07-17 11:38:36 +00:00
case StateChanged:
return "StateChanged"
case FolderRejected:
return "FolderRejected"
2014-09-06 15:31:23 +00:00
case ConfigSaved:
return "ConfigSaved"
case DownloadProgress:
return "DownloadProgress"
case RemoteDownloadProgress:
return "RemoteDownloadProgress"
case FolderSummary:
return "FolderSummary"
case FolderCompletion:
return "FolderCompletion"
case FolderErrors:
return "FolderErrors"
2015-08-23 19:56:10 +00:00
case DevicePaused:
return "DevicePaused"
case DeviceResumed:
return "DeviceResumed"
2015-08-26 22:49:06 +00:00
case FolderScanProgress:
return "FolderScanProgress"
case FolderPaused:
return "FolderPaused"
case FolderResumed:
return "FolderResumed"
case ListenAddressesChanged:
return "ListenAddressesChanged"
case LoginAttempt:
return "LoginAttempt"
case FolderWatchStateChanged:
return "FolderWatchStateChanged"
case Failure:
return "Failure"
2014-07-13 19:07:24 +00:00
default:
return "Unknown"
}
}
func (t EventType) MarshalText() ([]byte, error) {
return []byte(t.String()), nil
}
func (t *EventType) UnmarshalJSON(b []byte) error {
var s string
if err := json.Unmarshal(b, &s); err != nil {
return err
}
*t = UnmarshalEventType(s)
return nil
}
func UnmarshalEventType(s string) EventType {
switch s {
case "Starting":
return Starting
case "StartupComplete":
return StartupComplete
case "DeviceDiscovered":
return DeviceDiscovered
case "DeviceConnected":
return DeviceConnected
case "DeviceDisconnected":
return DeviceDisconnected
case "DeviceRejected":
return DeviceRejected
case "LocalChangeDetected":
return LocalChangeDetected
case "RemoteChangeDetected":
return RemoteChangeDetected
case "LocalIndexUpdated":
return LocalIndexUpdated
case "RemoteIndexUpdated":
return RemoteIndexUpdated
case "ItemStarted":
return ItemStarted
case "ItemFinished":
return ItemFinished
case "StateChanged":
return StateChanged
case "FolderRejected":
return FolderRejected
case "ConfigSaved":
return ConfigSaved
case "DownloadProgress":
return DownloadProgress
case "RemoteDownloadProgress":
return RemoteDownloadProgress
case "FolderSummary":
return FolderSummary
case "FolderCompletion":
return FolderCompletion
case "FolderErrors":
return FolderErrors
case "DevicePaused":
return DevicePaused
case "DeviceResumed":
return DeviceResumed
case "FolderScanProgress":
return FolderScanProgress
case "FolderPaused":
return FolderPaused
case "FolderResumed":
return FolderResumed
case "ListenAddressesChanged":
return ListenAddressesChanged
case "LoginAttempt":
return LoginAttempt
case "FolderWatchStateChanged":
return FolderWatchStateChanged
case "Failure":
return Failure
default:
return 0
}
}
2014-07-13 19:07:24 +00:00
const BufferSize = 64
type Logger interface {
suture.Service
Log(t EventType, data interface{})
Subscribe(mask EventType) Subscription
}
type logger struct {
subs []*subscription
nextSubscriptionIDs []int
nextGlobalID int
timeout *time.Timer
events chan Event
funcs chan func(context.Context)
toUnsubscribe chan *subscription
stop chan struct{}
2014-07-13 19:07:24 +00:00
}
type Event struct {
// Per-subscription sequential event ID. Named "id" for backwards compatibility with the REST API
SubscriptionID int `json:"id"`
// Global ID of the event across all subscriptions
GlobalID int `json:"globalID"`
Time time.Time `json:"time"`
Type EventType `json:"type"`
Data interface{} `json:"data"`
2014-07-13 19:07:24 +00:00
}
type Subscription interface {
C() <-chan Event
Poll(timeout time.Duration) (Event, error)
Mask() EventType
Unsubscribe()
2014-07-13 19:07:24 +00:00
}
type subscription struct {
mask EventType
events chan Event
toUnsubscribe chan *subscription
timeout *time.Timer
ctx context.Context
}
2014-07-13 19:07:24 +00:00
var (
ErrTimeout = errors.New("timeout")
ErrClosed = errors.New("closed")
)
func NewLogger() Logger {
l := &logger{
timeout: time.NewTimer(time.Second),
events: make(chan Event, BufferSize),
funcs: make(chan func(context.Context)),
toUnsubscribe: make(chan *subscription),
}
// Make sure the timer is in the stopped state and hasn't fired anything
// into the channel.
if !l.timeout.Stop() {
<-l.timeout.C
2014-07-13 19:07:24 +00:00
}
return l
2014-07-13 19:07:24 +00:00
}
2020-11-17 12:19:04 +00:00
func (l *logger) Serve(ctx context.Context) error {
loop:
for {
select {
case e := <-l.events:
// Incoming events get sent
l.sendEvent(e)
case fn := <-l.funcs:
// Subscriptions are handled here.
fn(ctx)
case s := <-l.toUnsubscribe:
l.unsubscribe(s)
case <-ctx.Done():
break loop
}
}
// Closing the event channels corresponds to what happens when a
// subscription is unsubscribed; this stops any BufferedSubscription,
// makes Poll() return ErrClosed, etc.
for _, s := range l.subs {
close(s.events)
}
2020-11-17 12:19:04 +00:00
return nil
}
func (l *logger) Log(t EventType, data interface{}) {
l.events <- Event{
Time: time.Now(),
Type: t,
Data: data,
// SubscriptionID and GlobalID are set in sendEvent
}
}
func (l *logger) sendEvent(e Event) {
l.nextGlobalID++
dl.Debugln("log", l.nextGlobalID, e.Type, e.Data)
e.GlobalID = l.nextGlobalID
for i, s := range l.subs {
if s.mask&e.Type != 0 {
e.SubscriptionID = l.nextSubscriptionIDs[i]
l.nextSubscriptionIDs[i]++
l.timeout.Reset(eventLogTimeout)
timedOut := false
2014-07-13 19:07:24 +00:00
select {
case s.events <- e:
case <-l.timeout.C:
// if s.events is not ready, drop the event
timedOut = true
}
// If stop returns false it already sent something to the
// channel. If we didn't already read it above we must do so now
// or we get a spurious timeout on the next loop.
if !l.timeout.Stop() && !timedOut {
<-l.timeout.C
2014-07-13 19:07:24 +00:00
}
}
}
}
func (l *logger) Subscribe(mask EventType) Subscription {
res := make(chan Subscription)
l.funcs <- func(ctx context.Context) {
dl.Debugln("subscribe", mask)
s := &subscription{
mask: mask,
events: make(chan Event, BufferSize),
toUnsubscribe: l.toUnsubscribe,
timeout: time.NewTimer(0),
ctx: ctx,
}
// We need to create the timeout timer in the stopped, non-fired state so
// that Subscription.Poll() can safely reset it and select on the timeout
// channel. This ensures the timer is stopped and the channel drained.
if runningTests {
// Make the behavior stable when running tests to avoid randomly
// varying test coverage. This ensures, in practice if not in
// theory, that the timer fires and we take the true branch of the
// next if.
runtime.Gosched()
}
if !s.timeout.Stop() {
<-s.timeout.C
}
l.subs = append(l.subs, s)
l.nextSubscriptionIDs = append(l.nextSubscriptionIDs, 1)
res <- s
}
return <-res
2014-07-13 19:07:24 +00:00
}
func (l *logger) unsubscribe(s *subscription) {
dl.Debugln("unsubscribe", s.mask)
for i, ss := range l.subs {
if s == ss {
last := len(l.subs) - 1
l.subs[i] = l.subs[last]
l.subs[last] = nil
l.subs = l.subs[:last]
l.nextSubscriptionIDs[i] = l.nextSubscriptionIDs[last]
l.nextSubscriptionIDs[last] = 0
l.nextSubscriptionIDs = l.nextSubscriptionIDs[:last]
break
}
}
close(s.events)
2014-07-13 19:07:24 +00:00
}
func (l *logger) String() string {
return fmt.Sprintf("events.Logger/@%p", l)
}
// Poll returns an event from the subscription or an error if the poll times
// out of the event channel is closed. Poll should not be called concurrently
// from multiple goroutines for a single subscription.
func (s *subscription) Poll(timeout time.Duration) (Event, error) {
Implement facility based logger, debugging via REST API This implements a new debug/trace infrastructure based on a slightly hacked up logger. Instead of the traditional "if debug { ... }" I've rewritten the logger to have no-op Debugln and Debugf, unless debugging has been enabled for a given "facility". The "facility" is just a string, typically a package name. This will be slightly slower than before; but not that much as it's mostly a function call that returns immediately. For the cases where it matters (the Debugln takes a hex.Dump() of something for example, and it's not in a very occasional "if err != nil" branch) there is an l.ShouldDebug(facility) that is fast enough to be used like the old "if debug". The point of all this is that we can now toggle debugging for the various packages on and off at runtime. There's a new method /rest/system/debug that can be POSTed a set of facilities to enable and disable debug for, or GET from to get a list of facilities with descriptions and their current debug status. Similarly a /rest/system/log?since=... can grab the latest log entries, up to 250 of them (hardcoded constant in main.go) plus the initial few. Not implemented in this commit (but planned) is a simple debug GUI available on /debug that shows the current log in an easily pasteable format and has checkboxes to enable the various debug facilities. The debug instructions to a user then becomes "visit this URL, check these boxes, reproduce your problem, copy and paste the log". The actual log viewer on the hypothetical /debug URL can poll regularly for new log entries and this bypass the 250 line limit. The existing STTRACE=foo variable is still obeyed and just sets the start state of the system.
2015-10-03 15:25:21 +00:00
dl.Debugln("poll", timeout)
2014-07-25 12:50:14 +00:00
s.timeout.Reset(timeout)
2014-07-13 19:07:24 +00:00
select {
case e, ok := <-s.events:
if !ok {
return e, ErrClosed
}
if runningTests {
// Make the behavior stable when running tests to avoid randomly
// varying test coverage. This ensures, in practice if not in
// theory, that the timer fires and we take the true branch of
// the next if.
s.timeout.Reset(0)
runtime.Gosched()
}
if !s.timeout.Stop() {
// The timeout must be stopped and possibly drained to be ready
// for reuse in the next call.
<-s.timeout.C
}
2014-07-13 19:07:24 +00:00
return e, nil
case <-s.timeout.C:
2014-07-13 19:07:24 +00:00
return Event{}, ErrTimeout
}
}
func (s *subscription) C() <-chan Event {
return s.events
}
func (s *subscription) Mask() EventType {
return s.mask
}
func (s *subscription) Unsubscribe() {
select {
case s.toUnsubscribe <- s:
case <-s.ctx.Done():
}
}
type bufferedSubscription struct {
sub Subscription
2014-07-13 19:07:24 +00:00
buf []Event
next int
cur int // Current SubscriptionID
2014-07-13 19:07:24 +00:00
mut sync.Mutex
cond *sync.TimeoutCond
2014-07-13 19:07:24 +00:00
}
type BufferedSubscription interface {
Since(id int, into []Event, timeout time.Duration) []Event
Mask() EventType
}
func NewBufferedSubscription(s Subscription, size int) BufferedSubscription {
bs := &bufferedSubscription{
2014-07-13 19:07:24 +00:00
sub: s,
buf: make([]Event, size),
2015-04-22 22:54:31 +00:00
mut: sync.NewMutex(),
2014-07-13 19:07:24 +00:00
}
bs.cond = sync.NewTimeoutCond(bs.mut)
2014-07-13 19:07:24 +00:00
go bs.pollingLoop()
return bs
}
func (s *bufferedSubscription) pollingLoop() {
for ev := range s.sub.C() {
2014-07-13 19:07:24 +00:00
s.mut.Lock()
s.buf[s.next] = ev
s.next = (s.next + 1) % len(s.buf)
s.cur = ev.SubscriptionID
2014-07-13 19:07:24 +00:00
s.cond.Broadcast()
s.mut.Unlock()
}
}
func (s *bufferedSubscription) Since(id int, into []Event, timeout time.Duration) []Event {
2014-07-13 19:07:24 +00:00
s.mut.Lock()
defer s.mut.Unlock()
// 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
}
}
2014-07-13 19:07:24 +00:00
}
for i := s.next; i < len(s.buf); i++ {
if s.buf[i].SubscriptionID > id {
2014-07-13 19:07:24 +00:00
into = append(into, s.buf[i])
}
}
for i := 0; i < s.next; i++ {
if s.buf[i].SubscriptionID > id {
2014-07-13 19:07:24 +00:00
into = append(into, s.buf[i])
}
}
return into
}
func (s *bufferedSubscription) Mask() EventType {
return s.sub.Mask()
}
// Error returns a string pointer suitable for JSON marshalling errors. It
2015-11-12 02:20:34 +00:00
// retains the "null on success" semantics, but ensures the error result is a
// string regardless of the underlying concrete error type.
func Error(err error) *string {
if err == nil {
return nil
}
str := err.Error()
return &str
}
type noopLogger struct{}
var NoopLogger Logger = &noopLogger{}
2020-11-17 12:19:04 +00:00
func (*noopLogger) Serve(ctx context.Context) error { return nil }
func (*noopLogger) Stop() {}
func (*noopLogger) Log(t EventType, data interface{}) {}
func (*noopLogger) Subscribe(mask EventType) Subscription {
return &noopSubscription{}
}
type noopSubscription struct{}
func (*noopSubscription) C() <-chan Event {
return nil
}
func (*noopSubscription) Poll(timeout time.Duration) (Event, error) {
return Event{}, errNoop
}
func (s *noopSubscription) Mask() EventType {
return 0
}
func (*noopSubscription) Unsubscribe() {}