goque

package module
v1.0.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 22, 2016 License: MIT Imports: 9 Imported by: 0

README

Goque GoDoc License Go Report Card Build Status

Goque provides embedded, disk-based implementations of stack, queue, and priority queue data structures.

Motivation for creating this project was the need for a persistent priority queue that remained performant while growing well beyond the available memory of a given machine. While there are many packages for Go offering queues, they all seem to be memory based and/or standalone solutions that are not embeddable within an application.

Instead of using an in-memory heap structure to store data, everything is stored using the Go port of LevelDB. This results in very little memory being used no matter the size of the database, while read and write performance remains near constant.

Features

  • Provides stack (LIFO), queue (FIFO), and priority queue structures.
  • Stacks and queues (but not priority queues) are interchangeable.
  • Persistent, disk-based.
  • Optimized for fast inserts and reads.
  • Goroutine safe.
  • Designed to work with large datasets outside of RAM/memory.

Installation

Fetch the package from GitHub:

go get github.com/beeker1121/goque

Import to your project:

import "github.com/beeker1121/goque"

Usage

Stack

Stack is a LIFO (last in, first out) data structure.

Create or open a stack:

s, err := goque.OpenStack("data_dir")
...
defer s.Close()

Create a new item:

item := goque.NewItem([]byte("item value"))
// or
item := goque.NewItemString("item value")
// or
item, err := goque.NewItemObject(Object{X:1})

Push an item:

err := s.Push(item)

Pop an item:

item, err := s.Pop()
...
fmt.Println(item.ID)         // 1
fmt.Println(item.Key)        // [0 0 0 0 0 0 0 1]
fmt.Println(item.Value)      // [105 116 101 109 32 118 97 108 117 101]
fmt.Println(item.ToString()) // item value

// Decode to object.
var obj Object
err := item.ToObject(&obj)
...
fmt.Printf("%+v\n", obj) // {X:1}

Peek the next stack item:

item, err := s.Peek()
// or
item, err := s.PeekByOffset(1)
// or
item, err := s.PeekByID(1)

Update an item in the stack:

err := s.Update(item, []byte("new value"))
// or
err := s.UpdateString(item, "new value")
// or
err := s.UpdateObject(item, Object{X:2})

Delete the stack and underlying database:

s.Drop()
Queue

Queue is a FIFO (first in, first out) data structure.

Methods

Create or open a queue:

q, err := goque.OpenQueue("data_dir")
...
defer q.Close()

Create a new item:

item := goque.NewItem([]byte("item value"))
// or
item := goque.NewItemString("item value")
// or
item, err := goque.NewItemObject(Object{X:1})

Enqueue an item:

err := q.Enqueue(item)

Dequeue an item:

item, err := q.Dequeue()
...
fmt.Println(item.ID)         // 1
fmt.Println(item.Key)        // [0 0 0 0 0 0 0 1]
fmt.Println(item.Value)      // [105 116 101 109 32 118 97 108 117 101]
fmt.Println(item.ToString()) // item value

// Decode to object.
var obj Object
err := item.ToObject(&obj)
...
fmt.Printf("%+v\n", obj) // {X:1}

Peek the next queue item:

item, err := q.Peek()
// or
item, err := q.PeekByOffset(1)
// or
item, err := q.PeekByID(1)

Update an item in the queue:

err := q.Update(item, []byte("new value"))
// or
err := q.UpdateString(item, "new value")
// or
err := q.UpdateObject(item, Object{X:2})

Delete the queue and underlying database:

q.Drop()
Priority Queue

PriorityQueue is a FIFO (first in, first out) queue with priority levels.

Methods

Create or open a priority queue:

pq, err := goque.OpenPriorityQueue("data_dir", goque.ASC)
...
defer pq.Close()

Create a new item:

item := goque.NewPriorityItem([]byte("item value"), 0)
// or
item := goque.NewPriorityItemString("item value", 0)
// or
item, err := goque.NewPriorityItemObject(Object{X:1}, 0)

Enqueue an item:

err := pq.Enqueue(item)

Dequeue an item:

item, err := pq.Dequeue()
// or
item, err := pq.DequeueByPriority(0)
...
fmt.Println(item.ID)         // 1
fmt.Println(item.Priority)   // 0
fmt.Println(item.Key)        // [0 0 0 0 0 0 0 1]
fmt.Println(item.Value)      // [105 116 101 109 32 118 97 108 117 101]
fmt.Println(item.ToString()) // item value

// Decode to object.
var obj Object
err := item.ToObject(&obj)
...
fmt.Printf("%+v\n", obj) // {X:1}

Peek the next priority queue item:

item, err := pq.Peek()
// or
item, err := pq.PeekByOffset(1)
// or
item, err := pq.PeekByPriorityID(0, 1)

Update an item in the priority queue:

err := pq.Update(item, []byte("new value"))
// or
err := pq.UpdateString(item, "new value")
// or
err := pq.UpdateObject(item, Object{X:2})

Delete the priority queue and underlying database:

pq.Drop()

Benchmarks

Benchmarks were run on a Google Compute Engine n1-standard-1 machine (1 vCPU 3.75 GB of RAM):

go test -bench=.
PASS
BenchmarkPriorityQueueEnqueue     200000              8102 ns/op             442 B/op          5 allocs/op
BenchmarkPriorityQueueDequeue     200000             18602 ns/op            1161 B/op         17 allocs/op
BenchmarkQueueEnqueue             200000              7582 ns/op             399 B/op          5 allocs/op
BenchmarkQueueDequeue             200000             19317 ns/op            1071 B/op         17 allocs/op
BenchmarkStackPush                200000              7847 ns/op             399 B/op          5 allocs/op
BenchmarkStackPop                 200000             18950 ns/op            1081 B/op         17 allocs/op

Thanks

syndtr (https://github.com/syndtr) - LevelDB port to Go
bogdanovich (https://github.com/bogdanovich/siberite) - Server based queue for Go using LevelDB
connor4312 (https://github.com/connor4312) - Recommending BoltDB/LevelDB, helping with structure
bwmarrin (https://github.com/bwmarrin) - Recommending BoltDB/LevelDB
zeroZshadow (https://github.com/zeroZshadow) - Code review and optimization

Documentation

Overview

Package goque provides embedded, disk-based implementations of stack, queue, and priority queue data structures.

Motivation for creating this project was the need for a persistent priority queue that remained performant while growing well beyond the available memory of a given machine. While there are many packages for Go offering queues, they all seem to be memory based and/or standalone solutions that are not embeddable within an application.

Instead of using an in-memory heap structure to store data, everything is stored using the Go port of LevelDB (https://github.com/syndtr/goleveldb). This results in very little memory being used no matter the size of the database, while read and write performance remains near constant.

See README.md or visit https://github.com/beeker1121/goque for more info.

Example (Object)

ExampleObject demonstrates enqueuing a struct object. 

package main

import (
	"fmt"

	"github.com/beeker1121/goque"
)

func main() {
	// Open/create a queue.
	q, err := goque.OpenQueue("data_dir")
	if err != nil {
		fmt.Println(err)
		return
	}
	defer q.Close()

	// Define our struct.
	type object struct {
		X int
		Y int
	}

	// Create a new item with our struct.
	item, err := goque.NewItemObject(object{X: 1, Y: 2})
	if err != nil {
		fmt.Println(err)
		return
	}

	// Enqueue the item.
	err = q.Enqueue(item)
	if err != nil {
		fmt.Println(err)
		return
	}

	fmt.Println(item.ID)    // 1
	fmt.Println(item.Key)   // [0 0 0 0 0 0 0 1]
	fmt.Println(item.Value) // [105 116 101 109 32 118 97 108 117 101]

	// Dequeue the item.
	deqItem, err := q.Dequeue()
	if err != nil {
		fmt.Println(err)
		return
	}

	// Create a variable to hold our decoded struct in.
	var obj object

	// Decode the item into our struct type.
	if err := deqItem.ToObject(&obj); err != nil {
		fmt.Println(err)
		return
	}

	fmt.Printf("%+v\n", obj) // {X:1 Y:2}
}

Output:
Example (PriorityQueue)

ExampleQueue demonstrates the implementation of a Goque queue. 

package main

import (
	"fmt"

	"github.com/beeker1121/goque"
)

func main() {
	// Open/create a priority queue.
	pq, err := goque.OpenPriorityQueue("data_dir", goque.ASC)
	if err != nil {
		fmt.Println(err)
		return
	}
	defer pq.Close()

	// Create a new item.
	item := goque.NewPriorityItem([]byte("item value"), 0)

	// Enqueue the item.
	err = pq.Enqueue(item)
	if err != nil {
		fmt.Println(err)
		return
	}

	fmt.Println(item.ID)         // 1
	fmt.Println(item.Priority)   // 0
	fmt.Println(item.Key)        // [0 0 0 0 0 0 0 1]
	fmt.Println(item.Value)      // [105 116 101 109 32 118 97 108 117 101]
	fmt.Println(item.ToString()) // item value

	// Change the item value in the queue.
	err = pq.Update(item, []byte("new item value"))
	if err != nil {
		fmt.Println(err)
		return
	}

	fmt.Println(item.ToString()) // new item value

	// Dequeue the next item.
	deqItem, err := pq.Dequeue()
	if err != nil {
		fmt.Println(err)
		return
	}

	fmt.Println(deqItem.ToString()) // new item value
}

Output:
Example (Queue)

ExampleQueue demonstrates the implementation of a Goque queue. 

package main

import (
	"fmt"

	"github.com/beeker1121/goque"
)

func main() {
	// Open/create a queue.
	q, err := goque.OpenQueue("data_dir")
	if err != nil {
		fmt.Println(err)
		return
	}
	defer q.Close()

	// Create a new item.
	item := goque.NewItem([]byte("item value"))

	// Enqueue the item.
	err = q.Enqueue(item)
	if err != nil {
		fmt.Println(err)
		return
	}

	fmt.Println(item.ID)         // 1
	fmt.Println(item.Key)        // [0 0 0 0 0 0 0 1]
	fmt.Println(item.Value)      // [105 116 101 109 32 118 97 108 117 101]
	fmt.Println(item.ToString()) // item value

	// Change the item value in the queue.
	err = q.Update(item, []byte("new item value"))
	if err != nil {
		fmt.Println(err)
		return
	}

	fmt.Println(item.ToString()) // new item value

	// Dequeue the next item.
	deqItem, err := q.Dequeue()
	if err != nil {
		fmt.Println(err)
		return
	}

	fmt.Println(deqItem.ToString()) // new item value
}

Output:
Example (Stack)

ExampleStack demonstrates the implementation of a Goque stack. 

package main

import (
	"fmt"

	"github.com/beeker1121/goque"
)

func main() {
	// Open/create a stack.
	s, err := goque.OpenStack("data_dir")
	if err != nil {
		fmt.Println(err)
		return
	}
	defer s.Close()

	// Create a new item.
	item := goque.NewItem([]byte("item value"))

	// Push it onto the stack.
	err = s.Push(item)
	if err != nil {
		fmt.Println(err)
		return
	}

	fmt.Println(item.ID)         // 1
	fmt.Println(item.Key)        // [0 0 0 0 0 0 0 1]
	fmt.Println(item.Value)      // [105 116 101 109 32 118 97 108 117 101]
	fmt.Println(item.ToString()) // item value

	// Change the item value in the stack.
	err = s.Update(item, []byte("new item value"))
	if err != nil {
		fmt.Println(err)
		return
	}

	fmt.Println(item.ToString()) // new item value

	// Pop an item off the stack.
	popItem, err := s.Pop()
	if err != nil {
		fmt.Println(err)
		return
	}

	fmt.Println(popItem.ToString()) // new item value
}

Output:

Index

Examples

Constants

View Source 
const (
	ASC  order = iota // Set priority level 0 as most important.
	DESC              // Set priority level 255 as most important.
)

Defines which priority order to dequeue in.

Variables

View Source 
var (
	// ErrIncompatibleType is returned when the opener type is
	// incompatible with the stored Goque type.
	ErrIncompatibleType = errors.New("goque: Opener type is incompatible with stored Goque type")

	// ErrEmpty is returned when the stack or queue is empty.
	ErrEmpty = errors.New("goque: Stack or queue is empty")

	// ErrOutOfBounds is returned when the ID used to lookup an item
	// is outside of the range of the stack or queue.
	ErrOutOfBounds = errors.New("goque: ID used is outside range of stack or queue")

	// ErrDBClosed is returned when the Close function has already
	// been called, causing the stack or queue to close, as well as
	// its underlying database.
	ErrDBClosed = errors.New("goque: Database is closed")
)

Functions

This section is empty.

Types

type Item 

type Item struct {
	ID    uint64
	Key   []byte
	Value []byte
}

Item represents an entry in either a stack or queue.

func NewItem 

func NewItem(value []byte) *Item

NewItem creates a new item for use with a stack or queue.

func NewItemObject 

func NewItemObject(value interface{}) (*Item, error)

NewItemObject is a helper function for NewItem that accepts any value type, which is then encoded into a byte slice using encoding/gob.

func NewItemString 

func NewItemString(value string) *Item

NewItemString is a helper function for NewItem that accepts a value as a string rather than a byte slice.

func (*Item) ToObject 

func (i *Item) ToObject(value interface{}) error

ToObject decodes the item value into the given value type using encoding/gob.

The value passed to this method should be a pointer to a variable of the type you wish to decode into. The variable pointed to will hold the decoded object.

func (*Item) ToString 

func (i *Item) ToString() string

ToString returns the item value as a string.

type PriorityItem 

type PriorityItem struct {
	ID       uint64
	Priority uint8
	Key      []byte
	Value    []byte
}

PriorityItem represents an entry in a priority queue.

func NewPriorityItem 

func NewPriorityItem(value []byte, priority uint8) *PriorityItem

NewPriorityItem creates a new item for use with a priority queue.

func NewPriorityItemObject 

func NewPriorityItemObject(value interface{}, priority uint8) (*PriorityItem, error)

NewPriorityItemObject is a helper function for NewPriorityItem that accepts any value type, which is then encoded into a byte slice using encoding/gob.

func NewPriorityItemString 

func NewPriorityItemString(value string, priority uint8) *PriorityItem

NewPriorityItemString is a helper function for NewPriorityItem that accepts a value as a string rather than a byte slice.

func (*PriorityItem) ToObject 

func (pi *PriorityItem) ToObject(value interface{}) error

ToObject decodes the item value into the given value type using encoding/gob.

The value passed to this method should be a pointer to a variable of the type you wish to decode into. The variable pointed to will hold the decoded object.

func (*PriorityItem) ToString 

func (pi *PriorityItem) ToString() string

ToString returns the priority item value as a string.

type PriorityQueue 

type PriorityQueue struct {
	sync.RWMutex
	DataDir string
	// contains filtered or unexported fields
}

PriorityQueue is a standard FIFO (first in, first out) queue with priority levels.

func OpenPriorityQueue 

func OpenPriorityQueue(dataDir string, order order) (*PriorityQueue, error)

OpenPriorityQueue opens a priority queue if one exists at the given directory. If one does not already exist, a new priority queue is created.

func (*PriorityQueue) Close 

func (pq *PriorityQueue) Close()

Close closes the LevelDB database of the priority queue.

func (*PriorityQueue) Dequeue 

func (pq *PriorityQueue) Dequeue() (*PriorityItem, error)

Dequeue removes the next item in the priority queue and returns it.

func (*PriorityQueue) DequeueByPriority 

func (pq *PriorityQueue) DequeueByPriority(priority uint8) (*PriorityItem, error)

DequeueByPriority removes the next item in the given priority level and returns it.

func (*PriorityQueue) Drop 

func (pq *PriorityQueue) Drop()

Drop closes and deletes the LevelDB database of the priority queue.

func (*PriorityQueue) Enqueue 

func (pq *PriorityQueue) Enqueue(item *PriorityItem) error

Enqueue adds an item to the priority queue.

func (*PriorityQueue) Length 

func (pq *PriorityQueue) Length() uint64

Length returns the total number of items in the priority queue.

func (*PriorityQueue) Peek 

func (pq *PriorityQueue) Peek() (*PriorityItem, error)

Peek returns the next item in the priority queue without removing it.

func (*PriorityQueue) PeekByOffset 

func (pq *PriorityQueue) PeekByOffset(offset uint64) (*PriorityItem, error)

PeekByOffset returns the item located at the given offset, starting from the head of the queue, without removing it.

func (*PriorityQueue) PeekByPriorityID 

func (pq *PriorityQueue) PeekByPriorityID(priority uint8, id uint64) (*PriorityItem, error)

PeekByPriorityID returns the item with the given ID and priority without removing it.

func (*PriorityQueue) Update 

func (pq *PriorityQueue) Update(item *PriorityItem, newValue []byte) error

Update updates an item in the priority queue without changing its position.

func (*PriorityQueue) UpdateObject 

func (pq *PriorityQueue) UpdateObject(item *PriorityItem, newValue interface{}) error

UpdateObject is a helper function for Update that accepts any value type, which is then encoded into a byte slice using encoding/gob.

func (*PriorityQueue) UpdateString 

func (pq *PriorityQueue) UpdateString(item *PriorityItem, newValue string) error

UpdateString is a helper function for Update that accepts a value as a string rather than a byte slice.

type Queue 

type Queue struct {
	sync.RWMutex
	DataDir string
	// contains filtered or unexported fields
}

Queue is a standard FIFO (first in, first out) queue.

func OpenQueue 

func OpenQueue(dataDir string) (*Queue, error)

OpenQueue opens a queue if one exists at the given directory. If one does not already exist, a new queue is created.

func (*Queue) Close 

func (q *Queue) Close()

Close closes the LevelDB database of the queue.

func (*Queue) Dequeue 

func (q *Queue) Dequeue() (*Item, error)

Dequeue removes the next item in the queue and returns it.

func (*Queue) Drop 

func (q *Queue) Drop()

Drop closes and deletes the LevelDB database of the queue.

func (*Queue) Enqueue 

func (q *Queue) Enqueue(item *Item) error

Enqueue adds an item to the queue.

func (*Queue) Length 

func (q *Queue) Length() uint64

Length returns the total number of items in the queue.

func (*Queue) Peek 

func (q *Queue) Peek() (*Item, error)

Peek returns the next item in the queue without removing it.

func (*Queue) PeekByID 

func (q *Queue) PeekByID(id uint64) (*Item, error)

PeekByID returns the item with the given ID without removing it.

func (*Queue) PeekByOffset 

func (q *Queue) PeekByOffset(offset uint64) (*Item, error)

PeekByOffset returns the item located at the given offset, starting from the head of the queue, without removing it.

func (*Queue) Update 

func (q *Queue) Update(item *Item, newValue []byte) error

Update updates an item in the queue without changing its position.

func (*Queue) UpdateObject 

func (q *Queue) UpdateObject(item *Item, newValue interface{}) error

UpdateObject is a helper function for Update that accepts any value type, which is then encoded into a byte slice using encoding/gob.

func (*Queue) UpdateString 

func (q *Queue) UpdateString(item *Item, newValue string) error

UpdateString is a helper function for Update that accepts a value as a string rather than a byte slice.

type Stack 

type Stack struct {
	sync.RWMutex
	DataDir string
	// contains filtered or unexported fields
}

Stack is a standard LIFO (last in, first out) stack.

func OpenStack 

func OpenStack(dataDir string) (*Stack, error)

OpenStack opens a stack if one exists at the given directory. If one does not already exist, a new stack is created.

func (*Stack) Close 

func (s *Stack) Close()

Close closes the LevelDB database of the stack.

func (*Stack) Drop 

func (s *Stack) Drop()

Drop closes and deletes the LevelDB database of the stack.

func (*Stack) Length 

func (s *Stack) Length() uint64

Length returns the total number of items in the stack.

func (*Stack) Peek 

func (s *Stack) Peek() (*Item, error)

Peek returns the next item in the stack without removing it.

func (*Stack) PeekByID 

func (s *Stack) PeekByID(id uint64) (*Item, error)

PeekByID returns the item with the given ID without removing it.

func (*Stack) PeekByOffset 

func (s *Stack) PeekByOffset(offset uint64) (*Item, error)

PeekByOffset returns the item located at the given offset, starting from the head of the stack, without removing it.

func (*Stack) Pop 

func (s *Stack) Pop() (*Item, error)

Pop removes the next item in the stack and returns it.

func (*Stack) Push 

func (s *Stack) Push(item *Item) error

Push adds an item to the stack.

func (*Stack) Update 

func (s *Stack) Update(item *Item, newValue []byte) error

Update updates an item in the stack without changing its position.

func (*Stack) UpdateObject 

func (s *Stack) UpdateObject(item *Item, newValue interface{}) error

UpdateObject is a helper function for Update that accepts any value type, which is then encoded into a byte slice using encoding/gob.

func (*Stack) UpdateString 

func (s *Stack) UpdateString(item *Item, newValue string) error

UpdateString is a helper function for Update that accepts a value as a string rather than a byte slice.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.