ston1th/docstore
1
1
Fork 0
Code Issues 3 Releases Activity

better jobs queue

Browse source
This commit is contained in:
ston1th 2022-10-06 18:48:41 +02:00
commit cca34832b7
2 changed files with 23 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

5
pkg/server/handler.go
View file

@ -166,7 +166,7 @@ func indexHandler(ctx *Context) {
if fpath, ok = hasTrimPrefix(path, core.IndexPrefix); ok { if fpath, ok = hasTrimPrefix(path, core.IndexPrefix); ok {
log = log.WithValues("file", fpath) log = log.WithValues("file", fpath)
log.V(2).Info("started indexing file") log.V(2).Info("started indexing file")
err = ctx.Srv.SQ.Add(fpath) err = ctx.Srv.SQ.Add(fpath, nil)
if err != nil { if err != nil {
log.Error(err, "error indexing file") log.Error(err, "error indexing file")
} }
@ -214,12 +214,11 @@ func indexHandler(ctx *Context) {
ctx.Srv.DB.IndexMutex.Set(uint64(len(files))) ctx.Srv.DB.IndexMutex.Set(uint64(len(files)))
for _, f := range files { for _, f := range files {
if !ctx.Srv.DB.IsIndexed(f) { if !ctx.Srv.DB.IsIndexed(f) {
err := ctx.Srv.SQ.Add(f) err := ctx.Srv.SQ.Add(f, ctx.Srv.DB.IndexMutex.Inc)
if err != nil { if err != nil {
log.Error(err, "error indexing file", "file", f) log.Error(err, "error indexing file", "file", f)
} }
} }
ctx.Srv.DB.IndexMutex.Inc()
} }
}() }()
} else { } else {

33
pkg/server/queue.go
View file

@ -5,20 +5,25 @@ package server
import "context" import "context"
type ScanQueue struct { type ScanQueue struct {
ctx context.Context ctx context.Context
srv *HTTPServer srv *HTTPServer
pchan chan string jobs chan job
}
type job struct {
path string
callback func()
} }
func NewScanQueue(ctx context.Context, srv *HTTPServer, buffer int) *ScanQueue { func NewScanQueue(ctx context.Context, srv *HTTPServer, buffer int) *ScanQueue {
return &ScanQueue{ return &ScanQueue{
ctx: ctx, ctx: ctx,
srv: srv, srv: srv,
pchan: make(chan string, buffer), jobs: make(chan job, buffer),
} }
} }
func (sq *ScanQueue) Add(path string) (err error) { func (sq *ScanQueue) Add(path string, callback func()) (err error) {
err = sq.srv.FS.AddScan(path) err = sq.srv.FS.AddScan(path)
if err != nil { if err != nil {
return return
@ -27,7 +32,7 @@ func (sq *ScanQueue) Add(path string) (err error) {
select { select {
case <-sq.ctx.Done(): case <-sq.ctx.Done():
return return
case sq.pchan <- path: case sq.jobs <- job{path, callback}:
} }
}() }()
return return
@ -37,16 +42,20 @@ func (sq *ScanQueue) Scan() {
for { for {
select { select {
case <-sq.ctx.Done(): case <-sq.ctx.Done():
//close(sq.pchan) //close(sq.jobs)
return return
case p := <-sq.pchan: case j := <-sq.jobs:
sq.scanFile(p) sq.scanFile(j)
} }
} }
} }
func (sq *ScanQueue) scanFile(path string) { func (sq *ScanQueue) scanFile(j job) {
srv := sq.srv srv := sq.srv
path := j.path
if j.callback != nil {
defer j.callback()
}
defer srv.FS.RemoveScan(path) defer srv.FS.RemoveScan(path)
log := srv.Log.WithValues("file", path) log := srv.Log.WithValues("file", path)
file, txt, err := srv.Scanner.Scan(path) file, txt, err := srv.Scanner.Scan(path)