Updated caching

Author: djthorpe

pkg/sqlite3/cache.go

@@ -1,17 +1,70 @@
 package sqlite3
 
+import (
+	"sync"
+
+	// Namespace Imports
+	. "github.com/djthorpe/go-errors"
+	sqlite3 "github.com/djthorpe/go-sqlite/sys/sqlite3"
+	multierror "github.com/hashicorp/go-multierror"
+)
+
 ////////////////////////////////////////////////////////////////////////////////
 // TYPES
 
 // PoolCache caches prepared statements and profiling information for
 // statements so it's possible to see slow queries, etc.
-type PoolCache struct {
+type PoolCache struct{}
+
+type ConnCache struct {
+	sync.Mutex
+	sync.Map
 }
 
+////////////////////////////////////////////////////////////////////////////////
+// GLOBALS
+
+const (
+	defaultCapacity = 100
+)
+
 ////////////////////////////////////////////////////////////////////////////////
 // PUBLIC METHODS
 
-// Return a prepared statement from the cache
-func (cache *PoolCache) Prepare(q string) (*Results, error) {
+// Return a prepared statement from the cache, or prepare a new statement
+// and put it in the cache before returning
+func (cache *ConnCache) Prepare(conn *sqlite3.ConnEx, q string) (*Results, error) {
+	if conn == nil {
+		return nil, ErrInternalAppError
+	}
+	st, _ := cache.Map.Load(q)
+	if st == nil {
+		// Prepare a statement and store in cache
+		var err error
+		cache.Mutex.Lock()
+		defer cache.Mutex.Unlock()
+		if st, err = conn.Prepare(q); err != nil {
+			return nil, err
+		} else {
+			cache.Map.Store(q, st)
+		}
+	} else {
+		// Increment counter by one
+		st.(*sqlite3.StatementEx).Inc(1)
+	}
+	return NewResults(st.(*sqlite3.StatementEx)), nil
+}
+
+// Close all conn cache prepared statements
+func (cache *ConnCache) Close() error {
+	var result error
+	cache.Map.Range(func(key, value interface{}) bool {
+		if err := value.(*sqlite3.StatementEx).Close(); err != nil {
+			result = multierror.Append(result, err)
+		}
+		return true
+	})
 
+	// Return any errors
+	return result
 }

pkg/sqlite3/cache_test.go

@@ -0,0 +1,54 @@
+package sqlite3_test
+
+import (
+	"context"
+	"math/rand"
+	"sync"
+	"testing"
+
+	// Module imports
+	sqlite3 "github.com/djthorpe/go-sqlite/sys/sqlite3"
+
+	// Namespace Imports
+	. "github.com/djthorpe/go-sqlite"
+	. "github.com/djthorpe/go-sqlite/pkg/lang"
+	. "github.com/djthorpe/go-sqlite/pkg/sqlite3"
+)
+
+func Test_Cache_001(t *testing.T) {
+	// Create a connection
+	conn, err := OpenPath(":memory:", sqlite3.DefaultFlags)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	// Perform caching in a transaction
+	conn.Do(context.Background(), 0, func(txn SQTransaction) error {
+		// Read values from the cache
+		var wg sync.WaitGroup
+		for i := 0; i < 99; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				n := rand.Uint32() % 5
+				if r, err := txn.Query(Q("SELECT ", n)); err != nil {
+					t.Error(err)
+				} else {
+					t.Log(r)
+				}
+			}()
+		}
+
+		// Wait for all goroutines to complete
+		wg.Wait()
+
+		// Return success
+		return nil
+	})
+
+	// Close cache, release resources
+	if err := conn.Close(); err != nil {
+		t.Fatal(err)
+	}
+}

pkg/sqlite3/conn.go

@@ -22,6 +22,8 @@ import (
 type Conn struct {
 	sync.Mutex
 	*sqlite3.ConnEx
+	ConnCache
+
 	c   chan struct{}
 	ctx context.Context
 }
@@ -150,23 +152,21 @@ func (conn *Conn) Do(ctx context.Context, flag SQTxnFlag, fn func(SQTransaction)
 }
 
 // Execute SQL statement and invoke a callback for each row of results which may return true to abort
-func (txn *Txn) Query(st SQStatement, v ...interface{}) (SQResult, error) {
+func (txn *Txn) Query(st SQStatement, v ...interface{}) (SQResults, error) {
 	if st == nil {
 		return nil, ErrBadParameter.With("Query")
 	}
 
-	// TODO: Get prepared version of query from the cache
-	stx, err := txn.Conn.Prepare(st.Query())
+	// Get a results object
+	r, err := txn.ConnCache.Prepare(txn.Conn.ConnEx, st.Query())
 	if err != nil {
 		return nil, err
 	}
 
-	// Create a results object and return it
-	results, err := stx.Exec(0, v...)
-	if err != nil {
-		stx.Close()
+	// Execute first query
+	if err := r.NextQuery(v...); err != nil {
 		return nil, err
+	} else {
+		return r, nil
 	}
-
-	return NewResults(results), nil
 }

pkg/sqlite3/pool.go

@@ -40,7 +40,6 @@ type Pool struct {
 	sync.WaitGroup
 	sync.Pool
 	PoolConfig
-	PoolCache
 
 	errs   chan<- error
 	ctx    context.Context
@@ -301,7 +300,11 @@ func (p *Pool) put(conn *Conn) {
 	n := atomic.AddInt32(&p.n, -1)
 	if n < p.Max() {
 		p.Pool.Put(conn)
-	} else if err := conn.Close(); err != nil {
+		return
+	}
+
+	// Close connection and remove from cache
+	if err := conn.Close(); err != nil {
 		p.err(err)
 	}
 }

pkg/sqlite3/result.go

@@ -0,0 +1,63 @@
+package sqlite3
+
+import (
+	"errors"
+	"fmt"
+	"io"
+
+	// Packages
+	sqlite3 "github.com/djthorpe/go-sqlite/sys/sqlite3"
+	// Namespace imports
+	//. "github.com/djthorpe/go-errors"
+)
+
+////////////////////////////////////////////////////////////////////////////////
+// TYPES
+
+// PoolConfig is the starting configuration for a pool
+type Results struct {
+	st      *sqlite3.StatementEx
+	results *sqlite3.Results
+	n       uint // next statement to execute
+}
+
+////////////////////////////////////////////////////////////////////////////////
+// LIFECYCLE
+
+func NewResults(st *sqlite3.StatementEx) *Results {
+	r := new(Results)
+	r.st, r.n = st, 0
+	return r
+}
+
+////////////////////////////////////////////////////////////////////////////////
+// STRINGIFY
+
+func (r *Results) String() string {
+	str := "<results"
+	if n := r.st.Count(); n > 0 {
+		str += fmt.Sprint(" cached (", n, ")")
+	}
+	str += fmt.Sprint(" ", r.st)
+	return str + ">"
+}
+
+////////////////////////////////////////////////////////////////////////////////
+// PUBLIC METHODS
+
+// NextQuery will execute the next query in the statement, return io.EOF if there
+// are no more statements. In order to read the rows, repeatedly read the rows
+// using the Next function.
+func (r *Results) NextQuery(v ...interface{}) error {
+	if results, err := r.st.Exec(r.n, v...); errors.Is(err, sqlite3.SQLITE_DONE) {
+		return io.EOF
+	} else if err != nil {
+		return err
+	} else {
+		r.results = results
+		r.n++
+		// TODO: Set LastInsertId, Changes, Columns, etc.
+		fmt.Println(r.results)
+		return nil
+	}
+}

pkg/sqlite3/results.go

@@ -88,28 +88,25 @@ type SQConnection interface {
 // SQTransaction is an sqlite transaction
 type SQTransaction interface {
 	// Query and return a set of results
-	Query(SQStatement, ...interface{}) (SQResult, error)
+	Query(SQStatement, ...interface{}) (SQResults, error)
 }
 
-// SQResult increments over returned rows from a query
-type SQResult interface {
+// SQResults increments over returned rows from a query
+type SQResults interface {
 	// Return next row, returns nil when all rows consumed
-	Next() []interface{}
+	//Next() []interface{}
 
 	// Return next map of values, or nil if no more rows
-	NextMap() map[string]interface{}
+	//NextMap() map[string]interface{}
 
 	// NextQuery executes the next query or returns io.EOF
 	NextQuery(...interface{}) error
 
-	// Close the rows, and free up any resources
-	Close() error
-
 	// Return Last RowID inserted of last statement
-	LastInsertId() int64
+	//LastInsertId() int64
 
 	// Return number of changes made of last statement
-	RowsAffected() int64
+	//RowsAffected() int64
 }
 
 // SQAuth is an interface for authenticating an action