Depends is a library intended to do queue re-ordering based on read/write dependencies, not the reliable brand for underware, pads, and protection. This library utilizes Manifold for stream abstraction and for asyncronous workflows in order to maximize compatibility with any of the abstractions that implement Manifold sources and/or sinks.
This library is utilizing Clojure 1.9 (alpha 10), clojure.spec and clojure.test.check.
It's on Clojars!
Depends is fairly straight forward, there are only a handful (so far,) of functions that you should be thinking about when using Depends:
First, there is depends/dependify
which creates a single instance of
a dependency manager. Its argument are an input (anything sourceable by
manifold,) and output (anything sinkable by Manifold,) and a map describing the
options to be applied. Its output is a map of the internals for that instance.
(def input (manifold.stream/stream 10))
(def output (manifold.stream/stream 10))
(depends/dependify input output {:max-waiting 10 :clean-up-interval 2500})
However, you don't need to use Manifold streams for your abstractions. Manifold supports several kinds of abstractions out of the box and core.async channels is one of them.
(def input (clojure.core.async/chan 10))
(def output (clojure.core.async/chan 10))
(depends/dependify input output)
Currently, the option map for dependify
currently accepts two options:
:max-waiting
is used to describe the maximum number of items that can hold on to a lock at once. This mechanism is to provide back pressure on the incoming stream so system resources aren't consumed until they're gone.:clean-up-interval
is used to describe how often Depends should scan for realized locks to be evicted from the dependency map. By default, this value is 15000 milliseconds.
Whatever stream abstraction is used for the output
argument will get items
with the structure {:depends/data ... :depends/complete << … >> }
The
message that came in from input is what is in :depends/data
and
:depends/complete
is a Manifold deferred which, when realized will release
its lock on any dependencies that message may have had. So with that said,
Depends can't just figure out what your dependencies are. That would be crazy!
You need to specify that for every message that comes into the input
a unique identifier that indicates what it's doing. This is important because even
though something isn't ready to run yet, it doesn't mean that something that
arrived after it, can't go ahead of any tasks that are waiting. There are
:read
and :write
dependencies. A :read
on the same identifier
will not block subsequent :read
dependencies with the same identifier.
:write
however, must wait for all :read
events that came before it
to complete and blocks all subsequent :read
and :write
events that
come after a :write
.
It's very easy to signal that the task is complete. You simply need to realize
the :depends/complete
deferred value. That's it.
(def input (s/stream 10))
(def output (s/stream 10))
(def system (depends/dependify input output))
(s/put!
input
^{:dependencies {[{:anything "you want"} 123] :write}}
["Only objects that implement the IMetas protocol can be used."])
;;; << true >>
(s/put!
input
^{:dependencies {[{:anything "you want"} 123] :read}}
{:more "data in" :even-more "places"})
;;; << true >>
(s/put! input {:data "with no dependencies"})
;;; << true >>
(def i1 (s/take! output))
i1
;;; << {:depends/data ["Only objects that implement the IMetas protocol can be used."], :depends/complete << … >>} >>
(def i2 (s/take! output))
i2
;;; << {:depends/data {:data "with no dependencies"}, :depends/complete << … >>} >>
(def i3 (s/take! output))
i3
;;; << … >>
(d/success! (:depends/complete @i1) true)
;;; true
i1
;;; << {:depends/data ["Only objects that implement the IMetas protocol can be used."], :depends/complete << true >>} >>
i3
;;; << {:depends/data {:more "data in", :even-more "places"}, :depends/complete << … >>} >>
As you can see, once the deferred on the first item was completed, the second item showed up but, not until the third message, which had no dependencies, was processed before the first was "completed".
What if something goes wrong? Will these locks last forever and deadlock everything that comes after it? Out of the box, yes, that's what will happen however, it's not hard to make a simple fix for this. Manifold already provides a timeout mechanism for deferred values and Depends has a stream mapping function that will automatically apply a timeout for every item in a stream coming its way. So you could apply this by doing the following:
(def input (s/stream 10))
(def output (s/stream 10))
(def pre-output (s/stream))
(s/connect (depends/map-timeout pre-output 1000) output)
(def system (depends/dependify input pre-output))
Now every message that gets put on the output buffered stream will have no less than 1 second to complete before its lock it automatically dropped.
TODO: Finish the README.
- VLACS© [email protected] 2016
- Jon Doane [email protected] 2016
Distributed under the Eclipse Public License either version 1.0 or (at your option) any later version.