aboutsummaryrefslogtreecommitdiff
path: root/worker/types/worker.go
diff options
context:
space:
mode:
authorSimon Ser <contact@emersion.fr>2019-04-27 15:56:38 +0000
committerDrew DeVault <sir@cmpwn.com>2019-04-27 14:28:26 -0400
commit9ef2a57b51601cc0e18824bbfe8026891da1a38d (patch)
treef7f5ff2ba4e467dab56af13ef6993eac8addd481 /worker/types/worker.go
parent2159eb876e7e04e81f65e64b1d742ad832890289 (diff)
worker/types: fix Worker.Callbacks race condition
Worker.Process* functions were called in different goroutines than Worker.Post*. Protect the map with a mutex. Also make the map unexported to prevent external unprotected accesses. Worker.Process* functions used to delete items from the map. However they didn't delete the element they retrieved: callbacks[msg.InResponseTo()] was read while callbacks[msg] was deleted. I'm not sure I understand why. I tried to delete the element that was accessed - but this broke everything (UI froze at "Connecting..."). I don't believe any elements were actually removed from the map, so the new code just doesn't remove anything.
Diffstat (limited to 'worker/types/worker.go')
-rw-r--r--worker/types/worker.go56
1 files changed, 41 insertions, 15 deletions
diff --git a/worker/types/worker.go b/worker/types/worker.go
index 1ff16a4..0ed216a 100644
--- a/worker/types/worker.go
+++ b/worker/types/worker.go
@@ -2,6 +2,7 @@ package types
import (
"log"
+ "sync"
)
type Backend interface {
@@ -9,11 +10,42 @@ type Backend interface {
}
type Worker struct {
- Actions chan WorkerMessage
- Backend Backend
- Callbacks map[WorkerMessage]func(msg WorkerMessage)
- Messages chan WorkerMessage
- Logger *log.Logger
+ Backend Backend
+ Actions chan WorkerMessage
+ Messages chan WorkerMessage
+ Logger *log.Logger
+
+ callbacks map[WorkerMessage]func(msg WorkerMessage) // protected by mutex
+ mutex sync.Mutex
+}
+
+func NewWorker(logger *log.Logger) *Worker {
+ return &Worker{
+ Actions: make(chan WorkerMessage, 50),
+ Messages: make(chan WorkerMessage, 50),
+ Logger: logger,
+ callbacks: make(map[WorkerMessage]func(msg WorkerMessage)),
+ }
+}
+
+func (worker *Worker) setCallback(msg WorkerMessage,
+ cb func(msg WorkerMessage)) {
+
+ if cb != nil {
+ worker.mutex.Lock()
+ worker.callbacks[msg] = cb
+ worker.mutex.Unlock()
+ }
+}
+
+func (worker *Worker) getCallback(msg WorkerMessage) (func(msg WorkerMessage),
+ bool) {
+
+ worker.mutex.Lock()
+ cb, ok := worker.callbacks[msg]
+ worker.mutex.Unlock()
+
+ return cb, ok
}
func (worker *Worker) PostAction(msg WorkerMessage,
@@ -26,9 +58,7 @@ func (worker *Worker) PostAction(msg WorkerMessage,
}
worker.Actions <- msg
- if cb != nil {
- worker.Callbacks[msg] = cb
- }
+ worker.setCallback(msg, cb)
}
func (worker *Worker) PostMessage(msg WorkerMessage,
@@ -41,9 +71,7 @@ func (worker *Worker) PostMessage(msg WorkerMessage,
}
worker.Messages <- msg
- if cb != nil {
- worker.Callbacks[msg] = cb
- }
+ worker.setCallback(msg, cb)
}
func (worker *Worker) ProcessMessage(msg WorkerMessage) WorkerMessage {
@@ -52,9 +80,8 @@ func (worker *Worker) ProcessMessage(msg WorkerMessage) WorkerMessage {
} else {
worker.Logger.Printf("(ui)<= %T\n", msg)
}
- if cb, ok := worker.Callbacks[msg.InResponseTo()]; ok {
+ if cb, ok := worker.getCallback(msg.InResponseTo()); ok {
cb(msg)
- delete(worker.Callbacks, msg)
}
return msg
}
@@ -65,9 +92,8 @@ func (worker *Worker) ProcessAction(msg WorkerMessage) WorkerMessage {
} else {
worker.Logger.Printf("<-(ui) %T\n", msg)
}
- if cb, ok := worker.Callbacks[msg.InResponseTo()]; ok {
+ if cb, ok := worker.getCallback(msg.InResponseTo()); ok {
cb(msg)
- delete(worker.Callbacks, msg)
}
return msg
}