goque

package module
v2.0.1+incompatible Latest Latest
Warning

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

 Go to latest Published: Jul 31, 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.

IMPORTANT UPDATE: Goque has been updated to v2, which introduces a completely new API for the Enqueue, Push, and Update methods. Please refer the v2 release for more information on exactly what has been changed. The prior API can still be found at release v1.0.2.

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()

Push an item:

item, err := s.Push([]byte("item value"))
// or
item, err := s.PushString("item value")
// or
item, err := s.PushObject(Object{X:1})

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:

item, err := s.Update(1, []byte("new value"))
// or
item, err := s.UpdateString(1, "new value")
// or
item, err := s.UpdateObject(1, 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()

Enqueue an item:

item, err := q.Enqueue([]byte("item value"))
// or
item, err := q.EnqueueString("item value")
// or
item, err := q.EnqueueObject(Object{X:1})

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:

item, err := q.Update(1, []byte("new value"))
// or
item, err := q.UpdateString(1, "new value")
// or
item, err := q.UpdateObject(1, 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()

Enqueue an item:

item, err := pq.Enqueue(0, []byte("item value"))
// or
item, err := pq.EnqueueString(0, "item value")
// or
item, err := pq.EnqueueObject(0, Object{X:1})

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 58 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:

item, err := pq.Update(0, 1, []byte("new value"))
// or
item, err := pq.UpdateString(0, 1, "new value")
// or
item, err := pq.UpdateObject(0, 1, 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              8104 ns/op             522 B/op          7 allocs/op
BenchmarkPriorityQueueDequeue     200000             18622 ns/op            1166 B/op         17 allocs/op
BenchmarkQueueEnqueue             200000              8049 ns/op             487 B/op          7 allocs/op
BenchmarkQueueDequeue             200000             18970 ns/op            1089 B/op         17 allocs/op
BenchmarkStackPush                200000              8145 ns/op             487 B/op          7 allocs/op
BenchmarkStackPop                 200000             18947 ns/op            1097 B/op         17 allocs/op
ok      github.com/beeker1121/goque     22.549s

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
nstafie (https://github.com/nstafie) - Help with structure

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
	}

	// Enqueue an object.
	item, err := q.EnqueueObject(object{X: 1, Y: 2})
	if err != nil {
		fmt.Println(err)
		return
	}

	fmt.Println(item.ID)  // 1
	fmt.Println(item.Key) // [0 0 0 0 0 0 0 1]

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

	// Create variable to hold our object in.
	var obj object

	// Decode 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}

	// Delete the queue and its database.
	q.Drop()
}

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()

	// Enqueue the item.
	item, err := pq.Enqueue(0, []byte("item value"))
	if err != nil {
		fmt.Println(err)
		return
	}

	fmt.Println(item.ID)         // 1
	fmt.Println(item.Priority)   // 0
	fmt.Println(item.Key)        // [0 58 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.
	item, err = pq.Update(item.Priority, item.ID, []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

	// Delete the queue and its database.
	pq.Drop()
}

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()

	// Enqueue an item.
	item, err := q.Enqueue([]byte("item value"))
	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.
	item, err = q.Update(item.ID, []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

	// Delete the queue and its database.
	q.Drop()
}

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()

	// Push an item onto the stack.
	item, err := s.Push([]byte("item value"))
	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.
	item, err = s.Update(item.ID, []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

	// Delete the stack and its database.
	s.Drop()
}

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 (*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 (*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(priority uint8, value []byte) (*PriorityItem, error)

Enqueue adds an item to the priority queue.

func (*PriorityQueue) EnqueueObject 

func (pq *PriorityQueue) EnqueueObject(priority uint8, value interface{}) (*PriorityItem, error)

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

func (*PriorityQueue) EnqueueString 

func (pq *PriorityQueue) EnqueueString(priority uint8, value string) (*PriorityItem, error)

EnqueueString is a helper function for Enqueue that accepts a value as a string rather than a byte slice.

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(priority uint8, id uint64, newValue []byte) (*PriorityItem, error)

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

func (*PriorityQueue) UpdateObject 

func (pq *PriorityQueue) UpdateObject(priority uint8, id uint64, newValue interface{}) (*PriorityItem, 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(priority uint8, id uint64, newValue string) (*PriorityItem, 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(value []byte) (*Item, error)

Enqueue adds an item to the queue.

func (*Queue) EnqueueObject 

func (q *Queue) EnqueueObject(value interface{}) (*Item, error)

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

func (*Queue) EnqueueString 

func (q *Queue) EnqueueString(value string) (*Item, error)

EnqueueString is a helper function for Enqueue that accepts a value as a string rather than a byte slice.

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(id uint64, newValue []byte) (*Item, error)

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

func (*Queue) UpdateObject 

func (q *Queue) UpdateObject(id uint64, newValue interface{}) (*Item, 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(id uint64, newValue string) (*Item, 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(value []byte) (*Item, error)

Push adds an item to the stack.

func (*Stack) PushObject 

func (s *Stack) PushObject(value interface{}) (*Item, error)

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

func (*Stack) PushString 

func (s *Stack) PushString(value string) (*Item, error)

PushString is a helper function for Push that accepts a value as a string rather than a byte slice.

func (*Stack) Update 

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

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

func (*Stack) UpdateObject 

func (s *Stack) UpdateObject(id uint64, newValue interface{}) (*Item, 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(id uint64, newValue string) (*Item, 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.