<commit_before>package dht

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"

	peer "github.com/jbenet/go-ipfs/peer"
	queue "github.com/jbenet/go-ipfs/peer/queue"
	kb "github.com/jbenet/go-ipfs/routing/kbucket"
	u "github.com/jbenet/go-ipfs/util"
)

// This file implements the Routing interface for the IpfsDHT struct.

// Basic Put/Get

// PutValue adds value corresponding to given Key.
// This is the top level "Store" operation of the DHT
func (dht *IpfsDHT) PutValue(key u.Key, value []byte) error {
	ctx := context.TODO()

	query := &dhtQuery{}
	query.peers = queue.NewXORDistancePQ(key)

	// get the peers we need to announce to
	for _, route := range dht.routingTables {
		peers := route.NearestPeers(kb.ConvertKey(key), KValue)
		for _, p := range peers {
			if p == nil {
				// this shouldn't be happening.
				panic("p should not be nil")
			}

			query.peers.Enqueue(p)
		}
	}

	query.qfunc = func(ctx context.Context, p *peer.Peer) (interface{}, []*peer.Peer, error) {
		dht.putValueToNetwork(ctx, p, string(key), value)
		return nil, nil, nil
	}

	_, err := query.Run(ctx, query.peers.Len())
	return err
}

// GetValue searches for the value corresponding to given Key.
// If the search does not succeed, a multiaddr string of a closer peer is
// returned along with util.ErrSearchIncomplete
func (dht *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
	ll := startNewRPC("GET")
	defer ll.EndAndPrint()

	ctx, _ := context.WithTimeout(context.TODO(), timeout)

	// If we have it local, dont bother doing an RPC!
	// NOTE: this might not be what we want to do...
	val, err := dht.getLocal(key)
	if err == nil {
		ll.Success = true
		u.DOut("Found local, returning.\n")
		return val, nil
	}

	// get closest peers in the routing tables
	routeLevel := 0
	closest := dht.routingTables[routeLevel].NearestPeers(kb.ConvertKey(key), PoolSize)
	if closest == nil || len(closest) == 0 {
		return nil, kb.ErrLookupFailure
	}

	query := &dhtQuery{}
	query.peers = queue.NewXORDistancePQ(key)

	// get the peers we need to announce to
	for _, p := range closest {
		query.peers.Enqueue(p)
	}

	// setup the Query Function
	query.qfunc = func(ctx context.Context, p *peer.Peer) (interface{}, []*peer.Peer, error) {
		return dht.getValueOrPeers(ctx, p, key, routeLevel)
	}

	// run it!
	result, err := query.Run(ctx, query.peers.Len())
	if err != nil {
		return nil, err
	}

	byt, ok := result.([]byte)
	if !ok {
		return nil, fmt.Errorf("received non-byte slice value")
	}
	return byt, nil
}

// Value provider layer of indirection.
// This is what DSHTs (Coral and MainlineDHT) do to store large values in a DHT.

// Provide makes this node announce that it can provide a value for the given key
func (dht *IpfsDHT) Provide(key u.Key) error {
	ctx := context.TODO()

	dht.providers.AddProvider(key, dht.self)
	peers := dht.routingTables[0].NearestPeers(kb.ConvertKey(key), PoolSize)
	if len(peers) == 0 {
		return kb.ErrLookupFailure
	}

	for _, p := range peers {
		err := dht.putProvider(ctx, p, string(key))
		if err != nil {
			return err
		}
	}
	return nil
}

// FindProvidersAsync runs FindProviders and sends back results over a channel
func (dht *IpfsDHT) FindProvidersAsync(key u.Key, count int, timeout time.Duration) chan *peer.Peer {
	ctx, _ := context.WithTimeout(context.TODO(), timeout)

	peerOut := make(chan *peer.Peer, count)
	go func() {
		ps := newPeerSet()
		provs := dht.providers.GetProviders(key)
		for _, p := range provs {
			count--
			// NOTE: assuming that this list of peers is unique
			ps.Add(p)
			peerOut <- p
			if count <= 0 {
				return
			}
		}

		peers := dht.routingTables[0].NearestPeers(kb.ConvertKey(key), AlphaValue)
		for _, pp := range peers {
			ppp := pp
			go func() {
				pmes, err := dht.findProvidersSingle(ctx, ppp, key, 0)
				if err != nil {<commit_msg>fix(routing/dht) match the routing interface<commit_after>package dht

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"

	peer "github.com/jbenet/go-ipfs/peer"
	queue "github.com/jbenet/go-ipfs/peer/queue"
	kb "github.com/jbenet/go-ipfs/routing/kbucket"
	u "github.com/jbenet/go-ipfs/util"
)

// This file implements the Routing interface for the IpfsDHT struct.

// Basic Put/Get

// PutValue adds value corresponding to given Key.
// This is the top level "Store" operation of the DHT
func (dht *IpfsDHT) PutValue(key u.Key, value []byte) error {
	ctx := context.TODO()

	query := &dhtQuery{}
	query.peers = queue.NewXORDistancePQ(key)

	// get the peers we need to announce to
	for _, route := range dht.routingTables {
		peers := route.NearestPeers(kb.ConvertKey(key), KValue)
		for _, p := range peers {
			if p == nil {
				// this shouldn't be happening.
				panic("p should not be nil")
			}

			query.peers.Enqueue(p)
		}
	}

	query.qfunc = func(ctx context.Context, p *peer.Peer) (interface{}, []*peer.Peer, error) {
		dht.putValueToNetwork(ctx, p, string(key), value)
		return nil, nil, nil
	}

	_, err := query.Run(ctx, query.peers.Len())
	return err
}

// GetValue searches for the value corresponding to given Key.
// If the search does not succeed, a multiaddr string of a closer peer is
// returned along with util.ErrSearchIncomplete
func (dht *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
	ll := startNewRPC("GET")
	defer ll.EndAndPrint()

	ctx, _ := context.WithTimeout(context.TODO(), timeout)

	// If we have it local, dont bother doing an RPC!
	// NOTE: this might not be what we want to do...
	val, err := dht.getLocal(key)
	if err == nil {
		ll.Success = true
		u.DOut("Found local, returning.\n")
		return val, nil
	}

	// get closest peers in the routing tables
	routeLevel := 0
	closest := dht.routingTables[routeLevel].NearestPeers(kb.ConvertKey(key), PoolSize)
	if closest == nil || len(closest) == 0 {
		return nil, kb.ErrLookupFailure
	}

	query := &dhtQuery{}
	query.peers = queue.NewXORDistancePQ(key)

	// get the peers we need to announce to
	for _, p := range closest {
		query.peers.Enqueue(p)
	}

	// setup the Query Function
	query.qfunc = func(ctx context.Context, p *peer.Peer) (interface{}, []*peer.Peer, error) {
		return dht.getValueOrPeers(ctx, p, key, routeLevel)
	}

	// run it!
	result, err := query.Run(ctx, query.peers.Len())
	if err != nil {
		return nil, err
	}

	byt, ok := result.([]byte)
	if !ok {
		return nil, fmt.Errorf("received non-byte slice value")
	}
	return byt, nil
}

// Value provider layer of indirection.
// This is what DSHTs (Coral and MainlineDHT) do to store large values in a DHT.

// Provide makes this node announce that it can provide a value for the given key
func (dht *IpfsDHT) Provide(key u.Key) error {
	ctx := context.TODO()

	dht.providers.AddProvider(key, dht.self)
	peers := dht.routingTables[0].NearestPeers(kb.ConvertKey(key), PoolSize)
	if len(peers) == 0 {
		return kb.ErrLookupFailure
	}

	for _, p := range peers {
		err := dht.putProvider(ctx, p, string(key))
		if err != nil {
			return err
		}
	}
	return nil
}

// FindProvidersAsync runs FindProviders and sends back results over a channel
func (dht *IpfsDHT) FindProvidersAsync(key u.Key, count int, timeout time.Duration) <-chan *peer.Peer {
	ctx, _ := context.WithTimeout(context.TODO(), timeout)

	peerOut := make(chan *peer.Peer, count)
	go func() {
		ps := newPeerSet()
		provs := dht.providers.GetProviders(key)
		for _, p := range provs {
			count--
			// NOTE: assuming that this list of peers is unique
			ps.Add(p)
			peerOut <- p
			if count <= 0 {
				return
			}
		}

		peers := dht.routingTables[0].NearestPeers(kb.ConvertKey(key), AlphaValue)
		for _, pp := range peers {
			ppp := pp
			go func() {
				pmes, err := dht.findProvidersSingle(ctx, ppp, key, 0)
				if err != nil {
{'fraction_non_alphanumeric': 0.13497786211258697, 'fraction_numerical': 0.0017710309930423782, 'mean_word_length': 3.4844015882019286, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '1035498', 'n_tokens_mistral': 3007, 'n_tokens_neox': 2824, 'n_words': 832}