{"size": 32454, "ext": "scrbl", "max_stars_repo_path": "scribblings/majordomo2.scrbl", "max_stars_repo_name": "dstorrs/majordomo2", "max_stars_repo_stars_event_min_datetime": null, "max_stars_repo_stars_event_max_datetime": null, "max_issues_repo_path": "scribblings/majordomo2.scrbl", "max_issues_repo_name": "dstorrs/majordomo2", "max_issues_repo_issues_event_min_datetime": null, "max_issues_repo_issues_event_max_datetime": null, "max_forks_repo_path": "scribblings/majordomo2.scrbl", "max_forks_repo_name": "dstorrs/majordomo2", "max_forks_repo_forks_event_min_datetime": null, "max_forks_repo_forks_event_max_datetime": null}
{"max_stars_count": null, "max_issues_count": null, "max_forks_count": null, "avg_line_length": 54.5445378151, "max_line_length": 701, "alphanum_fraction": 0.6002341776}
#lang scribble/manual

@(require (for-label racket majordomo2)
          racket/sandbox
          scribble/example)

@title{majordomo2}

@author{David K. Storrs}

@defmodule[majordomo2]

@section{Introduction}

@racketmodname[majordomo2] is a task manager.  It obsoletes the original @racketmodname[majordomo] package.

Major features include:

@itemlist[
          @item{Restart tasks that fail}
               @item{Carry state across restarts}
               @item{Automatic parallelization on request}
               @item{Pre-process results }
               @item{Sort results after pre-processing}
               @item{Post-process results}
               ]

@section{Demonstration}

The provided functions and data types are defined in the @secref["API"] section, but here are examples of practical usage.

@subsection{Simple Example}


@(define eval
   (call-with-trusted-sandbox-configuration
    (lambda ()
      (parameterize ([sandbox-output 'string]
                     [sandbox-error-output 'string]
                     [sandbox-memory-limit 50])
        [make-evaluator #:requires (list 'majordomo2) 'racket]))))

@examples[
          #:eval eval
          #:label #f

          (define (build-email-text msg names)
            (for/list ([name names])
              (format "Dear ~a: ~a" name msg)))

          (define names '("fred" "barney" "betty" "wilma" "bam-bam"))

          (define jarvis (start-majordomo))

          (define result-channel
            (add-task jarvis build-email-text "hi there" names))

          (channel? result-channel)

          (define result  (sync result-channel))
          (task? result) ; #t
          (pretty-print (task.data result))
          ;    '("Dear fred: hi there"
          ;      "Dear barney: hi there"
          ;      "Dear betty: hi there"
          ;      "Dear wilma: hi there"
          ;      "Dear bam-bam: hi there")

          @#reader scribble/comment-reader
          ; stop-majordomo shuts down all tasks that were added to that instance.  This
          @#reader scribble/comment-reader
          ; shuts down the instance custodian which will shut down the custodian for all
          @#reader scribble/comment-reader
          ; the tasks.

	  (stop-majordomo jarvis)
	  ]


@subsection{Sorting Task Results}

@examples[
          #:eval eval
          #:label #f

	  ;(require majordomo2)

          (define jarvis (start-majordomo))

          @#reader scribble/comment-reader
          (pretty-print
           ; Do the same thing as we did in the prior section, but get the results back
           ; sorted.
           (task.data (sync (add-task jarvis build-email-text "hi there" names
                                      #:sort-op string<?))))
          ; Result:
          ; '("Dear bam-bam: hi there"
          ;   "Dear barney: hi there"
          ;   "Dear betty: hi there"
          ;   "Dear fred: hi there"
          ;   "Dear wilma: hi there")

          @#reader scribble/comment-reader
          (define (mock-network-function . ip-addrs)
            ; In real life, this would (e.g.) connect to the specified IP address,
            ; send a message, and return whether or not it succeeded.  For purposes of
            ; this demonstration we'll have it return an arbitrary status.
            ;
            ; Note that tasks include a unique ID that is being returned.  This is
            ; autogenerated if not provided.
            (for/list ([addr ip-addrs]
                       [status (in-cycle '(success failure timeout))])
              (list (task.id (current-task)) addr status)))


          @#reader scribble/comment-reader
          (pretty-print
           ; This shows the baseline return value from mock-network-function before we do
           ; any sorting.
           (task.data (sync (add-task jarvis mock-network-function
                                      "4.4.4.4" "8.8.8.8" "172.67.188.90" "104.21.48.235"))))
          ; Returns:
          ; '((task-77546 "4.4.4.4" success)
          ;   (task-77546 "8.8.8.8" failure)
          ;   (task-77546 "172.67.188.90" timeout)
          ;   (task-77546 "104.21.48.235" success))

          @#reader scribble/comment-reader
          (pretty-print
           ; Now we'll sort the results based on their status.  NB: sort-cache-keys? is 
           ; unnecessary here since the key is cheap to calculate, but it's included for
	   ; sake of demonstration.
           (task.data (sync (add-task jarvis mock-network-function
                                      "4.4.4.4" "8.8.8.8" "172.67.188.90" "104.21.48.235"
				      #:sort-op symbol<?
				      #:sort-key last
				      #:sort-cache-keys? #t))))
          ; Returns:
          ; '((task-77546 "4.4.4.4" success)
          ;   (task-77546 "8.8.8.8" failure)
          ;   (task-77546 "172.67.188.90" timeout)
          ;   (task-77546 "104.21.48.235" success))

	  (stop-majordomo jarvis)
	  ]


@subsection{Pre- and Post-Processing}

@examples[
          #:eval eval
          #:label #f

          (define jarvis (start-majordomo))

          @#reader scribble/comment-reader
          (pretty-print
           ; This is a contrived and overly fancy example, but it demonstrates the
           ; functionality.  We'll generate the strings, append some text to the end after
           ; they've been generated, sort the strings by their length, and then put it all
           ; into titlecase.
           (task.data (sync (add-task jarvis build-email-text "hi there" names
                                      #:sort-op <
                                      #:sort-key string-length
                                      #:pre  (curry map (curryr string-append ", my friend."))
                                      #:post (curry map string-titlecase)))))
          ; Result for those reading the code:
          ; '("Dear Fred: Hi There, My Friend."
          ;   "Dear Betty: Hi There, My Friend."
          ;   "Dear Wilma: Hi There, My Friend."
          ;   "Dear Barney: Hi There, My Friend."
          ;   "Dear Bam-Bam: Hi There, My Friend.")

	  (stop-majordomo jarvis)
	  ]


@subsection[#:tag "filtering"]{Filtering Task Results}

If you're going to use pre/post processing and/or sorting, it's a good idea to ensure that all elements in your data set are appropriate.  To make this easy you can filter the results before preprocessing, sorting, and postprocessing happen in order to get rid of bad results.

@examples[
          #:eval eval
          #:label #f

          (define jarvis (start-majordomo))

          (define (simple)
            (success '(1 2 3 4 5 6 #f 'error)))

          @#reader scribble/comment-reader
          ;  Filtering the results happens before pre-processing
          (define result (sync (add-task jarvis
                                         simple
                                         #:filter  (and/c number? even?)
                                         #:pre     (λ (lst)
                                                     (if (memf odd? lst)
                                                         (raise 'failed)
                                                         lst))
                                         #:sort-op >)))
          (task.data result)

	  (stop-majordomo jarvis)
	  ]

@subsection{Restarting}

When a task fails, either because it threw an error or simply timed out, it will be restarted if there are retries left.  By default there will be 3 retries (meaning a total of 4 attempts to run the task), but you can use the @racket[#:retries] keyword to specify how many you want.  The argument must be a natural number or @racket[+inf.0].

@examples[
          #:eval eval
          #:label #f

          (define jarvis (start-majordomo))

          (define (failing-func)
            (displayln "inside failing-func")
            (raise-arguments-error 'failing-func "I don't feel so good"))

          @#reader scribble/comment-reader
	  (pretty-print
           ; If a function raises an error then the result will contain the value raised.
           ; By default it will restart 3 times, but in this example we don't want it to
           ; restart at all so we will specify 0 retries.
           (format "Data after failure was: ~a"
                   (task.data (sync (add-task jarvis failing-func #:retries 0)))))


          @#reader scribble/comment-reader
          (define (func-times-out . args)
            ; Sometimes an action will time out without explicitly throwing an error.  If
            ; so then if it has retries left it will be restarted again with all of its
            ; original arguments.  We can use the 'data' value in (current-task) to carry
            ; state across the restart.
            (define state (task.data (current-task)))
            (match state
              [(? hash-empty?)
               (displayln "Initial start.  Hello!")
               (update-data (hash-set* state
                                       'remaining-args args
                                       'results '()))
               (displayln "Intentional timeout to demonstrate restart.")
               (sync never-evt)]
              [(hash-table ('remaining-args (list current remaining ...)))
               (displayln (format "Marking ~a as processed..." current))
               (update-data (hash-set* state
                                       'remaining-args remaining
                                       'results (cons current
                                                      (hash-ref state 'results '()))))
               (displayln "Intentional timeout to demonstrate restart.")
               (sync never-evt)])
            (displayln "Goodbye."))


          @#reader scribble/comment-reader
          (let ([result (sync (add-task jarvis func-times-out 1 2 3 4 5 6
                                        #:keepalive 0.1 ; max time to complete/keepalive
                                        #:post (λ (h)
                                                 (hash-set h
                                                           'results (reverse (hash-ref h 'results '()))))))])
            ; Be sure to use struct* in your match pattern instead of struct.  The
	    ; task struct contains private fields that are not documented
	    ; or provided. Use the @racketmodname[struct-plus-plus] reflection API if you
	    ; really need to dig out the full field list.
            (match-define (struct* task ([status status] [data data])) result)
            (displayln (format "Final status was: ~v" status))
            (displayln (format "Final data was: ~v" data))
            )

          @#reader scribble/comment-reader
          (define (long-running-task)
            ; If a task is going to take a long time, it can periodically send a keepalive
            ; so that the manager knows not to restart it.  You can use @racket[update-data],
            ; @racket[success], @racket[failure], or @racket[keepalive] for that.
            (for ([i 10])
              (sleep 0.1)
              (keepalive))
            (success 'finished))

          @#reader scribble/comment-reader
          (let ([result (sync (add-task jarvis long-running-task #:retries 0 #:keepalive 0.25))])
            ; In this case we are saying that the task has failed if the manager doesn't
            ; hear from it after 0.25 seconds.
            (displayln (format "Final status of long-running task was: ~v" (task.status result)))
            (displayln (format "Final data of long-running task was: ~v" (task.data result))))

          @#reader scribble/comment-reader
          (let ([result (sync (add-task jarvis long-running-task #:retries 0 #:keepalive 0.05))])
            ; Here we let the task timeout.
            (displayln (format "Final status of long-running task was: ~v" (task.status result)))
            (displayln (format "Final data of long-running task was: ~v" (task.data result))))

	  (stop-majordomo jarvis)
	  ]

@subsection[#:tag "running-in-parallel"]{Parallelization}

@examples[
          #:eval eval
          #:label #f

          (define jarvis (start-majordomo))

          @#reader scribble/comment-reader
          (pretty-print
           ; Run the task in series to start with in order to show the baseline data.  (We
           ; already did this in a previous section but it's useful to repeat it here.)
           ; Note that the task ID is the same throughout.
           (task.data (sync (add-task jarvis mock-network-function
                                      "4.4.4.4" "8.8.8.8" "172.67.188.90" "104.21.48.235"))))

          @#reader scribble/comment-reader
          (pretty-print
           ; Parallelize the task such that each argument in the list is farmed out to a
           ; separate sub task and the results are compiled back together. NB: Because of
           ; how mock-network-function is defined, this results in an extra layer of
           ; listing, as we're getting multiple lists each containing the results of
           ; processing a single argument instead of one list containing the results of
           ; processing each argument in turn.  See below for how to handle this.  Note
           ; that the task ID is different for each entry.
           (task.data (sync (add-task jarvis mock-network-function
                                      #:parallel? #t
                                      "4.4.4.4" "8.8.8.8" "172.67.188.90" "104.21.48.235"))))

          @#reader scribble/comment-reader
          (pretty-print
           ; Same as above, but we'll append the sublists together in order to get back to
           ; the original results.  Note that once again the task ID is different for each
           ; entry.
           (task.data (sync (add-task jarvis mock-network-function
                                      #:parallel? #t
                                      #:post (curry apply append)
                                      "4.4.4.4" "8.8.8.8" "172.67.188.90" "104.21.48.235"))))


	  (stop-majordomo jarvis)
          ]

@subsection[#:tag "unwrapping"]{Unwrapping}

Often you'll have a function that wants individual arguments but you have the arguments as a list, because you generated them using a @racket[map] or got them from a database using @racket[query-rows] or etc.  You can save yourself an unnecessary @racket[apply] by using the @racket[#:unwrap?] argument.

@examples[
          #:eval eval
          #:label #f

          (define jarvis (start-majordomo))

          (define args (for/list ([i 5]) i))
          @#reader scribble/comment-reader
          ; This fails because we need to pass 0 1 2 3 4 separately instead of as a list
          (pretty-print
           (task.data (sync (add-task jarvis + args))))

          @#reader scribble/comment-reader
          ; This works because we are telling the task to unwrap the list into separate args
          (task.data (sync (add-task jarvis + args #:unwrap? #t)))
          ]

@subsection[#:tag "flattening"]{Flattening}

Sometimes, usually when running in parallel, you'll have a task where the data field ends up containing one or more tasks, each of which might also be nested, and what you actually want is the data from those subtasks.  You could dig that out on the far end or you could have the system do it for you.

@examples[
          #:eval eval
          #:label #f

          (define jarvis (start-majordomo))

        (define (make-task x)
          (task++ #:data x))

          @#reader scribble/comment-reader
          (let* ([jarvis (start-majordomo)]
               [without-flatten (sync (add-task jarvis make-task 0 1 2 #:parallel? #t))]
               [nested (sync (add-task jarvis make-task (for/list ([i 3]) (make-task i)) #:unwrap? #t #:parallel? #t))]
               [with-flatten    (sync (add-task jarvis make-task 0 1 2 (make-task (make-task 3))
                                                #:parallel? #t
                                                #:flatten-nested-tasks? #t))])
          (printf "data from without-flatten: ~v\n" (task.data without-flatten))
          (printf "data from nested tasks in without-flatten: ~v\n" (map task.data (task.data without-flatten)))
          (printf "data via explicit call to flatten-nested-tasks: ~v\n" (task.data (flatten-nested-tasks without-flatten)))
	  ;
	  ; Data from nested tasks gets lifted regardless of nesting depth
          (printf "data via #:flatten-nested-tasks?: ~v\n" (task.data with-flatten))

          (stop-majordomo jarvis))
          ]

Note that flattening loses the id and status fields of the subtasks.


@subsection[#:tag "convenience-functions"]{Convenience Functions}

Some operations are common enough that it's worth having a short form to avoid boilerplate.

@examples[
          #:eval eval
          #:label #f

          @#reader scribble/comment-reader
          ; A task that simply returns a specified value.  Generates its own majordomo
          (task.data (sync (task-return-value 'hello)))

          @#reader scribble/comment-reader
          ; Simplify adding task, syncing for result, and fetching data
          (get-task-data (start-majordomo) add1 7)
          ]

@section[#:tag "API"]{API}

@defproc*[([(start-majordomo) majordomo?]
           [(stop-majordomo) void?])]{Start and stop a @racket[majordomo] instance and all tasks it manages.  An instance contains a @racket[custodian] which manages resources created by any tasks given to that instance.  These resources will be cleaned up when @racket[stop-majordomo] is called.}

@defproc*[([(majordomo.id [instance majordomo])  any/c]
           [(majordomo-id [instance majordomo])  any/c])]{Retrieve the unique ID for this particular instance.  It's possible to have as many majordomo instances as desired.  You might have multiple ones in order to group tasks together so that it's easy to shut down a specific group without disturbing others.}

@defproc[(majordomo? [val any/c]) boolean?]{Predicate for the @racket[majordomo] struct.}

@defproc[(task-status/c [val (or/c 'success 'failure 'unspecified 'timeout)]) boolean?]{Contract for legal task status values.}

@defproc[
         #:kind "task constructor"
         (task++ [#:id     id     symbol?       (gensym "task-")]
                 [#:status status task-status/c 'unspecified]
                 [#:data   data   any/c         (hash)])
         task?]{A task struct encapsulates the details of a task.  It has a unique ID as an aid in debugging and organization, a status to identify what the outcome of the task was, and the data generated by the execution of the task.}

@defproc[(task? [the-task any/c]) boolean?]{Predicate for identifying tasks.}

@defproc*[([(task.id [the-task task?]) symbol?]
           [(task-id [the-task task?]) symbol?])
          ]{Accessors for the @racketid[id] field of a @racket[task] struct.}

@defproc*[([(task.status [the-task task?]) symbol?]
           [(task-status [the-task task?]) symbol?])
          ]{Accessors for the @racketid[status] field of a @racket[task] struct.}

@defproc*[([(task.data [the-task task?]) symbol?]
           [(task-data [the-task task?]) symbol?])
          ]{Accessors for the @racketid[data] field of a @racket[task] struct. See @secref["task-data"] for how the data field is used.}


@defproc*[([(is-success? [the-task task?]) boolean?]
           [(is-not-success? [the-task task?]) boolean?]
           [(is-failure? [the-task task?]) boolean?]
           [(is-timeout? [the-task task?]) boolean?]
           [(is-unspecified-status? [the-task task?]) boolean?])
          ]{Functions for testing the status of a @racketid[task] struct. @racket[is-not-success?] returns @racket[#t] if the task has any status other than @racket['success].}

@defproc[(set-task-id [the-task task?] [val symbol?]) task?]{Functional setter for the @racketid[id] field.}

@defproc[(set-task-status [the-task task?] [val task-status/c]) task?]{Functional setter for the @racketid[status] field.}

@defproc[(set-task-data [the-task task?] [val any/c]) task?]{Functional setter for the @racketid[data] field.  Prefer @racket[update-data] instead of using this directly, since that will also update the @racket[current-task] parameter and send a keepalive to the manager thread to delay restarting.}

@defparam[current-task t task? #:value #f]{A parameter that tracks the currently-running task.  It is updated when you call the @racket[update-data], @racket[success], and @racket[failure] functions.}

@defproc[(keepalive) void?]{Notify the manager that the task is still running.  This resets the timer and prevents the manager from restarting the task.}

@defproc[(success [arg any/c the-unsupplied-arg]) void?]{If @racketid[arg] was NOT @racket[the-unsupplied-arg], set the @racketid[data] field of @racket[current-task] to @racketid[arg].

Set the @racketid[status] field of @racket[current-task] to @racket['success].  Tell the manager that the worker has completed.  This will cause the manager to send the value of @racket[current-task] to the customer.}

@defproc[(failure [arg any/c the-unsupplied-arg]) void?]{If @racketid[arg] was NOT @racket[the-unsupplied-arg], set the @racketid[data] field of @racket[current-task] to @racketid[arg].

Set the @racketid[status] field of @racket[current-task] to @racket['failure].  Tell the manager that the worker has completed.  This will cause the manager to send the value of @racket[current-task] to the customer.}

@defproc[(update-data [val any/c]) void?]{Set the @racketid[data] field of @racket[current-task] to @racketid[val]. Send a keepalive to the manager.}


@subsection[#:tag "task-data"]{Task Data}

The @racket[current-task] parameter holds a @racket[task] structure that can be used to carry state across restarts, such as which arguments have already been processed. (Since the task is started with the same arguments every time.)  The data field is also useful for returning a value from the action.  See @secref["task-results"] for details.

There are three functions that an action can call to manipulate the contents of the @racket[current-task] struct:

@itemlist[
          @item{@racket[(update-data val)] will both update the data field of the @racket[current-task] and send a keepalive to the manager indicating that the worker is still running and does not need to be restarted.}
               @item{@racket[(success)] or @racket[(success arg)] will set the @racketid[status] field of the @racket[current-task] to @racket['success] and tell the manager that the worker has successfully completed.  If an argument was provided then the @racket[data] field will be updated to that value.}
               @item{@racket[(failure)] or @racket[(failure arg)] will set the @racketid[status] field of the @racket[current-task] to @racket['failure] and tell the manager that the worker has completed but failed.  If an argument was provided then the @racket[data] field will be updated to that value.}
               ]

If none of these functions is ever called then the data field will be set as follows:

@itemlist[
          @item{If the action completes without error, @racket[(success val)] will be called, where @racketid[val] is the return value of the action.}
               @item{If the action raises an error, @racket[(failure e)] will be called, where @racketid[e] is the raised value.}
               @item{If the action uses up all its retries and then times out, the data field is left undisturbed and the status is set to @racket['timeout].}
               ]

@subsection[#:tag "task-results"]{Task Results}

@racket[add-task] returns a channel.  When the task finishes, the content of the @racket[current-task] parameter is placed onto the channel.  The customer may retrieve the struct (via @racket[sync], @racket[channel-get], etc) and examine the @racketid[status] and @racketid[data] fields in order to determine how the task completed and what the final result was.

@section{Running tasks}

Tasks are created inside, and managed by, a @racket[majordomo] instance.

@defproc[(add-task [jarvis majordomo?]
                   [action (unconstrained-domain-> any/c)]
                   [arg                any/c] ...
                   [#:keepalive              keepalive-time (and/c real? (not/c negative?)) 5]
                   [#:retries                retries        (or/c natural-number/c +inf.0) 3]
                   [#:parallel?              parallel?      boolean? #f]
                   [#:unwrap?                unwrap?        boolean? #f]
                   [#:flatten-nested-tasks?  flatten-nested-tasks?    boolean? #f]
                   [#:filter                 filter-func    (or/c #f procedure?) #f]
                   [#:pre                    pre            procedure? identity]
                   [#:sort-op                sort-op        (or/c #f (-> any/c any/c any/c)) #f]
                   [#:sort-key               sort-key       (-> any/c any/c) identity]
                   [#:sort-cache-keys?       cache-keys?    boolean? #f]
                   [#:post                   post           procedure? identity])
         channel?]{
                   Add a task to a majordomo instance.

                       Two threads are created, a worker and a manager.  The worker does @racket[(apply action args)].  (See @secref["parallel-processing"] for an exception.) The manager does a @racket[sync/timeout keepalive-time] on the worker.  If the worker times out then the worker thread is killed and a new thread is started using the same arguments as the original and the most recent value of @racket[current-task], thereby allowing state to be carried over from one attempt to the next.  The keepalive timer resets whenever the worker does any of the following:

                       @itemlist[
                                 @item{Terminate (either naturally or by raising an error)}
                                      @item{Call @racket[keepalive]}
                                      @item{Call @racket[update-data]}
                                      @item{Call @racket[success]}
                                      @item{Call @racket[failure]}
                                      ]

                       Arguments are as follows:

                       @racketid[#:keepalive-time] The duration within which the worker must either terminate or notify the manager that it's still running.

                       @racketid[#:retries] The number of times that the manager should restart the worker before giving up.  (Note:  This counts @italic{retries}, not maximum attempts.  The maximum number of times your action will be started is retries + 1, with the +1 being the initial attempt.)

                       @racketid[#:parallel?] Whether the action should be run in parallel. See @secref["parallel-processing"].

                       @racketid[#:unwrap?] Whether the arguments should be used as-is or one layer of listing should be removed.  Saves you the trouble of currying in an @racket[apply]. See @secref["unwrapping"].

                       @racketid[#:flatten-nested-tasks?] If your action procedure produces a task that contains tasks you can have the system automatically lift the subtasks out and combine their data into the top-level task so that you don't have to dig it out on the far end.  You will lose the id and status fields of the subtasks if you do this.  See @secref["flattening"].

                       @racketid[#:filter] Filter the data after flattening (if any) and before passing it to the preprocessing function.  See @secref["filtering"].   

                       @racketid[#:pre] Pre-processes the results of the action.  The default preprocessor is @racket[identity].  This happens after flattening and before sorting.

                       @racketid[#:sort-op], @racketid[#:sort-key], @racketid[#:sort-cache-keys?] Whether and how to sort the results of the action.  They are passed as the (respectively) @racketid[less-than?], @racket[#:key extract-key] and @racket[#:cache-keys? cache-keys?] arguments to a @racket[sort] call.  Sorting happens after preprocessing and before postprocessing.

                       @racketid[#:post] Postprocesses the results of the action immediately before returning them.  The default is @racket[identity].

                       Obviously, if the results of your function are not a list, either leave @racketid[#:sort-op] as @racket[#f] so that it doesn't try to sort and fail, or else use @racket[#:pre list] to make it a list before sorting is applied.

Consolidating previous information, the pipeline goes:

@racketid[flatten] > @racketid[filter] > @racketid[preprocess] > @racketid[sort] > @racketid[postprocess]

Each step in the pipeline is optional.
                       }

@defproc[(task-return-value [val any/c]) channel?]{This provides a way to return a specific value using an interface equivalent to the normal @racket[add-task] system.  It returns a channel which syncs to a task with a data field containing the specified value.  A convenience function equivalent to:

            @racket[(add-task (start-majordomo) identity val)]}

@defproc[(get-task-data [jarvis majordomo?]
                        [action (unconstrained-domain-> any/c)]
                        [arg                any/c] ...
                        [#:keepalive              keepalive-time (and/c real? (not/c negative?)) 5]
                        [#:retries                retries        (or/c natural-number/c +inf.0) 3]
                        [#:parallel?              parallel?      boolean? #f]
                        [#:unwrap?                unwrap?        boolean? #f]
                        [#:flatten-nested-tasks?  flatten-nested-tasks?    boolean? #f]
                        [#:filter                 filter-func    (or/c #f procedure?) #f]
                        [#:pre                    pre            procedure? identity]
                        [#:sort-op                sort-op        (or/c #f (-> any/c any/c any/c)) #f]
                        [#:sort-key               sort-key       (-> any/c any/c) identity]
                        [#:sort-cache-keys?       cache-keys?    boolean? #f]
                        [#:post                   post           procedure? identity])
         channel?]{A convenience function that fetches the data for an action.  It is equivalent to:

                     @racket[(task.data (sync (add-task jarvis action args keyword-args)))]

                     The arguments and keywords all have the same meaning as @racket[add-task].}

@section[#:tag "parallel-processing"]{Parallel Processing}

As shown in the @secref["running-in-parallel"] demonstration, tasks can be parallelized simply by passing @racket[#:parallel? #t].  In this case @racket[add-task] will call itself recursively, creating subtasks and passing each of them one of the arguments in turn.  The main task will then wait for the subtasks to complete, aggregate their results into a list, and treat it as normal by running it through some or all of flattening, preprocessing, sorting, and postprocessing.  The subtasks will be run with the same @racket[#:keepalive] and @racket[#:retries] values as the main tasks but everything else will be the default, meaning that all preprocessing and sorting will happen in the main task.

Caveats:

@itemlist[
          @item{If your function returns a list and you run it in parallel mode then you will end up with a list of one-element lists.  If this isn't what you want then you can use @racket[#:pre (curry apply append)] to resolve it back into a single list.}
               @item{Obviously, your subtasks are running in separate threads.  Their @racket[current-task] will not be the same as the one in the original task.}
               @item{Only the data from the subtasks is returned, meaning that the status values are unavailable to the final customer.  One solution to this is to include the status into the data field or to throw an exception.  Better solutions are solicited.}
               ]

@section{Notes}

The @racket[task] structure is defined via the @racketmodname[struct-plus-plus] module, giving it a keyword constructor, dotted accessors, reflection data, etc.  As stated above, not all accessors are exported, so if you need to use @racket[match] on a struct, use @racket[struct*] instead of @racket[struct].
