Decomplect dvals into dcpaths and dvals.

Author: llasram

resources/data_readers.clj

@@ -2,5 +2,6 @@
  hadoop.fs/path parkour.fs/path,
  hadoop.mapreduce/job parkour.mapreduce/job,
  java.net/uri parkour.fs/uri,
- parkour/dval parkour.io.dval/distcache-dval,
+ parkour/dcpath parkour.io.dval/dcpath-reader,
+ parkour/dval parkour.io.dval/dval-reader,
  }

src/clojure/parkour/io/dval.clj

@@ -6,17 +6,42 @@
             [parkour.util :as util :refer [doto-let]])
   (:import [java.io Writer]
            [java.net URI]
-           [clojure.lang IDeref IPending]
+           [clojure.lang IDeref IObj IPending]
+           [org.apache.hadoop.fs Path]
            [org.apache.hadoop.filecache DistributedCache]))
 
-;; Distributed value
-(deftype DVal [value readv params dcm]
-  Object
-  (toString [_]
-    (let [v (if (realized? value) @value :pending)]
-      (str "#<DVal: " (pr-str v) ">")))
-  IDeref (deref [_] @value)
-  IPending (isRealized [_] (realized? value)))
+(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 path))))]
+    (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."
@@ -28,74 +53,83 @@
       (let [uri-s (str uri), n (- (count uri-s) (count fragment) 1)]
         (fs/uri (subs uri-s 0 n))))))
 
-(defn ^:private ->entry
-  "Parse `remote` and `local` into mapping tuple of (fragment, cache path)."
-  [^URI remote local]
+(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})))]
-      [fragment (fs/path source)])))
+      (fs/path source))))
 
-(defn ^:private distcache-dval
-  "Remote-side dval data reader, reconstituting as a delay."
-  [[readv params cnames]]
+(defn ^:private dcpath-reader
+  "EDN tagged-literal reader for dcpaths."
+  [dcname]
   (let [remotes (seq (DistributedCache/getCacheFiles cser/*conf*))
-        locals (seq (DistributedCache/getLocalCacheFiles cser/*conf*))
-        _ (when (not= (count remotes) (count locals))
-            (throw (ex-info "cache files do not match local files"
-                            {:remotes remotes, :locals locals})))
-        entries (map ->entry remotes locals)
-        cname->source (->> entries (remove nil?) (into {}))
-        sources (map cname->source cnames)
-        args (concat params sources)]
-    (delay (apply readv args))))
+        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 writer]
-  (let [repr (if (nil? cser/*conf*)
-               (str dval)
-               (let [v (.-readv dval), p (.-params dval), dcm (.-dcm dval)
-                     literal (pr-str [v p (-> dcm keys vec)])]
-                 (fs/distcache! cser/*conf* dcm)
-                 (str "#parkour/dval " literal)))]
-    (.write writer repr)))
+  [^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 ^:private cache-name
-  "Generate distinct distcache-entry name for `source`."
-  [source] (str (gensym "dval-") "-" (-> source fs/path .getName)))
+  [x] (reify IDeref (deref [_] x), IPending (isRealized [_] true)))
 
-(defn ^:private dval
-  "Return a dval which locally holds `value` and remotely will deserialize by
-applying var `readv` to the concatenation of `params` and distributed copies of
-`sources`."
-  ([value readv sources] (dval value readv nil sources))
-  ([value readv params sources]
-     (let [dcm (into {} (map (juxt cache-name fs/uri) sources))
-           value (if (instance? IDeref value) value (identity-ref value))]
-       (DVal. value readv params dcm))))
+(defn value-dval
+  "Return a dval which locally holds `value` and remotely will deserialize as a
+delay over applying var `readv` to `args`."
+  [value readv & args] (apply dval* (identity-ref value) readv args))
 
 (defn load-dval
-  "Return a dval which will deserialize by applying var `readv` to the
+  "Return a delay-like dval which will realize by applying var `readv` to the
 concatenation of `params` and `sources` locally and distributed copies of
 `sources` remotely."
   ([readv sources] (load-dval readv nil sources))
   ([readv params sources]
-     (let [args (concat params sources)
-           value (delay (apply readv args))]
-       (dval value readv params sources))))
+     (apply dval readv (concat params (map dcpath sources)))))
 
 (defn copy-dval
   "Like `load-dval`, but first copy `sources` to transient locations."
@@ -114,7 +148,7 @@ deserialize by calling var `readv` with a distributed copy of the transient
 serialization path."
   [writef readv value]
   (let [source (doto (transient-path) (writef value))]
-    (dval value readv [source])))
+    (value-dval value readv (dcpath source))))
 
 (defn edn-dval
   "EDN-serialize `value` to a transient location and yield a wrapping dval."