<commit_before>;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) 2016 <NAME> <<EMAIL>>

(ns executors.core
  "A executos service abstraction layer."
  (:import java.util.concurrent.ForkJoinPool
           java.util.concurrent.CompletableFuture
           java.util.concurrent.ExecutorService
           java.util.concurrent.ThreadFactory
           java.util.concurrent.TimeUnit
           java.util.concurrent.Supplier
           java.util.concurrent.ScheduledExecutorService
           java.util.concurrent.Executors))

(def ^:const +max-priority+ Thread/MAX_PRIORITY)
(def ^:const +min-priority+ Thread/MIN_PRIORITY)
(def ^:const +norm-priority+ Thread/NORM_PRIORITY)

;; --- Protocols

(defprotocol IExecutor
  (^:private -execute [_ task] "Execute a task in a executor.")
  (^:private -submit [_ task] "Submit a task and return a promise."))

(defprotocol IScheduledExecutor
  (^:provate -schedule [_ ms task] "Schedule a task to execute in a future."))

(defprotocol IScheduledTask
  "A cancellation abstraction."
  (-cancel [_])
  (-cancelled? [_]))

;; --- Implementation

(defn- thread-factory-adapter
  "Adapt a simple clojure function into a
  ThreadFactory instance."
  [func]
  (reify ThreadFactory
    (^Thread newThread [_ ^Runnable runnable]
      (func runnable))))

(defn- thread-factory
  [{:keys [daemon priority]
    :or {daemon true
         priority Thread/NORM_PRIORITY}}]
  (thread-factory-adapter
   (fn [runnable]
     (let [thread (Thread. ^Runnable runnable)]
       (.setDaemon thread daemon)
       (.setPriority thread priority)
       thread))))

(defn- resolve-thread-factory
  [opts]
  (cond
    (map? opts) (thread-factory opts)
    (fn? opts) (thread-factory-adapter opts)
    (instance? ThreadFactory opts) opts
    :else (throw (ex-info "Invalid thread factory" {}))))

(deftype ScheduledTask [^Future fut]
  clojure.lang.IDeref
  (deref [_]
    (.get fut))

  clojure.lang.IBlockingDeref
  (deref [_ ms default]
    (try
      (.get fut ms TimeUnit/MILLISECONDS)
        (catch TimeoutException e
          default)))

    clojure.lang.IPending
    (isRealized [_] (and (.isDone fut)
                         (not (.isCancelled fut))))

    IScheduledTask
    (-cancelled? [_]
      (.isCancelled fut))

    (-cancel [_]
      (when-not (.isCancelled fut)
        (.cancel fut true))))

(extend-type ExecutorService
  IExecutor
  (-execute [this task]
    (CompletableFuture/runAsync ^Runnable task this))

  IExecutorService
  (-submit [this task]
    (let [supplier (reify Supplier (get [_] (task)))]
      (CompletableFuture/supplyAsync supplier this))))

(extend-type ScheduledExecutorService
  IScheduledExecutor
  (-schedule [_ ms func]
    (let [fut (.schedule this func ms TimeUnit/MILLISECONDS)]
      (ScheduledTask. fut))))

;; --- Public Api (Pool Constructors)

(defn common-pool
  "Get the common pool."
  []
  (ForkJoinPool/commonPool))

(defn cached
  "A cached thread pool constructor."
  ([]
   (Executors/newCachedThreadPool))
  ([opts]
   (let [factory (resolve-thread-factory opts)]
     (Executors/newCachedThreadPool factory))))<commit_msg>Fix various bugs on executors module.<commit_after>;; This Source Code Form is subject to the terms of the Mozilla Public
;; License, v. 2.0. If a copy of the MPL was not distributed with this
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
;;
;; Copyright (c) 2016 <NAME> <<EMAIL>>

(ns executors.core
  "A executos service abstraction layer."
  (:import java.util.function.Supplier
           java.util.concurrent.ForkJoinPool
           java.util.concurrent.Future
           java.util.concurrent.CompletableFuture
           java.util.concurrent.ExecutorService
           java.util.concurrent.TimeoutException
           java.util.concurrent.ThreadFactory
           java.util.concurrent.TimeUnit
           java.util.concurrent.ScheduledExecutorService
           java.util.concurrent.Executors))

(def ^:const +max-priority+ Thread/MAX_PRIORITY)
(def ^:const +min-priority+ Thread/MIN_PRIORITY)
(def ^:const +norm-priority+ Thread/NORM_PRIORITY)

;; --- Protocols

(defprotocol IExecutor
  (^:private -execute [_ task] "Execute a task in a executor.")
  (^:private -submit [_ task] "Submit a task and return a promise."))

(defprotocol IScheduledExecutor
  (^:provate -schedule [_ ms task] "Schedule a task to execute in a future."))

(defprotocol IScheduledTask
  "A cancellation abstraction."
  (-cancel [_])
  (-cancelled? [_]))

;; --- Implementation

(defn- thread-factory-adapter
  "Adapt a simple clojure function into a
  ThreadFactory instance."
  [func]
  (reify ThreadFactory
    (^Thread newThread [_ ^Runnable runnable]
      (func runnable))))

(defn- thread-factory
  [{:keys [daemon priority]
    :or {daemon true
         priority Thread/NORM_PRIORITY}}]
  (thread-factory-adapter
   (fn [runnable]
     (let [thread (Thread. ^Runnable runnable)]
       (.setDaemon thread daemon)
       (.setPriority thread priority)
       thread))))

(defn- resolve-thread-factory
  [opts]
  (cond
    (map? opts) (thread-factory opts)
    (fn? opts) (thread-factory-adapter opts)
    (instance? ThreadFactory opts) opts
    :else (throw (ex-info "Invalid thread factory" {}))))

(deftype ScheduledTask [^Future fut]
  clojure.lang.IDeref
  (deref [_]
    (.get fut))

  clojure.lang.IBlockingDeref
  (deref [_ ms default]
    (try
      (.get fut ms TimeUnit/MILLISECONDS)
        (catch TimeoutException e
          default)))

    clojure.lang.IPending
    (isRealized [_] (and (.isDone fut)
                         (not (.isCancelled fut))))

    IScheduledTask
    (-cancelled? [_]
      (.isCancelled fut))

    (-cancel [_]
      (when-not (.isCancelled fut)
        (.cancel fut true))))

(extend-type ExecutorService
  IExecutor
  (-execute [this task]
    (CompletableFuture/runAsync ^Runnable task this))

  (-submit [this task]
    (let [supplier (reify Supplier (get [_] (task)))]
      (CompletableFuture/supplyAsync supplier this))))

(extend-type ScheduledExecutorService
  IScheduledExecutor
  (-schedule [this ms func]
    (let [fut (.schedule this func ms TimeUnit/MILLISECONDS)]
      (ScheduledTask. fut))))

;; --- Public Api (Pool Constructors)

(defn common-pool
  "Get the common pool."
  []
  (ForkJoinPool/commonPool))

(defn cached
  "A cached thread pool constructor."
  ([]
   (Executors/newCachedThreadPool))
  ([opts]
   (let [factory (resolve-thread-factory opts)]
     (Executors/newCachedThreadPool factory))))
{'fraction_non_alphanumeric': 0.12357013847080073, 'fraction_numerical': 0.002408187838651415, 'mean_word_length': 3.142768079800499, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, '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': '5255002', 'n_tokens_mistral': 2165, 'n_tokens_neox': 2061, 'n_words': 572}