priorityq

README

Thread-Safe Lock Free Priority Queue

Background

This project is based on Dr. Chris Okasaki's paper, Optimal Purely Functional Priority Queues. You can also read my own blog for background, motivation, and analysis of the repository. The underlying fundamental data structure in this repository is the Bootstrapped Skew Binomial Queue, which is outlined in Dr. Okasaki's paper. At one abstraction level higher, a separate data structure, concisely labeled as a "Parallel Queue", exists that enables thread-safe (and lock free!) concurrent dequeue operations at the cost of sacrificing absolute correctness.

The motivation behind the thread-safe version of the datastructure stemmed from a generally pragmatic use case of priority queues in which:

• thread-safe concurrent access is desirable
• returning the absolute highest priority item in a saturated environment is not necessarily required

It's also worth noting that a skip list would provide a generally comparable implementation of a priority queue, but enqueue() and dequeue() are swapped in terms of run-time complexity between O(1) and O(log N). However, the priority queues implemented here provide the distinct advantage that meld() is supported as a constant time operation, making it ideal for any environment where separate priority queues are routinely merged.

API

For the TLDR, check out public.go to understand the access and instantiation patterns of the data structures provided.

General Time Complexity

Enqueue: O(1)

Peek: O(1)

Dequeue: O(log N)

Length: O(1)

Meld: O(1)

Usage:

type ArbitraryDataType struct {
    ArbitraryData string
    score         int
}

func (a ArbitraryDataType) Score() int {
  return a.score
}

arbitraryScore := 100
q1 := priorityQ.NewImmutableSynchronousQ()

/**********  ENQUEUE  **********/
q1 = q1.Enqueue(
    ArbitraryDataType{
        "some_data",
        arbitraryScore,
    }
)


/**********  PEEK  **********/
highestPriorityData := q1.Peek()


/**********  DEQUEUE  **********/
highestPriorityData, q1 = q1.Dequeue()
casted, ok := highestPriorityData.(ArbitraryDataType)


/**********  MELD  **********/
anotherQ := priorityQ.NewImmutableSynchronousQ()
melded := q1.Meld(anotherQ)

All Cases

• Constant time enqueue

Use Cases for the Mutable Parallel Priority Queue

• Thread safe priority queue
• Constant time dequeue on the hot path (fast dequeues are an illusion where tree restructuring occurs in the background after a value is returned to the caller)
• A relatively high priority element, but not necessarily the highest priority element, is an acceptable value during concurrent dequeue operations
• Best case constant time meld

Use Cases for the Immutable Synchronous Priority Queue

• Not inherently thread safe
• Constant time meld
• Complete immutability is desirable (Any reference to a particular version of the data structure can be maintained indefinitely)
• Logarithmic time dequeues

Documentation

Index

• Constants
• Variables
• func MaxParallelism() int
• type ParallelQ
  • func (q ParallelQ) BlockUntilNoPending()
  • func (q ParallelQ) Dequeue() (PriorityScorer, PriorityQ)
  • func (q ParallelQ) Enqueue(priority PriorityScorer) PriorityQ
  • func (q ParallelQ) IsEmpty() bool
  • func (q ParallelQ) Length() int
  • func (q ParallelQ) Meld(otherQ PriorityQ) PriorityQ
  • func (q ParallelQ) Peek() PriorityScorer
• type PriorityQ
  • func NewImmutableSynchronousQ() PriorityQ
  • func NewMutableParallelQ() PriorityQ
• type PriorityScorer
• type SynchronousQ
  • func (s SynchronousQ) Dequeue() (PriorityScorer, PriorityQ)
  • func (b SynchronousQ) DequeueWithMergeCallback(mergeCallback func([]node, ...*skewBinomial) skewBinomial) (PriorityScorer, bootstrappedSkewBinomial)
  • func (s SynchronousQ) Enqueue(priority PriorityScorer) PriorityQ
  • func (b SynchronousQ) IsEmpty() bool
  • func (b SynchronousQ) Length() int
  • func (s SynchronousQ) Meld(otherQ PriorityQ) PriorityQ
  • func (b SynchronousQ) Peek() PriorityScorer
  • func (b SynchronousQ) Score() int64
  • func (b SynchronousQ) String() string
• type ThreadSafeList
  • func NewThreadSafeList() ThreadSafeList
  • func (t ThreadSafeList) Count() int
  • func (t *ThreadSafeList) DeleteObject(object unsafe.Pointer) (success bool)
  • func (t *ThreadSafeList) InsertObject(object unsafe.Pointer, lessThanFn func(unsafe.Pointer, unsafe.Pointer) bool)
  • func (t ThreadSafeList) Iter() <-chan unsafe.Pointer
  • func (t ThreadSafeList) LengthGreaterThan(n int) bool
  • func (t ThreadSafeList) Peek() unsafe.Pointer
  • func (t *ThreadSafeList) PopFirst(failAddress unsafe.Pointer) unsafe.Pointer
  • func (t *ThreadSafeList) PopHead() unsafe.Pointer
  • func (t ThreadSafeList) PopNth(n int, failAddress unsafe.Pointer) unsafe.Pointer
• type TrySplitParameter

Constants

const MELD_PERIOD = 2

const PARALLELISM_COEFF = 2

this number is based off of simple tuning experimentation

Variables

var FAIL_ADDRESS unsafe.Pointer = unsafe.Pointer(new(int8))

var NULL_OBJECT = nullObject{}

Functions

func MaxParallelism

func MaxParallelism() int

Types

type ParallelQ

type ParallelQ struct {
	// contains filtered or unexported fields
}

func (ParallelQ) BlockUntilNoPending

func (q ParallelQ) BlockUntilNoPending()

func (ParallelQ) Dequeue

func (q ParallelQ) Dequeue() (PriorityScorer, PriorityQ)

func (ParallelQ) Enqueue

func (q ParallelQ) Enqueue(priority PriorityScorer) PriorityQ

func (ParallelQ) IsEmpty

func (q ParallelQ) IsEmpty() bool

func (ParallelQ) Length

func (q ParallelQ) Length() int

func (ParallelQ) Meld

func (q ParallelQ) Meld(otherQ PriorityQ) PriorityQ

func (ParallelQ) Peek

func (q ParallelQ) Peek() PriorityScorer

type PriorityQ

type PriorityQ interface {
	Enqueue(priority PriorityScorer) PriorityQ
	Peek() PriorityScorer
	Meld(otherQ PriorityQ) PriorityQ
	Dequeue() (PriorityScorer, PriorityQ)
	Length() int
	IsEmpty() bool
}

func NewImmutableSynchronousQ

func NewImmutableSynchronousQ() PriorityQ

func NewMutableParallelQ

func NewMutableParallelQ() PriorityQ

type PriorityScorer

type PriorityScorer interface {
	Score() int64
}

type SynchronousQ

type SynchronousQ struct {
	// contains filtered or unexported fields
}

func (SynchronousQ) Dequeue

func (s SynchronousQ) Dequeue() (PriorityScorer, PriorityQ)

func (SynchronousQ) DequeueWithMergeCallback

func (b SynchronousQ) DequeueWithMergeCallback(mergeCallback func([]node, ...*skewBinomial) skewBinomial) (PriorityScorer, bootstrappedSkewBinomial)

func (SynchronousQ) Enqueue

func (s SynchronousQ) Enqueue(priority PriorityScorer) PriorityQ

func (SynchronousQ) IsEmpty

func (b SynchronousQ) IsEmpty() bool

func (SynchronousQ) Length

func (b SynchronousQ) Length() int

func (SynchronousQ) Meld

func (s SynchronousQ) Meld(otherQ PriorityQ) PriorityQ

func (SynchronousQ) Peek

func (b SynchronousQ) Peek() PriorityScorer

func (SynchronousQ) Score

func (b SynchronousQ) Score() int64

func (SynchronousQ) String

func (b SynchronousQ) String() string

type ThreadSafeList

type ThreadSafeList struct {
	// contains filtered or unexported fields
}

func NewThreadSafeList

func NewThreadSafeList() ThreadSafeList

func (ThreadSafeList) Count

func (t ThreadSafeList) Count() int

func (*ThreadSafeList) DeleteObject

func (t *ThreadSafeList) DeleteObject(object unsafe.Pointer) (success bool)

func (*ThreadSafeList) InsertObject

func (t *ThreadSafeList) InsertObject(object unsafe.Pointer, lessThanFn func(unsafe.Pointer, unsafe.Pointer) bool)

func (ThreadSafeList) Iter

func (t ThreadSafeList) Iter() <-chan unsafe.Pointer

func (ThreadSafeList) LengthGreaterThan

func (t ThreadSafeList) LengthGreaterThan(n int) bool

func (ThreadSafeList) Peek

func (t ThreadSafeList) Peek() unsafe.Pointer

func (*ThreadSafeList) PopFirst

func (t *ThreadSafeList) PopFirst(failAddress unsafe.Pointer) unsafe.Pointer

func (*ThreadSafeList) PopHead

func (t *ThreadSafeList) PopHead() unsafe.Pointer

func (ThreadSafeList) PopNth

func (t ThreadSafeList) PopNth(n int, failAddress unsafe.Pointer) unsafe.Pointer

type TrySplitParameter

type TrySplitParameter struct {
	N int
	// contains filtered or unexported fields
}