github-vet / archived-findings Goto Github PK

		for tIdx, d := range block.Data.Data {
			// ensure that we don't have too many concurrent validation workers
			v.Support.Acquire(context.Background(), 1)

			go func(index int, data []byte) {
				defer v.Support.Release(1)

				v.validateTx(&blockValidationRequest{
					d:     data,
					block: block,
					tIdx:  index,
				}, results)
			}(tIdx, d)
		}

	for i := 0; i <= 20; i++ {
		go func() {
			fmt.Print(i, " ")
		}()
	}

				for _, sig := range sigs {
					// we need to resend these signals so it can be caught in the
					// main routine...
					go func() { sigCh <- sig }()
					if sysSig, ok := sig.(syscall.Signal); ok {
						if sysSig != syscall.SIGHUP {
							gotSig = true
						}
					}
				}

	for i, cfg := range testCfgs {
		donec := make(chan error)
		go func() {
			// without timeout, dial continues forever on ipv4 black hole
			c, err := New(cfg)
			if c != nil || err == nil {
				t.Errorf("#%d: new client should fail", i)
			}
			donec <- err
		}()

		time.Sleep(10 * time.Millisecond)

		select {
		case err := <-donec:
			t.Errorf("#%d: dial didn't wait (%v)", i, err)
		default:
		}

		select {
		case <-time.After(5 * time.Second):
			t.Errorf("#%d: failed to timeout dial on time", i)
		case err := <-donec:
			if err != context.DeadlineExceeded {
				t.Errorf("#%d: unexpected error %v, want %v", i, err, context.DeadlineExceeded)
			}
		}
	}

    for _, listener := range Serv.Listeners {
        log.Printf("Now serving on: %s\n", listener.Listener.Addr().String())
        go func() {
            for {
                /* Accept new connection */
                conn, err := listener.Accept()
                if err != nil {
                    log.Printf("Accept connection failed: %s\n", err.Error())
                }

                /* Run this in a separate goroutine to get straight back to accepting */
                go func() {
                    /* Setup new client */
                    client := NewIrcClient(conn)
                    client.Init()

                    /* Add the client to the server client map */
                    Server.AddClient(client)

                    /* Start the client loop :) */
                    go client.Serve()
                }()
            }
        }()
    }

	for i, cfg := range testCfgs {
		donec := make(chan error)
		go func() {
			// without timeout, dial continues forever on ipv4 black hole
			c, err := New(cfg)
			if c != nil || err == nil {
				t.Errorf("#%d: new client should fail", i)
			}
			donec <- err
		}()

		time.Sleep(10 * time.Millisecond)

		select {
		case err := <-donec:
			t.Errorf("#%d: dial didn't wait (%v)", i, err)
		default:
		}

		select {
		case <-time.After(5 * time.Second):
			t.Errorf("#%d: failed to timeout dial on time", i)
		case err := <-donec:
			if err != context.DeadlineExceeded {
				t.Errorf("#%d: unexpected error %v, want %v", i, err, context.DeadlineExceeded)
			}
		}
	}

	for _, network := range []string{"tcp", "unix", "unixpacket"} {
		if !nettest.TestableNetwork(network) {
			continue
		}
		ln, err := nettest.NewLocalListener(network)
		if err != nil {
			t.Error(err)
			continue
		}
		defer func() {
			path := ln.Addr().String()
			ln.Close()
			if network == "unix" || network == "unixpacket" {
				os.Remove(path)
			}
		}()
		c, err := net.Dial(ln.Addr().Network(), ln.Addr().String())
		if err != nil {
			t.Error(err)
			continue
		}
		defer c.Close()
		if _, err := netreflect.SocketOf(c); err != nil {
			t.Error(err)
			continue
		}
	}

	for _, eventType := range table {
		out, in := io.Pipe()

		decoder := restclientwatch.NewDecoder(streaming.NewDecoder(out, getDecoder()), getDecoder())

		expect := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo"}}
		encoder := json.NewEncoder(in)
		go func() {
			data, err := runtime.Encode(scheme.Codecs.LegacyCodec(v1.SchemeGroupVersion), expect)
			if err != nil {
				t.Fatalf("Unexpected error %v", err)
			}
			event := metav1.WatchEvent{
				Type:   string(eventType),
				Object: runtime.RawExtension{Raw: json.RawMessage(data)},
			}
			if err := encoder.Encode(&event); err != nil {
				t.Errorf("Unexpected error %v", err)
			}
			in.Close()
		}()

		done := make(chan struct{})
		go func() {
			action, got, err := decoder.Decode()
			if err != nil {
				t.Fatalf("Unexpected error %v", err)
			}
			if e, a := eventType, action; e != a {
				t.Errorf("Expected %v, got %v", e, a)
			}
			if e, a := expect, got; !apiequality.Semantic.DeepDerivative(e, a) {
				t.Errorf("Expected %v, got %v", e, a)
			}
			t.Logf("Exited read")
			close(done)
		}()
		<-done

		done = make(chan struct{})
		go func() {
			_, _, err := decoder.Decode()
			if err == nil {
				t.Errorf("Unexpected nil error")
			}
			close(done)
		}()
		<-done

		decoder.Close()
	}

	for i := 0; i < int(n); i++ {
		go func() {
			r, e := spawn(s, i, p)
			results <- r
			errs <- e
		}()
	}

	for _, s1 := range swarms {
		log.Debugf("-------------------------------------------------------")
		log.Debugf("%s ping pong round", s1.LocalPeer())
		log.Debugf("-------------------------------------------------------")

		_, cancel := context.WithCancel(ctx)
		got := map[peer.ID]int{}
		errChan := make(chan error, MsgNum*len(swarms))
		streamChan := make(chan inet.Stream, MsgNum)

		// send out "ping" x MsgNum to every peer
		go func() {
			defer close(streamChan)

			var wg sync.WaitGroup
			send := func(p peer.ID) {
				defer wg.Done()

				// first, one stream per peer (nice)
				stream, err := s1.NewStream(ctx, p)
				if err != nil {
					errChan <- err
					return
				}

				// send out ping!
				for k := 0; k < MsgNum; k++ { // with k messages
					msg := "ping"
					log.Debugf("%s %s %s (%d)", s1.LocalPeer(), msg, p, k)
					if _, err := stream.Write([]byte(msg)); err != nil {
						errChan <- err
						continue
					}
				}

				// read it later
				streamChan <- stream
			}

			for _, s2 := range swarms {
				if s2.LocalPeer() == s1.LocalPeer() {
					continue // dont send to self...
				}

				wg.Add(1)
				go send(s2.LocalPeer())
			}
			wg.Wait()
		}()

		// receive "pong" x MsgNum from every peer
		go func() {
			defer close(errChan)
			count := 0
			countShouldBe := MsgNum * (len(swarms) - 1)
			for stream := range streamChan { // one per peer
				defer stream.Close()

				// get peer on the other side
				p := stream.Conn().RemotePeer()

				// receive pings
				msgCount := 0
				msg := make([]byte, 4)
				for k := 0; k < MsgNum; k++ { // with k messages

					// read from the stream
					if _, err := stream.Read(msg); err != nil {
						errChan <- err
						continue
					}

					if string(msg) != "pong" {
						errChan <- fmt.Errorf("unexpected message: %s", msg)
						continue
					}

					log.Debugf("%s %s %s (%d)", s1.LocalPeer(), msg, p, k)
					msgCount++
				}

				got[p] = msgCount
				count += msgCount
			}

			if count != countShouldBe {
				errChan <- fmt.Errorf("count mismatch: %d != %d", count, countShouldBe)
			}
		}()

		// check any errors (blocks till consumer is done)
		for err := range errChan {
			if err != nil {
				t.Error(err.Error())
			}
		}

		log.Debugf("%s got pongs", s1.LocalPeer())
		if (len(swarms) - 1) != len(got) {
			t.Errorf("got (%d) less messages than sent (%d).", len(got), len(swarms))
		}

		for p, n := range got {
			if n != MsgNum {
				t.Error("peer did not get all msgs", p, n, "/", MsgNum)
			}
		}

		cancel()
		<-time.After(10 * time.Millisecond)
	}

	for i, tt := range tests {
		clus := NewClusterV3(t, &ClusterConfig{Size: 3})

		wAPI := toGRPC(clus.RandClient()).Watch
		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer cancel()
		wStream, err := wAPI.Watch(ctx)
		if err != nil {
			t.Fatalf("#%d: wAPI.Watch error: %v", i, err)
		}

		err = wStream.Send(tt.watchRequest)
		if err != nil {
			t.Fatalf("#%d: wStream.Send error: %v", i, err)
		}

		// ensure watcher request created a new watcher
		cresp, err := wStream.Recv()
		if err != nil {
			t.Errorf("#%d: wStream.Recv error: %v", i, err)
			clus.Terminate(t)
			continue
		}
		if !cresp.Created {
			t.Errorf("#%d: did not create watchid, got %+v", i, cresp)
			clus.Terminate(t)
			continue
		}
		if cresp.Canceled {
			t.Errorf("#%d: canceled watcher on create %+v", i, cresp)
			clus.Terminate(t)
			continue
		}

		createdWatchId := cresp.WatchId
		if cresp.Header == nil || cresp.Header.Revision != 1 {
			t.Errorf("#%d: header revision got +%v, wanted revison 1", i, cresp)
			clus.Terminate(t)
			continue
		}

		// asynchronously create keys
		go func() {
			for _, k := range tt.putKeys {
				kvc := toGRPC(clus.RandClient()).KV
				req := &pb.PutRequest{Key: []byte(k), Value: []byte("bar")}
				if _, err := kvc.Put(context.TODO(), req); err != nil {
					t.Fatalf("#%d: couldn't put key (%v)", i, err)
				}
			}
		}()

		// check stream results
		for j, wresp := range tt.wresps {
			resp, err := wStream.Recv()
			if err != nil {
				t.Errorf("#%d.%d: wStream.Recv error: %v", i, j, err)
			}

			if resp.Header == nil {
				t.Fatalf("#%d.%d: unexpected nil resp.Header", i, j)
			}
			if resp.Header.Revision != wresp.Header.Revision {
				t.Errorf("#%d.%d: resp.Header.Revision got = %d, want = %d", i, j, resp.Header.Revision, wresp.Header.Revision)
			}

			if wresp.Created != resp.Created {
				t.Errorf("#%d.%d: resp.Created got = %v, want = %v", i, j, resp.Created, wresp.Created)
			}
			if resp.WatchId != createdWatchId {
				t.Errorf("#%d.%d: resp.WatchId got = %d, want = %d", i, j, resp.WatchId, createdWatchId)
			}

			if !reflect.DeepEqual(resp.Events, wresp.Events) {
				t.Errorf("#%d.%d: resp.Events got = %+v, want = %+v", i, j, resp.Events, wresp.Events)
			}
		}

		rok, nr := waitResponse(wStream, 1*time.Second)
		if !rok {
			t.Errorf("unexpected pb.WatchResponse is received %+v", nr)
		}

		// can't defer because tcp ports will be in use
		clus.Terminate(t)
	}

	for _, host := range p.Hosts {
		go func() {
			results <- deploy(p, host)
		}()
	}

	for i := 0; i < 10; i++ {
		var wg sync.WaitGroup
		wg.Add(len(users))
		for u := range users {
			go func(user string) {
				defer wg.Done()
				token := fmt.Sprintf("%s(%d)", user, i)
				ctx := context.WithValue(context.WithValue(context.TODO(), AuthenticateParamIndex{}, uint64(1)), AuthenticateParamSimpleTokenPrefix{}, token)
				if _, err := as.Authenticate(ctx, user, "123"); err != nil {
					t.Fatal(err)
				}
				if _, err := as.AuthInfoFromCtx(ctx); err != nil {
					t.Fatal(err)
				}
			}(u)
		}
		time.Sleep(time.Millisecond)
		wg.Wait()
	}

	for i := range thenOps {
		clus.Members[0].Stop(t)
		<-clus.Members[0].StopNotify()

		donec := make(chan struct{})
		go func() {
			_, err := kv.Txn(context.TODO()).Then(thenOps[i]...).Commit()
			if err != nil {
				t.Fatalf("expected response, got error %v", err)
			}
			donec <- struct{}{}
		}()
		// wait for txn to fail on disconnect
		time.Sleep(100 * time.Millisecond)

		// restart node; client should resume
		clus.Members[0].Restart(t)
		select {
		case <-donec:
		case <-time.After(2 * clus.Members[1].ServerConfig.ReqTimeout()):
			t.Fatalf("waited too long")
		}
	}

	for _, step := range d.Release.Spec.Lifecycle.V1 {
		stepShared := step.Shared()
		stepID := stepShared.ID
		if stepID != requestedStep {
			continue
		}

		if ok := d.maybeAbortDueToMissingRequirement(
			stepShared.Requires,
			c,
			requestedStep,
		); !ok {
			return
		}

		errChan := make(chan error)
		d.StepProgress.Store(stepID, daemontypes.JSONProgress("v2router", map[string]interface{}{
			"status":  "working",
			"message": "working",
		}))
		go func() {
			errChan <- d.StepExecutor(d, step)
		}()
		// hack, give it 10 ms in case its an instant step. Hydrate and send will read progress from the syncMap
		time.Sleep(10 * time.Millisecond)

		d.completeMut.Lock()
		d.hydrateAndSend(step, c)
		go func() {
			defer d.completeMut.Unlock()
			d.handleAsync(errChan, debug, step, stepID)
		}()
		return
	}

	for _, addr := range binds {
		go func(bind string) {
			if err := http.ListenAndServe(bind, monitorServeMux); err != nil {
				log.Error("http.ListenAndServe(\"%s\", pprofServeMux) error(%v)", addr, err)
				panic(err)
			}
		}(addr)
	}

	for _, u := range cfg.LCUrls {
		sctx := newServeCtx(cfg.logger)
		if u.Scheme == "http" || u.Scheme == "unix" {
			if !cfg.ClientTLSInfo.Empty() {
				if cfg.logger != nil {
					cfg.logger.Warn("scheme is HTTP while key and cert files are present; ignoring key and cert files", zap.String("client-url", u.String()))
				} else {
					plog.Warningf("The scheme of client url %s is HTTP while peer key/cert files are presented. Ignored key/cert files.", u.String())
				}
			}
			if cfg.ClientTLSInfo.ClientCertAuth {
				if cfg.logger != nil {
					cfg.logger.Warn("scheme is HTTP while --client-cert-auth is enabled; ignoring client cert auth for this URL", zap.String("client-url", u.String()))
				} else {
					plog.Warningf("The scheme of client url %s is HTTP while client cert auth (--client-cert-auth) is enabled. Ignored client cert auth for this url.", u.String())
				}
			}
		}
		if (u.Scheme == "https" || u.Scheme == "unixs") && cfg.ClientTLSInfo.Empty() {
			return nil, fmt.Errorf("TLS key/cert (--cert-file, --key-file) must be provided for client url %s with HTTPs scheme", u.String())
		}

		network := "tcp"
		addr := u.Host
		if u.Scheme == "unix" || u.Scheme == "unixs" {
			network = "unix"
			addr = u.Host + u.Path
		}
		sctx.network = network

		sctx.secure = u.Scheme == "https" || u.Scheme == "unixs"
		sctx.insecure = !sctx.secure
		if oldctx := sctxs[addr]; oldctx != nil {
			oldctx.secure = oldctx.secure || sctx.secure
			oldctx.insecure = oldctx.insecure || sctx.insecure
			continue
		}

		if sctx.l, err = net.Listen(network, addr); err != nil {
			return nil, err
		}
		// net.Listener will rewrite ipv4 0.0.0.0 to ipv6 [::], breaking
		// hosts that disable ipv6. So, use the address given by the user.
		sctx.addr = addr

		if fdLimit, fderr := runtimeutil.FDLimit(); fderr == nil {
			if fdLimit <= reservedInternalFDNum {
				if cfg.logger != nil {
					cfg.logger.Fatal(
						"file descriptor limit of etcd process is too low; please set higher",
						zap.Uint64("limit", fdLimit),
						zap.Int("recommended-limit", reservedInternalFDNum),
					)
				} else {
					plog.Fatalf("file descriptor limit[%d] of etcd process is too low, and should be set higher than %d to ensure internal usage", fdLimit, reservedInternalFDNum)
				}
			}
			sctx.l = transport.LimitListener(sctx.l, int(fdLimit-reservedInternalFDNum))
		}

		if network == "tcp" {
			if sctx.l, err = transport.NewKeepAliveListener(sctx.l, network, nil); err != nil {
				return nil, err
			}
		}

		defer func() {
			if err == nil {
				return
			}
			sctx.l.Close()
			if cfg.logger != nil {
				cfg.logger.Warn(
					"closing peer listener",
					zap.String("address", u.Host),
					zap.Error(err),
				)
			} else {
				plog.Info("stopping listening for client requests on ", u.Host)
			}
		}()
		for k := range cfg.UserHandlers {
			sctx.userHandlers[k] = cfg.UserHandlers[k]
		}
		sctx.serviceRegister = cfg.ServiceRegister
		if cfg.EnablePprof || cfg.Debug {
			sctx.registerPprof()
		}
		if cfg.Debug {
			sctx.registerTrace()
		}
		sctxs[addr] = sctx
	}

	for i, tt := range tests {
		txn := s.Write()
		done := make(chan struct{}, 1)
		go func() {
			tt()
			done <- struct{}{}
		}()
		select {
		case <-done:
			t.Fatalf("#%d: operation failed to be blocked", i)
		case <-time.After(10 * time.Millisecond):
		}

		txn.End()
		select {
		case <-done:
		case <-time.After(10 * time.Second):
			testutil.FatalStack(t, fmt.Sprintf("#%d: operation failed to be unblocked", i))
		}
	}

	for _, test := range tests {
		defaultConfig := NewDefaultConfig()
		defaultConfig.PollingWaitTime = waitTime
		configs := &configContainer{
			client: &defaultConfig,
			server: &serverConfig{
				scope: test.testScope,
			},
		}

		endpoint := newFakeMarathonEndpoint(t, configs)
		defer endpoint.Close()

		errCh := make(chan error)
		go func() {
			errCh <- endpoint.Client.WaitOnApplication(test.appName, test.timeout)
		}()

		select {
		case <-time.After(400 * time.Millisecond):
			assert.Fail(t, fmt.Sprintf("%s: WaitOnApplication did not complete in time", test.desc))
		case err := <-errCh:
			if test.shouldSucceed {
				assert.NoError(t, err, test.desc)
			} else {
				assert.IsType(t, err, ErrTimeoutError, test.desc)
			}
		}
	}

	for _, u := range cfg.LCUrls {
		sctx := newServeCtx()

		if u.Scheme == "http" || u.Scheme == "unix" {
			if !cfg.ClientTLSInfo.Empty() {
				plog.Warningf("The scheme of client url %s is HTTP while peer key/cert files are presented. Ignored key/cert files.", u.String())
			}
			if cfg.ClientTLSInfo.ClientCertAuth {
				plog.Warningf("The scheme of client url %s is HTTP while client cert auth (--client-cert-auth) is enabled. Ignored client cert auth for this url.", u.String())
			}
		}
		if (u.Scheme == "https" || u.Scheme == "unixs") && cfg.ClientTLSInfo.Empty() {
			return nil, fmt.Errorf("TLS key/cert (--cert-file, --key-file) must be provided for client url %s with HTTPs scheme", u.String())
		}

		proto := "tcp"
		addr := u.Host
		if u.Scheme == "unix" || u.Scheme == "unixs" {
			proto = "unix"
			addr = u.Host + u.Path
		}

		sctx.secure = u.Scheme == "https" || u.Scheme == "unixs"
		sctx.insecure = !sctx.secure
		if oldctx := sctxs[addr]; oldctx != nil {
			oldctx.secure = oldctx.secure || sctx.secure
			oldctx.insecure = oldctx.insecure || sctx.insecure
			continue
		}

		if sctx.l, err = net.Listen(proto, addr); err != nil {
			return nil, err
		}
		// net.Listener will rewrite ipv4 0.0.0.0 to ipv6 [::], breaking
		// hosts that disable ipv6. So, use the address given by the user.
		sctx.addr = addr

		if fdLimit, fderr := runtimeutil.FDLimit(); fderr == nil {
			if fdLimit <= reservedInternalFDNum {
				plog.Fatalf("file descriptor limit[%d] of etcd process is too low, and should be set higher than %d to ensure internal usage", fdLimit, reservedInternalFDNum)
			}
			sctx.l = transport.LimitListener(sctx.l, int(fdLimit-reservedInternalFDNum))
		}

		if proto == "tcp" {
			if sctx.l, err = transport.NewKeepAliveListener(sctx.l, "tcp", nil); err != nil {
				return nil, err
			}
		}

		plog.Info("listening for client requests on ", u.Host)
		defer func() {
			if err != nil {
				sctx.l.Close()
				plog.Info("stopping listening for client requests on ", u.Host)
			}
		}()
		for k := range cfg.UserHandlers {
			sctx.userHandlers[k] = cfg.UserHandlers[k]
		}
		sctx.serviceRegister = cfg.ServiceRegister
		if cfg.EnablePprof || cfg.Debug {
			sctx.registerPprof()
		}
		if cfg.Debug {
			sctx.registerTrace()
		}
		sctxs[addr] = sctx
	}

	for name, kex := range kexAlgoMap {
		a, b := memPipe()

		s := make(chan kexResultErr, 1)
		c := make(chan kexResultErr, 1)
		var magics handshakeMagics
		go func() {
			r, e := kex.Client(a, rand.Reader, &magics)
			c <- kexResultErr{r, e}
		}()
		go func() {
			r, e := kex.Server(b, rand.Reader, &magics, testSigners["ecdsa"])
			s <- kexResultErr{r, e}
		}()

		clientRes := <-c
		serverRes := <-s
		if clientRes.err != nil {
			t.Errorf("client: %v", clientRes.err)
		}
		if serverRes.err != nil {
			t.Errorf("server: %v", serverRes.err)
		}
		if !reflect.DeepEqual(clientRes.result, serverRes.result) {
			t.Errorf("kex %q: mismatch %#v, %#v", name, clientRes.result, serverRes.result)
		}
	}

	for _, fileName := range args[1:] {
		workerPool.Add()
		fmt.Printf("Writing %s\n", fileName)

		go func(f string, wp *swg.SizedWaitGroup) {
			ct, err := writeMRTFile(bc, f, sessID, filts)
			results <- writeMRTResult{fileName: f, msgCt: ct, err: err}
			wp.Done()
		}(fileName, &workerPool)
	}

	for req := 0; req < concurrency; req++ {
		c.wg.Add(1)
		totalSentRequests.Add(1)
		go func(n int, p params) {
			defer c.wg.Done()
			defer totalSentRequests.Done()

			var err error
			for i := 1; i <= AttemptLimit; i++ {
				c.wg.Add(1)
				defer c.wg.Done()
				select {
				case <-finishCh:
					return
				default:
				}

				// rehydrate the body (it is drained each read)
				if len(originalRequestBody) > 0 {
					p.req.Body = ioutil.NopCloser(bytes.NewBuffer(originalRequestBody))
				}
				if len(originalBody) > 0 {
					p.body = bytes.NewBuffer(originalBody)
				}

				var resp *http.Response
				// route the calls
				switch p.method {
				case "Do":
					resp, err = httpClient.Do(p.req)
				case "Get":
					resp, err = httpClient.Get(p.url)
				case "Head":
					resp, err = httpClient.Head(p.url)
				case "Post":
					resp, err = httpClient.Post(p.url, p.bodyType, p.body)
				case "PostForm":
					resp, err = httpClient.PostForm(p.url, p.data)
				default:
					err = ErrUnexpectedMethod
				}

				// Early return if we have a valid result
				// Only retry (ie, continue the loop) on 5xx status codes
				if err == nil && resp.StatusCode < 500 {
					multiplexCh <- result{resp: resp, err: err, req: n, retry: i}
					return
				}

				c.log(ErrEntry{
					Time:    time.Now(),
					Method:  p.method,
					Verb:    p.verb,
					URL:     p.url,
					Request: n,
					Retry:   i + 1, // would remove, but would break backward compatibility
					Attempt: i,
					Err:     err,
				})

				// if it is the last iteration, grab the result (which is an error at this point)
				if i == AttemptLimit {
					multiplexCh <- result{resp: resp, err: err}
					return
				}

				// if we are retrying, we should close this response body to free the fd
				if resp != nil {
					resp.Body.Close()
				}

				// prevent a 0 from causing the tick to block, pass additional microsecond
				<-time.After(c.Backoff(i) + 1*time.Microsecond)
			}
		}(req, p)
	}

	for i := range tests {
		lkv, closeLKV, err := leasing.NewKV(clus.Client(0), "foo/", concurrency.WithTTL(1))
		testutil.AssertNil(t, err)
		defer closeLKV()

		if _, err = lkv.Get(context.TODO(), "abc"); err != nil {
			t.Fatal(err)
		}

		// down endpoint lkv uses for keepalives
		clus.Members[0].Stop(t)
		if err := waitForLeasingExpire(clus.Client(1), "foo/abc"); err != nil {
			t.Fatal(err)
		}
		waitForExpireAck(t, lkv)

		ctx, cancel := context.WithCancel(context.TODO())
		errc := make(chan error, 1)
		go func() { errc <- tests[i](ctx, lkv) }()
		// some delay to get past for ctx.Err() != nil {} loops
		time.Sleep(100 * time.Millisecond)
		cancel()

		select {
		case err := <-errc:
			if err != ctx.Err() {
				t.Errorf("#%d: expected %v, got %v", i, ctx.Err(), err)
			}
		case <-time.After(5 * time.Second):
			t.Errorf("#%d: timed out waiting for cancel", i)
		}
		clus.Members[0].Restart(t)
	}

	for i := 0; i < 10; i++ {
		var wg sync.WaitGroup
		wg.Add(len(users))
		for u := range users {
			go func(user string) {
				defer wg.Done()
				token := fmt.Sprintf("%s(%d)", user, i)
				ctx := context.WithValue(context.WithValue(context.TODO(), "index", uint64(1)), "simpleToken", token)
				if _, err := as.Authenticate(ctx, user, "123"); err != nil {
					t.Fatal(err)
				}
				if _, err := as.AuthInfoFromCtx(ctx); err != nil {
					t.Fatal(err)
				}
			}(u)
		}
		time.Sleep(time.Millisecond)
		wg.Wait()
	}

		for run := 0; run < testRuns; run++ {
			barrier := make(chan struct{})
			errorsCh := make(chan error, workers)

			go func() {
				for w := 0; w < workers; w++ {
					wg.Add(1)
					go func(run, worker int) {
						defer wg.Done()
						ctx, _ := context.WithDeadline(context.Background(), time.Now().Add(5*time.Second))
						time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
						// all workers will block here until the for loop above has launched all the worker go-routines
						// this is to ensure we fire all the workers off at the same
						<-barrier
						_, err := provider.NewAccessResponse(ctx, request)
						errorsCh <- err
					}(run, w)
				}

				// wait until all workers have completed their work
				wg.Wait()
				close(errorsCh)
			}()

			// let the race begin!
			// all worker go-routines will now attempt to hit the "NewAccessResponse" method
			close(barrier)

			// process worker results

			// successCount is the number of workers that were able to call "NewAccessResponse" without receiving an error.
			// if the successCount at the end of a test run is bigger than one, it means that multiple access/refresh tokens
			// were issued using the same refresh token! - https://knowyourmeme.com/memes/scared-hamster
			var successCount int
			for err := range errorsCh {
				if err != nil {
					if e := (&fosite.RFC6749Error{}); errors.As(err, &e) {
						switch e.ErrorField {

						// change logic below when the refresh handler starts returning 'fosite.ErrInvalidRequest' for other reasons.
						// as of now, this error is only returned due to concurrent transactions competing to refresh using the same token.

						case fosite.ErrInvalidRequest.ErrorField:
							// the error description copy is defined by RFC 6749 and should not be different regardless of
							// the underlying transactional aware storage backend used by hydra
							assert.Equal(t, fosite.ErrInvalidRequest.DescriptionField, e.DescriptionField)
							// the database error debug copy will be different depending on the underlying database used
							switch dbName {
							case dbal.DriverMySQL:
							case dbal.DriverPostgreSQL, dbal.DriverCockroachDB:
								var matched bool
								for _, errSubstr := range []string{
									// both postgreSQL & cockroachDB return error code 40001 for consistency errors as a result of
									// using the REPEATABLE_READ isolation level
									"SQLSTATE 40001",
									// possible if one worker starts the transaction AFTER another worker has successfully
									// refreshed the token and committed the transaction
									"not_found",
									// postgres: duplicate key value violates unique constraint "hydra_oauth2_access_request_id_idx": Unable to insert or update resource because a resource with that value exists already: The request could not be completed due to concurrent access
									"duplicate key",
									// cockroach: restart transaction: TransactionRetryWithProtoRefreshError: TransactionRetryError: retry txn (RETRY_WRITE_TOO_OLD - WriteTooOld flag converted to WriteTooOldError): "sql txn" meta={id=7f069400 key=/Table/62/2/"02a55d6e-509b-4d7a-8458-5828b2f831a1"/0 pri=0.00598277 epo=0 ts=1600955431.566576173,2 min=1600955431.566576173,0 seq=6} lock=true stat=PENDING rts=1600955431.566576173,2 wto=false max=1600955431.566576173,0: Unable to serialize access due to a concurrent update in another session: The request could not be completed due to concurrent access
									"RETRY_WRITE_TOO_OLD",
									// postgres: pq: deadlock detected
									"deadlock detected",
									// postgres: pq: could not serialize access due to concurrent update: Unable to serialize access due to a concurrent update in another session: The request could not be completed due to concurrent access
									"concurrent update",
								} {
									if strings.Contains(e.DebugField, errSubstr) {
										matched = true
										break
									}
								}

								assert.True(t, matched, "received an unexpected kind of `fosite.ErrInvalidRequest`\n"+
									"DB version: %s\n"+
									"Error description: %s\n"+
									"Error debug: %s\n"+
									"Error hint: %s\n"+
									"Raw error: %T %+v\n"+
									"Raw cause: %T %+v",
									storageVersion,
									e.DescriptionField,
									e.DebugField,
									e.HintField,
									err, err,
									e, e)
							}
						case fosite.ErrServerError.ErrorField:
							// this happens when there is an error with the storage
							if dbName == dbal.DriverCockroachDB && strings.Contains(e.DebugField, "RETRY_WRITE_TOO_OLD") {
								break
							}
							fallthrough
						default:
							// unfortunately, MySQL does not offer the same behaviour under the "REPEATABLE_READ" isolation
							// level so we have to relax this assertion just for MySQL for the time being as server_errors
							// resembling the following can be returned:
							//
							//    Error 1213: Deadlock found when trying to get lock; try restarting transaction
							if dbName != dbal.DriverMySQL {
								t.Errorf("an unexpected RFC6749 error with the name %q was returned.\n"+
									"Hint: has the refresh token error handling changed in fosite? If so, you need to add further "+
									"assertions here to cover the additional errors that are being returned by the handler.\n"+
									"DB version: %s\n"+
									"Error description: %s\n"+
									"Error debug: %s\n"+
									"Error hint: %s\n"+
									"Raw error: %+v",
									e.ErrorField,
									storageVersion,
									e.DescriptionField,
									e.DebugField,
									e.HintField,
									err)
							}
						}
					} else {
						t.Errorf("expected underlying error to be of type '*fosite.RFC6749Error', but it was "+
							"actually of type %T: %+v - DB version: %s", err, err, storageVersion)
					}
				} else {
					successCount++
				}
			}

			// IMPORTANT - skip consistency check for MySQL :(
			//
			// different DBMS's provide different consistency guarantees when using the "REPEATABLE_READ" isolation level
			// Currently, MySQL's implementation of "REPEATABLE_READ" makes it possible for multiple concurrent requests
			// to successfully utilize the same refresh token. Therefore, we skip the assertion below.
			//
			// TODO: this needs to be addressed by making it possible to use different isolation levels for various authorization
			//       flows depending on the underlying hydra storage backend. For example, if using MySQL, hydra should force
			//       the transaction isolation level to be "Serializable" when a request to the token handler is received.

			switch dbName {
			case dbal.DriverMySQL:
			case dbal.DriverPostgreSQL, dbal.DriverCockroachDB:
				require.Equal(t, 1, successCount, "CRITICAL: in test iteration %d, %d out of %d workers "+
					"were able to use the refresh token. Exactly ONE was expected to be have been successful.",
					run,
					successCount,
					workers)
			}

			// reset state for the next test iteration
			assert.NoError(t, dbRegistry.OAuth2Storage().RevokeRefreshToken(ctx, request.ID))
			assert.NoError(t, dbRegistry.OAuth2Storage().CreateRefreshTokenSession(ctx, tokenSignature, request))
		}

	for i := 0; i < b.N; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			b.Errorf("TestFailed: %d", i)
		}(i)
	}

	for i, tt := range tests {
		clus := NewClusterV3(t, &ClusterConfig{Size: 3})

		wAPI := toGRPC(clus.RandClient()).Watch
		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer cancel()
		wStream, err := wAPI.Watch(ctx)
		if err != nil {
			t.Fatalf("#%d: wAPI.Watch error: %v", i, err)
		}

		err = wStream.Send(tt.watchRequest)
		if err != nil {
			t.Fatalf("#%d: wStream.Send error: %v", i, err)
		}

		// ensure watcher request created a new watcher
		cresp, err := wStream.Recv()
		if err != nil {
			t.Errorf("#%d: wStream.Recv error: %v", i, err)
			clus.Terminate(t)
			continue
		}
		if !cresp.Created {
			t.Errorf("#%d: did not create watchid, got %+v", i, cresp)
			clus.Terminate(t)
			continue
		}
		if cresp.Canceled {
			t.Errorf("#%d: canceled watcher on create %+v", i, cresp)
			clus.Terminate(t)
			continue
		}

		createdWatchId := cresp.WatchId
		if cresp.Header == nil || cresp.Header.Revision != 1 {
			t.Errorf("#%d: header revision got +%v, wanted revison 1", i, cresp)
			clus.Terminate(t)
			continue
		}

		// asynchronously create keys
		go func() {
			for _, k := range tt.putKeys {
				kvc := toGRPC(clus.RandClient()).KV
				req := &pb.PutRequest{Key: []byte(k), Value: []byte("bar")}
				if _, err := kvc.Put(context.TODO(), req); err != nil {
					t.Fatalf("#%d: couldn't put key (%v)", i, err)
				}
			}
		}()

		// check stream results
		for j, wresp := range tt.wresps {
			resp, err := wStream.Recv()
			if err != nil {
				t.Errorf("#%d.%d: wStream.Recv error: %v", i, j, err)
			}

			if resp.Header == nil {
				t.Fatalf("#%d.%d: unexpected nil resp.Header", i, j)
			}
			if resp.Header.Revision != wresp.Header.Revision {
				t.Errorf("#%d.%d: resp.Header.Revision got = %d, want = %d", i, j, resp.Header.Revision, wresp.Header.Revision)
			}

			if wresp.Created != resp.Created {
				t.Errorf("#%d.%d: resp.Created got = %v, want = %v", i, j, resp.Created, wresp.Created)
			}
			if resp.WatchId != createdWatchId {
				t.Errorf("#%d.%d: resp.WatchId got = %d, want = %d", i, j, resp.WatchId, createdWatchId)
			}

			if !reflect.DeepEqual(resp.Events, wresp.Events) {
				t.Errorf("#%d.%d: resp.Events got = %+v, want = %+v", i, j, resp.Events, wresp.Events)
			}
		}

		rok, nr := waitResponse(wStream, 1*time.Second)
		if !rok {
			t.Errorf("unexpected pb.WatchResponse is received %+v", nr)
		}

		// can't defer because tcp ports will be in use
		clus.Terminate(t)
	}

	for i := range thenOps {
		clus.Members[0].Stop(t)
		<-clus.Members[0].StopNotify()

		donec := make(chan struct{})
		go func() {
			_, err := kv.Txn(context.TODO()).Then(thenOps[i]...).Commit()
			if err != nil {
				t.Fatalf("expected response, got error %v", err)
			}
			donec <- struct{}{}
		}()
		// wait for txn to fail on disconnect
		time.Sleep(100 * time.Millisecond)

		// restart node; client should resume
		clus.Members[0].Restart(t)
		select {
		case <-donec:
		case <-time.After(2 * clus.Members[1].ServerConfig.ReqTimeout()):
			t.Fatalf("waited too long")
		}
	}

	for i, v := range s {
		go func() {
			println(i) // want "loop variable i captured by func literal"
			println(v) // want "loop variable v captured by func literal"
		}()
	}

	for _, addr := range pprofBind {
		go func() {
			if err := http.ListenAndServe(addr, pprofServeMux); err != nil {
				log.Error("http.ListenAndServe(\"%s\", pprofServeMux) error(%v)", addr, err)
				panic(err)
			}
		}()
	}

	for i, msg := range playerMap {
		go func() {
			// Mimic lag, sleep the code for 'lag' milliseconds
			time.Sleep(time.Duration(lag) * time.Millisecond)
			m.Broadcast(msg)
		}()
		delete(playerMap,i)
	}

		for _, event := range events {
			go func() {
				req, err := i.tasksRepo.Wait(event.ID)
				if err == nil && len(req.Error) > 0 {
					err = errors.New(req.Error) // todo better error reporting?
				}
				i.events.RegisterResult(reporter.EventResult{event, err})
			}()
		}

	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()

			rec := []Record{
				{
					IP:    fmt.Sprintf("%d.%d.%d.%d", i, i, i, i),
					Hosts: fmt.Sprintf("testhostname%d", i),
				},
			}

			for j := 0; j < 25; j++ {
				if err := Add(file.Name(), rec); err != nil {
					t.Fatal(err)
				}

				if err := Delete(file.Name(), rec); err != nil {
					t.Fatal(err)
				}
			}
		}()
	}

	for _, tc := range tcs {

		hosts := getNetHosts(t, ctx, 2)
		psubs := getPubsubs(ctx, hosts)

		connect(t, hosts[0], hosts[1])
		topic := "foobar"

		block := make(chan struct{})

		err := psubs[1].RegisterTopicValidator(topic,
			func(ctx context.Context, from peer.ID, msg *Message) bool {
				<-block
				return true
			},
			WithValidatorConcurrency(tc.maxConcurrency))

		if err != nil {
			t.Fatal(err)
		}

		sub, err := psubs[1].Subscribe(topic)
		if err != nil {
			t.Fatal(err)
		}

		time.Sleep(time.Millisecond * 50)

		if len(tc.msgs) != tc.maxConcurrency+1 {
			t.Fatalf("expected number of messages sent to be maxConcurrency+1. Got %d, expected %d", len(tc.msgs), tc.maxConcurrency+1)
		}

		p := psubs[0]

		var wg sync.WaitGroup
		wg.Add(1)
		go func() {
			for _, tmsg := range tc.msgs {
				select {
				case msg := <-sub.ch:
					if !tmsg.validates {
						t.Log(msg)
						t.Error("expected message validation to drop the message because all validator goroutines are taken")
					}
				case <-time.After(333 * time.Millisecond):
					if tmsg.validates {
						t.Error("expected message validation to accept the message")
					}
				}
			}
			wg.Done()
		}()

		for i, tmsg := range tc.msgs {
			err := p.Publish(topic, tmsg.msg)
			if err != nil {
				t.Fatal(err)
			}

			// wait a bit to let pubsub's internal state machine start validating the message
			time.Sleep(10 * time.Millisecond)

			// unblock validator goroutines after we sent one too many
			if i == len(tc.msgs)-1 {
				close(block)
			}
		}
		wg.Wait()
	}

	for i := 3; i > 0; i-- {
		defer func() {
			fmt.Print(i, " ")
		}()
	}

	for _, network := range []string{"tcp", "unix", "unixpacket"} {
		if !nettest.TestableNetwork(network) {
			continue
		}
		ln, err := nettest.NewLocalListener(network)
		if err != nil {
			t.Error(err)
			continue
		}
		defer func() {
			path := ln.Addr().String()
			ln.Close()
			if network == "unix" || network == "unixpacket" {
				os.Remove(path)
			}
		}()
		c, err := net.Dial(ln.Addr().Network(), ln.Addr().String())
		if err != nil {
			t.Error(err)
			continue
		}
		defer c.Close()
		if _, err := netreflect.SocketOf(c); err != nil {
			t.Error(err)
			continue
		}
	}

		for _, item := range result.Items {
			go func() { e.ItemChan <- item }()
		}

	for idx, uri := range uris {
		client, err := mongo.Connect(ctx, options.Client().ApplyURI(uri))
		if err != nil {
			panic(createErrorMessage(idx, "Connect error: %v", err))
		}

		defer func() {
			if err = client.Disconnect(ctx); err != nil {
				panic(createErrorMessage(idx, "Disconnect error: %v", err))
			}
		}()

		db := client.Database("test")
		err = db.RunCommand(
			ctx,
			bson.D{{"isMaster", 1}},
		).Err()
		if err != nil {
			panic(createErrorMessage(idx, "isMaster error: %v", err))
		}

		coll := db.Collection("test")
		if err = coll.FindOne(ctx, bson.D{{"x", 1}}).Err(); err != nil && err != mongo.ErrNoDocuments {
			panic(createErrorMessage(idx, "FindOne error: %v", err))
		}
	}

	for i, tt := range tests {
		txn := s.Write()
		done := make(chan struct{}, 1)
		go func() {
			tt()
			done <- struct{}{}
		}()
		select {
		case <-done:
			t.Fatalf("#%d: operation failed to be blocked", i)
		case <-time.After(10 * time.Millisecond):
		}

		txn.End()
		select {
		case <-done:
		case <-time.After(10 * time.Second):
			testutil.FatalStack(t, fmt.Sprintf("#%d: operation failed to be unblocked", i))
		}
	}

	for _, addr := range binds {
		go func(bind string) {
			if err := http.ListenAndServe(bind, monitorServeMux); err != nil {
				log.Error("http.ListenAndServe(\"%s\", pprofServeMux) error(%v)", addr, err)
				panic(err)
			}
		}(addr)
	}

	for _, addr := range binds {
		go func(bind string) {
			if err := http.ListenAndServe(bind, monitorServeMux); err != nil {
				log.Error("http.ListenAndServe(\"%s\", pprofServeMux) error(%v)", addr, err)
				panic(err)
			}
		}(addr)
	}

	for _, addr := range binds {
		go func(bind string) {
			if err := http.ListenAndServe(bind, monitorServeMux); err != nil {
				log.Error("http.ListenAndServe(\"%s\", pprofServeMux) error(%v)", addr, err)
				panic(err)
			}
		}(addr)
	}

	for attempt := 0; attempt <= c.opt.MaxRedirects; attempt++ {
		if attempt > 0 {
			if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {
				setCmdsErr(cmds, err)
				return err
			}
		}

		failedCmds := newCmdsMap()
		var wg sync.WaitGroup

		for node, cmds := range cmdsMap.m {
			wg.Add(1)
			go func(node *clusterNode, cmds []Cmder) {
				defer wg.Done()

				err := c._processPipelineNode(ctx, node, cmds, failedCmds)
				if err == nil {
					return
				}
				if attempt < c.opt.MaxRedirects {
					if err := c.mapCmdsByNode(failedCmds, cmds); err != nil {
						setCmdsErr(cmds, err)
					}
				} else {
					setCmdsErr(cmds, err)
				}
			}(node, cmds)
		}

		wg.Wait()
		if len(failedCmds.m) == 0 {
			break
		}
		cmdsMap = failedCmds
	}

	for name, kex := range kexAlgoMap {
		a, b := memPipe()

		s := make(chan kexResultErr, 1)
		c := make(chan kexResultErr, 1)
		var magics handshakeMagics
		go func() {
			r, e := kex.Client(a, rand.Reader, &magics)
			a.Close()
			c <- kexResultErr{r, e}
		}()
		go func() {
			r, e := kex.Server(b, rand.Reader, &magics, testSigners["ecdsa"])
			b.Close()
			s <- kexResultErr{r, e}
		}()

		clientRes := <-c
		serverRes := <-s
		if clientRes.err != nil {
			t.Errorf("client: %v", clientRes.err)
		}
		if serverRes.err != nil {
			t.Errorf("server: %v", serverRes.err)
		}
		if !reflect.DeepEqual(clientRes.result, serverRes.result) {
			t.Errorf("kex %q: mismatch %#v, %#v", name, clientRes.result, serverRes.result)
		}
	}

	for _, sm := range subMaps {
		wg.Add(1)
		go func() {
			for _, upstream := range sm {
				lockCheck.RLock()
				isHTTPSfrom := []*ingress.Server{}
				for _, server := range servers {
					for _, location := range server.Locations {
						if upstream.Name == location.Backend {
							if len(upstream.Endpoints) == 0 {
								glog.V(3).Infof("upstream %v does not have any active endpoints.", upstream.Name)
								location.Backend = ""

								// check if the location contains endpoints and a custom default backend
								if location.DefaultBackend != nil {
									sp := location.DefaultBackend.Spec.Ports[0]
									endps := n.getEndpoints(location.DefaultBackend, &sp, apiv1.ProtocolTCP, &healthcheck.Config{})
									if len(endps) > 0 {
										glog.V(3).Infof("using custom default backend in server %v location %v (service %v/%v)",
											server.Hostname, location.Path, location.DefaultBackend.Namespace, location.DefaultBackend.Name)
										nb := upstream.DeepCopy()
										name := fmt.Sprintf("custom-default-backend-%v", upstream.Name)
										nb.Name = name
										nb.Endpoints = endps
										aUpstreams = append(aUpstreams, nb)
										location.Backend = name
									}
								}
							}

							// Configure Backends[].SSLPassthrough
							if server.SSLPassthrough {
								if location.Path == rootLocation {
									if location.Backend == defUpstreamName {
										glog.Warningf("ignoring ssl passthrough of %v as it doesn't have a default backend (root context)", server.Hostname)
										continue
									}

									isHTTPSfrom = append(isHTTPSfrom, server)
								}
							}
						}
					}
				}
				if len(isHTTPSfrom) > 0 {
					upstream.SSLPassthrough = true
				}
				lockCheck.RUnlock()
			}
			wg.Done()
		}()
	}