(ns durable-queue
  (:require
    [clojure.java.io :as io]
    [byte-streams :as bs]
    [clojure.string :as str]
    [primitive-math :as p]
    [taoensso.nippy :as nippy])
  (:import
    [java.lang.reflect
     Method
     Field]
    [java.util.concurrent
     LinkedBlockingQueue
     TimeoutException
     TimeUnit]
    [java.util.concurrent.atomic
     AtomicLong]
    [java.util.zip
     CRC32]
    [java.util.concurrent.locks
     ReentrantReadWriteLock]
    [java.io
     Writer
     File
     RandomAccessFile
     IOException]
    [java.nio.channels
     FileChannel
     FileChannel$MapMode]
    [java.nio
     ByteBuffer
     MappedByteBuffer]
    [java.lang.ref
     WeakReference]))

;;;

(defmacro ^:private with-lock [lock & body]
  `(let [^ReentrantReadWriteLock lock# ~lock
         read-lock# (.readLock lock#)]
     (do
       (.lock read-lock#)
       (try
         ~@body
         (finally
           (.unlock read-lock#))))))

(defmacro ^:private with-exclusive-lock [lock & body]
  `(let [^ReentrantReadWriteLock lock# ~lock
         write-lock# (.writeLock lock#)]
     (do
       (.lock write-lock#)
       (try
         ~@body
         (finally
           (.unlock write-lock#))))))

;;;

(defn- checksum ^long [^long length ^bytes ary]
  (let [crc (CRC32.)]
    (dotimes [i 4]
      (.update crc (p/>> length i)))
    (.update crc ary)
    (.getValue crc)))

;;;

(def ^:private ^:const header-size 14)

(defprotocol ITask
  (^:private status [_] "Returns the task status")
  (^:private status! [_ status] "Sets the task status"))

(defprotocol ITaskSlab
  (^:private unmap [_] "Temporarily releases mapped byte buffer until it's needed again.")
  (^:private mapped? [_] "Returns true if the slab is actively mapped into memory.")
  (^:private sync! [_])
  (^:private invalidate [_ offset len])
  (^:private ^ByteBuffer buffer [_])
  (^:private append-to-slab! [_ descriptor])
  (^:private read-write-lock [_]))

(defmacro ^:private with-buffer [[buf slab] & body]
  `(with-lock (read-write-lock ~slab)
     (when-let [~buf (buffer ~slab)]
       ~@body)))

;;;

(defn create-buffer [filename size]
  (let [raf (doto (RandomAccessFile. (io/file filename) "rw")
              (.setLength size))]
    (try
      (let [fc (.getChannel raf)]
        (try
          (let [buf (.map fc FileChannel$MapMode/READ_WRITE 0 size)]
            (doto buf
              (.put 0 (byte 0))
              .force))
          (finally
            (.close fc))))
      (finally
        (.close raf)))))

(defn load-buffer
  ([filename]
     (load-buffer filename nil nil))
  ([filename offset length]
     (let [_ (assert (.exists (io/file filename)))
           raf (RandomAccessFile. (io/file filename) "rw")]
       (try
         (let [fc (.getChannel raf)]
           (try
             (.map fc
               FileChannel$MapMode/READ_WRITE
               (or offset 0)
               (or length (.length raf)))
             (finally
               (.close fc))))
         (finally
           (.close raf))))))

(let [clean (delay
              (doto (.getMethod
                      (Class/forName "sun.misc.Cleaner")
                      "clean"
                      nil)
                (.setAccessible true)))]
  (defn- unmap-buffer
    "A delightful endrun on the JVM's mmap GC mechanism"
    [^ByteBuffer buf]
    (when (.isDirect buf)
      (try

        (let [^Method clean @clean
              cleaner (doto (.getMethod (class buf) "cleaner" nil)
                        (.setAccessible true))]
          (.invoke clean
            (.invoke cleaner buf nil)
            nil))
        (catch Throwable e
          ;; not much we can do here, sadly
          )))))

(defn- force-buffer
  [^MappedByteBuffer buf offset length]
  (.force buf))

;;;

;; a single task within a slab, assumes that the buffer is sliced around
;; the task's boundaries
(defrecord Task
  [slab
   ^long offset
   ^long length
   status
   deserializer]
  clojure.lang.IDeref
  (deref [_]
    (deserializer))
  ITask
  (status [_]
    (with-buffer [buf slab]
      (or @status
        (let [s (case (.get buf (p/+ offset 1))
                  0 :incomplete
                  1 :in-progress
                  2 :complete)]
          (reset! status s)
          s))))
  (status! [_ s]
    (with-buffer [buf slab]
      (reset! status s)
      (.put buf (p/+ offset 1)
        (case s
          :incomplete 0
          :in-progress 1
          :complete 2))
      (invalidate slab (p/+ offset 1) 1)
      nil)))

(defn- task [slab offset len]
  (Task.
    slab
    offset
    len
    (atom nil)
    (fn []
      (with-buffer [buf slab]
        (let [^ByteBuffer buf (-> buf
                                (.position offset)
                                ^ByteBuffer
                                (.limit (+ offset len))
                                .slice)
              checksum' (.getLong buf 2)
              ary (bs/to-byte-array (.position buf header-size))]
          (when-not (== (checksum (.getInt buf 10) ary) checksum')
            (throw (IOException. "checksum mismatch")))
          (nippy/thaw ary))))))

(defmethod print-method Task [t ^Writer w]
  (.write w
    (str "< " (status t) " | " (pr-str @t) " >")))

;;;

;; the byte layout is
;; [ exists?  : int8
;;   state    : int8
    checksum : int64
;;   size     : int32
;;   payload  : array ]
  valid values for ' exists ' is 0 ( no ) , 1 ( yes )
  valid values for ' state ' is 0 ( unclaimed ) , 1 ( in progress ) , 2 ( complete )
(defn- slab->task-seq
  "Takes a slab, and returns a sequence of the tasks it contains."
  ([slab]
     (slab->task-seq slab 0))
  ([slab ^long pos]
     (with-buffer [buf slab]
       (try
         (let [^ByteBuffer buf' (.position buf pos)]

           ;; is there a next task, and is there space left in the buffer?
           (when (and
                   (<= header-size (.remaining buf'))
                   (== 1 (.get buf')))

             (lazy-seq
               (with-buffer [buf slab]
                 (let [^ByteBuffer buf' (.position buf (p/inc pos))
                       status (.get buf')
                       checksum (.getLong buf')
                       size (.getInt buf')]

                   ;; this shouldn't be necessary, but let's not gratuitously
                   ;; overreach our bounds
                   (when (< size (.remaining buf'))
                     (cons

                       (task
                         slab
                         pos
                         (+ header-size size))

                       (slab->task-seq
                         slab
                         (+ pos header-size size)))))))))
         (catch Throwable e
           ;; this implies unrecoverable corruption
           nil
           )))))

(deftype TaskSlab
  [filename
   q-name
   queue
   buf        ;; a clearable atom holding the buffer
   position   ;; an atom storing the write position of the slab
   lock
   dirty      ;; an atom containing an interval of dirty bytes
   ]
  ITaskSlab

  (read-write-lock [_]
    lock)

  (buffer [this]
    (let [buf (or @buf
                (swap! buf
                  (fn [buf]
                    (or buf (load-buffer filename)))))]
      (.duplicate ^ByteBuffer buf)))

  (mapped? [_]
    (boolean @buf))

  (unmap [_]
    (with-exclusive-lock lock
      (when-let [b @buf]
        (reset! buf nil)
        (unmap-buffer b))))

  (invalidate [_ start' len]
    (let [end' (+ start' len)]
      (swap! dirty
        (fn [[start end]]
          [(min start start') (max end end')]))))

  (sync! [this]
    (let [[start end] @dirty]
      (when (< start end)
        (with-buffer [_ this]
          (let [buf @buf]
            (force-buffer buf start (- end start))
            (compare-and-set! dirty [start end] [Integer/MAX_VALUE 0])
            nil)))))

  (append-to-slab! [this descriptor]
    (with-buffer [buf this]
      (let [ary (nippy/freeze descriptor)
            cnt (count ary)
            pos @position
            ^ByteBuffer buf (.position buf pos)]

        (when (> (.remaining buf) (+ (count ary) header-size))
          ;; write to the buffer
          (doto buf
            (.position pos)
            (.put (byte 1))   ;; exists
            (.put (byte 0))   ;; incomplete
            (.putLong (checksum cnt ary))
            (.putInt cnt)
            (.put ary)
            (.put (byte 0))) ;; next doesn't exist

          (swap! position + header-size cnt)

          (invalidate this pos (+ header-size cnt))

          ;; return a task to enqueue in-memory
          (task
            this
            pos
            (+ header-size cnt))))))

  clojure.lang.Seqable
  (seq [this]
    (slab->task-seq this))

  Comparable
  (compareTo [_ x]
    (assert (instance? TaskSlab x))
    (compare filename (.filename ^TaskSlab x))))

(def ^:private fs-monitor (Object.))

(defn- delete-slab
  [^TaskSlab slab]
  (locking fs-monitor
    (unmap slab)
    (.delete (io/file (.filename slab)))))

(defn- create-slab
  "Creates a new slab file, ensuring a new file name that is lexicographically greater than
   any existing files for that queue name."
  ([directory q-name queue size]
     (locking fs-monitor
       (let [pattern (re-pattern (str "^" q-name "_(\\d{6}$)"))
             last-number (->> directory
                           io/file
                           .listFiles
                           (map #(.getName ^File %))
                           (map #(second (re-find pattern %)))
                           (remove nil?)
                           (map #(Long/parseLong %))
                           sort
                           last)
             n (if last-number (inc last-number) 0)
             f (io/file (str directory "/" q-name "_" (format "%06d" n)))]

         (when-not (.createNewFile f)
           (throw (IOException. (str "Could not create new slab file at " (.getAbsolutePath f)))))

         (TaskSlab.
           (.getAbsolutePath f)
           q-name
           queue
           (atom (create-buffer f size))
           (atom 0)
           (ReentrantReadWriteLock.)
           (atom [Integer/MAX_VALUE 0]))))))

(defn- file->slab
  "Transforms a file into a slab representing that file's contents."
  [filename q-name queue]
  (let [pos (atom 0)
        slab (TaskSlab.
               filename
               q-name
               queue
               (atom nil)
               pos
               (ReentrantReadWriteLock.)
               (atom [Integer/MAX_VALUE 0]))
        len (->> slab
              (map :length)
              (reduce +))]
    (reset! pos len)
    (unmap slab)
    slab))

(defn- directory->queue-name->slab-files
  "Returns a map of queue names onto slab files for that queue."
  [directory]
  (let [queue->file (->> directory
                      io/file
                      .listFiles
                      (filter #(re-find #"^\w+_\d{6}$" (.getName ^File %)))
                      (group-by #(second (re-find #"^(\w+)_\d{6}$" (.getName ^File %)))))]
    (zipmap
      (keys queue->file)
      (map
        (fn [files]
          (->> files
            (map #(.getAbsolutePath ^File %))
            sort))
        (vals queue->file)))))

;;;

(defn- initial-stats [^long count]
  {:enqueued (AtomicLong. count)
   :retried  (AtomicLong. 0)
   :completed (AtomicLong. 0)})

(defn- immediate-stats [^LinkedBlockingQueue q {:keys [enqueued retried completed]}]
  (let [cnt (.size q)
        completed (.get ^AtomicLong completed)
        enqueued (.get ^AtomicLong enqueued)]
    {:enqueued enqueued
     :retried (.get ^AtomicLong retried)
     :completed completed
     :in-progress (- (- enqueued completed) cnt)}))

;;;

(defprotocol IQueues
  (^:private mark-complete! [_ q-name])
  (^:private mark-retry! [_ q-name])
  (delete! [_]
    "Deletes all files associated with the queues.")
  (stats [_]
    "Returns a map of queue names onto information about the immediate state of the queue.")
  (fsync [_]
    "Forces an fsync on all modified files.")
  (take!
    [_ q-name]
    [_ q-name timeout timeout-val]
    "A blocking dequeue from `name`.  If `timeout` is specified, returns `timeout-val` if
     no task is available within `timeout` milliseconds.")
  (put!
    [_ q-name task-descriptor]
    [_ q-name task-descriptor timeout]
    "A blocking enqueue to `name`.  If `timeout` is specified, returns `false` if unable to
     enqueue within `timeout` milliseconds, `true` otherwise."))

(defn queues
  "Creates a point of interaction for queues, backed by disk storage in `directory`.

   The following options can be specified:

       max-queue-size - the maximum number of elements that can be in the queue before `put!`
                        blocks.  Defaults to `Integer/MAX_VALUE`.

       complete? - a predicate that is run on pre-existing tasks to check if they were already
                   completed.  If the tasks in the queue are non-idempotent, this must be
                   specified for correct behavior.  Defaults to always returning false.

       slab-size - The size, in bytes, of the backing files for the queue.  Defaults to 16mb.

       fsync-put? - if true, each `put!` will force an fsync.  Defaults to true.

       fsync-take? - if true, each `take!` will force an fsync.  Defaults to false."
  ([directory]
     (queues directory nil))
  ([directory
    {:keys [max-queue-size
            complete?
            slab-size
            fsync-put?
            fsync-take?
            fsync-threshold
            fsync-interval]
     :or {max-queue-size Integer/MAX_VALUE
          complete? nil
          slab-size (* 64 1024 1024)
          fsync-put? true
          fsync-take? false}}]

     (assert
       (not
         (and
           (or fsync-threshold fsync-interval)
           (or fsync-take? fsync-put?)))
       "Both batch and per-task fsync options are enabled, which is probably not what you intended.")

     (.mkdirs (io/file directory))

     (let [

           queue (memoize (fn [_] (LinkedBlockingQueue. (int max-queue-size))))
           queue-name->files (directory->queue-name->slab-files directory)

           ;; core state stores
           queue-name->slabs (atom
                               (zipmap
                                 (keys queue-name->files)
                                 (->> queue-name->files
                                   (map
                                     (fn [[queue-name files]]
                                       (map #(file->slab % queue-name (queue queue-name)) files)))
                                   vec)))

           queue-name->stats (atom
                               (zipmap
                                 (keys queue-name->files)
                                 (map
                                   #(initial-stats (count (queue %)))
                                   (keys queue-name->files))))

           queue-name->current-slab (atom {})

           ;; initialize
           slabs (->> @queue-name->slabs vals (apply concat))
           slab->count (zipmap
                         slabs
                         (map #(atom (count (seq %))) slabs))
           create-new-slab (fn [q-name]
                             (let [slab (create-slab directory q-name (queue q-name) slab-size)
                                   empty-slabs (->> (@queue-name->slabs q-name)
                                                 (filter (fn [slab]
                                                           (->> slab
                                                             (remove #(= :complete (status %)))
                                                             empty?)))
                                                 set)]

                               ;; delete empty slabs
                               (doseq [s empty-slabs]
                                 (delete-slab s))

                               ;; update list of active slabs
                               (swap! queue-name->slabs update-in [q-name]
                                 #(conj (vec (remove empty-slabs %)) slab))

  unmap all slabs but the first ( which is being consumed )
                               ;; and the last (which is being written to)
                               (doseq [s (-> (@queue-name->slabs q-name) rest butlast)]
                                 (unmap s))
                               slab))

           populate-stats! #(when-not (contains? @queue-name->stats %)
                              (swap! queue-name->stats assoc % (initial-stats 0)))

           this-ref (promise)

           action-counter (AtomicLong. 0)

           mark-action! (if fsync-threshold
                          (fn []
                            (when (zero? (rem (.incrementAndGet action-counter) fsync-threshold))
                              (fsync @this-ref)))
                          (fn []))]

       ;;
       (when fsync-interval
         (future
           (let [ref (WeakReference. @this-ref)]
             (while (.get ref)
               (when-let [q (.get ref)]
                 (try
                   (let [start (System/currentTimeMillis)]
                     (fsync q)
                     (let [end (System/currentTimeMillis)]
                       (Thread/sleep (max 0 (- fsync-interval (- end start))))))
                   (catch Throwable e
                     )))))))

       ;; populate queues with pre-existing tasks
       (let [empty-slabs (atom #{})]
         (doseq [[q slabs] @queue-name->slabs]
           (let [^LinkedBlockingQueue q' (queue q)]
             (doseq [slab slabs]
               (let [tasks (->> slab
                             (map #(vary-meta % assoc
                                     ::this this-ref
                                     ::queue q'
                                     ::queue-name q
                                     ::fsync? fsync-take?))
                             (remove #(or (= :complete (status %))
                                        (and complete? (complete? @%)))))]

                 (if (empty? tasks)

                   ;; if there aren't any active tasks, just delete the slab
                   (do
                     (delete-slab slab)
                     (swap! empty-slabs conj slab))

                   (do
                     (doseq [task tasks]
                       (status! task :incomplete)
                       (when-not (.offer q' task)
                         (throw
                           (IllegalArgumentException.
                             "'max-queue-size' insufficient to hold existing tasks."))))
                     (unmap slab)))))

             (let [^AtomicLong counter (get-in @queue-name->stats [q :enqueued])]
               (.addAndGet counter (count (queue q))))))

         (swap! queue-name->slabs
           (fn [m]
             (->> m
               (map
                 (fn [[q slabs]]
                   [q (remove @empty-slabs slabs)]))
               (into {})))))

       (deliver this-ref
         (reify

           java.io.Closeable
           (close [_]
             (doseq [s (->> @queue-name->slabs vals (apply concat))]
               (unmap s)))

           IQueues

           (delete! [this]
             (doseq [s (->> @queue-name->slabs vals (apply concat))]
               (unmap s)
               (delete-slab s)))

           (fsync [_]
             (doseq [slab (->> @queue-name->slabs vals (apply concat))]
               (sync! slab)))

           (mark-retry! [_ q-name]
             (mark-action!)
             (populate-stats! q-name)
             (let [^AtomicLong retry-counter (get-in @queue-name->stats [q-name :retried])]
               (.incrementAndGet retry-counter)))

           (mark-complete! [_ q-name]
             (mark-action!)
             (populate-stats! q-name)
             (let [^AtomicLong retry-counter (get-in @queue-name->stats [q-name :completed])]
               (.incrementAndGet retry-counter)))

           (stats [_]
             (let [ks (keys @queue-name->stats)]
               (zipmap ks
                 (map
                   (fn [q-name]
                     (merge
                       {:num-slabs (-> @queue-name->slabs (get q-name) count)
                        :num-active-slabs (->> (get @queue-name->slabs q-name)
                                            (filter mapped?)
                                            count)}
                       (immediate-stats (queue q-name) (get @queue-name->stats q-name))))
                   ks))))

           (take! [this q-name timeout timeout-val]
             (let [q-name (munge (name q-name))
                   ^LinkedBlockingQueue q (queue q-name)]
               (try
                 (if-let [t (if (zero? timeout)
                              (.poll q)
                              (.poll q timeout TimeUnit/MILLISECONDS))]

                   (let [slab (:slab t)]

                     ;; if we've moved onto a new slab, unmap all but the current and
                     ;; last slabs
                     (let [old-slab (@queue-name->current-slab q-name)]
                       (when-not (= slab old-slab)
                         (swap! queue-name->current-slab assoc q-name slab)
                         (doseq [s (->> (get @queue-name->slabs q-name)
                                     butlast
                                     (remove #(= slab %)))]
                           (unmap s))))

                     (status! t :in-progress)
                     ;; we don't need to fsync here, because in-progress and incomplete
                     ;; are effectively equivalent on restart

                     t)
                   timeout-val)
                 (catch TimeoutException _
                   timeout-val))))

           (take! [this q-name]
             (take! this q-name Long/MAX_VALUE nil))

           (put! [_ q-name task-descriptor timeout]
             (let [q-name (munge (name q-name))
                   ^LinkedBlockingQueue q (queue q-name)
                   slab! (fn []
                           (let [slabs (@queue-name->slabs q-name)
                                 slab  (last slabs)
                                 task  (when slab
                                         (append-to-slab! slab task-descriptor))

                                 ;; if no task was created, we need to create a new slab file
                                 ;; and try again
                                 slab  (if task
                                         slab
                                         (create-new-slab q-name))
                                 task  (or task (append-to-slab! slab task-descriptor))]

                             (when-not task
                               (throw
                                 (IllegalArgumentException.
                                   (str "Can't enqueue task whose serialized representation is larger than :slab-size, which is currently " slab-size))))

                             (when fsync-put?
                               (sync! slab))
                             task))

                   queue! (fn [task]
                            (if (zero? timeout)
                              (.offer q task)
                              (.offer q task timeout TimeUnit/MILLISECONDS)))]
               (if-let [val (locking q
                              (queue!
                                (vary-meta (slab!) assoc
                                  ::this this-ref
                                  ::queue-name q-name
                                  ::queue q
                                  ::fsync? fsync-take?)))]
                 (do
                   (populate-stats! q-name)
                   (let [^AtomicLong counter (get-in @queue-name->stats [q-name :enqueued])]
                     (.incrementAndGet counter))
                   true)
                 false)))

           (put! [this q-name task-descriptor]
             (put! this q-name task-descriptor Long/MAX_VALUE))))

       @this-ref)))

;;;

(defn task-seq
  "Returns an infinite lazy sequence of tasks for `q-name`."
  [qs q-name]
  (lazy-seq
    (cons
      (take! qs q-name)
      (task-seq qs q-name))))

(defn immediate-task-seq
  "Returns a finite lazy sequence of tasks for `q-name` which terminates once there are
   no more tasks immediately available."
  [qs q-name]
  (lazy-seq
    (let [task (take! qs q-name 0 ::none)]
      (when-not (= ::none task)
        (cons
          task
          (immediate-task-seq qs q-name))))))

(defn interval-task-seq
  "Returns a lazy sequence of tasks that can be consumed in `interval` milliseconds.  This will
   terminate after that time has elapsed, even if there are still tasks immediately available."
  [qs q-name interval]
  (let [now (System/currentTimeMillis)]
    (lazy-seq
      (let [now' (System/currentTimeMillis)
            remaining (- interval (- now' now))]
        (when (pos? remaining)
          (let [task (take! qs q-name remaining ::none)]
            (when-not (= ::none task)
              (cons
                task
                (interval-task-seq qs q-name (- interval (- (System/currentTimeMillis) now)))))))))))

(defn complete!
  "Marks a task as complete."
  [task]
  (if (identical? :complete (status task))
    false
    (do
      (status! task :complete)
      (when (-> task meta ::fsync?)
        (sync! (:slab task)))
      (mark-complete! @(-> task meta ::this) (-> task meta ::queue-name))
      true)))

(defn retry!
  "Marks a task as available for retry."
  [task]
  (if (or
        (identical? :complete (status task))
        (identical? :incomplete (status task)))
    false
    (do
      (status! task :incomplete)
      (when (-> task meta ::fsync?)
        (sync! (:slab task)))
      (mark-retry! @(-> task meta ::this) (-> task meta ::queue-name))
      (let [^LinkedBlockingQueue q (-> task meta ::queue)]
        (.put q task))
      true)))





 not much we can do here, sadly

 a single task within a slab, assumes that the buffer is sliced around
 the task's boundaries

 the byte layout is
 [ exists?  : int8
   state    : int8
   size     : int32
   payload  : array ]
 is there a next task, and is there space left in the buffer?
 this shouldn't be necessary, but let's not gratuitously
 overreach our bounds
 this implies unrecoverable corruption
 a clearable atom holding the buffer
 an atom storing the write position of the slab
 an atom containing an interval of dirty bytes
 write to the buffer
 exists
 incomplete
 next doesn't exist
 return a task to enqueue in-memory


 core state stores
 initialize
 delete empty slabs
 update list of active slabs
 and the last (which is being written to)

 populate queues with pre-existing tasks
 if there aren't any active tasks, just delete the slab
 if we've moved onto a new slab, unmap all but the current and
 last slabs
 we don't need to fsync here, because in-progress and incomplete
 are effectively equivalent on restart
 if no task was created, we need to create a new slab file
 and try again

(ns durable-queue
  (:require
    [clojure.java.io :as io]
    [byte-streams :as bs]
    [clojure.string :as str]
    [primitive-math :as p]
    [taoensso.nippy :as nippy])
  (:import
    [java.lang.reflect
     Method
     Field]
    [java.util.concurrent
     LinkedBlockingQueue
     TimeoutException
     TimeUnit]
    [java.util.concurrent.atomic
     AtomicLong]
    [java.util.zip
     CRC32]
    [java.util.concurrent.locks
     ReentrantReadWriteLock]
    [java.io
     Writer
     File
     RandomAccessFile
     IOException]
    [java.nio.channels
     FileChannel
     FileChannel$MapMode]
    [java.nio
     ByteBuffer
     MappedByteBuffer]
    [java.lang.ref
     WeakReference]))


(defmacro ^:private with-lock [lock & body]
  `(let [^ReentrantReadWriteLock lock# ~lock
         read-lock# (.readLock lock#)]
     (do
       (.lock read-lock#)
       (try
         ~@body
         (finally
           (.unlock read-lock#))))))

(defmacro ^:private with-exclusive-lock [lock & body]
  `(let [^ReentrantReadWriteLock lock# ~lock
         write-lock# (.writeLock lock#)]
     (do
       (.lock write-lock#)
       (try
         ~@body
         (finally
           (.unlock write-lock#))))))


(defn- checksum ^long [^long length ^bytes ary]
  (let [crc (CRC32.)]
    (dotimes [i 4]
      (.update crc (p/>> length i)))
    (.update crc ary)
    (.getValue crc)))


(def ^:private ^:const header-size 14)

(defprotocol ITask
  (^:private status [_] "Returns the task status")
  (^:private status! [_ status] "Sets the task status"))

(defprotocol ITaskSlab
  (^:private unmap [_] "Temporarily releases mapped byte buffer until it's needed again.")
  (^:private mapped? [_] "Returns true if the slab is actively mapped into memory.")
  (^:private sync! [_])
  (^:private invalidate [_ offset len])
  (^:private ^ByteBuffer buffer [_])
  (^:private append-to-slab! [_ descriptor])
  (^:private read-write-lock [_]))

(defmacro ^:private with-buffer [[buf slab] & body]
  `(with-lock (read-write-lock ~slab)
     (when-let [~buf (buffer ~slab)]
       ~@body)))


(defn create-buffer [filename size]
  (let [raf (doto (RandomAccessFile. (io/file filename) "rw")
              (.setLength size))]
    (try
      (let [fc (.getChannel raf)]
        (try
          (let [buf (.map fc FileChannel$MapMode/READ_WRITE 0 size)]
            (doto buf
              (.put 0 (byte 0))
              .force))
          (finally
            (.close fc))))
      (finally
        (.close raf)))))

(defn load-buffer
  ([filename]
     (load-buffer filename nil nil))
  ([filename offset length]
     (let [_ (assert (.exists (io/file filename)))
           raf (RandomAccessFile. (io/file filename) "rw")]
       (try
         (let [fc (.getChannel raf)]
           (try
             (.map fc
               FileChannel$MapMode/READ_WRITE
               (or offset 0)
               (or length (.length raf)))
             (finally
               (.close fc))))
         (finally
           (.close raf))))))

(let [clean (delay
              (doto (.getMethod
                      (Class/forName "sun.misc.Cleaner")
                      "clean"
                      nil)
                (.setAccessible true)))]
  (defn- unmap-buffer
    "A delightful endrun on the JVM's mmap GC mechanism"
    [^ByteBuffer buf]
    (when (.isDirect buf)
      (try

        (let [^Method clean @clean
              cleaner (doto (.getMethod (class buf) "cleaner" nil)
                        (.setAccessible true))]
          (.invoke clean
            (.invoke cleaner buf nil)
            nil))
        (catch Throwable e
          )))))

(defn- force-buffer
  [^MappedByteBuffer buf offset length]
  (.force buf))


(defrecord Task
  [slab
   ^long offset
   ^long length
   status
   deserializer]
  clojure.lang.IDeref
  (deref [_]
    (deserializer))
  ITask
  (status [_]
    (with-buffer [buf slab]
      (or @status
        (let [s (case (.get buf (p/+ offset 1))
                  0 :incomplete
                  1 :in-progress
                  2 :complete)]
          (reset! status s)
          s))))
  (status! [_ s]
    (with-buffer [buf slab]
      (reset! status s)
      (.put buf (p/+ offset 1)
        (case s
          :incomplete 0
          :in-progress 1
          :complete 2))
      (invalidate slab (p/+ offset 1) 1)
      nil)))

(defn- task [slab offset len]
  (Task.
    slab
    offset
    len
    (atom nil)
    (fn []
      (with-buffer [buf slab]
        (let [^ByteBuffer buf (-> buf
                                (.position offset)
                                ^ByteBuffer
                                (.limit (+ offset len))
                                .slice)
              checksum' (.getLong buf 2)
              ary (bs/to-byte-array (.position buf header-size))]
          (when-not (== (checksum (.getInt buf 10) ary) checksum')
            (throw (IOException. "checksum mismatch")))
          (nippy/thaw ary))))))

(defmethod print-method Task [t ^Writer w]
  (.write w
    (str "< " (status t) " | " (pr-str @t) " >")))


    checksum : int64
  valid values for ' exists ' is 0 ( no ) , 1 ( yes )
  valid values for ' state ' is 0 ( unclaimed ) , 1 ( in progress ) , 2 ( complete )
(defn- slab->task-seq
  "Takes a slab, and returns a sequence of the tasks it contains."
  ([slab]
     (slab->task-seq slab 0))
  ([slab ^long pos]
     (with-buffer [buf slab]
       (try
         (let [^ByteBuffer buf' (.position buf pos)]

           (when (and
                   (<= header-size (.remaining buf'))
                   (== 1 (.get buf')))

             (lazy-seq
               (with-buffer [buf slab]
                 (let [^ByteBuffer buf' (.position buf (p/inc pos))
                       status (.get buf')
                       checksum (.getLong buf')
                       size (.getInt buf')]

                   (when (< size (.remaining buf'))
                     (cons

                       (task
                         slab
                         pos
                         (+ header-size size))

                       (slab->task-seq
                         slab
                         (+ pos header-size size)))))))))
         (catch Throwable e
           nil
           )))))

(deftype TaskSlab
  [filename
   q-name
   queue
   lock
   ]
  ITaskSlab

  (read-write-lock [_]
    lock)

  (buffer [this]
    (let [buf (or @buf
                (swap! buf
                  (fn [buf]
                    (or buf (load-buffer filename)))))]
      (.duplicate ^ByteBuffer buf)))

  (mapped? [_]
    (boolean @buf))

  (unmap [_]
    (with-exclusive-lock lock
      (when-let [b @buf]
        (reset! buf nil)
        (unmap-buffer b))))

  (invalidate [_ start' len]
    (let [end' (+ start' len)]
      (swap! dirty
        (fn [[start end]]
          [(min start start') (max end end')]))))

  (sync! [this]
    (let [[start end] @dirty]
      (when (< start end)
        (with-buffer [_ this]
          (let [buf @buf]
            (force-buffer buf start (- end start))
            (compare-and-set! dirty [start end] [Integer/MAX_VALUE 0])
            nil)))))

  (append-to-slab! [this descriptor]
    (with-buffer [buf this]
      (let [ary (nippy/freeze descriptor)
            cnt (count ary)
            pos @position
            ^ByteBuffer buf (.position buf pos)]

        (when (> (.remaining buf) (+ (count ary) header-size))
          (doto buf
            (.position pos)
            (.putLong (checksum cnt ary))
            (.putInt cnt)
            (.put ary)

          (swap! position + header-size cnt)

          (invalidate this pos (+ header-size cnt))

          (task
            this
            pos
            (+ header-size cnt))))))

  clojure.lang.Seqable
  (seq [this]
    (slab->task-seq this))

  Comparable
  (compareTo [_ x]
    (assert (instance? TaskSlab x))
    (compare filename (.filename ^TaskSlab x))))

(def ^:private fs-monitor (Object.))

(defn- delete-slab
  [^TaskSlab slab]
  (locking fs-monitor
    (unmap slab)
    (.delete (io/file (.filename slab)))))

(defn- create-slab
  "Creates a new slab file, ensuring a new file name that is lexicographically greater than
   any existing files for that queue name."
  ([directory q-name queue size]
     (locking fs-monitor
       (let [pattern (re-pattern (str "^" q-name "_(\\d{6}$)"))
             last-number (->> directory
                           io/file
                           .listFiles
                           (map #(.getName ^File %))
                           (map #(second (re-find pattern %)))
                           (remove nil?)
                           (map #(Long/parseLong %))
                           sort
                           last)
             n (if last-number (inc last-number) 0)
             f (io/file (str directory "/" q-name "_" (format "%06d" n)))]

         (when-not (.createNewFile f)
           (throw (IOException. (str "Could not create new slab file at " (.getAbsolutePath f)))))

         (TaskSlab.
           (.getAbsolutePath f)
           q-name
           queue
           (atom (create-buffer f size))
           (atom 0)
           (ReentrantReadWriteLock.)
           (atom [Integer/MAX_VALUE 0]))))))

(defn- file->slab
  "Transforms a file into a slab representing that file's contents."
  [filename q-name queue]
  (let [pos (atom 0)
        slab (TaskSlab.
               filename
               q-name
               queue
               (atom nil)
               pos
               (ReentrantReadWriteLock.)
               (atom [Integer/MAX_VALUE 0]))
        len (->> slab
              (map :length)
              (reduce +))]
    (reset! pos len)
    (unmap slab)
    slab))

(defn- directory->queue-name->slab-files
  "Returns a map of queue names onto slab files for that queue."
  [directory]
  (let [queue->file (->> directory
                      io/file
                      .listFiles
                      (filter #(re-find #"^\w+_\d{6}$" (.getName ^File %)))
                      (group-by #(second (re-find #"^(\w+)_\d{6}$" (.getName ^File %)))))]
    (zipmap
      (keys queue->file)
      (map
        (fn [files]
          (->> files
            (map #(.getAbsolutePath ^File %))
            sort))
        (vals queue->file)))))


(defn- initial-stats [^long count]
  {:enqueued (AtomicLong. count)
   :retried  (AtomicLong. 0)
   :completed (AtomicLong. 0)})

(defn- immediate-stats [^LinkedBlockingQueue q {:keys [enqueued retried completed]}]
  (let [cnt (.size q)
        completed (.get ^AtomicLong completed)
        enqueued (.get ^AtomicLong enqueued)]
    {:enqueued enqueued
     :retried (.get ^AtomicLong retried)
     :completed completed
     :in-progress (- (- enqueued completed) cnt)}))


(defprotocol IQueues
  (^:private mark-complete! [_ q-name])
  (^:private mark-retry! [_ q-name])
  (delete! [_]
    "Deletes all files associated with the queues.")
  (stats [_]
    "Returns a map of queue names onto information about the immediate state of the queue.")
  (fsync [_]
    "Forces an fsync on all modified files.")
  (take!
    [_ q-name]
    [_ q-name timeout timeout-val]
    "A blocking dequeue from `name`.  If `timeout` is specified, returns `timeout-val` if
     no task is available within `timeout` milliseconds.")
  (put!
    [_ q-name task-descriptor]
    [_ q-name task-descriptor timeout]
    "A blocking enqueue to `name`.  If `timeout` is specified, returns `false` if unable to
     enqueue within `timeout` milliseconds, `true` otherwise."))

(defn queues
  "Creates a point of interaction for queues, backed by disk storage in `directory`.

   The following options can be specified:

       max-queue-size - the maximum number of elements that can be in the queue before `put!`
                        blocks.  Defaults to `Integer/MAX_VALUE`.

       complete? - a predicate that is run on pre-existing tasks to check if they were already
                   completed.  If the tasks in the queue are non-idempotent, this must be
                   specified for correct behavior.  Defaults to always returning false.

       slab-size - The size, in bytes, of the backing files for the queue.  Defaults to 16mb.

       fsync-put? - if true, each `put!` will force an fsync.  Defaults to true.

       fsync-take? - if true, each `take!` will force an fsync.  Defaults to false."
  ([directory]
     (queues directory nil))
  ([directory
    {:keys [max-queue-size
            complete?
            slab-size
            fsync-put?
            fsync-take?
            fsync-threshold
            fsync-interval]
     :or {max-queue-size Integer/MAX_VALUE
          complete? nil
          slab-size (* 64 1024 1024)
          fsync-put? true
          fsync-take? false}}]

     (assert
       (not
         (and
           (or fsync-threshold fsync-interval)
           (or fsync-take? fsync-put?)))
       "Both batch and per-task fsync options are enabled, which is probably not what you intended.")

     (.mkdirs (io/file directory))

     (let [

           queue (memoize (fn [_] (LinkedBlockingQueue. (int max-queue-size))))
           queue-name->files (directory->queue-name->slab-files directory)

           queue-name->slabs (atom
                               (zipmap
                                 (keys queue-name->files)
                                 (->> queue-name->files
                                   (map
                                     (fn [[queue-name files]]
                                       (map #(file->slab % queue-name (queue queue-name)) files)))
                                   vec)))

           queue-name->stats (atom
                               (zipmap
                                 (keys queue-name->files)
                                 (map
                                   #(initial-stats (count (queue %)))
                                   (keys queue-name->files))))

           queue-name->current-slab (atom {})

           slabs (->> @queue-name->slabs vals (apply concat))
           slab->count (zipmap
                         slabs
                         (map #(atom (count (seq %))) slabs))
           create-new-slab (fn [q-name]
                             (let [slab (create-slab directory q-name (queue q-name) slab-size)
                                   empty-slabs (->> (@queue-name->slabs q-name)
                                                 (filter (fn [slab]
                                                           (->> slab
                                                             (remove #(= :complete (status %)))
                                                             empty?)))
                                                 set)]

                               (doseq [s empty-slabs]
                                 (delete-slab s))

                               (swap! queue-name->slabs update-in [q-name]
                                 #(conj (vec (remove empty-slabs %)) slab))

  unmap all slabs but the first ( which is being consumed )
                               (doseq [s (-> (@queue-name->slabs q-name) rest butlast)]
                                 (unmap s))
                               slab))

           populate-stats! #(when-not (contains? @queue-name->stats %)
                              (swap! queue-name->stats assoc % (initial-stats 0)))

           this-ref (promise)

           action-counter (AtomicLong. 0)

           mark-action! (if fsync-threshold
                          (fn []
                            (when (zero? (rem (.incrementAndGet action-counter) fsync-threshold))
                              (fsync @this-ref)))
                          (fn []))]

       (when fsync-interval
         (future
           (let [ref (WeakReference. @this-ref)]
             (while (.get ref)
               (when-let [q (.get ref)]
                 (try
                   (let [start (System/currentTimeMillis)]
                     (fsync q)
                     (let [end (System/currentTimeMillis)]
                       (Thread/sleep (max 0 (- fsync-interval (- end start))))))
                   (catch Throwable e
                     )))))))

       (let [empty-slabs (atom #{})]
         (doseq [[q slabs] @queue-name->slabs]
           (let [^LinkedBlockingQueue q' (queue q)]
             (doseq [slab slabs]
               (let [tasks (->> slab
                             (map #(vary-meta % assoc
                                     ::this this-ref
                                     ::queue q'
                                     ::queue-name q
                                     ::fsync? fsync-take?))
                             (remove #(or (= :complete (status %))
                                        (and complete? (complete? @%)))))]

                 (if (empty? tasks)

                   (do
                     (delete-slab slab)
                     (swap! empty-slabs conj slab))

                   (do
                     (doseq [task tasks]
                       (status! task :incomplete)
                       (when-not (.offer q' task)
                         (throw
                           (IllegalArgumentException.
                             "'max-queue-size' insufficient to hold existing tasks."))))
                     (unmap slab)))))

             (let [^AtomicLong counter (get-in @queue-name->stats [q :enqueued])]
               (.addAndGet counter (count (queue q))))))

         (swap! queue-name->slabs
           (fn [m]
             (->> m
               (map
                 (fn [[q slabs]]
                   [q (remove @empty-slabs slabs)]))
               (into {})))))

       (deliver this-ref
         (reify

           java.io.Closeable
           (close [_]
             (doseq [s (->> @queue-name->slabs vals (apply concat))]
               (unmap s)))

           IQueues

           (delete! [this]
             (doseq [s (->> @queue-name->slabs vals (apply concat))]
               (unmap s)
               (delete-slab s)))

           (fsync [_]
             (doseq [slab (->> @queue-name->slabs vals (apply concat))]
               (sync! slab)))

           (mark-retry! [_ q-name]
             (mark-action!)
             (populate-stats! q-name)
             (let [^AtomicLong retry-counter (get-in @queue-name->stats [q-name :retried])]
               (.incrementAndGet retry-counter)))

           (mark-complete! [_ q-name]
             (mark-action!)
             (populate-stats! q-name)
             (let [^AtomicLong retry-counter (get-in @queue-name->stats [q-name :completed])]
               (.incrementAndGet retry-counter)))

           (stats [_]
             (let [ks (keys @queue-name->stats)]
               (zipmap ks
                 (map
                   (fn [q-name]
                     (merge
                       {:num-slabs (-> @queue-name->slabs (get q-name) count)
                        :num-active-slabs (->> (get @queue-name->slabs q-name)
                                            (filter mapped?)
                                            count)}
                       (immediate-stats (queue q-name) (get @queue-name->stats q-name))))
                   ks))))

           (take! [this q-name timeout timeout-val]
             (let [q-name (munge (name q-name))
                   ^LinkedBlockingQueue q (queue q-name)]
               (try
                 (if-let [t (if (zero? timeout)
                              (.poll q)
                              (.poll q timeout TimeUnit/MILLISECONDS))]

                   (let [slab (:slab t)]

                     (let [old-slab (@queue-name->current-slab q-name)]
                       (when-not (= slab old-slab)
                         (swap! queue-name->current-slab assoc q-name slab)
                         (doseq [s (->> (get @queue-name->slabs q-name)
                                     butlast
                                     (remove #(= slab %)))]
                           (unmap s))))

                     (status! t :in-progress)

                     t)
                   timeout-val)
                 (catch TimeoutException _
                   timeout-val))))

           (take! [this q-name]
             (take! this q-name Long/MAX_VALUE nil))

           (put! [_ q-name task-descriptor timeout]
             (let [q-name (munge (name q-name))
                   ^LinkedBlockingQueue q (queue q-name)
                   slab! (fn []
                           (let [slabs (@queue-name->slabs q-name)
                                 slab  (last slabs)
                                 task  (when slab
                                         (append-to-slab! slab task-descriptor))

                                 slab  (if task
                                         slab
                                         (create-new-slab q-name))
                                 task  (or task (append-to-slab! slab task-descriptor))]

                             (when-not task
                               (throw
                                 (IllegalArgumentException.
                                   (str "Can't enqueue task whose serialized representation is larger than :slab-size, which is currently " slab-size))))

                             (when fsync-put?
                               (sync! slab))
                             task))

                   queue! (fn [task]
                            (if (zero? timeout)
                              (.offer q task)
                              (.offer q task timeout TimeUnit/MILLISECONDS)))]
               (if-let [val (locking q
                              (queue!
                                (vary-meta (slab!) assoc
                                  ::this this-ref
                                  ::queue-name q-name
                                  ::queue q
                                  ::fsync? fsync-take?)))]
                 (do
                   (populate-stats! q-name)
                   (let [^AtomicLong counter (get-in @queue-name->stats [q-name :enqueued])]
                     (.incrementAndGet counter))
                   true)
                 false)))

           (put! [this q-name task-descriptor]
             (put! this q-name task-descriptor Long/MAX_VALUE))))

       @this-ref)))


(defn task-seq
  "Returns an infinite lazy sequence of tasks for `q-name`."
  [qs q-name]
  (lazy-seq
    (cons
      (take! qs q-name)
      (task-seq qs q-name))))

(defn immediate-task-seq
  "Returns a finite lazy sequence of tasks for `q-name` which terminates once there are
   no more tasks immediately available."
  [qs q-name]
  (lazy-seq
    (let [task (take! qs q-name 0 ::none)]
      (when-not (= ::none task)
        (cons
          task
          (immediate-task-seq qs q-name))))))

(defn interval-task-seq
  "Returns a lazy sequence of tasks that can be consumed in `interval` milliseconds.  This will
   terminate after that time has elapsed, even if there are still tasks immediately available."
  [qs q-name interval]
  (let [now (System/currentTimeMillis)]
    (lazy-seq
      (let [now' (System/currentTimeMillis)
            remaining (- interval (- now' now))]
        (when (pos? remaining)
          (let [task (take! qs q-name remaining ::none)]
            (when-not (= ::none task)
              (cons
                task
                (interval-task-seq qs q-name (- interval (- (System/currentTimeMillis) now)))))))))))

(defn complete!
  "Marks a task as complete."
  [task]
  (if (identical? :complete (status task))
    false
    (do
      (status! task :complete)
      (when (-> task meta ::fsync?)
        (sync! (:slab task)))
      (mark-complete! @(-> task meta ::this) (-> task meta ::queue-name))
      true)))

(defn retry!
  "Marks a task as available for retry."
  [task]
  (if (or
        (identical? :complete (status task))
        (identical? :incomplete (status task)))
    false
    (do
      (status! task :incomplete)
      (when (-> task meta ::fsync?)
        (sync! (:slab task)))
      (mark-retry! @(-> task meta ::this) (-> task meta ::queue-name))
      (let [^LinkedBlockingQueue q (-> task meta ::queue)]
        (.put q task))
      true)))
