From f8ffd68060dc572c0d73c91b5702435bc2d432a5 Mon Sep 17 00:00:00 2001 From: Jesse Hallam Date: Thu, 22 Nov 2018 04:53:44 -0500 Subject: [PATCH] Webhub race condition (#9863) * fix webconn close semantics Avoid race conditions in WebConn on shutdown by closing channels to guarantee all readers are notified. Wrap this with sync.Once to avoid closing the channel more than once. * web_hub_test.go * webhub: fix race condition on shutdown Ensure that if the webhub shuts down in the process of sending, the caller unblocks given that the webhub will no longer consume incoming events. * panic if app shutdown takes >30 seconds * simplify WebConn::Pump channel semantics too --- api4/apitestlib.go | 18 +++++++++++++++++- app/apptestlib.go | 19 ++++++++++++++++++- app/web_conn.go | 20 +++++++++++++------- app/web_hub.go | 22 +++++++++++++++++----- app/web_hub_test.go | 43 +++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 108 insertions(+), 14 deletions(-) diff --git a/api4/apitestlib.go b/api4/apitestlib.go index 0a843dfab5..ec047d5a68 100644 --- a/api4/apitestlib.go +++ b/api4/apitestlib.go @@ -179,10 +179,26 @@ func SetupConfig(updateConfig func(cfg *model.Config)) *TestHelper { return setupTestHelper(false, updateConfig) } +func (me *TestHelper) ShutdownApp() { + done := make(chan bool) + go func() { + me.App.Shutdown() + close(done) + }() + + select { + case <-done: + case <-time.After(30 * time.Second): + // panic instead of t.Fatal to terminate all tests in this package, otherwise the + // still running App could spuriously fail subsequent tests. + panic("failed to shutdown App within 30 seconds") + } +} + func (me *TestHelper) TearDown() { utils.DisableDebugLogForTest() - me.App.Shutdown() + me.ShutdownApp() os.Remove(me.tempConfigPath) utils.EnableDebugLogForTest() diff --git a/app/apptestlib.go b/app/apptestlib.go index 01dc6b30e1..81569c90fd 100644 --- a/app/apptestlib.go +++ b/app/apptestlib.go @@ -391,8 +391,25 @@ func (me *TestHelper) CreateScheme() (*model.Scheme, []*model.Role) { return scheme, roles } +func (me *TestHelper) ShutdownApp() { + done := make(chan bool) + go func() { + me.App.Shutdown() + close(done) + }() + + select { + case <-done: + case <-time.After(30 * time.Second): + // panic instead of t.Fatal to terminate all tests in this package, otherwise the + // still running App could spuriously fail subsequent tests. + panic("failed to shutdown App within 30 seconds") + } +} + func (me *TestHelper) TearDown() { - me.App.Shutdown() + me.ShutdownApp() + os.Remove(me.tempConfigPath) if err := recover(); err != nil { StopTestStore() diff --git a/app/web_conn.go b/app/web_conn.go index 7eea5d3bb3..49e85f62da 100644 --- a/app/web_conn.go +++ b/app/web_conn.go @@ -5,6 +5,7 @@ package app import ( "fmt" + "sync" "sync/atomic" "time" @@ -39,6 +40,7 @@ type WebConn struct { AllChannelMembers map[string]string LastAllChannelMembersTime int64 Sequence int64 + closeOnce sync.Once endWritePump chan struct{} pumpFinished chan struct{} } @@ -59,8 +61,8 @@ func (a *App) NewWebConn(ws *websocket.Conn, session model.Session, t goi18n.Tra UserId: session.UserId, T: t, Locale: locale, - endWritePump: make(chan struct{}, 2), - pumpFinished: make(chan struct{}, 1), + endWritePump: make(chan struct{}), + pumpFinished: make(chan struct{}), } wc.SetSession(&session) @@ -72,7 +74,9 @@ func (a *App) NewWebConn(ws *websocket.Conn, session model.Session, t goi18n.Tra func (wc *WebConn) Close() { wc.WebSocket.Close() - wc.endWritePump <- struct{}{} + wc.closeOnce.Do(func() { + close(wc.endWritePump) + }) <-wc.pumpFinished } @@ -105,16 +109,18 @@ func (c *WebConn) SetSession(v *model.Session) { } func (c *WebConn) Pump() { - ch := make(chan struct{}, 1) + ch := make(chan struct{}) go func() { c.writePump() - ch <- struct{}{} + close(ch) }() c.readPump() - c.endWritePump <- struct{}{} + c.closeOnce.Do(func() { + close(c.endWritePump) + }) <-ch c.App.HubUnregister(c) - c.pumpFinished <- struct{}{} + close(c.pumpFinished) } func (c *WebConn) readPump() { diff --git a/app/web_hub.go b/app/web_hub.go index f65bec931d..8427576362 100644 --- a/app/web_hub.go +++ b/app/web_hub.go @@ -346,7 +346,10 @@ func (a *App) UpdateWebConnUserActivity(session model.Session, activityAt int64) } func (h *Hub) Register(webConn *WebConn) { - h.register <- webConn + select { + case h.register <- webConn: + case <-h.didStop: + } if webConn.IsAuthenticated() { webConn.SendHello() @@ -356,22 +359,31 @@ func (h *Hub) Register(webConn *WebConn) { func (h *Hub) Unregister(webConn *WebConn) { select { case h.unregister <- webConn: - case <-h.stop: + case <-h.didStop: } } func (h *Hub) Broadcast(message *model.WebSocketEvent) { if h != nil && h.broadcast != nil && message != nil { - h.broadcast <- message + select { + case h.broadcast <- message: + case <-h.didStop: + } } } func (h *Hub) InvalidateUser(userId string) { - h.invalidateUser <- userId + select { + case h.invalidateUser <- userId: + case <-h.didStop: + } } func (h *Hub) UpdateActivity(userId, sessionToken string, activityAt int64) { - h.activity <- &WebConnActivityMessage{UserId: userId, SessionToken: sessionToken, ActivityAt: activityAt} + select { + case h.activity <- &WebConnActivityMessage{UserId: userId, SessionToken: sessionToken, ActivityAt: activityAt}: + case <-h.didStop: + } } func getGoroutineId() int { diff --git a/app/web_hub_test.go b/app/web_hub_test.go index 8702acb21c..83b58489a3 100644 --- a/app/web_hub_test.go +++ b/app/web_hub_test.go @@ -5,6 +5,7 @@ import ( "net/http" "net/http/httptest" "testing" + "time" "github.com/gorilla/websocket" goi18n "github.com/nicksnyder/go-i18n/i18n" @@ -60,3 +61,45 @@ func TestHubStopWithMultipleConnections(t *testing.T) { defer wc2.Close() defer wc3.Close() } + +// TestHubStopRaceCondition verifies that attempts to use the hub after it has shutdown does not +// block the caller indefinitely. +func TestHubStopRaceCondition(t *testing.T) { + th := Setup().InitBasic() + defer th.TearDown() + + s := httptest.NewServer(http.HandlerFunc(dummyWebsocketHandler(t))) + + th.App.HubStart() + wc1 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id) + defer wc1.Close() + + hub := th.App.Srv.Hubs[0] + th.App.HubStop() + time.Sleep(5 * time.Second) + + done := make(chan bool) + go func() { + wc4 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id) + wc5 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id) + hub.Register(wc4) + hub.Register(wc5) + + hub.UpdateActivity("userId", "sessionToken", 0) + + for i := 0; i <= BROADCAST_QUEUE_SIZE; i++ { + hub.Broadcast(&model.WebSocketEvent{}) + } + + hub.InvalidateUser("userId") + hub.Unregister(wc4) + hub.Unregister(wc5) + close(done) + }() + + select { + case <-done: + case <-time.After(15 * time.Second): + t.Fatalf("hub call did not return within 15 seconds after stop") + } +}