syndicate-lang
/
syndicate-2017
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

untyped flink: use interest as request for jobs

This commit is contained in:
Sam Caldwell 2019-10-16 16:21:00 -04:00
parent d29afb6679
commit 80ebab5ed7
1 changed files with 10 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
racket/syndicate/examples/actor/flink.rkt
View File

@ -116,11 +116,14 @@ TaskRunners.
Job Submission Protocol Job Submission Protocol
----------------------- -----------------------
Finally, Clients submit their jobs to the JobManager by asserting a Job, which is a (job ID (Listof Task)). Finally, Clients submit their jobs to the JobManager by asserting interest
The JobManager then performs the job and, when finished, asserts (job-finished ID TaskResult) (observe (job-completion ID (Listof Task) ★))
The JobManager then performs the job and, when finished, asserts
(job-completion ID (Listof Task) TaskResults)
|# |#
(assertion-struct job (id tasks)) (struct job (id tasks) #:transparent)
(assertion-struct job-finished (id data)) (assertion-struct job-completion (id data result))
;; --------------------------------------------------------------------------------------------------- ;; ---------------------------------------------------------------------------------------------------
;; Logging ;; Logging
@ -304,7 +307,7 @@ The JobManager then performs the job and, when finished, asserts (job-finished I
(define (received-answer! id) (define (received-answer! id)
(requests-in-flight (hash-update (requests-in-flight) id sub1))) (requests-in-flight (hash-update (requests-in-flight) id sub1)))
(during (job $job-id $tasks) (during (observe (job-completion $job-id $tasks _))
(log "JM receives job ~a" job-id) (log "JM receives job ~a" job-id)
(define-values (ready not-ready) (partition task-ready? tasks)) (define-values (ready not-ready) (partition task-ready? tasks))
(field [ready-tasks ready] (field [ready-tasks ready]
@ -388,7 +391,7 @@ The JobManager then performs the job and, when finished, asserts (job-finished I
(empty? (ready-tasks)) (empty? (ready-tasks))
(empty? (waiting-tasks))) (empty? (waiting-tasks)))
(log "JM finished with job ~a" job-id) (log "JM finished with job ~a" job-id)
(react (assert (job-finished job-id data)))] (react (assert (job-completion job-id tasks data)))]
[else [else
;; TODO - in MapReduce, there should be either 1 waiting task, or 0, meaning the job is done. ;; TODO - in MapReduce, there should be either 1 waiting task, or 0, meaning the job is done.
(define still-waiting (define still-waiting
@ -434,8 +437,7 @@ The JobManager then performs the job and, when finished, asserts (job-finished I
;; Job -> Void ;; Job -> Void
(define (spawn-client j) (define (spawn-client j)
(spawn (spawn
(assert j) (on (asserted (job-completion (job-id j) (job-tasks j) $data))
(on (asserted (job-finished (job-id j) $data))
(printf "job done!\n~a\n" data)))) (printf "job done!\n~a\n" data))))
;; --------------------------------------------------------------------------------------------------- ;; ---------------------------------------------------------------------------------------------------