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

3 files changed, 60 insertions, 19 deletions

diff --git a/integrationtests/commandutils.go b/integrationtests/commandutils.go
index 23b9c37..d5b5987 100644
--- a/integrationtests/commandutils.go
+++ b/integrationtests/commandutils.go
@@ -12,9 +12,6 @@ import (
 	"time"
 )
 
-// The exit code and the Go error of the command terminated.
-type exitPromise func() (int, error)
-
 func runCommand(ctx context.Context, t *testing.T, stdoutFile, cmdStr string,
 	args ...string) (int, error) {
 
@@ -51,7 +48,7 @@ func runCommandRetry(ctx context.Context, t *testing.T, retries int, stdoutFile,
 }
 
 func startCommand(ctx context.Context, t *testing.T, cmdStr string,
-	args ...string) (<-chan string, <-chan string, exitPromise, error) {
+	args ...string) (<-chan string, <-chan string, <-chan error, error) {
 
 	stdoutCh := make(chan string)
 	stderrCh := make(chan string)
@@ -90,10 +87,33 @@ func startCommand(ctx context.Context, t *testing.T, cmdStr string,
 		close(stderrCh)
 	}()
 
-	return stdoutCh, stderrCh, func() (int, error) {
-		err := cmd.Wait()
-		return exitCodeFromError(err), err
-	}, nil
+	cmdErrCh := make(chan error)
+	go func() {
+		cmdErrCh <- cmd.Wait()
+	}()
+
+	return stdoutCh, stderrCh, cmdErrCh, nil
+}
+
+func waitForCommand(ctx context.Context, t *testing.T,
+	stdoutCh, stderrCh <-chan string, cmdErrCh <-chan error) {
+
+	for {
+		select {
+		case line, ok := <-stdoutCh:
+			if ok {
+				t.Log(line)
+			}
+		case line, ok := <-stderrCh:
+			if ok {
+				t.Log(line)
+			}
+		case cmdErr := <-cmdErrCh:
+			t.Log(fmt.Sprintf("Command finished with with exit code %d: %v",
+				exitCodeFromError(cmdErr), cmdErr))
+			return
+		}
+	}
 }
 
 func exitCodeFromError(err error) int {
diff --git a/integrationtests/dmap_test.go b/integrationtests/dmap_test.go
index 7d8b8b5..2de679b 100644
--- a/integrationtests/dmap_test.go
+++ b/integrationtests/dmap_test.go
@@ -15,7 +15,6 @@ func TestDMap(t *testing.T) {
 		return
 	}
 	inFile := "mapr_testdata.log"
-	stdoutFile := "dmap.stdout.tmp"
 	csvFile := "dmap.csv.tmp"
 	expectedCsvFile := "dmap.csv.expected"
 	queryFile := fmt.Sprintf("%s.query", csvFile)
@@ -25,14 +24,23 @@ func TestDMap(t *testing.T) {
 		"avg($goroutines),min(concurrentConnections),max(lifetimeConnections) "+
 		"group by $hostname outfile %s", csvFile)
 
-	_, err := runCommand(context.TODO(), t, stdoutFile,
-		"../dmap", "--query", query, inFile)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	stdoutCh, stderrCh, cmdErrCh, err := startCommand(ctx, t, "../dmap",
+		"--query", query,
+		"--logger", "stdout",
+		"--logLevel", "error",
+		"--noColor",
+		inFile)
 
 	if err != nil {
 		t.Error(err)
 		return
 	}
 
+	waitForCommand(ctx, t, stdoutCh, stderrCh, cmdErrCh)
+
 	if err := compareFiles(t, csvFile, expectedCsvFile); err != nil {
 		t.Error(err)
 		return
@@ -42,7 +50,6 @@ func TestDMap(t *testing.T) {
 		return
 	}
 
-	os.Remove(stdoutFile)
 	os.Remove(csvFile)
 	os.Remove(queryFile)
 }
@@ -100,16 +107,31 @@ func TestDMap3(t *testing.T) {
 		"avg($goroutines),min($goroutines) group by $time order by count($time) "+
 		"outfile %s", csvFile)
 
-	// Read many input files at once.
-	args := []string{"--logLevel", "trace", "--pprof", "localhost:8080", "--query", query}
-	for i := 0; i < 100; i++ {
-		args = append(args, inFile)
-	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	stdoutCh, stderrCh, cmdErrCh, err := startCommand(ctx, t, "../dmap",
+		"--query", query,
+		"--logger", "stdout",
+		"--logLevel", "info",
+		"--noColor",
+		inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile,
+		inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile,
+		inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile,
+		inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile,
+		inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile,
+		inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile,
+		inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile,
+		inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile,
+		inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile,
+		inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile, inFile)
 
-	if _, err := runCommand(context.TODO(), t, stdoutFile, "../dmap", args...); err != nil {
+	if err != nil {
 		t.Error(err)
 		return
 	}
+	waitForCommand(ctx, t, stdoutCh, stderrCh, cmdErrCh)
+
 	if err := compareFilesContents(t, csvFile, expectedCsvFile); err != nil {
 		t.Error(err)
 		return
diff --git a/integrationtests/dtail_test.go b/integrationtests/dtail_test.go
index c6d0107..2b4d6de 100644
--- a/integrationtests/dtail_test.go
+++ b/integrationtests/dtail_test.go
@@ -45,7 +45,6 @@ func TestDTailWithServer(t *testing.T) {
 		return
 	}
 
-	// TODO: In testmode, the client should not try to manipulate any known_hosts files.
 	// TODO: In testmode, never read a config file (use none for all commands)
 	clientCh, _, _, err := startCommand(ctx, t,
 		"../dtail",