scylla-bench.core.clj

(ns scylla-bench.core
  (:require [clojurewerkz.cassaforte.client :as client]
            [clojurewerkz.cassaforte.cql    :as cql]
            [clojurewerkz.cassaforte.query  :as dsl]))

(def STEP-SIZE 1000)
(def counter (agent [nil, 0]))

(defn connect []
  (client/connect
   ["192.168.50.4"]
   ;; ["127.0.0.1"]
   {:keyspace "Clojure" :protocol-version 3}))

(defn prepare []
  (let [connection (connect)]
    (cql/create-table connection :msisdns
                      (dsl/column-definitions {:msisdn :bigint,
                                               :uid :varchar,
                                               :primary-key [:msisdn]}))))

(defn step [time1 count]
  (if time1
    (let [time2 (System/currentTimeMillis)
          delta (- time2 time1)
          tps (* 1000 (/ STEP-SIZE delta))]
      (do
        (prn (str "Done: " count " Elapsed: " delta "ms TPS: " (float tps)))
        time2))
    (System/currentTimeMillis)))

(defn save-request [[time, value]]
  (if (zero? (mod value STEP-SIZE))
    [(step time value), (inc value)]
    [time, (inc value)]))

(defn send-requests [num-requests]
  (let [connection (connect)]
    (dotimes [n num-requests] (do
                                (cql/insert connection :msisdns {:msisdn (rand-int 79999999),
                                                                 :uid (str n)})
                                (send counter save-request)))))

(defn benchmark [num-workers num-requests]
  (do
    (send counter (fn [&_] [nil, 0]))
    (let [requests-per-worker (quot num-requests num-workers)
          futures (for [_ (range num-workers)]
                    (future (send-requests requests-per-worker)))]
      (doseq [f futures] @f))))