lil-driver.clj

(require '[datomic.api :as d]
         '[clojure.stacktrace :as st])

;; Knobs
(def datoms-per-tx 1000)

;; Monitoring
(def start-time (atom 0))
(def transaction-count (atom 0))


(defn partition-txs
  "Given a seq of transactions, figure out how many datoms in the first
  transaction, and then partition to ensure each group of transactions
  has *datoms-per-tx* datoms in it."
  [txs]
  (let [representative    (first txs)
        tx-size           (reduce + (map count representative))
        txs-per-partition (max (quot datoms-per-tx tx-size) 1)]
    (prn {:op                :partition-txs
          :txs-per-partition txs-per-partition})
    (partition-all txs-per-partition txs)))
    
(defn load-generated-data
  "Call a generator to create entities. Load num-samples of them into
   the database, using parallel threads for better throughput."
  [conn generator num-samples]
  (let [db (d/db conn)]
    (reset! start-time (System/nanoTime))
    (prn {:op :start})
    (doseq [{:keys [tx result counter start tod]}
            (pmap #(let [ctr (swap! transaction-count inc)
                         tx  (apply concat %)]
                     (try
                       {:tx      tx
                        :start   (. System (nanoTime))
                        :result  (d/transact-async conn tx)
                        :counter ctr}
                       (catch Throwable e
                         {:result  (delay (throw e))
                          :counter ctr
                          :tx      tx})))
                  (partition-txs (repeatedly num-samples generator)))]
      (try
        (when @result
          (prn {:op         :tx
                :ctr        counter
                :elapsed-ms (quot (- (System/nanoTime) start)
                                  1000000)}))
        (catch Throwable e
          (prn {:op   :tx
                :ctr  counter
                :err  (st/root-cause e)
                :tx   tx})
          (when tx
            @(d/transact-async conn tx)))))
    (let [scheduled?  (d/request-index conn)]
      (prn {:op :final-index
            :scheduled? scheduled?}))
    (prn {:op :end
          :elapsed-ms (quot (- (System/nanoTime) @start-time)
                            1000000)})))