diff --git a/project.clj b/project.clj index a127613..3f09a9d 100644 --- a/project.clj +++ b/project.clj @@ -5,14 +5,19 @@ :url "http://www.eclipse.org/legal/epl-v10.html"} :dependencies [[org.clojure/clojure "1.9.0"] [org.clojure/core.async "0.4.474"] + [org.clojure/math.combinatorics "0.1.4"] ;; web [aero "1.1.3"] [http-kit "2.3.0"] [compojure "1.6.1"] + [com.taoensso/nippy "2.14.0"] [javax.servlet/javax.servlet-api "3.1.0"] + [org.apache.kafka/kafka-streams "1.1.0"] + [org.apache.kafka/kafka-clients "1.1.0"] ;;utils [slingshot "0.12.2"] + [spootnik/kinsky "0.1.22"] [cheshire "5.8.0"]] :java-source-paths ["src/java"] :main ^:skip-aot skirmish.core @@ -21,5 +26,8 @@ :profiles { :uberjar {:aot :all} :dev {:jvm-opts ["-XX:-OmitStackTraceInFastThrow"] - :dependencies [[midje "1.9.1"]]}}) + :dependencies [[midje "1.9.1"] + [stylefruits/gniazdo "1.0.1"] + [com.stuartsierra/frequencies "0.1.0"] + [clj-http "3.9.0"]]}}) diff --git a/resources/config.edn b/resources/config.edn index 1492e47..cf200e0 100644 --- a/resources/config.edn +++ b/resources/config.edn @@ -1,12 +1,23 @@ { + :kafka + {:bootstrap.servers ["localhost:9092"] + :topics-prefix "skirmish-01-"} + + + :webserver ;; possible config from http-kit's run-server - {:port #long #or [#env PORT 8080] - :ip "0.0.0.0" - :max-body 524288 ;; 512kb, small but enough - :max-ws 524288} + {:port #long #or [#env PORT 8080] + :ip "0.0.0.0" + :max-body 524288 ;; 512kb, small but enough + :max-ws 524288} :securities ;; list of tradable securities - [{:sec-id :foo} - {:sec-id :bar} - {:sec-id :toto} - {:sec-id :tata}]} \ No newline at end of file + [{:sec-id :foo} + {:sec-id :bar} + {:sec-id :toto} + {:sec-id :tata}] + :benchmark { + :number-of-requests-per-securities 30}} + + + diff --git a/src/skirmish/cli/benchmark.clj b/src/skirmish/cli/benchmark.clj new file mode 100644 index 0000000..5e52996 --- /dev/null +++ b/src/skirmish/cli/benchmark.clj @@ -0,0 +1,116 @@ +(ns skirmish.cli.benchmark + (:require [clojure.math.combinatorics :refer [cartesian-product]] + [clojure.core.async :refer [go chan ! close! to-chan close!] :as async] + [gniazdo.core :as ws] + [clj-http.client :as client] + [com.stuartsierra.frequencies :as freq] + [clojure.edn :as edn])) + + +(def ids (atom 0)) + + +(defn web-socket-to-channel [url output-ch] + (let [on-receive #(go (>! output-ch %)) + headers {"accept" "application/edn"}] + (ws/connect url :headers headers :on-receive on-receive))) + + +(defn create-trades-channel [securities] + (let [output (chan 1 (map edn/read-string)) + url #(format "ws://localhost:%s/api/1/securities/%s/trades" 8080 %) + xf (comp (map :sec-id) + (map name) + (map url) + (map #(web-socket-to-channel % output)))] + (doall (sequence xf securities)) + (async/pub output :book.order/order-id))) + + +(defn create-orders-channel [securities] + (let [output (chan 1 (map edn/read-string)) + url #(format "ws://localhost:%s/api/1/securities/%s/orders" 8080 %) + xf (comp (map :sec-id) + (map name) + (map url) + (map #(web-socket-to-channel % output)))] + (doall (sequence xf securities)) + (async/pub output :book.order/id))) + + + +(defn create-requests-payload-for-security [i security id] + (let [rd (new java.util.Random i) + rand-bool (= (mod (. rd nextInt) 2) 0) ;; java random number generator REALLY sucks + type (if rand-bool :sell :buy) + price (+ (* 3 (. rd nextGaussian)) 30) + int-price (int (* price 100)) + quantity (reduce + 1 (map #(. rd nextInt (inc %)) (range 100)))] + + {:type type :quantity quantity :price int-price :security security :benchmark/id id})) + + +(defn create-requests-for-security [[i security]] + (let [req-id (swap! ids inc) + payload (create-requests-payload-for-security i (str (:sec-id security)) req-id) + sec-id-str (-> security :sec-id name) + params {:throw-exceptions false :content-type :edn :headers {"accept" "application/edn"}} + url (format "http://localhost:%s/api/1/securities/%s" 8080 sec-id-str)] + (fn [] + (let [now (System/nanoTime)] + (client/post url (assoc params :form-params (assoc payload :benchmark/request-time now))))))) + + +(defn make-requests [n securities] + (map create-requests-for-security (cartesian-product (range n) securities))) + + +(defn no-more-then-n-simultaneously [f input-chan n] + (let [buffer (chan)] + (go (async/pipeline n buffer (map f) input-chan)) + buffer)) + +(defn decode-response [response] + (let [now (System/nanoTime) + ret {} + ret (assoc ret :status (:status response)) + ret (assoc ret :order (edn/read-string (:body response))) + ret (assoc ret :duration (/ (- now (get-in ret [:order :benchmark/request-time])) (Math/pow 10 9)))] + ret)) + + +(defn benchmark [requests order trades] + (let [decoded-responses (chan 1 (map decode-response)) + responses (no-more-then-n-simultaneously #(%) (to-chan requests) 10)] + (go (async/pipe responses decoded-responses)) + decoded-responses)) + + +(defn compute-result [results new-result] + (let [status-code (:status new-result) + duration (:duration new-result) + old-status-code-count (get-in results [:status status-code] 0) + ret results + ret (assoc-in ret [:status status-code] (inc old-status-code-count)) + ret (assoc ret :durations (conj (:durations ret) duration))] + ret)) + +(defn pp-durations [durations] + (let [freq-map (frequencies durations)] + (print (freq/stats freq-map)))) + +(defn display-results [results] + (println results) + (println "number of request: " (:n results)) + (println "status map: " (:status results)) + (pp-durations (:durations results))) + +(defn -main [config] + (let [benchmark-config (:benchmark config) + requests (-> benchmark-config :number-of-requests-per-securities (make-requests (:securities config))) + trades (create-trades-channel (:securities config)) + orders (create-orders-channel (:securities config)) + results (benchmark requests orders trades)] + (display-results (async/ "config.edn" clojure.java.io/resource read-config) diff --git a/src/skirmish/rest/marketdata.clj b/src/skirmish/rest/marketdata.clj index cb203ea..48de8d5 100644 --- a/src/skirmish/rest/marketdata.clj +++ b/src/skirmish/rest/marketdata.clj @@ -1 +1,13 @@ -(ns skirmish.rest.marketdata) +(ns skirmish.rest.marketdata + (:require [compojure.core :refer :all])) + + +(defn create-routes [exchange-apis] + (routes + (context "/api/1" [] + (GET "/securities" [] ) + (GET "/securities/:sec-id" [] (get-security exchange-apis)) + (POST "/securities/:sec-id" [] (create-order exchange-apis)) + (GET "/securities/:sec-id/trades" [] (listen-to-trades exchange-apis)) + (GET "/securities/:sec-id/orders" [] (listen-to-orders exchange-apis))) + (not-found "not found"))) diff --git a/test/skirmish/cli/benchmark_test.clj b/test/skirmish/cli/benchmark_test.clj new file mode 100644 index 0000000..0a86ba2 --- /dev/null +++ b/test/skirmish/cli/benchmark_test.clj @@ -0,0 +1,30 @@ +(ns skirmish.cli.benchmark-test + (:require [skirmish.cli.benchmark :refer :all] + [midje.sweet :refer :all] + [clojure.core.async :as async])) + +(facts "creating simple orders" + (fact "buy order" + (let [buy-order (create-requests-payload-for-security 666 (str :equity/ibm) 1) + expt {:benchmark/id 1 :price 2922 :quantity 2476 :security ":equity/ibm" :type :buy}] + buy-order => expt));; + (fact "sell order" + (let [buy-order (create-requests-payload-for-security 667 (str :equity/ibm) 2) + expt {:benchmark/id 2 :price 2740 :quantity 2516 :security ":equity/ibm" :type :sell}] + buy-order => expt))) + +(facts "the utility functions" + (fact "no-more-then-n-simultaneously" + (let [simultatious (atom []) + input-chan (async/to-chan (range 10)) + sleep-time #(+ (mod (+ (* 5142 %) 8765) 200) 50) + f (fn [i] + (swap! simultatious #(conj % {:id i :status :start :value 1})) + (Thread/sleep (sleep-time i)) + (swap! simultatious #(conj % {:id i :status :end :value -1})) + i) + res (async/ 3 + (first simultatious-over-time) => 0 + (last simultatious-over-time) => 0)))