Merge pull request #16 from ipfs/feat/fifo-equal-priority

When priority is equal, use FIFO

Author: hannahhoward

go.mod

@@ -3,6 +3,7 @@ module github.com/ipfs/go-peertaskqueue
 go 1.16
 
 require (
+	github.com/benbjohnson/clock v1.1.0
 	github.com/ipfs/go-ipfs-pq v0.0.2
 	github.com/libp2p/go-libp2p-core v0.0.1
 	github.com/multiformats/go-multihash v0.0.5 // indirect

go.sum

@@ -1,4 +1,6 @@
 github.com/aead/siphash v1.0.1/go.mod h1:Nywa3cDsYNNK3gaciGTWPwHt0wlpNV15vwmswBAUSII=
+github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
+github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/btcsuite/btcd v0.0.0-20190213025234-306aecffea32 h1:qkOC5Gd33k54tobS36cXdAzJbeHaduLtnLQQwNoIi78=
 github.com/btcsuite/btcd v0.0.0-20190213025234-306aecffea32/go.mod h1:DrZx5ec/dmnfpw9KyYoQyYo7d0KEvTkk/5M/vbZjAr8=
 github.com/btcsuite/btclog v0.0.0-20170628155309-84c8d2346e9f/go.mod h1:TdznJufoqS23FtqVCzL0ZqgP5MqXbb4fg/WgDys70nA=

peertask/peertask.go

@@ -15,7 +15,7 @@ var FIFOCompare = func(a, b *QueueTask) bool {
 // PriorityCompare respects the target peer's task priority. For tasks involving
 // different peers, the oldest task is prioritized.
 var PriorityCompare = func(a, b *QueueTask) bool {
-	if a.Target == b.Target {
+	if a.Target == b.Target && a.Priority != b.Priority {
 		return a.Priority > b.Priority
 	}
 	return FIFOCompare(a, b)

peertracker/peertracker.go

@@ -2,13 +2,15 @@ package peertracker
 
 import (
 	"sync"
-	"time"
 
+	"github.com/benbjohnson/clock"
 	pq "github.com/ipfs/go-ipfs-pq"
 	"github.com/ipfs/go-peertaskqueue/peertask"
 	peer "github.com/libp2p/go-libp2p-core/peer"
 )
 
+var clockInstance = clock.New()
+
 // TaskMerger is an interface that is used to merge new tasks into the active
 // and pending queues
 type TaskMerger interface {
@@ -144,7 +146,7 @@ func (p *PeerTracker) SetIndex(i int) {
 
 // PushTasks adds a group of tasks onto a peer's queue
 func (p *PeerTracker) PushTasks(tasks ...peertask.Task) {
-	now := time.Now()
+	now := clockInstance.Now()
 
 	p.activelk.Lock()
 	defer p.activelk.Unlock()

peertracker/peertracker_test.go

@@ -2,7 +2,9 @@ package peertracker
 
 import (
 	"testing"
+	"time"
 
+	"github.com/benbjohnson/clock"
 	"github.com/ipfs/go-peertaskqueue/peertask"
 	"github.com/ipfs/go-peertaskqueue/testutil"
 )
@@ -568,3 +570,60 @@ func TestRemoveActive(t *testing.T) {
 		t.Fatal("Expected tasks in order")
 	}
 }
+
+func TestPushPopEqualTaskPriorities(t *testing.T) {
+	partner := testutil.GeneratePeers(1)[0]
+	clock := clock.NewMock()
+	oldClock := clockInstance
+	clockInstance = clock
+	t.Cleanup(func() {
+		clockInstance = oldClock
+	})
+	tracker := New(partner, &DefaultTaskMerger{}, 1)
+
+	tasks := []peertask.Task{
+		{
+			Topic:    "1",
+			Priority: 10,
+			Work:     1,
+		},
+		{
+			Topic:    "2",
+			Priority: 10,
+			Work:     1,
+		},
+		{
+			Topic:    "3",
+			Priority: 10,
+			Work:     1,
+		},
+	}
+	tracker.PushTasks(tasks[0])
+	clock.Add(10 * time.Millisecond)
+	tracker.PushTasks(tasks[1])
+	clock.Add(10 * time.Millisecond)
+	tracker.PushTasks(tasks[2])
+	popped, _ := tracker.PopTasks(1)
+	if len(popped) != 1 {
+		t.Fatal("Expected 1 task")
+	}
+	if popped[0].Topic != "1" {
+		t.Fatal("Expected first task")
+	}
+	tracker.TaskDone(popped[0])
+	popped, _ = tracker.PopTasks(1)
+	if len(popped) != 1 {
+		t.Fatal("Expected 1 task")
+	}
+	if popped[0].Topic != "2" {
+		t.Fatal("Expected second task")
+	}
+	tracker.TaskDone(popped[0])
+	popped, _ = tracker.PopTasks(1)
+	if len(popped) != 1 {
+		t.Fatal("Expected 1 task")
+	}
+	if popped[0].Topic != "3" {
+		t.Fatal("Expected third task")
+	}
+}