Merge pull request #248 from riking/ccache

Compile fixes, doc updates, visibility into caching

socketserver/server/backend.go

@@ -18,6 +18,7 @@ import (
 	"github.com/FrankerFaceZ/FrankerFaceZ/socketserver/server/naclform"
 	cache "github.com/patrickmn/go-cache"
 	"golang.org/x/crypto/nacl/box"
+	"golang.org/x/sync/singleflight"
 )
 
 const bPathAnnounceStartup = "/startup"
@@ -29,6 +30,7 @@ type backendInfo struct {
 	HTTPClient    http.Client
 	baseURL       string
 	responseCache *cache.Cache
+	reloadGroup   singleflight.Group
 
 	postStatisticsURL  string
 	addTopicURL        string
@@ -49,7 +51,8 @@ func setupBackend(config *ConfigFile) *backendInfo {
 
 	b.HTTPClient.Timeout = 60 * time.Second
 	b.baseURL = config.BackendURL
-	b.responseCache = cache.New(60*time.Second, 120*time.Second)
+	// size in bytes of string payload
+	b.responseCache = cache.New(60*time.Second, 10*time.Minute)
 
 	b.announceStartupURL = fmt.Sprintf("%s%s", b.baseURL, bPathAnnounceStartup)
 	b.addTopicURL = fmt.Sprintf("%s%s", b.baseURL, bPathAddTopic)
@@ -88,12 +91,18 @@ func (bfe ErrForwardedFromBackend) Error() string {
 }
 
 // ErrAuthorizationNeeded is emitted when the backend replies with HTTP 401.
+//
 // Indicates that an attempt to validate `ClientInfo.TwitchUsername` should be attempted.
 var ErrAuthorizationNeeded = errors.New("Must authenticate Twitch username to use this command")
 
-// SendRemoteCommandCached performs a RPC call on the backend, but caches responses.
+// SendRemoteCommandCached performs a RPC call on the backend, checking for a
+// cached response first.
+//
+// If a cached, but expired, response is found, the existing value is returned
+// and the cache is updated in the background.
 func (backend *backendInfo) SendRemoteCommandCached(remoteCommand, data string, auth AuthInfo) (string, error) {
-	cached, ok := backend.responseCache.Get(getCacheKey(remoteCommand, data))
+	cacheKey := getCacheKey(remoteCommand, data)
+	cached, ok := backend.responseCache.Get(cacheKey)
 	if ok {
 		return cached.(string), nil
 	}
@@ -101,9 +110,21 @@ func (backend *backendInfo) SendRemoteCommandCached(remoteCommand, data string,
 }
 
 // SendRemoteCommand performs a RPC call on the backend by POSTing to `/cmd/$remoteCommand`.
+//
 // The form data is as follows: `clientData` is the JSON in the `data` parameter
-// (should be retrieved from ClientMessage.Arguments), and either `username` or
+// (should be retrieved from ClientMessage.Arguments), `username` is AuthInfo.TwitchUsername,
-// `usernameClaimed` depending on whether AuthInfo.UsernameValidates is true is AuthInfo.TwitchUsername.
+// and `authenticated` is 1 or 0 depending on AuthInfo.UsernameValidated.
+//
+// 401 responses return an ErrAuthorizationNeeded.
+//
+// Non-2xx responses return the response body as an error to the client (application/json
+// responses are sent as-is, non-json are sent as a JSON string).
+//
+// If a 2xx response has the FFZ-Cache header, its value is used as a minimum number of
+// seconds to cache the response for. (Responses may be cached for longer, see
+// SendRemoteCommandCached and the cache implementation.)
+//
+// A successful response updates the Statistics.Health.Backend map.
 func (backend *backendInfo) SendRemoteCommand(remoteCommand, data string, auth AuthInfo) (responseStr string, err error) {
 	destURL := fmt.Sprintf("%s/cmd/%s", backend.baseURL, remoteCommand)
 	healthBucket := fmt.Sprintf("/cmd/%s", remoteCommand)
@@ -166,7 +187,11 @@ func (backend *backendInfo) SendRemoteCommand(remoteCommand, data string, auth A
 			return "", fmt.Errorf("The RPC server returned a non-integer cache duration: %v", err)
 		}
 		duration := time.Duration(durSecs) * time.Second
-		backend.responseCache.Set(getCacheKey(remoteCommand, data), responseStr, duration)
+		backend.responseCache.Set(
+			getCacheKey(remoteCommand, data),
+			responseStr,
+			duration,
+		)
 	}
 
 	now := time.Now().UTC()

socketserver/server/commands.go

@@ -32,14 +32,17 @@ var commandHandlers = map[Command]CommandHandler{
 	"track_follow":  C2STrackFollow,
 	"emoticon_uses": C2SEmoticonUses,
 	"survey":        C2SSurvey,
+}
 
-	"twitch_emote":          C2SHandleBunchedCommand,
+var bunchedCommands = []Command{
-	"get_link":              C2SHandleBunchedCommand,
+	"get_display_name",
-	"get_display_name":      C2SHandleBunchedCommand,
+	"get_emote",
-	"get_emote":             C2SHandleBunchedCommand,
+	"get_emote_set",
-	"get_emote_set":         C2SHandleBunchedCommand,
+	"get_link",
-	"has_logs":              C2SHandleBunchedCommand,
+	"get_itad_plain",
-	"update_follow_buttons": C2SHandleRemoteCommand,
+	"get_itad_prices",
+	"get_name_history",
+	"has_logs",
 }
 
 func setupInterning() {
@@ -71,6 +74,12 @@ func DispatchC2SCommand(conn *websocket.Conn, client *ClientInfo, msg ClientMess
 	handler, ok := commandHandlers[msg.Command]
 	if !ok {
 		handler = C2SHandleRemoteCommand
+
+		for _, v := range bunchedCommands {
+			if msg.Command == v {
+				handler = C2SHandleBunchedCommand
+			}
+		}
 	}
 
 	CommandCounter <- msg.Command

socketserver/server/handlecore.go

@@ -205,7 +205,6 @@ func HTTPHandleRootURL(w http.ResponseWriter, r *http.Request) {
 		updateSysMem()
 
 		if Statistics.SysMemFreeKB > 0 && Statistics.SysMemFreeKB < Configuration.MinMemoryKBytes {
-			atomic.AddUint64(&Statistics.LowMemDroppedConnections, 1)
 			w.WriteHeader(503)
 			fmt.Fprint(w, "error: low memory")
 			return

socketserver/server/stats.go

@@ -38,9 +38,8 @@ type StatsData struct {
 	MemoryInUseKB uint64
 	MemoryRSSKB   uint64
 
-	LowMemDroppedConnections uint64
+	ResponseCacheItems int
-
+	MemPerClientBytes  uint64
-	MemPerClientBytes uint64
 
 	CpuUsagePct float64
 
@@ -84,7 +83,7 @@ func commandCounter() {
 }
 
 // StatsDataVersion is the version of the StatsData struct.
-const StatsDataVersion = 7
+const StatsDataVersion = 8
 const pageSize = 4096
 
 var cpuUsage struct {
@@ -170,6 +169,7 @@ func updatePeriodicStats() {
 
 	{
 		Statistics.Uptime = nowUpdate.Sub(Statistics.StartTime).String()
+		Statistics.ResponseCacheItems = Backend.responseCache.ItemCount()
 	}
 
 	{

socketserver/server/subscriptions.go

@@ -156,8 +156,11 @@ func UnsubscribeSingleChat(client *ClientInfo, channelName string) {
 //   - write lock to SubscriptionInfos
 //   - write lock to ClientInfo
 func UnsubscribeAll(client *ClientInfo) {
-	if StopAcceptingConnections {
+	select {
-		return // no need to remove from a high-contention list when the server is closing
+	case <-StopAcceptingConnectionsCh:
+		// Skip high-contention client removal operations while server shutting down
+		return
+	default:
 	}
 
 	GlobalSubscriptionLock.Lock()