Introducing progress-message handler

handler/builder.go

@@ -54,48 +54,30 @@ func (handler *InoHandler) rebuildEnvironmentLoop() {
 		// Regenerate preprocessed sketch!
 		done := make(chan bool)
 		go func() {
-			{
-				// Request a new progress token
-				req := &lsp.WorkDoneProgressCreateParams{Token: "arduinoLanguageServerRebuild"}
-				var resp lsp.WorkDoneProgressCreateResult
-				if err := handler.StdioConn.Call(context.Background(), "window/workDoneProgress/create", req, &resp, nil); err != nil {
-					log.Printf("    !!! could not create report progress: %s", err)
-					<-done
-					return
-				}
-			}
 
-			req := &lsp.ProgressParams{Token: "arduinoLanguageServerRebuild"}
-			req.Value = lsp.WorkDoneProgressBegin{
+			handler.progressHandler.Create("arduinoLanguageServerRebuild")
+			handler.progressHandler.Begin("arduinoLanguageServerRebuild", &lsp.WorkDoneProgressBegin{
 				Title: "Building sketch",
-			}
-			if err := handler.StdioConn.Notify(context.Background(), "$/progress", req, nil); err != nil {
-				log.Printf("    !!! could not report progress: %s", err)
-			}
+			})
+
 			count := 0
 			dots := []string{".", "..", "..."}
 			for {
 				select {
 				case <-time.After(time.Millisecond * 400):
 					msg := "compiling" + dots[count%3]
 					count++
-					req.Value = lsp.WorkDoneProgressReport{Message: &msg}
-					if err := handler.StdioConn.Notify(context.Background(), "$/progress", req, nil); err != nil {
-						log.Printf("    !!! could not report progress: %s", err)
-					}
+					handler.progressHandler.Report("arduinoLanguageServerRebuild", &lsp.WorkDoneProgressReport{Message: &msg})
 				case <-done:
 					msg := "done"
-					req.Value = lsp.WorkDoneProgressEnd{Message: &msg}
-					if err := handler.StdioConn.Notify(context.Background(), "$/progress", req, nil); err != nil {
-						log.Printf("    !!! could not report progress: %s", err)
-					}
+					handler.progressHandler.End("arduinoLanguageServerRebuild", &lsp.WorkDoneProgressEnd{Message: &msg})
 					return
 				}
 			}
 		}()
 
 		handler.dataMux.Lock()
-		handler.initializeWorkbench(nil)
+		handler.initializeWorkbench(context.Background(), nil)
 		handler.dataMux.Unlock()
 		done <- true
 		close(done)

handler/handler.go

@@ -47,6 +47,7 @@ type InoHandler struct {
 
 	stdioNotificationCount  int64
 	clangdNotificationCount int64
+	progressHandler         *ProgressProxyHandler
 
 	clangdStarted              *sync.Cond
 	dataMux                    sync.RWMutex
@@ -83,6 +84,9 @@ func NewInoHandler(stdio io.ReadWriteCloser, board lsp.Board) *InoHandler {
 	stdStream := jsonrpc2.NewBufferedStream(stdio, jsonrpc2.VSCodeObjectCodec{})
 	var stdHandler jsonrpc2.Handler = jsonrpc2.HandlerWithError(handler.HandleMessageFromIDE)
 	handler.StdioConn = jsonrpc2.NewConn(context.Background(), stdStream, stdHandler)
+
+	handler.progressHandler = NewProgressProxy(handler.StdioConn)
+
 	if enableLogging {
 		log.Println("Initial board configuration:", board)
 	}
@@ -178,7 +182,7 @@ func (handler *InoHandler) HandleMessageFromIDE(ctx context.Context, conn *jsonr
 			defer handler.dataMux.Unlock()
 
 			log.Printf("LS  --- initializing workbench (running)")
-			handler.initializeWorkbench(p)
+			handler.initializeWorkbench(ctx, p)
 
 			// clangd should be running now...
 			handler.clangdStarted.Broadcast()
@@ -410,8 +414,6 @@ func (handler *InoHandler) HandleMessageFromIDE(ctx context.Context, conn *jsonr
 		err = handler.ClangdConn.Notify(ctx, req.Method, params)
 	} else {
 		log.Printf(prefix + "sent to Clang")
-		ctx, cancel := context.WithTimeout(ctx, 800*time.Millisecond)
-		defer cancel()
 		result, err = lsp.SendRequest(ctx, handler.ClangdConn, req.Method, params)
 	}
 	if err == nil && handler.buildSketchSymbolsLoad {
@@ -450,7 +452,7 @@ func (handler *InoHandler) exit() {
 	os.Exit(1)
 }
 
-func (handler *InoHandler) initializeWorkbench(params *lsp.InitializeParams) error {
+func (handler *InoHandler) initializeWorkbench(ctx context.Context, params *lsp.InitializeParams) error {
 	currCppTextVersion := 0
 	if params != nil {
 		log.Printf("    --> initialize(%s)\n", params.RootURI)
@@ -495,8 +497,6 @@ func (handler *InoHandler) initializeWorkbench(params *lsp.InitializeParams) err
 			},
 		}
 
-		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
-		defer cancel()
 		if err := handler.ClangdConn.Notify(ctx, "textDocument/didChange", syncEvent); err != nil {
 			log.Println("    error reinitilizing clangd:", err)
 			return err
@@ -513,7 +513,7 @@ func (handler *InoHandler) initializeWorkbench(params *lsp.InitializeParams) err
 		}
 
 		clangdStream := jsonrpc2.NewBufferedStream(clangdStdio, jsonrpc2.VSCodeObjectCodec{})
-		clangdHandler := jsonrpc2.AsyncHandler(jsonrpc2.HandlerWithError(handler.FromClangd))
+		clangdHandler := AsyncHandler{jsonrpc2.HandlerWithError(handler.FromClangd)}
 		handler.ClangdConn = jsonrpc2.NewConn(context.Background(), clangdStream, clangdHandler)
 
 		// Send initialization command to clangd
@@ -1447,20 +1447,55 @@ func (handler *InoHandler) FromClangd(ctx context.Context, connection *jsonrpc2.
 	}
 	defer log.Printf(prefix + "(done)")
 
-	log.Printf(prefix + "(queued)")
-	switch req.Method {
-	case // No locking required
-		"$/progress",
-		"window/workDoneProgress/create":
-	case // Read lock
-		"textDocument/publishDiagnostics",
-		"workspace/applyEdit":
-		handler.dataMux.RLock()
-		defer handler.dataMux.RUnlock()
-	default: // Default to read lock
-		handler.dataMux.RLock()
-		defer handler.dataMux.RUnlock()
+	if req.Method == "window/workDoneProgress/create" {
+		params := lsp.WorkDoneProgressCreateParams{}
+		if err := json.Unmarshal(*req.Params, &params); err != nil {
+			log.Printf(prefix+"error decoding window/workDoneProgress/create: %v", err)
+			return nil, err
+		}
+		handler.progressHandler.Create(params.Token)
+		return &lsp.WorkDoneProgressCreateResult{}, nil
 	}
+
+	if req.Method == "$/progress" {
+		// data may be of many different types...
+		log.Printf(prefix + "decoding progress...")
+		params := lsp.ProgressParams{}
+		if err := json.Unmarshal(*req.Params, &params); err != nil {
+			log.Printf(prefix+"error decoding progress: %v", err)
+			return nil, err
+		}
+		id := params.Token
+
+		var begin lsp.WorkDoneProgressBegin
+		if err := json.Unmarshal(*params.Value, &begin); err == nil {
+			log.Printf(prefix+"begin %s %v", id, begin)
+			handler.progressHandler.Begin(id, &begin)
+			return nil, nil
+		}
+
+		var report lsp.WorkDoneProgressReport
+		if err := json.Unmarshal(*params.Value, &report); err == nil {
+			log.Printf(prefix+"report %s %v", id, report)
+			handler.progressHandler.Report(id, &report)
+			return nil, nil
+		}
+
+		var end lsp.WorkDoneProgressEnd
+		if err := json.Unmarshal(*params.Value, &end); err == nil {
+			log.Printf(prefix+"end %s %v", id, end)
+			handler.progressHandler.End(id, &end)
+			return nil, nil
+		}
+
+		log.Printf(prefix + "error unsupported $/progress: " + string(*params.Value))
+		return nil, errors.New("unsupported $/progress: " + string(*params.Value))
+	}
+
+	// Default to read lock
+	log.Printf(prefix + "(queued)")
+	handler.dataMux.RLock()
+	defer handler.dataMux.RUnlock()
 	log.Printf(prefix + "(running)")
 
 	params, err := lsp.ReadParams(req.Method, req.Params)

handler/progress.go

@@ -0,0 +1,192 @@
+package handler
+
+import (
+	"context"
+	"log"
+	"sync"
+
+	"github.com/bcmi-labs/arduino-language-server/lsp"
+	"github.com/bcmi-labs/arduino-language-server/streams"
+	"github.com/sourcegraph/jsonrpc2"
+)
+
+type ProgressProxyHandler struct {
+	conn               *jsonrpc2.Conn
+	mux                sync.Mutex
+	actionRequiredCond *sync.Cond
+	proxies            map[string]*progressProxy
+}
+
+type progressProxyStatus int
+
+const (
+	progressProxyNew progressProxyStatus = iota
+	progressProxyCreated
+	progressProxyBegin
+	progressProxyReport
+	progressProxyEnd
+)
+
+type progressProxy struct {
+	currentStatus  progressProxyStatus
+	requiredStatus progressProxyStatus
+	beginReq       *lsp.WorkDoneProgressBegin
+	reportReq      *lsp.WorkDoneProgressReport
+	endReq         *lsp.WorkDoneProgressEnd
+}
+
+func NewProgressProxy(conn *jsonrpc2.Conn) *ProgressProxyHandler {
+	res := &ProgressProxyHandler{
+		conn:    conn,
+		proxies: map[string]*progressProxy{},
+	}
+	res.actionRequiredCond = sync.NewCond(&res.mux)
+	go res.handlerLoop()
+	return res
+}
+
+func (p *ProgressProxyHandler) handlerLoop() {
+	defer streams.CatchAndLogPanic()
+
+	p.mux.Lock()
+	defer p.mux.Unlock()
+
+	for {
+		p.actionRequiredCond.Wait()
+
+		for id, proxy := range p.proxies {
+			for proxy.currentStatus != proxy.requiredStatus {
+				p.handleProxy(id, proxy)
+			}
+		}
+
+		// Cleanup ended proxies
+		for id, proxy := range p.proxies {
+			if proxy.currentStatus == progressProxyEnd {
+				delete(p.proxies, id)
+			}
+		}
+	}
+}
+
+func (p *ProgressProxyHandler) handleProxy(id string, proxy *progressProxy) {
+	ctx := context.Background()
+	switch proxy.currentStatus {
+	case progressProxyNew:
+		p.mux.Unlock()
+		var res lsp.WorkDoneProgressCreateResult
+		err := p.conn.Call(ctx, "window/workDoneProgress/create", &lsp.WorkDoneProgressCreateParams{Token: id}, &res)
+		p.mux.Lock()
+
+		if err != nil {
+			log.Printf("ProgressHandler: error creating token %s: %v", id, err)
+		} else {
+			proxy.currentStatus = progressProxyCreated
+		}
+
+	case progressProxyCreated:
+		p.mux.Unlock()
+		err := p.conn.Notify(ctx, "$/progress", lsp.ProgressParams{
+			Token: id,
+			Value: lsp.Raw(proxy.beginReq),
+		})
+		p.mux.Lock()
+
+		proxy.beginReq = nil
+		if err != nil {
+			log.Printf("ProgressHandler: error sending begin req token %s: %v", id, err)
+		} else {
+			proxy.currentStatus = progressProxyBegin
+		}
+
+	case progressProxyBegin:
+		if proxy.requiredStatus == progressProxyReport {
+			p.mux.Unlock()
+			err := p.conn.Notify(ctx, "$/progress", &lsp.ProgressParams{
+				Token: id,
+				Value: lsp.Raw(proxy.reportReq)})
+			p.mux.Lock()
+
+			proxy.reportReq = nil
+			if err != nil {
+				log.Printf("ProgressHandler: error sending begin req token %s: %v", id, err)
+			} else {
+				proxy.requiredStatus = progressProxyBegin
+			}
+
+		} else if proxy.requiredStatus == progressProxyEnd {
+			p.mux.Unlock()
+			err := p.conn.Notify(ctx, "$/progress", &lsp.ProgressParams{
+				Token: id,
+				Value: lsp.Raw(proxy.endReq),
+			})
+			p.mux.Lock()
+
+			proxy.endReq = nil
+			if err != nil {
+				log.Printf("ProgressHandler: error sending begin req token %s: %v", id, err)
+			} else {
+				proxy.currentStatus = progressProxyEnd
+			}
+
+		}
+	}
+}
+
+func (p *ProgressProxyHandler) Create(id string) {
+	p.mux.Lock()
+	defer p.mux.Unlock()
+
+	if _, opened := p.proxies[id]; opened {
+		// Already created
+		return
+	}
+
+	p.proxies[id] = &progressProxy{
+		currentStatus:  progressProxyNew,
+		requiredStatus: progressProxyCreated,
+	}
+	p.actionRequiredCond.Broadcast()
+}
+
+func (p *ProgressProxyHandler) Begin(id string, req *lsp.WorkDoneProgressBegin) {
+	p.mux.Lock()
+	defer p.mux.Unlock()
+
+	proxy, ok := p.proxies[id]
+	if !ok {
+		return
+	}
+
+	proxy.beginReq = req
+	proxy.requiredStatus = progressProxyBegin
+	p.actionRequiredCond.Broadcast()
+}
+
+func (p *ProgressProxyHandler) Report(id string, req *lsp.WorkDoneProgressReport) {
+	p.mux.Lock()
+	defer p.mux.Unlock()
+
+	proxy, ok := p.proxies[id]
+	if !ok {
+		return
+	}
+
+	proxy.reportReq = req
+	proxy.requiredStatus = progressProxyReport
+	p.actionRequiredCond.Broadcast()
+}
+
+func (p *ProgressProxyHandler) End(id string, req *lsp.WorkDoneProgressEnd) {
+	p.mux.Lock()
+	defer p.mux.Unlock()
+
+	proxy, ok := p.proxies[id]
+	if !ok {
+		return
+	}
+
+	proxy.endReq = req
+	proxy.requiredStatus = progressProxyEnd
+	p.actionRequiredCond.Broadcast()
+}

handler/syncer.go

@@ -13,5 +13,11 @@ type AsyncHandler struct {
 
 // Handle handles a request or notification
 func (ah AsyncHandler) Handle(ctx context.Context, conn *jsonrpc2.Conn, req *jsonrpc2.Request) {
-	go ah.handler.Handle(ctx, conn, req)
+	switch req.Method {
+	case // Request that should not be parallelized
+		"window/workDoneProgress/create", "$/progress":
+		ah.handler.Handle(ctx, conn, req)
+	default: // By default process all requests in parallel
+		go ah.handler.Handle(ctx, conn, req)
+	}
 }