diff --git a/tsc/internal/ipc/conn_sync.go b/tsc/internal/ipc/conn_sync.go index a0b0cecb3c3d1..cdbebc3a0338f 100644 --- a/tsc/internal/ipc/conn_sync.go +++ b/tsc/internal/ipc/conn_sync.go @@ -28,15 +28,21 @@ type SyncConn struct { // This ensures that concurrent calls from handler goroutines (e.g., project code // spawning goroutines that invoke filesystem callbacks) don't corrupt the stream. mu sync.Mutex + + calls int + reading bool + turn sync.Cond } // NewSyncConn creates a new sync connection with the given transport and handler. func NewSyncConn(rwc io.ReadWriteCloser, protocol Protocol, handler Handler) *SyncConn { - return &SyncConn{ + c := &SyncConn{ rwc: rwc, protocol: protocol, handler: handler, } + c.turn.L = &c.mu + return c } // SetCollectTiming enables or disables per-request server processing-time @@ -70,7 +76,7 @@ func (c *SyncConn) Run(ctx context.Context) error { } if msg.IsRequest() { - if err := c.handleRequest(ctx, msg); err != nil { + if err := c.handleRequest(ctx, msg, 0); err != nil { return err } } else if msg.IsNotification() { @@ -82,13 +88,21 @@ func (c *SyncConn) Run(ctx context.Context) error { } } +func (c *SyncConn) lockTurn(depth int) { + c.mu.Lock() + for c.calls != depth { + c.turn.Wait() + } + c.reading = depth > 0 +} + // handleRequest processes an incoming request. -func (c *SyncConn) handleRequest(ctx context.Context, msg *Message) (retErr error) { +func (c *SyncConn) handleRequest(ctx context.Context, msg *Message, depth int) (retErr error) { // Intercept the meta-requests for collected server timing before dispatching // to the handler, so they are answered directly and not themselves recorded. switch msg.Method { case string(MethodGetServerTiming): - c.mu.Lock() + c.lockTurn(depth) writeErr := c.protocol.WriteResponse(msg.ID, serverTimingSnapshot(c.timing)) c.mu.Unlock() if writeErr != nil { @@ -99,7 +113,7 @@ func (c *SyncConn) handleRequest(ctx context.Context, msg *Message) (retErr erro if c.timing != nil { c.timing.reset() } - c.mu.Lock() + c.lockTurn(depth) writeErr := c.protocol.WriteResponse(msg.ID, nil) c.mu.Unlock() if writeErr != nil { @@ -122,7 +136,7 @@ func (c *SyncConn) handleRequest(ctx context.Context, msg *Message) (retErr erro stack := string(debug.Stack()) err = fmt.Errorf("panic: %v\n%s", r, stack) - c.mu.Lock() + c.lockTurn(depth) writeErr := c.protocol.WriteError(msg.ID, &jsonrpc.ResponseError{ Code: jsonrpc.CodeInternalError, Message: err.Error(), @@ -141,7 +155,7 @@ func (c *SyncConn) handleRequest(ctx context.Context, msg *Message) (retErr erro c.timing.record(msg.Method, time.Since(start)) } - c.mu.Lock() + c.lockTurn(depth) defer c.mu.Unlock() var writeErr error @@ -175,6 +189,17 @@ func (c *SyncConn) Call(ctx context.Context, method string, params any) (json.Va // 3. We need to ensure write/read pairs are atomic c.mu.Lock() defer c.mu.Unlock() + for c.reading { + c.turn.Wait() + } + c.calls++ + c.reading = true + depth := c.calls + defer func() { + c.calls-- + c.reading = false + c.turn.Broadcast() + }() id := jsonrpc.NewIDString(method) @@ -202,8 +227,10 @@ func (c *SyncConn) Call(ctx context.Context, method string, params any) (json.Va if msg.IsRequest() { // A synchronous client callback may make a nested API request. Release // the protocol lock while handling it so nested callbacks can proceed. + c.reading = false + c.turn.Broadcast() c.mu.Unlock() - err := c.handleRequest(ctx, msg) + err := c.handleRequest(ctx, msg, depth) c.mu.Lock() if err != nil { return nil, err @@ -211,9 +238,11 @@ func (c *SyncConn) Call(ctx context.Context, method string, params any) (json.Va continue } if msg.IsNotification() { + c.reading = false + c.turn.Broadcast() c.mu.Unlock() c.handleNotification(ctx, msg) - c.mu.Lock() + c.lockTurn(depth) continue } return nil, fmt.Errorf("ipc: unexpected message while waiting for %q response", method) diff --git a/tsc/internal/ipc/conn_sync_test.go b/tsc/internal/ipc/conn_sync_test.go index cc10ce1523861..750798c20b42e 100644 --- a/tsc/internal/ipc/conn_sync_test.go +++ b/tsc/internal/ipc/conn_sync_test.go @@ -4,7 +4,10 @@ import ( "context" "errors" "io" + "net" + "sync" "testing" + "testing/synctest" "github.com/microsoft/TypeScript/tsc/internal/ipc" "github.com/microsoft/TypeScript/tsc/internal/json" @@ -78,3 +81,89 @@ func TestSyncConnRunReturnsPanicResponseWriteFailure(t *testing.T) { assert.Assert(t, errors.Is(err, responseErr), "expected panic response write error, got %v", err) assert.ErrorContains(t, err, "original panic: handler panic") } + +type gatedHandler map[string]chan struct{} + +func (h gatedHandler) HandleRequest(_ context.Context, _ string, params json.Value) (any, error) { + <-h[string(params)] + return params, nil +} + +func (gatedHandler) HandleNotification(context.Context, string, json.Value) error { + return nil +} + +func TestSyncConnAnswersNestedRequestsInStackOrder(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + server, client := net.Pipe() + handler := gatedHandler{`"a"`: make(chan struct{}), `"b"`: make(chan struct{})} + conn := ipc.NewSyncConn(server, ipc.NewJSONRPCProtocol(server), handler) + peer := ipc.NewJSONRPCProtocol(client) + answers := map[string]string{} + var serve func() json.Value + serve = func() json.Value { + for { + msg, err := peer.ReadMessage() + if err != nil { + return nil + } + if msg.IsResponse() { + return msg.Result + } + _ = peer.WriteRequest(jsonrpc.NewIDString("resolve"), "resolve", msg.Params) + answers[string(msg.Params)] = string(serve()) + _ = peer.WriteResponse(msg.ID, msg.Params) + } + } + go serve() + + var wg sync.WaitGroup + var a, b json.Value + wg.Go(func() { a, _ = conn.Call(t.Context(), "callback", json.Value(`"a"`)) }) + synctest.Wait() + wg.Go(func() { b, _ = conn.Call(t.Context(), "callback", json.Value(`"b"`)) }) + synctest.Wait() + close(handler[`"a"`]) + synctest.Wait() + close(handler[`"b"`]) + wg.Wait() + assert.NilError(t, client.Close()) + + assert.DeepEqual(t, answers, map[string]string{`"a"`: `"a"`, `"b"`: `"b"`}) + assert.Equal(t, string(a), `"a"`) + assert.Equal(t, string(b), `"b"`) + }) +} + +type callingNotificationHandler struct { + noOpHandler + conn *ipc.SyncConn +} + +func (h *callingNotificationHandler) HandleNotification(ctx context.Context, _ string, _ json.Value) error { + _, err := h.conn.Call(ctx, "inner", nil) + return err +} + +func TestSyncConnNotificationHandlerCanCall(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + server, client := net.Pipe() + handler := &callingNotificationHandler{} + conn := ipc.NewSyncConn(server, ipc.NewJSONRPCProtocol(server), handler) + handler.conn = conn + peer := ipc.NewJSONRPCProtocol(client) + go func() { + outer, _ := peer.ReadMessage() + _ = peer.WriteNotification("notify", nil) + inner, _ := peer.ReadMessage() + _ = peer.WriteResponse(inner.ID, inner.Method) + _ = peer.WriteResponse(outer.ID, outer.Method) + }() + + result, err := conn.Call(t.Context(), "outer", nil) + assert.NilError(t, err) + assert.Equal(t, string(result), `"outer"`) + }) +}