Common Lisp implementation of the REDUCERS protocol

reducers.lisp

;;
;; originally posted:  http://paste.lisp.org/display/132849
;;

;; The fastest and easiest way to learn the essential concepts and motivation
;; for a more generalized notion of collections based on the minimal abstraction
;; of REDUCTIBLE, Rich Hickey gives an excellent overview on his original Clojure
;; implementation of core.reducers.  Note that it is not currently part of the 
;; default Clojure environment due to the requirement of the ForkJoin concurrency
;; facilities introduced in Java7 (or Java6 + JSR167Y)
;;
;; http://clojure.com/blog/2012/05/08/reducers-a-library-and-model-for-collection-processing.html

;; Once comfortable with the basic idea, the next question you might have is about
;; the implementation of the reducers provided as part of the standard baseline 
;; library and what the process would entail for one to extend similar capabilities
;; to a new kind of collection. This follow-up article discusses the implementation 
;; of new, customized reducers for this purpose. It finishes by touching on a few
;; basic ideas related to how parallel traversal of collections is able to become
;; conveniently transparent to the user-level code and introduces the fold operation
;; which provides that consistent user interface.
;;
;; http://clojure.com/blog/2012/05/15/anatomy-of-reducer.html

;; More about the concepts of collection parallelism and their relationship to the
;; reducer abstraction is described in this friendly overview of the key ideas. 
;; it also reviews them in relation to some well known prior art from Guy Steele's 
;; influential 2009 lecture on the subject.
;;
;; http://blog.guillermowinkler.com/blog/2013/12/01/whats-so-great-about-reducers/

;; There is a freely available chapter excerpted from "The Data Analysis Cookbook"
;; with a practical, detailed tutorial on concurrency and reducers. By demonstration
;; of concrete examples and use cases it shows that to model the concept of a 
;; collection simply as something that is REDUCTable results in a simple, elegant, 
;; abstraction that implicitly provides for inherent concurrent processing of content.
;;
;; http://ow.ly/tCfYl

;; So there seems a lot to like about getting parallelism "for free" by modeling 
;; it on the abstraction of reducers.  But what might be the cost of getting it
;; for free?  This paper describes an empirical study of the performance characteristics
;; and relative merits of ForkJoin concurrency used by reducers in comparison to
;; the more well known concurrency model of MapReduce.
;;
;; http://www.macs.hw.ac.uk/cs/techreps/docs/files/HW-MACS-TR-0096.pdf



;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; COMMON-LISP REDUCERS IMPLEMENTATION
;;
;; The current code makes use of the 'Extensible Sequences' extension available in
;; SBCL and ABCL to provide the basis of SEQUENCE class and REDUCE generic function
;; Naturally, if reducers were to be incorporated into CL21, we would want to 
;; replace these with our own equivalent.
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;


(ql:quickload :alexandria)

(defclass reducer (sequence standard-object)
  ((coll :initarg :coll)
   (xf :initarg :xf)))

(defun reducer (coll xf)
  (make-instance 'reducer :coll coll :xf xf))

(defclass folder (reducer sequence standard-object)
  ((coll :initarg :coll)
   (xf :initarg :xf)))

(defun folder (coll xf)
  (make-instance 'folder :coll coll :xf xf))

(defgeneric fold (n combinef reducef sequence))

(defmethod fold (n combinef reducef (sequence folder))
  (with-slots (coll xf) sequence
    (fold n combinef (funcall xf reducef) coll)))

(defmethod fold (n combinef reducef (sequence sequence))
    (sequence:reduce reducef sequence :initial-value (funcall combinef)))

(defmethod fold (n combinef reducef (sequence vector))
  (reduce combinef (lparallel:preduce-partial reducef sequence :initial-value (funcall combinef) :parts n)))

(defmethod sequence:reduce (function (sequence reducer)
			    &key from-end (start 0) end (initial-value nil iv-p))
  (with-slots (coll xf) sequence
    (sequence:reduce (funcall xf function)
		     coll
		     :initial-value (if iv-p
					initial-value
					(funcall function)))))

(defun mapping (f)
  (lambda (f1)
    (lambda (result input)
      (funcall f1 result (funcall f input)))))

(defun filtering (pred)
  (lambda (f1)
    (lambda (result input)
      (if (funcall pred input)
	  (funcall f1 result input)
	  result))))

(defun mapcatting (f)
  (lambda (f1)
    (lambda (result input)
      (sequence:reduce f1 (funcall f input) :initial-value result))))

(defun rmap (f coll)
  (folder coll (mapping f)))

(defun rfilter (pred coll)
  (folder coll (filtering pred)))

(defun rmapcat (f coll)
  (reducer coll (mapcatting f)))

(defun rcons (&optional x y)
  (when (or x y) (cons y x)))

(defmacro rbind (&rest reducers)
  (let ((reducers
	 (mapcar (lambda (reducer) `(alexandria:curry #',(car reducer) ,@(cdr reducer))) reducers)))
    `(alexandria:compose ,@reducers)))



;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; REDUCER EXAMPLES
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

;;
;;  (sequence:reduce #'+ (rmap #'1+ (rfilter #'evenp (alexandria:iota 100))))
;;
;;  (let ((pipeline   (rbind (rmap #'1+) (rfilter #'evenp)))
;;        (collection (alexandria:iota 100)))
;;     (sequence:reduce #'+ (funcall pipeline collection)))
;;;