deduplicate blocks in queue

worker/worker.go

@@ -140,15 +140,30 @@ func (w *Worker) start(c Config) {
 }
 
 type BlockList struct {
-	list list.List
+	list    list.List
+	uniques map[util.Key]*list.Element
 }
 
 func (s *BlockList) PushFront(b *blocks.Block) {
-	s.list.PushFront(b)
+	if s.uniques == nil {
+		s.uniques = make(map[util.Key]*list.Element)
+	}
+	_, ok := s.uniques[b.Key()]
+	if !ok {
+		e := s.list.PushFront(b)
+		s.uniques[b.Key()] = e
+	}
 }
 
 func (s *BlockList) Push(b *blocks.Block) {
-	s.list.PushBack(b)
+	if s.uniques == nil {
+		s.uniques = make(map[util.Key]*list.Element)
+	}
+	_, ok := s.uniques[b.Key()]
+	if !ok {
+		e := s.list.PushBack(b)
+		s.uniques[b.Key()] = e
+	}
 }
 
 func (s *BlockList) Pop() *blocks.Block {
@@ -157,7 +172,9 @@ func (s *BlockList) Pop() *blocks.Block {
 	}
 	e := s.list.Front()
 	s.list.Remove(e)
-	return e.Value.(*blocks.Block)
+	b := e.Value.(*blocks.Block)
+	delete(s.uniques, b.Key())
+	return b
 }
 
 func (s *BlockList) Len() int {

worker/worker_test.go

 package worker
 
-import "testing"
+import (
+	blocks "github.com/jbenet/go-ipfs/blocks"
+	"testing"
+)
 
 func TestStartClose(t *testing.T) {
 	numRuns := 50
@@ -12,3 +15,49 @@ func TestStartClose(t *testing.T) {
 		w.Close()
 	}
 }
+
+func TestQueueDeduplication(t *testing.T) {
+	numUniqBlocks := 5 // arbitrary
+
+	var firstBatch []*blocks.Block
+	for i := 0; i < numUniqBlocks; i++ {
+		firstBatch = append(firstBatch, blockFromInt(i))
+	}
+
+	// to get different pointer values and prevent the implementation from
+	// cheating. The impl must check equality using Key.
+	var secondBatch []*blocks.Block
+	for i := 0; i < numUniqBlocks; i++ {
+		secondBatch = append(secondBatch, blockFromInt(i))
+	}
+	var workQueue BlockList
+
+	for _, b := range append(firstBatch, secondBatch...) {
+		workQueue.Push(b)
+	}
+	for i := 0; i < numUniqBlocks; i++ {
+		b := workQueue.Pop()
+		if b.Key() != firstBatch[i].Key() {
+			t.Fatal("list is not FIFO")
+		}
+	}
+	if b := workQueue.Pop(); b != nil {
+		t.Fatal("the workQueue did not de-duplicate the blocks")
+	}
+}
+
+func TestPushPopPushPop(t *testing.T) {
+	var workQueue BlockList
+	orig := blockFromInt(1)
+	dup := blockFromInt(1)
+	workQueue.PushFront(orig)
+	workQueue.Pop()
+	workQueue.Push(dup)
+	if workQueue.Len() != 1 {
+		t.Fatal("the block list's internal state is corrupt")
+	}
+}
+
+func blockFromInt(i int) *blocks.Block {
+	return blocks.NewBlock([]byte(string(i)))
+}