<commit_before>// Copyright 2016 PingCAP, Inc.
//
// 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,
// See the License for the specific language governing permissions and
// limitations under the License.

package pd

import (
	"bufio"
	"net"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/deadline"
	"github.com/ngaut/log"
	"github.com/pingcap/kvproto/pkg/msgpb"
	"github.com/pingcap/kvproto/pkg/pdpb"
	"github.com/pingcap/kvproto/pkg/util"
	"github.com/pingcap/pd/pkg/metrics"
	"github.com/twinj/uuid"
)

const (
	pdRPCPrefix      = "/pd/rpc"
	connectPDTimeout = time.Second * 3
	netIOTimeout     = time.Second
)

const (
	readBufferSize  = 8 * 1024
	writeBufferSize = 8 * 1024
)

const maxPipelineRequest = 10000

// errInvalidResponse represents response message is invalid.
var errInvalidResponse = errors.New("invalid response")

type tsoRequest struct {
	done     chan error
	physical int64
	logical  int64
}

type storeRequest struct {
	pbReq  *pdpb.GetStoreRequest
	done   chan error
	pbResp *pdpb.GetStoreResponse
}

type regionRequest struct {
	pbReq  *pdpb.GetRegionRequest
	done   chan error
	pbResp *pdpb.GetRegionResponse
}

type clusterConfigRequest struct {
	pbReq  *pdpb.GetClusterConfigRequest
	done   chan error
	pbResp *pdpb.GetClusterConfigResponse
}

type rpcWorker struct {
	addr      string
	clusterID uint64
	requests  chan interface{}
	wg        sync.WaitGroup
	quit      chan struct{}
}

func newRPCWorker(addr string, clusterID uint64) *rpcWorker {
	w := &rpcWorker{
		addr:      addr,
		clusterID: clusterID,
		requests:  make(chan interface{}, maxPipelineRequest),
		quit:      make(chan struct{}),
	}
	w.wg.Add(1)
	go w.work()
	return w
}

func (w *rpcWorker) stop(err error) {
	close(w.quit)
	w.wg.Wait()

	for i := 0; i < len(w.requests); i++ {
		req := <-w.requests
		switch r := req.(type) {
		case *tsoRequest:
			r.done <- err
		case *clusterConfigRequest:
			r.done <- err
		case *storeRequest:
			r.done <- err
		case *regionRequest:
			r.done <- err
		}
	}
}

func (w *rpcWorker) work() {
	defer w.wg.Done()

RECONNECT:
	log.Infof("[pd] connect to pd server %v", w.addr)
	conn, err := rpcConnect(w.addr)
	if err != nil {
		log.Warnf("[pd] failed connect pd server: %v, will retry later", err)

		select {
		case <-time.After(netIOTimeout):
			goto RECONNECT
		case <-w.quit:<commit_msg>pd-client: fix a bug that hangs requests forever. (#323)<commit_after>// Copyright 2016 PingCAP, Inc.
//
// 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,
// See the License for the specific language governing permissions and
// limitations under the License.

package pd

import (
	"bufio"
	"net"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/deadline"
	"github.com/ngaut/log"
	"github.com/pingcap/kvproto/pkg/msgpb"
	"github.com/pingcap/kvproto/pkg/pdpb"
	"github.com/pingcap/kvproto/pkg/util"
	"github.com/pingcap/pd/pkg/metrics"
	"github.com/twinj/uuid"
)

const (
	pdRPCPrefix      = "/pd/rpc"
	connectPDTimeout = time.Second * 3
	netIOTimeout     = time.Second
)

const (
	readBufferSize  = 8 * 1024
	writeBufferSize = 8 * 1024
)

const maxPipelineRequest = 10000

// errInvalidResponse represents response message is invalid.
var errInvalidResponse = errors.New("invalid response")

type tsoRequest struct {
	done     chan error
	physical int64
	logical  int64
}

type storeRequest struct {
	pbReq  *pdpb.GetStoreRequest
	done   chan error
	pbResp *pdpb.GetStoreResponse
}

type regionRequest struct {
	pbReq  *pdpb.GetRegionRequest
	done   chan error
	pbResp *pdpb.GetRegionResponse
}

type clusterConfigRequest struct {
	pbReq  *pdpb.GetClusterConfigRequest
	done   chan error
	pbResp *pdpb.GetClusterConfigResponse
}

type rpcWorker struct {
	addr      string
	clusterID uint64
	requests  chan interface{}
	wg        sync.WaitGroup
	quit      chan struct{}
}

func newRPCWorker(addr string, clusterID uint64) *rpcWorker {
	w := &rpcWorker{
		addr:      addr,
		clusterID: clusterID,
		requests:  make(chan interface{}, maxPipelineRequest),
		quit:      make(chan struct{}),
	}
	w.wg.Add(1)
	go w.work()
	return w
}

func (w *rpcWorker) stop(err error) {
	close(w.quit)
	w.wg.Wait()

	n := len(w.requests)
	for i := 0; i < n; i++ {
		req := <-w.requests
		switch r := req.(type) {
		case *tsoRequest:
			r.done <- err
		case *clusterConfigRequest:
			r.done <- err
		case *storeRequest:
			r.done <- err
		case *regionRequest:
			r.done <- err
		}
	}
}

func (w *rpcWorker) work() {
	defer w.wg.Done()

RECONNECT:
	log.Infof("[pd] connect to pd server %v", w.addr)
	conn, err := rpcConnect(w.addr)
	if err != nil {
		log.Warnf("[pd] failed connect pd server: %v, will retry later", err)

		select {
		case <-time.After(netIOTimeout):
			goto RECONNECT
		case <-w.quit:
{'fraction_non_alphanumeric': 0.11822660098522167, 'fraction_numerical': 0.012953840540047437, 'mean_word_length': 3.4102976669348353, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6308814', 'n_tokens_mistral': 2079, 'n_tokens_neox': 1931, 'n_words': 547}