watch response latency

latency.go

package main

import (
	"flag"
	"fmt"
	"os"
	"runtime"
	"sync/atomic"
	"time"

	"github.com/coreos/etcd/clientv3"
	"golang.org/x/net/context"
)

func main() {
	intervalP := flag.Duration("t", 100*time.Millisecond, "")
	flag.Parse()
	interval := *intervalP

	runtime.GOMAXPROCS(1)
	// clus := integration.NewClusterV3(nil, &integration.ClusterConfig{Size: 1})
	// defer clus.Terminate(nil)
	// etcdcli := clus.RandClient()
	etcdcli, err := clientv3.New(clientv3.Config{
		Endpoints: []string{"http://127.0.0.1:2379"},
	})
	if err != nil {
		panic(err)
	}

	ctx := context.Background()
	var counter uint64

	go func() {
		for {
			for i := 0; i < 100; i++ {
				go func() {
					ctx2, cancel := context.WithCancel(ctx)
					go func() {
						time.Sleep(interval)
						_, err := etcdcli.Get(ctx2, "/abc")
						if err != nil {
							if err != context.Canceled {
								fmt.Println("read error:", err)
								os.Exit(1)
							}
						}
					}()
					time.Sleep(interval)
					cancel()
				}()
			}
			for i := 0; i < 10; i++ {
				go func() {
					time.Sleep(interval)
					_, err := etcdcli.Put(ctx, "/abc", "haha")
					if err != nil {
						fmt.Println("put error:", err)
						os.Exit(1)
					}
					atomic.AddUint64(&counter, 1)
				}()
			}
			time.Sleep(interval)
		}
	}()

	fmt.Println("watching..========================")
	wcount := uint64(0)
	for {
		now := time.Now()
		wch := etcdcli.Watch(ctx, "/", clientv3.WithPrefix(), clientv3.WithRev(2))
		select {
		case r := <-wch:
			if r.Err() != nil {
				fmt.Println("watch error:", r.Err())
				os.Exit(1)
			}
			wcount++
			pcount := atomic.LoadUint64(&counter)
			fmt.Println("count diff:", pcount-wcount)
			dur := time.Since(now)
			if dur > time.Duration(10*time.Second) {
				panic("too long")
			}
			fmt.Println("time:", dur)
			for _, e := range r.Events {
				// fmt.Printf("event: %s\n", e.Kv.Key)
				e = e
				break
			}
		case <-time.After(5 * time.Second):
			fmt.Println("timeout")
			os.Exit(1)
		}
		time.Sleep(interval)
	}
}