<commit_before>// Copyright 2014 The Cockroach 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.
//
// Author: <NAME> (<EMAIL>)

package gossip

import (
	"net"
	"sync"

	"google.golang.org/grpc"

	"github.com/cockroachdb/cockroach/roachpb"
	"github.com/cockroachdb/cockroach/util"
	"github.com/cockroachdb/cockroach/util/grpcutil"
	"github.com/cockroachdb/cockroach/util/log"
	"github.com/cockroachdb/cockroach/util/stop"
)

// server maintains an array of connected peers to which it gossips
// newly arrived information on a periodic basis.
type server struct {
	stopper *stop.Stopper

	mu       sync.Mutex                             // Protects the fields below
	is       *infoStore                             // The backing infostore
	incoming nodeSet                                // Incoming client node IDs
	nodeMap  map[util.UnresolvedAddr]roachpb.NodeID // Incoming client's local address -> node ID
	tighten  chan roachpb.NodeID                    // Channel of too-distant node IDs
	sent     int                                    // Count of infos sent from this server to clients
	received int                                    // Count of infos received from clients
	ready    *sync.Cond                             // Broadcasts wakeup to waiting gossip requests

	simulationCycler *sync.Cond // Used when simulating the network to signal next cycle
}

// newServer creates and returns a server struct.
func newServer(stopper *stop.Stopper) *server {
	s := &server{
		stopper:  stopper,
		is:       newInfoStore(0, util.UnresolvedAddr{}, stopper),
		incoming: makeNodeSet(minPeers),
		nodeMap:  make(map[util.UnresolvedAddr]roachpb.NodeID),
		tighten:  make(chan roachpb.NodeID, 1),
	}
	s.ready = sync.NewCond(&s.mu)
	return s
}

func (s *server) gossipSender(argsPtr **Request, senderFn func(*Response) error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	reply := new(Response)

	for {
		if !s.stopper.RunTask(func() {
			args := *argsPtr
			delta := s.is.delta(args.Nodes)
			if infoCount := len(delta); infoCount > 0 {
				if log.V(1) {
					log.Infof("node %d returned %d info(s) to node %d", s.is.NodeID, infoCount, args.NodeID)
				}

				*reply = Response{
					NodeID: s.is.NodeID,
					Nodes:  s.is.getNodes(),
					Delta:  delta,
				}

				s.mu.Unlock()
				err := senderFn(reply)
				s.mu.Lock()
				if err != nil {
					if !grpcutil.IsClosedConnection(err) {
						log.Error(err)
					}
					return
				}
				s.sent += infoCount
			}
		}) {
			return
		}

		s.ready.Wait()
	}
}

// Gossip receives gossiped information from a peer node.
// The received delta is combined with the infostore, and this
// node's own gossip is returned to requesting client.
func (s *server) Gossip(stream Gossip_GossipServer) error {
	args, err := stream.Recv()
	if err != nil {
		return err
	}

	ctx := stream.Context()
	syncChan := make(chan struct{}, 1)
	send := func(reply *Response) error {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case syncChan <- struct{}{}:
			defer func() { <-syncChan }()
			return stream.Send(reply)
		}
	}

	defer func() { syncChan <- struct{}{} }()

	s.mu.Lock()
	defer s.mu.Unlock()

	// Verify that there aren't multiple incoming connections from the same
	// node. This can happen when bootstrap connections are initiated through
	// a load balancer.
	if _, ok := s.nodeMap[args.Addr]; ok {
		return util.Errorf("duplicate connection from node at %s", args.Addr)
	}

	s.stopper.RunWorker(func() {
		s.gossipSender(&args, send)
	})

	reply := new(Response)

	// This loop receives gossip from the client. It does not attempt to send the
	// server's gossip to the client.
	for {
		if args.NodeID != 0 {
			// Decide whether or not we can accept the incoming connection
			// as a permanent peer.
			if s.incoming.hasNode(args.NodeID) {
				// Do nothing.
			} else if s.incoming.hasSpace() {
				s.incoming.addNode(args.NodeID)
				s.nodeMap[args.Addr] = args.NodeID

				defer func(nodeID roachpb.NodeID, addr util.UnresolvedAddr) {<commit_msg>gossip: fix incoming-request-during-shutdown race<commit_after>// Copyright 2014 The Cockroach 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.
//
// Author: <NAME> (<EMAIL>)

package gossip

import (
	"net"
	"sync"

	"google.golang.org/grpc"

	"github.com/cockroachdb/cockroach/roachpb"
	"github.com/cockroachdb/cockroach/util"
	"github.com/cockroachdb/cockroach/util/grpcutil"
	"github.com/cockroachdb/cockroach/util/log"
	"github.com/cockroachdb/cockroach/util/stop"
)

// server maintains an array of connected peers to which it gossips
// newly arrived information on a periodic basis.
type server struct {
	stopper *stop.Stopper

	mu       sync.Mutex                             // Protects the fields below
	is       *infoStore                             // The backing infostore
	incoming nodeSet                                // Incoming client node IDs
	nodeMap  map[util.UnresolvedAddr]roachpb.NodeID // Incoming client's local address -> node ID
	tighten  chan roachpb.NodeID                    // Channel of too-distant node IDs
	sent     int                                    // Count of infos sent from this server to clients
	received int                                    // Count of infos received from clients
	ready    *sync.Cond                             // Broadcasts wakeup to waiting gossip requests

	simulationCycler *sync.Cond // Used when simulating the network to signal next cycle
}

// newServer creates and returns a server struct.
func newServer(stopper *stop.Stopper) *server {
	s := &server{
		stopper:  stopper,
		is:       newInfoStore(0, util.UnresolvedAddr{}, stopper),
		incoming: makeNodeSet(minPeers),
		nodeMap:  make(map[util.UnresolvedAddr]roachpb.NodeID),
		tighten:  make(chan roachpb.NodeID, 1),
	}
	s.ready = sync.NewCond(&s.mu)
	return s
}

func (s *server) gossipSender(argsPtr **Request, senderFn func(*Response) error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	reply := new(Response)

	for {
		if !s.stopper.RunTask(func() {
			args := *argsPtr
			delta := s.is.delta(args.Nodes)
			if infoCount := len(delta); infoCount > 0 {
				if log.V(1) {
					log.Infof("node %d returned %d info(s) to node %d", s.is.NodeID, infoCount, args.NodeID)
				}

				*reply = Response{
					NodeID: s.is.NodeID,
					Nodes:  s.is.getNodes(),
					Delta:  delta,
				}

				s.mu.Unlock()
				err := senderFn(reply)
				s.mu.Lock()
				if err != nil {
					if !grpcutil.IsClosedConnection(err) {
						log.Error(err)
					}
					return
				}
				s.sent += infoCount
			}
		}) {
			return
		}

		s.ready.Wait()
	}
}

// Gossip receives gossiped information from a peer node.
// The received delta is combined with the infostore, and this
// node's own gossip is returned to requesting client.
func (s *server) Gossip(stream Gossip_GossipServer) error {
	args, err := stream.Recv()
	if err != nil {
		return err
	}

	ctx := stream.Context()
	syncChan := make(chan struct{}, 1)
	send := func(reply *Response) error {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case syncChan <- struct{}{}:
			defer func() { <-syncChan }()
			return stream.Send(reply)
		}
	}

	defer func() { syncChan <- struct{}{} }()

	s.mu.Lock()
	defer s.mu.Unlock()

	// Verify that there aren't multiple incoming connections from the same
	// node. This can happen when bootstrap connections are initiated through
	// a load balancer.
	if _, ok := s.nodeMap[args.Addr]; ok {
		return util.Errorf("duplicate connection from node at %s", args.Addr)
	}

	// Starting the worker in a task prevents data races during shutdown.
	if !s.stopper.RunTask(func() {
		s.stopper.RunWorker(func() {
			s.gossipSender(&args, send)
		})
	}) {
		return nil
	}

	reply := new(Response)

	// This loop receives gossip from the client. It does not attempt to send the
	// server's gossip to the client.
	for {
		if args.NodeID != 0 {
			// Decide whether or not we can accept the incoming connection
			// as a permanent peer.
			if s.incoming.hasNode(args.NodeID) {
				// Do nothing.
			} else if s.incoming.hasSpace() {
				s.incoming.addNode(args.NodeID)
				s.nodeMap[args.Addr] = args.NodeID

				defer func(nodeID roachpb.NodeID, addr util.UnresolvedAddr) {
{'fraction_non_alphanumeric': 0.11379460400348129, 'fraction_numerical': 0.0030461270670147957, 'mean_word_length': 2.6698602794411177, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1151457', 'n_tokens_mistral': 3232, 'n_tokens_neox': 2871, 'n_words': 910}