clj-commons / durable-queue Goto Github PK

I'm starting to work with this library and Clojure 1.8.0, and as-is, this isn't working well - but if I preload the byte-streams v0.2.2 and exclude that from durable-queues, then it seems to work just fine.

I would suggest upgrading to [byte-streams "0.2.2"] in this project, and then it should be fine.

In the put! function, slab! is called before queue!, writing the data to the slab before offering to the queue.

               (if-let [val (locking q
                              (queue!
                                (vary-meta (slab!) assoc
                                  ::this this-ref
                                  ::queue-name q-name
                                  ::queue q
                                  ::fsync? fsync-take?)))]

The consequence is that the slab files can get larger than the max-queue-size, and upon application restart, exceptions get raised when the slabs are loaded and don't fit into the queue.

(IllegalArgumentException.
                             "'max-queue-size' insufficient to hold existing tasks."))

I've noticed an annoying issue that seems like it might be avoidable.

Let's say that I'm using durable-queue in a project:

(defonce q (queues "/tmp" {}))

and then you're off and having fun and it's working great, and your MacBook Pro connects to some WiFi hotspot at home - or work, and macOS creates a bunch of temp files:

/tmp/wifi-01-25-2017_12:34:56.log

and when the code starts, there are permission issues on these files, and the code exceptions. When I delete these log files, everything is fine.

My question is - can these files in /tmp that aren't durable queue data files - be excluded from the search on startup?

Hi,

Should we consider put! and take! threadsafe?

I have one thread doing puts in bursts at regular interval, and N workers on their own respective threads consuming the queue. Sometimes I get into a situation where I have 1 in progress task and all other workers waiting on tasks while the queue keeps growing (essentially deadlocked at this point). In the code I take!, deref and mark completed tasks immediately (did that to exclude a bug on complete!).
Should I serialize the takes or is this a potential bug? I ll try to come up with a repro at work tomorrow.

I also get the occasional negative in-progress while testing:

09:22:14.938 [clojure-agent-send-off-pool-0] INFO sink.testdq - {foo {:num-slabs 1, :num-active-slabs 1, :enqueued 10947, :retried 0, :completed 10947, :in-progress -1}}

Thanks for the lib by the way, it s super useful.

This library has been a workhorse across java updates. It looks like a bit of work will be required to maintain dependability with the release of java 13 which touched on many aspects of java.nio. The type-specific Buffer classes of java.nio now have absolute (as opposed to relative) bulk get and put methods.

java.nio.ByteBuffer and the other buffer types in java.nio now define absolute bulk get and put methods to transfer contiguous sequences of bytes without regard to or effect on the buffer position.

After upgrading to java 13, I started seeing the following error:

No matching method put found taking 2 args for class java.nio.DirectByteBuffer

Linking back to the source:
https://github.com/Factual/durable-queue/blob/83452cbcc4c15a2244747e77b54a96a23e027691/src/durable_queue.clj#L176

I hope it's a small thing to fix.

> (put! q "example.com" "foo")
true
> (stats q)
{"example.com" {:num-slabs 1, :num-active-slabs 1, :enqueued 1, :retried 0, :completed 0, :in-progress 0}}
> (def q (queues "/tmp"))
> (stats q)
{"com" {:num-slabs 1, :num-active-slabs 0, :enqueued 1, :retried 0, :completed 0, :in-progress 0}}

From my testing, it appears all the characters before the last '.' in the queue name is lost when reloading the queue.

I had another process writing temporary files with names matching #"\w+_\d+" (2015_05_01_xxxxxx.zip) to the same ./tmp directory where durable-queue stored its slabs. After a while, I noticed those temporary files occasionally disappear before the worker process would get a chance to clean them up. Turned out, durable-queue was deleting them, thinking they were empty slabs.

While, I suppose, I shouldn’t have taken the /tmp example in README so seriously, at least it seems worth noting in the docs that the directory is better be dedicated to the process. A stricter regexp (i.e. #"^\w+_\d+$") could also help reduce the risk of collisions.

Specifically: https://github.com/Factual/durable-queue/blob/master/src/durable_queue.clj#L731

It looks like put! will always evaluate to nil, not to true/false based on whether or not the put times out.

I'm not sure if this is as expected, but I found this behavior when I was trying stuff in REPL.

> (dq/put! q :test "test task")
> (def t (dq/take! q :test))
> (dq/complete! t)
true
> (get (dq/stats q) "test")
{:in-progress 0, :completed 1, :retried 0, :enqueued 1, :num-slabs 1, :num-active-slabs 1}
> (dq/complete! t)
true
> (get (dq/stats q) "test")
{:in-progress -1, :completed 2, :retried 0, :enqueued 1, :num-slabs 1, :num-active-slabs 1}

Is this expected behavior and the caller is responsible of keeping track of task status?
If so, is there possibility of making status function public, so that we can check the status of the task?(currently it's private)

[Question]
Here's one I see in my process.
"out" {:enqueued 8332, :retried 0, :completed 8333, :in-progress -1}

I'm unsure how to handle this, as I depend on the values to decide how to progress (completion marking etc.)

Thanks!

Hi,

I've successfully implemented a durable queue according to the stated use case, as an in-process device that does both the puts and takes. However, initially I had a misgiving, and I had two processes, one doing the puts, the other the takes. This did not work. Despite the fact that the queue object was pointing to the same underlying file, the puts of one process did not result in tasks visible to takes in the second process. Now I am just curious, what are the technical reasons for this. Intuitively, I though that no further synchronization was needed beyond fsync to make this work.

Thank you in advance.

I assume this is a version conflict, either of Java or Clojure or durable-queue. My code works fine on my local Macbook Pro, but I created a new EC2 instance with Ubuntu 18.04 and then I added Java:

java -version
openjdk version "1.8.0_191"
OpenJDK Runtime Environment (build 1.8.0_191-8u191-b12-0ubuntu0.18.04.1-b12)
OpenJDK 64-Bit Server VM (build 25.191-b12, mixed mode)

On the EC2 instance I get this error on put!

INFO: java.lang.NoSuchMethodError: java.util.zip.Checksum.update([B)V
  java.lang.NoSuchMethodError: java.util.zip.Checksum.update([B)V
 at durable_queue$checksum.invokeStatic (durable_queue.clj:64)
    durable_queue$checksum.invokePrim (durable_queue.clj:-1)
    durable_queue.TaskSlab.append_to_slab_BANG_ (durable_queue.clj:314)
    durable_queue$queues$reify__6549$slab_BANG___6570.invoke (durable_queue.clj:702)
    durable_queue$queues$reify__6549$fn__6575.invoke (durable_queue.clj:719)
    durable_queue$queues$reify__6549.put_BANG_ (durable_queue.clj:717)
    durable_queue$queues$reify__6549.put_BANG_ (durable_queue.clj:732)
    humongorous_nlp.core$cycle_to_database$fn__13673$fn__13693.invoke (core.clj:665)
    humongorous_nlp.core$cycle_to_database$fn__13673.invoke (core.clj:663)
    clojure.lang.AFn.run (AFn.java:22)
    java.util.concurrent.Executors$RunnableAdapter.call (Executors.java:511)
    java.util.concurrent.FutureTask.runAndReset (FutureTask.java:308)
    java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301 (ScheduledThreadPoolExecutor.java:180)
    java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run (ScheduledThreadPoolExecutor.java:294)
    java.util.concurrent.ThreadPoolExecutor.runWorker (ThreadPoolExecutor.java:1149)
    java.util.concurrent.ThreadPoolExecutor$Worker.run (ThreadPoolExecutor.java:624)
    java.lang.Thread.run (Thread.java:748)

Is this a matter of upgrading Java, or is this a matter of doing a data conversion before calling put! ?

I'm puzzled why this works on my Mac but not the EC2 instance.

Would it be possible to create a version of the queue that would not be durable for testing purposes? Maybe you could pass an optional parameter to set it to in memory on mode.

Hey @ztellman. Thanks for the awesome library!

I noticed that queue names with dashes are automatically converted to underscores. Intentional?

(def dir "/tmp")
(def conn (d/queues dir {}))
(d/put! conn "a-b-c" 42)
(d/stats conn)
;; => {"a_b_c" {:in-progress 0, :completed 0, :retried 0, :enqueued 1, :num-slabs 1, :num-active-slabs 1}}