DEV Community: Kevin Wan

Circuit Breaker Explained

Kevin Wan — Mon, 12 Sep 2022 13:27:53 +0000

Circuit Breaker refers to a mechanism that suspends trading in the stock market for a certain period of time during trading hours when price fluctuations reach a certain target (melting point). This mechanism is like a circuit breaker that blows when the current is too high, hence the name. The purpose of the meltdown mechanism is to prevent systemic risk, to give the market more time to calm down, to avoid the spread of panic leading to market-wide volatility, and thus to prevent the occurrence of large-scale stock price declines.

Similarly, in the design of distributed systems, there should be a mechanism for meltdown. The circuit breaker is usually configured on the client side (caller side), when the client initiates a request to the server side, the error on the server side keeps increasing, then the circuit breaker may be triggered, after triggering the circuit breaker the client's request is no longer sent to the server side, but the request is rejected directly on the client side, thus the server side can be protected from overload. The server side mentioned here may be rpc service, http service, or mysql, redis, etc. Note that circuit breaker is a lossy mechanism, and some degradation strategies may be needed to work with it when it is circuit breaker enabled.

Circuit Breaker Principle

Modern microservice architectures are basically distributed, and the whole distributed system is composed of a very large number of microservices. Different services invoke each other and form a complex invocation chain. If a service in the complex invocation link is unstable, it may cascade and eventually the whole link may hang. Therefore, we need to circuit breaker and downgrade unstable service dependencies to temporarily cut off unstable service calls to avoid local instabilities that can lead to an avalanche of the whole distributed system.

To put it bluntly, I think circuit breaker is like a kind of proxy for those services that are prone to exceptions. This proxy can record the number of errors that occurred in recent calls, and then decide whether to continue the operation or return the error immediately.

The circuit breaker maintains a circuit breaker state machine internally, and the state machine transitions are related as shown in the following diagram.

The circuit breaker has three states.

Closed state: also the initial state, we need a call failure counter, if the call fails, it makes the number of failures plus 1. If the number of recent failures exceeds the threshold of allowed failures in a given time, it switches to the Open state, when a timeout clock is turned on, when the timeout clock time is reached, it switches to the Half Open state, the timeout time is set to give the system a chance to fix the error that caused the call to fail in order to return to the normal working state. In the Closed state, the error count is time-based. This prevents the circuit breaker from entering the Open state due to an accidental error, and can also be based on the number of consecutive failures.
Open state: In this state, the client request returns an immediate error response without invoking the server side.
Half-Open state: allows a certain number of client de-calls to the server, and if these requests have a successful call to the service, then it can be assumed that the error that caused the call to fail before has been corrected, at which point the circuit breaker switches to the Closed state and the error counter is reset. The Half-Open state is effective in preventing the recovering service from being hit again by a sudden influx of requests.

The following diagram shows the logic of the circuit breaker implementation in Netflix's open source project Hystrix.

From this flowchart, it can be seen that.

a request comes, first allowRequest() function to determine whether in the circuit breaker, if not, then release, if so, but also to see whether a circuit breaker time slice reached, if the circuit breaker time slice to, also release, otherwise directly return an error.
each call has two functions makeSuccess(duration) and makeFailure(duration) to count how many are successful or failed within a certain duration.
the condition isOpen() to determine whether to circuit breaker, is to calculate the failure/(success+failure) the current error rate, if higher than a threshold, then the circuit breaker is open, otherwise closed.
Hystrix will maintain a data in memory which records the statistics of the request results for each cycle, elements that exceed the length of time will be deleted.

Circuit breaker implementation

After understanding the principle of fusing, let's implement a set of fusers by ourselves.

Those who are familiar with go-zero know that fusing in go-zero does not use the way described above, but refers to "Google SRE" to adopt An adaptive fusing mechanism, what are the benefits of this adaptive approach? The following section will do a comparison based on these two mechanisms.

Below we implement a set of our own circuit breaker based on the fusing principle introduced above.

Code path: go-zero/core/breaker/hystrixbreaker.go

The default state of the circuit breaker is Closed, when the circuit breaker is opened the default cooling time is 5 seconds, when the circuit breaker is in HalfOpen state the default detection time is 200 milliseconds, by default the rateTripFunc method is used to determine whether to trigger the circuit breaker, the rule is that the sampling is greater than or equal to 200 and the error rate is greater than 50%, and a sliding window is used to record the total number of requests and number of errors.

func newHystrixBreaker() *hystrixBreaker {
  bucketDuration := time.Duration(int64(window) / int64(buckets))
  stat := collection.NewRollingWindow(buckets, bucketDuration)
  return &hystrixBreaker{
    state: Closed,
    coolingTimeout: defaultCoolingTimeout,
    detectTimeout: defaultDetectTimeout,
    tripFunc: rateTripFunc(defaultErrRate, defaultMinSample),
    stat: stat,
    now: time,
  }
}

func rateTripFunc(rate float64, minSamples int64) TripFunc {
  return func(rollingWindow *collection.RollingWindow) bool {
    var total, errs int64
    rollingWindow.Reduce(func(b *collection.Bucket) {
      total += b.Count
      errs += int64(b.Sum)
    })
    errRate := float64(errs) / float64(total)
    return total >= minSamples && errRate > rate
  }
}

The doReq method is called for each request. In this method, first determine if this request is rejected by the accept() method, and if rejected, return the circuit breaker error directly. Otherwise, req() is executed to really initiate the server-side call, and b.markSuccess() and b.markFailure() are called for success and failure respectively

func (b *hystrixBreaker) doReq(req func() error, fallback func(error) error, acceptable Acceptable) error {
  if err := b.accept(); err ! = nil {
    if fallback ! = nil {
      return fallback(err)
    }
    return err
  }

  defer func() {
    if e := recover(); e ! = nil {
      b.markFailure()
      panic(e)
    }
  }()

  err := req()
  if acceptable(err) {
    b.markSuccess()
  } else {
    b.markFailure()
  }

  return err
}

In the accept() method, first get the current circuit breaker state, when the circuit breaker is in the Closed state directly return, that normal processing of this request.

When the current state is Open, determine whether the cooling time is expired, if not expired, then directly return the circuit breaker error to reject the request, if expired, then the circuit breaker state is changed to HalfOpen, the main purpose of the cooling time is to give the server some time to recover from failure, to avoid continuous requests to the server to hit the hang.

When the current state is HalfOpen, first determine the probe interval to avoid probing too often, the default use of 200 milliseconds as the probe interval.

func (b *hystrixBreaker) accept() error {
  b.mux.Lock()
  switch b.getState() {
  case Open:
    now := b.now()
    if b.openTime.Add(b.coolingTimeout).After(now) {
      b.mux.Unlock()
      return ErrServiceUnavailable
    }
    if b.getState() == Open {
      atomic.StoreInt32((*int32)(&b.state), int32(HalfOpen))
      atomic.StoreInt32(&b.halfopenSuccess, 0)
      b.lastRetryTime = now
      b.mux.Unlock()
    } else {
      b.mux.Unlock()
      return ErrServiceUnavailable
    }
  case HalfOpen:
    now := b.now()
    if b.lastRetryTime.Add(b.detectTimeout).After(now) {
      b.mux.Unlock()
      return ErrServiceUnavailable
    }
    b.lastRetryTime = now
    b.mux.Unlock()
  case Closed:
    b.mux.Unlock()
  }

  return nil
}

If this request returns normally, the markSuccess() method is called. If the current circuit breaker is in the HalfOpen state, it determines whether the current number of successful probes is greater than the default number of successful probes, and if it is greater, the state of the circuit breaker is updated to Closed.

func (b *hystrixBreaker) markSuccess() {
  b.mux.Lock()
  switch b.getState() {
  case Open:
    b.mux.Unlock()
  case HalfOpen:
    Atomic.AddInt32(&b.halfopenSuccess, 1)
    if atomic.LoadInt32(&b.halfopenSuccess) > defaultHalfOpenSuccesss {
      atomic.StoreInt32((*int32)(&b.state), int32(Closed))
      b.stat.Reduce(func(b *collection.Bucket) {
        b.Count = 0
        b.Sum = 0
      })
    }
    b.mux.Unlock()
  case Closed:
    b.stat.Add(1)
    b.mux.Unlock()
  }
}

In markFailure() method, if the current state is Closed by executing tripFunc to determine if the circuit breaker condition is met, if it is met then change the circuit breaker state to Open state.

func (b *hystrixBreaker) markFailure() {
  b.mux.Lock()
  b.stat.Add(0)
  switch b.getState() {
  case Open:
    b.mux.Unlock()
  case HalfOpen:
    b.openTime = b.now()
    atomic.StoreInt32((*int32)(&b.state), int32(Open))
    b.mux.Unlock()
  case Closed:
    if b.tripFunc ! = nil && b.tripFunc(b.stat) {
      b.openTime = b.now()
      atomic.StoreInt32((*int32)(&b.state), int32(Open))
    }
    b.mux.Unlock()
  }
}

The overall logic of the implementation of the circuit breaker is relatively simple, read the code can basically understand, this part of the code implementation is rather rushed, there may be bugs, if you find bugs can always contact me for correction.

hystrixBreaker and googlebreaker comparison

Next, compare the fusing effect of the two fuses.

This part of the sample code is under: go-zero/example

The user-api and user-rpc services are defined respectively. user-api acts as a client to request user-rpc, and user-rpc acts as a server to respond to client requests.

In the example method of user-rpc, there is a 20% chance of returning an error.

func (l *UserInfoLogic) UserInfo(in *user.UserInfoRequest) (*user.UserInfoResponse, error) {
  ts := time.Now().UnixMilli()
  if in.UserId == int64(1) {
    if ts%5 == 1 {
      return nil, status.Error(codes.Internal, "internal error")
    }
    return &user.UserInfoResponse{
      UserId: 1,
      Name: "jack",
    }, nil

  }
  return &user.UserInfoResponse{}, nil
}

In the example method of user-api, make a request to user-rpc and then use the prometheus metric to record the number of normal requests.

var metricSuccessReqTotal = metric.NewCounterVec(&metric.CounterVecOpts{
  Namespace: "circuit_breaker",
  Subsystem: "requests",
  Name: "req_total",
  Help: "test for circuit breaker",
  Labels: []string{"method"},
})

func (l *UserInfoLogic) UserInfo() (resp *types.UserInfoResponse, err error) {
  for {
    _, err := l.svcCtx.UserRPC.UserInfo(l.ctx, &user.UserInfoRequest{UserId: int64(1)})
    if err ! = nil && err == breaker.ErrServiceUnavailable {
      fmt.Println(err)
      continue
    }
    metricSuccessReqTotal.Inc("UserInfo")
  }

  return &types.UserInfoResponse{}, nil
}

Start both services and then observe the number of normal requests under both fusing strategies.

The normal request rate for the googleBreaker is shown in the following figure.

The normal request rate for the hystrixBreaker is shown in the following graph.

From the above experimental results, we can see that the normal request count of go-zero's built-in googleBreaker is higher than that of hystrixBreaker. This is because hystrixBreaker maintains three states. When entering the Open state, a cooling clock is used in order to avoid continuing to cause pressure on the server to initiate requests, and no requests are spared during this time, meanwhile, after changing from the HalfOpen state to the Closed state, a large number of requests will be sent to the server again instantly, when the server It is likely that the server has not yet recovered, thus causing the circuit breaker to change to the Open state again. The googleBreaker is an adaptive fusing strategy, and does not require multiple states, and will not be like the hystrixBreaker as a one-size-fits-all, but will handle as many requests as possible, which is not what we expect, after all, the circuit breaker is detrimental to the customer. Let's learn the go-zero built-in googleBreaker.

Source code interpretation

googleBreaker code path in: go-zero/core/breaker/googlebreaker.go

In the doReq() method by accept() method to determine whether the circuit breaker is triggered, if the circuit breaker is triggered then return error, here if the callback function is defined then you can perform callbacks, such as doing some degraded data processing. If the request is normal then markSuccess() adds 1 to the total number of requests and normal requests, if the request fails by markFailure then only add 1 to the total number of requests.

func (b *googleBreaker) doReq(req func() error, fallback func(err error) error, acceptable Acceptable) error {
  if err := b.accept(); err ! = nil {
    if fallback ! = nil {
      return fallback(err)
    }

    return err
  }

  defer func() {
    if e := recover(); e ! = nil {
      b.markFailure()
      panic(e)
    }
  }()

  err := req()
  if acceptable(err) {
    b.markSuccess()
  } else {
    b.markFailure()
  }

  return err
}

Determine whether the circuit breaker is triggered by calculation in the accept() method.

In this algorithm, two request counts need to be recorded, which are

Total number of requests (requests): the sum of the number of requests initiated by the caller
Number of requests normally processed (accepts): the number of requests normally processed by the server

In the normal case, these two values are equal. As the called party's service starts rejecting requests with exceptions, the value of the number of requests accepted (accepts) starts to gradually become smaller than the number of requests (requests), at which point the caller can continue sending requests until requests = K * accepts, and once this limit is exceeded, the circuit breaker opens and new requests will be discarded locally with a certain probability of returning errors directly, the probability is calculated as follows.

max(0, (requests - K * accepts) / (requests + 1))

By modifying the K(multiplier) in the algorithm, the sensitivity of the circuit breaker can be adjusted, when decreasing the multiplier will make the adaptive circuit breaker algorithm more sensitive, when increasing the multiplier will make the adaptive circuit breaker algorithm less sensitive, for example, suppose the upper limit of the caller's requests is adjusted from requests = 2 * accepts to requests = 1.1 * accepts then it means that one out of every ten requests from the caller will trigger the circuit breaker.

func (b *googleBreaker) accept() error {
  accepts, total := b.history()
  weightedAccepts := b.k * float64(accepts)
  // https://landing.google.com/sre/sre-book/chapters/handling-overload/#eq2101
  dropRatio := math.Max(0, (float64(total-protection)-weightedAccepts)/float64(total+1))
  if dropRatio <= 0 {
    return nil
  }

  if b.proba.TrueOnProba(dropRatio) {
    return ErrServiceUnavailable
  }

  return nil
}

history counts the total number of current requests and the number of normally processed requests from the sliding window.

func (b *googleBreaker) history() (accepts, total int64) {
  b.stat.Reduce(func(b *collection.Bucket) {
    accepts += int64(b.Sum)
    total += b.Count
  })

  return
}

Conclusion

This article introduces a client-side throttling mechanism in service governance - circuit breaker. There are three states to be implemented in the hystrix circuit breaker policy, Open, HalfOpen and Closed, and the timing of switching between the different states is described in detail in the above article, so you can read it again and again to understand it, or better yet, implement it yourself. For go-zero built-in circuit breaker is no state, if you have to say its state, then there are only two cases of open and closed, it is based on the success rate of the current request adaptive discard request, is a more flexible circuit breaker strategy, discard request probability with the number of requests processed normally changing, the more requests processed normally the lower the probability of discarding requests, and vice versa the higher the probability of discarding requests. The higher the probability of dropping a request, the higher the probability of dropping a request.

Although the principle of circuit breaker is the same, the effect may be different due to different implementation mechanisms, so you can choose a circuit breaker strategy that meets the business scenario in actual production.

I hope this article will be helpful to you.

Code for this article: https://github.com/zhoushuguang/go-zero/tree/circuit-breaker

Reference

https://martinfowler.com/bliki/CircuitBreaker.html

https://github.com/Netflix/Hystrix/wiki/How-it-Works

Project address

https://github.com/zeromicro/go-zero

Welcome to use go-zero and star to support us!

5 minutes to add RESTful APIs for your gRPC services

Kevin Wan — Mon, 22 Aug 2022 14:35:00 +0000

gRPC service with HTTP interface?

While go-zero brought an excellent RESTful and gRPC service development experience to the developers, more expectations arised.

I want to write code only once
I want both the gRPC and HTTP interfaces

It makes sense! You see what users say.

User A: a set of logic, HTTP and gRPC together.

User B: if go-zero can simplify this step I feel it will become the the best microservices framework, ever.

So I fell into a deep thought: the user is never wrong, but do we want to provide it?

Here comes the article.

Let's write gRPC service first

We are too familiar with this service. Create a new directory, let's call it grpc-restufl, and put a sum.proto file in it

syntax = "proto3";

package sum;
option go_package="./pb";

message SumRequest {
    int64 a = 1;
    int64 b = 2;
}

message SumResponse {
    int64 result = 1;
}

service Sum {
    rpc Add(SumRequest) returns (SumResponse) {}
}

One-click generation, you know.

$ goctl rpc protoc --go_out=. --go-grpc_out=. --zrpc_out=. sum.proto

See what you get

.
├── etc
│ └── sum.yaml
├── go.mod
├── internal
│ ├── config
│ │ └── config.go
│ ├── logic
│ │ └── addlogic.go
│ ├── server
│ │ └── sumserver.go
│ └── svc
│ └── servicecontext.go
├─ pb
│ ├── sum.pb.go
│ └── sum_grpc.pb.go
├─ sum
│ └── sum.go
├── sum.go
└── sum.proto

To implement the business logic, modify the Add method in internal/logic/addlogic.go as follows.

func (l *AddLogic) Add(in *pb.SumRequest) (*pb.SumResponse, error) {
    return &pb.SumResponse{
        Result: in.A+in.B,
    }, nil
}

You can run it, and the business logic is there (although it is very simple, for demo purpose.)

$ go mod tidy && go run sum.go
Starting rpc server at 127.0.0.1:8080...

For those who are familiar with go-zero, there is no highlight (new knowledge) here, let's go ahead ~

Provide HTTP interface

Update go-zero

First, let's update go-zero to v1.4.0 version,

$ go get -u github.com/zeromicro/go-zero@latest

Modify the proto file

Modify sum.proto, I created a new sum-api.proto, as follows

syntax = "proto3";

package sum;
option go_package="./pb";

import "google/api/annotations.proto";

message SumRequest {
    int64 a = 1;
    int64 b = 2;
}

message SumResponse {
    int64 result = 1;
}

service Sum {
    rpc Add(SumRequest) returns (SumResponse) {
        option (google.api.http) = {
            post: "/v1/sum"
            body: "*"
        };
    }
}

Generate proto descriptor file

protoc --include_imports --proto_path=. --descriptor_set_out=sum.pb sum-api.proto

Modify the configuration file

The modified internal/config/config.go looks like this (partially)

type Config struct {
    zrpc.RpcServerConf
    Gateway gateway.GatewayConf
Gateway.GatewayConf }

The modified etc/sum.yaml is as follows

Gateway:
  Name: gateway
  Port: 8081
  Upstreams:
    - Grpc:
        Endpoints:
          - localhost:8080
      ProtoSets:
        - sum.pb

Modify the main function

Create gateway and use ServiceGroup to manage gRPC server and gateway server, part of the code is as follows.

    gw := gateway.MustNewServer(c.Gateway)
    group := service.NewServiceGroup()
    group.Add(s)
    group.Add(gw)
    defer group.Stop()

    fmt.Printf("Starting rpc server at %s... \n", c.ListenOn)
    fmt.Printf("Starting gateway at %s:%d... \n", c.Gateway.Host, c.Gateway.Port)
    group.Start()

Great job!

Let's start the service

$ go run sum.go
Starting rpc server at 127.0.0.1:8080...
Starting gateway at 0.0.0.0:8081...

Test it with curl

$ curl -i -H "Content-Type: application/json" -d '{"a":2, "b":3}' localhost:8081/v1/sum
HTTP/1.1 200 OK
Content-Type: application/json; charset=utf-8
Traceparent: 00-ad5b7df7a834a1c05ee64999e3310811-195ba1f4f9956cc4-00
Date: Mon, 18 Jul 2022 14:33:11 GMT
Content-Length: 20

{
  "result": "5"
}

And look at the link information in our gateway and gRPC logs that corresponds to what the client received, awesome!

{"@timestamp": "2022-07-18T22:33:11.437+08:00", "caller": "serverinterceptors/statinterceptor.go:76", "content": "127.0.0.1:61635 - /sum. Sum/Add - {\"a\":2,\"b\":3}", "duration": "0.0ms", "level": "info", "span": "b3c85cd32a76f8c9", "trace":" ad5b7df7a834a1c05ee64999e3310811"}
{"@timestamp": "2022-07-18T22:33:11.438+08:00", "caller": "handler/loghandler.go:197", "content":"[HTTP] 200 - POST /v1/sum - 127.0.0.1: 61662 - curl/7.79.1", "duration": "0.7ms", "level": "info", "span": "195ba1f4f9956cc4", "trace": "ad5b7df7a834a1c05ee64999e3310811"}

Conclusion

You see, adding the HTTP interface to our gRPC service is very easy? Isn't it?

Also, don't underestimate this simple gateway, the configuration will automatically load balance if it is docked to the gRPC service found behind it, and you can also customize the middleware to control it whatever you want.

By the way, the full code for this example is at.

https://github.com/kevwan/grpc-restful

Project address

https://github.com/zeromicro/go-zero

Feel free to use go-zero and star to support us!

Understanding and tuning parameters of connection pools

Kevin Wan — Tue, 21 Jun 2022 05:12:21 +0000

Did you feel the same?

When you are developing backend services, do you often have the following questions?

How many connections does the database connection pool have?
How long does the lifetime of each connection last?
When a connection is abnormally disconnected, is it the server side or the client side that actively disconnected?
Does the underlying library have KeepAlive requests when there are no requests for a long time?

Handling complex network situations is always one of the key points and difficulties of backend development. Are you also suffering from debugging various network situations?

Why I wrote tproxy?

When I was doing backend development and writing go-zero, I would often need to monitor network connections and analyze the request packets. For example.

Analyzing when a gRPC connection connects and reconnects, and adjusting various parameters accordingly, e.g. MaxConnectionIdle
Analyze the MySQL connection pool, how many connections are currently available, and what the connection lifetime policy is
It can also be used to observe and analyze any TCP connection to see if the server is actively disconnecting or the client is actively disconnecting, etc.

Installation of tproxy

$ GOPROXY=https://goproxy.cn/,direct go install github.com/kevwan/tproxy@latest

Or use the docker image: ``Shell

`Shell $ docker run --rm -it -p <listen-port>:<listen-port> -p <remote-port>:<remote-port> kevinwan/tproxy:v1 tproxy -l 0.0.0.0 -p <listen-port> -r host. docker.internal:<remote-port> `

arm64 system:

`Shell $ docker run --rm -it -p <listen-port>:<listen-port> -p <remote-port>:<remote-port> kevinwan/tproxy:v1-arm64 tproxy -l 0.0.0.0 -p <listen-port> - r host.docker.internal:<remote-port> `

Usage of tproxy

`Shell $ tproxy --help Usage of tproxy: -d duration the delay to relay packets -l string Local address to listen on (default "localhost") -p int Local port to listen on -q Quiet mode, only prints connection open/close and stats, default false -r string Remote address (host:port) to connect -t string The type of protocol, currently support grpc `

Analyze gRPC connections

`Shell tproxy -p 8088 -r localhost:8081 -t grpc -d 100ms `

Listening on localhost and port 8088
Redirect requests to localhost:8081
Analyze packet format as gRPC
Packet delay of 100ms

where we can see the initialization and back and forth of a request from gRPC, and we can see the first request with the stream id as 1.

For example, gRPC has a MaxConnectionIdle parameter that sets how long the connection will be closed after idle, and we can directly observe that the server will send an http2 GoAway packet after the time is up.

For example, if I set MaxConnectioinIdle to 5 minutes, the connection is automatically closed after 5 minutes of no requests, and then a new connection is created.

Analyzing MySQL Connections

Let's analyze the impact of MySQL connection pooling settings on the connection pool. For example, if I set the parameters to

`Go maxIdleConns = 5 maxOpenConns = 8 maxLifetime = time.Minute ... conn.SetMaxIdleConns(maxIdleConns) conn.SetMaxOpenConns(maxOpenConns) conn.SetConnMaxLifetime(maxLifetime) `

We set MaxIdleConns and MaxOpenConns to different values, then we do a benchmark test with hey

`Shell hey -c 10 -z 10s "http://localhost:8888/lookup?url=go-zero.dev" `

We did a benchmark test with a concurrency of 10QPS and a duration of 10 seconds, and the connection result is as follows.

As we can see.

2000+ connections were established in 10 seconds
The service constantly closing existing connections and reopening new ones
Each time the connection is used and put back, it may exceed MaxIdleConns, and then the connection will be closed
When a new request comes to get the connection, it is found that the number of connections is less than MaxOpenConns, but there are no more connections available, so a new connection is created.

This is the reason why we often see a lot of TIME_WAIT connections for MySQL.

Then we set MaxIdleConns and MaxOpenConns to the same value and do the same benchmark test again:

We can see that.

8 connections have been maintained constant
After a minute of idle (ConnMaxLifetime), all connections are closed

The ConnMaxLifetime here must be set smaller than the wait_timeout, you can get the wait_timeout value from the following method.

I recommend setting a value of less than 5 minutes, because some switches will clean up idle connections for 5 minutes, for example, when we are doing social apps, the heartbeat packets usually do not exceed 5 minutes. The specific reasons can be seen in

https://github.com/zeromicro/go-zero/blob/master/core/stores/sqlx/sqlmanager.go#L65

There is a paragraph in issue #257 of go-sql-driver that also talks about ConnMaxLifetime, as follows.

14400 sec is too long. one minutes is enough for most use cases.
Even if you configure entire your DC (OS, switch, router, etc...) , TCP connection may be lost from various reasons. (bugs in router firmware, unstable power voltage, electric nose, etc...)

So if you don't know how to set MySQL connection pool parameters, you can refer to go-zero's settings.

Also, ConnMaxIdleTime has no effect on the above benchmark test results, so you don't really need to set it.

If you have any questions about the above settings, or if you think something is wrong, please feel free to leave comments and let's discuss.

Project address

tproxy: https://github.com/kevwan/tproxy

go-zero: https://github.com/zeromicro/go-zero

Welcome to use and star to support us!

Sending type-safe HTTP requests with Go

Kevin Wan — Mon, 23 May 2022 04:58:37 +0000

Scenarios

For Gophers, we basically write code for clients to request. Sometimes we need to request RESTful APIs provided by third parties. At this time, we feel that it's hard to assemble the requests, not difficult, but error-prone.

For example, if we want to send a request like this, it looks very simple, but it's still tedious to actually write it.

POST /articles/5/update?device=ios HTTP/1.1
Host: go-zero.dev
Authorization: Bearer <jwt-token>

{"author": "kevin", "body": "this is not important!", "title": "my title", "type":6}

Go native way

This API is actually quite simple and we can write it directly from scratch.

func main() {
    var buf bytes.
    encoder := json.NewEncoder(&buf)
    params := map[string]interface{}{
        "title":  "my title",
        "body":   "this is not important!",
        "author": "kevin",
        "type":   6,
    }
    if err := encoder.Encode(params); err ! = nil {
        fmt.Fprintln(os.Stderr, err)
        return
    }

    url := fmt.Sprintf("http://go-zero.dev/articles/%d/update?device=%s", 5, "ios")
    req, err := http.NewRequest(http.MethodPost, url, &buf)
    if err ! = nil {
        fmt.Fprintln(os.Stderr, err)
        return
    }

    req.Header.Add("Authorization", "Bearer <jwt-token>")
    cli := http.Client{}
    resp, err := cli.Do(req)
    if err ! = nil {
        fmt.Fprintln(os.Stderr, err)
        return
    }

    io.Copy(os.Stdout, resp.Body)
}

We ran a test and found that we didn't get 200 OK, dumped the packet and the request looks as follows. Can you think of the reason for the failure?

POST /articles/5/update?device=ios HTTP/1.1
Host: go-zero.dev
User-Agent: Go-http-client/1.1
Content-Length: 79
Authorization: Bearer <jwt-token>
Accept-Encoding: gzip

{"author": "kevin", "body": "this is not important!", "title": "my title", "type":6}

The specific reasons for the failure will be discussed below, so let's explain this code first. You can see that the map[string]interface{} is used for the splicing parameters, for each field we can't check if the type matches. Only when we send it out and receive 200 OK from the server, we can confirm that it is passed correctly. For example, the type parameter is used here as int type, we may write it as string type by mistake. But it is still hard to find out that this parameter is written incorrectly without requesting it.

So let's see how the httpc package in go-zero is used for type safety.

`httpc` implementation

Let's see how the code for requesting with the httpc package is written.

const url = "http://go-zero.dev/articles/:id/update"

type UpdateArticle struct {
    ID            int    `path: "id"`
    Device        string `form: "device,options=ios,android,web,desktop"`
    Authorization string `header: "Authorization"`
    Title         string `json: "title"`
    Body          string `json: "body"`
    Author        string `json: "author"`
    Type          int    `json: "type"`
}

func main() {
    data := &UpdateArticle{
        ID:            5,
        Device:        "ios",
        Authorization: "Bearer <jwt-token>",
        Title:         "my title",
        Body:          "this is not important!",
        Author:        "kevin",
        Type:          6,
    }

    resp, err := httpc.Do(context.Background(), http.MethodPost, url, data)
    if err ! = nil {
        fmt.Fprintln(os.Stderr, err)
        return
    }

    io.Copy(os.Stdout, resp.Body)
}

Let's verify the code by sending a request, the result is as expected.

POST /articles/5/update?device=ios HTTP/1.1
Host: go-zero.dev
User-Agent: Go-http-client/1.1
Content-Length: 79
Content-Type: application/json; charset=utf-8
Authorization: Bearer <jwt-token>
Accept-Encoding: gzip

{"author": "kevin", "body": "this is not important!", "title": "my title", "type":6}

Did you find out that, in contrast to the previous one, there is one more header was set, Content-Type: application/json; charset=utf-8, and we forgot to set Content-Type in the previous code.

httpc implementation is very easy to understand by defining the type of the request and sending with Do. With support for path, form, header and json as shown in our code, it is very easy and type-safe to send HTTP requests.

More Capabilities

In addition to the ease of use and type-safety shown above, the httpc package has the following features.

timeout control with context. You can pass ctx for the requests.
automatic integration of OpenTelemetry. The trace-id, span-id returned by the server will be automatically written in the log, so that the client and server can work together to investigate the problems.
use httpc.Service to get the circuit breaker ability. When the server side has problems, it will automatically stop sending the requests, to avoid useless requests and reduce the pressure on the server side.

Project address

https://github.com/zeromicro/go-zero

Welcome to use go-zero and star to support us!

Developing a RESTful API with Go

Kevin Wan — Sun, 15 May 2022 14:29:19 +0000

When to use RESTful API

For most startups, we should focus more on delivering the products in the early stage of business. The monolithic services have the advantages of simple architecture, easy deployment, and better development productivity, which can help us achieve the product requirements quickly. While we use monolithic services to deliver products quickly, we also need to reserve the possibility for business incresement, so we usually split different business modules clearly in monolithic services.

Shopping mall monolithic service architecture

We take the mall as an example to build a monolithic service. The mall service is generally relatively complex and consists of multiple modules, the more important modules include account, product and order modules, etc. Each module will have its own independent business logic, and each module will also depend on some others. For example, the order module and the product module will depend on the account module. In the monolithic application this kind of dependency is usually accomplished by method calls between modules. Monolithic services generally share storage resources, such as MySQL and Redis.

The overall architecture of monolithic services is relatively simple, which is also the advantage of monolithic services. Customer requests are parsed through DNS and forwarded to the mall's backend services through Nginx. Mall services are deployed on cloud hosts. In order to achieve greater throughput and high availability, the service will generally deployed with multiple copies. This simple architecture can carry high throughput if well optimized.

For example, a request for order details interface /order/detail is routed to the order module, which relies on the account module and the product module to compose the complete order details back to the user, and multiple modules in a single service generally share the database and cache.

Monolithic Service

The next section describes how to quickly implement a mall monolithic service based on go-zero. Devs who have used go-zero know that we provide an API format file to describe the Restful API, and then we can generate the corresponding code by goctl with one command, we just need to fill in the corresponding business logic in the logic files. The mall service contains several modules, and in order to make the modules independent from each other, different modules are defined by separate APIs, but all the APIs are defined for the same service (mall-api).

Create user.api, order.api, product.api and mall.api in the api directory, where mall.api is the aggregated api file. Other api files are imported via import directives.

api
|-- mall.api
|-- order.api
|-- product.api
|-- user.api

Mall API Definition

mall.api is defined as follows, where syntax = "v1" means that this is the v1 syntax of zero-api.

syntax = "v1"

import "user.api"
import "order.api"
import "product.api"

Account module API definition

View user details
Get all orders for a user

user.api is defined as follows.

syntax = "v1"

type (
    UserRequest {
        ID int64 `path:"id"`
    }

    UserReply {
        ID      int64   `json:"id"`
        Name    string  `json:"name"`
        Balance float64 `json:"balance"`
    }

    UserOrdersRequest {
        ID int64 `path:"id"`
    }

    UserOrdersReply {
        ID       string `json:"id"`
        State    uint32 `json:"state"`
        CreateAt string `json:"create_at"`
    }
)

service mall-api {
    @handler UserHandler
    get /user/:id (UserRequest) returns (UserReply)

    @handler UserOrdersHandler
    get /user/:id/orders (UserOrdersRequest) returns (UserOrdersReply)
}

Order module API definition

Get order details
Generate orders

order.api is defined as follows.

syntax = "v1"

type (
    OrderRequest {
        ID string `path:"id"`
    }

    OrderReply {
        ID       string `json:"id"`
        State    uint32 `json:"state"`
        CreateAt string `json:"create_at"`
    }

    OrderCreateRequest {
        ProductID int64 `json:"product_id"`
    }

    OrderCreateReply {
        Code int `json:"code"`
    }
)

service mall-api {
    @handler OrderHandler
    get /order/:id (OrderRequest) returns (OrderReply)

    @handler OrderCreateHandler
    post /order/create (OrderCreateRequest) returns (OrderCreateReply)
}

Product module API definition

View product details

product.api is defined as follows.

syntax = "v1"

type ProductRequest {
    ID int64 `path:"id"`
}

type ProductReply {
    ID    int64   `json:"id"`
    Name  string  `json:"name"`
    Price float64 `json:"price"`
    Count int64   `json:"count"`
}

service mall-api {
    @handler ProductHandler
    get /product/:id (ProductRequest) returns (ProductReply)
}

Generating the monolithic service

With the API already defined, generating a service with the API becomes very simple, we use goctl to generate the monolithic service code.

$ goctl api go -api api/mall.api -dir .

The generated code is structured as follows.

.
├── api
│ ├── mall.api
│ ├── order.api
│ ├── product.api
│ └── user.api
├── etc
│ └── mall-api.yaml
├─ internal
│ ├── config
│ │ └── config.go
│ ├── handler
│ │ ├── ordercreatehandler.go
│ │ ├── orderhandler.go
│ │ ├── producthandler.go
│ │ ├── routes.go
│ │ ├── userhandler.go
│ │ └─ userordershandler.go
│ ├─ logic
│ │ ├─ ordercreatelogic.go
│ │ ├── orderlogic.go
│ │ ├── productlogic.go
│ │ ├── userlogic.go
│ │ └── userorderslogic.go
│ ├── svc
│ │ └── servicecontext.go
│ └── types
│ └── types.go
└── mall.go

Let's explain the generated files.

api: holds the API description file
etc: used to define the project configuration, all configuration items can be written in mall-api.yaml
internal/config: the configuration definition of the service
internal/handler: the implementation of the handler corresponding to the routes defined in the API file
internal/logic: used to put the business logic corresponding to each route, the reason for the distinction between handler and logic is to make the business processing part as less dependent as possible, to separate HTTP requests from the logic processing code, and to facilitate the subsequent splitting into RPC service
internal/svc: used to define the dependencies of the business logic processing, we can create the dependent resources in the main function and pass them to handler and logic via ServiceContext
internal/types: defines the API request and response data structures
mall.go: the file where the main function is located, with the same name as the service in the API definition, minus the -api suffix

The generated service can be run without any modification:

$ go run mall.go
Starting server at 0.0.0.0:8888...

Implementing the business logic

Next, let's implement the business logic. The logic will be simple for demonstration purposes, not real business logic.

First, let's implement the logic of getting all orders for users. Since there is no order-related information in the user module, we need to rely on the order module to query the orders of users, so we add a dependency on OrderLogic in UserOrdersLogic.

type UserOrdersLogic struct {
    logx.Logger
    ctx        context.Context
    svcCtx     *svc.ServiceContext
    orderLogic *OrderLogic
}

func NewUserOrdersLogic(ctx context.Context, svcCtx *svc.ServiceContext) *UserOrdersLogic {
    return &UserOrdersLogic{
        Logger:     logx.WithContext(ctx),
        ctx:        ctx,
        svcCtx:     svcCtx,
        orderLogic: NewOrderLogic(ctx, svcCtx),
    }
}

Implement a method in OrderLogic to query all orders based on user id

func (l *OrderLogic) ordersByUser(uid int64) ([]*types.OrderReply, error) {
    if uid == 123 {
        // It should actually be queried from database or cache
        return []*types.OrderReply{
            {
                ID:       "236802838635",
                State:    1,
                CreateAt: "2022-5-12 22:59:59",
            },
            {
                ID:       "236802838636",
                State:    1,
                CreateAt: "2022-5-10 20:59:59",
            },
        }, nil
    }

    return nil, nil
}

Call the ordersByUser method in the UserOrders method of UserOrdersLogic.

func (l *UserOrdersLogic) UserOrders(req *types.UserOrdersRequest) (*types.UserOrdersReply, error) {
    orders, err := l.orderLogic.ordersByUser(req.ID)
    if err ! = nil {
        return nil, err
    }

    return &types.UserOrdersReply{
        Orders: orders,
    }, nil
}

At this point we restart the mall-api service and request all the user's orders in the browser.

http://localhost:8888/user/123/orders

The return result is as follows, as we expected

{
    "orders": [
        {
            "id": "236802838635",
            "state": 1,
            "create_at": "2022-5-12 22:59:59"
        },
        {
            "id": "236802838636",
            "state": 1,
            "create_at": "2022-5-10 20:59:59"
        }
    ]
}

Next we'll implement the logic for creating an order. To create an order we first need to see if the item in stock is enough, so we need to rely on the item module in the order module.

type OrderCreateLogic struct {
    logx.Logger
    ctx          context.Context
    svcCtx       *svc.ServiceContext
    productLogic *ProductLogic
    productLogic *ProductLogic
}

func NewOrderCreateLogic(ctx context.Context, svcCtx *svc.ServiceContext) *OrderCreateLogic {
    return &OrderCreateLogic{
        Logger:       logx.WithContext(ctx),
        ctx:          ctx,
        svcCtx:       svcCtx,
        productLogic: NewProductLogic(ctx, svcCtx),
    }
}

The logic for creating an order is as follows.

const (
    success = 0
    failure = -1
)

func (l *OrderCreateLogic) OrderCreate(req *types.OrderCreateRequest) (*types.OrderCreateReply, error) {
    product, err := l.productLogic.productByID(req.ProductID)
    if err ! = nil {
        return nil, err
    }

    if product.Count > 0 {
        return &types.OrderCreateReply{Code: success}, nil
    }

    return &types.OrderCreateReply{Code: failure}, nil
}

The logic of the dependent product module is as follows.

func (l *ProductLogic) Product(req *types.ProductRequest) (*types.ProductReply, error) {
    return l.productByID(req.ID)
}

func (l *ProductLogic) productByID(id int64) (*types.ProductReply, error) {
    return &types.ProductReply{
        ID: id,
        Name: "apple watch 3",
        Price: 3333.33,
        Count: 99,
    }, nil
}

The above shows that using go-zero to develop a monolithic service is very simple, which helps us to develop quickly. And we also separated modules, which also provides the possibility of changing to microservices later.

Summary

The above example shows that it is very simple to use go-zero to develop monolithic services. You only need to define the api file, and then the goctl tool can automatically generate the project code. We only need to fill in the business logic code in the logic package. In this article we just demonstrated how to quickly develop monolithic services based on go-zero, which does not involve databases. In fact, goctl can also generate CRUD and cache code with one command.

And for different business scenarios, customization can also be achieved through customizing templates. And customized templates can be shared within the team through remote git repositories, which can be very efficient for team collaboration.

Project address

https://github.com/zeromicro/go-zero

Welcome to use go-zero and star to support us!

How we define and load configuration from files.

Kevin Wan — Mon, 09 May 2022 13:50:19 +0000

Background

When we write applications, we basically use configuration files, from various shells to nginx, etc., all have their own configuration files. Although this is not too difficult, the configuration items are relatively cumbersome, and parsing and verifying them can be troublesome. In this article, I will tell how we simplify the definition and parsing of configuration files.

Scenario

If we want to write a Restful API service, the configuration items will probably have the following contents.

Host, the listening IP, if not filled, the default is 0.0.0.0
Port, the port to listen to, required, can only be a number, greater than or equal to 80, less than 65535
LogMode, logging mode, only file or console can be selected
Verbose, see whether to output detailed logs, optional, default is false
MaxConns, the maximum number of concurrent connections allowed, default 10000
Timeout, timeout setting, default 3s
CpuThreshold, sets the threshold for CPU usage to trigger load shedding, default 900, 1000m means 100%

Previously we used json for the configuration file, but there was a problem with json that we couldn't add comments, so we switched to yaml format.

Now let's see how easy it is to define and parse such a configuration file with go-zero ~

Defining the configuration

First, we need to define the above configuration requirements into a Go structure, as follows.

RestfulConf struct {
    Host string `json:",default=0.0.0.0"`
    Port int `json:",range=[80,65535)"`
    LogMode string `json:",options=[file,console]"`
    Verbose bool `json:",optional"`
    MaxConns int `json:",default=10000"`
    Timeout time.Duration `json:",default=3s"`
    CpuThreshold int64 `json:",default=900,range=[0:1000]"`
}

As you can see, we have certain definitions and restrictions for each configuration item, some of which are defined as follows.

default, the default value is used if the configuration is not filled, you can see that 3s in it will be automatically resolved into time.Duration type
optional, you can leave the item blank, if not, use the zero value of its type
range, which restricts the number type to the given range
options, which restricts the configured value to only one of the given values

And, some attributes can be used together, such as.

default and range can be used together to both add a range restriction and provide a default value
default and options can be used together to add option restrictions and provide a default value

Configuration files

Because we give a lot of default values when defining the configuration, as well as using optional to specify as optional, so our configuration file has relatively few configuration items that can use the default values without writing explicitly, as follows.

# Because many of them have default values, you only need to write the ones
# that need to be specified and the ones that don't have default values
Port: 8080
LogMode: console
# You can read the values of environment variables
MaxBytes: ${MAX_BYTES}

Here is a note, if the value of the configuration chars are all numbers, and you define the configuration type is string, for example, some people often use 123456 to test the password, but the password will generally be defined as string, the configuration should be written as follows (just an example, passwords are generally not recommended to be written in the configuration file)

Password: "123456"

Here the double quotes can not be missed, otherwise type mismatch will be reported, because yaml parser will parse 123456 as int.

Loading configuration files

We have the configuration definition (config.go) and the configuration file (config.yaml), the next step is to load the configuration file, which can be done in three ways.

must be loaded successfully, otherwise the program exits, we generally use so, if the configuration is not correct, the program will not be able to continue

// If error, exit the program directly
var config RestfulConf
conf.MustLoad("config.yaml", &config)

The default code generated by go-zero's goctl tool also uses MustLoad to load the configuration file

Load the configuration and handle the error on your own

// handle errors on your own
var config RestfulConf
// For simplicity, LoadConfig will be changed to Load later,
// LoadConfig has been marked as Deprecated.
if err := conf.LoadConfig("config.yaml", &config); err ! = nil {
    log.Fatal(err)
}

Load the configuration and read the environment variables

// Automatically read environment variables
var config RestfulConf
conf.MustLoad(configFile, &config, conf.UseEnv())

Here why we need to explicitly specify conf.UseEnv(), because if default to use, you may need to escape when writing specific characters in the configuration, so the default does not read the environment variables.

Implementation principle

We usually use encoding/json or the corresponding yaml library when implementing yaml/json parsing, but for go-zero, we need to have more precise control over unmarshal, which leads us to customize yaml/json parsing ourselves, the complete code is here.

Configuration file code: https://github.com/zeromicro/go-zero/tree/master/core/conf

yaml/json parsing code: https://github.com/zeromicro/go-zero/tree/master/core/mapping

This is also a good example of how reflect can be used and how unit tests can be used to ensure correctness in complex scenarios.

Summary

I always recommend the idea of Fail Fast, we do the same on loading configuration files, once there is an error, immediately exit, so that ops will find the problem in time when deploying services, because the process can not get up running.

All the configuration items of go-zero services are loaded and automatically verified in this way, including the configuration of many of the tools I wrote are also based on this, hope it helps!

Project address

https://github.com/zeromicro/go-zero

Welcome to use go-zero and star to support us!

A load shedding protector against bursting requests

Kevin Wan — Tue, 03 May 2022 02:27:51 +0000

Why do we need load shedding?

In a microservice cluster, the invocation chain is complex, and as a service provider, it needs a mechanism to protect itself from being overwhelmed by mindless invocations and to ensure the high availability of its own services.

The most common protection mechanism is flow limiting mechanism, the premise of using flow limiter is to know the maximum number of concurrency that it can handle, usually through pressure testing to get the maximum number of concurrency before going online, and the flow limiting parameters of each interface are different in the daily request process, while the system has been constantly iterating its processing capacity will often change, each time before going online, it is necessary to pressure test and then adjust the flow limiting The parameters become very tedious.

So is there a more concise flow limiting mechanism that can achieve maximum self-protection?

What is adaptive load shedding

Adaptive load shedding is a very intelligent way to protect the service itself, and dynamically determine whether it needs to be downgraded based on the system load of the service itself.

Design goals.

to ensure that the system is not dragged down.
maintain the system throughput under the premise of system stability.

The key then is how to measure the load of the service itself?

Judging high load depends on two main indicators.

whether the cpu is overloaded.
whether the maximum number of concurrency is overloaded.

When the above two points are satisfied at the same time, it means that the service is in a high load state, and adaptive load reduction is performed.

It should also be noted that the cpu load and concurrency of the high concurrency scenario tend to fluctuate greatly, and we call this phenomenon burr from the data, which may lead to frequent auto-dowload operations, so we generally obtain the average value of the indicators over a period of time to make the indicators more smooth. The implementation can be done by accurately recording the metrics over a period of time and then directly calculating the average value, but it takes up some system resources.

There is a statistical algorithm: exponential moving average, which can be used to estimate the local average of variables, so that the update of variables is related to the historical values over time, and the average can be estimated without recording all the historical local variables, which saves valuable server resources.

Principle of Sliding Average Algorithm Refer to this article for a very clear explanation.

The variable V is noted as Vt at time t, and θt is the value of the variable V taken at time t. That is, Vt=θt when the sliding average model is not used, and after using the sliding average model, Vt is updated by the following formula.

Vt=β⋅Vt-1+(1-β)⋅θt

Vt = θt for β = 0
β = 0.9, which is approximately the average of the last 10 θt values
β = 0.99, which is approximately the average of the last 100 θt values

Code Implementation

Next, let's look at the code implementation of go-zero adaptive load reduction.

core/load/adaptiveshedder.go

Adaptive downgrade interface definition.

// callback function
Promise interface {
    // This function is called back when the request succeeds
    Pass()
    // Callback when the request fails
    Fail()
}

// Drop the interface definition
Shedder interface {
    // Drop check
    // 1. Allow the call, you need to manually execute Promise.accept()/reject() to report the actual execution task structure
    // 2. Reject the call and it will return err: Service overloaded error ErrServiceOverloaded
    Allow() (Promise, error)
}

The interface definition is very concise meaning it is actually very simple to use, exposing an `Allow()(Promise, error) to the outside world.

Example of go-zero usage.

The business only needs to call the method to determine whether to dowload, if it is dowloaded then directly end the process, otherwise the implementation of the business finally use the return value Promise according to the results of the implementation of the callback results can be.

func UnarySheddingInterceptor(shedder load.Shedder, metrics *stat.Metrics) grpc.UnaryServerInterceptor {
    ensureSheddingStat()

    return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler) (val interface{}, err error) {
        sheddingStat.IncrementTotal()
        var promise load.Promise
        // Check if it was downgraded
        promise, err = shedder.Allow()
        // drop the load and record the relevant logs and metrics
        if err ! = nil {
            metrics.AddDrop()
            sheddingStat.IncrementDrop()
            return
        }
        // Final callback execution result
        defer func() {
            // Failed to execute
            if err == context.DeadlineExceeded {
                promise.Fail()
            // Execution succeeded
            } else {
                sheddingStat.IncrementPass()
                Promise.Pass()
            }
        }()
        // Execute the business method
        return handler(ctx, req)
    }
}

Interface implementation class definition.

There are three main types of properties

cpu load threshold: exceeding this value means the cpu is in a high load state. 2.
cooling period: if the service has been down loaded before, then it will enter the cooling period, the purpose is to prevent the load from being down during the process of down loading and immediately pressurized resulting in back and forth jitter. Because it takes some time to reduce the load, you should continue to check whether the number of concurrency exceeds the limit during the cooling-off period, and continue to discard requests if the limit is exceeded.
concurrency: the number of concurrency currently being processed, the average number of concurrency currently being processed, and the number of requests and response time in the most recent period, in order to calculate whether the number of concurrency currently being processed is greater than the maximum number of concurrency that can be carried by the system.

// option parameter pattern
ShedderOption func(opts *shedderOptions)

// Optional configuration parameters
shedderOptions struct {
    // sliding time window size
    window time.Duration
    // Number of sliding time windows
    buckets int
    // cpu load threshold
    cpuThreshold int64
cpuThreshold int64 }

// adaptiveShedder struct, need to implement Shedder interface
adaptiveShedder struct {
    // cpu load threshold
    // higher than the threshold means high load needs to be down loaded to ensure service
    cpuThreshold int64
    // How many buckets in 1s
    windows int64
    // number of concurrent
    flying int64
    // Number of concurrent sliding smoothing
    avgFlying float64
    // Spin lock, one service shares one drop
    // Lock must be applied when counting the number of requests currently being processed
    // lossless concurrency, improves performance
    avgFlyingLock syncx.SpinLock
    // Last rejection time
    dropTime *syncx.AtomicDuration
    // whether it has been rejected recently
    droppedRecently *syncx.AtomicBool
    // request count statistics, with a sliding time window to record metrics for the most recent period
    passCounter *collection.RollingWindow
    // response time statistics, with a sliding time window to record metrics over the most recent period
    rtCounter *collection.RollingWindow
}

Adaptive dropload constructor.

func NewAdaptiveShedder(opts . .ShedderOption) Shedder {
    // To ensure code uniformity
    // return the default empty implementation when the developer closes, to achieve code uniformity
    // go-zero uses this design in many places, such as Breaker, the logging component
    if !enabled.True() {
        return newNopShedder()
    }
    // options mode sets optional configuration parameters
    options := shedderOptions{
        // Default statistics for the last 5s
        window: defaultWindow,
        // default bucket count of 50
        buckets: defaultBuckets,
        // cpu load
        cpuThreshold: defaultCpuThreshold,
    }
    for _, opt := range opts {
        opt(&options)
    }
    // Calculate the interval for each window, default is 100ms
    bucketDuration := options.window / time.Duration(options.buckets)
    return &adaptiveShedder{
        // cpu load
        cpuThreshold: options.cpuThreshold,
        // How many sliding window units are contained in 1s of time
        windows: int64(time.Second / bucketDuration),
        // last rejection time
        dropTime: syncx.NewAtomicDuration(),
        // whether it has been rejected recently
        droppedRecently: syncx.NewAtomicBool(),
        // qps statistics, sliding time window
        // ignore the current writing window (bucket), incomplete time period may lead to data exceptions
        NewRollingWindow(options.buckets, bucketDuration.), // passCounter: collection,
            IgnoreCurrentBucket()),
        // Response time statistics, sliding time window
        // Ignore the current writing window (bucket), incomplete time period may lead to data exceptions
        rtCounter: collection.NewRollingWindow(options.buckets, bucketDuration,
            collection.IgnoreCurrentBucket()),
    }
}

Drop check Allow().

Check if the current request should be dropped, and if it is dropped the business side needs to interrupt the request directly to protect the service, which also means that the downgrade takes effect while entering the cooling off period. If released then return promise and wait for the business side to execute the callback function to perform the metrics statistics.

// Drop check
func (as *adaptiveShedder) Allow() (Promise, error) {
    // Check if the request was dropped
    if as.shouldDrop() {
        // Set the drop time
        as.dropTime.Set(timex.Now())
        // Recently dropped
        as.droppedRecently.Set(true)
        // return overloaded
        return nil, ErrServiceOverloaded
    }
    // Add 1 to the number of requests being processed
    as.addFlying(1)
    // Each allowed request here returns a new promise object
    // promise holds the drop pointer object internally
    return &promise{
        start: timex.Now(),
        shedder: as,
    }, nil
}

Check if shouldDrop().

// whether the request should be dropped or not
func (as *adaptiveShedder) shouldDrop() bool {
    // The current cpu load exceeds the threshold
    // The service should continue to check the load and try to drop the request while it is on cooldown
    if as.systemOverloaded() || as.stillHot() {
        // Check if the concurrency being processed exceeds the current maximum number of concurrency that can be hosted
        // Discard the request if it exceeds it
        if as.highThru() {
            flying := atomic.LoadInt64(&as.flying)
            as.avgFlyingLock.Lock()
            avgFlying := as.avgFlying
            as.avgFlyingLock.Unlock()
            msg := fmt.Sprintf(
                "dropreq, cpu: %d, maxPass: %d, minRt: %.2f, hot: %t, flying: %d, avgFlying: %.2f",
                stat.CpuUsage(), as.maxPass(), as.minRt(), as.stillHot(), flying, avgFlying)
            logx.Error(msg)
            stat.Report(msg)
            return true
        }
    }
    return false
}

cpu threshold check systemOverloaded().

The cpu load value calculation algorithm uses a sliding average algorithm to prevent burrs. Sampling every 250ms β is 0.95, which is roughly equivalent to the average of 20 cpu loadings in history, with a time period of about 5s.

// whether the cpu is overloaded
func (as *adaptiveShedder) systemOverloaded() bool {
    return systemOverloadChecker(as.cpuThreshold)
}

// cpu checker function
systemOverloadChecker = func(cpuThreshold int64) bool {
        return stat.CpuUsage() >= cpuThreshold
}

// cpu sliding average
curUsage := internal.RefreshCpu()
prevUsage := atomic.LoadInt64(&cpuUsage)
// cpu = cpuᵗ-¹ * beta + cpuᵗ * (1 - beta)
// sliding average algorithm
usage := int64(float64(prevUsage)*beta + float64(curUsage)*(1 - beta))
atomic.StoreInt64(&cpuUsage, usage)

Check if the system is in a cooling period stillHot:

Determine if the current system is in a cooling period, if it is, it should continue to try to check if the request is dropped. The main purpose is to prevent the system in the process of recovery from overload load has not yet come down, immediately increase the pressure again resulting in back and forth jitter, at this time should try to continue to discard the request.

func (as *adaptiveShedder) stillHot() bool {
    // No requests have been dropped recently
    // means the service is fine
    if !as.droppedRecently.True() {
        return false
    }
    // Not on cooldown
    dropTime := as.dropTime.Load()
    if dropTime == 0 {
        return false
    }
    // cool down time defaults to 1s
    hot := timex.Since(dropTime) < coolOffDuration
    // not in the cool off period, normal processing of requests in progress
    if !hot {
        // reset the drop record
        as.droppedRecently.Set(false)
    }

    return hot
}

Check the number of concurrency currently being processed highThru().

Once current concurrency being processed > concurrency carrying limit then go into a down load state.

Why do we need to add locks here? Because adaptive downgrading is used globally to ensure that the concurrency average is correct.

Why do we need to add spin locks here? Because the concurrent processing can be done without blocking other goroutine execution tasks, using lock-free concurrency to improve performance.

func (as *adaptiveShedder) highThru() bool {
    // locking
    as.avgFlyingLock.Lock()
    // Get the sliding average
    // update at the end of each request
    avgFlying := as.avgFlying
    // Unlock
    as.avgFlyingLock.Unlock()
    // Maximum number of concurrent requests at this point in the system
    maxFlight := as.maxFlight()
    // whether the number of concurrent processes and the average number of concurrent processes is greater than the system's maximum number of concurrent processes
    return int64(avgFlying) > maxFlight && atomic.LoadInt64(&as.flying) > maxFlight
}

How do I get the number of concurrency being processed versus the average concurrency?

The current concurrency count is actually very simple: +1 concurrency for each allowed request, -1 for the promise object callback after the request is completed, and the average concurrency can be solved using the sliding average algorithm.

type promise struct {
    // request start time
    // count the time spent on request processing
    start time.
    shedder *adaptiveShedder
shedder *adaptiveShedder }

func (p *promise) Fail() {
    // End of request, -1 for the number of requests currently being processed
    p.shedder.addFlying(-1)
}

func (p *promise) Pass() {
    // response time in milliseconds
    rt := float64(timex.Since(p.start)) / float64(time.Millisecond)
    // end of request, number of requests currently being processed -1
    p.shedder.addFlying(-1)
    p.shedder.rtCounter.Add(math.Ceil(rt))
    p.shedder.passCounter.Add(1)
}

func (as *adaptiveShedder) addFlying(delta int64) {
    flying := atomic.AddInt64(&as.flying, delta)
    // After the request ends, count the concurrency of requests currently being processed
    if delta < 0 {
        as.avgFlyingLock.Lock()
        // Estimate the average number of requests for the current service over the recent time period
        as.avgFlying = as.avgFlying*flyingBeta + float64(flying)*(1-flyingBeta)
        as.avgFlyingLock.Unlock()
    }
}

It's not enough to get the current system count, we also need to know the maximum number of concurrent requests the system can handle, i.e. the maximum number of concurrent requests.

The number of requests passed and the response time are achieved by a sliding window, which can be implemented in the Adaptive Fuse article.

The maximum concurrency of the current system = the maximum number of passes per unit time of the window * the minimum response time per unit time of the window.

// Calculate the maximum concurrency of the system per second
// max concurrency = max requests (qps) * min response time (rt)
func (as *adaptiveShedder) maxFlight() int64 {
    // windows = buckets per second
    // maxQPS = maxPASS * windows
    // minRT = min average response time in milliseconds
    // maxQPS * minRT / milliseconds_per_second
    // as.maxPass() * as.windows - maximum qps per bucket * number of buckets contained in 1s
    // as.minRt()/1e3 - the smallest average response time of all buckets in the window / 1000ms here to convert to seconds
    return int64(math.Max(1, float64(as.maxPass()*as.windows)*(as.minRt()/1e3)))
}    

// Sliding time window with multiple buckets
// find the one with the highest number of requests
// Each bucket takes up internal ms
// qps refers to the number of requests in 1s, qps: maxPass * time.Second/internal
func (as *adaptiveShedder) maxPass() int64 {
    var result float64 = 1
    // The bucket with the highest number of requests in the current time window
    as.passCounter.Reduce(func(b *collection.Bucket) {
        if b.Sum > result {
            result = b.Sum
        }
    })

    return int64(result)
}

// Sliding time window with multiple buckets
// Calculate the minimum average response time
// because it is necessary to calculate the maximum number of concurrency the system can handle in a recent period of time
func (as *adaptiveShedder) minRt() float64 {
    // default is 1000ms
    result := defaultMinRt

    as.rtCounter.Reduce(func(b *collection.Bucket) {
        if b.Count <= 0 {
            return
        }
        // request average response time
        avg := math.Round(b.Sum / float64(b.Count))
        if avg < result {
            result = avg
        }
    })

    return result
}

Reference

Google BBR Congestion Control Algorithm

Principle of sliding average algorithm

go-zero adaptive load shedding

Project address

https://github.com/zeromicro/go-zero

Feel free to use go-zero and star to support us!

My best practices on Go fuzzing

Kevin Wan — Tue, 03 May 2022 02:07:19 +0000

As programmers, we often hope that our code is bug-free! But the fact is that bug-free can only be disproved, not proven. The upcoming release of Go 1.18 officially provides a great tool to help us prove it (in most cases) - go fuzzing.

Go 1.18 is all about generics, but I really think go fuzzing is the most useful feature of Go 1.18 at the moment, even compared with generics!

In this article, we'll take a closer look at go fuzzing:

What is it?
How to use go fuzzing?
What are the best practices?

What is go fuzzing

According to the official documentation, go fuzzing is a way to automate tests by continuously giving different inputs to a program and analyzing the code coverage to intelligently find the failed cases. The problems found in the test are usually difficult to find.

How to use go fuzzing

The official rules for writing fuzz tests are.

The function must start with Fuzz, the only argument is *testing.F, and there is no return value
Fuzz tests must be in the *_test.go file
The fuzz target in the above image is a method call (*testing.F).Fuzz with *testing.T as the first argument, and then arguments called fuzzing arguments, with no return value
There can be only one fuzz target in each fuzz test
Calling f.Add(...) requires that the arguments be of the same order and type as the fuzzing arguments
fuzzing arguments only supports the following types.
- string, []byte
- int, int8, int16, int32/rune, int64
- uint, uint8/byte, uint16, uint32, uint64
- float32, float64
- bool
fuzz target Do not rely on global state, it will run in parallel.

Run `fuzzing tests`

If I write a fuzzing test, e.g.



// See https://github.com/zeromicro/go-zero/blob/master/core/mr/mapreduce_fuzz_test.go for the specific code
func FuzzMapReduce(f *testing.F) {
  ...
}

Then we can execute it like this.



go test -fuzz=MapReduce

We would get something like the following result.



fuzz: elapsed: 0s, gathering baseline coverage: 0/2 completed
fuzz: elapsed: 0s, gathering baseline coverage: 2/2 completed, now fuzzing with 10 workers
fuzz: elapsed: 3s, execs: 3338 (1112/sec), new interesting: 56 (total: 57)
fuzz: elapsed: 6s, execs: 6770 (1144/sec), new interesting: 62 (total: 63)
fuzz: elapsed: 9s, execs: 10157 (1129/sec), new interesting: 69 (total: 70)
fuzz: elapsed: 12s, execs: 13586 (1143/sec), new interesting: 72 (total: 73)
^Cfuzz: elapsed: 13s, execs: 14031 (1084/sec), new interesting: 72 (total: 73)
PASS
ok github.com/zeromicro/go-zero/core/mr 13.169s

The ^C is because I pressed Ctrl-C to terminate the test, see the official documentation for details.

best practices in go-zero

Based on my experience with go-zero, I've summarized the best practices in four preliminary steps.

define fuzzing arguments, first figure out how to define fuzzing arguments, and write fuzzing target with the given fuzzing arguments. 2.
think about how to write fuzzing target, the focus here is on how to verify the correctness of the results, because fuzzing arguments are given "randomly", so there should be a general method to verify the results
think about how to print the result of a failed case, so that a new unit test can be generated
write a new unit test based on the results of the failed fuzzing test, this new unit test will be used to debug the problems found by the fuzzing test and solidify them for the CI.

The next step is to show the above steps with a simple array summation function. The actual case in go-zero is slightly more complicated, and I will give the internal go-zero landing case at the end of the article for you to write complex scenarios.

Here is a bug-injected implementation of the summation code.



func Sum(vals []int64) int64 {
  var total int64

  for _, val := range vals {
    if val%1e5 ! = 0 {
      total += val
    }
  }

  return total
}

1. Define `fuzzing arguments`

You need to give at least one fuzzing argument, otherwise go fuzzing can't generate test code, so even if we don't have good input, we need to define a fuzzing argument that will have an impact on the result, here we use the number of slice elements as fuzzing arguments, and then Go fuzzing will automatically generate different arguments to simulate the test based on the code coverage that is run.



func FuzzSum(f *testing.F) {
  f.Add(10)
  f.Fuzz(func(t *testing.T, n int) {
    n %= 20
    ...
  })
}

Here n is the number of elements that go fuzzing can simulate in a slice. To make sure the number of elements is not too high, we limit it to 20 (0 is fine) and we add a corpus of 10 (called corpus in go fuzzing), which is the value that makes go fuzzing cold start, it doesn't matter how much it is.

2. How to write the `fuzzing target`

This step focuses on writing a verifiable fuzzing target, writing test code based on the given fuzzing arguments, and generating data to verify the correctness of the results.

For our Sum function, it is actually relatively simple to generate a random slice of n elements and then sum up the expected result. As follows.



func FuzzSum(f *testing.F) {
  rand.Seed(time.Now().UnixNano())

  f.Add(10)
  f.Fuzz(func(t *testing.T, n int) {
    n %= 20
    var vals []int64
    var expect int64
    for i := 0; i < n; i++ {
      val := rand.Int63() % 1e6
      vals = append(vals, val)
      expect += val
    }

    assert.Equal(t, expect, Sum(vals))
  })
}

This code is still very easy to understand, their own summation and Sum summation to do a comparison only, will not explain in detail. But complex scenarios you need to think carefully about how to write the verification code, but this will not be too difficult, too difficult, it may not be enough understanding of the test function or simplify.

At this point, you can run fuzzing tests with the following command, and the result will be similar to the following.



$ go test -fuzz=Sum
fuzz: elapsed: 0s, gathering baseline coverage: 0/2 completed
fuzz: elapsed: 0s, gathering baseline coverage: 2/2 completed, now fuzzing with 10 workers
fuzz: elapsed: 0s, execs: 6672 (33646/sec), new interesting: 7 (total: 6)
--- FAIL: FuzzSum (0.21s)
    --- FAIL: FuzzSum (0.00s)
        sum_fuzz_test.go:34:
              Error Trace: sum_fuzz_test.go:34
                                  value.go:556
                                  value.go:339
                                  fuzz.go:334
              Error: Not equal:
                            expected: 8736932
                            actual : 8636932
              Test: FuzzSum

    Failing input written to testdata/fuzz/FuzzSum/739002313aceff0ff5ef993030bbde9115541cabee2554e6c9f3faaf581f2004
    To re-run:
    go test -run=FuzzSum/739002313aceff0ff5ef993030bbde9115541cabee2554e6c9f3faaf581f2004
FAIL
exit status 1
FAIL github.com/kevwan/fuzzing 0.614s

So here's the problem! We see the result is not right, but we can hardly analyze why it is not right, you taste carefully, this output above, how do you analyze?

3. How to print the input for the failed case

For the above failed test, if we can print out the input and form a simple test case, then we can debug it directly. It is better to copy/paste the printed input directly to the new test case, if the format is not right, you need to adjust the format line by line for so many lines of input is too tired, and it may not be only one failure case.

So we changed the code to the following.



func FuzzSum(f *testing.F) {
  rand.Seed(time.Now().UnixNano())

  f.Add(10)
  f.Fuzz(func(t *testing.T, n int) {
    n %= 20
    var vals []int64
    var expect int64
    var buf strings.
    buf.WriteString("\n")
    for i := 0; i < n; i++ {
      val := rand.Int63() % 1e6
      vals = append(vals, val)
      expect += val
      buf.WriteString(fmt.Sprintf("%d,\n", val))
    }

    assert.Equal(t, expect, Sum(vals), buf.String())
  })
}

Running the command again gives the following result.



$ go test -fuzz=Sum
fuzz: elapsed: 0s, gathering baseline coverage: 0/2 completed
fuzz: elapsed: 0s, gathering baseline coverage: 2/2 completed, now fuzzing with 10 workers
fuzz: elapsed: 0s, execs: 1402 (10028/sec), new interesting: 10 (total: 8)
--- FAIL: FuzzSum (0.16s)
    --- FAIL: FuzzSum (0.00s)
        sum_fuzz_test.go:34:
              Error Trace: sum_fuzz_test.go:34
                                  value.go:556
                                  value.go:339
                                  fuzz.go:334
              Error: Not equal:
                            expected: 5823336
                            actual : 5623336
              Test: FuzzSum
              Messages:
                            799023,
                            110387,
                            811082,
                            115543,
                            859422,
                            997646,
                            200000,
                            399008,
                            7905,
                            931332,
                            591988,

    Failing input written to testdata/fuzz/FuzzSum/26d024acf85aae88f3291bf7e1c6f473eab8b051f2adb1bf05d4491bc49f5767
    To re-run:
    go test -run=FuzzSum/26d024acf85aae88f3291bf7e1c6f473eab8b051f2adb1bf05d4491bc49f5767
FAIL
exit status 1
FAIL github.com/kevwan/fuzzing 0.602s

4. Write a new test case

Based on the output of the failure case above, we can copy/paste the following code, but of course the framework is written by ourselves and the input parameters can be copied in directly.



func TestSumFuzzCase1(t *testing.T) {
  vals := []int64{
    799023,
    110387,
    811082,
    115543,
    859422,
    997646,
    200000,
    399008,
    7905,
    931332,
    591988,
  }
  assert.Equal(t, int64(5823336), Sum(vals))
}

This makes it easy to debug and to add a valid unit test to ensure that the bug never comes up again.

`go fuzzing` more experience

Go versioning issues

I believe that Go 1.18 has been released, and most projects will not immediately upgrade their online code to 1.18, so what if the testing.F introduced by go fuzzing doesn't work?

If the online (go.mod) is not upgraded to Go 1.18, but we are fully recommended to upgrade locally, then we just need to put the above FuzzSum into a file with a name like sum_fuzz_test.go and add the following directive to the header.



//go:build go1.18
// +build go1.18

Note: The third line must be a blank line, otherwise it will become a package comment.

This way we don't report errors online no matter which version we use, and we run fuzz testing usually locally and are not affected.

go fuzzing does not reproduce failures

The above steps are for simple cases, but sometimes the problem becomes complicated when the input from the failure case forms a new unit test that does not reproduce the problem (especially if there is a goroutine deadlock problem), as you can see in the following output.



go test -fuzz=MapReduce
fuzz: elapsed: 0s, gathering baseline coverage: 0/2 completed
fuzz: elapsed: 0s, gathering baseline coverage: 2/2 completed, now fuzzing with 10 workers
fuzz: elapsed: 3s, execs: 3681 (1227/sec), new interesting: 54 (total: 55)
...
fuzz: elapsed: 1m21s, execs: 92705 (1101/sec), new interesting: 85 (total: 86)
--- FAIL: FuzzMapReduce (80.96s)
    fuzzing process hung or terminated unexpectedly: exit status 2
    Failing input written to testdata/fuzz/FuzzMapReduce/ee6a61e8c968adad2e629fba11984532cac5d177c4899d3e0b7c2949a0a3d840
    To re-run:
    go test -run=FuzzMapReduce/ee6a61e8c968adad2e629fba11984532cac5d177c4899d3e0b7c2949a0a3d840
FAIL
exit status 1
FAIL github.com/zeromicro/go-zero/core/mr 81.471s

In this case, it just tells us that the fuzzing process is stuck or has ended abnormally, with a status code of 2. In this case, the re-run is not normally reproduced. Why does it simply return an error code of 2? I went through the source code of go fuzzing, and each fuzzing test is run by a separate process, and then go fuzzing throws away the process output of the fuzzing test and just shows the status code. So how do we solve this problem?

After careful analysis, I decided to write a regular unit test code like fuzzing test myself, which would ensure that the failures are in the same process and would print the error message to the standard output, the code is roughly as follows.



func TestSumFuzzRandom(t *testing.T) {
  const times = 100000
  rand.Seed(time.Now().UnixNano())

  for i := 0; i < times; i++ {
    n := rand.Intn(20)
    var vals []int64
    var expect int64
    var buf strings.
    buf.WriteString("\n")
    for i := 0; i < n; i++ {
      val := rand.Int63() % 1e6
      vals = append(vals, val)
      expect += val
      buf.WriteString(fmt.Sprintf("%d,\n", val))
    }

    assert.Equal(t, expect, Sum(vals), buf.String())
  }
}

This way we can do a simple simulation of go fuzzing ourselves, but with any errors we can get clear output. Here maybe we study through go fuzzing, or there are other ways to control it, if you know, thanks for letting me know.

But this simulated case that takes a long time to run, we wouldn't want it to be executed every time at CI, so I put it in a separate file named something like sum_fuzzcase_test.go and added the following directive to the header of the file.



//go:build fuzz
// +build fuzz

This way we can add -tags fuzz when we need to run this mock case, e.g.



go test -tags fuzz . /...

Complex usage examples

The above is an example, still relatively simple, if you encounter complex scenarios do not know how to write, you can first see how go-zero is landed go fuzzing, as follows.

MapReduce - https://github.com/zeromicro/go-zero/tree/master/core/mr
stringx - https://github.com/zeromicro/go-zero/tree/master/core/stringx

Project address

https://github.com/zeromicro/go-zero

Feel free to use go-zero and star to support us!

Best practices on developing monolithic services in Go

Kevin Wan — Mon, 25 Apr 2022 06:39:07 +0000

Why to write this best practices

For many startups, we should focus more on delivering application early, and at this time the user base is small and the QPS is very low, we should use a simpler technical architecture to accelerate the delivery of the application, and this is where the advantages of monoliths come into play.
As I often mentioned in my presentations, while we use monoliths to deliver applications quickly, we also need to consider the future possibilities when developing applications, and we can clearly split modules inside the monolith.
Many devs asked what's the best practices on developing monoliths with go-zero.

As go-zero is a widely used microservices framework, which I have precipitated during the complete development of several large projects. We have fully considered the scenario of monolithic service development.

The monolithic architecture using go-zero, as shown in the figure, can also support a large volume of business scale, where Service is a monolithic service of multiple Pods.

I'll share in detail how to use go-zero to quickly develop a monolithic service with multiple modules.

Monolithic example

Let's use an upload and download monolithic service to explain the best practices of go-zero monolithic service development, why use such an example?

The go-zero community often asks how to define API files for uploading files and then use goctl to generate the code automatically. When I first saw this kind of questions, I thought it was strange, why not use a service like OSS? I found many scenarios where the user needs to upload excel files, and then the server discards the file after parsing it. One is that the file is small, and the second is that the service is not serving a large amount of users, so we don't need to bring in OSS, which I think is quite reasonable.
The go-zero community also asking how to download files by defining an API file and then goctl automatically generating it. The reason why such questions are asked through Go is that there are generally two reasons: one is that the business is just starting, so it's easier to lay out a service to get all things done; the other is that I hope to take the advantages of go-zero's built-in JWT authentication.

This is just an example, no need to go deeper into whether uploading and downloading should be written in Go. So let's see how we can solve such a monolithic service, which we call the file service, with go-zero. The architecture is as follows.

Monolithic implementation

`API` definition

Devs who have used go-zero know that we provide an API format file to describe the RESTful API, and then we can generate the corresponding code by goctl with one shot, and we only need to fill in the corresponding business logic in the logic file. Let's see how the download and upload services define the API.

`Download` API definition

The sample requirement is as follows.

Download a file named <filename> through the /static/<filename> path
Just return the content of the file directly

We create a file named download.api in the api directory with the following content.



syntax = "v1"

type DownloadRequest {
  File string `path: "file"`
}

service file-api {
  @handler DownloadHandler
  get /static/:file(DownloadRequest)
}

The syntax of zero-api is relatively self-explanatory and means the following.

syntax = "v1" means that this is the v1 syntax of zero-api
type DownloadRequest defines the request format for Download
service file-api defines the request route for Download

`Upload` API definition

The sample requirement is as follows.

Upload a file via the /upload path
Return the upload status via json, where code can be used to express a richer scenario than HTTP code

We create a file called upload.api in the api directory with the following content.



syntax = "v1"

type UploadResponse {
  Code int `json: "code"`
}

service file-api {
  @handler UploadHandler
  post /upload returns (UploadResponse)
}

Explain as follows.

syntax = "v1" means this is the zero-api v1 syntax
type UploadResponse defines the return format of Upload
service file-api defines the request route for Upload

Here comes the problem

We have defined the Download and Upload services, but how can we put them into a service?

I don't know if you have noticed some details.

either Download or Upload, we prefixed the request and response data definition, and did not use directly such as Request or Response
we define service in download.api and upload.api with file-api as the service name, not download-api and upload-api respectively

The purpose of this is to automatically generate the corresponding Go code when we put the two services into the monolithic service. Let's see how to merge Download and Upload together~

Defining the monolithic service `API`

For simplicity reasons, goctl only supports accepting a single API file as a parameter, the issue of accepting multiple API files is not discussed here and may be supported later if we figure out a simple and efficient solution.

We create a new file.api file in the api directory with the following content.



syntax = "v1"

import "download.api"
import "upload.api"

This way we import both Download and Upload services like #include in C/C++. But there are a few things to keep in mind.

the defined structs cannot be renamed
the service name must be the same in all files

The outermost API file can also contain part of the same service definition, but we recommend to keep it symmetrical, unless these APIs really belong to the parent level, e.g. the same logical level as Download and Upload, then they should not be defined in file.api.

At this point, we have the following file structure.



.
└── api
    ├── download.api
    ├── file.api
    └── upload.api

Generating monolithic service

Now that we have the API interface defined, the next step is pretty straightforward for go-zero (of course, defining the API is pretty straightforward, isn't it?). Let's use goctl to generate the monolithic service code.



$ goctl api go -api api/file.api -dir .

Let's take a look at the generated file structure.



.
├── api
│ ├── download.api
│ ├── file.api
│ └── upload.api
├── etc
│ └── file-api.yaml
├── file.go
├─ go.mod
├── go.sum
└── internal
    ├─ config
    │ └─ config.go
    ├─ handler
    │ ├── downloadhandler.go
    │ ├── routes.go
    │ └── uploadhandler.go
    ├─ logic
    │ ├── downloadlogic.go
    │ └── uploadlogic.go
    ├── svc
    │ └── servicecontext.go
    └─ types
        └─ types.go

Let's explain the layout of the project by directory.

api directory: the API interface description file we defined earlier, no need to talk much
etc directory: this is for the yaml configuration files, all configuration items can be written in the file-api.yaml file
file.go: the file where the main function is located, with the same name as service, removed -api suffix
internal/config directory: the configuration definition of the service
internal/handler directory: the handler implementation of the routes defined in the API file
internal/logic directory: used to put the business processing logic corresponding to each route, the reason for the distinction between handler and logic is to minimize the dependency of the business processing part, to isolate HTTP requests from the logic processing code, and to facilitate the subsequent splitting into RPC service as needed
internal/svc directory: used to define the dependencies for business logic processing, we can create the dependent resources in main and pass them to handler and logic via ServiceContext
internal/types directory: defines the API request and response data structure

Let's not change anything, let's run it and see how it works.



$ go run file.go -f etc/file-api.yaml
Starting server at 0.0.0.0:8888...

Implementing the business logic

Next we need to implement the relevant business logic, but the logic here is really just for demonstration purposes, so don't pay too much attention to the implementation details, just understand that we should write the business logic in the logic layer.

The following things are done here.

Add the Path setting in the configuration item to place the uploaded files, and by default I wrote the current directory, because it is an example, as follows.



type Config struct {
  RestConf
  // New
  Path string `json:",default=." `
}

Adjusted the request body size limit as follows.



Name: file-api
Host: localhost
Port: 8888
# New
MaxBytes: 1073741824

Since Download needs to write the file to the client, we passed ResponseWriter as io.Writer to the logic layer, and the modified code is as follows



func (l *DownloadLogic) Download(req *types.DownloadRequest) error {
  logx.Infof("download %s", req.File)
  body, err := ioutil.ReadFile(req.File)
  if err ! = nil {
    return err
  }

  n, err := l.writer.Write(body)
  if err ! = nil {
    return err
  }

  if n < len(body) {
    return io.ErrClosedPipe
  }

  return nil
}

Since Upload needs to read the files uploaded by the user, we pass http.Request to the logic layer and the modified code is as follows.



func (l *UploadLogic) Upload() (resp *types.UploadResponse, err error) {
  l.r.ParseMultipartForm(maxFileSize)
  file, handler, err := l.r.FormFile("myFile")
  if err ! = nil {
    return nil, err
  }
  defer file.Close()

  logx.Infof("upload file: %+v, file size: %d, MIME header: %+v",
    handler.Filename, handler.Size, handler.Header)

  tempFile, err := os.Create(path.Join(l.svcCtx.Config.Path, handler.Filename))
  if err ! = nil {
    return nil, err
  }
  defer tempFile.Close()
  io.Copy(tempFile, file)

  return &types.UploadResponse{
    Code: 0,
  }, nil
}

Full code: https://github.com/zeromicro/zero-examples/tree/main/monolithic

We can start the file monolithic service by running the following command.



$ go run file.go -f etc/file-api.yaml

The Download service can be verified with curl:



$ curl -i "http://localhost:8888/static/file.go"
HTTP/1.1 200 OK
Traceparent: 00-831431c47d162b4decfb6b30fb232556-dd3b383feb1f13a9-00
Date: Mon, 25 Apr 2022 01:50:58 GMT
Content-Length: 584
Content-Type: text/plain; charset=utf-8

...

The sample repository contains upload.html, the browser can open this file to try the Upload service.

Summary of monolithic development

Let me summarize the complete process of developing a monolithic service with go-zero as follows.

define the API files for each submodule, e.g. download.api and upload.api
define the general API file, e.g. file.api. Use it to import the API files of each submodule defined in step 1
generate the monolithic service framework code via the goctl api go command
add and adjust the configuration to implement the business logic of the corresponding submodule

In addition, goctl can generate CRUD and cache code according to SQL in one shot, which can help you develop monolithic services more quickly.

Project address

https://github.com/zeromicro/go-zero

Welcome to use go-zero and star to support us!

Elegantly manage multiple services in one process

Kevin Wan — Thu, 21 Apr 2022 15:45:21 +0000

Background

I was often asked if it is possible to put API gateway and RPC service in the same process, and how to do it? There are also devs who put the external service and the message queue in the same process. Let's see how to solve such a problem more elegantly.

Problem

Let's use two mocking services as an example. We have two services that need to be started on two different ports in one process. The code is as following.

package main

import (
  "fmt"
  "net/http"
)

func morning(w http.ResponseWriter, req *http.Request) {
  fmt.Fprintln(w, "morning!")
}

func evening(w http.ResponseWriter, req *http.Request) {
  fmt.Fprintln(w, "evening!")
}

type Morning struct{}

func (m Morning) Start() {
  http.HandleFunc("/morning", morning)
  http.ListenAndServe("localhost:8080", nil)
}

func (m Morning) Stop() {
  fmt.Println("Stop morning service...")
}

type Evening struct{}

func (e Evening) Start() {
  http.HandleFunc("/evening", evening)
  http.ListenAndServe("localhost:8081", nil)
}

func (e Evening) Stop() {
  fmt.Println("Stop evening service...")
}

func main() {
  // todo: start both services here
}

The code is simple enough to have a request for the morning service and the service returns morning! and a request for the evening service and the service returns evening. Let's try to implement it~

First try

Is it possible to start both services in main? Let's try.

func main() {
  var morning Morning
  morning.Start()
  defer morning.Stop()

  var evening Evening
  Start()
  Stop()
}

After starting, let's verify with curl.

$ curl -i http://localhost:8080/morning
HTTP/1.1 200 OK
Date: Mon, 18 Apr 2022 02:10:34 GMT
Content-Length: 9
Content-Type: text/plain; charset=utf-8

morning!
$ curl -i http://localhost:8081/evening
curl: (7) Failed to connect to localhost port 8081 after 4 ms: Connection refused

Why only morning succeeds, but evening cannot be requested?

Let's add a print statement to main and try

func main() {
  fmt.Println("Start morning service...")
  var morning Morning
  morning.Start()
  defer morning.Stop()

  fmt.Println("Start evening service...")
  var evening Evening
  Start()
  Stop() defer evening.
}

Re-run and see.

$ go run main.go
Start morning service...

It only prints Start morning service..., so the evening service is not started at all. The reason for this is that morning.Start() blocks the current goroutine, so the subsequent code is not executed.

Second try

This is where WaitGroup come in handy. As the name implies, WaitGroup is used to wait a group of operations and wait for them to be done so that the waiting goroutine can continue. Let's try it out.

func main() {
  var wg sync.WaitGroup
  wg.Add(2)

  go func() {
    defer wg.Done()
    fmt.Println("Start morning service...")
    var morning Morning
    defer morning.Stop()
    morning.Start()
  Start() }()

  go func() {
    defer wg.Done()
    fmt.Println("Start evening service...")
    var evening Evening
    defer evening.Stop()
    evening.Start()
  Start() }()

  wg.Wait()
}

Try it.

$ go run main.go
Start evening service...
Start morning service...

Okay, both services are up, so let's verify with curl.

$ curl -i http://localhost:8080/morning
HTTP/1.1 200 OK
Date: Mon, 18 Apr 2022 02:28:33 GMT
Content-Length: 9
Content-Type: text/plain; charset=utf-8

morning!
$ curl -i http://localhost:8081/evening
HTTP/1.1 200 OK
Date: Mon, 18 Apr 2022 02:28:36 GMT
Content-Length: 9
Content-Type: text/plain; charset=utf-8

evening!

It works, and we see that our process with WaitGroup is

remember that we have several services that need wait
add the services one by one
wait for all services to finish

Let's see how go-zero handle it~

Third try

In go-zero, we provide a ServiceGroup to easily manage the start and stop of multiple services. Let's see how it is done by using it in our scenario.

import "github.com/zeromicro/go-zero/core/service"

// more code here

func main() {
  group := service.NewServiceGroup()
  defer group.Stop()
  group.Add(Morning{})
  group.Add(Evening{})
  group.Start()
}

As you can see, the code is much more readable and we can't accidentally miscalculate how many to add to WaitGroup. And ServiceGroup also makes sure that services that Start later Stop first, which is the same as defer, and makes it easier to clean up resources.

The ServiceGroup not only manages the Start/Stop of each service, but also provides graceful shutdown, which actively calls the Stop method of each service when a SIGTERM signal is received, and for HTTP services, you can exit gracefully with server.Shutdown for HTTP services and server.GracefulStop() for gRPC services.

Summary

The implementation of ServiceGroup is simple enough, with 82 lines of code.

$ cloc core/service/servicegroup.go
------------------------------------------------------------------
Language files blank comment code
------------------------------------------------------------------
Go           1    22      14   82
------------------------------------------------------------------

The code is short and concise, every service (Restful, RPC, MQ) is basically managed by ServiceGroup in go-zero, which is very convenient and worth reading.

Project address

https://github.com/zeromicro/go-zero

Welcome to use go-zero and star to support us!

The pitfalls on using timeout in Go.

Kevin Wan — Mon, 18 Apr 2022 14:15:17 +0000

Problems

Timeouts are important for programs that connect to external resources or that otherwise need to bound execution time.
Too long server-side processing will cost too much resources, resulting in a decline in concurrency, and even service unavailability.

Why Go timeout control is necessary?

Go is normally used to write backend services. Generally, a request is completed by multiple serial or parallel subtasks. Each subtask may issue another internal request. When the request times out, it’s better to return quickly and release the occupied resources, such as goroutines, file descriptors, etc.

Common timeout control on the server side

In-process processing
Serving client requests, such as HTTP or RPC requests
Calling other services, including calling RPC or accessing DB, etc.

What if there is no timeout control?

In order to simplify, we take a request function hardWork as an example. It does not matter what it is used for. As the name suggests, it may be slow to process.



func hardWork(job interface{}) error {
   time.Sleep(time.Minute)
   return nil
}

func requestWork(ctx context.Context, job interface{}) error {
   return hardWork(job)
}

When we use this kind of code to serve, the familiar image shows up for one minute. I guess most people can’t wait that long, but the server is still working on that even the page is closed. And the processing resources are not released to serve other requests.

This article will not go deep into other details, only focus on the timeout implementation.

Let’s take a look at how timeout work and what kind of pitfalls shall we take care.

Version 1

Before read further, think about how to implement the timeout of the function.

Here is our first try:



func requestWork(ctx context.Context, job interface{}) error {
   ctx, cancel := context.WithTimeout(context.Background(), time.Second*2)
   defer cancel()

   done := make(chan error)
   go func() {
     done <- hardWork(job)
   }()

   select {
   case err := <-done:
     return err
   case <-ctx.Done():
     return ctx.Err()
   }
}

Let’s write a main function to test it.



func main() {
   const total = 1000
   var wg sync.WaitGroup
   wg.Add(total)
   now := time.Now()

   for i := 0; i <total; i++ {
     go func() {
       defer wg.Done()
       requestWork(context.Background(), "any")
     }()
   }

   wg.Wait()
   fmt.Println("elapsed:", time.Since(now))
}

Run it to we’ll see.



➜ go run timeout.go
elapsed: 2.005725931s

The timeout has taken effect. But is it all done?

goroutine leak

Let’s add a line of code at the end of the main function to see how many goroutines are there.



time.Sleep(time.Minute*2)
fmt.Println("number of goroutines:", runtime.NumGoroutine())

Sleep 2 minutes is to wait for all tasks to be done, then we print the current number of goroutines. Let’s run it and see the result.



➜ go run timeout.go
elapsed: 2.005725931s
number of goroutines: 1001

Oops, the goroutine leaked, let’s see why this happens? First, the requestWork function exits after a timeout of 2 seconds. When the requestWork function exits, the done channel isn't being received by any goroutine. When the code done <- hardWork(job) is executed, it will always be stuck and cannot write. This kind of problem causes each timeout request to occupy a goroutine forever. This is a seious problem. Each goroutine takes 2-4K bytes memory, and when the memory is exhausted, the process exits unexpectedly.

So how to fix it? In fact, it is very simple, the only thing that we need to do is to set the buffer size to 1 when make chan, as below:



done := make(chan error, 1)

In this way, done <- hardWork(job) can be written regardless of whether it timed out or not, without getting stuck in the goroutine. With this method, someone may ask if there will be a problem if you write to a channel that is not being received by any goroutine. In Go, the channel is not like the resources like file descriptor. It does not have to be closed, it is just an object, close (channel) is only used to tell the receivers that there is nothing to write, no other purpose.

After changing one line of code, let’s test it again.



➜ go run timeout.go
elapsed: 2.005655146s
number of goroutines: 1

The goroutine leaking problem is gone. Awesome!

panic cannot be captured

Let’s change the code of the hardWork function to



panic("oops")

Modify the main function to catch the exceptions as below:



go func() {
   defer func() {
     if p := recover(); p != nil {
       fmt.Println("oops, panic")
     }
   }()

   defer wg.Done()
   requestWork(context.Background(), "any")
}()

If you execute the code, you will find that panic cannot be captured. The reason is that other goroutines cannot capture panic generated in the goroutine from within requestWork.

The solution is to add panicChan to the requestWork. Similarly, the buffer size of panicChan needs to be 1, as below:



func requestWork(ctx context.Context, job interface{}) error {
   ctx, cancel := context.WithTimeout(context.Background(), time.Second*2)
   defer cancel()

   done := make(chan error, 1)
   panicChan := make(chan interface{}, 1)
   go func() {
     defer func() {
       if p := recover(); p != nil {
         panicChan <- p
       }
     }()

     done <- hardWork(job)
   }()

   select {
   case err := <-done:
     return err
   case p := <-panicChan:
     panic(p)
   case <-ctx.Done():
     return ctx.Err()
   }
}

With this code, panics can be handled outside of requestWork.

Is the timeout period correct?

The above implementation of requestWork ignores the incoming ctx parameter. If ctx has a timeout setting, we must pay attention to whether the incoming timeout is less than 2 seconds we given here. If it is, we need to use the given timeout setting in ctx argument. Fortunately, context.WithTimeout compares the timeout and sets to the less one, so we just modify the code like below:



ctx, cancel := context.WithTimeout(ctx, time.Second*2)

Data race

In our example, requestWork just returns an error parameter. If you need to return multiple parameters, we need to pay attention to data race, which can be solved by mutex. For specific implementation, please refer to go-zero/zrpc/internal/serverinterceptors/timeoutinterceptor.go, I won’t go deep into the details here.

Complete example



package main

import (

   "context"

   "fmt"

   "runtime"

   "sync"

   "time"

)

func hardWork(job interface{}) error {

   time.Sleep(time.Second * 10)

   return nil

}

func requestWork(ctx context.Context, job interface{}) error {

   ctx, cancel := context.WithTimeout(ctx, time.Second*2)

   defer cancel()

done := make(chan error, 1)

   panicChan := make(chan interface{}, 1)

   go func() {

     defer func() {

       if p := recover(); p != nil {

         panicChan <- p

       }

     }()

 <span class="n">done</span> <span class="o">&lt;-</span> <span class="n">hardWork</span><span class="p">(</span><span class="n">job</span><span class="p">)</span>



}()

select {

   case err := <-done:

     return err

   case p := <-panicChan:

     panic(p)

   case <-ctx.Done():

     return ctx.Err()

   }

}

func main() {

   const total = 10

   var wg sync.WaitGroup

   wg.Add(total)

   now := time.Now()

for i := 0; i <total; i++ {

     go func() {

       defer func() {

         if p := recover(); p != nil {

           fmt.Println("oops, panic")

         }

       }()

   <span class="k">defer</span> <span class="n">wg</span><span class="o">.</span><span class="n">Done</span><span class="p">()</span>
   <span class="n">requestWork</span><span class="p">(</span><span class="n">context</span><span class="o">.</span><span class="n">Background</span><span class="p">(),</span> <span class="s">"any"</span><span class="p">)</span>
 <span class="p">}()</span>



}

wg.Wait()

   fmt.Println("elapsed:", time.Since(now))

   time.Sleep(time.Second * 20)

   fmt.Println("number of goroutines:", runtime.NumGoroutine())

}

More timeout examples

GitHub Project

https://github.com/zeromicro/go-zero

Welcome to use go-zero and star to support us!

Implement a timing wheel for millions of concurrent tasks.

Kevin Wan — Mon, 11 Apr 2022 11:28:07 +0000

For systems that contain lots of delayed in-process tasks. If we use lots of timers to handle the tasks, there will be lots of idle goroutines and lots of memory consumed. Lots of gourtines also consume more CPU to schedule them.

This article introduces the TimingWheel in go-zero, which allows developers to schedule lots of delayed tasks. As for delayed tasks, two options are usually available.

Timer, timers are used for one-off tasks. It represents a single event in the future. You tell the timer how long you want to wait, and it provides a channel that will be notified at that time.
TimingWheel, which maintains an array of task groups, and each slot maintains a chain of stored tasks. When execution starts, the timer executes the tasks in one slot at specified intervals.

Option 2 reduces the maintenance of tasks from priority queue O(nlog(n)) to bidirectional linked table O(1), and the execution of tasks also requires only polling for tasks at one point in time O(N), without putting in and removing elements O(nlog(n)), as in the case of the priority queue.

Let's look at our own use of TimingWheel in go-zero:

TimingWheel in cache

Let's start with the use of TimingWheel in the cache of collection.



timingWheel, err := NewTimingWheel(time.Second, slots, func(k, v interface{}) {
  key, ok := k.(string)
  if !ok {
    return
  }
  cache.Del(key)
})
if err ! = nil {
  return nil, err
}

cache.timingWheel = timingWheel

This is the initialization of cache which also initializes TimingWheel to clean the expired key.

interval: time interval to check the tasks
numSlots: the number of time slots
execute: the function to process tasks

The execution function in cache is deleting the expired key, and this expiration calls are controlled by TimingWheel to proceed.

Initialization



// The initialization of TimingWheel
func newTimingWheelWithClock(interval time.Duration, numSlots int, execute Execute, ticker timex.Ticker) (
  *TimingWheel, error) {
  tw := &TimingWheel{
    interval: interval, // time frame interval
    ticker: ticker, // the ticker to trigger the execution
    slots: make([]*list.List, numSlots), // the slots to put tasks
    timers: NewSafeMap(), // map to store task with {key, value}
    tickedPos: numSlots - 1, // the previous ticked position
    execute: execute, // execute function
    numSlots: numSlots, // the number of slots
    setChannel: make(chan timingEntry), // the channel to set tasks
    moveChannel: make(chan baseEntry), // the channel to move tasks
    removeChannel: make(chan interface{}), // the channel to remove tasks
    drainChannel: make(chan func(key, value interface{})), // the channel to drain tasks
    stopChannel: make(chan lang.PlaceholderType), // the channel to stop TimingWheel
  }
  // Prepare all the lists stored in the slot
  tw.initSlots()
  // start asynchronous concurrent process, use channel for task communication and passing
  go tw.run()

  return tw, nil
}

The above is a more visual representation of the "time wheel" of TimingWheel, and the details will be explained later around this diagram.

go tw.run() creates a goroutine to do the tick notification.



func (tw *TimingWheel) run() {
  for {
    select {
    // Timer does time push -> scanAndRunTasks()
    case <-tw.ticker.Chan():
      tw.onTick()
    // add task will input task to setChannel
    case task := <-tw.setChannel:
      tw.setTask(&task)
      ...
    }
  }
}

As you can see, the timer execution starts at initialization and spins in the internal time slot, and then the bottom keeps fetching the task from the list in the slot and giving it to the execute execution.

Task Operation

The next step is to set the cache key.



func (c *Cache) Set(key string, value interface{}) {
  c.lock.
  _, ok := c.data[key]
  c.data[key] = value
  c.lruCache.add(key)
  c.lock.Unlock()

  expiry := c.unstableExpiry.AroundDuration(c.expiry)
  if ok {
    c.timingWheel.MoveTimer(key, expiry)
  } else {
    c.timingWheel.SetTimer(key, value, expiry)
  }
}

check if the key exists in the data map
if it exists, update expire by calling MoveTimer()
otherwise, set the key with expiry by calling SetTimer()

So the use of TimingWheel is clear, developers can add or update according to their needs.

Also, if we read the source code, we will find that SetTimer(), MoveTimer() is to send the task to the channel, and the task operation of the channel will be continuously taken out by the goroutine created in run().

SetTimer() -> setTask().

not exist task: getPostion -> pushBack to list -> setPosition

exist task: get from timers -> moveTask()

MoveTimer() -> moveTask()

From the above call chain, there is a function that is called by all: moveTask()



func (tw *TimingWheel) moveTask(task baseEntry) {
  // timers: Map => get [positionEntry "pos, task"] by key
  val, ok := tw.timers.Get(task.key)
  if !ok {
    return
  }

  timer := val.(*positionEntry)
  // {delay < interval} => delay is less than a time frame interval,
  // the task should be executed immediately
  if task.delay < tw.interval {
    threading.GoSafe(func() {
      tw.execute(timer.item.key, timer.item.value)
    })
    return
  }
  // If > interval, calculate the new pos, circle out of the time wheel by delaying the time
  pos, circle := tw.getPositionAndCircle(task.delay)
  if pos >= timer.pos {
    timer.item.circle = circle
    // Record the offset of the move
    timer.item.diff = pos - timer.pos
  } else if circle > 0 {
    // move to the level of (circle-1)
    circle --
    timer.item.circle = circle
    // because it's an array, add numSlots [which is the equivalent of going to the next level]
    timer.item.diff = tw.numSlots + pos - timer.pos
  } else {
    // If offset is ahead of schedule, task is still on the first level
    // mark the old task for deletion and requeue it for execution
    timer.item.removed = true
    newItem := &timingEntry{
      baseEntry: task,
      value: timer.item.value,
    }
    tw.slots[pos].PushBack(newItem)
    tw.setTimerPosition(pos, newItem)
  }
}

The above process has the following cases.

delay < internal: because < single time precision, it means that this task needs to be executed immediately
delay for changes.
- new >= old: <newPos, newCircle, diff>
- newCircle > 0: compute diff and convert circle to the next level, so diff + numslots
- If the delay is simply shortened, remove the old task marker, rejoin the list, and wait for the next round of loops to be executed

Execute

Previously in the initialization, the timer in run() keeps advancing, and the process of advancing is mainly to pass the task in the list to the execution of execute func. Let's start with the execution of the timer.



// Timer 'execute every internal'
func (tw *TimingWheel) onTick() {
  // update the current tick position on every execution
  tw.tickedPos = (tw.tickedPos + 1) % tw.numSlots
  // Get the chain of tasks stored in the tick position at this time
  l := tw.slots[tw.tickedPos]
  tw.scanAndRunTasks(l)
}

Immediately following this is how to execute execute.



func (tw *TimingWheel) scanAndRunTasks(l *list.List) {
  // store the tasks {key, value} that currently need to be executed [the arguments needed by execute, passed in turn to execute]
  var tasks []timingTask

  for e := l.Front(); e ! = nil; {
    task := e.Value.(*timingEntry)
    // mark for deletion, do the real deletion in scan, delete data from map
    if task.removed {
      next := e.Next()
      l.Remove(e)
      tw.timers.Del(task.key)
      e = next
      continue
    } else if task.circle > 0 {
      // the current execution point has expired, but at the same time it's not in the first level,
      // so even though the current level is done, it drops to the next level
      // but it doesn't modify pos
      task.circle--
      e = e.Next()
      continue
    } else if task.diff > 0 {
      // because the diff has already been marked, it needs to go into the queue again
      next := e.Next()
      l.Remove(e)
      pos := (tw.tickedPos + task.diff) % tw.numSlots
      tw.slots[pos].PushBack(task)
      tw.setTimerPosition(pos, task)
      task.diff = 0
      e = next
      continue
    }
    // the above cases are all non-executable cases, those that can be executed will be added to tasks
    tasks = append(tasks, timingTask{
      key: task.key,
      value: task.value,
    })
    next := e.Next()
    l.Remove(e)
    tw.timers.Del(task.key)
    e = next
  }
  // for range tasks, then just execute each task->execute
  tw.runTasks(tasks)
}

The specific branching situation is explained in the comments, which can be combined with the previous moveTask(), where circle descends and diff computation is the focus of the associated two functions.

As for the diff calculation, it involves the calculation of pos, circle.



// interval: 4min, d: 60min, numSlots: 16, tickedPos = 15
// step = 15, pos = 14, circle = 0
func (tw *TimingWheel) getPositionAndCircle(d time.Duration) (pos int, circle int) {
  steps := int(d / tw.interval)
  pos = (tw.tickedPos + steps) % tw.numSlots
  circle = (steps - 1) / tw.numSlots
  return
}

The above process can be simplified to the following.



steps = d / interval
pos = step % numSlots - 1
circle = (step - 1) / numSlots

Summarize

TimingWheel relies on the timer to drive the time forward while taking out the tasks from the doubly linked list in the current time frame and passing them to execute for execution. Because it is driven by internal fixed time step, there may be: a 60s task, internal = 1s, so it will run 59 times noop.
in the expansion time, take circle layering, so that you can constantly reuse the original numSlots, because the timer is constantly loop through circle by circle. Any number of tasks can be put into the fixed size of slots. This design can break the long time limit without creating additional data structures.

Also in go-zero there are many practical toolkits, using them for improving service performance and development efficiency.

Project address

https://github.com/zeromicro/go-zero

Welcome to use go-zero and star to support us!

DEV Community: Kevin Wan

Circuit Breaker Explained

Circuit Breaker Principle

Circuit breaker implementation

hystrixBreaker and googlebreaker comparison

Source code interpretation

Conclusion

Reference

Project address

5 minutes to add RESTful APIs for your gRPC services

gRPC service with HTTP interface?

Let's write gRPC service first

Provide HTTP interface

Update go-zero

Modify the proto file

Generate proto descriptor file

Modify the configuration file

Modify the main function

Great job!

Conclusion

Project address

Understanding and tuning parameters of connection pools

Did you feel the same?

Why I wrote tproxy?

Installation of tproxy

Usage of tproxy

Analyze gRPC connections

Analyzing MySQL Connections

Project address

Sending type-safe HTTP requests with Go

Scenarios

Go native way

httpc implementation

More Capabilities

Project address

Developing a RESTful API with Go

When to use RESTful API

Shopping mall monolithic service architecture

Monolithic Service

Mall API Definition

Account module API definition

Order module API definition

Product module API definition

Generating the monolithic service

Implementing the business logic

Summary

Project address

How we define and load configuration from files.

Background

Scenario

Defining the configuration

Configuration files

Loading configuration files

Implementation principle

Summary

Project address

A load shedding protector against bursting requests

Why do we need load shedding?

What is adaptive load shedding

Code Implementation

Reference

Project address

My best practices on Go fuzzing

What is go fuzzing

How to use go fuzzing

Run fuzzing tests

best practices in go-zero

1. Define fuzzing arguments

2. How to write the fuzzing target

3. How to print the input for the failed case

4. Write a new test case

go fuzzing more experience

Go versioning issues

go fuzzing does not reproduce failures

Complex usage examples

Project address

Best practices on developing monolithic services in Go

Why to write this best practices

Monolithic example

Monolithic implementation

`httpc` implementation

Run `fuzzing tests`

1. Define `fuzzing arguments`

2. How to write the `fuzzing target`

`go fuzzing` more experience

`API` definition

`Download` API definition

`Upload` API definition

Defining the monolithic service `API`