From 82b043005ad084226e23b5befb1a278d111b39df Mon Sep 17 00:00:00 2001 From: Harshavardhana Date: Wed, 13 May 2015 23:05:48 -0700 Subject: [PATCH] Avoid one last memory copy --- pkg/api/api_router.go | 2 +- pkg/storage/drivers/memory/memory.go | 54 +++++++++---------- ...{memory_intelligent.go => memory_cache.go} | 24 ++++----- 3 files changed, 38 insertions(+), 42 deletions(-) rename pkg/storage/drivers/memory/{memory_intelligent.go => memory_cache.go} (85%) diff --git a/pkg/api/api_router.go b/pkg/api/api_router.go index 429bbad2e..3bfbf7aa0 100644 --- a/pkg/api/api_router.go +++ b/pkg/api/api_router.go @@ -63,7 +63,7 @@ func HTTPHandler(driver drivers.Driver) http.Handler { // h = quota.BandwidthCap(h, 100*1024*1024, time.Duration(24*time.Hour)) // h = quota.RequestLimit(h, 100, time.Duration(30*time.Minute)) // h = quota.RequestLimit(h, 1000, time.Duration(24*time.Hour)) - h = quota.ConnectionLimit(h, 5) + h = quota.ConnectionLimit(h, 4) h = logging.LogHandler(h) return h } diff --git a/pkg/storage/drivers/memory/memory.go b/pkg/storage/drivers/memory/memory.go index 18c60b1a1..2c20ac361 100644 --- a/pkg/storage/drivers/memory/memory.go +++ b/pkg/storage/drivers/memory/memory.go @@ -43,7 +43,7 @@ import ( type memoryDriver struct { storedBuckets map[string]storedBucket lock *sync.RWMutex - objects *Intelligent + objects *Cache } type storedBucket struct { @@ -69,7 +69,7 @@ func Start(maxSize uint64, expiration time.Duration) (chan<- string, <-chan erro var memory *memoryDriver memory = new(memoryDriver) memory.storedBuckets = make(map[string]storedBucket) - memory.objects = NewIntelligent(maxSize, expiration) + memory.objects = NewCache(maxSize, expiration) memory.lock = new(sync.RWMutex) memory.objects.OnExpired = memory.expiredObject @@ -207,30 +207,6 @@ func (memory *memoryDriver) CreateObject(bucket, key, contentType, expectedMD5Su return md5sum, iodine.New(err, nil) } -// getMD5AndData - this is written as a wrapper to capture md5sum and data in a more memory efficient way -func getMD5AndData(reader io.Reader) ([]byte, []byte, error) { - hash := md5.New() - var data []byte - - var err error - var length int - for err == nil { - byteBuffer := make([]byte, 1024*1024) - length, err = reader.Read(byteBuffer) - // While hash.Write() wouldn't mind a Nil byteBuffer - // It is necessary for us to verify this and break - if length == 0 { - break - } - hash.Write(byteBuffer[0:length]) - data = append(data, byteBuffer[0:length]...) - } - if err != io.EOF { - return nil, nil, err - } - return hash.Sum(nil), data, nil -} - // createObject - PUT object to memory buffer func (memory *memoryDriver) createObject(bucket, key, contentType, expectedMD5Sum string, size int64, data io.Reader) (string, error) { memory.lock.RLock() @@ -267,15 +243,35 @@ func (memory *memoryDriver) createObject(bucket, key, contentType, expectedMD5Su } expectedMD5Sum = hex.EncodeToString(expectedMD5SumBytes) } - md5SumBytes, readBytes, err := getMD5AndData(data) - if err != nil { + + // calculate md5 + hash := md5.New() + var readBytes []byte + + var err error + var length int + for err == nil { + byteBuffer := make([]byte, 1024*1024) + length, err = data.Read(byteBuffer) + // While hash.Write() wouldn't mind a Nil byteBuffer + // It is necessary for us to verify this and break + if length == 0 { + break + } + hash.Write(byteBuffer[0:length]) + readBytes = append(readBytes, byteBuffer[0:length]...) + } + if err != io.EOF { return "", iodine.New(err, nil) } + go debug.FreeOSMemory() + md5SumBytes := hash.Sum(nil) totalLength := len(readBytes) + memory.lock.Lock() memory.objects.Set(objectKey, readBytes) memory.lock.Unlock() - // de-allocating + // setting up for de-allocation readBytes = nil md5Sum := hex.EncodeToString(md5SumBytes) diff --git a/pkg/storage/drivers/memory/memory_intelligent.go b/pkg/storage/drivers/memory/memory_cache.go similarity index 85% rename from pkg/storage/drivers/memory/memory_intelligent.go rename to pkg/storage/drivers/memory/memory_cache.go index ed63123af..e72ae9c46 100644 --- a/pkg/storage/drivers/memory/memory_intelligent.go +++ b/pkg/storage/drivers/memory/memory_cache.go @@ -23,9 +23,9 @@ import ( var zeroExpiration = time.Duration(0) -// Intelligent holds the required variables to compose an in memory cache system +// Cache holds the required variables to compose an in memory cache system // which also provides expiring key mechanism and also maxSize -type Intelligent struct { +type Cache struct { // Mutex is used for handling the concurrent // read/write requests for cache sync.Mutex @@ -62,12 +62,12 @@ type Stats struct { Expired uint64 } -// NewIntelligent creates an inmemory cache +// NewCache creates an inmemory cache // // maxSize is used for expiring objects before we run out of memory // expiration is used for expiration of a key from cache -func NewIntelligent(maxSize uint64, expiration time.Duration) *Intelligent { - return &Intelligent{ +func NewCache(maxSize uint64, expiration time.Duration) *Cache { + return &Cache{ items: map[string]interface{}{}, updatedAt: map[string]time.Time{}, expiration: expiration, @@ -76,7 +76,7 @@ func NewIntelligent(maxSize uint64, expiration time.Duration) *Intelligent { } // Stats get current cache statistics -func (r *Intelligent) Stats() Stats { +func (r *Cache) Stats() Stats { return Stats{ Bytes: r.currentSize, Items: uint64(len(r.items)), @@ -85,7 +85,7 @@ func (r *Intelligent) Stats() Stats { } // ExpireObjects expire objects in go routine -func (r *Intelligent) ExpireObjects(gcInterval time.Duration) { +func (r *Cache) ExpireObjects(gcInterval time.Duration) { r.stopExpireTimer = make(chan struct{}) ticker := time.NewTicker(gcInterval) go func() { @@ -103,7 +103,7 @@ func (r *Intelligent) ExpireObjects(gcInterval time.Duration) { } // Get returns a value of a given key if it exists -func (r *Intelligent) Get(key string) (interface{}, bool) { +func (r *Cache) Get(key string) (interface{}, bool) { r.Lock() defer r.Unlock() value, ok := r.items[key] @@ -115,7 +115,7 @@ func (r *Intelligent) Get(key string) (interface{}, bool) { } // Set will persist a value to the cache -func (r *Intelligent) Set(key string, value interface{}) { +func (r *Cache) Set(key string, value interface{}) { r.Lock() // remove random key if only we reach the maxSize threshold, // if not assume infinite memory @@ -135,7 +135,7 @@ func (r *Intelligent) Set(key string, value interface{}) { } // Expire expires keys which have expired -func (r *Intelligent) Expire() { +func (r *Cache) Expire() { r.Lock() defer r.Unlock() for key := range r.items { @@ -146,7 +146,7 @@ func (r *Intelligent) Expire() { } // Delete deletes a given key if exists -func (r *Intelligent) Delete(key string) { +func (r *Cache) Delete(key string) { if _, ok := r.items[key]; ok { r.currentSize -= uint64(len(r.items[key].([]byte))) delete(r.items, key) @@ -158,7 +158,7 @@ func (r *Intelligent) Delete(key string) { } } -func (r *Intelligent) isValid(key string) bool { +func (r *Cache) isValid(key string) bool { updatedAt, ok := r.updatedAt[key] if !ok { return false