Merge pull request #596 from harshavardhana/pr_out_avoid_one_last_memory_copy_wip

This commit is contained in:
Harshavardhana 2015-05-14 11:29:59 -07:00
commit 4f7ae8af92
3 changed files with 38 additions and 42 deletions

View File

@ -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
}

View File

@ -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)

View File

@ -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