Fix deadlock around aggregating data + server max concurrent file read limiter

3 files changed, 7 insertions, 29 deletions

diff --git a/internal/io/fs/catfile.go b/internal/io/fs/catfile.go
index 01c15ba..e4676f3 100644
--- a/internal/io/fs/catfile.go
+++ b/internal/io/fs/catfile.go
@@ -6,9 +6,7 @@ type CatFile struct {
 }
 
 // NewCatFile returns a new file catter.
-func NewCatFile(filePath string, globID string, serverMessages chan<- string,
-	limiter chan struct{}) CatFile {
-
+func NewCatFile(filePath string, globID string, serverMessages chan<- string) CatFile {
 	return CatFile{
 		readFile: readFile{
 			filePath:       filePath,
@@ -17,7 +15,6 @@ func NewCatFile(filePath string, globID string, serverMessages chan<- string,
 			retry:          false,
 			canSkipLines:   false,
 			seekEOF:        false,
-			limiter:        limiter,
 		},
 	}
 }
diff --git a/internal/io/fs/readfile.go b/internal/io/fs/readfile.go
index 28cbe58..5815aa3 100644
--- a/internal/io/fs/readfile.go
+++ b/internal/io/fs/readfile.go
@@ -38,7 +38,6 @@ type readFile struct {
 	canSkipLines bool
 	// Seek to the EOF before processing file?
 	seekEOF bool
-	limiter chan struct{}
 }
 
 // String returns the string representation of the readFile
@@ -66,25 +65,7 @@ func (f readFile) Retry() bool {
 func (f readFile) Start(ctx context.Context, ltx lcontext.LContext,
 	lines chan<- line.Line, re regex.Regex) error {
 
-	dlog.Common.Debug("readFile", f)
-	defer func() {
-		select {
-		case <-f.limiter:
-		default:
-		}
-	}()
-
-	select {
-	case f.limiter <- struct{}{}:
-	default:
-		select {
-		case f.serverMessages <- dlog.Common.Warn(f.filePath, f.globID,
-			"Server limit reached. Queuing file..."):
-		case <-ctx.Done():
-			return nil
-		}
-		f.limiter <- struct{}{}
-	}
+	dlog.Common.Trace("readFile", f)
 
 	fd, err := os.Open(f.filePath)
 	if err != nil {
@@ -156,7 +137,9 @@ func (f readFile) makeReader(fd *os.File) (reader *bufio.Reader, err error) {
 	return
 }
 
-func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan *bytes.Buffer, truncate <-chan struct{}) error {
+func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan *bytes.Buffer,
+	truncate <-chan struct{}) error {
+
 	var offset uint64
 	reader, err := f.makeReader(fd)
 	if err != nil {
@@ -250,6 +233,7 @@ func (f readFile) filterWithoutLContext(ctx context.Context, rawLines <-chan *by
 				return
 			}
 			if filteredLine, ok := f.transmittable(line, len(lines), cap(lines), re); ok {
+				//dlog.Common.Trace("TODO", "lines", lines, len(lines), cap(lines))
 				select {
 				case lines <- filteredLine:
 				case <-ctx.Done():
diff --git a/internal/io/fs/tailfile.go b/internal/io/fs/tailfile.go
index b03b45d..7a40ac4 100644
--- a/internal/io/fs/tailfile.go
+++ b/internal/io/fs/tailfile.go
@@ -6,9 +6,7 @@ type TailFile struct {
 }
 
 // NewTailFile returns a new file tailer.
-func NewTailFile(filePath string, globID string, serverMessages chan<- string,
-	limiter chan struct{}) TailFile {
-
+func NewTailFile(filePath string, globID string, serverMessages chan<- string) TailFile {
 	return TailFile{
 		readFile: readFile{
 			filePath:       filePath,
@@ -17,7 +15,6 @@ func NewTailFile(filePath string, globID string, serverMessages chan<- string,
 			retry:          true,
 			canSkipLines:   true,
 			seekEOF:        true,
-			limiter:        limiter,
 		},
 	}
 }