Updated indexer

Author: djthorpe

etc/server.yaml

@@ -40,7 +40,7 @@ sqlite3:
 
   # Set trace to true to enable the ability to profile queries. Profiling information
   # can be displayed through the API.
-  trace: true
+  trace: false
 
   # Set max number of connections that can be simultaneously opened
   max: 100

pkg/indexer/queue.go

@@ -17,13 +17,13 @@ type Queue struct {
 }
 
 type QueueEvent struct {
-	Event
+	EventType
 	Name string
 	Path string
 	Info fs.FileInfo
 }
 
-type Event uint
+type EventType uint
 
 ///////////////////////////////////////////////////////////////////////////////
 // GLOBALS
@@ -33,7 +33,7 @@ const (
 )
 
 const (
-	EventNone Event = iota
+	EventNone EventType = iota
 	EventAdd
 	EventRemove
 	EventReindexStarted
@@ -151,7 +151,7 @@ func (q *Queue) Count() int {
 ///////////////////////////////////////////////////////////////////////////////
 // PRIVATE METHODS
 
-func (q *Queue) add(e Event, name, path string, info fs.FileInfo) {
+func (q *Queue) add(e EventType, name, path string, info fs.FileInfo) {
 	q.RWMutex.Lock()
 	defer q.RWMutex.Unlock()
 	// This assumes the key does not exist

pkg/indexer/schema.go

@@ -93,12 +93,12 @@ func CreateSchema(ctx context.Context, conn SQConnection, schema string, tokeniz
 			return err
 		}
 		if _, err := txn.Query(N(searchTriggerDeleteName).WithSchema(schema).CreateTrigger(filesTableName,
-			Q("INSERT INTO ", searchTableName, " (", searchTableName, ",rowid, name, parent, filename) VALUES ('delete', old.rowid, old.name, old.parent, old.filename)"),
+			Q("INSERT INTO ", searchTableName, " (", searchTableName, ", rowid, name, parent, filename) VALUES ('delete', old.rowid, old.name, old.parent, old.filename)"),
 		).After().Delete().IfNotExists()); err != nil {
 			return err
 		}
 		if _, err := txn.Query(N(searchTriggerUpdateName).WithSchema(schema).CreateTrigger(filesTableName,
-			Q("INSERT INTO ", searchTableName, " (", searchTableName, ",rowid, name, parent, filename) VALUES ('delete', old.rowid, old.name, old.parent, old.filename)"),
+			Q("INSERT INTO ", searchTableName, " (", searchTableName, ", rowid, name, parent, filename) VALUES ('delete', old.rowid, old.name, old.parent, old.filename)"),
 			Q("INSERT INTO ", searchTableName, " (rowid, name, parent, filename) VALUES (new.rowid, new.name, new.parent, new.filename)"),
 		).After().Update().IfNotExists()); err != nil {
 			return err

pkg/indexer/store.go

@@ -12,17 +12,19 @@ import (
 
 	// Import namepaces
 	. "github.com/djthorpe/go-errors"
+	. "github.com/mutablelogic/go-server"
 	. "github.com/mutablelogic/go-sqlite"
 )
 
 ///////////////////////////////////////////////////////////////////////////////
 // TYPES
 
 type Store struct {
-	pool    SQPool
-	queue   *Queue
-	workers uint
-	schema  string
+	pool     SQPool
+	queue    *Queue
+	renderer Renderer
+	workers  uint
+	schema   string
 }
 
 type operation struct {
@@ -42,12 +44,13 @@ var (
 // LIFECYCLE
 
 // Create a new store object
-func NewStore(pool SQPool, schema string, queue *Queue, workers uint) *Store {
+func NewStore(pool SQPool, schema string, queue *Queue, r Renderer, workers uint) *Store {
 	s := new(Store)
 	s.pool = pool
 	s.queue = queue
+	s.renderer = r
 
-	// Create workers - use cpus
+	// Create workers - use double number of cores by default
 	if workers == 0 {
 		s.workers = defaultWorkers
 	} else {
@@ -59,6 +62,11 @@ func NewStore(pool SQPool, schema string, queue *Queue, workers uint) *Store {
 		pool.SetMax(int(s.workers))
 	}
 
+	// Reduce number of workers if the pool max does not match
+	if pool.Max() > 0 {
+		s.workers = uintMin(s.workers, uint(pool.Max()))
+	}
+
 	// Get a database connection
 	conn := pool.Get()
 	if conn == nil {
@@ -105,6 +113,18 @@ func (s *Store) Run(ctx context.Context, errs chan<- error) error {
 	return result
 }
 
+///////////////////////////////////////////////////////////////////////////////
+// STRINGIFY
+
+func (s *Store) String() string {
+	str := "<store"
+	str += fmt.Sprint(" workers=", s.workers)
+	if s.schema != "" {
+		str += fmt.Sprintf(" schema=%q", s.schema)
+	}
+	return str + ">"
+}
+
 ///////////////////////////////////////////////////////////////////////////////
 // PUBLIC METHODS
 
@@ -174,7 +194,7 @@ func (s *Store) worker(ctx context.Context, id uint, errs chan<- error) error {
 }
 
 func (s *Store) process(evt *QueueEvent) operation {
-	switch evt.Event {
+	switch evt.EventType {
 	case EventAdd:
 		if replace, args := Replace(s.schema, evt); replace != nil {
 			return operation{replace, args}

pkg/indexer/util.go

@@ -33,3 +33,10 @@ func boolToInt64(v bool) int64 {
 	}
 	return 0
 }
+
+func uintMin(a, b uint) uint {
+	if a < b {
+		return a
+	}
+	return b
+}

plugin/indexer/main.go

@@ -83,11 +83,12 @@ func New(ctx context.Context, provider Provider) Plugin {
 	}
 
 	// Create a queue, indexers and stores
+	// TODO: Add a renderer interface for the store
 	q := indexer.NewQueueWithCapacity(defaultCapacity)
 	if q == nil {
 		provider.Print(ctx, "unable to create queue")
 		return nil
-	} else if store := indexer.NewStore(p.pool, schema, q, 0); store == nil {
+	} else if store := indexer.NewStore(p.pool, schema, q, nil, 0); store == nil {
 		provider.Print(ctx, "unable to create store")
 		return nil
 	} else {
@@ -102,8 +103,6 @@ func New(ctx context.Context, provider Provider) Plugin {
 		}
 	}
 
-	fmt.Println("got", p)
-
 	// Return success
 	return p
 }
@@ -113,8 +112,14 @@ func New(ctx context.Context, provider Provider) Plugin {
 
 func (p *plugin) String() string {
 	str := "<indexer"
+	if p.store != nil {
+		str += fmt.Sprint(" store=", p.store)
+	}
+	for name, index := range p.index {
+		str += fmt.Sprintf(" %q=%v", name, index)
+	}
 	if p.pool != nil {
-		str += fmt.Sprint(" ", p.pool)
+		str += fmt.Sprint(" pool=", p.pool)
 	}
 	return str + ">"
 }