When priority is equal, use FIFO

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)

peertaskqueue.go

@@ -3,6 +3,7 @@ package peertaskqueue
 import (
 	"sync"
 
+	"github.com/benbjohnson/clock"
 	pq "github.com/ipfs/go-ipfs-pq"
 	"github.com/ipfs/go-peertaskqueue/peertask"
 	"github.com/ipfs/go-peertaskqueue/peertracker"
@@ -171,7 +172,7 @@ func (ptq *PeerTaskQueue) PushTasks(to peer.ID, tasks ...peertask.Task) {
 
 	peerTracker, ok := ptq.peerTrackers[to]
 	if !ok {
-		peerTracker = peertracker.New(to, ptq.taskMerger, ptq.maxOutstandingWorkPerPeer)
+		peerTracker = peertracker.New(to, ptq.taskMerger, ptq.maxOutstandingWorkPerPeer, clock.New())
 		ptq.pQueue.Push(peerTracker)
 		ptq.peerTrackers[to] = peerTracker
 		ptq.callHooks(to, peerAdded)

peertracker/peertracker.go

@@ -2,8 +2,8 @@ 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"
@@ -36,6 +36,7 @@ func (*DefaultTaskMerger) Merge(task peertask.Task, existing *peertask.Task) {
 type PeerTracker struct {
 	target peer.ID
 
+	clock clock.Clock
 	// Tasks that are pending being made active
 	pendingTasks map[peertask.Topic]*peertask.QueueTask
 	// Tasks that have been made active
@@ -59,14 +60,15 @@ type PeerTracker struct {
 }
 
 // New creates a new PeerTracker
-func New(target peer.ID, taskMerger TaskMerger, maxActiveWorkPerPeer int) *PeerTracker {
+func New(target peer.ID, taskMerger TaskMerger, maxActiveWorkPerPeer int, clock clock.Clock) *PeerTracker {
 	return &PeerTracker{
 		target:               target,
 		taskQueue:            pq.New(peertask.WrapCompare(peertask.PriorityCompare)),
 		pendingTasks:         make(map[peertask.Topic]*peertask.QueueTask),
 		activeTasks:          make(map[*peertask.Task]struct{}),
 		taskMerger:           taskMerger,
 		maxActiveWorkPerPeer: maxActiveWorkPerPeer,
+		clock:                clock,
 	}
 }
 
@@ -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 := p.clock.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"
 )
@@ -11,7 +13,7 @@ const testMaxActiveWorkPerPeer = 100
 
 func TestEmpty(t *testing.T) {
 	partner := testutil.GeneratePeers(1)[0]
-	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer)
+	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer, clock.New())
 
 	tasks, _ := tracker.PopTasks(100)
 	if len(tasks) != 0 {
@@ -21,7 +23,7 @@ func TestEmpty(t *testing.T) {
 
 func TestPushPop(t *testing.T) {
 	partner := testutil.GeneratePeers(1)[0]
-	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer)
+	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer, clock.New())
 
 	tasks := []peertask.Task{
 		{
@@ -42,7 +44,7 @@ func TestPushPop(t *testing.T) {
 
 func TestPopNegativeOrZeroSize(t *testing.T) {
 	partner := testutil.GeneratePeers(1)[0]
-	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer)
+	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer, clock.New())
 
 	tasks := []peertask.Task{
 		{
@@ -64,7 +66,7 @@ func TestPopNegativeOrZeroSize(t *testing.T) {
 
 func TestPushPopSizeAndOrder(t *testing.T) {
 	partner := testutil.GeneratePeers(1)[0]
-	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer)
+	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer, clock.New())
 
 	tasks := []peertask.Task{
 		{
@@ -118,7 +120,7 @@ func TestPushPopSizeAndOrder(t *testing.T) {
 
 func TestPopFirstItemAlways(t *testing.T) {
 	partner := testutil.GeneratePeers(1)[0]
-	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer)
+	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer, clock.New())
 
 	tasks := []peertask.Task{
 		{
@@ -149,7 +151,7 @@ func TestPopFirstItemAlways(t *testing.T) {
 
 func TestPopItemsToCoverTargetWork(t *testing.T) {
 	partner := testutil.GeneratePeers(1)[0]
-	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer)
+	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer, clock.New())
 
 	tasks := []peertask.Task{
 		{
@@ -185,7 +187,7 @@ func TestPopItemsToCoverTargetWork(t *testing.T) {
 
 func TestRemove(t *testing.T) {
 	partner := testutil.GeneratePeers(1)[0]
-	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer)
+	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer, clock.New())
 
 	tasks := []peertask.Task{
 		{
@@ -217,7 +219,7 @@ func TestRemove(t *testing.T) {
 
 func TestRemoveMulti(t *testing.T) {
 	partner := testutil.GeneratePeers(1)[0]
-	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer)
+	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer, clock.New())
 
 	tasks := []peertask.Task{
 		{
@@ -249,7 +251,7 @@ func TestRemoveMulti(t *testing.T) {
 
 func TestTaskDone(t *testing.T) {
 	partner := testutil.GeneratePeers(1)[0]
-	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer)
+	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer, clock.New())
 
 	tasks := []peertask.Task{
 		{
@@ -301,7 +303,7 @@ func (*permissiveTaskMerger) Merge(task peertask.Task, existing *peertask.Task)
 
 func TestReplaceTaskPermissive(t *testing.T) {
 	partner := testutil.GeneratePeers(1)[0]
-	tracker := New(partner, &permissiveTaskMerger{}, testMaxActiveWorkPerPeer)
+	tracker := New(partner, &permissiveTaskMerger{}, testMaxActiveWorkPerPeer, clock.New())
 
 	tasks := []peertask.Task{
 		{
@@ -340,7 +342,7 @@ func TestReplaceTaskPermissive(t *testing.T) {
 
 func TestReplaceTaskSize(t *testing.T) {
 	partner := testutil.GeneratePeers(1)[0]
-	tracker := New(partner, &permissiveTaskMerger{}, testMaxActiveWorkPerPeer)
+	tracker := New(partner, &permissiveTaskMerger{}, testMaxActiveWorkPerPeer, clock.New())
 
 	tasks := []peertask.Task{
 		{
@@ -393,7 +395,7 @@ func TestReplaceTaskSize(t *testing.T) {
 
 func TestReplaceActiveTask(t *testing.T) {
 	partner := testutil.GeneratePeers(1)[0]
-	tracker := New(partner, &permissiveTaskMerger{}, testMaxActiveWorkPerPeer)
+	tracker := New(partner, &permissiveTaskMerger{}, testMaxActiveWorkPerPeer, clock.New())
 
 	tasks := []peertask.Task{
 		{
@@ -432,7 +434,7 @@ func TestReplaceActiveTask(t *testing.T) {
 
 func TestReplaceActiveTaskNonPermissive(t *testing.T) {
 	partner := testutil.GeneratePeers(1)[0]
-	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer)
+	tracker := New(partner, &DefaultTaskMerger{}, testMaxActiveWorkPerPeer, clock.New())
 
 	tasks := []peertask.Task{
 		{
@@ -470,7 +472,7 @@ func TestReplaceActiveTaskNonPermissive(t *testing.T) {
 
 func TestReplaceTaskThatIsActiveAndPending(t *testing.T) {
 	partner := testutil.GeneratePeers(1)[0]
-	tracker := New(partner, &permissiveTaskMerger{}, testMaxActiveWorkPerPeer)
+	tracker := New(partner, &permissiveTaskMerger{}, testMaxActiveWorkPerPeer, clock.New())
 
 	tasks := []peertask.Task{
 		{
@@ -521,7 +523,7 @@ func TestReplaceTaskThatIsActiveAndPending(t *testing.T) {
 
 func TestRemoveActive(t *testing.T) {
 	partner := testutil.GeneratePeers(1)[0]
-	tracker := New(partner, &permissiveTaskMerger{}, testMaxActiveWorkPerPeer)
+	tracker := New(partner, &permissiveTaskMerger{}, testMaxActiveWorkPerPeer, clock.New())
 
 	tasks := []peertask.Task{
 		{
@@ -568,3 +570,55 @@ func TestRemoveActive(t *testing.T) {
 		t.Fatal("Expected tasks in order")
 	}
 }
+
+func TestPushPopEqualTaskPriorities(t *testing.T) {
+	partner := testutil.GeneratePeers(1)[0]
+	clock := clock.NewMock()
+	tracker := New(partner, &DefaultTaskMerger{}, 1, clock)
+
+	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")
+	}
+}