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

progress on flink

This commit is contained in:
Sam Caldwell 2019-05-21 17:23:45 -04:00
parent 96e9431e15
commit deca0a82be
1 changed files with 68 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

128
racket/typed/examples/roles/flink.rkt
View File

@ -130,7 +130,8 @@ The JobManager then performs the job and, when finished, asserts (job-finished I
(TaskManager ID Int) (TaskManager ID Int)
(Observe (TaskManager ★/t ★/t)) (Observe (TaskManager ★/t ★/t))
(Job ID (List PendingTask)) (Job ID (List PendingTask))
(Observe (Job ★/t ★/t)))) (Observe (Job ★/t ★/t))
(JobFinished ID TaskResult)))
;; --------------------------------------------------------------------------------------------------- ;; ---------------------------------------------------------------------------------------------------
;; Util Macros ;; Util Macros
@ -328,66 +329,96 @@ The JobManager then performs the job and, when finished, asserts (job-finished I
(during (job (bind job-id ID) (bind tasks (List PendingTask))) (during (job (bind job-id ID) (bind tasks (List PendingTask)))
(log "JM receives job ~a" job-id) (log "JM receives job ~a" job-id)
(define n-r/r (partition-ready-tasks tasks)) (define-tuple (not-ready ready) (partition-ready-tasks tasks))
(define ready (select 1 n-r/r))
(define not-ready (select 0 n-r/r))
#;(define-values (ready not-ready) (partition task-ready? tasks))
(field [ready-tasks (List ConcreteTask) ready] (field [ready-tasks (List ConcreteTask) ready]
[waiting-tasks (List PendingTask) not-ready] [waiting-tasks (List PendingTask) not-ready]
[tasks-in-progress Int 0]) [tasks-in-progress Int 0])
#;(begin/dataflow (begin/dataflow
(define slots (slots-available)) (define slots (slots-available))
(define-values (ts readys) (define-tuple (ts readys)
(split-at/lenient (ready-tasks) slots)) (split-at/lenient (ref ready-tasks) slots))
(for ([t ts]) (for ([t ts])
(perform-task t push-results)) #f
#;(perform-task t push-results))
(unless (empty? ts) (unless (empty? ts)
;; the empty? check may be necessary to avoid a dataflow loop ;; the empty? check may be necessary to avoid a dataflow loop
(ready-tasks readys))) (set! ready-tasks readys)))
;; Task -> Void ;; Task -> Void
#;(define (add-ready-task! t) (define (add-ready-task! [t : ConcreteTask])
;; TODO - use functional-queue.rkt from ../../ ;; TODO - use functional-queue.rkt from ../../
(log "JM marks task ~a as ready" (task-id t)) (match-define (task (bind tid TaskID) discard) t)
(ready-tasks (cons t (ready-tasks)))) (log "JM marks task ~a as ready" tid)
(set! ready-tasks (cons t (ref ready-tasks))))
;; ID Data -> Void
;; Update any dependent tasks with the results of the given task, moving
;; them to the ready queue when possible
(define (push-results [task-id : TaskID]
[data : TaskResult])
(cond
[(and (zero? (ref tasks-in-progress))
(empty? (ref ready-tasks))
(empty? (ref waiting-tasks)))
(log "JM finished with job ~a" job-id)
(start-facet done (assert (job-finished job-id data)))]
[else
;; TODO - in MapReduce, there should be either 1 waiting task, or 0, meaning the job is done.
(define still-waiting
(for/fold ([ts : (List PendingTask) (list)])
([t (ref waiting-tasks)])
(define t+ (task+data t task-id data))
(match (task-ready? t+)
[(some (bind ready ConcreteTask))
(add-ready-task! ready)
ts]
[discard
(cons t+ ts)])))
(set! waiting-tasks still-waiting)]))
;; Task (ID TaskResult -> Void) -> Void ;; Task (ID TaskResult -> Void) -> Void
;; Requires (task-ready? t) ;; Requires (task-ready? t)
#;(define (perform-task t k) (define (perform-task [t : ConcreteTask]
(react [k : (→fn TaskID TaskResult (Tuple))]
(define task-facet (current-facet-id)) -> ★/t)
(on-start (tasks-in-progress (add1 (tasks-in-progress)))) (start-facet perform
(on-stop (tasks-in-progress (sub1 (tasks-in-progress)))) (on start (set! tasks-in-progress (add1 (ref tasks-in-progress))))
(match-define (task this-id desc) t) (on stop (set! tasks-in-progress (sub1 (ref tasks-in-progress))))
(match-define (task (bind this-id TaskID) (bind desc ConcreteWork)) t)
(log "JM begins on task ~a" this-id) (log "JM begins on task ~a" this-id)
(define (select-a-task-manager) (define (select-a-task-manager)
(react (start-facet this-facet
(begin/dataflow (begin/dataflow
(define mngr (define mngr?
(for/first ([(id slots) (in-hash (task-managers))] (for/first ([(id slots) (ref task-managers)]
#:when (positive? (- slots (hash-ref (requests-in-flight) id 0)))) #:when (positive? (- slots (hash-ref/failure (ref requests-in-flight) id 0))))
id)) id))
(when mngr (match mngr?
(take-slot! mngr) [(some (bind mngr ID))
(stop-current-facet (assign-task mngr)))))) (take-slot! mngr)
(stop this-facet
#;(assign-task mngr))]
[none
#f])
#f)))
;; ID -> ... ;; ID -> ...
(define (assign-task mngr) (define (assign-task [mngr : ID])
(react (start-facet this-facet
(define this-facet (current-facet-id)) (on (retracted (task-manager mngr discard))
(on (retracted (task-manager mngr _))
;; our task manager has crashed ;; our task manager has crashed
(stop-current-facet (select-a-task-manager))) (stop this-facet (select-a-task-manager)))
(on-start (on start
;; N.B. when this line was here, and not after `(when mngr ...)` above, ;; N.B. when this line was here, and not after `(when mngr ...)` above,
;; things didn't work. I think that due to script scheduling, all ready ;; things didn't work. I think that due to script scheduling, all ready
;; tasks were being assigned to the manager ;; tasks were being assigned to the manager
#;(take-slot! mngr) #;(take-slot! mngr)
(react (stop-when (asserted (task-state mngr job-id this-id _)) (start-facet take-slot
(received-answer! mngr))) (stop-when (asserted (task-state mngr job-id this-id discard))
(task-assigner t job-id mngr (received-answer! mngr)))
#;(task-assigner t job-id mngr
(lambda () (lambda ()
;; need to find a new task manager ;; need to find a new task manager
;; don't think we need a release-slot! here, because if we've heard back from a task manager, ;; don't think we need a release-slot! here, because if we've heard back from a task manager,
@ -396,32 +427,9 @@ The JobManager then performs the job and, when finished, asserts (job-finished I
(stop-facet this-facet (select-a-task-manager))) (stop-facet this-facet (select-a-task-manager)))
(lambda (results) (lambda (results)
(log "JM receives the results of task ~a" this-id) (log "JM receives the results of task ~a" this-id)
(stop-facet task-facet (k this-id results))))))) (stop-facet perform (k this-id results)))))))
(on-start (select-a-task-manager)))) (on start (select-a-task-manager))))
;; ID Data -> Void
;; Update any dependent tasks with the results of the given task, moving
;; them to the ready queue when possible
#;(define (push-results task-id data)
(cond
[(and (zero? (tasks-in-progress))
(empty? (ready-tasks))
(empty? (waiting-tasks)))
(log "JM finished with job ~a" job-id)
(react (assert (job-finished job-id data)))]
[else
;; TODO - in MapReduce, there should be either 1 waiting task, or 0, meaning the job is done.
(define still-waiting
(for/fold ([ts '()])
([t (in-list (waiting-tasks))])
(define t+ (task+data t task-id data))
(cond
[(task-ready? t+)
(add-ready-task! t+)
ts]
[else
(cons t+ ts)])))
(waiting-tasks still-waiting)]))
#f)))) #f))))