Lots of tests, bugfixing concurrency, large-scale outputs

Author: lrsjohnson

code/datasets.scm

@@ -1,12 +1,16 @@
 ;;; User-facing operations
 (define (mrs:create-data-set) (create-mrq-data-set))
 (define (mrs:create-sink-data-set) (create-sink-data-set))
-(define (mrs:create-file-writer-data-set) (create-file-writer-data-set))
+(define (mrs:create-file-writer-data-set filename)
+  (create-file-writer-data-set filename))
 (define (mrs:create-output-data-set) (create-output-data-set))
 
 ;;; General data-set operations
 (define *data-sets* '())
 
+(define (clear-data-sets)
+  (set! *data-sets* '()))
+
 (define (register-data-set data-set)
   (set! *data-sets* (cons data-set *data-sets*)))
 
@@ -175,8 +179,9 @@
     (lambda ()
       (if (file-writer-data-set-done data-set)
           (begin
-            (create-ds-elt-done)
-            (set-file-writer-data-set-done! data-set #f))
+	    (pp 'file-writer-get-done)
+            (set-file-writer-data-set-done! data-set #f)
+            (create-ds-elt-done))
           *empty-ds-elt*)))
   file-writer-data-set?)
 
@@ -238,7 +243,10 @@
 
 (defhandler ds-add-elt
   (lambda (data-set val)
-    ((output-data-set-output-callback data-set) val))
+    (let ((lock (ds-get-lock data-set)))
+      (conspire:acquire-lock lock)
+      ((output-data-set-output-callback data-set) val)
+      (conspire:unlock lock)))
   output-data-set?)
 
 (defhandler ds-get-reader

code/load.scm

@@ -6,4 +6,6 @@
 (load "data-loader")
 (load "multi-reader-queue")
 (load "workers")
-(load "pipes")
+(load "pipes")
+(load "main")
+(load "util")

code/main.scm

@@ -13,6 +13,7 @@
                       (callback 
                        (ds-elt-key ds-elt)
                        (ds-elt-value ds-elt)))))))
+	 (clear-data-sets)
          (thunk)
          (flush-input-data-sets)
          (let lp ()
@@ -32,19 +33,19 @@
       output-value)))
 
 #|
-  (define (test1)
+  (define (test-print-streaming)
     (define ds-input (mrs:create-data-set))
     (define ds1 (mrs:create-data-set))
     (mrs:print-streaming ds-input 'ds1)
     (mrs:feed-value-list ds-input '(1 2 3 4)))
-  (mrs:run-computation test1)
+  (mrs:run-computation test-print-streaming)
  ;->  (ds1 0 1)
  ;    (ds1 1 2)
  ;    (ds1 2 3)
  ;    (ds1 3 4)
  ;    (ds1 done)
 
-  (define (test2)
+  (define (test-basic-map))
     (define ds-input (mrs:create-data-set))
     (define ds1 (mrs:create-data-set))
     (mrs:map
@@ -54,14 +55,14 @@
      ds1)
     (mrs:print-streaming ds1 'ds1)
     (mrs:feed-value-list ds-input '(1 2 3 4)))
-  (mrs:run-computation test2)
+  (mrs:run-computation test-basic-map)
  ;   (ds1 3 40)
  ;   (ds1 0 10)
  ;   (ds1 1 20)
  ;   (ds1 2 30)
  ;   (ds1 done)
 
- (define (test3)
+ (define (test-chain-of-maps)
     (define ds-input (mrs:create-data-set))
     (define ds1 (mrs:create-data-set))
     (mrs:map
@@ -76,7 +77,7 @@
      ds1)
     (mrs:print-streaming ds1 'ds1)
     (mrs:feed-value-list ds-input '(1 2 3 4)))
-  (mrs:run-computation test3)
+  (mrs:run-computation test-chain-of-maps)
  ;  (ds1 1 20)
  ;  (ds1 2 30)
  ;  (ds1 3 40)
@@ -87,7 +88,7 @@
  ;  (ds1 0 11)
  ;  (ds1 done)
 
-  (define (test4)
+  (define (test-aggregate)
     (define ds-input (mrs:create-data-set))
     (define ds1 (mrs:create-data-set))
     (mrs:aggregate
@@ -97,14 +98,14 @@
     (mrs:feed-value-list ds-input '(1 2 3 4))
     (mrs:feed-value-list ds-input '(3 3 3 3))
     (mrs:feed-value-list ds-input '(1 4 3 4)))
-  (mrs:run-computation test4)
+  (mrs:run-computation test-aggregate)
  ; (ds1 3 (4 3 4))
  ; (ds1 2 (3 3 3))
  ; (ds1 1 (4 3 2))
  ; (ds1 0 (1 3 1))
  ; (ds1 done)
 
- (define (test5)
+ (define (test-map-reduce)
     (define ds-input (mrs:create-data-set))
     (define ds1 (mrs:create-data-set))
     (define ds2 (mrs:create-data-set))
@@ -123,7 +124,7 @@
     (mrs:feed-value-list ds-input '(1 1 3 5))
     (mrs:feed-value-list ds-input '(2 2 1 4))
     (mrs:feed-value-list ds-input '(3 1 0 3)))
-  (mrs:run-computation test5)
+  (mrs:run-computation test-map-reduce)
  ; (ds1 3 50)
  ; (ds1 0 20)
  ; (ds1 0 10)
@@ -161,5 +162,131 @@
     (mrs:feed-value-list ds-input '(1 2 3 4)))
   (mrs:run-computation test-with-output-data-set)
  ;;; Note that this returns directly!
- ;-> ((1 20) (0 10) (3 40) (2 30))
+ ;Value -> ((1 20) (0 10) (3 40) (2 30))
+
+ (define (test-with-user-specified-callback)
+   (define ds-input (mrs:create-data-set))
+   (define ds-output (mrs:create-output-data-set))
+   (mrs:map
+    (lambda (key value)
+      (mrs:emit key (* 10 value)))
+    ds-input
+    ds-output)
+   (mrs:feed-value-list ds-input '(1 2 3 4)))
+ (mrs:run-computation-with-callback
+  test-with-user-specified-callback
+  (lambda (k v)
+    (pp (list 'user k v))))
+ ;;; Printed:
+ ; (user 1 20)
+ ; (user 2 30)
+ ; (user 3 40)
+ ; (user 0 10)
+
+ (define (test-multiple-writers-to-output)
+    (define ds-input (mrs:create-data-set))
+    (define ds-output (mrs:create-output-data-set))
+    (mrs:map
+     (lambda (key value)
+       (mrs:emit key (* 10 value)))
+     ds-input
+     ds-output)
+    (mrs:map
+     (lambda (key value)
+       (mrs:emit key (* 11 value)))
+     ds-input
+     ds-output)
+    (mrs:feed-value-list ds-input '(1 2 3 4)))
+  (mrs:run-computation test-multiple-writers-to-output)
+ ;;; Note that this returns directly!
+ ;Value -> ((3 44) (2 33) (1 22) (0 11) (0 10) (3 40) (2 30) (1 20))
+
+ (define (test-file-writer-output-no-dependency)
+   (define ds-file-input (mrs:create-data-set))
+   (define ds-file-output (mrs:create-file-writer-data-set "output.txt"))
+   (define ds-main-input (mrs:create-data-set))
+   (define ds-main-output (mrs:create-output-data-set))
+   (mrs:map
+    (lambda (key value)
+      (pp 'file-map)
+      (conspire:thread-yield)
+      (pp 'emit)
+      (mrs:emit key (* 10 value)))
+    ds-file-input
+    ds-file-output)
+   (mrs:map
+    (lambda (key value)
+      (pp 'main-map)
+      (mrs:emit key (* 11 value)))
+    ds-main-input
+    ds-main-output)
+   (mrs:feed-value-list ds-main-input '(1 2 3))
+   (mrs:feed-value-list ds-file-input '(1 2 3 4 5 6 7 8 9 10 11 12)))
+ (mrs:run-computation test-file-writer-output-no-dependency)
+
+ ;;; Returned
+ ;Value 40: ((0 11) (2 33) (1 22))
+
+ ;;; BUT the file output.txt is empty! This is because the main
+ ;;; dataflow finished and returned before the file writing was
+ ;;; finished. We can fix that by making the done of the main dataflow
+ ;;; depend on the completion of the others:
+
+
+ (define (test-file-writer-output-with-dependency)
+   (define ds-file-input (mrs:create-data-set))
+   (define ds-file-output (mrs:create-file-writer-data-set "output.txt"))
+   (define ds-main-input (mrs:create-data-set))
+   (define ds-main-output (mrs:create-output-data-set))
+   (mrs:map
+    (lambda (key value)
+      (conspire:thread-yield)
+      (mrs:emit key (* 10 value)))
+    ds-file-input
+    ds-file-output)
+   (mrs:map
+    (lambda (key value)
+      (mrs:emit key (* 11 value)))
+    ds-main-input
+    ds-main-output)
+   (mrs:depends-on
+    ds-file-output
+    ds-main-output)
+   (mrs:feed-value-list ds-main-input '(1 2 3))
+   (mrs:feed-value-list ds-file-input '(1 2 3 4 5 6 7 8 9 10 11 12)))
+ (mrs:run-computation test-file-writer-output-with-dependency)
+ ;Value 28: ((0 11) (2 33) (1 22))
+ ;; Value is returned to the REPL
+
+ ;; And, unlike the test case above, the "output.txt" file contains all
+ ;; 12 key-value pairs:
+ ; (2 30)
+ ; (3 40)
+ ; (4 50)
+ ; (0 10)
+ ; (1 20)
+ ; (6 70)
+ ; (7 80)
+ ; (8 90)
+ ; (9 100)
+ ; (5 60)
+ ; (11 120)
+ ; (10 110)
+
+
+ (define (test-large-data-fact)
+   (define ds-fib-input (mrs:create-data-set))
+   (define ds-fib-output (mrs:create-file-writer-data-set "output.txt"))
+   (mrs:map
+    (lambda (key value)
+      (mrs:emit key (fact value)))
+    ds-fib-input
+    ds-fib-output)
+   (mrs:feed-value-list ds-fib-input (make-range 500)))
+ (mrs:run-computation test-large-data-fact)
+
+ ;;; Works! The output file has the first 500 factorials. They are not
+ ;;; in strictly increasing ordering, demonstrating that they are
+ ;;; indeed being distributed across our workers.
+
 |#

code/operations.scm

@@ -73,7 +73,9 @@
   (define (mm-func emit)
     (lambda (ds-elt)
       (if (ds-elt-done? ds-elt)
-          (emit ds-elt))))
+	  (begin
+	    (pp 'depends-on-done)
+	    (emit ds-elt)))))
   (make-distributor mm-func ds-in ds-out 1))
 
 

code/util.scm

@@ -0,0 +1,50 @@
+(define (make-range n)
+  (let lp ((i (- n 1))
+	   (range '()))
+    (if (< i 0)
+	range
+	(lp (- i 1)
+	    (cons i range)))))
+
+#|
+(make-range 5)
+:-> (0 1 2 3 4)
+|#
+
+(define (fact n)
+  (cond ((= n 0) 1)
+	(else (* n (fact (- n 1))))))
+
+#|
+ (map fact (make-range 10))
+ ;Value: (1 1 2 6 24 120 720 5040 40320 362880)
+|#
+
+(define (fib n)
+  (cond ((< n 2) n)
+	(else (+ (fib (- n 1))
+		 (fib (- n 2))))))
+
+#|
+(map fib (make-range 10))
+;Value:: (0 1 1 2 3 5 8 13 21 34)
+|#
+
+(define (sum lst)
+  (reduce + 0 lst))
+
+#|
+(sum '(1 3 2))
+;-> 6
+
+(sum '(1 2 3 4 5))
+;-> 15
+
+
+(sum '())
+;-> 0
+
+|#
+
+	   
+