Merge pull request #4 from m4ng0squ4sh/master

Performance improve and small fixes

Author: yosssi

reaper/reaper.go

@@ -8,6 +8,10 @@ import (
 	"github.com/yosssi/boltstore/shared"
 )
 
+//##############//
+//### Public ###//
+//##############//
+
 // Run invokes a reap function as a goroutine.
 func Run(db *bolt.DB, options Options) (chan<- struct{}, <-chan struct{}) {
 	options.setDefault()
@@ -22,66 +26,114 @@ func Quit(quitC chan<- struct{}, doneC <-chan struct{}) {
 	<-doneC
 }
 
+//###############//
+//### Private ###//
+//###############//
+
 func reap(db *bolt.DB, options Options, quitC <-chan struct{}, doneC chan<- struct{}) {
-	var prevKey []byte
-	for {
-		err := db.View(func(tx *bolt.Tx) error {
-			bucket := tx.Bucket(options.BucketName)
-			if bucket == nil {
-				return nil
-			}
+	// Create a new ticker
+	ticker := time.NewTicker(options.CheckInterval)
 
-			c := bucket.Cursor()
+	defer func() {
+		// Stop the ticker
+		ticker.Stop()
+	}()
 
-			var i int
+	var prevKey []byte
 
-			for k, v := c.Seek(prevKey); ; k, v = c.Next() {
-				// If we hit the end of our sessions then
-				// exit and start over next time.
-				if k == nil {
-					prevKey = nil
+	for {
+		select {
+		case <-quitC: // Check if a quit signal is sent.
+			doneC <- struct{}{}
+			return
+		case <-ticker.C: // Check if the ticker fires a signal.
+			// This slice is a buffer to save all expired session keys.
+			expiredSessionKeys := make([][]byte, 0)
+
+			// Start a bolt read transaction.
+			err := db.View(func(tx *bolt.Tx) error {
+				bucket := tx.Bucket(options.BucketName)
+				if bucket == nil {
 					return nil
 				}
 
-				i++
+				c := bucket.Cursor()
 
-				session, err := shared.Session(v)
-				if err != nil {
-					return err
-				}
+				var i int
+				var isExpired bool
+
+				for k, v := c.Seek(prevKey); ; k, v = c.Next() {
+					// If we hit the end of our sessions then
+					// exit and start over next time.
+					if k == nil {
+						prevKey = nil
+						return nil
+					}
 
-				if shared.Expired(session) {
-					err := db.Update(func(txu *bolt.Tx) error {
-						return txu.Bucket(options.BucketName).Delete(k)
-					})
+					i++
+
+					// The flag if the session is expired
+					isExpired = false
+
+					session, err := shared.Session(v)
 					if err != nil {
-						return err
+						// Just remove the session with the invalid session data.
+						// Log the error first.
+						log.Printf("boltstore: removing session from database with invalid value: %v", err)
+						isExpired = true
+					} else if shared.Expired(session) {
+						isExpired = true
 					}
-				}
 
-				if options.BatchSize == i {
-					// Store the current key to the previous key.
-					// Copy the byte slice key, because this data is
-					// not safe outside of this transaction.
-					prevKey = make([]byte, len(k))
-					copy(prevKey, k)
-					return nil
+					if isExpired {
+						// Copy the byte slice key, because this data is
+						// not safe outside of this transaction.
+						temp := make([]byte, len(k))
+						copy(temp, k)
+
+						// Add it to the expired sessios keys slice
+						expiredSessionKeys = append(expiredSessionKeys, temp)
+					}
+
+					if options.BatchSize == i {
+						// Store the current key to the previous key.
+						// Copy the byte slice key, because this data is
+						// not safe outside of this transaction.
+						prevKey = make([]byte, len(k))
+						copy(prevKey, k)
+						return nil
+					}
 				}
+			})
+
+			if err != nil {
+				log.Printf("boltstore: obtain expired sessions error: %v", err)
 			}
-		})
 
-		if err != nil {
-			log.Println(err.Error())
-		}
+			if len(expiredSessionKeys) > 0 {
+				// Remove the expired sessions from the database
+				err = db.Update(func(txu *bolt.Tx) error {
+					// Get the bucket
+					b := txu.Bucket(options.BucketName)
+					if b == nil {
+						return nil
+					}
 
-		// Check if a quit signal is sent.
-		select {
-		case <-quitC:
-			doneC <- struct{}{}
-			return
-		default:
-		}
+					// Remove all expired sessions in the slice
+					for _, key := range expiredSessionKeys {
+						err = b.Delete(key)
+						if err != nil {
+							return err
+						}
+					}
+
+					return nil
+				})
 
-		time.Sleep(options.CheckInterval)
+				if err != nil {
+					log.Printf("boltstore: remove expired sessions error: %v", err)
+				}
+			}
+		}
 	}
 }

reaper/reaper_test.go

@@ -1,10 +1,10 @@
 package reaper
 
 import (
+	"code.google.com/p/gogoprotobuf/proto"
 	"fmt"
 	"testing"
 	"time"
-	"code.google.com/p/gogoprotobuf/proto"
 
 	"github.com/boltdb/bolt"
 	"github.com/yosssi/boltstore/shared"
@@ -62,7 +62,12 @@ func Test_reap(t *testing.T) {
 
 	// When shared.Session returns an error
 	err = db.Update(func(tx *bolt.Tx) error {
-		return tx.Bucket(bucketName).Put([]byte("test"), []byte("value"))
+		session := shared.NewSession([]byte{}, -1)
+		data, err := proto.Marshal(session)
+		if err != nil {
+			return err
+		}
+		return tx.Bucket(bucketName).Put([]byte("test"), data)
 	})
 	if err != nil {
 		t.Error(err.Error())

shared/consts.go

@@ -15,6 +15,6 @@ const (
 
 // Defaults for reaper.Options
 const (
-	DefaultBatchSize     = 10
-	DefaultCheckInterval = time.Second
+	DefaultBatchSize     = 100
+	DefaultCheckInterval = time.Minute
 )