Feat/meshlib sidecarless v1

client/call_execution.go

@@ -0,0 +1,34 @@
+package client
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/cloudwego/kitex/pkg/endpoint"
+	"github.com/cloudwego/kitex/pkg/remote"
+)
+
+func (kc *kClient) buildClientCall(ctx context.Context, kind remote.ClientCallType, handler remote.ClientTransHandler, next endpoint.Endpoint) (endpoint.Endpoint, error) {
+	if factory, ok := kc.opt.RemoteOpt.CliHandlerFactory.(remote.ClientCallFactory); ok {
+		call, err := factory.NewClientCall(ctx, remote.ClientCallConfig{
+			Type: kind, ServiceInfo: kc.svcInfo, RPCConfig: kc.opt.Configs,
+			RemoteOption: kc.opt.RemoteOpt, MetaHandlers: kc.opt.MetaHandlers,
+			TransHandler: handler, RunTask: workerPool.RunTask,
+		}, next)
+		if errors.Is(err, remote.ErrUnsupportedClientCall) {
+			return func(context.Context, interface{}, interface{}) error { return err }, nil
+		}
+		if err != nil {
+			return nil, err
+		}
+		if call == nil {
+			return nil, fmt.Errorf("client call factory %T returned a nil endpoint", factory)
+		}
+		return call, nil
+	}
+	if kind == remote.ClientCallUnary {
+		return rpcTimeoutMW(ctx).ToMiddleware()(next), nil
+	}
+	return next, nil
+}

client/call_execution_test.go

@@ -0,0 +1,95 @@
+package client
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	internalclient "github.com/cloudwego/kitex/internal/client"
+	"github.com/cloudwego/kitex/pkg/endpoint"
+	"github.com/cloudwego/kitex/pkg/remote"
+	"github.com/cloudwego/kitex/pkg/rpcinfo"
+)
+
+type callFactory struct {
+	remote.ClientTransHandlerFactory
+	kinds       []remote.ClientCallType
+	unsupported bool
+	failure     error
+}
+
+func (f *callFactory) NewClientCall(_ context.Context, cfg remote.ClientCallConfig, next endpoint.Endpoint) (endpoint.Endpoint, error) {
+	f.kinds = append(f.kinds, cfg.Type)
+	if f.unsupported && cfg.Type == remote.ClientCallStreaming {
+		return nil, remote.ErrUnsupportedClientCall
+	}
+	if f.failure != nil {
+		return nil, f.failure
+	}
+	return next, nil
+}
+
+func TestCallFactorySupportsBothCallTypes(t *testing.T) {
+	o := internalclient.NewOptions(nil)
+	t.Cleanup(func() { _ = o.RemoteOpt.ConnPool.Close() })
+	factory := &callFactory{}
+	o.RemoteOpt.CliHandlerFactory = factory
+	kc := &kClient{opt: o}
+	for _, kind := range []remote.ClientCallType{remote.ClientCallUnary, remote.ClientCallStreaming} {
+		invoked := false
+		ep, err := kc.buildClientCall(context.Background(), kind, nil, func(context.Context, interface{}, interface{}) error { invoked = true; return nil })
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err = ep(context.Background(), nil, nil); err != nil || !invoked {
+			t.Fatalf("kind=%v invoked=%v err=%v", kind, invoked, err)
+		}
+	}
+	if len(factory.kinds) != 2 || factory.kinds[0] != remote.ClientCallUnary || factory.kinds[1] != remote.ClientCallStreaming {
+		t.Fatal(factory.kinds)
+	}
+}
+
+func TestUnsupportedCallTypeDoesNotRejectOtherCalls(t *testing.T) {
+	o := internalclient.NewOptions(nil)
+	t.Cleanup(func() { _ = o.RemoteOpt.ConnPool.Close() })
+	o.RemoteOpt.CliHandlerFactory = &callFactory{unsupported: true}
+	kc := &kClient{opt: o}
+	for _, kind := range []remote.ClientCallType{remote.ClientCallUnary, remote.ClientCallStreaming} {
+		invoked := false
+		ep, err := kc.buildClientCall(context.Background(), kind, nil, func(context.Context, interface{}, interface{}) error { invoked = true; return nil })
+		if err != nil {
+			t.Fatal(err)
+		}
+		err = ep(context.Background(), nil, nil)
+		if kind == remote.ClientCallStreaming {
+			if !errors.Is(err, remote.ErrUnsupportedClientCall) || invoked {
+				t.Fatalf("%v %v", err, invoked)
+			}
+		} else if err != nil || !invoked {
+			t.Fatalf("%v %v", err, invoked)
+		}
+	}
+}
+
+func TestDefaultCallExecutionKeepsTimeoutAndStreamingIndependent(t *testing.T) {
+	o := internalclient.NewOptions(nil)
+	t.Cleanup(func() { _ = o.RemoteOpt.ConnPool.Close() })
+	_ = rpcinfo.AsMutableRPCConfig(o.Configs).SetRPCTimeout(time.Second)
+	kc := &kClient{opt: o}
+	for _, kind := range []remote.ClientCallType{remote.ClientCallUnary, remote.ClientCallStreaming} {
+		ctx := rpcinfo.NewCtxWithRPCInfo(context.Background(), rpcinfo.NewRPCInfo(nil, nil, rpcinfo.NewInvocation("", "Echo"), o.Configs, rpcinfo.NewRPCStats()))
+		deadline := false
+		ep, err := kc.buildClientCall(context.Background(), kind, nil, func(ctx context.Context, _, _ interface{}) error { _, deadline = ctx.Deadline(); return nil })
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err = ep(ctx, nil, nil); err != nil {
+			t.Fatal(err)
+		}
+		if deadline != (kind == remote.ClientCallUnary) {
+			t.Fatalf("kind=%v deadline=%v", kind, deadline)
+		}
+	}
+}

client/client.go

@@ -135,7 +135,7 @@ func (kc *kClient) init() (err error) {
 	mw := kc.initMiddlewares(ctx)
 	kc.initDebugService()
 	kc.richRemoteOption()
-	if err = kc.buildInvokeChain(mw); err != nil {
+	if err = kc.buildInvokeChain(ctx, mw); err != nil {
 		return err
 	}
 	if err = kc.warmingUp(); err != nil {
@@ -277,8 +277,9 @@ func (kc *kClient) initLBCache() error {
 
 type middleware struct {
 	// unary middleware
-	mws  []endpoint.Middleware      // wrap the mws at `context mw -> customized mws -> other mws`
-	uMws []endpoint.UnaryMiddleware // wrap the mws at `service cb mw -> xds router -> rpctimeout mw -> customized unary mws`
+	mws      []endpoint.Middleware      // wrap the mws at `context mw -> customized mws -> other mws`
+	uMws     []endpoint.UnaryMiddleware // service circuit breaker and XDS router
+	userUMws []endpoint.UnaryMiddleware // inside the call execution boundary
 
 	// streaming middleware
 	smws []endpoint.Middleware  // wrap the mws at `xds router -> context mw -> customized mws -> other mws`
@@ -318,9 +319,8 @@ func (kc *kClient) initMiddlewares(ctx context.Context) (mw middleware) {
 		// integrate xds if enabled
 		mw.uMws = append(mw.uMws, kc.opt.XDSRouterMiddleware.ToUnaryMiddleware())
 	}
-	mw.uMws = append(mw.uMws, rpcTimeoutMW(ctx))
 	kc.opt.UnaryOptions.InitMiddlewares(ctx)
-	mw.uMws = append(mw.uMws, kc.opt.UnaryOptions.UnaryMiddlewares...)
+	mw.userUMws = kc.opt.UnaryOptions.UnaryMiddlewares
 
 	// 3. stream middlewares
 	kc.opt.Streaming.InitMiddlewares(ctx)
@@ -476,22 +476,31 @@ func (kc *kClient) richRemoteOption() {
 	}
 }
 
-func (kc *kClient) buildInvokeChain(mw middleware) error {
-	innerHandlerEp, err := kc.invokeHandleEndpoint()
+func (kc *kClient) buildInvokeChain(ctx context.Context, mw middleware) error {
+	innerHandlerEp, handler, err := kc.invokeHandleEndpoint()
 	if err != nil {
 		return err
 	}
 	eps := endpoint.Chain(mw.mws...)(innerHandlerEp)
+	eps = endpoint.Endpoint(endpoint.UnaryChain(mw.userUMws...)(endpoint.UnaryEndpoint(eps)))
+	eps, err = kc.buildClientCall(ctx, remote.ClientCallUnary, handler, eps)
+	if err != nil {
+		return err
+	}
 
 	kc.eps = endpoint.UnaryChain(mw.uMws...)(func(ctx context.Context, req, resp interface{}) (err error) {
 		return eps(ctx, req, resp)
 	})
 
-	innerStreamingEp, err := kc.invokeStreamingEndpoint()
+	innerStreamingEp, streamHandler, err := kc.invokeStreamingEndpoint()
 	if err != nil {
 		return err
 	}
 	sEps := endpoint.Chain(mw.smws...)(innerStreamingEp)
+	sEps, err = kc.buildClientCall(ctx, remote.ClientCallStreaming, streamHandler, sEps)
+	if err != nil {
+		return err
+	}
 	kc.sEps = cep.StreamChain(mw.sMws...)(func(ctx context.Context) (st streaming.ClientStream, err error) {
 		resp := &streaming.Result{}
 		err = sEps(ctx, nil, resp)
@@ -503,10 +512,10 @@ func (kc *kClient) buildInvokeChain(mw middleware) error {
 	return nil
 }
 
-func (kc *kClient) invokeHandleEndpoint() (endpoint.Endpoint, error) {
+func (kc *kClient) invokeHandleEndpoint() (endpoint.Endpoint, remote.ClientTransHandler, error) {
 	transPipl, err := newCliTransHandler(kc.opt.RemoteOpt)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	return func(ctx context.Context, req, resp interface{}) (err error) {
 		var sendMsg remote.Message
@@ -543,7 +552,7 @@ func (kc *kClient) invokeHandleEndpoint() (endpoint.Endpoint, error) {
 		recvMsg = remote.NewMessage(resp, ri, remote.Reply, remote.Client)
 		err = cli.Recv(ctx, ri, recvMsg)
 		return err
-	}, nil
+	}, transPipl, nil
 }
 
 // Close is not concurrency safe.

client/rpctimeout_pool.go

@@ -279,8 +279,9 @@ type timeoutContext struct {
 	dl time.Time
 	ch chan struct{}
 
-	mu  sync.Mutex
-	err error
+	mu        sync.Mutex
+	err       error // Protected by mu.
+	callbacks *timeoutAfterFunc
 }
 
 func newTimeoutContext(ctx context.Context, timeout time.Duration) *timeoutContext {
@@ -318,10 +319,68 @@ func (p *timeoutContext) Err() error {
 }
 
 func (p *timeoutContext) Cancel(err error) {
+	var callbacks *timeoutAfterFunc
 	p.mu.Lock()
 	if err != nil && p.err == nil {
 		p.err = err
 		close(p.ch)
+		callbacks, p.callbacks = p.callbacks, nil
+		for f := callbacks; f != nil; f = f.next {
+			f.pending = false
+		}
+	}
+	p.mu.Unlock()
+	for f := callbacks; f != nil; {
+		next, callback := f.next, f.callback
+		f.next, f.prev, f.callback = nil, nil, nil
+		go callback()
+		f = next
+	}
+}
+
+type timeoutAfterFunc struct {
+	parent     *timeoutContext
+	prev, next *timeoutAfterFunc
+	callback   func()
+	pending    bool // Protected by parent.mu; false once stopped or dispatched.
+}
+
+// AfterFunc implements the callback hook recognized by context.AfterFunc.
+// Registration and stop need no watcher goroutine. Cancellation still runs each
+// callback asynchronously, and a successful stop prevents it from starting.
+func (p *timeoutContext) AfterFunc(callback func()) func() bool {
+	f := &timeoutAfterFunc{parent: p}
+	p.mu.Lock()
+	if p.err != nil || p.Context.Err() != nil {
+		p.mu.Unlock()
+		go callback()
+		return f.stop
+	}
+	f.pending, f.callback, f.next = true, callback, p.callbacks
+	if f.next != nil {
+		f.next.prev = f
 	}
+	p.callbacks = f
 	p.mu.Unlock()
+	return f.stop
+}
+
+func (f *timeoutAfterFunc) stop() bool {
+	p := f.parent
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if !f.pending {
+		return false
+	}
+	f.pending = false
+	if f.prev != nil {
+		f.prev.next = f.next
+	} else {
+		p.callbacks = f.next
+	}
+	if f.next != nil {
+		f.next.prev = f.prev
+	}
+	f.prev, f.next, f.callback = nil, nil, nil
+	return true
 }

client/stream.go

@@ -160,7 +160,7 @@ func (kc *kClient) StreamX(ctx context.Context, method string) (streaming.Client
 	return cs, err
 }
 
-func (kc *kClient) invokeStreamingEndpoint() (endpoint.Endpoint, error) {
+func (kc *kClient) invokeStreamingEndpoint() (endpoint.Endpoint, remote.ClientTransHandler, error) {
 	var handler remote.ClientTransHandler
 	var err error
 	if kc.opt.RemoteOpt.GRPCStreamingCliHandlerFactory != nil {
@@ -169,7 +169,7 @@ func (kc *kClient) invokeStreamingEndpoint() (endpoint.Endpoint, error) {
 		handler, err = kc.opt.RemoteOpt.CliHandlerFactory.NewTransHandler(kc.opt.RemoteOpt)
 	}
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	// recvEP and sendEP are the endpoints for the new stream interface,
@@ -194,7 +194,7 @@ func (kc *kClient) invokeStreamingEndpoint() (endpoint.Endpoint, error) {
 		rresp.ClientStream = clientStream
 		rresp.Stream = clientStream.GetGRPCStream()
 		return
-	}, nil
+	}, handler, nil
 }
 
 const (