<commit_before>/*
Copyright IBM Corp. 2016 All Rights Reserved.

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 state

import (
	"bytes"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"

	pb "github.com/golang/protobuf/proto"
	"github.com/hyperledger/fabric/core/committer"
	"github.com/hyperledger/fabric/gossip/api"
	"github.com/hyperledger/fabric/gossip/comm"
	common2 "github.com/hyperledger/fabric/gossip/common"
	"github.com/hyperledger/fabric/gossip/gossip"
	"github.com/hyperledger/fabric/gossip/util"
	"github.com/hyperledger/fabric/protos/common"
	proto "github.com/hyperledger/fabric/protos/gossip"
	"github.com/op/go-logging"
)

// GossipStateProvider is the interface to acquire sequences of the ledger blocks
// capable to full fill missing blocks by running state replication and
// sending request to get missing block to other nodes
type GossipStateProvider interface {
	// Retrieve block with sequence number equal to index
	GetBlock(index uint64) *common.Block

	AddPayload(payload *proto.Payload) error

	// Stop terminates state transfer object
	Stop()
}

const (
	defPollingPeriod       = 200 * time.Millisecond
	defAntiEntropyInterval = 10 * time.Second
)

// GossipStateProviderImpl the implementation of the GossipStateProvider interface
// the struct to handle in memory sliding window of
// new ledger block to be acquired by hyper ledger
type GossipStateProviderImpl struct {
	// MessageCryptoService
	mcs api.MessageCryptoService

	// Chain id
	chainID string

	// The gossiping service
	gossip gossip.Gossip

	// Channel to read gossip messages from
	gossipChan <-chan *proto.GossipMessage

	commChan <-chan proto.ReceivedMessage

	// Flag which signals for termination
	stopFlag int32

	mutex sync.RWMutex

	// Queue of payloads which wasn't acquired yet
	payloads PayloadsBuffer

	committer committer.Committer

	logger *logging.Logger

	done sync.WaitGroup
}

// NewGossipStateProvider creates initialized instance of gossip state provider
func NewGossipStateProvider(chainID string, g gossip.Gossip, committer committer.Committer, mcs api.MessageCryptoService) GossipStateProvider {
	logger := util.GetLogger(util.LoggingStateModule, "")

	gossipChan, _ := g.Accept(func(message interface{}) bool {
		// Get only data messages
		return message.(*proto.GossipMessage).IsDataMsg() &&
			bytes.Equal(message.(*proto.GossipMessage).Channel, []byte(chainID))
	}, false)

	remoteStateMsgFilter := func(message interface{}) bool {
		receivedMsg := message.(proto.ReceivedMessage)
		msg := receivedMsg.GetGossipMessage()
		if !msg.IsRemoteStateMessage() {
			return false
		}
		// If we're not running with authentication, no point
		// in enforcing access control
		if !receivedMsg.GetConnectionInfo().IsAuthenticated() {
			return true
		}
		connInfo := receivedMsg.GetConnectionInfo()
		authErr := mcs.VerifyByChannel(msg.Channel, connInfo.Identity, connInfo.Auth.Signature, connInfo.Auth.SignedData)
		if authErr != nil {
			logger.Warning("Got unauthorized state transfer request from", string(connInfo.Identity))
			return false
		}
		return true
	}

	// Filter message which are only relevant for state transfer
	_, commChan := g.Accept(remoteStateMsgFilter, true)

	height, err := committer.LedgerHeight()

	if err != nil {
		logger.Error("Could not read ledger info to obtain current ledger height due to: ", err)
		// Exiting as without ledger it will be impossible
		// to deliver new blocks
		return nil
	}

	s := &GossipStateProviderImpl{
		// MessageCryptoService
		mcs: mcs,

		// Chain ID
		chainID: chainID,
<commit_msg>[FAB-2642] Avoid overflow of seqNum in peer metastate<commit_after>/*
Copyright IBM Corp. 2016 All Rights Reserved.

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 state

import (
	"bytes"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"

	pb "github.com/golang/protobuf/proto"
	"github.com/hyperledger/fabric/core/committer"
	"github.com/hyperledger/fabric/gossip/api"
	"github.com/hyperledger/fabric/gossip/comm"
	common2 "github.com/hyperledger/fabric/gossip/common"
	"github.com/hyperledger/fabric/gossip/gossip"
	"github.com/hyperledger/fabric/gossip/util"
	"github.com/hyperledger/fabric/protos/common"
	proto "github.com/hyperledger/fabric/protos/gossip"
	"github.com/op/go-logging"
)

// GossipStateProvider is the interface to acquire sequences of the ledger blocks
// capable to full fill missing blocks by running state replication and
// sending request to get missing block to other nodes
type GossipStateProvider interface {
	// Retrieve block with sequence number equal to index
	GetBlock(index uint64) *common.Block

	AddPayload(payload *proto.Payload) error

	// Stop terminates state transfer object
	Stop()
}

const (
	defPollingPeriod       = 200 * time.Millisecond
	defAntiEntropyInterval = 10 * time.Second
)

// GossipStateProviderImpl the implementation of the GossipStateProvider interface
// the struct to handle in memory sliding window of
// new ledger block to be acquired by hyper ledger
type GossipStateProviderImpl struct {
	// MessageCryptoService
	mcs api.MessageCryptoService

	// Chain id
	chainID string

	// The gossiping service
	gossip gossip.Gossip

	// Channel to read gossip messages from
	gossipChan <-chan *proto.GossipMessage

	commChan <-chan proto.ReceivedMessage

	// Flag which signals for termination
	stopFlag int32

	mutex sync.RWMutex

	// Queue of payloads which wasn't acquired yet
	payloads PayloadsBuffer

	committer committer.Committer

	logger *logging.Logger

	done sync.WaitGroup
}

// NewGossipStateProvider creates initialized instance of gossip state provider
func NewGossipStateProvider(chainID string, g gossip.Gossip, committer committer.Committer, mcs api.MessageCryptoService) GossipStateProvider {
	logger := util.GetLogger(util.LoggingStateModule, "")

	gossipChan, _ := g.Accept(func(message interface{}) bool {
		// Get only data messages
		return message.(*proto.GossipMessage).IsDataMsg() &&
			bytes.Equal(message.(*proto.GossipMessage).Channel, []byte(chainID))
	}, false)

	remoteStateMsgFilter := func(message interface{}) bool {
		receivedMsg := message.(proto.ReceivedMessage)
		msg := receivedMsg.GetGossipMessage()
		if !msg.IsRemoteStateMessage() {
			return false
		}
		// If we're not running with authentication, no point
		// in enforcing access control
		if !receivedMsg.GetConnectionInfo().IsAuthenticated() {
			return true
		}
		connInfo := receivedMsg.GetConnectionInfo()
		authErr := mcs.VerifyByChannel(msg.Channel, connInfo.Identity, connInfo.Auth.Signature, connInfo.Auth.SignedData)
		if authErr != nil {
			logger.Warning("Got unauthorized state transfer request from", string(connInfo.Identity))
			return false
		}
		return true
	}

	// Filter message which are only relevant for state transfer
	_, commChan := g.Accept(remoteStateMsgFilter, true)

	height, err := committer.LedgerHeight()
	if height == 0 {
		// Panic here since this is an indication of invalid situation which should not happen in normal
		// code path.
		logger.Panic("Committer height cannot be zero, ledger should include at least one block (genesis).")
	}

	if err != nil {
		logger.Error("Could not read ledger info to obtain current ledger height due to: ", err)
		// Exiting as without ledger it will be impossible
		// to deliver new blocks
		return nil
	}

	s := &GossipStateProviderImpl{
		// MessageCryptoService
		mcs: mcs,

		// Chain ID
		chainID: chainID,

{'fraction_non_alphanumeric': 0.08684744952813284, 'fraction_numerical': 0.004897861665272966, 'mean_word_length': 4.335882727852135, 'pattern_counts': {'":': 0, '<': 7, '<?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': '6304934', 'n_tokens_mistral': 2612, 'n_tokens_neox': 2398, 'n_words': 802}