// growslice handles slice growth during append.
// It is passed the slice element type, the old slice, and the desired new minimum capacity,
// and it returns a new slice with at least that capacity, with the old data
// copied into it.
// The new slice's length is set to the old slice's length,
// NOT to the new requested capacity.
// This is for codegen convenience. The old slice's length is used immediately
// to calculate where to write new values during an append.
// TODO: When the old backend is gone, reconsider this decision.
// The SSA backend might prefer the new length or to return only ptr/cap and save stack space.
func growslice(et *_type, old slice, cap int) slice {
    ......

    newcap := old.cap
    doublecap := newcap + newcap
    if cap > doublecap {
        newcap = cap
    } else {
        if old.cap < 1024 {
            newcap = doublecap
        } else {
            // Check 0 < newcap to detect overflow
            // and prevent an infinite loop.
            for 0 < newcap && newcap < cap {
                newcap += newcap / 4
            }
            // Set newcap to the requested cap when
            // the newcap calculation overflowed.
            if newcap <= 0 {
                newcap = cap
            }
        }
    }
    ......

package main

import (
    "bytes"
    "fmt"
    "runtime"
    "strconv"
    "sync"
)

func GetGoid() uint64 {
    b := make([]byte, 64)
    b = b[:runtime.Stack(b, false)]
    b = bytes.TrimPrefix(b, []byte("goroutine "))
    b = b[:bytes.IndexByte(b, ' ')]
    n, _ := strconv.ParseUint(string(b), 10, 64)
    return n
}

func main() {
    fmt.Println("main", GetGoid())
    var wg sync.WaitGroup
    for i := 0; i < 20; i++ {
        i := i
        wg.Add(1)
        go func() {
            defer wg.Done()
            fmt.Println(i, GetGoid())
        }()
    }
    wg.Wait()
}

package main

import (
    "context"
    "fmt"
    "google.golang.org/grpc"
    "log"
    . "test/protobuf"
)

func main() {
    conn, err := grpc.Dial("localhost:1234", grpc.WithInsecure())
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()

    // protobuf生成Client
    client := NewHelloServiceClient(conn)
    reply, err := client.Hello(context.Background(), &String{Value: "hello"})
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println(reply.GetValue())

package main

import (
    "context"
    "google.golang.org/grpc"
    "log"
    "net"
    . "test/protobuf"
)

type HelloServiceImpl struct{}

func (p *HelloServiceImpl) Hello(ctx context.Context, args *String) (*String, error) {
    reply := &String{Value: "hello:" + args.GetValue()}
    return reply, nil
}

func main() {
    grpcServer := grpc.NewServer()
    // protobuf生成注册函数
    RegisterHelloServiceServer(grpcServer, new(HelloServiceImpl))

    lis, err := net.Listen("tcp", ":1234")
    if err != nil {
        log.Fatal(err)
    }
    grpcServer.Serve(lis)

package main

import (
    "context"
    "fmt"
    "google.golang.org/grpc"
    "io"
    "log"
    . "test/protobuf"
    "time"
)

func main() {
    conn, err := grpc.Dial("localhost:1234", grpc.WithInsecure())
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()

    client := NewHelloServiceClient(conn)

    // 先获取stream对象
    stream, err := client.Channel(context.Background())
    if err != nil {
        log.Fatal(err)
    }

    // 模拟客户端发送数据
    go func() {
        for {
            if err := stream.Send(&String{Value: "hi"}); err != nil {
                log.Fatal(err)
            }
            time.Sleep(time.Second)
        }
    }()

    // 循环接收数据
    for {
        reply, err := stream.Recv()
        if err != nil {
            if err == io.EOF {
                break
            }
            log.Fatal(err)
        }
        fmt.Println(reply.GetValue())
    }
}

package main

import (
    "context"
    "google.golang.org/grpc"
    "io"
    "log"
    "net"
    . "test/protobuf"
)

type HelloServiceImpl struct{}

func (p *HelloServiceImpl) Hello(ctx context.Context, args *String) (*String, error) {
    reply := &String{Value: "hello:" + args.GetValue()}
    return reply, nil
}

// 循环接收客户端的数据，数据重新组装后，通过stream又发给客户端；双向流数据的发送和接收是独立的
func (p *HelloServiceImpl) Channel(stream HelloService_ChannelServer) error {
    for {
        args, err := stream.Recv()
        if err != nil {
            if err == io.EOF {
                return nil
            }
            return err
        }

        reply := &String{Value: "hello:" + args.GetValue()}

        err = stream.Send(reply)
        if err != nil {
            return err
        }
    }
}

func main() {
    grpcServer := grpc.NewServer()
    RegisterHelloServiceServer(grpcServer, new(HelloServiceImpl))

    lis, err := net.Listen("tcp", ":1234")
    if err != nil {
        log.Fatal(err)
    }
    grpcServer.Serve(lis)
}

package main

import (
    "io"
    "net/http"
    "net/rpc"
    "net/rpc/jsonrpc"
)

type HelloService struct {}

func (p *HelloService) Hello(request string, reply *string) error {
    *reply = "hello:" + request
    return nil
}

func main() {
    rpc.RegisterName("HelloService", new(HelloService))

    // 在处理函数中基于http.ResponseWriter和http.Request类型的参数
    // 构造一个io.ReadWriteCloser类型的conn通道
    http.HandleFunc("/hello", func(w http.ResponseWriter, r *http.Request) {
        var conn io.ReadWriteCloser = struct {
            io.Writer
            io.ReadCloser
        }{
            ReadCloser: r.Body,
            Writer:     w,
        }

        rpc.ServeRequest(jsonrpc.NewServerCodec(conn))
    })

    http.ListenAndServe(":1234", nil)
}

curl localhost:1234/hello -X POST \
    --data '{"method":"HelloService.Hello","params":["hello"],"id":0}'

package main

import (
    . "./server"
    "log"
    "net"
    "net/rpc"
    "net/rpc/jsonrpc"
)

func main() {
    RegisterHelloService(new(HelloService))

    listener, err := net.Listen("tcp", ":1234")
    if err != nil {
        log.Fatal("ListenTCP error:", err)
    }

    for {
        conn, err := listener.Accept()
        if err != nil {
            log.Fatal("Accept error:", err)
        }

        // json编解码器包装
        go rpc.ServeCodec(jsonrpc.NewServerCodec(conn))
    }
}

package server

import (
    "net/rpc"
)

const HelloServiceName = "path/to/pkg.HelloService"

type HelloServiceInterface interface {
    Hello(request string, reply *string) error
}

func RegisterHelloService(svc HelloServiceInterface) error {
    return rpc.RegisterName(HelloServiceName, svc)
}

type HelloService struct {}

func (p *HelloService) Hello(request string, reply *string) error {
    *reply = "hello:" + request
    return nil
}

package main

import (
    . "../server"
    "fmt"
    "log"
    "net"
    "net/rpc"
    "net/rpc/jsonrpc"
)

type HelloServiceClient struct {
    *rpc.Client
}

//var _ HelloServiceInterface = (*HelloServiceClient)(nil)

func DialHelloService(network, address string) (*HelloServiceClient, error) {
    // rpc连接建立在io.ReadWriteCloser接口之上
    conn, err := net.Dial(network, address)
    if err != nil {
        return nil, err
    }
    // json编解码器包装
    client := rpc.NewClientWithCodec(jsonrpc.NewClientCodec(conn))

    return &HelloServiceClient{Client: client}, nil
}

func (p *HelloServiceClient) Hello(request string, reply *string) error {
    return p.Client.Call(HelloServiceName+".Hello", request, reply)
}

func main() {
    client, err := DialHelloService("tcp", "localhost:1234")
    if err != nil {
        log.Fatal("dialing:", err)
    }

    var reply string
    err = client.Hello("world", &reply)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println(reply)
}

package main

import (
    . "./server"
    "log"
    "net"
    "net/rpc"
)

func main() {
    RegisterHelloService(new(HelloService))

    listener, err := net.Listen("tcp", ":1234")
    if err != nil {
        log.Fatal("ListenTCP error:", err)
    }

    for {
        conn, err := listener.Accept()
        if err != nil {
            log.Fatal("Accept error:", err)
        }

        go rpc.ServeConn(conn)
    }
}

package main

import (
    . "../server"
    "fmt"
    "log"
    "net/rpc"
)

type HelloServiceClient struct {
    *rpc.Client
}

var _ HelloServiceInterface = (*HelloServiceClient)(nil)

func DialHelloService(network, address string) (*HelloServiceClient, error) {
    // 与rpc server建立rpc连接
    c, err := rpc.Dial(network, address)
    if err != nil {
        return nil, err
    }
    return &HelloServiceClient{Client: c}, nil
}

func (p *HelloServiceClient) Hello(request string, reply *string) error {
    // 调用rpc方法
    return p.Client.Call(HelloServiceName+".Hello", request, reply)
}

func main() {
    client, err := DialHelloService("tcp", "localhost:1234")
    if err != nil {
        log.Fatal("dialing:", err)
    }

    var reply string
    err = client.Hello("world", &reply)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println(reply)
}

package server

import (
    "net/rpc"
)

const HelloServiceName = "path/to/pkg.HelloService"

type HelloServiceInterface interface {
    Hello(request string, reply *string) error
}

// rpc.Register将对象类型下所有符合Go rpc规则的方法都注册为rpc方法，
// rpc方法托管在HelloServiceName命名空间下
func RegisterHelloService(svc HelloServiceInterface) error {
    return rpc.RegisterName(HelloServiceName, svc)
}

type HelloService struct {}

func (p *HelloService) Hello(request string, reply *string) error {
    *reply = "hello:" + request
    return nil
}

package sets

import (
    "reflect"
    "sort"
)

// Empty is public since it is used by some pkg API objects for conversions between external
// string arrays and pkg sets, and conversion logic requires public types today.
type Empty struct{}

// sets.String is a set of strings, implemented via map[string]struct{} for minimal memory consumption.
type String map[string]Empty

// NewString creates a String from a list of values.
func NewString(items ...string) String {
    ss := String{}
    ss.Insert(items...)
    return ss
}

// StringKeySet creates a String from a keys of a map[string](? extends interface{}).
// If the value passed in is not actually a map, this will panic.
func StringKeySet(theMap interface{}) String {
    v := reflect.ValueOf(theMap)
    ret := String{}

    for _, keyValue := range v.MapKeys() {
        ret.Insert(keyValue.Interface().(string))
    }
    return ret
}

// Insert adds items to the set.
func (s String) Insert(items ...string) {
    for _, item := range items {
        s[item] = Empty{}
    }
}

// Delete removes all items from the set.
func (s String) Delete(items ...string) {
    for _, item := range items {
        delete(s, item)
    }
}

// Has returns true if and only if item is contained in the set.
func (s String) Has(item string) bool {
    _, contained := s[item]
    return contained
}

// HasAll returns true if and only if all items are contained in the set.
func (s String) HasAll(items ...string) bool {
    for _, item := range items {
        if !s.Has(item) {
            return false
        }
    }
    return true
}

// HasAny returns true if any items are contained in the set.
func (s String) HasAny(items ...string) bool {
    for _, item := range items {
        if s.Has(item) {
            return true
        }
    }
    return false
}

// Difference returns a set of objects that are not in s2
// For example:
// s1 = {a1, a2, a3}
// s2 = {a1, a2, a4, a5}
// s1.Difference(s2) = {a3}
// s2.Difference(s1) = {a4, a5}
func (s String) Difference(s2 String) String {
    result := NewString()
    for key := range s {
        if !s2.Has(key) {
            result.Insert(key)
        }
    }
    return result
}

// Union returns a new set which includes items in either s1 or s2.
// For example:
// s1 = {a1, a2}
// s2 = {a3, a4}
// s1.Union(s2) = {a1, a2, a3, a4}
// s2.Union(s1) = {a1, a2, a3, a4}
func (s1 String) Union(s2 String) String {
    result := NewString()
    for key := range s1 {
        result.Insert(key)
    }
    for key := range s2 {
        result.Insert(key)
    }
    return result
}

// Intersection returns a new set which includes the item in BOTH s1 and s2
// For example:
// s1 = {a1, a2}
// s2 = {a2, a3}
// s1.Intersection(s2) = {a2}
func (s1 String) Intersection(s2 String) String {
    var walk, other String
    result := NewString()
    if s1.Len() < s2.Len() {
        walk = s1
        other = s2
    } else {
        walk = s2
        other = s1
    }
    for key := range walk {
        if other.Has(key) {
            result.Insert(key)
        }
    }
    return result
}

// IsSuperset returns true if and only if s1 is a superset of s2.
func (s1 String) IsSuperset(s2 String) bool {
    for item := range s2 {
        if !s1.Has(item) {
            return false
        }
    }
    return true
}

// Equal returns true if and only if s1 is equal (as a set) to s2.
// Two sets are equal if their membership is identical.
// (In practice, this means same elements, order doesn't matter)
func (s1 String) Equal(s2 String) bool {
    return len(s1) == len(s2) && s1.IsSuperset(s2)
}

type sortableSliceOfString []string

func (s sortableSliceOfString) Len() int           { return len(s) }
func (s sortableSliceOfString) Less(i, j int) bool { return lessString(s[i], s[j]) }
func (s sortableSliceOfString) Swap(i, j int)      { s[i], s[j] = s[j], s[i] }

// List returns the contents as a sorted string slice.
func (s String) List() []string {
    res := make(sortableSliceOfString, 0, len(s))
    for key := range s {
        res = append(res, key)
    }
    sort.Sort(res)
    return []string(res)
}

// UnsortedList returns the slice with contents in random order.
func (s String) UnsortedList() []string {
    res := make([]string, 0, len(s))
    for key := range s {
        res = append(res, key)
    }
    return res
}

// Returns a single element from the set.
func (s String) PopAny() (string, bool) {
    for key := range s {
        s.Delete(key)
        return key, true
    }
    var zeroValue string
    return zeroValue, false
}

// Len returns the size of the set.
func (s String) Len() int {
    return len(s)
}

func lessString(lhs, rhs string) bool {
    return lhs < rhs
}

package ip

import (
    "errors"
    "fmt"
    "net"
    "syscall"

    "github.com/vishvananda/netlink"
)

func getIfaceAddrs(iface *net.Interface) ([]netlink.Addr, error) {
    link := &netlink.Device{
        netlink.LinkAttrs{
            Index: iface.Index,
        },
    }

    return netlink.AddrList(link, syscall.AF_INET)
}

func GetIfaceIP4Addr(iface *net.Interface) (net.IP, error) {
    addrs, err := getIfaceAddrs(iface)
    if err != nil {
        return nil, err
    }

    // prefer non link-local addr
    var ll net.IP

    for _, addr := range addrs {
        if addr.IP.To4() == nil {
            continue
        }

        if addr.IP.IsGlobalUnicast() {
            return addr.IP, nil
        }

        if addr.IP.IsLinkLocalUnicast() {
            ll = addr.IP
        }
    }

    if ll != nil {
        // didn't find global but found link-local. it'll do.
        return ll, nil
    }

    return nil, errors.New("No IPv4 address found for given interface")
}

func GetIfaceIP4AddrMatch(iface *net.Interface, matchAddr net.IP) error {
    addrs, err := getIfaceAddrs(iface)
    if err != nil {
        return err
    }

    for _, addr := range addrs {
        // Attempt to parse the address in CIDR notation
        // and assert it is IPv4
        if addr.IP.To4() != nil {
            if addr.IP.To4().Equal(matchAddr) {
                return nil
            }
        }
    }

    return errors.New("No IPv4 address found for given interface")
}

func GetDefaultGatewayIface() (*net.Interface, error) {
    routes, err := netlink.RouteList(nil, syscall.AF_INET)
    if err != nil {
        return nil, err
    }

    for _, route := range routes {
        if route.Dst == nil || route.Dst.String() == "0.0.0.0/0" {
            if route.LinkIndex <= 0 {
                return nil, errors.New("Found default route but could not determine interface")
            }
            return net.InterfaceByIndex(route.LinkIndex)
        }
    }

    return nil, errors.New("Unable to find default route")
}

func GetInterfaceByIP(ip net.IP) (*net.Interface, error) {
    ifaces, err := net.Interfaces()
    if err != nil {
        return nil, err
    }

    for _, iface := range ifaces {
        err := GetIfaceIP4AddrMatch(&iface, ip)
        if err == nil {
            return &iface, nil
        }
    }

    return nil, errors.New("No interface with given IP found")
}

func DirectRouting(ip net.IP) (bool, error) {
    routes, err := netlink.RouteGet(ip)
    if err != nil {
        return false, fmt.Errorf("couldn't lookup route to %v: %v", ip, err)
    }
    if len(routes) == 1 && routes[0].Gw == nil {
        // There is only a single route and there's no gateway (i.e. it's directly connected)
        return true, nil
    }
    return false, nil
}

// EnsureV4AddressOnLink ensures that there is only one v4 Addr on `link` and it equals `ipn`.
// If there exist multiple addresses on link, it returns an error message to tell callers to remove additional address.
func EnsureV4AddressOnLink(ipn IP4Net, link netlink.Link) error {
    addr := netlink.Addr{IPNet: ipn.ToIPNet()}
    existingAddrs, err := netlink.AddrList(link, netlink.FAMILY_V4)
    if err != nil {
        return err
    }

    // flannel will never make this happen. This situation can only be caused by a user, so get them to sort it out.
    if len(existingAddrs) > 1 {
        return fmt.Errorf("link has incompatible addresses. Remove additional addresses and try again. %#v", link)
    }

    // If the device has an incompatible address then delete it. This can happen if the lease changes for example.
    if len(existingAddrs) == 1 && !existingAddrs[0].Equal(addr) {
        if err := netlink.AddrDel(link, &existingAddrs[0]); err != nil {
            return fmt.Errorf("failed to remove IP address %s from %s: %s", ipn.String(), link.Attrs().Name, err)
        }
        existingAddrs = []netlink.Addr{}
    }

    // Actually add the desired address to the interface if needed.
    if len(existingAddrs) == 0 {
        if err := netlink.AddrAdd(link, &addr); err != nil {
            return fmt.Errorf("failed to add IP address %s to %s: %s", ipn.String(), link.Attrs().Name, err)
        }
    }

    return nil
}

package main

func If(condition bool, trueVal, falseVal interface{}) interface{} {
    if condition {
        return trueVal
    }
    return falseVal
}

func main() {
    a, b := 2, 3
    max := If(a > b, a, b).(int)
    println(max)
}

package main

//参数定义在结构体内部
type Op struct {
    userInfo *models.UserInfo
    pagination *models.Pagination
    state string
}

type OpOption func(*Op)

//变量赋值給实例
func (op *Op) applyOpts(opts []OpOption) {
    for _, opt := range opts {
        opt(op)
    }
}

func WithUserInfo(userInfo *models.UserInfo) OpOption {
    return func(op *Op) { op.userInfo = userInfo }
}

func WithPagination(pagination *models.Pagination) OpOption {
    return func(op *Op) { op.pagination = pagination }
}

func WithState(state string) OpOption {
    return func(op *Op) { op.state = state }
}

func OpGet(opts ...OpOption) Op {
    ret := Op{}
    ret.applyOpts(opts)
    return ret
}

package main

import "fmt"

//回调函数
func XX(x int) int {
    return x * x
}

//中间函数
//接受一个生成平方数的函数作为参数
func returnXXNumber(x int, getXXNumber func(int) int) int {
    return getXXNumber(x)
}

//起始函数
func main() {
    x := 2
    fmt.Println(returnXXNumber(x, XX))
}

package main

import "fmt"

func add() func(int) int {
    sum := 0
    return func(x int) int {
        sum += x
        return sum
    }
}

func main() {
    //add()就是一个闭包，并赋值给pos和neg，pos的闭包函数和neg的闭包函数被绑定在各自的sum变量上；
    //两个闭包函数的sum变量之间没有任何关系
    pos, neg := add(), add()
    for i := 0; i < 5; i++ {
        fmt.Println(
            pos(i),
            neg(-i),
        )
    }
}

package main

//fibonacci函数完成核心算法、核心数据存储, 不负责for循环
func fibonacci() func() int {
    b1 := 1
    b2 := 0
    bc := 0
    return func() int {
        bc = b1 + b2
        b1 = b2
        b2 = bc
        return bc
    }
}

func main() {
    f := fibonacci()
    for i := 0; i < 10; i++ {
        fmt.Println(f())
    }
}

package main

import (
    "context"
    "fmt"
    "log"
    "strings"
    "time"

    "github.com/coreos/etcd/clientv3"
    "github.com/coreos/etcd/clientv3/concurrency"
)

var (
    endpoints = "http://10.10.10.10:52379"
)

func main() {
    cli, err := clientv3.New(clientv3.Config{
        Endpoints:   strings.Split(endpoints, ","),
        DialTimeout: 3 * time.Second,
    })
    if err != nil {
        panic(err)
    }
    defer cli.Close()

    // create two separate sessions for lock competition
    s1, err := concurrency.NewSession(cli)
    if err != nil {
        log.Fatal(err)
    }
    defer s1.Close()
    m1 := concurrency.NewMutex(s1, "/my-lock/")

    s2, err := concurrency.NewSession(cli)
    if err != nil {
        log.Fatal(err)
    }
    defer s2.Close()
    m2 := concurrency.NewMutex(s2, "/my-lock/")

    // acquire lock for s1
    if err := m1.Lock(context.TODO()); err != nil {
        log.Fatal(err)
    }
    fmt.Println("acquired lock for s1")

    m2Locked := make(chan struct{})
    go func() {
        defer close(m2Locked)
        // wait until s1 is locks /my-lock/
        if err := m2.Lock(context.TODO()); err != nil {
            log.Fatal(err)
        }
    }()

    if err := m1.Unlock(context.TODO()); err != nil {
        log.Fatal(err)
    }
    fmt.Println("released lock for s1")

    <-m2Locked
    fmt.Println("acquired lock for s2")

    // Output:
    // acquired lock for s1
    // released lock for s1
    // acquired lock for s2
}

func CommonPaginate(x interface{}, offset int, limit int) (int, int) {
    xLen := reflect.ValueOf(x).Len()
    if offset+1 > xLen {
        offset = xLen - 1
        if offset < 0 {
            offset = 0
        }
    }
    end := offset + limit
    if end > xLen {
        end = xLen
    }
    return offset, end
}

package main

import (
    "fmt"
    "github.com/moby/moby/pkg/pubsub"
    "strings"
    "sync"
    "time"
)

func main() {
    p := pubsub.NewPublisher(100*time.Millisecond, 10)

    golang := p.SubscribeTopic(func(v interface{}) bool {
        if key, ok := v.(string); ok {
            if strings.HasPrefix(key, "golang:") {
                return true
            }
        }
        return false
    })
    docker := p.SubscribeTopic(func(v interface{}) bool {
        if key, ok := v.(string); ok {
            if strings.HasPrefix(key, "docker:") {
                return true
            }
        }
        return false
    })

    go p.Publish("hi")
    go p.Publish("golang: https://golang.org")
    go p.Publish("docker: https://www.docker.com/")
    time.Sleep(1)

    wg := &sync.WaitGroup{}
    wg.Add(2)

    go func() {
        defer func() {
            wg.Done()
        }()
        fmt.Println("golang topic:", <-golang)
    }()
    go func() {
        defer func() {
            wg.Done()
        }()
        fmt.Println("docker topic:", <-docker)
    }()
    wg.Wait()
}

package main

import (
    "fmt"
    "time"
)

type Periodic interface {
    Do() error
    Frequency() time.Duration
    Name() string
    FirstRun() bool
}

type PeriodicCanceller func()

func DoPeriodic(pSlice []Periodic) PeriodicCanceller {
    doneCh := make(chan struct{})
    for _, p := range pSlice {
        go func(p Periodic) {
            if p.FirstRun() {
                err := p.Do()
                if err != nil {
                    fmt.Printf("periodic job ran and returned error (%s)\n", err)
                } else {
                    fmt.Printf("periodic job %s ran\n", p.Name())
                }
            }
            ticker := time.NewTicker(p.Frequency())
            for {
                select {
                case <-ticker.C:
                    err := p.Do()
                    if err != nil {
                        fmt.Printf("periodic job ran and returned error (%s)\n", err)
                    }
                case <-doneCh:
                    ticker.Stop()
                    return
                }
            }
        }(p)
    }
    return func() {
        close(doneCh)
    }
}

func NewRefreshData(frequency time.Duration, name string, firstRun bool) Periodic {
    return &refreshData{
        frequency: frequency,
        name:      name,
        firstRun:  firstRun,
    }
}

type refreshData struct {
    frequency time.Duration
    name      string
    firstRun  bool
}

func (r *refreshData) Do() error {
    fmt.Printf("time: %s, %s Do xxx\n", time.Now(), r.name)
    return nil
}

func (r *refreshData) Frequency() time.Duration {
    return r.frequency
}

func (r *refreshData) FirstRun() bool {
    return r.firstRun
}

func (r *refreshData) Name() string {
    return r.name
}

func main() {
    var refreshInterval = 3
    freshness := time.Duration(refreshInterval) * time.Second
    periodicRefresh :=  NewRefreshData(freshness, "refresh", false)
    newPeriodicRefresh :=  NewRefreshData(5*time.Second, "test", true)
    toDo := []Periodic{periodicRefresh, newPeriodicRefresh}
    DoPeriodic(toDo)

    select {
    }
}

package main

import (
    "fmt"
    "time"
)

func main() {
    var (
        DefaultInterval = 1
    )

    done := make(chan bool)
    ticker := time.NewTicker(time.Duration(DefaultInterval) * time.Second)

    fmt.Println("begin!")
    go func() {
        for {
            select {
            case <- done:
                fmt.Println("stop!")
                ticker.Stop()
                return
            case <- ticker.C:
                fmt.Printf("time: %s, msg: trigger the periodic timer.\n", time.Now())
            }
        }
    }()
    time.Sleep(5*time.Second)
    done <- true
    fmt.Println("end!")
}

/*
Copyright 2014 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
    http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package wait

import (
    "context"
    "errors"
    "math"
    "math/rand"
    "sync"
    "time"

    "k8s.io/apimachinery/pkg/util/runtime"
    "k8s.io/utils/clock"
)

// For any test of the style:
//   ...
//   <- time.After(timeout):
//      t.Errorf("Timed out")
// The value for timeout should effectively be "forever." Obviously we don't want our tests to truly lock up forever, but 30s
// is long enough that it is effectively forever for the things that can slow down a run on a heavily contended machine
// (GC, seeks, etc), but not so long as to make a developer ctrl-c a test run if they do happen to break that test.
var ForeverTestTimeout = time.Second * 30

// NeverStop may be passed to Until to make it never stop.
var NeverStop <-chan struct{} = make(chan struct{})

// Group allows to start a group of goroutines and wait for their completion.
type Group struct {
    wg sync.WaitGroup
}

func (g *Group) Wait() {
    g.wg.Wait()
}

// StartWithChannel starts f in a new goroutine in the group.
// stopCh is passed to f as an argument. f should stop when stopCh is available.
func (g *Group) StartWithChannel(stopCh <-chan struct{}, f func(stopCh <-chan struct{})) {
    g.Start(func() {
        f(stopCh)
    })
}

// StartWithContext starts f in a new goroutine in the group.
// ctx is passed to f as an argument. f should stop when ctx.Done() is available.
func (g *Group) StartWithContext(ctx context.Context, f func(context.Context)) {
    g.Start(func() {
        f(ctx)
    })
}

// Start starts f in a new goroutine in the group.
func (g *Group) Start(f func()) {
    g.wg.Add(1)
    go func() {
        defer g.wg.Done()
        f()
    }()
}

// Forever calls f every period for ever.
//
// Forever is syntactic sugar on top of Until.
func Forever(f func(), period time.Duration) {
    Until(f, period, NeverStop)
}

// Until loops until stop channel is closed, running f every period.
//
// Until is syntactic sugar on top of JitterUntil with zero jitter factor and
// with sliding = true (which means the timer for period starts after the f
// completes).
func Until(f func(), period time.Duration, stopCh <-chan struct{}) {
    JitterUntil(f, period, 0.0, true, stopCh)
}

// UntilWithContext loops until context is done, running f every period.
//
// UntilWithContext is syntactic sugar on top of JitterUntilWithContext
// with zero jitter factor and with sliding = true (which means the timer
// for period starts after the f completes).
func UntilWithContext(ctx context.Context, f func(context.Context), period time.Duration) {
    JitterUntilWithContext(ctx, f, period, 0.0, true)
}

// NonSlidingUntil loops until stop channel is closed, running f every
// period.
//
// NonSlidingUntil is syntactic sugar on top of JitterUntil with zero jitter
// factor, with sliding = false (meaning the timer for period starts at the same
// time as the function starts).
func NonSlidingUntil(f func(), period time.Duration, stopCh <-chan struct{}) {
    JitterUntil(f, period, 0.0, false, stopCh)
}

// NonSlidingUntilWithContext loops until context is done, running f every
// period.
//
// NonSlidingUntilWithContext is syntactic sugar on top of JitterUntilWithContext
// with zero jitter factor, with sliding = false (meaning the timer for period
// starts at the same time as the function starts).
func NonSlidingUntilWithContext(ctx context.Context, f func(context.Context), period time.Duration) {
    JitterUntilWithContext(ctx, f, period, 0.0, false)
}

// JitterUntil loops until stop channel is closed, running f every period.
//
// If jitterFactor is positive, the period is jittered before every run of f.
// If jitterFactor is not positive, the period is unchanged and not jittered.
//
// If sliding is true, the period is computed after f runs. If it is false then
// period includes the runtime for f.
//
// Close stopCh to stop. f may not be invoked if stop channel is already
// closed. Pass NeverStop to if you don't want it stop.
func JitterUntil(f func(), period time.Duration, jitterFactor float64, sliding bool, stopCh <-chan struct{}) {
    BackoffUntil(f, NewJitteredBackoffManager(period, jitterFactor, &clock.RealClock{}), sliding, stopCh)
}

// BackoffUntil loops until stop channel is closed, run f every duration given by BackoffManager.
//
// If sliding is true, the period is computed after f runs. If it is false then
// period includes the runtime for f.
func BackoffUntil(f func(), backoff BackoffManager, sliding bool, stopCh <-chan struct{}) {
    var t clock.Timer
    for {
        select {
        case <-stopCh:
            return
        default:
        }

        if !sliding {
            t = backoff.Backoff()
        }

        func() {
            defer runtime.HandleCrash()
            f()
        }()

        if sliding {
            t = backoff.Backoff()
        }

        // NOTE: b/c there is no priority selection in golang
        // it is possible for this to race, meaning we could
        // trigger t.C and stopCh, and t.C select falls through.
        // In order to mitigate we re-check stopCh at the beginning
        // of every loop to prevent extra executions of f().
        select {
        case <-stopCh:
            if !t.Stop() {
                <-t.C()
            }
            return
        case <-t.C():
        }
    }
}

// JitterUntilWithContext loops until context is done, running f every period.
//
// If jitterFactor is positive, the period is jittered before every run of f.
// If jitterFactor is not positive, the period is unchanged and not jittered.
//
// If sliding is true, the period is computed after f runs. If it is false then
// period includes the runtime for f.
//
// Cancel context to stop. f may not be invoked if context is already expired.
func JitterUntilWithContext(ctx context.Context, f func(context.Context), period time.Duration, jitterFactor float64, sliding bool) {
    JitterUntil(func() { f(ctx) }, period, jitterFactor, sliding, ctx.Done())
}

// Jitter returns a time.Duration between duration and duration + maxFactor *
// duration.
//
// This allows clients to avoid converging on periodic behavior. If maxFactor
// is 0.0, a suggested default value will be chosen.
func Jitter(duration time.Duration, maxFactor float64) time.Duration {
    if maxFactor <= 0.0 {
        maxFactor = 1.0
    }
    wait := duration + time.Duration(rand.Float64()*maxFactor*float64(duration))
    return wait
}

// ErrWaitTimeout is returned when the condition exited without success.
var ErrWaitTimeout = errors.New("timed out waiting for the condition")

// ConditionFunc returns true if the condition is satisfied, or an error
// if the loop should be aborted.
type ConditionFunc func() (done bool, err error)

// ConditionWithContextFunc returns true if the condition is satisfied, or an error
// if the loop should be aborted.
//
// The caller passes along a context that can be used by the condition function.
type ConditionWithContextFunc func(context.Context) (done bool, err error)

// WithContext converts a ConditionFunc into a ConditionWithContextFunc
func (cf ConditionFunc) WithContext() ConditionWithContextFunc {
    return func(context.Context) (done bool, err error) {
        return cf()
    }
}

// runConditionWithCrashProtection runs a ConditionFunc with crash protection
func runConditionWithCrashProtection(condition ConditionFunc) (bool, error) {
    return runConditionWithCrashProtectionWithContext(context.TODO(), condition.WithContext())
}

// runConditionWithCrashProtectionWithContext runs a
// ConditionWithContextFunc with crash protection.
func runConditionWithCrashProtectionWithContext(ctx context.Context, condition ConditionWithContextFunc) (bool, error) {
    defer runtime.HandleCrash()
    return condition(ctx)
}

// Backoff holds parameters applied to a Backoff function.
type Backoff struct {
    // The initial duration.
    Duration time.Duration
    // Duration is multiplied by factor each iteration, if factor is not zero
    // and the limits imposed by Steps and Cap have not been reached.
    // Should not be negative.
    // The jitter does not contribute to the updates to the duration parameter.
    Factor float64
    // The sleep at each iteration is the duration plus an additional
    // amount chosen uniformly at random from the interval between
    // zero and `jitter*duration`.
    Jitter float64
    // The remaining number of iterations in which the duration
    // parameter may change (but progress can be stopped earlier by
    // hitting the cap). If not positive, the duration is not
    // changed. Used for exponential backoff in combination with
    // Factor and Cap.
    Steps int
    // A limit on revised values of the duration parameter. If a
    // multiplication by the factor parameter would make the duration
    // exceed the cap then the duration is set to the cap and the
    // steps parameter is set to zero.
    Cap time.Duration
}

// Step (1) returns an amount of time to sleep determined by the
// original Duration and Jitter and (2) mutates the provided Backoff
// to update its Steps and Duration.
func (b *Backoff) Step() time.Duration {
    if b.Steps < 1 {
        if b.Jitter > 0 {
            return Jitter(b.Duration, b.Jitter)
        }
        return b.Duration
    }
    b.Steps--

    duration := b.Duration

    // calculate the next step
    if b.Factor != 0 {
        b.Duration = time.Duration(float64(b.Duration) * b.Factor)
        if b.Cap > 0 && b.Duration > b.Cap {
            b.Duration = b.Cap
            b.Steps = 0
        }
    }

    if b.Jitter > 0 {
        duration = Jitter(duration, b.Jitter)
    }
    return duration
}

// contextForChannel derives a child context from a parent channel.
//
// The derived context's Done channel is closed when the returned cancel function
// is called or when the parent channel is closed, whichever happens first.
//
// Note the caller must *always* call the CancelFunc, otherwise resources may be leaked.
func contextForChannel(parentCh <-chan struct{}) (context.Context, context.CancelFunc) {
    ctx, cancel := context.WithCancel(context.Background())

    go func() {
        select {
        case <-parentCh:
            cancel()
        case <-ctx.Done():
        }
    }()
    return ctx, cancel
}

// BackoffManager manages backoff with a particular scheme based on its underlying implementation. It provides
// an interface to return a timer for backoff, and caller shall backoff until Timer.C() drains. If the second Backoff()
// is called before the timer from the first Backoff() call finishes, the first timer will NOT be drained and result in
// undetermined behavior.
// The BackoffManager is supposed to be called in a single-threaded environment.
type BackoffManager interface {
    Backoff() clock.Timer
}

type exponentialBackoffManagerImpl struct {
    backoff              *Backoff
    backoffTimer         clock.Timer
    lastBackoffStart     time.Time
    initialBackoff       time.Duration
    backoffResetDuration time.Duration
    clock                clock.Clock
}

// NewExponentialBackoffManager returns a manager for managing exponential backoff. Each backoff is jittered and
// backoff will not exceed the given max. If the backoff is not called within resetDuration, the backoff is reset.
// This backoff manager is used to reduce load during upstream unhealthiness.
func NewExponentialBackoffManager(initBackoff, maxBackoff, resetDuration time.Duration, backoffFactor, jitter float64, c clock.Clock) BackoffManager {
    return &exponentialBackoffManagerImpl{
        backoff: &Backoff{
            Duration: initBackoff,
            Factor:   backoffFactor,
            Jitter:   jitter,

            // the current impl of wait.Backoff returns Backoff.Duration once steps are used up, which is not
            // what we ideally need here, we set it to max int and assume we will never use up the steps
            Steps: math.MaxInt32,
            Cap:   maxBackoff,
        },
        backoffTimer:         nil,
        initialBackoff:       initBackoff,
        lastBackoffStart:     c.Now(),
        backoffResetDuration: resetDuration,
        clock:                c,
    }
}

func (b *exponentialBackoffManagerImpl) getNextBackoff() time.Duration {
    if b.clock.Now().Sub(b.lastBackoffStart) > b.backoffResetDuration {
        b.backoff.Steps = math.MaxInt32
        b.backoff.Duration = b.initialBackoff
    }
    b.lastBackoffStart = b.clock.Now()
    return b.backoff.Step()
}

// Backoff implements BackoffManager.Backoff, it returns a timer so caller can block on the timer for exponential backoff.
// The returned timer must be drained before calling Backoff() the second time
func (b *exponentialBackoffManagerImpl) Backoff() clock.Timer {
    if b.backoffTimer == nil {
        b.backoffTimer = b.clock.NewTimer(b.getNextBackoff())
    } else {
        b.backoffTimer.Reset(b.getNextBackoff())
    }
    return b.backoffTimer
}

type jitteredBackoffManagerImpl struct {
    clock        clock.Clock
    duration     time.Duration
    jitter       float64
    backoffTimer clock.Timer
}

// NewJitteredBackoffManager returns a BackoffManager that backoffs with given duration plus given jitter. If the jitter
// is negative, backoff will not be jittered.
func NewJitteredBackoffManager(duration time.Duration, jitter float64, c clock.Clock) BackoffManager {
    return &jitteredBackoffManagerImpl{
        clock:        c,
        duration:     duration,
        jitter:       jitter,
        backoffTimer: nil,
    }
}

func (j *jitteredBackoffManagerImpl) getNextBackoff() time.Duration {
    jitteredPeriod := j.duration
    if j.jitter > 0.0 {
        jitteredPeriod = Jitter(j.duration, j.jitter)
    }
    return jitteredPeriod
}

// Backoff implements BackoffManager.Backoff, it returns a timer so caller can block on the timer for jittered backoff.
// The returned timer must be drained before calling Backoff() the second time
func (j *jitteredBackoffManagerImpl) Backoff() clock.Timer {
    backoff := j.getNextBackoff()
    if j.backoffTimer == nil {
        j.backoffTimer = j.clock.NewTimer(backoff)
    } else {
        j.backoffTimer.Reset(backoff)
    }
    return j.backoffTimer
}

// ExponentialBackoff repeats a condition check with exponential backoff.
//
// It repeatedly checks the condition and then sleeps, using `backoff.Step()`
// to determine the length of the sleep and adjust Duration and Steps.
// Stops and returns as soon as:
// 1. the condition check returns true or an error,
// 2. `backoff.Steps` checks of the condition have been done, or
// 3. a sleep truncated by the cap on duration has been completed.
// In case (1) the returned error is what the condition function returned.
// In all other cases, ErrWaitTimeout is returned.
func ExponentialBackoff(backoff Backoff, condition ConditionFunc) error {
    for backoff.Steps > 0 {
        if ok, err := runConditionWithCrashProtection(condition); err != nil || ok {
            return err
        }
        if backoff.Steps == 1 {
            break
        }
        time.Sleep(backoff.Step())
    }
    return ErrWaitTimeout
}

// Poll tries a condition func until it returns true, an error, or the timeout
// is reached.
//
// Poll always waits the interval before the run of 'condition'.
// 'condition' will always be invoked at least once.
//
// Some intervals may be missed if the condition takes too long or the time
// window is too short.
//
// If you want to Poll something forever, see PollInfinite.
func Poll(interval, timeout time.Duration, condition ConditionFunc) error {
    return PollWithContext(context.Background(), interval, timeout, condition.WithContext())
}

// PollWithContext tries a condition func until it returns true, an error,
// or when the context expires or the timeout is reached, whichever
// happens first.
//
// PollWithContext always waits the interval before the run of 'condition'.
// 'condition' will always be invoked at least once.
//
// Some intervals may be missed if the condition takes too long or the time
// window is too short.
//
// If you want to Poll something forever, see PollInfinite.
func PollWithContext(ctx context.Context, interval, timeout time.Duration, condition ConditionWithContextFunc) error {
    return poll(ctx, false, poller(interval, timeout), condition)
}

// PollUntil tries a condition func until it returns true, an error or stopCh is
// closed.
//
// PollUntil always waits interval before the first run of 'condition'.
// 'condition' will always be invoked at least once.
func PollUntil(interval time.Duration, condition ConditionFunc, stopCh <-chan struct{}) error {
    ctx, cancel := contextForChannel(stopCh)
    defer cancel()
    return PollUntilWithContext(ctx, interval, condition.WithContext())
}

// PollUntilWithContext tries a condition func until it returns true,
// an error or the specified context is cancelled or expired.
//
// PollUntilWithContext always waits interval before the first run of 'condition'.
// 'condition' will always be invoked at least once.
func PollUntilWithContext(ctx context.Context, interval time.Duration, condition ConditionWithContextFunc) error {
    return poll(ctx, false, poller(interval, 0), condition)
}

// PollInfinite tries a condition func until it returns true or an error
//
// PollInfinite always waits the interval before the run of 'condition'.
//
// Some intervals may be missed if the condition takes too long or the time
// window is too short.
func PollInfinite(interval time.Duration, condition ConditionFunc) error {
    return PollInfiniteWithContext(context.Background(), interval, condition.WithContext())
}

// PollInfiniteWithContext tries a condition func until it returns true or an error
//
// PollInfiniteWithContext always waits the interval before the run of 'condition'.
//
// Some intervals may be missed if the condition takes too long or the time
// window is too short.
func PollInfiniteWithContext(ctx context.Context, interval time.Duration, condition ConditionWithContextFunc) error {
    return poll(ctx, false, poller(interval, 0), condition)
}

// PollImmediate tries a condition func until it returns true, an error, or the timeout
// is reached.
//
// PollImmediate always checks 'condition' before waiting for the interval. 'condition'
// will always be invoked at least once.
//
// Some intervals may be missed if the condition takes too long or the time
// window is too short.
//
// If you want to immediately Poll something forever, see PollImmediateInfinite.
func PollImmediate(interval, timeout time.Duration, condition ConditionFunc) error {
    return PollImmediateWithContext(context.Background(), interval, timeout, condition.WithContext())
}

// PollImmediateWithContext tries a condition func until it returns true, an error,
// or the timeout is reached or the specified context expires, whichever happens first.
//
// PollImmediateWithContext always checks 'condition' before waiting for the interval.
// 'condition' will always be invoked at least once.
//
// Some intervals may be missed if the condition takes too long or the time
// window is too short.
//
// If you want to immediately Poll something forever, see PollImmediateInfinite.
func PollImmediateWithContext(ctx context.Context, interval, timeout time.Duration, condition ConditionWithContextFunc) error {
    return poll(ctx, true, poller(interval, timeout), condition)
}

// PollImmediateUntil tries a condition func until it returns true, an error or stopCh is closed.
//
// PollImmediateUntil runs the 'condition' before waiting for the interval.
// 'condition' will always be invoked at least once.
func PollImmediateUntil(interval time.Duration, condition ConditionFunc, stopCh <-chan struct{}) error {
    ctx, cancel := contextForChannel(stopCh)
    defer cancel()
    return PollImmediateUntilWithContext(ctx, interval, condition.WithContext())
}

// PollImmediateUntilWithContext tries a condition func until it returns true,
// an error or the specified context is cancelled or expired.
//
// PollImmediateUntilWithContext runs the 'condition' before waiting for the interval.
// 'condition' will always be invoked at least once.
func PollImmediateUntilWithContext(ctx context.Context, interval time.Duration, condition ConditionWithContextFunc) error {
    return poll(ctx, true, poller(interval, 0), condition)
}

// PollImmediateInfinite tries a condition func until it returns true or an error
//
// PollImmediateInfinite runs the 'condition' before waiting for the interval.
//
// Some intervals may be missed if the condition takes too long or the time
// window is too short.
func PollImmediateInfinite(interval time.Duration, condition ConditionFunc) error {
    return PollImmediateInfiniteWithContext(context.Background(), interval, condition.WithContext())
}

// PollImmediateInfiniteWithContext tries a condition func until it returns true
// or an error or the specified context gets cancelled or expired.
//
// PollImmediateInfiniteWithContext runs the 'condition' before waiting for the interval.
//
// Some intervals may be missed if the condition takes too long or the time
// window is too short.
func PollImmediateInfiniteWithContext(ctx context.Context, interval time.Duration, condition ConditionWithContextFunc) error {
    return poll(ctx, true, poller(interval, 0), condition)
}

// Internally used, each of the the public 'Poll*' function defined in this
// package should invoke this internal function with appropriate parameters.
// ctx: the context specified by the caller, for infinite polling pass
// a context that never gets cancelled or expired.
// immediate: if true, the 'condition' will be invoked before waiting for the interval,
// in this case 'condition' will always be invoked at least once.
// wait: user specified WaitFunc function that controls at what interval the condition
// function should be invoked periodically and whether it is bound by a timeout.
// condition: user specified ConditionWithContextFunc function.
func poll(ctx context.Context, immediate bool, wait WaitWithContextFunc, condition ConditionWithContextFunc) error {
    if immediate {
        done, err := runConditionWithCrashProtectionWithContext(ctx, condition)
        if err != nil {
            return err
        }
        if done {
            return nil
        }
    }

    select {
    case <-ctx.Done():
        // returning ctx.Err() will break backward compatibility
        return ErrWaitTimeout
    default:
        return WaitForWithContext(ctx, wait, condition)
    }
}

// WaitFunc creates a channel that receives an item every time a test
// should be executed and is closed when the last test should be invoked.
type WaitFunc func(done <-chan struct{}) <-chan struct{}

// WithContext converts the WaitFunc to an equivalent WaitWithContextFunc
func (w WaitFunc) WithContext() WaitWithContextFunc {
    return func(ctx context.Context) <-chan struct{} {
        return w(ctx.Done())
    }
}

// WaitWithContextFunc creates a channel that receives an item every time a test
// should be executed and is closed when the last test should be invoked.
//
// When the specified context gets cancelled or expires the function
// stops sending item and returns immediately.
type WaitWithContextFunc func(ctx context.Context) <-chan struct{}

// WaitFor continually checks 'fn' as driven by 'wait'.
//
// WaitFor gets a channel from 'wait()'', and then invokes 'fn' once for every value
// placed on the channel and once more when the channel is closed. If the channel is closed
// and 'fn' returns false without error, WaitFor returns ErrWaitTimeout.
//
// If 'fn' returns an error the loop ends and that error is returned. If
// 'fn' returns true the loop ends and nil is returned.
//
// ErrWaitTimeout will be returned if the 'done' channel is closed without fn ever
// returning true.
//
// When the done channel is closed, because the golang `select` statement is
// "uniform pseudo-random", the `fn` might still run one or multiple time,
// though eventually `WaitFor` will return.
func WaitFor(wait WaitFunc, fn ConditionFunc, done <-chan struct{}) error {
    ctx, cancel := contextForChannel(done)
    defer cancel()
    return WaitForWithContext(ctx, wait.WithContext(), fn.WithContext())
}

// WaitForWithContext continually checks 'fn' as driven by 'wait'.
//
// WaitForWithContext gets a channel from 'wait()'', and then invokes 'fn'
// once for every value placed on the channel and once more when the
// channel is closed. If the channel is closed and 'fn'
// returns false without error, WaitForWithContext returns ErrWaitTimeout.
//
// If 'fn' returns an error the loop ends and that error is returned. If
// 'fn' returns true the loop ends and nil is returned.
//
// context.Canceled will be returned if the ctx.Done() channel is closed
// without fn ever returning true.
//
// When the ctx.Done() channel is closed, because the golang `select` statement is
// "uniform pseudo-random", the `fn` might still run one or multiple times,
// though eventually `WaitForWithContext` will return.
func WaitForWithContext(ctx context.Context, wait WaitWithContextFunc, fn ConditionWithContextFunc) error {
    waitCtx, cancel := context.WithCancel(context.Background())
    defer cancel()
    c := wait(waitCtx)
    for {
        select {
        case _, open := <-c:
            ok, err := runConditionWithCrashProtectionWithContext(ctx, fn)
            if err != nil {
                return err
            }
            if ok {
                return nil
            }
            if !open {
                return ErrWaitTimeout
            }
        case <-ctx.Done():
            // returning ctx.Err() will break backward compatibility
            return ErrWaitTimeout
        }
    }
}

// poller returns a WaitFunc that will send to the channel every interval until
// timeout has elapsed and then closes the channel.
//
// Over very short intervals you may receive no ticks before the channel is
// closed. A timeout of 0 is interpreted as an infinity, and in such a case
// it would be the caller's responsibility to close the done channel.
// Failure to do so would result in a leaked goroutine.
//
// Output ticks are not buffered. If the channel is not ready to receive an
// item, the tick is skipped.
func poller(interval, timeout time.Duration) WaitWithContextFunc {
    return WaitWithContextFunc(func(ctx context.Context) <-chan struct{} {
        ch := make(chan struct{})

        go func() {
            defer close(ch)

            tick := time.NewTicker(interval)
            defer tick.Stop()

            var after <-chan time.Time
            if timeout != 0 {
                // time.After is more convenient, but it
                // potentially leaves timers around much longer
                // than necessary if we exit early.
                timer := time.NewTimer(timeout)
                after = timer.C
                defer timer.Stop()
            }

            for {
                select {
                case <-tick.C:
                    // If the consumer isn't ready for this signal drop it and
                    // check the other channels.
                    select {
                    case ch <- struct{}{}:
                    default:
                    }
                case <-after:
                    return
                case <-ctx.Done():
                    return
                }
            }
        }()

        return ch
    })
}

// ExponentialBackoffWithContext works with a request context and a Backoff. It ensures that the retry wait never
// exceeds the deadline specified by the request context.
func ExponentialBackoffWithContext(ctx context.Context, backoff Backoff, condition ConditionFunc) error {
    for backoff.Steps > 0 {
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
        }

        if ok, err := runConditionWithCrashProtection(condition); err != nil || ok {
            return err
        }

        if backoff.Steps == 1 {
            break
        }

        waitBeforeRetry := backoff.Step()
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-time.After(waitBeforeRetry):
        }
    }

    return ErrWaitTimeout
}

package main

type Result struct {
    Error error
}

func updateNode(node string) error {
    return nil
}

func main() {
    nodeList := make([]string, 10)
    checkStatus := func(done <-chan interface{}, nodeList ...string) <-chan Result {
        results := make(chan Result)
        go func() {
            defer close(results)
            for _, node := range nodeList {
                var result Result
                // update node
                err := updateNode(node)
                result = Result{Error: err}
                select {
                case <- done:
                    return
                case results <- result:
                }
            }
        }()
        return results
    }

    done := make(chan interface{})
    defer close(done)

    for result := range checkStatus(done, nodeList...) {
        if result.Error != nil {
            panic(result.Error)
        }
    }
}

package main

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), thresholdTime)
    defer cancel()
    g, ctx := errgroup.WithContext(ctx)

    if felixLive {
        g.Go(func() error {
            if err := checkFelixHealth(ctx, felixLivenessEp, "liveness"); err != nil {
                return fmt.Errorf("calico/node is not ready: Felix is not live: %+v", err)
            }
            return nil
        })
    }

    if birdLive {
        g.Go(func() error {
            if err := checkServiceIsLive([]string{"confd", "bird"}); err != nil {
                return fmt.Errorf("calico/node is not ready: bird/confd is not live: %+v", err)
            }
            return nil
        })
    }

    if bird6Live {
        g.Go(func() error {
            if err := checkServiceIsLive([]string{"confd", "bird6"}); err != nil {
                return fmt.Errorf("calico/node is not ready: bird6/confd is not live: %+v", err)
            }
            return nil
        })
    }

    if felixReady {
        g.Go(func() error {
            if err := checkFelixHealth(ctx, felixReadinessEp, "readiness"); err != nil {
                return fmt.Errorf("calico/node is not ready: felix is not ready: %+v", err)
            }
            return nil
        })
    }

    if bird {
        g.Go(func() error {
            if err := checkBIRDReady("4", thresholdTime); err != nil {
                return fmt.Errorf("calico/node is not ready: BIRD is not ready: %+v", err)
            }
            return nil
        })
    }

    if bird6 {
        g.Go(func() error {
            if err := checkBIRDReady("6", thresholdTime); err != nil {
                return fmt.Errorf("calico/node is not ready: BIRD6 is not ready: %+v", err)
            }
            return nil
        })
    }
    if err := g.Wait(); err != nil {
        fmt.Printf("%s", err)
        os.Exit(1)
    }
}

package main

func main() {
    var (
        mu   = sync.Mutex{}
        wg   = sync.WaitGroup{}
        errs []error
    )
    appendError := func(err error) {
        mu.Lock()
        defer mu.Unlock()
        errs = append(errs, err)
    }

    results := make([]schedulerapi.HostPriorityList, len(priorityConfigs), len(priorityConfigs))

    // DEPRECATED: we can remove this when all priorityConfigs implement the
    // Map-Reduce pattern.
    for i := range priorityConfigs {
        if priorityConfigs[i].Function != nil {
            wg.Add(1)
            go func(index int) {
                defer wg.Done()
                var err error
                results[index], err = priorityConfigs[index].Function(pod, nodeNameToInfo, nodes)
                if err != nil {
                    appendError(err)
                }
            }(i)
        } else {
            results[i] = make(schedulerapi.HostPriorityList, len(nodes))
        }
    }

    // Wait for all computations to be finished.
    wg.Wait()
    if len(errs) != 0 {
        return schedulerapi.HostPriorityList{}, errors.NewAggregate(errs)
    }
}

package workqueue

import (
    "context"
    "sync"

    utilruntime "k8s.io/apimachinery/pkg/util/runtime"
)

type DoWorkPieceFunc func(piece int)

type options struct {
    chunkSize int
}

type Options func(*options)

// WithChunkSize allows to set chunks of work items to the workers, rather than
// processing one by one.
// It is recommended to use this option if the number of pieces significantly
// higher than the number of workers and the work done for each item is small.
func WithChunkSize(c int) func(*options) {
    return func(o *options) {
        o.chunkSize = c
    }
}

// ParallelizeUntil is a framework that allows for parallelizing N
// independent pieces of work until done or the context is canceled.
func ParallelizeUntil(ctx context.Context, workers, pieces int, doWorkPiece DoWorkPieceFunc, opts ...Options) {
    if pieces == 0 {
        return
    }
    o := options{}
    for _, opt := range opts {
        opt(&o)
    }
    chunkSize := o.chunkSize
    if chunkSize < 1 {
        chunkSize = 1
    }

    chunks := ceilDiv(pieces, chunkSize)
    toProcess := make(chan int, chunks)
    for i := 0; i < chunks; i++ {
        toProcess <- i
    }
    //不影响读
    close(toProcess)

    var stop <-chan struct{}
    if ctx != nil {
        stop = ctx.Done()
    }
    if chunks < workers {
        workers = chunks
    }
    wg := sync.WaitGroup{}
    wg.Add(workers)
    for i := 0; i < workers; i++ {
        go func() {
            defer utilruntime.HandleCrash()
            defer wg.Done()
            //并发从channel中读取
            for chunk := range toProcess {
                start := chunk * chunkSize
                end := start + chunkSize
                if end > pieces {
                    end = pieces
                }
                for p := start; p < end; p++ {
                    select {
                    case <-stop:
                        return
                    default:
                        doWorkPiece(p)
                    }
                }
            }
        }()
    }
    wg.Wait()
}

func ceilDiv(a, b int) int {
    return (a + b - 1) / b
}

package main

import (
    "sync/atomic"
    "unsafe"
)

// LFStack 无锁栈
// 使用链表实现
type LFStack struct {
    head unsafe.Pointer // 栈顶
}

// Node 节点
type Node struct {
    val  int32
    next unsafe.Pointer
}

// NewLFStack NewLFStack
func NewLFStack() *LFStack {
    n := unsafe.Pointer(&Node{})
    return &LFStack{head: n}
}

// Push 入栈
func (s *LFStack) Push(v int32) {
    n := &Node{val: v}

    for {
        // 先取出栈顶
        old := atomic.LoadPointer(&s.head)
        n.next = old
        if atomic.CompareAndSwapPointer(&s.head, old, unsafe.Pointer(n)) {
            return
        }
    }
}

// Pop 出栈，没有数据时返回 nil
func (s *LFStack) Pop() int32 {
    for {
        // 先取出栈顶
        old := atomic.LoadPointer(&s.head)
        if old == nil {
            return 0
        }

        oldNode := (*Node)(old)
        // 取出下一个节点
        next := atomic.LoadPointer(&oldNode.next)
        // 重置栈顶
        if atomic.CompareAndSwapPointer(&s.head, old, next) {
            return oldNode.val
        }
    }
}

package main

import (
    "fmt"
    "reflect"
    "runtime"
    "time"
)

var (
    MaxWorker = 10
)

type Payload struct {
    Num int
}

//待执行的工作
type Job struct {
    Payload Payload
}

//任务channal
var JobQueue chan Job

//执行任务的工作者单元
type Worker struct {
    WorkerPool chan chan Job //工作者池--每个元素是一个工作者的私有任务channal
    JobChannel chan Job      //每个工作者单元包含一个任务管道 用于获取任务
    quit       chan bool     //退出信号
    no         int           //编号
}

//创建一个新工作者单元
func NewWorker(workerPool chan chan Job, no int) Worker {
    fmt.Println("创建了工作者", no)
    return Worker{
        WorkerPool: workerPool,
        JobChannel: make(chan Job),
        quit:       make(chan bool),
        no:         no,
    }
}

//循环监听任务和结束信号
func (w Worker) Start() {
    go func() {
        for {
            // register the current worker into the worker queue.
            w.WorkerPool <- w.JobChannel
            fmt.Println("w.WorkerPool <- w.JobChannel", w)
            select {
            case job := <-w.JobChannel:
                fmt.Println("job := <-w.JobChannel")
                // 收到任务
                fmt.Println(job)
                time.Sleep(100 * time.Second)
            case <-w.quit:
                // 收到退出信号
                return
            }
        }
    }()
}

// 停止信号
func (w Worker) Stop() {
    go func() {
        w.quit <- true
    }()
}

//调度中心
type Dispatcher struct {
    //工作者池
    WorkerPool chan chan Job
    //工作者数量
    MaxWorkers int
}

//创建调度中心
func NewDispatcher(maxWorkers int) *Dispatcher {
    pool := make(chan chan Job, maxWorkers)
    return &Dispatcher{WorkerPool: pool, MaxWorkers: maxWorkers}
}

//工作者池的初始化
func (d *Dispatcher) Run() {
    // starting n number of workers
    for i := 1; i < d.MaxWorkers+1; i++ {
        worker := NewWorker(d.WorkerPool, i)
        worker.Start()
    }
    go d.dispatch()
}

//调度
func (d *Dispatcher) dispatch() {
    for {
        select {
        case job := <-JobQueue:
            fmt.Println("job := <-JobQueue:")
            go func(job Job) {
                //等待空闲worker (任务多的时候会阻塞这里)
                jobChannel := <-d.WorkerPool
                fmt.Println("jobChannel := <-d.WorkerPool", reflect.TypeOf(jobChannel))
                // 将任务放到上述woker的私有任务channal中
                jobChannel <- job
                fmt.Println("jobChannel <- job")
            }(job)
        }
    }
}

func main() {
    JobQueue = make(chan Job, 10)
    dispatcher := NewDispatcher(MaxWorker)
    dispatcher.Run()
    time.Sleep(1 * time.Second)
    go addQueue()
    time.Sleep(1000 * time.Second)
}

func addQueue() {
    for i := 0; i < 20; i++ {
        // 新建一个任务
        payLoad := Payload{Num: i}
        work := Job{Payload: payLoad}
        // 任务放入任务队列channal
        JobQueue <- work
        fmt.Println("JobQueue <- work", i)
        fmt.Println("当前协程数:", runtime.NumGoroutine())
        time.Sleep(100 * time.Millisecond)
    }
}