Merge pull request #636 from harshavardhana/pr_out_an_attempt_to_make_connection_limit_bufferred_channel_wip

main.go

@@ -46,9 +46,9 @@ var flags = []cli.Flag{
 		},
 	*/
 	cli.IntFlag{
-		Name:  "conn-limit",
+		Name:  "ratelimit",
 		Value: 16,
-		Usage: "Set per IP connection limit quota for server: [DEFAULT: 16]",
+		Usage: "Limit for total concurrent requests: [DEFAULT: 16]",
 	},
 	cli.StringFlag{
 		Name:  "cert",
@@ -84,11 +84,11 @@ func getAPIServerConfig(c *cli.Context) httpserver.Config {
 	}
 	tls := (certFile != "" && keyFile != "")
 	return httpserver.Config{
-		Address:         c.GlobalString("address"),
+		Address:   c.GlobalString("address"),
-		TLS:             tls,
+		TLS:       tls,
-		CertFile:        certFile,
+		CertFile:  certFile,
-		KeyFile:         keyFile,
+		KeyFile:   keyFile,
-		ConnectionLimit: c.GlobalInt("conn-limit"),
+		RateLimit: c.GlobalInt("ratelimit"),
 	}
 }
 

pkg/api/api_router.go

@@ -31,8 +31,8 @@ type minioAPI struct {
 
 // Config api configurable parameters
 type Config struct {
-	ConnectionLimit int
+	RateLimit int
-	driver          drivers.Driver
+	driver    drivers.Driver
 }
 
 // GetDriver - get a an existing set driver
@@ -69,11 +69,12 @@ func HTTPHandler(config Config) http.Handler {
 	handler = timeValidityHandler(handler)
 	handler = ignoreResourcesHandler(handler)
 	handler = validateAuthHeaderHandler(handler)
-	//	h = quota.BandwidthCap(h, 25*1024*1024, time.Duration(30*time.Minute))
+	//	handler = quota.BandwidthCap(h, 25*1024*1024, time.Duration(30*time.Minute))
-	//	h = quota.BandwidthCap(h, 100*1024*1024, time.Duration(24*time.Hour))
+	//	handler = quota.BandwidthCap(h, 100*1024*1024, time.Duration(24*time.Hour))
-	//	h = quota.RequestLimit(h, 100, time.Duration(30*time.Minute))
+	//	handler = quota.RequestLimit(h, 100, time.Duration(30*time.Minute))
-	//	h = quota.RequestLimit(h, 1000, time.Duration(24*time.Hour))
+	//	handler = quota.RequestLimit(h, 1000, time.Duration(24*time.Hour))
-	handler = quota.ConnectionLimit(handler, config.ConnectionLimit)
+	//      handler = quota.ConnectionLimit(handler, config.ConnectionLimit)
+	handler = quota.RateLimit(handler, config.RateLimit)
 	handler = logging.LogHandler(handler)
 	return handler
 }

pkg/api/api_test.go

@@ -124,7 +124,7 @@ func setDummyAuthHeader(req *http.Request) {
 }
 
 func setConfig(driver drivers.Driver) Config {
-	conf := Config{ConnectionLimit: 16}
+	conf := Config{RateLimit: 16}
 	conf.SetDriver(driver)
 	return conf
 }

pkg/api/quota/conn_limit.go

@@ -16,74 +16,35 @@
 
 package quota
 
-import (
+import "net/http"
-	"net"
-	"net/http"
-	"sync"
-
-	"github.com/minio/minio/pkg/utils/log"
-)
 
 // requestLimitHandler
 type connLimit struct {
-	sync.RWMutex
+	handler         http.Handler
-	handler     http.Handler
+	connectionQueue chan bool
-	connections map[uint32]int
-	limit       int
 }
 
-func (c *connLimit) IsLimitExceeded(ip uint32) bool {
+func (c *connLimit) Add() {
-	if c.connections[ip] >= c.limit {
+	c.connectionQueue <- true
-		return true
-	}
-	return false
-}
-
-func (c *connLimit) GetUsed(ip uint32) int {
-	return c.connections[ip]
-}
-
-func (c *connLimit) Add(ip uint32) {
-	c.Lock()
-	defer c.Unlock()
-	count := c.connections[ip]
-	count = count + 1
-	c.connections[ip] = count
 	return
 }
 
-func (c *connLimit) Remove(ip uint32) {
+func (c *connLimit) Remove() {
-	c.Lock()
+	<-c.connectionQueue
-	defer c.Unlock()
+	return
-	count, _ := c.connections[ip]
-	count = count - 1
-	if count <= 0 {
-		delete(c.connections, ip)
-		return
-	}
-	c.connections[ip] = count
 }
 
 // ServeHTTP is an http.Handler ServeHTTP method
 func (c *connLimit) ServeHTTP(w http.ResponseWriter, req *http.Request) {
-	host, _, _ := net.SplitHostPort(req.RemoteAddr)
+	c.Add()
-	longIP := longIP{net.ParseIP(host)}.IptoUint32()
-	if c.IsLimitExceeded(longIP) {
-		hosts, _ := net.LookupAddr(uint32ToIP(longIP).String())
-		log.Debug.Printf("Connection limit reached - Host: %s, Total Connections: %d\n", hosts, c.GetUsed(longIP))
-		writeErrorResponse(w, req, ConnectionLimitExceeded, req.URL.Path)
-		return
-	}
-	c.Add(longIP)
-	defer c.Remove(longIP)
 	c.handler.ServeHTTP(w, req)
+	c.Remove()
 }
 
 // ConnectionLimit limits the number of concurrent connections
 func ConnectionLimit(h http.Handler, limit int) http.Handler {
 	return &connLimit{
-		handler:     h,
+		handler:         h,
-		connections: make(map[uint32]int),
+		connectionQueue: make(chan bool, limit),
-		limit:       limit,
 	}
 }

pkg/api/quota/rate_limiter.go

@@ -0,0 +1,50 @@
+/*
+ * Minimalist Object Storage, (C) 2015 Minio, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package quota
+
+import "net/http"
+
+// rateLimit
+type rateLimit struct {
+	handler   http.Handler
+	rateQueue chan bool
+}
+
+func (c *rateLimit) Add() {
+	c.rateQueue <- true // fill in the queue
+	return
+}
+
+func (c *rateLimit) Remove() {
+	<-c.rateQueue // invalidate the queue, after the request is served
+	return
+}
+
+// ServeHTTP is an http.Handler ServeHTTP method
+func (c *rateLimit) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	c.Add()                     // add
+	c.handler.ServeHTTP(w, req) // serve
+	c.Remove()                  // remove
+}
+
+// RateLimit limits the number of concurrent http requests
+func RateLimit(handle http.Handler, limit int) http.Handler {
+	return &rateLimit{
+		handler:   handle,
+		rateQueue: make(chan bool, limit),
+	}
+}

pkg/server/httpserver/httpserver.go

@@ -25,11 +25,11 @@ import (
 
 // Config - http server config
 type Config struct {
-	Address         string
+	Address   string
-	TLS             bool
+	TLS       bool
-	CertFile        string
+	CertFile  string
-	KeyFile         string
+	KeyFile   string
-	ConnectionLimit int
+	RateLimit int
 }
 
 // Server - http server related

pkg/server/server.go

@@ -43,7 +43,7 @@ type MemoryFactory struct {
 func (f MemoryFactory) GetStartServerFunc() StartServerFunc {
 	return func() (chan<- string, <-chan error) {
 		_, _, driver := memory.Start(f.MaxMemory, f.Expiration)
-		conf := api.Config{ConnectionLimit: f.ConnectionLimit}
+		conf := api.Config{RateLimit: f.RateLimit}
 		conf.SetDriver(driver)
 		ctrl, status, _ := httpserver.Start(api.HTTPHandler(conf), f.Config)
 		return ctrl, status
@@ -60,7 +60,7 @@ type FilesystemFactory struct {
 func (f FilesystemFactory) GetStartServerFunc() StartServerFunc {
 	return func() (chan<- string, <-chan error) {
 		_, _, driver := fs.Start(f.Path)
-		conf := api.Config{ConnectionLimit: f.ConnectionLimit}
+		conf := api.Config{RateLimit: f.RateLimit}
 		conf.SetDriver(driver)
 		ctrl, status, _ := httpserver.Start(api.HTTPHandler(conf), f.Config)
 		return ctrl, status
@@ -90,7 +90,7 @@ type DonutFactory struct {
 func (f DonutFactory) GetStartServerFunc() StartServerFunc {
 	return func() (chan<- string, <-chan error) {
 		_, _, driver := donut.Start(f.Paths)
-		conf := api.Config{ConnectionLimit: f.ConnectionLimit}
+		conf := api.Config{RateLimit: f.RateLimit}
 		conf.SetDriver(driver)
 		ctrl, status, _ := httpserver.Start(api.HTTPHandler(conf), f.Config)
 		return ctrl, status