beeker1121 / goque Goto Github PK

The Update method in stacks and queues did not check if the item actually existed within the stack or queue. Commit 303be53 fixes this.

Currently there is a bug on levelDB, unfortunately the implication is the error is also present in goqueue.

(google/leveldb#593)

This error consist on levelDB not getting rid off old ldb files. Currently I am working on an application that handles a high amount of incoming data that has to be stored while being processed. After a couple days running I end up with high amount of ldb files (> 2000).

Then once I restart my application and the gopqueue wants to do a recovery, it ends up exceeding the allowed number of open files descriptors on my system.

I would like to fix this issue by using the CompactRange function from leveldb while my program is executing. Nevertheless, due to the lack of exposure to the db struct from goqueue and no Compact/Clean function from the queue struct itself, this is not possible.

It would be nice if either the db struct could be exposed or if there was a Clean/Compact call from the queue itself.

It can happen that OpenQueue() and the like fail due to underlying DB corruption (e. g., corrupted manifest file after a power outage).

In this case, it would be nice if goque exposed the repair functionality of its underlying DB, for example through an additional function OpenQueueRepair() which attempts to repair the DB if it fails to open.

I'm looking for exactly this but need to use it with RPC for other processes to access the queue from a daemon. I was just wondering if there's any way of getting a queue item and then confirming it's been stored before deleting it? Just in case the other process crashes before processing the item...

Thanks!

I've to store a struct type in the queue. I don't have much control on this struct type and it can be a nested json data that I get from a request. For example

{
"name": "string",
  "quota": {
    "compute": {},
    "storage": {}
  },
  "regime": "string",
  "request_id": "test",
  "requestor_id": "string",
  "service_group": "string",
  "service_resource_id": "string",
  "tags": [
    "string"
  ],
  "update_uri": "string"
}

So if I try to add this struct (of type map[string]interface{}) as a prefix queue using pq.EnqueueObject([]byte(k), v)
Then its not getting added to the queue.

I think gob module is not able to decode it.

Is there any way to be able to store this data ?

I'm thinking about changing the API of goque so it's a bit more idiomatic. I wanted to see what people would think of the change, if it's a good idea or not.

The API change would consist of taking out the Item object as a parameter to the Enqueue, Push, and Update methods. Instead, the value itself you wish to enqueue/push/update will be passed as the parameter, which then returns a new Item.

For example, instead of:
func (q *Queue) Enqueue(item *Item) error {

the method would be changed to:
func (q *Queue) Enqueue(value []byte) (*Item, error) {

The prior API would remain tagged as v1.0.2 with a legacy branch created and README.md updated, notifying everyone of the changes.

The Update function was bothering me, as there's no definition of what the Item object passed to it should contain. Do you think this API change would be a more idiomatic approach?

Would it be possible to make goque to reclaim disk space? The algorithm could be that if queue is empty and size of database file is larger than the initial extent (or a configurable limit) then delete it and re-create.

The motivation for this feature is that we want to use this queue for persisting some telemetry data when network connection goes down. The amount of data is quite substantial but these event do not occur often and application must stay online 24/7/365. So, the problem is that one outage can use a lot of disk space that is never returned back. Once system operates normally the queue will be very shallow.

The application may use few dozen queues at the same time but outages do not occur at the same time so waste of storage will multiply and can easily take system down if left unattended.

I suppose we could work around this problem by recreating the queues from outside but it would be nicer to have this feature in goque. Would it be possible to add this feature?

Hi,

I'm working on a restful microservice and using PrefixQueue.

pq, err := goque.OpenPrefixQueue(DATADIR)
gives error on concurrent access. That is, it fails on trying to access the db on concurrent API calls(or goroutines).

error message- "resource temporarily unavailable"
PS - I'm closing the db after using it.

Is there any workaround for this? I need to be able to access the DB concurrently

Is it possible to block on peek or pop?

Currently I'm just polling in a loop, but I guess in the empty-queue case its lightweight enough: q.RLock() then q.getItemByID(...) => q.Length() => q.tail - q.head, so basically just a lock and a comparison

If it makes the use case easier. I am writing/enqueing from multiple places but only reading/dequeing from one place. I don't mind spurious wakeups and I don't need to race against other goroutines which could also be waiting - I just want to block until the queue is not empty.

Many thanks

func foo(q *goque.Queue) {
    ticker := time.NewTicker(pollPeriod)
    defer ticker.Stop()
    
    for {
        select {
            case <-ticker.C: // blocks
        }
        
        item, err := q.Peek()
        if err == goque.ErrEmpty { continue }
        if err != nil { panic(err) }
        
        fmt.Println(item.ToString())
    }
}

func main() {
   // ...
   go foo(q)
   // ...
}

if an object contains a pointer and is written to a queue (did only test it with priority queues) then reading the object from the queue can result in a different object when the dereferenced pointer has a zero value. In that case the pointer itself is nil instead of a pointer to a zero-value.

See the output of the following program:

package main

import (
"fmt"
"github.com/beeker1121/goque"
"log"
)

type Obj struct {
A *int
B int
}

func (o Obj) String() string {
if o.A != nil {
return fmt.Sprintf("%v %v %v", o.A, *o.A, o.B)
} else {
return fmt.Sprintf("%v %v", o.A, o.B)
}
}

func main() {
queue, err := goque.OpenPriorityQueue("queuetest", goque.ASC)
if err != nil {
log.Fatal("OpenPriorityQueue()", err)
}

o1 := Obj{new(int), 1}
*o1.A = 0 // not necessary
    // *o1.A = 10 // this will work
    // o1.A = nil // will work, too

log.Println("Writing to queue:", o1)

_, err = queue.EnqueueObject(0, o1)
if err != nil {
    log.Println("EnqueueObject()", o1, err)
}

item, err := queue.Dequeue()
if err != nil {
    log.Fatal("Dequeue()", err)
}
o2 := Obj{}
err = item.ToObject(&o2)
if err != nil {
    log.Fatal("ToObject()", err)
}

log.Println("Read from queue:", o2)

queue.Drop()

}

If checkGoqueType fails within OpenPrefixQueue, there is no way to close the already-constructed object. On Windows, the file lock held will prevent deletion of the path.

Other types (queue, stack, priority queue) return an object, so prefix queue should, as well.

Cant import version 2.x.x to my project with go.mod
Error:

go mod vendor
go: errors parsing go.mod:  
require github.com/beeker1121/goque: version "v2.1.1" invalid: module contains a go.mod file, so major version must be compatible: should be v0 or v1, not v2

It want it to be github.com/beeker1121/goque/v2, or use only 1.x.x version
Example, how it was implemented in other project github.com/savsgio/atreugo
Thanks!

When a PriorityQueue is already closed or dropped the Length method still returns the number of elements available before close/drop.

Maybe 0 or an error would be a better return value in this case.

goque support new rocksdb ?

Will it block when dequeue-ing?