Limiter (concurrent control)

Overview

Restricted fluids are a service governance capability that is used to limit the combined use of services to protect their stability.Only the use of restricted fluids for rest,grpc services is described in this subdocument, without a limited flow algorithm.

Limit flow is typically single-node flows, cluster limit flow (average amount of limit flow value for cluster nodes, its essence or single nodal flow), distributive stream.

The current presentation is a single-node flow that is more appropriate than a single-point flow.

Preparation

  1. Create a go mod project
  1. $ mkdir -p ~/workspace/demo && cd ~/workspace/demo
  2. $ go mod init demo

rest service

Limit Configuration

  1. RestConf struct {
  2.     ...
  3.     MaxConns int    `json:",default=10000"` // 最大并发连接数，默认值为 10000 qps
  4.     ... 
  5. }

Sample

We use a demo to describe the use of the lower limit streamer.

  1. Create new directory demo project rest-limit-demo
  1. $ cd ~/workspace/demo
  2. $ mkdir rest-limit-demo && cd rest-limit-demo
  1. Create a limit.api file, copy the following to the file

limit.api

  1. syntax = "v1"
  3. service limit {
  4.     @handler ping
  5.     get /ping
  6. }
  1. Generate rest code
  1. $ cd ~/workspace/demo/rest-limit-demo
  2. $ goctl api go -api limit.api -dir .
  1. View directory structure
  1. $ tree
  2. .
  3. ├── etc
  4. │   └── limit.yaml
  5. ├── internal
  6. │   ├── config
  7. │   │   └── config.go
  8. │   ├── handler
  9. │   │   ├── pinghandler.go
  10. │   │   └── routes.go
  11. │   ├── logic
  12. │   │   └── pinglogic.go
  13. │   ├── svc
  14. │   │   └── servicecontext.go
  15. │   └── types
  16. │       └── types.go
  17. ├── limit.api
  18. └── limit.go

We modify the configuration, limit qps to 100, and then add a logic of blocking logic.

  1. Edit configuation

Change to <code> maxConns in ~/workspace/demo/etc/limit.yaml to 100

  1. Add Logic Code

Add a blocking logic to~/workspace/demo/rest-limit-demo/internal/logic/pinglogic.go 中的 Ping

Final code content below

  • limit.yaml
  • pinglogic.go

limit.yaml

  1. Name: limit
  2. Host: 0.0.0.0
  3. Port: 8888
  4. MaxConns: 100
  1. package logic
  3. import (
  4.     "context"
  5.     "time"
  7.     "github.com/zeromicro/go-zero/core/logx"
  9.     "demo/rest-limit-demo/internal/svc"
  10. )
  12. type PingLogic struct {
  13.     logx.Logger
  14.     ctx    context.Context
  15.     svcCtx *svc.ServiceContext
  16. }
  18. func NewPingLogic(ctx context.Context, svcCtx *svc.ServiceContext) *PingLogic {
  19.     return &PingLogic{
  20.         Logger: logx.WithContext(ctx),
  21.         ctx:    ctx,
  22.         svcCtx: svcCtx,
  23.     }
  24. }
  26. func (l *PingLogic) Ping() error {
  27.     time.Sleep(50 * time.Millisecond)
  28.     return nil
  29. }

Let’s run this easiest rest service first. We use the hey tool to simply pressure interface.

Start service

  1. $ cd ~/workspace/demo/rest-limit-demo
  2. $ go run limit.go

Separate terminal pressure

  1. # 我们用 hey 工具来进行压测，压测 90 个并发，执行 1 秒
  2. $ hey -z 1s -c 90 -q 1 'http://localhost:8888/ping'
  4. Summary:
  5.   Total:    1.1084 secs
  6.   Slowest:  0.1066 secs
  7.   Fastest:  0.0607 secs
  8.   Average:  0.0890 secs
  9.   Requests/sec: 81.1980
  12. Response time histogram:
  13.   0.061 [1] |■
  14.   0.065 [2] |■■■
  15.   0.070 [8] |■■■■■■■■■■■
  16.   0.074 [13]    |■■■■■■■■■■■■■■■■■
  17.   0.079 [5] |■■■■■■■
  18.   0.084 [0] |
  19.   0.088 [0] |
  20.   0.093 [2] |■■■
  21.   0.097 [23]    |■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
  22.   0.102 [30]    |■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
  23.   0.107 [6] |■■■■■■■■
  26. Latency distribution:
  27.   10% in 0.0682 secs
  28.   25% in 0.0742 secs
  29.   50% in 0.0961 secs
  30.   75% in 0.0983 secs
  31.   90% in 0.0996 secs
  32.   95% in 0.1039 secs
  33.   0% in 0.0000 secs
  35. Details (average, fastest, slowest):
  36.   DNS+dialup:   0.0054 secs, 0.0607 secs, 0.1066 secs
  37.   DNS-lookup:   0.0000 secs, 0.0000 secs, 0.0000 secs
  38.   req write:    0.0002 secs, 0.0000 secs, 0.0011 secs
  39.   resp wait:    0.0832 secs, 0.0576 secs, 0.0942 secs
  40.   resp read:    0.0001 secs, 0.0000 secs, 0.0012 secs
  42. Status code distribution:
  43.   [200] 90 responses

In terms of pressure measurements, 90 requests have been successful, and we have stepped up and count to see what will happen.

  1. # 我们用 hey 工具来进行压测，压测 110 个并发，执行 1 秒
  2. $ hey -z 1s -c 110 -q 1 'http://127.0.0.1:8888/ping'
  4. Summary:
  5.   Total:    1.0833 secs
  6.   Slowest:  0.0756 secs
  7.   Fastest:  0.0107 secs
  8.   Average:  0.0644 secs
  9.   Requests/sec: 101.5403
  12. Response time histogram:
  13.   0.011 [1] |■
  14.   0.017 [9] |■■■■■■■
  15.   0.024 [0] |
  16.   0.030 [0] |
  17.   0.037 [0] |
  18.   0.043 [0] |
  19.   0.050 [0] |
  20.   0.056 [0] |
  21.   0.063 [2] |■■
  22.   0.069 [45]    |■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
  23.   0.076 [53]    |■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
  26. Latency distribution:
  27.   10% in 0.0612 secs
  28.   25% in 0.0665 secs
  29.   50% in 0.0690 secs
  30.   75% in 0.0726 secs
  31.   90% in 0.0737 secs
  32.   95% in 0.0740 secs
  33.   99% in 0.0756 secs
  35. Details (average, fastest, slowest):
  36.   DNS+dialup:   0.0040 secs, 0.0107 secs, 0.0756 secs
  37.   DNS-lookup:   0.0000 secs, 0.0000 secs, 0.0000 secs
  38.   req write:    0.0001 secs, 0.0000 secs, 0.0025 secs
  39.   resp wait:    0.0602 secs, 0.0064 secs, 0.0741 secs
  40.   resp read:    0.0000 secs, 0.0000 secs, 0.0001 secs
  42. Status code distribution:
  43.   [200] 100 responses
  44.   [503] 10 responses

Based on pressure measurements, our services can only support 100 combinations, and more than 100 concurrent requests will be restricted to return to the 503 status code. Flow limitation-related errors also appear in the service’s logs at:

  1. {"@timestamp":"2023-02-09T18:41:55.500+08:00","caller":"internal/log.go:62","content":"(/ping - 127.0.0.1:64163) concurrent connections over 100, rejected with code 503","level":"error","span":"08fa61d49b694e63","trace":"622b2287f32ff2d45f4dfce3eec8e62c"}

gRPC service

grpc is an Intranet service that does not provide services externally but only for other services in-house, so we do not need to limit the flow of services by default.

If you really want to limit the flow, this can be done by grpc intermediaries, for example below.

We use one of the simplest grpc server to show.

  1. Create new directory demo project grpc-limit-demo
  1. $ cd ~/workspace/demo
  2. $ mkdir grpc-limit-demo && cd grpc-limit-demo
  1. Create a limit.proto file, copy the following to the file

limit.proto

  1. syntax = "proto3";
  3. package proto;
  5. option  go_package = "./proto";
  7. message PingReq{}
  8. message PingResp{}
  10. service limit{
  11.   rpc Ping(PingReq) returns (PingResp);
  12. }
  1. Generate grpc code
  1. $ cd ~/workspace/demo/grpc-limit-demo
  2. $ goctl rpc protoc limit.proto --go_out=.  --go-grpc_out=.  --zrpc_out=.
  1. View Directory
  1. $ tree
  2. .
  3. ├── etc
  4. │   └── limit.yaml
  5. ├── internal
  6. │   ├── config
  7. │   │   └── config.go
  8. │   ├── logic
  9. │   │   └── pinglogic.go
  10. │   ├── server
  11. │   │   └── limitserver.go
  12. │   └── svc
  13. │       └── servicecontext.go
  14. ├── limit
  15. │   └── limit.go
  16. ├── limit.go
  17. ├── limit.proto
  18. └── proto
  19.     ├── limit.pb.go
  20.     └── limit_grpc.pb.go
  22. 8 directories, 10 files

We have implemented ~/workspace/demo/grpc-limit-demo/internal/logic/limitlogic.go 中实现 Ping below：

imitlogic.go

  1. package logic
  3. import (
  4.     "context"
  5.     "time"
  7.     "demo/grpc-limit-demo/internal/svc"
  8.     "demo/grpc-limit-demo/proto"
  10.     "github.com/zeromicro/go-zero/core/logx"
  11. )
  13. type PingLogic struct {
  14.     ctx    context.Context
  15.     svcCtx *svc.ServiceContext
  16.     logx.Logger
  17. }
  19. func NewPingLogic(ctx context.Context, svcCtx *svc.ServiceContext) *PingLogic {
  20.     return &PingLogic{
  21.         ctx:    ctx,
  22.         svcCtx: svcCtx,
  23.         Logger: logx.WithContext(ctx),
  24.     }
  25. }
  27. func (l *PingLogic) Ping(in *proto.PingReq) (*proto.PingResp, error) {
  28.     time.Sleep(50*time.Millisecond)
  30.     return &proto.PingResp{}, nil
  31. }
  1. Add middleware in ~/workspace/demo/grpc-limit-demo/limit.go ：

limit.go

  1. package main
  3. import (
  4.     "context"
  5.     "flag"
  6.     "fmt"
  7.     "net/http"
  9.     "github.com/zeromicro/go-zero/core/logx"
  10.     "github.com/zeromicro/go-zero/core/syncx"
  11.     "google.golang.org/grpc/codes"
  12.     "google.golang.org/grpc/status"
  14.     "demo/grpc-limit-demo/internal/config"
  15.     "demo/grpc-limit-demo/internal/server"
  16.     "demo/grpc-limit-demo/internal/svc"
  17.     "demo/grpc-limit-demo/proto"
  19.     "github.com/zeromicro/go-zero/core/conf"
  20.     "github.com/zeromicro/go-zero/core/service"
  21.     "github.com/zeromicro/go-zero/zrpc"
  22.     "google.golang.org/grpc"
  23.     "google.golang.org/grpc/reflection"
  24. )
  26. var configFile = flag.String("f", "etc/limit.yaml", "the config file")
  28. func main() {
  29.     flag.Parse()
  31.     var c config.Config
  32.     conf.MustLoad(*configFile, &c)
  33.     ctx := svc.NewServiceContext(c)
  35.     s := zrpc.MustNewServer(c.RpcServerConf, func(grpcServer *grpc.Server) {
  36.         proto.RegisterLimitServer(grpcServer, server.NewLimitServer(ctx))
  38.         if c.Mode == service.DevMode || c.Mode == service.TestMode {
  39.             reflection.Register(grpcServer)
  40.         }
  41.     })
  42.     var n = 100
  43.     l := syncx.NewLimit(n)
  44.     s.AddUnaryInterceptors(func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
  45.         if l.TryBorrow() {
  46.             defer func() {
  47.                 if err := l.Return(); err != nil {
  48.                     logx.Error(err)
  49.                 }
  50.             }()
  51.             return handler(ctx, req)
  52.         } else {
  53.             logx.Errorf("concurrent connections over %d, rejected with code %d",
  54.                 n, http.StatusServiceUnavailable)
  55.             return nil, status.Error(codes.Unavailable, "concurrent connections over limit")
  56.         }
  57.     })
  58.     defer s.Stop()
  60.     fmt.Printf("Starting rpc server at %s...\n", c.ListenOn)
  61.     s.Start()
  62. }
  64. }

We will generate profiles ~/workspace/demo/grpc-limit-demo/etc/limit.yaml to delete the etcd configuration in the file and start grpc server using direct connections.

limit.yaml

  1. Name: limit.rpc
  2. ListenOn: 0.0.0.0:8080

Start service

  1. $ cd ~/workspace/demo/grpc-limit-demo
  2. $ go run limit.go

Now that the grpc server is available, we use ghz to pressure it.

  1. $ cd ~/workspace/demo/grpc-limit-demo
  2. # 压测 90 qps，共请求 110 次
  3. $ ghz --insecure --proto=limit.proto --call=proto.limit.Ping -d '{}' -c 90 -n 110  127.0.0.1:8080 
  5. Summary:
  6.   Count:        110
  7.   Total:        108.94 ms
  8.   Slowest:      60.51 ms
  9.   Fastest:      50.24 ms
  10.   Average:      55.73 ms
  11.   Requests/sec: 1009.75
  13. Response time histogram:
  14.   50.240 [1]  |∎∎
  15.   51.266 [14] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  16.   52.293 [9]  |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  17.   53.320 [3]  |∎∎∎∎∎∎
  18.   54.347 [9]  |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  19.   55.374 [5]  |∎∎∎∎∎∎∎∎∎∎
  20.   56.401 [9]  |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  21.   57.427 [20] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  22.   58.454 [19] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  23.   59.481 [14] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  24.   60.508 [7]  |∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  26. Latency distribution:
  27.   10 % in 50.88 ms 
  28.   25 % in 53.40 ms 
  29.   50 % in 56.81 ms 
  30.   75 % in 57.96 ms 
  31.   90 % in 58.93 ms 
  32.   95 % in 59.52 ms 
  33.   99 % in 60.26 ms 
  35. Status code distribution:
  36.   [OK]   110 responses

All requests can be seen to be successful, and we come back to 110 qps.

  1. $ cd ~/workspace/demo/grpc-limit-demo
  2. # 压测 110 qps，共请求 110 次
  3. $ ghz --insecure --proto=limit.proto --call=proto.limit.Ping -d '{}' -c 110 -n 110  127.0.0.1:8080
  5. Summary:
  6.   Count:        110
  7.   Total:        68.35 ms
  8.   Slowest:      67.53 ms
  9.   Fastest:      58.76 ms
  10.   Average:      58.03 ms
  11.   Requests/sec: 1609.33
  13. Response time histogram:
  14.   58.763 [1]  |∎∎
  15.   59.640 [7]  |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  16.   60.517 [12] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  17.   61.394 [7]  |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  18.   62.272 [10] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  19.   63.149 [15] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  20.   64.026 [14] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  21.   64.903 [19] |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  22.   65.780 [9]  |∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
  23.   66.657 [3]  |∎∎∎∎∎∎
  24.   67.534 [3]  |∎∎∎∎∎∎
  26. Latency distribution:
  27.   10 % in 59.81 ms 
  28.   25 % in 60.96 ms 
  29.   50 % in 62.96 ms 
  30.   75 % in 64.52 ms 
  31.   90 % in 65.29 ms 
  32.   95 % in 66.09 ms 
  33.   99 % in 67.03 ms 
  35. Status code distribution:
  36.   [Unavailable]   10 responses    
  37.   [OK]            100 responses   
  39. Error distribution:
  40.   [10]   rpc error: code = Unavailable desc = concurrent connections over limit

It can be seen, when the concurrent output exceeds 100, returns rpc error: code = Unavailable desc = concurrent connections over limit.

References

  1. Lightweight Pressure Tool hey
  2. grpc Pressure ghz