镜像站点 · 本页由第三方 GitHub 只读镜像提供,非 GitHub 官方站点,不接受任何登录或凭据输入。前往 github.com
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 38 additions & 9 deletions tsc/internal/ipc/conn_sync.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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() {
Expand All @@ -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 {
Expand All @@ -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 {
Expand All @@ -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(),
Expand All @@ -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
Expand Down Expand Up @@ -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)

Expand Down Expand Up @@ -202,18 +227,22 @@ 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
}
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)
Expand Down
89 changes: 89 additions & 0 deletions tsc/internal/ipc/conn_sync_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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"`)
})
}
Loading