<commit_before>(ns parkour.io.dval
  (:require [clojure.java.io :as io]
            [parkour (fs :as fs) (cser :as cser) (mapreduce :as mr)
             ,       (reducers :as pr)]
            [parkour.io.transient :refer [transient-path]]
            [parkour.util :as util :refer [doto-let]])
  (:import [java.io Writer]
           [java.net URI]
           [clojure.lang IDeref IObj IPending]
           [org.apache.hadoop.fs Path]
           [org.apache.hadoop.filecache DistributedCache]))

(defn ^:private cache-name
  "Generate distinct distcache-entry name for `source`."
  [source] (str (gensym "dval-") "-" (-> source fs/path .getName)))

(defn ^:private dcpath*
  [dcname path]
  (let [path (str (fs/path path))
        dcpath* (fn dcpath* [md]
                  (proxy [Path IObj] [path]
                    (meta [] md)
                    (withMeta [md] (dcpath* md))))]
    (dcpath* {:type ::dcpath, ::dcname dcname})))

(defn dcpath
  "Distributed-cache–able instance of `path`.  When serialized into a job
configuration, adds `path` to the distributed cache.  When deserialized from a
job configuration, reconstitutes as the cache path when available and as the
original remote path when not (i.e. under local- or mixed-mode job execution)."
  [path]
  (let [path (fs/path path)
        dcname (cache-name path)]
    (dcpath* dcname path)))

(defmethod print-method ::dcpath
  [path ^Writer writer]
  (if (nil? cser/*conf*)
    (print-method (with-meta path nil) writer)
    (let [^String dcname (-> path meta ::dcname)]
      (fs/distcache! cser/*conf* {dcname path})
      (.write writer "#parkour/dcpath \"")
      (.write writer dcname)
      (.write writer "\""))))

(defn ^:private unfragment
  "The provided `uri`, but without any fragment."
  {:tag `URI}
  [^URI uri]
  (let [fragment (.getFragment uri)]
    (if (nil? fragment)
      uri
      (let [uri-s (str uri), n (- (count uri-s) (count fragment) 1)]
        (fs/uri (subs uri-s 0 n))))))

(defn ^:private resolve-source
  "Resolve distributed cache `remote` and `local` URIs to the \"most local\"
available source path.  Result will usually be a local file path, but may be the
original remote path under mixed-mode job execution."
  [[^URI remote local]]
  (if-let [fragment (.getFragment remote)]
    (let [symlink (io/file fragment), symlink? (.exists symlink)
          local (io/file (str local)), local? (.exists local)
          remote (unfragment remote), remote? (= "file" (.getScheme remote))
          source (cond symlink? symlink, local? local, remote? remote
                       ;; Could localize, but issues: clean-up, directories
                       (mr/local-runner? cser/*conf*) remote
                       :else (throw (ex-info
                                     (str remote ": cannot locate local file")
                                     {:remote remote, :local local})))]
      (fs/path source))))

(defn ^:private dcpath-reader
  "EDN tagged-literal reader for dcpaths."
  [dcname]
  (let [remotes (seq (DistributedCache/getCacheFiles cser/*conf*))
        locals (or (seq (DistributedCache/getLocalCacheFiles cser/*conf*))
                   (map unfragment remotes))]
    (if (not= (count remotes) (count locals))
      (throw (ex-info "cache files do not match local files"
                      {:remotes remotes, :locals locals}))
      (->> (map vector remotes locals)
           (pr/ffilter (fn [[^URI r]] (= dcname (.getFragment r))))
           (resolve-source)
           (dcpath* dcname)))))

;; Distributed value
(deftype DVal [value form]
  Object
  (toString [_]
    (let [v (if (realized? value) @value :pending)]
      (str "#<DVal: " (pr-str v) ">")))
  IDeref (deref [_] @value)
  IPending (isRealized [_] (realized? value)))

(defmethod print-method DVal
  [^DVal dval ^Writer w]
  (if (nil? cser/*conf*)
    (.write w (str dval))
    (do
      (.write w "#parkour/dval ")
      (.write w (pr-str (.-form dval))))))

(defn ^:private dval-reader
  "EDN tagged-literal reader for dvals."
  [[f & args]] (delay (apply f args)))

(defn ^:private dval*
  "Return a dval which locally proxies to `valref` and remotely will deserialize
as a delay over applying var `readv` to `args`."
  [valref readv & args] (DVal. valref (cons readv args)))

(defn dval
  "Return a dval which acts as a delay over applying var `readv` to `args`."
  [readv & args] (apply dval* (delay (apply readv args)) readv args))

(defn ^:private identity-ref
  "Return reference which yields `x` when `deref`ed."
  [x] (reify IDeref (deref [_] x), IPending (isRealized [_] true)))

(defn value-dval
  "Return a dval which locally holds `value` and remotely will deserialize as a
delay over applying var `readv` to `args`."<commit_msg>Allow DVals to locally deserialize as DVals.<commit_after>(ns parkour.io.dval
  (:refer-clojure :exclude [eval])
  (:require [clojure.java.io :as io]
            [parkour (fs :as fs) (cser :as cser) (mapreduce :as mr)
             ,       (reducers :as pr)]
            [parkour.io.transient :refer [transient-path]]
            [parkour.util :as util :refer [doto-let]])
  (:import [java.io Writer]
           [java.net URI]
           [clojure.lang IDeref IObj IPending]
           [org.apache.hadoop.fs Path]
           [org.apache.hadoop.filecache DistributedCache]))

(defn ^:private cache-name
  "Generate distinct distcache-entry name for `source`."
  [source] (str (gensym "dval-") "-" (-> source fs/path .getName)))

(defn ^:private dcpath*
  [dcname path]
  (let [path (str (fs/path path))
        dcpath* (fn dcpath* [md]
                  (proxy [Path IObj] [path]
                    (meta [] md)
                    (withMeta [md] (dcpath* md))))]
    (dcpath* {:type ::dcpath, ::dcname dcname})))

(defn dcpath
  "Distributed-cache–able instance of `path`.  When serialized into a job
configuration, adds `path` to the distributed cache.  When deserialized from a
job configuration, reconstitutes as the cache path when available and as the
original remote path when not (i.e. under local- or mixed-mode job execution)."
  [path]
  (let [path (fs/path path)
        dcname (cache-name path)]
    (dcpath* dcname path)))

(defmethod print-method ::dcpath
  [path ^Writer writer]
  (if (nil? cser/*conf*)
    (print-method (with-meta path nil) writer)
    (let [^String dcname (-> path meta ::dcname)]
      (fs/distcache! cser/*conf* {dcname path})
      (.write writer "#parkour/dcpath \"")
      (.write writer dcname)
      (.write writer "\""))))

(defn ^:private unfragment
  "The provided `uri`, but without any fragment."
  {:tag `URI}
  [^URI uri]
  (let [fragment (.getFragment uri)]
    (if (nil? fragment)
      uri
      (let [uri-s (str uri), n (- (count uri-s) (count fragment) 1)]
        (fs/uri (subs uri-s 0 n))))))

(defn ^:private resolve-source
  "Resolve distributed cache `remote` and `local` URIs to the \"most local\"
available source path.  Result will usually be a local file path, but may be the
original remote path under mixed-mode job execution."
  [[^URI remote local]]
  (let [fragment (.getFragment remote)
        symlink (io/file fragment), symlink? (.exists symlink)
        local (io/file (str local)), local? (.exists local)
        remote (unfragment remote), remote? (= "file" (.getScheme remote))
        source (cond symlink? symlink, local? local, remote? remote
                     ;; Could localize, but issues: clean-up, directories
                     (nil? mr/*context*) remote
                     (mr/local-runner? cser/*conf*) remote
                     :else (throw (ex-info
                                   (str remote ": cannot locate local file")
                                   {:remote remote, :local local})))]
    (fs/path source)))

(defn ^:private dcpath-reader
  "EDN tagged-literal reader for dcpaths."
  [dcname]
  (let [remotes (seq (DistributedCache/getCacheFiles cser/*conf*))
        locals (or (seq (DistributedCache/getLocalCacheFiles cser/*conf*))
                   (map unfragment remotes))]
    (if (not= (count remotes) (count locals))
      (throw (ex-info "cache files do not match local files"
                      {:remotes remotes, :locals locals}))
      (->> (map vector remotes locals)
           (pr/ffilter (fn [[^URI r]] (= dcname (.getFragment r))))
           (resolve-source)
           (dcpath* dcname)))))

;; Distributed value
(deftype DVal [value form]
  Object
  (toString [_]
    (let [v (if (realized? value) @value :pending)]
      (str "#<DVal: " (pr-str v) ">")))
  IDeref (deref [_] @value)
  IPending (isRealized [_] (realized? value)))

(defmethod print-method DVal
  [^DVal dval ^Writer w]
  (if (nil? cser/*conf*)
    (.write w (str dval))
    (do
      (.write w "#parkour/dval ")
      (.write w (pr-str (.-form dval))))))

(defn ^:private dval-reader
  "EDN tagged-literal reader for dvals."
  [form]
  (let [value (delay (apply pr/funcall form))]
    (DVal. value form)))

(defn ^:private dval*
  "Return a dval which locally proxies to `valref` and remotely will deserialize
as a delay over applying var `readv` to `args`."
  [valref readv & args] (DVal. valref (cons readv args)))

(defn dval
  "Return a dval which acts as a delay over applying var `readv` to `args`."
  [readv & args] (apply dval* (delay (apply readv args)) readv args))

(defn eval
  "Evaluate `dval` form, recomputing value each time."
  [dval] (apply pr/funcall (.-form ^DVal dval)))

(defn ^:private identity-ref
  "Return reference which yields `x` when `deref`ed."
  [x] (reify IDeref (deref [_] x), IPending (isRealized [_] true)))

(defn value-dval
  "Return a dval which locally holds `value` and remotely will deserialize as a
delay over applying var `readv` to `args`."
{'fraction_non_alphanumeric': 0.14034365675481017, 'fraction_numerical': 0.00041156497582055766, 'mean_word_length': 2.6790310370931114, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2246364', 'n_tokens_mistral': 3190, 'n_tokens_neox': 3092, 'n_words': 1077}