<commit_before>;; Custom SSA terminators to CLJS core async.

(ns rewindable.ago
  (:require [cljs.core.async.impl.ioc-helpers :as ioc-helpers]
            [cljs.core.async.impl.buffers :as buffers]
            [cljs.core.async.impl.channels :as channels]
            [cljs.core.async.impl.dispatch :as dispatch]
            [cljs.core.async.impl.protocols :as protocols]))

(defn acopy [asrc adst & start-idx]
  (loop [idx (or start-idx 0)]
    (when (< idx (alength asrc))
      (aset adst idx (aget asrc idx))
      (recur (inc idx))))
  adst)

(defn dissoc-in [m [k & ks :as keys]]
  (if ks
    (if-let [next-map (get m k)]
      (let [new-map (dissoc-in next-map ks)]
        (if (seq new-map)
          (assoc m k new-map)
          (dissoc m k)))
      m)
    (dissoc m k)))

; --------------------------------------------------------

(defn make-ago-world []
  (let [last-id (atom 0)]
    (atom {:gen-id #(swap! last-id inc)
           :seqv [0]    ; Sequence numbers, grown when snapshot revived/branched.
           :bufs {}     ; Keyed by buf-id.
           :smas {}     ; Keyed by buf-id, value is state-machine array.
           :smas-new {} ; Same as :smas, but for new, not yet run goroutines.
           })))

(defn seqv+ [ago-world-now]
  (update-in ago-world-now [:seqv (dec (count (:seqv ago-world-now)))] inc))

; --------------------------------------------------------

; Persistent/immutable buffer implementation as an alternative to
; CLJS/core.async's default mutable RingBuffer, so that we can easily
; take ago-world snapshots.  Note: we have an unused length property
; in order to keep ManyToManyChannel happy.
(deftype FifoBuffer [ago-world seqv buf-id length max-length]
  Object
  (pop [_]
    ; Assumes nil is not a valid buffer item.
    (when-let [x (last (get-in @ago-world [:bufs buf-id]))]
      (swap! ago-world #(seqv+ (update-in % [:bufs buf-id] drop-last)))
      (when (empty? (get-in @ago-world [:bufs buf-id]))
        (swap! ago-world #(seqv+ (dissoc-in % [:bufs buf-id]))))
      x))

  (unshift [_ x]
    (swap! ago-world #(seqv+ (update-in % [:bufs buf-id] conj x)))
    nil)

  (unbounded-unshift [this x] ; From RingBuffer 'interface'.
    (.unshift this x))

  (resize [_] nil) ; From RingBuffer 'interface'.

  (cleanup [_ keep?]
    (swap! ago-world #(seqv+ (update-in % [:bufs buf-id]
                                        (fn [xs] (filter keep? xs))))))

  protocols/Buffer
  (full? [this]
    (and (>= max-length 0)
         (>= (count this) max-length)))
  (remove! [this]
    (.pop this))
  (add! [this itm]
    (.unshift this itm))

  cljs.core/ICounted
  (-count [this]
    (count (get-in @ago-world [:bufs buf-id]))))

(defn fifo-buffer [ago-world buf-id max-length]
  (FifoBuffer. ago-world (:seqv @ago-world) buf-id -123 max-length))

; --------------------------------------------------------

(defn ago-chan-buf [ago-world buf]
  (channels/ManyToManyChannel.
   (fifo-buffer ago-world (str (.-buf-id buf) "-takes") -1) 0
   (fifo-buffer ago-world (str (.-buf-id buf) "-puts") -1) 0
   buf false))

(defn ago-chan
  "Creates a channel with an optional buffer. If buf-or-n is a
  number, will create and use a buffer with that max-length."
  ([ago-world] (ago-chan ago-world nil))
  ([ago-world buf-or-n]
     (let [buf-or-n (if (= buf-or-n 0)
                      nil
                      buf-or-n)]
       (ago-chan-buf ago-world
                     (if (number? buf-or-n)
                       (fifo-buffer ago-world
                                    (str "ch-" ((:gen-id @ago-world)))
                                    buf-or-n)
                       buf-or-n)))))

; --------------------------------------------------------

(defn ago-reg-state-machine [ago-world state-machine-arr buf]
  (swap! ago-world #(seqv+ (assoc-in % [:smas-new (.-buf-id buf)] state-machine-arr))))

(defn ago-run-state-machine [ago-world state-machine-arr buf]
  (swap! ago-world #(-> %
                        (dissoc-in [:smas-new (.-buf-id buf)])
                        (assoc-in [:smas (.-buf-id buf)] state-machine-arr)
                        (seqv+))))

(defn ago-dereg-state-machine [ago-world buf]
  (swap! ago-world #(-> %
                        (dissoc-in [:smas-new (.-buf-id buf)])
                        (dissoc-in [:smas (.-buf-id buf)])
                        (seqv+))))

(defn ago-revive-state-machine [ago-world old-sma buf]
  (let [ch (rewindable.ago/ago-chan-buf ago-world buf)
        new-sma (acopy old-sma ((aget old-sma ioc-helpers/FN-IDX))
                       ioc-helpers/STATE-IDX) ; We depend on *-IDX ordering.
        new-sma2 (ioc/aset-all! new-sma ioc-helpers/USER-START-IDX ch)]
    (ago-reg-state-machine ago-world new-sma2 buf)
    (dispatch/run
     (fn []
       (ago-run-state-machine ago-world new-sma2 buf)
       (ioc-helpers/run-state-machine-wrapped new-sma2)))
    ch))

(defn ago-judge-state-machines [bufs-ss smas-ss smas-cur]
  (loop [ss-buf-ids (sort (keys smas-ss))
         cur-buf-ids (sort (keys smas-cur))
         acc-recycled-smas {} ; State machines shared by ss and cur (just need reset).
         acc-reborn-smas []]  ; State machines that need rebirth (were in ss only).
    (let [ss-buf-id (first ss-buf-ids)
          cur-buf-id (first cur-buf-ids)]
      (cond
       (= nil ss-buf-id cur-buf-id)
       [acc-recycled-smas acc-reborn-smas]

       (and ss-buf-id ; In ss but not in cur.
            (or (nil? cur-buf-id)
                (< ss-buf-id cur-buf-id)))
       (recur (rest ss-buf-ids) cur-buf-ids
              acc-recycled-smas
              (let [ss-buf (get bufs-ss ss-buf-id)
                    sma-old (get smas-ss ss-buf-id)]
                ; Later, can invoke (ago-revive-state-machine agw sma-old ss-buf).
                (conj acc-reborn-smas [sma-old ss-buf])))

       (and cur-buf-id ; In cur but not in ss, so drop cur's sma.
            (or (nil? ss-buf-id)
                (< cur-buf-id ss-buf-id)))
       (recur ss-buf-ids (rest cur-buf-ids)
              acc-recycled-smas
              acc-reborn-smas)

       (= ss-buf-id cur-buf-id) ; In both cur and ss.
       (let [sma-ss (get smas-ss ss-buf-id)
             sma-cur (get smas-cur cur-buf-id)]
         (acopy sma-ss sma-cur) ; Recycle the cur's sma.
         (recur (rest ss-buf-ids)
                (rest cur-buf-ids)
                (assoc acc-recycled-smas cur-buf-id sma-cur)
                acc-reborn-smas))

       :else (println "UNEXPECTED case in ago-judge-state-machines")))))

; --------------------------------------------------------

(defn ago-take [state blk ^not-native c]
  (cljs.core.async.impl.ioc-helpers/take! state blk c))

(defn ago-put [state blk ^not-native c val]
  (cljs.core.async.impl.ioc-helpers/put! state blk c val))

(defn ago-alts [state cont-block ports & rest]
  (apply cljs.core.async.impl.ioc-helpers/ioc-alts! state cont-block ports rest))

(defn ago-return-chan [state value]
  (let [^not-native c (aget state ioc-helpers/USER-START-IDX)]<commit_msg>Have ago-take reuse take! impl for future pre-birth checking.<commit_after>;; Custom SSA terminators to CLJS core async.

(ns rewindable.ago
  (:require-macros [cljs.core.async.impl.ioc-macros :as ioc-macros])
  (:require [cljs.core.async.impl.ioc-helpers :as ioc-helpers]
            [cljs.core.async.impl.buffers :as buffers]
            [cljs.core.async.impl.channels :as channels]
            [cljs.core.async.impl.dispatch :as dispatch]
            [cljs.core.async.impl.protocols :as protocols]))

(defn acopy [asrc adst & start-idx]
  (loop [idx (or start-idx 0)]
    (when (< idx (alength asrc))
      (aset adst idx (aget asrc idx))
      (recur (inc idx))))
  adst)

(defn dissoc-in [m [k & ks :as keys]]
  (if ks
    (if-let [next-map (get m k)]
      (let [new-map (dissoc-in next-map ks)]
        (if (seq new-map)
          (assoc m k new-map)
          (dissoc m k)))
      m)
    (dissoc m k)))

; --------------------------------------------------------

(defn make-ago-world []
  (let [last-id (atom 0)]
    (atom {:gen-id #(swap! last-id inc)
           :seqv [0]    ; Sequence numbers, grown when snapshot revived/branched.
           :bufs {}     ; Keyed by buf-id.
           :smas {}     ; Keyed by buf-id, value is state-machine array.
           :smas-new {} ; Same as :smas, but for new, not yet run goroutines.
           })))

(defn seqv+ [ago-world-now]
  (update-in ago-world-now [:seqv (dec (count (:seqv ago-world-now)))] inc))

; --------------------------------------------------------

; Persistent/immutable buffer implementation as an alternative to
; CLJS/core.async's default mutable RingBuffer, so that we can easily
; take ago-world snapshots.  Note: we have an unused length property
; in order to keep ManyToManyChannel happy.
(deftype FifoBuffer [ago-world seqv buf-id length max-length]
  Object
  (pop [_]
    ; Assumes nil is not a valid buffer item.
    (when-let [x (last (get-in @ago-world [:bufs buf-id]))]
      (swap! ago-world #(seqv+ (update-in % [:bufs buf-id] drop-last)))
      (when (empty? (get-in @ago-world [:bufs buf-id]))
        (swap! ago-world #(seqv+ (dissoc-in % [:bufs buf-id]))))
      x))

  (unshift [_ x]
    (swap! ago-world #(seqv+ (update-in % [:bufs buf-id] conj x)))
    nil)

  (unbounded-unshift [this x] ; From RingBuffer 'interface'.
    (.unshift this x))

  (resize [_] nil) ; From RingBuffer 'interface'.

  (cleanup [_ keep?]
    (swap! ago-world #(seqv+ (update-in % [:bufs buf-id]
                                        (fn [xs] (filter keep? xs))))))

  protocols/Buffer
  (full? [this]
    (and (>= max-length 0)
         (>= (count this) max-length)))
  (remove! [this]
    (.pop this))
  (add! [this itm]
    (.unshift this itm))

  cljs.core/ICounted
  (-count [this]
    (count (get-in @ago-world [:bufs buf-id]))))

(defn fifo-buffer [ago-world buf-id max-length]
  (FifoBuffer. ago-world (:seqv @ago-world) buf-id -123 max-length))

; --------------------------------------------------------

(defn ago-chan-buf [ago-world buf]
  (channels/ManyToManyChannel.
   (fifo-buffer ago-world (str (.-buf-id buf) "-takes") -1) 0
   (fifo-buffer ago-world (str (.-buf-id buf) "-puts") -1) 0
   buf false))

(defn ago-chan
  "Creates a channel with an optional buffer. If buf-or-n is a
  number, will create and use a buffer with that max-length."
  ([ago-world] (ago-chan ago-world nil))
  ([ago-world buf-or-n]
     (let [buf-or-n (if (= buf-or-n 0)
                      nil
                      buf-or-n)]
       (ago-chan-buf ago-world
                     (if (number? buf-or-n)
                       (fifo-buffer ago-world
                                    (str "ch-" ((:gen-id @ago-world)))
                                    buf-or-n)
                       buf-or-n)))))

; --------------------------------------------------------

(defn ago-reg-state-machine [ago-world state-machine-arr buf]
  (swap! ago-world #(seqv+ (assoc-in % [:smas-new (.-buf-id buf)] state-machine-arr))))

(defn ago-run-state-machine [ago-world state-machine-arr buf]
  (swap! ago-world #(-> %
                        (dissoc-in [:smas-new (.-buf-id buf)])
                        (assoc-in [:smas (.-buf-id buf)] state-machine-arr)
                        (seqv+))))

(defn ago-dereg-state-machine [ago-world buf]
  (swap! ago-world #(-> %
                        (dissoc-in [:smas-new (.-buf-id buf)])
                        (dissoc-in [:smas (.-buf-id buf)])
                        (seqv+))))

(defn ago-revive-state-machine [ago-world old-sma buf]
  (let [ch (rewindable.ago/ago-chan-buf ago-world buf)
        new-sma (acopy old-sma ((aget old-sma ioc-helpers/FN-IDX))
                       ioc-helpers/STATE-IDX) ; We depend on *-IDX ordering.
        new-sma2 (ioc/aset-all! new-sma ioc-helpers/USER-START-IDX ch)]
    (ago-reg-state-machine ago-world new-sma2 buf)
    (dispatch/run
     (fn []
       (ago-run-state-machine ago-world new-sma2 buf)
       (ioc-helpers/run-state-machine-wrapped new-sma2)))
    ch))

(defn ago-judge-state-machines [bufs-ss smas-ss smas-cur]
  (loop [ss-buf-ids (sort (keys smas-ss))
         cur-buf-ids (sort (keys smas-cur))
         acc-recycled-smas {} ; State machines shared by ss and cur (just need reset).
         acc-reborn-smas []]  ; State machines that need rebirth (were in ss only).
    (let [ss-buf-id (first ss-buf-ids)
          cur-buf-id (first cur-buf-ids)]
      (cond
       (= nil ss-buf-id cur-buf-id)
       [acc-recycled-smas acc-reborn-smas]

       (and ss-buf-id ; In ss but not in cur.
            (or (nil? cur-buf-id)
                (< ss-buf-id cur-buf-id)))
       (recur (rest ss-buf-ids) cur-buf-ids
              acc-recycled-smas
              (let [ss-buf (get bufs-ss ss-buf-id)
                    sma-old (get smas-ss ss-buf-id)]
                ; Later, can invoke (ago-revive-state-machine agw sma-old ss-buf).
                (conj acc-reborn-smas [sma-old ss-buf])))

       (and cur-buf-id ; In cur but not in ss, so drop cur's sma.
            (or (nil? ss-buf-id)
                (< cur-buf-id ss-buf-id)))
       (recur ss-buf-ids (rest cur-buf-ids)
              acc-recycled-smas
              acc-reborn-smas)

       (= ss-buf-id cur-buf-id) ; In both cur and ss.
       (let [sma-ss (get smas-ss ss-buf-id)
             sma-cur (get smas-cur cur-buf-id)]
         (acopy sma-ss sma-cur) ; Recycle the cur's sma.
         (recur (rest ss-buf-ids)
                (rest cur-buf-ids)
                (assoc acc-recycled-smas cur-buf-id sma-cur)
                acc-reborn-smas))

       :else (println "UNEXPECTED case in ago-judge-state-machines")))))

; --------------------------------------------------------

(defn fn-handler [f]
  (reify
    protocols/Handler
    (active? [_] true)
    (commit [_] f)))

(defn ago-take [state blk ^not-native c]
  (if-let [cb (protocols/take!
               c (fn-handler
                  (fn [x]
                    (ioc-macros/aset-all! state ioc-helpers/VALUE-IDX
                                          x ioc-helpers/STATE-IDX blk)
                    (ioc-helpers/run-state-machine-wrapped state))))]
    (do (ioc-macros/aset-all! state ioc-helpers/VALUE-IDX
                              @cb ioc-helpers/STATE-IDX blk)
        :recur)
    nil))

(defn ago-put [state blk ^not-native c val]
  (cljs.core.async.impl.ioc-helpers/put! state blk c val))

(defn ago-alts [state cont-block ports & rest]
  (apply cljs.core.async.impl.ioc-helpers/ioc-alts! state cont-block ports rest))

(defn ago-return-chan [state value]
  (let [^not-native c (aget state ioc-helpers/USER-START-IDX)]
{'fraction_non_alphanumeric': 0.18710648781823158, 'fraction_numerical': 0.0021899808376676703, 'mean_word_length': 1.7303811659192825, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 13, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6055446', 'n_tokens_mistral': 5250, 'n_tokens_neox': 5124, 'n_words': 1316}