move to p2p dir

queue/distance.go

@@ -0,0 +1,101 @@
+package queue
+
+import (
+	"container/heap"
+	"math/big"
+	"sync"
+
+	key "github.com/ipfs/go-ipfs/blocks/key"
+	peer "github.com/ipfs/go-ipfs/p2p/peer"
+	ks "github.com/ipfs/go-ipfs/routing/keyspace"
+)
+
+// peerMetric tracks a peer and its distance to something else.
+type peerMetric struct {
+	// the peer
+	peer peer.ID
+
+	// big.Int for XOR metric
+	metric *big.Int
+}
+
+// peerMetricHeap implements a heap of peerDistances
+type peerMetricHeap []*peerMetric
+
+func (ph peerMetricHeap) Len() int {
+	return len(ph)
+}
+
+func (ph peerMetricHeap) Less(i, j int) bool {
+	return -1 == ph[i].metric.Cmp(ph[j].metric)
+}
+
+func (ph peerMetricHeap) Swap(i, j int) {
+	ph[i], ph[j] = ph[j], ph[i]
+}
+
+func (ph *peerMetricHeap) Push(x interface{}) {
+	item := x.(*peerMetric)
+	*ph = append(*ph, item)
+}
+
+func (ph *peerMetricHeap) Pop() interface{} {
+	old := *ph
+	n := len(old)
+	item := old[n-1]
+	*ph = old[0 : n-1]
+	return item
+}
+
+// distancePQ implements heap.Interface and PeerQueue
+type distancePQ struct {
+	// from is the Key this PQ measures against
+	from ks.Key
+
+	// heap is a heap of peerDistance items
+	heap peerMetricHeap
+
+	sync.RWMutex
+}
+
+func (pq *distancePQ) Len() int {
+	pq.Lock()
+	defer pq.Unlock()
+	return len(pq.heap)
+}
+
+func (pq *distancePQ) Enqueue(p peer.ID) {
+	pq.Lock()
+	defer pq.Unlock()
+
+	distance := ks.XORKeySpace.Key([]byte(p)).Distance(pq.from)
+
+	heap.Push(&pq.heap, &peerMetric{
+		peer:   p,
+		metric: distance,
+	})
+}
+
+func (pq *distancePQ) Dequeue() peer.ID {
+	pq.Lock()
+	defer pq.Unlock()
+
+	if len(pq.heap) < 1 {
+		panic("called Dequeue on an empty PeerQueue")
+		// will panic internally anyway, but we can help debug here
+	}
+
+	o := heap.Pop(&pq.heap)
+	p := o.(*peerMetric)
+	return p.peer
+}
+
+// NewXORDistancePQ returns a PeerQueue which maintains its peers sorted
+// in terms of their distances to each other in an XORKeySpace (i.e. using
+// XOR as a metric of distance).
+func NewXORDistancePQ(fromKey key.Key) PeerQueue {
+	return &distancePQ{
+		from: ks.XORKeySpace.Key([]byte(fromKey)),
+		heap: peerMetricHeap{},
+	}
+}

queue/interface.go

@@ -0,0 +1,18 @@
+package queue
+
+import peer "github.com/ipfs/go-ipfs/p2p/peer"
+
+// PeerQueue maintains a set of peers ordered according to a metric.
+// Implementations of PeerQueue could order peers based on distances along
+// a KeySpace, latency measurements, trustworthiness, reputation, etc.
+type PeerQueue interface {
+
+	// Len returns the number of items in PeerQueue
+	Len() int
+
+	// Enqueue adds this node to the queue.
+	Enqueue(peer.ID)
+
+	// Dequeue retrieves the highest (smallest int) priority node
+	Dequeue() peer.ID
+}

queue/queue_test.go

@@ -0,0 +1,141 @@
+package queue
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	key "github.com/ipfs/go-ipfs/blocks/key"
+	peer "github.com/ipfs/go-ipfs/p2p/peer"
+	u "github.com/ipfs/go-ipfs/util"
+
+	context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
+)
+
+func TestQueue(t *testing.T) {
+
+	p1 := peer.ID("11140beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a31") // these aren't valid, because need to hex-decode.
+	p2 := peer.ID("11140beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a32") // these aren't valid, because need to hex-decode.
+	p3 := peer.ID("11140beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a33") // these aren't valid, because need to hex-decode.
+	p4 := peer.ID("11140beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a34") // these aren't valid, because need to hex-decode.
+	p5 := peer.ID("11140beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a31") // these aren't valid, because need to hex-decode.
+	// but they work.
+
+	// these are the peer.IDs' XORKeySpace Key values:
+	// [228 47 151 130 156 102 222 232 218 31 132 94 170 208 80 253 120 103 55 35 91 237 48 157 81 245 57 247 66 150 9 40]
+	// [26 249 85 75 54 49 25 30 21 86 117 62 85 145 48 175 155 194 210 216 58 14 241 143 28 209 129 144 122 28 163 6]
+	// [78 135 26 216 178 181 224 181 234 117 2 248 152 115 255 103 244 34 4 152 193 88 9 225 8 127 216 158 226 8 236 246]
+	// [125 135 124 6 226 160 101 94 192 57 39 12 18 79 121 140 190 154 147 55 44 83 101 151 63 255 94 179 51 203 241 51]
+
+	pq := NewXORDistancePQ(key.Key("11140beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a31"))
+	pq.Enqueue(p3)
+	pq.Enqueue(p1)
+	pq.Enqueue(p2)
+	pq.Enqueue(p4)
+	pq.Enqueue(p5)
+	pq.Enqueue(p1)
+
+	// should come out as: p1, p4, p3, p2
+
+	if d := pq.Dequeue(); d != p1 && d != p5 {
+		t.Error("ordering failed")
+	}
+
+	if d := pq.Dequeue(); d != p1 && d != p5 {
+		t.Error("ordering failed")
+	}
+
+	if d := pq.Dequeue(); d != p1 && d != p5 {
+		t.Error("ordering failed")
+	}
+
+	if pq.Dequeue() != p4 {
+		t.Error("ordering failed")
+	}
+
+	if pq.Dequeue() != p3 {
+		t.Error("ordering failed")
+	}
+
+	if pq.Dequeue() != p2 {
+		t.Error("ordering failed")
+	}
+
+}
+
+func newPeerTime(t time.Time) peer.ID {
+	s := fmt.Sprintf("hmmm time: %v", t)
+	h := u.Hash([]byte(s))
+	return peer.ID(h)
+}
+
+func TestSyncQueue(t *testing.T) {
+	tickT := time.Microsecond * 50
+	max := 5000
+	consumerN := 10
+	countsIn := make([]int, consumerN*2)
+	countsOut := make([]int, consumerN)
+
+	if testing.Short() {
+		max = 1000
+	}
+
+	ctx := context.Background()
+	pq := NewXORDistancePQ(key.Key("11140beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a31"))
+	cq := NewChanQueue(ctx, pq)
+	wg := sync.WaitGroup{}
+
+	produce := func(p int) {
+		defer wg.Done()
+
+		tick := time.Tick(tickT)
+		for i := 0; i < max; i++ {
+			select {
+			case tim := <-tick:
+				countsIn[p]++
+				cq.EnqChan <- newPeerTime(tim)
+			case <-ctx.Done():
+				return
+			}
+		}
+	}
+
+	consume := func(c int) {
+		defer wg.Done()
+
+		for {
+			select {
+			case <-cq.DeqChan:
+				countsOut[c]++
+				if countsOut[c] >= max*2 {
+					return
+				}
+			case <-ctx.Done():
+				return
+			}
+		}
+	}
+
+	// make n * 2 producers and n consumers
+	for i := 0; i < consumerN; i++ {
+		wg.Add(3)
+		go produce(i)
+		go produce(consumerN + i)
+		go consume(i)
+	}
+
+	wg.Wait()
+
+	sum := func(ns []int) int {
+		total := 0
+		for _, n := range ns {
+			total += n
+		}
+		return total
+	}
+
+	if sum(countsIn) != sum(countsOut) {
+		t.Errorf("didnt get all of them out: %d/%d", sum(countsOut), sum(countsIn))
+	}
+}

queue/sync.go

@@ -0,0 +1,84 @@
+package queue
+
+import (
+	context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
+	peer "github.com/ipfs/go-ipfs/p2p/peer"
+	logging "github.com/ipfs/go-ipfs/vendor/go-log-v1.0.0"
+)
+
+var log = logging.Logger("peerqueue")
+
+// ChanQueue makes any PeerQueue synchronizable through channels.
+type ChanQueue struct {
+	Queue   PeerQueue
+	EnqChan chan<- peer.ID
+	DeqChan <-chan peer.ID
+}
+
+// NewChanQueue creates a ChanQueue by wrapping pq.
+func NewChanQueue(ctx context.Context, pq PeerQueue) *ChanQueue {
+	cq := &ChanQueue{Queue: pq}
+	cq.process(ctx)
+	return cq
+}
+
+func (cq *ChanQueue) process(ctx context.Context) {
+	// construct the channels here to be able to use them bidirectionally
+	enqChan := make(chan peer.ID)
+	deqChan := make(chan peer.ID)
+
+	cq.EnqChan = enqChan
+	cq.DeqChan = deqChan
+
+	go func() {
+		log.Debug("processing")
+		defer log.Debug("closed")
+		defer close(deqChan)
+
+		var next peer.ID
+		var item peer.ID
+		var more bool
+
+		for {
+			if cq.Queue.Len() == 0 {
+				// log.Debug("wait for enqueue")
+				select {
+				case next, more = <-enqChan:
+					if !more {
+						return
+					}
+					// log.Debug("got", next)
+
+				case <-ctx.Done():
+					return
+				}
+
+			} else {
+				next = cq.Queue.Dequeue()
+				// log.Debug("peek", next)
+			}
+
+			select {
+			case item, more = <-enqChan:
+				if !more {
+					if cq.Queue.Len() > 0 {
+						return // we're done done.
+					}
+					enqChan = nil // closed, so no use.
+				}
+				// log.Debug("got", item)
+				cq.Queue.Enqueue(item)
+				cq.Queue.Enqueue(next) // order may have changed.
+				next = ""
+
+			case deqChan <- next:
+				// log.Debug("dequeued", next)
+				next = ""
+
+			case <-ctx.Done():
+				return
+			}
+		}
+
+	}()
+}