Futures (#441)

* Futures

* Type checking fixes

* Lintman

* Actual lintman

* Clean up a bit

* Tests

* Slightly smarter definitions of pcalls and pvalues

* Appropriately chunk futures in pmap

* Improve the docstring a bit

* Juicy

* Correct ordering issues

* pmap, pcalls, pvalues tests

* Use newer Docker image

* New dependencies I guess

* Fix the importer bug

* Fix seq iterator stack overflow

* Py 3.6 dependencies

* Fix MyPy

* Address attr cmp and eq changes

* Changelog

* Be literally a lot smarter

* Update changelog

* Type check it

Author: chrisrink10

CHANGELOG.md

@@ -5,6 +5,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+### Added
+ * Added support for `future`s (#441)
 
 ## [v0.1.dev13] - 2020-03-16
 ### Added

src/basilisp/core.lpy

@@ -42,9 +42,9 @@
   ^{:doc      "Creates a new sequence where o is the first element and seq is the rest.
                If seq is nil, return a list containing o. If seq is not a Seq, attempt
                to coerce it to a Seq and then cons o onto the resulting sequence."
-    :arglists '([v seq])}
+    :arglists '([o seq])}
   cons
-  (fn* cons [v seq] (basilisp.lang.runtime/cons v seq)))
+  (fn* cons [o seq] (basilisp.lang.runtime/cons o seq)))
 
 (def ^{:doc      "Return the metadata from o, or nil if there is no metadata."
        :arglists '([o])}
@@ -807,7 +807,8 @@
   (basilisp.lang.atom/Atom v))
 
 (defn realized?
-  "Return true if the delay or lazy sequence has been realized."
+  "Return true if the delay, future, lazy sequence, or promise has been
+  realized."
   [o]
   (.-is-realized o))
 
@@ -847,6 +848,56 @@
   [p v]
   (.deliver p v))
 
+;;;;;;;;;;;;;
+;; Futures ;;
+;;;;;;;;;;;;;
+
+(import* atexit)
+
+(def ^:dynamic *executor-pool*
+  (basilisp.lang.futures/ThreadPoolExecutor))
+
+(atexit/register (.-shutdown *executor-pool*))
+
+(defn future-call
+  "Call the no args function f in another thread. Returns a Future object.
+  The value returned by f can be fetched using `deref` or `@`, though
+  doing so may block unless the `deref` with a timeout argument is used."
+  ([f]
+   (future-call f *executor-pool*))
+  ([f pool]
+   (.submit pool f)))
+
+(defmacro future
+  "Execute the expressions of body in another thread. Returns a Future object.
+  The value returned by the body can be fetched using `deref` or `@`, though
+  doing so may block unless the `deref` with a timeout argument is used."
+  [& body]
+  `(future-call
+    (fn* []
+         ~@body)))
+
+(defn future-cancel
+  "Attempt to cancel the Future fut. If the future can be cancelled, return
+  true. Otherwise, return false."
+  [fut]
+  (.cancel fut))
+
+(defn future?
+  "Return true if x is a future, false otherwise."
+  [x]
+  (instance? basilisp.lang.futures/Future x))
+
+(defn future-cancelled?
+  "Return true if the Future fut has been cancelled, false otherwise."
+  [fut]
+  (.cancelled fut))
+
+(defn future-done?
+  "Return true if the Future fut is done, false otherwise."
+  [fut]
+  (.done fut))
+
 ;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; Arithmetic Functions ;;
 ;;;;;;;;;;;;;;;;;;;;;;;;;;
@@ -2571,6 +2622,58 @@
       (recur (ret))
       ret)))
 
+(import* multiprocessing)
+
+(def ^:dynamic *pmap-cpu-count*
+  (* 2 (multiprocessing/cpu-count)))
+
+(defn pmap
+  "Apply f as by map, but in parallel using futures.
+
+  This may only be useful for functions which are mainly blocked on IO, since
+  Python threads do not allow parallel computation using threads.
+
+  This function is not fully lazy. Chunks of elements from the input collection
+  are grabbed eagerly to spawn Futures of `f`. Elements after the chunk size in
+  the input collection are grabbed lazily. The chunk size is set by
+  `*pmap-cpu-count*` and defaults to 2 times the number of CPU cores on the
+  machine; the value of `*pmap-cpu-count*` is captured when `pmap` is first
+  called, so it should be safe to bind in the initial calling thread. The
+  sequence of futures is fed into a final (fully lazy) sequence to deref the
+  Future and return its final value."
+  ([f coll]
+   (lazy-seq
+    (when (seq coll)
+      (let [chunk-size *pmap-cpu-count*
+            futs       (mapv #(future (f %)) (take chunk-size coll))]
+        (concat (map deref futs)
+                (pmap f (drop chunk-size coll)))))))
+  ([f coll & colls]
+   (lazy-seq
+    (when (seq coll)
+      (let [chunk-size *pmap-cpu-count*
+            futs       (apply mapv (fn [& args]
+                                     (future (apply f args)))
+                              (take chunk-size coll)
+                              (map #(take chunk-size %) colls))]
+        (concat (map deref futs)
+                (apply pmap f (drop chunk-size coll) (map #(drop chunk-size %) colls))))))))
+
+(defn pcalls
+  "Return a lazy seq of the result of executing the no arg functions fns,
+  which will be evaluated in parallel."
+  [& fns]
+  (pmap (fn [f] (f)) fns))
+
+(defmacro pvalues
+  "Returns a lazy seq of the result of exprs, which will be evaluated in
+  parallel."
+  [& exprs]
+  `(pcalls
+    ~@(map (fn [expr]
+             (concat '(fn* []) [expr]))
+           exprs)))
+
 ;;;;;;;;;;;;;;;;;;;;;;
 ;; Random Functions ;;
 ;;;;;;;;;;;;;;;;;;;;;;

src/basilisp/lang/compiler/generator.py

@@ -467,6 +467,7 @@ def _var_ns_as_python_sym(name: str) -> str:
 _CORE_ALIAS = genname("core")
 _DELAY_ALIAS = genname("delay")
 _EXC_ALIAS = genname("exc")
+_FUTURES_ALIAS = genname("futures")
 _INTERFACES_ALIAS = genname("interfaces")
 _KW_ALIAS = genname("kw")
 _LIST_ALIAS = genname("llist")
@@ -488,6 +489,7 @@ def _var_ns_as_python_sym(name: str) -> str:
     "basilisp.core": _CORE_ALIAS,
     "basilisp.lang.delay": _DELAY_ALIAS,
     "basilisp.lang.exception": _EXC_ALIAS,
+    "basilisp.lang.futures": _FUTURES_ALIAS,
     "basilisp.lang.interfaces": _INTERFACES_ALIAS,
     "basilisp.lang.keyword": _KW_ALIAS,
     "basilisp.lang.list": _LIST_ALIAS,

src/basilisp/lang/futures.py

@@ -0,0 +1,84 @@
+from concurrent.futures import (  # noqa # pylint: disable=unused-import
+    Future as _Future,
+    ProcessPoolExecutor as _ProcessPoolExecutor,
+    ThreadPoolExecutor as _ThreadPoolExecutor,
+    TimeoutError as _TimeoutError,
+)
+from typing import Callable, Optional, TypeVar
+
+import attr
+
+from basilisp.lang.interfaces import IBlockingDeref
+
+_CMP_OFF = getattr(attr, "__version_info__", (0,)) >= (19, 2)
+
+T = TypeVar("T")
+
+
+@attr.s(  # type: ignore
+    auto_attribs=True,
+    **({"eq": True} if _CMP_OFF else {"cmp": False}),
+    frozen=True,
+    repr=False,
+    slots=True,
+)
+class Future(IBlockingDeref[T]):
+    _future: "_Future[T]"
+
+    def __repr__(self):  # pragma: no cover
+        return self._future.__repr__()
+
+    def cancel(self) -> bool:
+        return self._future.cancel()
+
+    def cancelled(self) -> bool:
+        return self._future.cancelled()
+
+    def deref(
+        self, timeout: Optional[float] = None, timeout_val: Optional[T] = None
+    ) -> Optional[T]:
+        try:
+            return self._future.result(timeout=timeout)
+        except _TimeoutError:
+            return timeout_val
+
+    def done(self) -> bool:
+        return self._future.done()
+
+    @property
+    def is_realized(self) -> bool:
+        return self.done()
+
+    # Pass `Future.result(timeout=...)` through so `Executor.map(...)` can
+    # still work with this Future wrapper.
+    def result(self, timeout: Optional[float] = None) -> T:
+        return self._future.result(timeout=timeout)
+
+
+# Basilisp's standard Future executor is the `ThreadPoolExecutor`, but since
+# it is set via a dynamic variable, it can be rebound using the binding macro.
+# Callers may wish to use a process pool if they have CPU bound work.
+
+
+class ProcessPoolExecutor(_ProcessPoolExecutor):  # pragma: no cover
+    def __init__(self, max_workers: Optional[int] = None):
+        super().__init__(max_workers=max_workers)
+
+    def submit(  # type: ignore
+        self, fn: Callable[..., T], *args, **kwargs
+    ) -> "Future[T]":
+        return Future(super().submit(fn, *args, **kwargs))
+
+
+class ThreadPoolExecutor(_ThreadPoolExecutor):
+    def __init__(
+        self,
+        max_workers: Optional[int] = None,
+        thread_name_prefix: str = "basilisp-futures",
+    ):
+        super().__init__(max_workers=max_workers, thread_name_prefix=thread_name_prefix)
+
+    def submit(  # type: ignore
+        self, fn: Callable[..., T], *args, **kwargs
+    ) -> "Future[T]":
+        return Future(super().submit(fn, *args, **kwargs))

src/basilisp/lang/runtime.py

@@ -409,6 +409,7 @@ class Namespace(ReferenceBase):
                 "basilisp.lang.compiler",
                 "basilisp.lang.delay",
                 "basilisp.lang.exception",
+                "basilisp.lang.futures",
                 "basilisp.lang.interfaces",
                 "basilisp.lang.keyword",
                 "basilisp.lang.list",
@@ -1052,10 +1053,10 @@ def deref(o, timeout_s=None, timeout_val=None):
     timeout_val are supplied, deref will wait at most timeout_s seconds,
     returning timeout_val if timeout_s seconds elapse and o has not
     returned."""
-    if isinstance(o, IDeref):
-        return o.deref()
-    elif isinstance(o, IBlockingDeref):
+    if isinstance(o, IBlockingDeref):
         return o.deref(timeout_s, timeout_val)
+    elif isinstance(o, IDeref):
+        return o.deref()
     raise TypeError(f"Object of type {type(o)} cannot be dereferenced")
 
 
@@ -1370,7 +1371,7 @@ def decorator(f):
 
 def _fn_with_meta(f, meta: Optional[lmap.Map]):
     """Return a new function with the given meta. If the function f already
-    has a meta map, then merge the """
+    has a meta map, then merge the new meta with the existing meta."""
 
     if not isinstance(meta, lmap.Map):
         raise TypeError("meta must be a map")

tests/basilisp/core_fns_test.lpy

@@ -1,4 +1,5 @@
 (ns tests.basilisp.core-fns-test
+  (:import time)
   (:require
    [basilisp.test :refer [deftest are is testing]]))
 
@@ -335,6 +336,60 @@
   (is (= "lo w" (subs "hello world" 3 7)))
   (is (thrown? python/IndexError (subs "hello world" 12 3))))
 
+(deftest futures-test
+  (testing "successful future"
+    (let [fut (future 1)]
+      (is (= 1 @fut))
+      (is (= 1 @fut))
+      (is (= false (future-cancel fut)))
+      (is (= false (future-cancelled? fut)))
+      (is (= true (future-done? fut)))
+      (is (= true (realized? fut)))))
+
+  (testing "timed deref of future"
+    (let [fut (future (time/sleep 3))]
+      (is (= :timed-out (deref fut 0.01 :timed-out)))
+      (is (= false (future-cancelled? fut)))
+      (is (= false (future-done? fut)))
+      ;; can't always cancel a sleep-ed Future
+      (future-cancel fut))))
+
+(deftest pmap-test
+  (binding [*pmap-cpu-count* 2]
+    (let [slow (fn slow [x]
+                 (time/sleep 0.5)
+                 (+ x 10))]
+      (is (= [] (vec (pmap slow []))))
+      (is (= [11] (vec (pmap slow [1]))))
+      (is (= [11 12 13 14 15 16 17 18 19]
+             (vec (pmap slow (range 1 10))))))))
+
+(deftest pcalls-test
+  (binding [*pmap-cpu-count* 2]
+    (let [slow (fn slow [x]
+                 (time/sleep 0.5)
+                 (+ x 10))]
+      (is (= [] (vec (pcalls))))
+      (is (= [11] (vec (pcalls #(slow 1)))))
+      (is (= [11 12 13 14]
+             (vec (pcalls #(slow 1)
+                          #(slow 2)
+                          #(slow 3)
+                          #(slow 4))))))))
+
+(deftest pvalues-test
+  (binding [*pmap-cpu-count* 2]
+    (let [slow (fn slow [x]
+                 (time/sleep 0.5)
+                 (+ x 10))]
+      (is (= [] (vec (pvalues))))
+      (is (= [11] (vec (pvalues (slow 1)))))
+      (is (= [11 12 13 14]
+             (vec (pvalues (slow 1)
+                           (slow 2)
+                           (slow 3)
+                           (slow 4))))))))
+
 (deftest to-array-test
   (is (= #py [] (to-array [])))
   (is (= #py [] (to-array '())))