X first cut on scalable parallel auth cache.

blob.go

@@ -16,6 +16,7 @@ import (
 	"net/http/httptest"
 	"regexp"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -23,7 +24,7 @@ import (
 type AuthReply struct {
 	// raw reply from auth backend & preAuthorizeHandler().
 	// recorded so we can replay it from auth cache to each client in full
-	// if access is rejected.	XXX for accepted too? (see WWW-Authenticate in preAuthorizeHandler)
+	// if e.g. access is rejected.
 	RawReply *httptest.ResponseRecorder
 
 	// decoded auth reply
@@ -32,46 +33,106 @@ type AuthReply struct {
 
 // Entry in authorization reply cache
 type AuthCacheEntry struct {
+	// FIXME we need to lock the entry only to "correctly" update Nhit on
+	// read side but we can tolerate some looses in Nhit and update it
+	// without mutex or atomic. Only -race complains...
+	sync.Mutex
+
 	AuthReply
 
 	Tauth int64         // in seconds	XXX needed?
-
-	// how many times this entry was hit when querying auth cache during
-	// the last refresh period.
-	Nhit int64
+	Nhit  int64         // how many times this entry was hit when querying
+                            // auth cache during the last refresh period.
+	ready chan struct{} // closed when entry is ready
 }
 
 // Authorization reply cache
 // {} project -> AuthCacheEntry
 //
 // XXX should be not only project (privateToken etc...)
-var authCache = make(map[string]*AuthCacheEntry)
+type AuthCache struct {
+	mu      sync.RWMutex // guards .cached
+	cached	map[string]*AuthCacheEntry
+}
+
+var authCache = AuthCache{}	// XXX make map ?
+
+// XXX u should not be in args?
+func (c *AuthCache) VerifyDownloadAccess(u *upstream, project string) AuthReply {
+	// first try to read from cache in parallel with other readers
+	c.mu.RLock()
+	auth := c.cached[project]
+	c.mu.RUnlock()
+
+have_entry:
+	// entry in cache - use it
+	if auth != nil {
+//have_entry:
+		<-auth.ready // make sure it is ready
+
+		auth.Lock()
+		auth.Nhit++
+		//log.Printf("authReply for %v cached ago: %v  (hits: %v)",
+		//	project,
+		//	time.Since(time.Unix(auth.Tauth, 0)),
+		//	auth.Nhit)
+		auth.Unlock()
+
+	// no entry - relock the cache in exclusive mode, create empty entry,
+	// and start filling it
+	} else {
+		c.mu.Lock()
+		// another ex-reader could be trying to create this entry
+		// simultaneously with us - recheck
+		auth = c.cached[project]
+		if auth != nil {
+			c.mu.Unlock()
+			goto have_entry
+		}
+
+		// new non-ready entry
+		auth = &AuthCacheEntry{ready: make(chan struct{})}
+		c.cached[project] = auth
+		c.mu.Unlock()
+
+		// this goroutine becomes responsible for quering auth backend
+		auth.AuthReply = askAuthBackend(u, project)
+		auth.Tauth = time.Now().Unix()
+		auth.Nhit = 0
+
+		// broadcast to other goroutines this entry is ready
+		close(auth.ready)
+
+		// launch entry refresher
+		go c.refreshEntry(auth, u, project)
+	}
+
+	return auth.AuthReply
+}
+
 
 // Time period for refreshing / removing unused entires in authCache
 const authCacheRefresh = 10 * time.Second	// XXX 30
 
 // Goroutine to refresh auth cache entry periodically while it is used.
 // if the entry is detected to be not used - remove it from cache and stop refreshing.
-func authRefreshEntry(u *upstream, project string) {
-	// XXX auth := authCache[project]
-	// and then use auth without authCache lookup ?
-
+func (c *AuthCache) refreshEntry(auth *AuthCacheEntry, u *upstream, project string) {
 	for {
 		time.Sleep(authCacheRefresh)
 
-		// XXX lock
-		auth, ok := authCache[project]
-		if !ok { // someone removed the entry from cache - no
-			log.Printf("AUTH refresh - %v entry removed", project)
-			break // no need to further refresh
-		}
+		auth.Lock()
+		nhit := auth.Nhit
+		auth.Unlock()
 
 		// clear cache entry if it is not used
-		log.Printf("AUTH refresh - %v #hit: %v", project, auth.Nhit)
-		if auth.Nhit == 0 { // not used - we can remove and stop refreshing
+		log.Printf("AUTH refresh - %v #hit: %v", project, nhit)
+		if nhit == 0 { // not used - we can remove and stop refreshing
 			log.Printf("AUTH - removing %v", project)
-			// XXX lock
-			delete(authCache, project)
+			// NOTE it is ok even if someone gets this auth in this time window
+			//      and use it for some time
+			c.mu.Lock()
+			delete(c.cached, project)
+			c.mu.Unlock()
 			break
 		}
 
@@ -79,10 +140,11 @@ func authRefreshEntry(u *upstream, project string) {
 		// XXX what if it stucks?
 		authReply := askAuthBackend(u, project)
 
-		// XXX lock
+		auth.Lock()
 		auth.AuthReply = authReply
 		auth.Tauth = time.Now().Unix()
 		auth.Nhit = 0
+		auth.Unlock()
 	}
 }
 
@@ -122,11 +184,12 @@ func askAuthBackend(u *upstream, project string) AuthReply {
 	return authReply
 }
 
+/*
 // Verify that download access is ok or not.
 // first we try to see authCache; if information is not there -> ask auth backend
 // download is ok if AuthReply.RepoPath != ""
 // XXX return -> *AuthReply ?
-func verifyDownloadAccess(w http.ResponseWriter, u *upstream, project string) AuthReply {
+func verifyDownloadAccess(u *upstream, project string) AuthReply {
 	// XXX lock authCache
 	auth, ok := authCache[project]
 	if ok {
@@ -147,6 +210,10 @@ func verifyDownloadAccess(w http.ResponseWriter, u *upstream, project string) Au
 
 	return authReply
 }
+*/
+func verifyDownloadAccess(u *upstream, project string) AuthReply {
+	return authCache.VerifyDownloadAccess(u, project)
+}
 
 // HTTP handler for `.../raw/<ref>/path`
 var projectRe = regexp.MustCompile(`^/[\w\.-]+/[\w\.-]+/`)
@@ -168,7 +235,7 @@ func handleGetBlobRaw(w http.ResponseWriter, r *gitRequest) {
 	refpath = refpath[4:]              // strip 'raw/...'
 
 	// Query download access auth for this project
-	authReply := verifyDownloadAccess(w, r.u, project)
+	authReply := verifyDownloadAccess(r.u, project)
 	if authReply.RepoPath == "" {
 		// access denied - copy auth reply to client in full -
 		// there are HTTP code and other headers / body relevant for