<commit_before>(ns circle.workers
  (:require [circle.backend.build.run :as run])
  (:require [circle.backend.project.circle :as circle])
  (:require [circle.backend.build.config :as config])
  (:use [clojure.tools.logging :only (infof errorf error)])
  (:require [circle.ruby :as ruby])
  (:use [circle.airbrake :only (airbrake)])
  (:require [circle.env :as env])
  (:require [fs])
  (:require [circle.model.project :as project]))

; TODO: find another home for this
(defn run-build-from-jruby
  [url]
  (let [build (config/build-from-url url)]
    (run/run-build build)))

;;; Workers. Handles starting workers, checking if they're done, and getting the
;;; result.
(defn call-clojure-from-ruby [f args]
  (ruby/->ruby (apply f args)))

(def worker-store (ref {}))

(defmacro log-future
  "Takes a function to be called in a future. Starts the future and logs the result at the end"
  [& body]
  `(future
     (try
       (let [result# (do ~@body)]
         (infof "%s returned %s" (quote  ~@body) result#)
         result#)
       (catch Exception e#
         (airbrake {:body (quote ~@body) :future true})
         (error e# "%s threw" (quote ~@body))
         (throw e#)))))

(defn start-worker [f & args]
  "Call fn with args as a worker. Returns the id of the worker"
  (let [fut (log-future
             (call-clojure-from-ruby f args))]
    (dosync
     (let [next-id (count @worker-store)]
       (infof "starting worker id %s for %s" next-id f)
       (alter worker-store assoc next-id fut)
       next-id))))

(defn fire-worker [f & args]
  "Start a worker, but don't wait for a response"
  (log-future (apply f args))
  nil)

(defn blocking-worker [f & args]
  "Start a worker and block until it returns"
  (call-clojure-from-ruby f args))

(defn worker-done? [id]
  "Return if the worker is done. Throw an NPE if there is no such worker"
  (dosync
   (let [as-int (int id)
         fut (get @worker-store as-int)]
     (future-done? fut))))

(defn wait-for-worker [id]
  "Block until the worker is done, and return it's result. Will only work once, next time it throws a NPE because the worker is no longer available"
  (dosync
   (let [as-int (int id)
         fut (get @worker-store as-int)
         ; TODO: when we dereference this, it may result in throwing the
         ; exception that occurred within the future. We want the stack trace
         ; that appears on error to be that stack-trace.
         result (deref fut)]

     (alter worker-store dissoc as-int)
     result)))

(defn worker-count []
  (dosync
   (count @worker-store)))<commit_msg>Extract the real exception out of the future.<commit_after>(ns circle.workers
  (:require [circle.backend.build.run :as run])
  (:require [circle.backend.project.circle :as circle])
  (:require [circle.backend.build.config :as config])
  (:use [clojure.tools.logging :only (infof errorf error)])
  (:require [circle.ruby :as ruby])
  (:use [circle.airbrake :only (airbrake)])
  (:require [circle.env :as env])
  (:require [fs])
  (:require [circle.model.project :as project]))

; TODO: find another home for this
(defn run-build-from-jruby
  [url]
  (let [build (config/build-from-url url)]
    (run/run-build build)))

;;; Workers. Handles starting workers, checking if they're done, and getting the
;;; result.
(defn call-clojure-from-ruby [f args]
  (ruby/->ruby (apply f args)))

(def worker-store (ref {}))

(defmacro log-future
  "Takes a function to be called in a future. Starts the future and logs the result at the end"
  [& body]
  `(future
     (try
       (let [result# (do ~@body)]
         (infof "%s returned %s" (quote  ~@body) result#)
         result#)
       (catch Exception e#
         (airbrake {:body (quote ~@body) :future true})
         (error e# "%s threw" (quote ~@body))
         (throw e#)))))

(defn start-worker [f & args]
  "Call fn with args as a worker. Returns the id of the worker"
  (let [fut (log-future
             (call-clojure-from-ruby f args))]
    (dosync
     (let [next-id (count @worker-store)]
       (infof "starting worker id %s for %s" next-id f)
       (alter worker-store assoc next-id fut)
       next-id))))

(defn fire-worker [f & args]
  "Start a worker, but don't wait for a response"
  (log-future (apply f args))
  nil)

(defn blocking-worker [f & args]
  "Start a worker and block until it returns"
  (call-clojure-from-ruby f args))

(defn worker-done? [id]
  "Return if the worker is done. Throw an NPE if there is no such worker"
  (dosync
   (let [as-int (int id)
         fut (get @worker-store as-int)]
     (future-done? fut))))

(defn wait-for-worker [id]
  "Block until the worker is done, and return it's result. Will only work once, next time it throws a NPE because the worker is no longer available"
  (dosync
   (let [as-int (int id)
         fut (get @worker-store as-int)]
     (alter worker-store dissoc as-int)
     ;; The actual clojure exception is wrapped in a ExecutionException.
     (try
       (let [result (deref fut)]
         result)
       (catch java.util.concurrent.ExecutionException e
         (throw (.getCause e)))))))

(defn worker-count []
  (dosync
   (count @worker-store)))
{'fraction_non_alphanumeric': 0.1359487279083317, 'fraction_numerical': 0.0, 'mean_word_length': 2.9493865030674846, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1422701', 'n_tokens_mistral': 1762, 'n_tokens_neox': 1705, 'n_words': 625}