Fix `Compile` gRPC call hangs when there is a lot of output (#2171)

* Added integration test

* Apply stream-send synchronization in Compile grpc call

* Apply the same pattern to all daemon gRPC calls

commands/daemon/stream.go

@@ -98,3 +98,26 @@ func consumeStreamFrom(reader func() ([]byte, error)) io.Reader {
 	}()
 	return r
 }
+
+// SynchronizedSender is a sender function with an extra protection for
+// concurrent writes, if multiple threads call the Send method they will
+// be blocked and serialized.
+type SynchronizedSender[T any] struct {
+	lock          sync.Mutex
+	protectedSend func(T) error
+}
+
+// Send the message using the underlyng stream.
+func (s *SynchronizedSender[T]) Send(value T) error {
+	s.lock.Lock()
+	err := s.protectedSend(value)
+	s.lock.Unlock()
+	return err
+}
+
+// NewSynchronizedSend takes a Send function and wraps it in a SynchronizedSender
+func NewSynchronizedSend[T any](send func(T) error) *SynchronizedSender[T] {
+	return &SynchronizedSender[T]{
+		protectedSend: send,
+	}
+}

internal/integrationtest/arduino-cli.go

@@ -388,14 +388,15 @@ func (inst *ArduinoCLIInstance) PlatformInstall(ctx context.Context, packager, a
 }
 
 // Compile calls the "Compile" gRPC method.
-func (inst *ArduinoCLIInstance) Compile(ctx context.Context, fqbn, sketchPath string) (commands.ArduinoCoreService_CompileClient, error) {
+func (inst *ArduinoCLIInstance) Compile(ctx context.Context, fqbn, sketchPath string, warnings string) (commands.ArduinoCoreService_CompileClient, error) {
 	compileCl, err := inst.cli.daemonClient.Compile(ctx, &commands.CompileRequest{
 		Instance:   inst.instance,
 		Fqbn:       fqbn,
 		SketchPath: sketchPath,
 		Verbose:    true,
+		Warnings:   warnings,
 	})
-	logCallf(">>> Compile(%v %v)\n", fqbn, sketchPath)
+	logCallf(">>> Compile(%v %v warnings=%v)\n", fqbn, sketchPath, warnings)
 	return compileCl, err
 }
 

internal/integrationtest/daemon/daemon_concurrency_test.go

+// This file is part of arduino-cli.
+//
+// Copyright 2022 ARDUINO SA (http://www.arduino.cc/)
+//
+// This software is released under the GNU General Public License version 3,
+// which covers the main part of arduino-cli.
+// The terms of this license can be found at:
+// https://www.gnu.org/licenses/gpl-3.0.en.html
+//
+// You can be released from the requirements of the above licenses by purchasing
+// a commercial license. Buying such a license is mandatory if you want to
+// modify or otherwise use the software for commercial activities involving the
+// Arduino software without disclosing the source code of your own applications.
+// To purchase a commercial license, send an email to license@arduino.cc.
+
+package daemon_test
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/arduino/arduino-cli/rpc/cc/arduino/cli/commands/v1"
+	"github.com/arduino/go-paths-helper"
+	"github.com/stretchr/testify/require"
+)
+
+func TestArduinoCliDaemonCompileWithLotOfOutput(t *testing.T) {
+	// See: https://github.com/arduino/arduino-cli/issues/2169
+
+	env, cli := createEnvForDaemon(t)
+	defer env.CleanUp()
+
+	_, _, err := cli.Run("core", "install", "arduino:avr")
+	require.NoError(t, err)
+
+	grpcInst := cli.Create()
+	require.NoError(t, grpcInst.Init("", "", func(ir *commands.InitResponse) {
+		fmt.Printf("INIT> %v\n", ir.GetMessage())
+	}))
+
+	sketchPath, err := paths.New("..", "testdata", "ManyWarningsSketch").Abs()
+	require.NoError(t, err)
+
+	testCompile := func() {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		compile, err := grpcInst.Compile(ctx, "arduino:avr:uno", sketchPath.String(), "all")
+		require.NoError(t, err)
+		msgCount := 0
+		for {
+			_, err := compile.Recv()
+			if err == io.EOF {
+				break
+			}
+			msgCount++
+			require.NoError(t, err)
+		}
+		fmt.Println("Received", msgCount, "messages.")
+	}
+
+	// The synchronization bug doesn't always happens, try 10 times to
+	// increase the chance to trigger it.
+	testCompile()
+	testCompile()
+	testCompile()
+	testCompile()
+	testCompile()
+	testCompile()
+	testCompile()
+	testCompile()
+	testCompile()
+	testCompile()
+}

internal/integrationtest/daemon/daemon_test.go

@@ -135,7 +135,7 @@ func TestDaemonCompileOptions(t *testing.T) {
 
 	// Build sketch (with errors)
 	sk := paths.New("testdata", "bare_minimum")
-	compile, err := grpcInst.Compile(context.Background(), "arduino:avr:uno:some_menu=bad", sk.String())
+	compile, err := grpcInst.Compile(context.Background(), "arduino:avr:uno:some_menu=bad", sk.String(), "")
 	require.NoError(t, err)
 	for {
 		msg, err := compile.Recv()
@@ -153,7 +153,7 @@ func TestDaemonCompileOptions(t *testing.T) {
 	}
 
 	// Build sketch (without errors)
-	compile, err = grpcInst.Compile(context.Background(), "arduino:avr:uno:some_menu=good", sk.String())
+	compile, err = grpcInst.Compile(context.Background(), "arduino:avr:uno:some_menu=good", sk.String(), "")
 	require.NoError(t, err)
 	for {
 		msg, err := compile.Recv()
@@ -180,7 +180,7 @@ func TestDaemonCompileAfterFailedLibInstall(t *testing.T) {
 
 	// Build sketch (with errors)
 	sk := paths.New("testdata", "bare_minimum")
-	compile, err := grpcInst.Compile(context.Background(), "", sk.String())
+	compile, err := grpcInst.Compile(context.Background(), "", sk.String(), "")
 	require.NoError(t, err)
 	for {
 		msg, err := compile.Recv()