Use primitive-math

Author: valerauko

src/manifold/bus.clj

@@ -6,7 +6,8 @@
      [stream :as s]
      [deferred :as d]
      [utils :refer [definterface+]]]
-    [potemkin.types :refer [deftype+]])
+    [potemkin.types :refer [deftype+]]
+    [clj-commons.primitive-math :as p])
   (:import
     [java.util.concurrent
      ConcurrentHashMap]
@@ -53,25 +54,25 @@
   (if (nil? ary)
     (object-array [x])
     (let [len  (Array/getLength ary)
-          ary' (object-array (inc len))]
+          ary' (object-array (p/inc len))]
       (System/arraycopy ary 0 ary' 0 len)
       (aset ^objects ary' len x)
       ary')))
 
 (defn- disj' [^objects ary x]
   (let [len (Array/getLength ary)]
     (if-let [idx (loop [i 0]
-                   (if (<= len i)
+                   (if (p/<= len i)
                      nil
                      (if (identical? x (aget ary i))
                        i
-                       (recur (inc i)))))]
-      (let [idx (long idx)]
-        (if (== 1 len)
+                       (recur (p/inc i)))))]
+      (let [idx (p/long idx)]
+        (if (p/== 1 len)
           nil
-          (let [ary' (object-array (dec len))]
+          (let [ary' (object-array (p/dec len))]
             (System/arraycopy ary 0 ary' 0 idx)
-            (System/arraycopy ary (inc idx) ary' idx (- len idx 1))
+            (System/arraycopy ary (p/inc idx) ary' idx (p/- len idx 1))
             ary')))
       ary)))
 

src/manifold/deferred.clj

@@ -10,6 +10,7 @@
     [manifold.time :as time]
     [manifold.utils :as utils :refer [assert-some definterface+]]
     [potemkin.types :refer [def-abstract-type reify+ defprotocol+ deftype+]]
+    [clj-commons.primitive-math :as p]
     [riddley.compiler :as compiler]
     [riddley.walk :as walk])
   (:import
@@ -689,7 +690,7 @@
     ([]
      (deferred (ex/executor)))
     ([executor]
-     (if (and (zero? (rem (.incrementAndGet created) 1024))
+     (if (and (p/zero? (rem (.incrementAndGet created) 1024))
               debug/*dropped-error-logging-enabled?*)
        (LeakAwareDeferred. nil ::unset nil (utils/mutex) (LinkedList.) nil false executor)
        (Deferred. nil ::unset nil (utils/mutex) (LinkedList.) nil false executor)))))
@@ -703,8 +704,8 @@
   #_{:inline
      (fn [val]
        (cond
-         (true? val)   'manifold.deferred/true-deferred-
-         (false? val)  'manifold.deferred/false-deferred-
+         (p/true? val)   'manifold.deferred/true-deferred-
+         (p/false? val)  'manifold.deferred/false-deferred-
          (nil? val)    'manifold.deferred/nil-deferred-
          :else         `(SuccessDeferred. ~val nil nil)))
    :inline-arities #{1}}
@@ -1026,7 +1027,7 @@
   "Like `chain`, but does not coerce deferrable values.  This is useful both when coercion
    is undesired, or for 2-4x better performance than `chain`."
   {:inline (fn [& args]
-             (if false #_(< 3 (count args))
+             (if false #_(p/< 3 (count args))
                (apply unroll-chain 'manifold.deferred/unwrap' args)
                `(chain'- nil ~@args)))}
   ([x]
@@ -1046,12 +1047,12 @@
    The returned deferred will only be realized once all functions have been applied and their
    return values realized.
 
-       @(chain 1 inc #(future (inc %))) => 3
+       @(chain 1 inc #(future (p/inc %))) => 3
 
        @(chain (future 1) inc inc) => 3
    "
   {:inline (fn [& args]
-             (if false #_(< 3 (count args))
+             (if false #_(p/< 3 (count args))
                (apply unroll-chain 'manifold.deferred/unwrap args)
                `(chain- nil ~@args)))}
   ([x]
@@ -1153,7 +1154,7 @@
 (defn finally
   "An equivalent of the finally clause, which takes a no-arg side-effecting function that executes
    no matter what the result.
-  
+
   Returns x unless a Throwable is thrown in f, in which case it returns an error-deferred."
   [x f]
   (if-let [d (->deferred x nil)]
@@ -1174,7 +1175,7 @@
 
         ;; no further results, decrement the counter one last time
         ;; and return the result if everything else has been realized
-        (if (zero? (.get counter))
+        (if (p/zero? (.get counter))
           (success-deferred (or (seq ary) (list)))
           d)
 
@@ -1196,7 +1197,7 @@
                 (on-realized (chain' x)
                              (fn [val]
                                (aset ary idx val)
-                               (when (zero? (.decrementAndGet counter))
+                               (when (p/zero? (.decrementAndGet counter))
                                  (success! d (seq ary))))
                              (fn [err]
                                (error! d err)))
@@ -1228,10 +1229,10 @@
     (clojure.core/loop [i 1]
       (if (= i n)
         a
-        (let [j (rand-int (inc i))]
+        (let [j (rand-int (p/inc i))]
           (aset a i (aget a j))
           (aset a j i)
-          (recur (inc i)))))))
+          (recur (p/inc i)))))))
 
 (defn alt'
   "Like `alt`, but only unwraps Manifold deferreds."
@@ -1240,7 +1241,7 @@
         cnt        (count vals)
         ^ints idxs (random-array cnt)]
     (clojure.core/loop [i 0]
-      (when (< i cnt)
+      (when (p/< i cnt)
         (let [i' (aget idxs i)
               x  (nth vals i')]
           (if (deferred? x)
@@ -1250,7 +1251,7 @@
               (do (on-realized (chain' x)
                                #(success! d %)
                                #(error! d %))
-                  (recur (inc i))))
+                  (recur (p/inc i))))
             (success! d x)))))
     d))
 
@@ -1324,9 +1325,9 @@
 
    (loop [i 1e6]
      (chain (future i)
-       #(if (zero? %)
+       #(if (p/zero? %)
           %
-          (recur (dec %)))))"
+          (recur (p/dec %)))))"
   [bindings & body]
   (let [vars     (->> bindings (partition 2) (map first))
         vals     (->> bindings (partition 2) (map second))
@@ -1479,15 +1480,15 @@
    Returns a deferred value, representing the value returned by the body.
 
       (let-flow [x (future 1)]
-        (+ x 1))
+        (p/+ x 1))
 
       (let-flow [x (future 1)
-                 y (future (+ x 1))]
-        (+ y 1))
+                 y (future (p/+ x 1))]
+        (p/+ y 1))
 
       (let [x (future 1)]
-        (let-flow [y (future (+ x 1))]
-          (+ y 1)))"
+        (let-flow [y (future (p/+ x 1))]
+          (p/+ y 1)))"
   [bindings & body]
   (expand-let-flow
     'manifold.deferred/chain

src/manifold/executor.clj

@@ -88,9 +88,9 @@
        (when (contains? stats Stats$Metric/QUEUE_LENGTH)
          {:queue-length (q #(.getQueueLength s %))})
        (when (contains? stats Stats$Metric/QUEUE_LATENCY)
-         {:queue-latency (q #(double (/ (.getQueueLatency s %) 1e6)))})
+         {:queue-latency (q #(p/double (p// (.getQueueLatency s %) 1e6)))})
        (when (contains? stats Stats$Metric/TASK_LATENCY)
-         {:task-latency (q #(double (/ (.getTaskLatency s %) 1e6)))})
+         {:task-latency (q #(p/double (p// (.getTaskLatency s %) 1e6)))})
        (when (contains? stats Stats$Metric/TASK_ARRIVAL_RATE)
          {:task-arrival-rate (q #(.getTaskArrivalRate s %))})
        (when (contains? stats Stats$Metric/TASK_COMPLETION_RATE)

src/manifold/stream.clj

@@ -7,6 +7,7 @@
     [clojure.core :as clj]
     [manifold.deferred :as d]
     [potemkin.types :refer [deftype+]]
+    [clj-commons.primitive-math :as p]
     [manifold.utils :as utils]
     [manifold.time :as time]
     [manifold.stream
@@ -494,7 +495,7 @@
         callback #(d/chain %
                            callback
                            (fn [result]
-                             (when (false? result)
+                             (when (p/false? result)
                                (d/success! complete true))
                              result))]
     (connect source (Callback. callback #(d/success! complete true) nil nil) nil)
@@ -567,7 +568,7 @@
          (cons x (stream->seq s timeout-interval)))))))
 
 (defn- periodically-
-  [stream period initial-delay f]
+  [stream ^long period initial-delay f]
   (let [cancel (promise)]
     (deliver cancel
              (time/every period
@@ -588,7 +589,7 @@
                                              (fn [x]
                                                (if-not x
                                                  (close! stream)
-                                                 (periodically- stream period (- period (rem (System/currentTimeMillis) period)) f)))))))
+                                                 (periodically- stream period (p/- period (rem (System/currentTimeMillis) period)) f)))))))
                              (catch Throwable e
                                (@cancel)
                                (close! stream)
@@ -600,8 +601,8 @@
    (let [s (stream 1)]
      (periodically- s period initial-delay f)
      (source-only s)))
-  ([period f]
-   (periodically period (- period (rem (System/currentTimeMillis) period)) f)))
+  ([^long period f]
+   (periodically period (p/- period (rem (System/currentTimeMillis) period)) f)))
 
 (declare zip)
 
@@ -745,7 +746,7 @@
 
 (defn mapcat
   "Equivalent to Clojure's `mapcat`, but for streams instead of sequences.
-  
+
   Note that just like `clojure.core/mapcat`, the provided function `f`
   must return a collection and not a stream."
   ([f s]
@@ -904,7 +905,7 @@
                               (identical? result ::timeout)
                               timeout-val
 
-                              (false? result)
+                              (p/false? result)
                               false
 
                               :else
@@ -926,7 +927,7 @@
                           (if (identical? default-val x)
                             x
                             (let [[size msg] x]
-                              (buf+ (- size))
+                              (buf+ (p/- ^long size))
                               msg))))]
       (if blocking?
         @val
@@ -944,7 +945,7 @@
 
                             :else
                             (let [[size msg] x]
-                              (buf+ (- size))
+                              (buf+ (p/- ^long size))
                               msg))))]
       (if blocking?
         @val
@@ -965,7 +966,7 @@
    (buffered-stream (constantly 1) buffer-size))
   ([metric limit]
    (buffered-stream metric limit identity))
-  ([metric limit description]
+  ([metric ^long limit description]
    (let [buf         (stream Integer/MAX_VALUE)
          buffer-size (AtomicLong. 0)
          last-put    (AtomicReference. (d/success-deferred true))
@@ -974,10 +975,10 @@
                          (let [buf' (.addAndGet buffer-size n)
                                buf  (unchecked-subtract buf' n)]
                            (cond
-                             (and (<= buf' limit) (< limit buf))
+                             (and (p/<= buf' limit) (p/< limit buf))
                              (-> last-put .get (d/success! true))
 
-                             (and (<= buf limit) (< limit buf'))
+                             (and (p/<= buf limit) (p/< limit buf'))
                              (-> last-put (.getAndSet (d/deferred)) (d/success! true))))))]
 
      (BufferedStream.
@@ -1013,7 +1014,7 @@
    (batch (constantly 1) batch-size nil s))
   ([max-size max-latency s]
    (batch (constantly 1) max-size max-latency s))
-  ([metric max-size max-latency s]
+  ([metric ^long max-size ^long max-latency s]
    (assert (pos? max-size))
 
    (let [buf (stream)
@@ -1022,16 +1023,16 @@
      (connect-via-proxy s buf s' {:description {:op "batch"}})
      (on-closed s' #(close! buf))
 
-     (d/loop [msgs [], size 0, earliest-message -1, last-message -1]
+     (d/loop [msgs [], ^long size 0, ^long earliest-message -1, ^long last-message -1]
        (cond
          (or
-           (== size max-size)
-           (and (< max-size size) (== (count msgs) 1)))
+           (p/== size max-size)
+           (and (p/< max-size size) (p/== (count msgs) 1)))
          (d/chain' (put! s' msgs)
                    (fn [_]
                      (d/recur [] 0 -1 -1)))
 
-         (> size max-size)
+         (p/> size max-size)
          (let [msg (peek msgs)]
            (d/chain' (put! s' (pop msgs))
                      (fn [_]
@@ -1045,7 +1046,7 @@
                      (take! buf ::empty)
                      (try-take! buf
                                 ::empty
-                                (- max-latency (- (System/currentTimeMillis) earliest-message))
+                                (p/- max-latency (p/- (System/currentTimeMillis) earliest-message))
                                 ::timeout))
                    (fn [msg]
                      (condp identical? msg
@@ -1064,8 +1065,8 @@
                        (let [time (System/currentTimeMillis)]
                          (d/recur
                            (conj msgs msg)
-                           (+ size (metric msg))
-                           (if (neg? earliest-message)
+                           (p/+ size ^long (metric msg))
+                           (if (p/< earliest-message 0)
                              time
                              earliest-message)
                            time)))))))
@@ -1080,15 +1081,15 @@
    this is set to one second's worth."
   ([max-rate s]
    (throttle max-rate max-rate s))
-  ([max-rate max-backlog s]
+  ([^long max-rate ^double max-backlog s]
    (let [buf    (stream)
          s'     (stream)
-         period (double (/ 1000 max-rate))]
+         period (p/double (p// 1000 max-rate))]
 
      (connect-via-proxy s buf s' {:description {:op "throttle"}})
      (on-closed s' #(close! buf))
 
-     (d/loop [backlog 0.0, read-start (System/currentTimeMillis)]
+     (d/loop [^double backlog 0.0, ^long read-start (System/currentTimeMillis)]
        (d/chain (take! buf ::none)
 
                 (fn [msg]
@@ -1100,11 +1101,11 @@
 
                 (fn [result]
                   (when result
-                    (let [elapsed  (double (- (System/currentTimeMillis) read-start))
-                          backlog' (min (+ backlog (- (/ elapsed period) 1)) max-backlog)]
-                      (if (<= 1 backlog')
-                        (- backlog' 1.0)
-                        (d/timeout! (d/deferred) (- period elapsed) 0.0)))))
+                    (let [elapsed  (p/double (p/- (System/currentTimeMillis) read-start))
+                          backlog' (p/min (p/+ backlog (p/- (p// elapsed period) 1.0)) max-backlog)]
+                      (if (p/<= 1.0 backlog')
+                        (p/- backlog' 1.0)
+                        (d/timeout! (d/deferred) (p/- period elapsed) 0.0)))))
 
                 (fn [backlog]
                   (if backlog