(ns coderloop.parallel-io

     (:refer-clojure :only [])

   (:require rlm.ns-rlm rlm.light-base))

 (rlm.ns-rlm/ns-clone rlm.light-base)

 

 (import '[java.io FileInputStream RandomAccessFile])

 (use '(clojure.contrib   [str-utils2 :only [join]]))

 (import 'java.util.concurrent.atomic.AtomicLong)

 

 ;;(set! *warn-on-reflection* true)

 

 

 

 

 

 ;;;; Reading

 

 (defn chunk-file

   "Partitions a file into n line-aligned chunks.  Returns a list of start and

   end byte offset pairs."

   [filename n]

   (with-open [file (java.io.RandomAccessFile. filename "r")]

     (let [offsets (for [offset (range 0 (.length file) (/ (.length file) n))]

 		    (do (when-not (zero? offset)

 			  (.seek file offset)

 			  (while (not= (.read file) (int \newline))))

 			(.getFilePointer file)))

 	  offsets (concat offsets [(.length file)])]

       (doall (partition 2 (interleave offsets (rest offsets)))))))

 

 

 (defn read-lines-range [file start-byte end-byte]

   "Returns a lazy sequence of lines from file between start-byte and end-byte."

   (let [reader (-> (doto (java.io.FileInputStream. file)

 		     (.skip start-byte))

 		   (java.io.BufferedInputStream. (* 8 131072))

 		   (java.io.InputStreamReader. "US-ASCII")

 		   (java.io.BufferedReader. 131072))]

     (letfn [(read-line [remaining]

 		       (lazy-seq

 			(if-let [line (and (pos? remaining) (.readLine reader))]

 			  (cons line (read-line (- remaining (.length line))))

 			  (.close reader))))]

       (read-line (- end-byte start-byte)))))

 

 (defn #^"[Ljava.lang.String;" dumbest-split

   [#^String s c #^"[Ljava.lang.String;" tokens]

   (let [len (dec (int (alength tokens)))]

     (loop [start (int 0)

 	   i (int 0)]

       (let [idx (int (.indexOf s (int c) (int start)))]

 	(if (or (neg? idx) (>= i len))

 	  (do (aset tokens i (.substring s start))

 	      tokens)

 	  (do (aset tokens i (.substring s start idx))

 	      (recur (inc idx) (inc i))))))))

 

 (defn parse-lines [lines]

   (let [ary (make-array String 12)]

     (for [#^String line lines

 	  :let [fields (dumbest-split line \space ary)

 		status (aget fields 8)

 		bytes (aget fields 9)

 		#^String ref (aget fields 10)]

 	  :when (= (aget fields 5) "\"GET")

 	  :when ('#{"200" "304" "404"} status)]

       {:client (aget fields 0)

        :url (aget fields 6)

        :status status

        :bytes (if (= bytes "-") 0 (Long/parseLong bytes))

        :ref (.substring ref 1 (dec (count ref)))})))

 

 ;;;; Tallying

 

 (defn bump! [map-atom #^String key #^Long delta]

   (if-let [#^AtomicLong counter (get @map-atom key)]

     (.addAndGet counter delta)

     (swap! map-atom #(assoc % (String. key)

 			    (if-let [#^AtomicLong counter (get % key)]

 			      (AtomicLong. (+ (.get counter) delta))

 			      (AtomicLong. delta))))))

 

 (def article-re #"^/ongoing/When/\d\d\dx/\d\d\d\d/\d\d/\d\d/[^ .]+$")

 

 (defn tally! [{:keys [url-hits url-bytes clients refs s404s]} records]

   (doseq [{:keys [#^String url bytes client status #^String ref]} records]

     (if (= status "404")

       (bump! s404s url 1)

       (do (bump! url-bytes url bytes)

 	  (when (when (.startsWith url "/ongoing/When/")

 		  (re-matches article-re url))

 	    (bump! url-hits url 1)

 	    (bump! clients client 1)

 	    (when-not (or (= ref "-")

 			  (.startsWith ref "http://www.tbray.org/ongoing/"))

 	      (bump! refs ref 1)))))))

 

 ;;;; Reporting

 

 (defn truncate [s n]

   (if (> (count s) n) (str (.substring s 0 n) "...") s))

 

 (defn print-top10 [results label & [shrink?]]

   (println "Top" label)

   (let [fmt (if shrink? " %9.1fM: %s" " %10d: %s")]

     (doseq [[k v] (take 10 results)]

       (let [v (if shrink? (/ v 1024.0 1024.0) (long v))]

 	(println (format fmt v (truncate k 60))))))

   (println))

 

 (defn sort-by-vals-desc [m]

   (sort-by #(- (val %)) m))

 

 

 

 (defn take-greatest-vals [n m]

   (when-let [m (seq m)]

     (reduce (fn [best x]

 	      (if (>= (val x) (val (last best)))

 		(vec (take n (sort-by-vals-desc (conj best x))))

 		best))

 	    [(first m)] (rest m))))

 

 (defn report [tallies state]

   (->> state

        (map (fn [[tally rows]] (str (count @rows) " " (name tally))))

        (join ", ")

        (println))

   (println)

   (->> tallies

        (pmap (fn [[tally & options]]

 	       (cons (take-greatest-vals 10 (sort-by-vals-desc @(state tally)))

 		     options)))

        (map #(apply print-top10 %))

        (dorun)))

 

 						 ;;;; Main

 

 (def tallies [[:url-hits  "URIs by hit"]

 	      [:url-bytes "URIs by bytes" :shrink]

 	      [:s404s	   "404s"]

 	      [:clients   "client addresses"]

 	      [:refs	   "referrers"]])

 

 (defn wf-atoms [file]

   (let [chunk-count (int (/ (.length (java.io.File. file)) (* 32 1024 1024)))

 	state (zipmap (map first tallies) (repeatedly #(atom {})))]

     (dorun

      (pmap (fn [[idx [start end]]]

 	     (println (str "Chunk " idx "/" chunk-count

 			   " (" start " -> " end ")"))

 	     (->> (read-lines-range file start end)

 		  (parse-lines)

 		  (tally! state)))

 	   (indexed (chunk-file file chunk-count))))

     (time (report tallies state))))