Fix race conditions during in-flight size checks (#3505)

3 files changed, 22 insertions, 13 deletions

diff --git a/weed/filer/reader_cache.go b/weed/filer/reader_cache.go
index b409dbf61..eb2308758 100644
--- a/weed/filer/reader_cache.go
+++ b/weed/filer/reader_cache.go
@@ -186,8 +186,8 @@ func (s *SingleChunkCacher) destroy() {
 	if s.data != nil {
 		mem.Free(s.data)
 		s.data = nil
+		close(s.cacheStartedCh)
 	}
-	close(s.cacheStartedCh)
 }
 
 func (s *SingleChunkCacher) readChunkAt(buf []byte, offset int64) (int, error) {
diff --git a/weed/server/filer_server_handlers.go b/weed/server/filer_server_handlers.go
index 4fb3c9609..153971f6e 100644
--- a/weed/server/filer_server_handlers.go
+++ b/weed/server/filer_server_handlers.go
@@ -2,14 +2,15 @@ package weed_server
 
 import (
 	"errors"
-	"github.com/seaweedfs/seaweedfs/weed/glog"
-	"github.com/seaweedfs/seaweedfs/weed/security"
-	"github.com/seaweedfs/seaweedfs/weed/util"
 	"net/http"
 	"strings"
 	"sync/atomic"
 	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/security"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+
 	"github.com/seaweedfs/seaweedfs/weed/stats"
 )
 
@@ -63,9 +64,11 @@ func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) {
 		// wait until in flight data is less than the limit
 		contentLength := getContentLength(r)
 		fs.inFlightDataLimitCond.L.Lock()
-		for fs.option.ConcurrentUploadLimit != 0 && atomic.LoadInt64(&fs.inFlightDataSize) > fs.option.ConcurrentUploadLimit {
-			glog.V(4).Infof("wait because inflight data %d > %d", fs.inFlightDataSize, fs.option.ConcurrentUploadLimit)
+		inFlightDataSize := atomic.LoadInt64(&fs.inFlightDataSize)
+		for fs.option.ConcurrentUploadLimit != 0 && inFlightDataSize > fs.option.ConcurrentUploadLimit {
+			glog.V(4).Infof("wait because inflight data %d > %d", inFlightDataSize, fs.option.ConcurrentUploadLimit)
 			fs.inFlightDataLimitCond.Wait()
+			inFlightDataSize = atomic.LoadInt64(&fs.inFlightDataSize)
 		}
 		fs.inFlightDataLimitCond.L.Unlock()
 		atomic.AddInt64(&fs.inFlightDataSize, contentLength)
diff --git a/weed/server/volume_server_handlers.go b/weed/server/volume_server_handlers.go
index 92f7cca1d..eec78a7db 100644
--- a/weed/server/volume_server_handlers.go
+++ b/weed/server/volume_server_handlers.go
@@ -45,15 +45,17 @@ func (vs *VolumeServer) privateStoreHandler(w http.ResponseWriter, r *http.Reque
 	case "GET", "HEAD":
 		stats.ReadRequest()
 		vs.inFlightDownloadDataLimitCond.L.Lock()
-		for vs.concurrentDownloadLimit != 0 && atomic.LoadInt64(&vs.inFlightDownloadDataSize) > vs.concurrentDownloadLimit {
+		inFlightDownloadSize := atomic.LoadInt64(&vs.inFlightDownloadDataSize)
+		for vs.concurrentDownloadLimit != 0 && inFlightDownloadSize > vs.concurrentDownloadLimit {
 			select {
 			case <-r.Context().Done():
 				glog.V(4).Infof("request cancelled from %s: %v", r.RemoteAddr, r.Context().Err())
 				return
 			default:
-				glog.V(4).Infof("wait because inflight download data %d > %d", vs.inFlightDownloadDataSize, vs.concurrentDownloadLimit)
+				glog.V(4).Infof("wait because inflight download data %d > %d", inFlightDownloadSize, vs.concurrentDownloadLimit)
 				vs.inFlightDownloadDataLimitCond.Wait()
 			}
+			inFlightDownloadSize = atomic.LoadInt64(&vs.inFlightDownloadDataSize)
 		}
 		vs.inFlightDownloadDataLimitCond.L.Unlock()
 		vs.GetOrHeadHandler(w, r)
@@ -66,17 +68,19 @@ func (vs *VolumeServer) privateStoreHandler(w http.ResponseWriter, r *http.Reque
 		if r.URL.Query().Get("type") != "replicate" && vs.concurrentUploadLimit != 0 {
 			startTime := time.Now()
 			vs.inFlightUploadDataLimitCond.L.Lock()
-			for vs.inFlightUploadDataSize > vs.concurrentUploadLimit {
+			inFlightUploadDataSize := atomic.LoadInt64(&vs.inFlightUploadDataSize)
+			for inFlightUploadDataSize > vs.concurrentUploadLimit {
 				//wait timeout check
 				if startTime.Add(vs.inflightUploadDataTimeout).Before(time.Now()) {
 					vs.inFlightUploadDataLimitCond.L.Unlock()
-					err := fmt.Errorf("reject because inflight upload data %d > %d, and wait timeout", vs.inFlightUploadDataSize, vs.concurrentUploadLimit)
+					err := fmt.Errorf("reject because inflight upload data %d > %d, and wait timeout", inFlightUploadDataSize, vs.concurrentUploadLimit)
 					glog.V(1).Infof("too many requests: %v", err)
 					writeJsonError(w, r, http.StatusTooManyRequests, err)
 					return
 				}
-				glog.V(4).Infof("wait because inflight upload data %d > %d", vs.inFlightUploadDataSize, vs.concurrentUploadLimit)
+				glog.V(4).Infof("wait because inflight upload data %d > %d", inFlightUploadDataSize, vs.concurrentUploadLimit)
 				vs.inFlightUploadDataLimitCond.Wait()
+				inFlightUploadDataSize = atomic.LoadInt64(&vs.inFlightUploadDataSize)
 			}
 			vs.inFlightUploadDataLimitCond.L.Unlock()
 		}
@@ -121,9 +125,11 @@ func (vs *VolumeServer) publicReadOnlyHandler(w http.ResponseWriter, r *http.Req
 	case "GET", "HEAD":
 		stats.ReadRequest()
 		vs.inFlightDownloadDataLimitCond.L.Lock()
-		for vs.concurrentDownloadLimit != 0 && atomic.LoadInt64(&vs.inFlightDownloadDataSize) > vs.concurrentDownloadLimit {
-			glog.V(4).Infof("wait because inflight download data %d > %d", vs.inFlightDownloadDataSize, vs.concurrentDownloadLimit)
+		inFlightDownloadSize := atomic.LoadInt64(&vs.inFlightDownloadDataSize)
+		for vs.concurrentDownloadLimit != 0 && inFlightDownloadSize > vs.concurrentDownloadLimit {
+			glog.V(4).Infof("wait because inflight download data %d > %d", inFlightDownloadSize, vs.concurrentDownloadLimit)
 			vs.inFlightDownloadDataLimitCond.Wait()
+			inFlightDownloadSize = atomic.LoadInt64(&vs.inFlightDownloadDataSize)
 		}
 		vs.inFlightDownloadDataLimitCond.L.Unlock()
 		vs.GetOrHeadHandler(w, r)