Make -P work with --recursive-max-concurrent=false

Author: karolba

main.go

@@ -245,18 +245,18 @@ func executeAndFlushTty(command []string) (exitCode int) {
 		log.Fatalf("Could not find executable %s: %v\n", command[0], err)
 	}
 
-	// this process won't be used for anything much more, let's cap memory usage a bit
-	// this reduces memory usage by a couple of megabytes when running a lot of executeAndFlushTtys
-	debug.SetMemoryLimit(0)
-	debug.FreeOSMemory()
-
 	process, err := os.StartProcess(path, command, &os.ProcAttr{
 		Files: standardFdToFile,
 	})
 	if err != nil {
 		log.Fatalf("Could not displaySequentially %s: %v\n", shellescape.QuoteCommand(command), err)
 	}
 
+	// this process won't be used for anything much more, let's cap memory usage a bit
+	// this reduces memory usage by a couple of megabytes when running a lot of executeAndFlushTtys
+	debug.SetMemoryLimit(0)
+	debug.FreeOSMemory()
+
 	processState, err := process.Wait()
 	if err != nil {
 		log.Fatalf("Could not wait for process %v, %v\n", shellescape.QuoteCommand(command), err)
@@ -291,10 +291,11 @@ func main() {
 		os.Exit(0)
 	}
 
-	if *flRecursiveProcessLimit {
-		if _, hasMasterLimitServer := os.LookupEnv(EnvGparallelChildLimitSocket); !hasMasterLimitServer {
-			createLimitServer()
-		}
+	if !*flRecursiveProcessLimit {
+		_ = os.Unsetenv(EnvGparallelChildLimitSocket)
+	}
+	if _, hasMasterLimitServer := os.LookupEnv(EnvGparallelChildLimitSocket); !hasMasterLimitServer {
+		createLimitServer()
 	}
 
 	processes := chann.New[*ProcessResult]()

recursivetasklimit.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"errors"
+	"io"
 	"io/fs"
 	"log"
 	"net"
@@ -15,6 +16,17 @@ import (
 
 const EnvGparallelChildLimitSocket = "_GPARALLEL_CHILD_LIMIT_SOCKET"
 
+func readOneByte(reader io.Reader) error {
+	var b [1]byte
+	_, err := reader.Read(b[:])
+	return err
+}
+
+func writeOneByte(writer io.Writer) error {
+	_, err := writer.Write([]byte{1})
+	return err
+}
+
 func serveClients(listener net.Listener) {
 	for {
 		conn, err := listener.Accept()
@@ -25,10 +37,9 @@ func serveClients(listener net.Listener) {
 			log.Fatalf("Error accepting connection on the %s unix socket: %v\n", os.Getenv(EnvGparallelChildLimitSocket), err)
 		}
 
-		_, _ = conn.Write([]byte{1})
+		_ = writeOneByte(conn)
 
-		var buf [1]byte
-		_, err = conn.Read(buf[:])
+		err = readOneByte(conn)
 		if errors.Is(err, net.ErrClosed) {
 			break
 		}
@@ -56,6 +67,9 @@ func createLimitServer() {
 		log.Fatalf("Couldn't listen on unix socket '%s': %v\n", listenPath, err)
 	}
 
+	// Every process has the ability to spawn 1 child of its own, and as many other children
+	// as there are active serveClients goroutines. That's why we spawn (*flMaxProcesses-1)
+	// of them.
 	for i := 0; i < *flMaxProcesses-1; i++ {
 		go serveClients(listener)
 	}
@@ -121,11 +135,7 @@ var recursiveTaskLimitClient = sync.OnceValue(func() (client struct {
 
 		mutex.Unlock()
 		ch := make(chan error)
-		go func() {
-			var b [1]byte
-			_, err = conn.Read(b[:])
-			ch <- err
-		}()
+		go func() { ch <- readOneByte(conn) }()
 		select {
 		case <-ctx.Done():
 			err = ctx.Err()
@@ -153,7 +163,7 @@ var recursiveTaskLimitClient = sync.OnceValue(func() (client struct {
 		}
 
 		if toClose != nil && toClose.conn != nil {
-			_, _ = toClose.conn.Write([]byte{2})
+			_ = writeOneByte(toClose.conn)
 			haveToClose("connection to master gparallel", toClose.conn)
 			toClose.conn = nil
 		} else if toClose != nil && toClose.cancel != nil {

runner.go

@@ -57,9 +57,7 @@ func (proc *ProcessResult) isAlive() bool {
 }
 
 func (proc *ProcessResult) wait() error {
-	if *flRecursiveProcessLimit {
-		defer recursiveTaskLimitClient().del(proc)
-	}
+	defer recursiveTaskLimitClient().del(proc)
 
 	// wait for both stdout and stderr if we opened two readers
 	<-proc.output.streamClosed
@@ -313,9 +311,7 @@ func runWithStdin(command []string, stdin io.Reader) (result *ProcessResult) {
 	result.originalCommand = command
 	result.exitCode = make(chan int)
 
-	if *flRecursiveProcessLimit {
-		recursiveTaskLimitClient().addWait(result)
-	}
+	recursiveTaskLimitClient().addWait(result)
 
 	if stdoutIsTty {
 		command = append([]string{executable(), "--_execute-and-flush-tty"}, command...)