syncthing/internal/events/events.go
Jakob Borg 7bf55dd14f events.Subscription.Poll does not seem to require locking
This is a large source of output from the new lock logging, and it
doesn't seem to accomplish anything useful that I can see. Running
integration with the race detector to make sure...
2015-04-24 11:25:42 +09:00

256 lines
4.5 KiB
Go

// Copyright (C) 2014 The Syncthing Authors.
//
// 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 http://mozilla.org/MPL/2.0/.
// Package events provides event subscription and polling functionality.
package events
import (
"errors"
stdsync "sync"
"time"
"github.com/syncthing/syncthing/internal/sync"
)
type EventType int
const (
Ping EventType = 1 << iota
Starting
StartupComplete
DeviceDiscovered
DeviceConnected
DeviceDisconnected
DeviceRejected
LocalIndexUpdated
RemoteIndexUpdated
ItemStarted
ItemFinished
StateChanged
FolderRejected
ConfigSaved
DownloadProgress
FolderSummary
FolderCompletion
AllEvents = (1 << iota) - 1
)
func (t EventType) String() string {
switch t {
case Ping:
return "Ping"
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 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 FolderSummary:
return "FolderSummary"
case FolderCompletion:
return "FolderCompletion"
default:
return "Unknown"
}
}
func (t EventType) MarshalText() ([]byte, error) {
return []byte(t.String()), nil
}
const BufferSize = 64
type Logger struct {
subs map[int]*Subscription
nextID int
mutex sync.Mutex
}
type Event struct {
ID int `json:"id"`
Time time.Time `json:"time"`
Type EventType `json:"type"`
Data interface{} `json:"data"`
}
type Subscription struct {
mask EventType
id int
events chan Event
}
var Default = NewLogger()
var (
ErrTimeout = errors.New("timeout")
ErrClosed = errors.New("closed")
)
func NewLogger() *Logger {
return &Logger{
subs: make(map[int]*Subscription),
mutex: sync.NewMutex(),
}
}
func (l *Logger) Log(t EventType, data interface{}) {
l.mutex.Lock()
if debug {
dl.Debugln("log", l.nextID, t.String(), data)
}
e := Event{
ID: l.nextID,
Time: time.Now(),
Type: t,
Data: data,
}
l.nextID++
for _, s := range l.subs {
if s.mask&t != 0 {
select {
case s.events <- e:
default:
// if s.events is not ready, drop the event
}
}
}
l.mutex.Unlock()
}
func (l *Logger) Subscribe(mask EventType) *Subscription {
l.mutex.Lock()
if debug {
dl.Debugln("subscribe", mask)
}
s := &Subscription{
mask: mask,
id: l.nextID,
events: make(chan Event, BufferSize),
}
l.nextID++
l.subs[s.id] = s
l.mutex.Unlock()
return s
}
func (l *Logger) Unsubscribe(s *Subscription) {
l.mutex.Lock()
if debug {
dl.Debugln("unsubscribe")
}
delete(l.subs, s.id)
close(s.events)
l.mutex.Unlock()
}
func (s *Subscription) Poll(timeout time.Duration) (Event, error) {
if debug {
dl.Debugln("poll", timeout)
}
to := time.After(timeout)
select {
case e, ok := <-s.events:
if !ok {
return e, ErrClosed
}
return e, nil
case <-to:
return Event{}, ErrTimeout
}
}
func (s *Subscription) C() <-chan Event {
return s.events
}
type BufferedSubscription struct {
sub *Subscription
buf []Event
next int
cur int
mut sync.Mutex
cond *stdsync.Cond
}
func NewBufferedSubscription(s *Subscription, size int) *BufferedSubscription {
bs := &BufferedSubscription{
sub: s,
buf: make([]Event, size),
mut: sync.NewMutex(),
}
bs.cond = stdsync.NewCond(bs.mut)
go bs.pollingLoop()
return bs
}
func (s *BufferedSubscription) pollingLoop() {
for {
ev, err := s.sub.Poll(60 * time.Second)
if err == ErrTimeout {
continue
}
if err == ErrClosed {
return
}
if err != nil {
panic("unexpected error: " + err.Error())
}
s.mut.Lock()
s.buf[s.next] = ev
s.next = (s.next + 1) % len(s.buf)
s.cur = ev.ID
s.cond.Broadcast()
s.mut.Unlock()
}
}
func (s *BufferedSubscription) Since(id int, into []Event) []Event {
s.mut.Lock()
defer s.mut.Unlock()
for id >= s.cur {
s.cond.Wait()
}
for i := s.next; i < len(s.buf); i++ {
if s.buf[i].ID > id {
into = append(into, s.buf[i])
}
}
for i := 0; i < s.next; i++ {
if s.buf[i].ID > id {
into = append(into, s.buf[i])
}
}
return into
}