Merge pull request #6 from ethanv2/event

Merge new event system, which drastically reduces CPU load, increases percentage time idle (preserving my battery life) and removes a few hacky pieces of redundant waiter code.

Author: ejv2

data/cache.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/dhowden/tag"
 	"github.com/ethanv2/podbit/data/escape"
+	ev "github.com/ethanv2/podbit/event"
 )
 
 // Possible cache errors.
@@ -35,6 +36,8 @@ type Cache struct {
 	downloadsMutex sync.RWMutex // Protects the below two variables
 	downloads      []*Download
 	ongoing        int
+
+	hndl ev.Handler
 }
 
 // Episode represents the data extracted from a single cached episode
@@ -85,10 +88,11 @@ func (c *Cache) guessDir() string {
 // Open opens and initialises the cache.
 // Should be called once and once only - further modifications
 // and cache mutations happen exclusively through other methods.
-func (c *Cache) Open() error {
+func (c *Cache) Open(hndl ev.Handler) error {
 	home, _ := os.UserHomeDir()
 	c.dir = strings.ReplaceAll(c.guessDir(), "~", home)
 	files, err := ioutil.ReadDir(c.dir)
+	c.hndl = hndl
 
 	if err != nil {
 		cerr := os.MkdirAll(c.dir, os.ModeDir|os.ModePerm)
@@ -175,9 +179,9 @@ func (c *Cache) Download(item *QueueItem) (id int, err error) {
 	}
 
 	if item.Youtube {
-		go dl.DownloadYoutube()
+		go dl.DownloadYoutube(c.hndl)
 	} else {
-		go dl.DownloadHTTP()
+		go dl.DownloadHTTP(c.hndl)
 	}
 
 	c.downloadsMutex.Lock()

data/data.go

@@ -11,11 +11,13 @@ import (
 	"net/url"
 	"os"
 	"time"
+
+	ev "github.com/ethanv2/podbit/event"
 )
 
 const (
 	// QueueReloadInterval is how often the queue will be reloaded.
-	QueueReloadInterval = time.Duration(15) * time.Second
+	QueueReloadInterval = time.Minute
 	// EpisodeCacheTime is how long an episode is allowed to stay in cache in seconds.
 	// Default value is three days (3 * 24 * 60 * 60).
 	EpisodeCacheTime = 259200
@@ -30,7 +32,7 @@ var (
 
 // InitData initialises all dependent data structures.
 // The only returned errors *will* be fatal to the program.
-func InitData() error {
+func InitData(hndl ev.Handler) error {
 	fmt.Print("Reading queue...")
 	err := Q.Open()
 	if err != nil {
@@ -46,7 +48,7 @@ func InitData() error {
 	fmt.Println("done")
 
 	fmt.Print("Initialising cache...")
-	err = Downloads.Open()
+	err = Downloads.Open(hndl)
 	if err != nil {
 		return err
 	}

data/download.go

@@ -10,6 +10,8 @@ import (
 	"strings"
 	"sync"
 	"time"
+
+	ev "github.com/ethanv2/podbit/event"
 )
 
 // YouTube downloading constants.
@@ -65,7 +67,7 @@ type Download struct {
 // (synchronously).
 //
 // Used internally by cache; avoid calling directly.
-func (d *Download) DownloadYoutube() {
+func (d *Download) DownloadYoutube(hndl ev.Handler) {
 	if !d.Elem.Youtube {
 		panic("download: downloading non-youtube with youtube-dl")
 	}
@@ -133,6 +135,8 @@ func (d *Download) DownloadYoutube() {
 		d.Percentage /= 100
 		d.mut.Unlock()
 
+		hndl.Post(ev.DownloadChanged)
+
 		select {
 		case <-d.Stop:
 			d.mut.Lock()
@@ -185,14 +189,16 @@ func (d *Download) DownloadYoutube() {
 	Downloads.downloadsMutex.Unlock()
 
 	d.mut.Unlock()
+
+	hndl.Post(ev.DownloadChanged)
 }
 
 // DownloadHTTP connects to the URL of the specified download
 // and downloads to download path on the calling thread
 // (synchronously)
 //
 // Used internally by cache; avoid calling directly.
-func (d *Download) DownloadHTTP() {
+func (d *Download) DownloadHTTP(hndl ev.Handler) {
 	resp, err := http.Get(d.Elem.URL)
 	if err != nil || resp.StatusCode != http.StatusOK {
 		d.mut.Lock()
@@ -234,6 +240,7 @@ outer:
 		d.Percentage = float64(d.Done) / float64(d.Size)
 		d.mut.Unlock()
 
+		hndl.Post(ev.DownloadChanged)
 		if Downloads.Ongoing() > 1 {
 			runtime.Gosched() // Give the other threads a turn
 		}

event/event.go

@@ -0,0 +1,14 @@
+package ev
+
+// Event classes.
+// ev.Handler passes events as integers between goroutines. These integers are
+// not arbitrary and are selected by the caller. This allows the receiving
+// goroutine to select which events are relevant to it using simple arithmetic
+// and/or a switch case.
+const (
+	Keystroke = iota
+	Resize
+	TrayMessage
+	PlayerChanged
+	DownloadChanged
+)

event/handle.go

@@ -0,0 +1,37 @@
+package ev
+
+// Handler is a multiplexer over multiple subsystems waiting for events in
+// parallel. When an event is delivered from one subsystem, handler bounces the
+// message to all other listening handlers.
+//
+// Handler is thread safe, provided that handlers are only added in sequence
+// and it is only ever passed by value after initialisation completes.
+type Handler struct {
+	msg  chan int
+	hndl []chan int
+}
+
+func NewHandler() *Handler {
+	return &Handler{
+		make(chan int, 1),
+		make([]chan int, 0, 3),
+	}
+}
+
+func (h Handler) Run() {
+	for msg := range h.msg {
+		for _, elem := range h.hndl {
+			elem <- msg
+		}
+	}
+}
+
+func (h *Handler) Register() chan int {
+	hndl := make(chan int, 1)
+	h.hndl = append(h.hndl, hndl)
+	return hndl
+}
+
+func (h Handler) Post(ev int) {
+	h.msg <- ev
+}

main.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/ethanv2/podbit/colors"
 	"github.com/ethanv2/podbit/data"
+	ev "github.com/ethanv2/podbit/event"
 	"github.com/ethanv2/podbit/sound"
 	"github.com/ethanv2/podbit/ui"
 
@@ -24,10 +25,10 @@ var (
 	homedir string
 	confdir string
 
-	redraw    = make(chan int)
+	events    *ev.Handler
 	keystroke = make(chan rune)
 	newMen    = make(chan ui.Menu)
-	exit      = make(chan int)
+	exit      = make(chan struct{})
 )
 
 func banner() {
@@ -91,8 +92,9 @@ func main() {
 	}
 	defer lock.Unlock()
 
+	events = ev.NewHandler()
 	now := time.Now()
-	err := data.InitData()
+	err := data.InitData(*events)
 	if err != nil {
 		fmt.Println("\n" + err.Error())
 		return
@@ -101,7 +103,7 @@ func main() {
 	go data.ReloadLoop()
 
 	fmt.Print("Initialising sound system...")
-	sound.Plr, err = sound.NewPlayer()
+	sound.Plr, err = sound.NewPlayer(events)
 	if err != nil {
 		fmt.Printf("\nError: Failed to initialise sound system: %s\n", err.Error())
 		os.Exit(1)
@@ -120,7 +122,7 @@ func main() {
 	initColors()
 	defer goncurses.End()
 
-	ui.InitUI(scr, ui.LibraryMenu, redraw, keystroke, newMen)
+	ui.InitUI(scr, ui.LibraryMenu, events, keystroke, newMen)
 	go ui.RenderLoop()
 
 	// Welcome message
@@ -132,8 +134,9 @@ func main() {
 		len(data.Q.Items), len(data.Q.GetPodcasts()),
 		startup.Seconds()))
 
-	// Initial UI draw
-	ui.Redraw(ui.RedrawAll)
+	// Run events handler and kickstart listeners
+	go events.Run()
+	events.Post(ev.Keystroke)
 
 	// Initialisation is done; use this thread as the input loop
 	ui.InputLoop(exit)

sound/sound.go

@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"github.com/ethanv2/podbit/data"
+	ev "github.com/ethanv2/podbit/event"
 
 	"github.com/blang/mpv"
 )
@@ -34,7 +35,7 @@ var (
 	// to idle mpv ready to receive instructions.
 	PlayerArgs = []string{"--idle", "--no-video", "--input-ipc-server=" + PlayerRPC}
 	// UpdateTime is the time between queue checks and supervision updates.
-	UpdateTime = 200 * time.Millisecond
+	UpdateTime = 500 * time.Millisecond
 )
 
 // Internal: Types of actions.
@@ -61,6 +62,10 @@ type WaitFunc func(u chan int)
 type Player struct {
 	proc *exec.Cmd
 
+	hndl   ev.Handler
+	event  chan int
+	dlchan chan struct{}
+
 	act chan int
 	dat chan interface{}
 
@@ -111,7 +116,7 @@ func endWait(u chan int) {
 func downloadWait(u chan int) {
 	var id int
 	for y := true; y && DownloadAtHead(Plr.download); y, id = data.Downloads.IsDownloading(Plr.download.Path) {
-		time.Sleep(UpdateTime)
+		<-Plr.dlchan
 	}
 
 	Plr.waiting = false
@@ -125,23 +130,19 @@ func downloadWait(u chan int) {
 		head--
 	}
 
-	u <- 1
-}
-
-func newWait(u chan int) {
-	for head >= len(queue) {
-		time.Sleep(UpdateTime)
-	}
-
-	Plr.waiting = false
+	Plr.hndl.Post(ev.PlayerChanged)
 	u <- 1
 }
 
 // NewPlayer constructs a new player. This does not yet
 // launch any processes or play any media.
-func NewPlayer() (p Player, err error) {
+func NewPlayer(events *ev.Handler) (p Player, err error) {
 	p.act = make(chan int)
 	p.dat = make(chan interface{})
+	p.dlchan = make(chan struct{}, 1)
+
+	p.hndl = *events
+	p.event = events.Register()
 
 	return
 }
@@ -377,10 +378,23 @@ func (p *Player) Wait() {
 
 	now, _ := p.ctrl.Filename()
 	for filename := now; filename == now; filename, _ = p.ctrl.Filename() {
+		if !p.isPaused() {
+			p.hndl.Post(ev.PlayerChanged)
+		}
 		time.Sleep(UpdateTime)
 	}
 }
 
+func (p *Player) Event(e int) {
+	switch e {
+	case ev.DownloadChanged:
+		select {
+		case p.dlchan <- struct{}{}:
+		default:
+		}
+	}
+}
+
 // Mainloop - Main sound handling thread loop
 //
 // Loops infinitely waiting for sound events, managing the queue, mpv
@@ -434,7 +448,9 @@ func Mainloop() {
 			select {
 			case <-u:
 				keepWaiting = false
-
+				Plr.hndl.Post(ev.PlayerChanged)
+			case e := <-Plr.event:
+				Plr.Event(e)
 			case action := <-Plr.act:
 				switch action {
 				case actTerm:
@@ -472,5 +488,7 @@ func Mainloop() {
 				}
 			}
 		}
+
+		Plr.hndl.Post(ev.PlayerChanged)
 	}
 }

ui/download.go

@@ -8,6 +8,7 @@ import (
 	"github.com/ethanv2/podbit/data"
 	"github.com/ethanv2/podbit/sound"
 	"github.com/ethanv2/podbit/ui/components"
+	ev "github.com/ethanv2/podbit/event"
 )
 
 var downloadHeadings []components.Column = []components.Column{
@@ -76,6 +77,10 @@ func (q *Downloads) Render(x, y int) {
 	q.tbl.Render()
 }
 
+func (q *Downloads) Should(event int) bool {
+	return event == ev.Keystroke || event == ev.DownloadChanged || event == ev.PlayerChanged
+}
+
 func (q *Downloads) Input(c rune) {
 	switch c {
 	case 'j':