typed flink: unify task-state and task-assignment, job and job-finished
This commit is contained in:
parent
18fdcdeff7
commit
5823cf32c3
|
@ -50,12 +50,12 @@ Task Delegation Protocol
|
||||||
|
|
||||||
Task Delegation has two roles, TaskAssigner (TA) and TaskPerformer (TP).
|
Task Delegation has two roles, TaskAssigner (TA) and TaskPerformer (TP).
|
||||||
|
|
||||||
A TaskAssigner asserts the association of a Task with a particular TaskPerformer
|
A TaskAssigner requests the performance of a Task with a particular TaskPerformer
|
||||||
through
|
through the assertion of interest
|
||||||
(task-assignment ID Task)
|
(observe (task-performance ID Task ★))
|
||||||
where the ID identifies the TP
|
where the ID identifies the TP
|
||||||
|#
|
|#
|
||||||
(assertion-struct task-assignment : TaskAssignment (assignee task))
|
(assertion-struct task-performance : TaskPerformance (assignee task desc))
|
||||||
#|
|
#|
|
||||||
A Task is a (task TaskID Work), where Work is one of
|
A Task is a (task TaskID Work), where Work is one of
|
||||||
- (map-work String)
|
- (map-work String)
|
||||||
|
@ -91,9 +91,9 @@ A TaskResult is a (Hashof String Natural), counting the occurrences of words
|
||||||
(define-type-alias ConcreteTask
|
(define-type-alias ConcreteTask
|
||||||
(Task TaskID ConcreteWork))
|
(Task TaskID ConcreteWork))
|
||||||
#|
|
#|
|
||||||
The TaskPerformer responds to a task-assignment by describing its state with respect
|
The TaskPerformer responds to a request by describing its state with respect
|
||||||
to that task,
|
to that task,
|
||||||
(task-state TaskID TaskStateDesc)
|
(task-performance ID Task TaskStateDesc)
|
||||||
|
|
||||||
A TaskStateDesc is one of
|
A TaskStateDesc is one of
|
||||||
- ACCEPTED, when the TP has the resources to perform the task. (TODO - not sure if this is ever visible, currently)
|
- ACCEPTED, when the TP has the resources to perform the task. (TODO - not sure if this is ever visible, currently)
|
||||||
|
@ -101,7 +101,6 @@ A TaskStateDesc is one of
|
||||||
- RUNNING, indicating that the task is being performed
|
- RUNNING, indicating that the task is being performed
|
||||||
- (finished TaskResult), describing the results
|
- (finished TaskResult), describing the results
|
||||||
|#
|
|#
|
||||||
(assertion-struct task-state : TaskState (task-id desc))
|
|
||||||
(define-constructor* (finished : Finished data))
|
(define-constructor* (finished : Finished data))
|
||||||
(define-type-alias TaskStateDesc
|
(define-type-alias TaskStateDesc
|
||||||
(U Symbol (Finished TaskResult)))
|
(U Symbol (Finished TaskResult)))
|
||||||
|
@ -117,44 +116,46 @@ TaskRunners.
|
||||||
|
|
||||||
(define-type-alias TaskAssigner
|
(define-type-alias TaskAssigner
|
||||||
(Role (assign)
|
(Role (assign)
|
||||||
(Shares (TaskAssignment ID ConcreteTask))
|
(Shares (Observe (TaskPerformance ID ConcreteTask ★/t)))
|
||||||
;; would be nice to say how the TaskIDs relate to each other
|
;; would be nice to say how the TaskIDs relate to each other
|
||||||
(Reacts (Asserted (TaskState TaskID ★/t))
|
(Reacts (Asserted (TaskPerformance ID ConcreteTask ★/t))
|
||||||
(Branch (Stop assign)
|
(Branch (Stop assign)
|
||||||
(Effs)))))
|
(Effs)))))
|
||||||
|
|
||||||
(define-type-alias TaskPerformer
|
(define-type-alias TaskPerformer
|
||||||
(Role (listen)
|
(Role (listen)
|
||||||
(During (TaskAssignment ID ConcreteTask)
|
(During (Observe (TaskPerformance ID ConcreteTask ★/t))
|
||||||
;; would be nice to say how the IDs and TaskIDs relate to each other
|
;; would be nice to say how the IDs and TaskIDs relate to each other
|
||||||
(Shares (TaskState TaskID TaskStateDesc)))))
|
(Shares (TaskPerformance TaskID TaskStateDesc)))))
|
||||||
|
|
||||||
#|
|
#|
|
||||||
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) TaskResult)
|
||||||
|
|
||||||
|#
|
|#
|
||||||
(require-struct job #:as Job #:from "flink-support.rkt")
|
(require-struct job #:as Job #:from "flink-support.rkt")
|
||||||
(assertion-struct job-finished : JobFinished (id data))
|
(assertion-struct job-completion : JobCompletion (id data result))
|
||||||
(define-type-alias JobDesc (Job ID (List InputTask)))
|
(define-type-alias JobDesc (Job ID (List InputTask)))
|
||||||
|
|
||||||
(define-type-alias τc
|
(define-type-alias τc
|
||||||
(U (TaskRunner ID)
|
(U (TaskRunner ID)
|
||||||
(TaskAssignment ID ConcreteTask)
|
(Observe (TaskPerformance ID ConcreteTask ★/t))
|
||||||
(Observe (TaskAssignment ID ★/t))
|
(TaskPerformance ID ConcreteTask TaskStateDesc)
|
||||||
(TaskState TaskID TaskStateDesc)
|
(Observe (Observe (TaskPerformance ID ★/t ★/t)))
|
||||||
(Observe (TaskState TaskID ★/t))
|
|
||||||
(JobManagerAlive)
|
(JobManagerAlive)
|
||||||
(Observe (JobManagerAlive))
|
(Observe (JobManagerAlive))
|
||||||
(Observe (TaskRunner ★/t))
|
(Observe (TaskRunner ★/t))
|
||||||
(TaskManager ID Int)
|
(TaskManager ID Int)
|
||||||
(Observe (TaskManager ★/t ★/t))
|
(Observe (TaskManager ★/t ★/t))
|
||||||
JobDesc
|
(JobCompletion ID (List InputTask) TaskResult)
|
||||||
(Observe (Job ★/t ★/t))
|
(Observe (JobCompletion ID (List InputTask) ★/t))
|
||||||
(JobFinished ID TaskResult)
|
(Observe (Observe (JobCompletion ★/t ★/t ★/t)))))
|
||||||
(Observe (JobFinished ID ★/t))))
|
|
||||||
|
|
||||||
;; ---------------------------------------------------------------------------------------------------
|
;; ---------------------------------------------------------------------------------------------------
|
||||||
;; Util Macros
|
;; Util Macros
|
||||||
|
@ -194,9 +195,10 @@ The JobManager then performs the job and, when finished, asserts (job-finished I
|
||||||
(assert (task-runner id))
|
(assert (task-runner id))
|
||||||
(on (retracted (task-manager tm-id _))
|
(on (retracted (task-manager tm-id _))
|
||||||
(stop runner))
|
(stop runner))
|
||||||
(during (task-assignment id (task $task-id $desc))
|
(during (observe (task-performance id $t _))
|
||||||
|
(match-define (task $task-id $desc) t)
|
||||||
(field [state TaskStateDesc ACCEPTED])
|
(field [state TaskStateDesc ACCEPTED])
|
||||||
(assert (task-state task-id (ref state)))
|
(assert (task-performance id t (ref state)))
|
||||||
;; since we currently finish everything in one turn, these changes to status aren't
|
;; since we currently finish everything in one turn, these changes to status aren't
|
||||||
;; actually visible.
|
;; actually visible.
|
||||||
(set! state RUNNING)
|
(set! state RUNNING)
|
||||||
|
@ -257,19 +259,19 @@ The JobManager then performs the job and, when finished, asserts (job-finished I
|
||||||
[none
|
[none
|
||||||
(error "need to call can-accept? before selecting a runner")]))
|
(error "need to call can-accept? before selecting a runner")]))
|
||||||
|
|
||||||
(during (task-assignment id (task $task-id $desc))
|
(during (observe (task-performance id $t _))
|
||||||
|
(match-define (task $task-id $desc) t)
|
||||||
(define status0 : TaskStateDesc
|
(define status0 : TaskStateDesc
|
||||||
(if (can-accept?)
|
(if (can-accept?)
|
||||||
RUNNING
|
RUNNING
|
||||||
OVERLOAD/ts))
|
OVERLOAD/ts))
|
||||||
(field [status TaskStateDesc status0])
|
(field [status TaskStateDesc status0])
|
||||||
(assert (task-state task-id (ref status)))
|
(assert (task-performance id t (ref status)))
|
||||||
(when (can-accept?)
|
(when (can-accept?)
|
||||||
(define runner (select-runner))
|
(define runner (select-runner))
|
||||||
(log "TM ~a assigns task ~a to runner ~a" id task-id runner)
|
(log "TM ~a assigns task ~a to runner ~a" id task-id runner)
|
||||||
(on stop (set! busy-runners (set-remove (ref busy-runners) runner)))
|
(on stop (set! busy-runners (set-remove (ref busy-runners) runner)))
|
||||||
(assert (task-assignment runner (task task-id desc)))
|
(on (asserted (task-performance runner t $st))
|
||||||
(on (asserted (task-state task-id $st))
|
|
||||||
(match st
|
(match st
|
||||||
[ACCEPTED #f]
|
[ACCEPTED #f]
|
||||||
[RUNNING #f]
|
[RUNNING #f]
|
||||||
|
@ -278,6 +280,7 @@ The JobManager then performs the job and, when finished, asserts (job-finished I
|
||||||
[(finished discard)
|
[(finished discard)
|
||||||
(set! status st)])))))))))
|
(set! status st)])))))))))
|
||||||
|
|
||||||
|
|
||||||
;; ---------------------------------------------------------------------------------------------------
|
;; ---------------------------------------------------------------------------------------------------
|
||||||
;; JobManager
|
;; JobManager
|
||||||
|
|
||||||
|
@ -369,7 +372,7 @@ The JobManager then performs the job and, when finished, asserts (job-finished I
|
||||||
(define (received-answer! [id : ID])
|
(define (received-answer! [id : ID])
|
||||||
(set! requests-in-flight (hash-update (ref requests-in-flight) id sub1)))
|
(set! requests-in-flight (hash-update (ref 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 pending (for/list ([t tasks])
|
(define pending (for/list ([t tasks])
|
||||||
(input->pending-task t)))
|
(input->pending-task t)))
|
||||||
|
@ -427,7 +430,6 @@ The JobManager then performs the job and, when finished, asserts (job-finished I
|
||||||
;; ID -> ...
|
;; ID -> ...
|
||||||
(define (assign-task [mngr : ID])
|
(define (assign-task [mngr : ID])
|
||||||
(start-facet assign
|
(start-facet assign
|
||||||
(assert (task-assignment mngr t))
|
|
||||||
(know (assigned-task mngr))
|
(know (assigned-task mngr))
|
||||||
(on (retracted (task-manager mngr _))
|
(on (retracted (task-manager mngr _))
|
||||||
;; our task manager has crashed
|
;; our task manager has crashed
|
||||||
|
@ -438,10 +440,10 @@ The JobManager then performs the job and, when finished, asserts (job-finished I
|
||||||
;; tasks were being assigned to the manager
|
;; tasks were being assigned to the manager
|
||||||
#;(take-slot! mngr)
|
#;(take-slot! mngr)
|
||||||
(start-facet take-slot
|
(start-facet take-slot
|
||||||
(on (asserted (task-state this-id _))
|
(on (asserted (task-performance mngr t _))
|
||||||
(stop take-slot
|
(stop take-slot
|
||||||
(received-answer! mngr)))))
|
(received-answer! mngr)))))
|
||||||
(on (asserted (task-state this-id $status))
|
(on (asserted (task-performance mngr t $status))
|
||||||
(match status
|
(match status
|
||||||
[ACCEPTED #f]
|
[ACCEPTED #f]
|
||||||
[RUNNING #f]
|
[RUNNING #f]
|
||||||
|
@ -487,7 +489,7 @@ The JobManager then performs the job and, when finished, asserts (job-finished I
|
||||||
(start-facet delegate-tasks
|
(start-facet delegate-tasks
|
||||||
(on (realize (tasks-finished job-id $data:TaskResult))
|
(on (realize (tasks-finished job-id $data:TaskResult))
|
||||||
(stop delegate-tasks
|
(stop delegate-tasks
|
||||||
(start-facet done (assert (job-finished job-id data)))))
|
(start-facet done (assert (job-completion job-id tasks data)))))
|
||||||
(begin/dataflow
|
(begin/dataflow
|
||||||
(define slots (slots-available))
|
(define slots (slots-available))
|
||||||
(define-tuple (ts readys)
|
(define-tuple (ts readys)
|
||||||
|
@ -505,9 +507,8 @@ The JobManager then performs the job and, when finished, asserts (job-finished I
|
||||||
(define (spawn-client [j : JobDesc])
|
(define (spawn-client [j : JobDesc])
|
||||||
(spawn τc
|
(spawn τc
|
||||||
(start-facet _
|
(start-facet _
|
||||||
(match-define (job $id _) j)
|
(match-define (job $id $tasks) j)
|
||||||
(assert j)
|
(on (asserted (job-completion id tasks $data))
|
||||||
(on (asserted (job-finished id $data))
|
|
||||||
(printf "job done!\n~a\n" data)))))
|
(printf "job done!\n~a\n" data)))))
|
||||||
|
|
||||||
;; ---------------------------------------------------------------------------------------------------
|
;; ---------------------------------------------------------------------------------------------------
|
||||||
|
|
Loading…
Reference in New Issue