Thanks for these clarifications—I would have added more comments to the code, but I actually wrote it yesterday as an experiment for myself.
The `place/context` form binds `terminate-pch` to the worker place's initial place-channel, the other end of which is part of the place descriptor value (bound to `place` in the loop inside the `start-places` function. So you are correct, it does not matter what value is sent to that place-channel: the worker place will always terminate when it receives a message that way. You are correct that, in this design, `get-work-pch` is shared among all of the worker places. This avoids the "busy wait" problem by letting the Racket scheduler deal with assigning jobs to the places. If you run (main) a few times, you will see that which place gets which work varies, and sometimes e.g. place 1 will do two jobs before place 0 does any. (You might be able to illustrate this more clearly by adding a call to `sleep` for a random delay to the handler for the work.) I think relying on the Racket scheduler to give out work to idle places will probably be more effective than essentially implementing your own scheduler. The comment about `send-work-pch` and `get-work-pch` has to do with the fact that each place effectively runs on its own instance of the Racket VM. That means that each place has its own instance of each module, so, if `send-work-pch` and `get-work-pch` were a module-level binding, each place would get their own version. That would mean that, for the worker places, nothing would ever send work and, for the main place, nothing would be listening for the work it sends. Your understanding of `send-work-pch` if essentially correct, though I would clarify that what is received is more a confirmation that a job was finished than a result. If you are trying to run arbitrary procedures that return results on the worker places, you presumably would want to implement a mechanism for associating each result to the procedure that produced it so it could be returned to the right place. The rationale for counting the number of jobs completed is to imitate the terminating step from your original example. Because `sync` makes a non-deterministic choice, in the worker places' loop, you cannot assume that `terminate-pch` being chosen means that `get-work-pch` was not also ready. Because `place-channel-put` is non-blocking, If you simply eliminated the result-counting step, `main` would tell the places to terminate as soon as it has finished sending all of the jobs, so the places might terminate with jobs left in the channel. Of course there are other ways to avoid this. In your original example, it wasn't a problem because the terminate message would always be behind all of the worker place's assigned jobs in the channel. Another alternative would be to re-write the worker places' loop function so that a message on `terminate-pch` wouldn't immediately terminate the place: instead the place will loop on any remaining work without blocking and will terminate when none is available. I'm pasting a revised version that takes this approach at the end of this message. I *think* yet another alternative, and perhaps the best one if I'm right that it would work, is not to explicitly terminate the places and just let them be garbage-collected when they are blocked on `get-work-pch` and `send-work-pch` becomes unreachable (because `main` has returned). This would also make `main` non-blocking. However, there are subtleties with readability and mutual blocking on place-channels, and I'd want to confirm that I'm right and the places really would get collected, which I don't have time to test right now. Of course, there are many more issues that would come up if you try to use the places to run arbitrary procedures (especially if you are interested in the results), some of which you identify, and I don't think I'm sufficiently expert in working with places to solve all of them in a general way. In terms of dealing with places that fail, one approach would be for `main` to launch a thread that would sync on the `place-dead-evt` for all of the places and call `exit` with the dead place's completion value (see: http://docs.racket-lang.org/reference/places.html#%28tech._completion._value%29) if it was non-zero. You could also try to recover, but that would be more complicated and might require knowing which job the place was working on when it died. A more complicated issue is that a client could send the places a (serialized) procedure that never returns. The right way of dealing with that would be a design decision. If your design is much like this example, in that the complete list of work is known when you call `main`, you could just decide that it is the caller's responsibility. I think the broader point, when you say you don't yet have a clear idea of what you want to use these places for and are trying to be completely generic, is that designing a general solution is much harder than solving for some specific scenario—even a relatively broad scenario, e.g. creating a pool of places to run a list of (serializable) thunks, ignore the results, and terminate when all the thunks have been run. I would definitely recommend becoming familiar with synchronizable events if you want to go further in this direction. Also, while they deal with concurrency rather than parallelism, I've found the tutorial <https://docs.racket-lang.org/more/> "More: Systems Programming with Racket" and the paper <https://www.cs.utah.edu/plt/publications/pldi04-ff.pdf> "Kill-Safe Synchronization Abstractions" helpful in thinking about these issues. -Philip Here is the revised version that avoids counting the number of jobs completed: #lang racket (define (place-output out-ch . args) (place-channel-put out-ch args)) (define (make-out-ch) (define-values (in-ch out-ch) (place-channel)) (thread (λ () (let loop () (apply printf (sync in-ch)) (newline) (loop)))) out-ch) (define (start-places get-work-pch out-ch) (for/list ([place-id (in-range (processor-count))]) (define place (place/context terminate-pch (place-output out-ch "Place ~a starting" place-id) (define (do-work data) ;; Include a random delay to make it more interesting. (define pause (random 5)) (place-output out-ch "Place ~s will do this work in ~a seconds: ~a." place-id pause data) (sleep pause) (place-output out-ch "Place ~s did this work: ~a." place-id data)) (define (post-terminate-loop) ;; This runs after the place has received a terminate message. (define work-remaining? ;; Do another job if any are remaining, but don't block. (sync/timeout 0 (wrap-evt get-work-pch do-work))) (cond [work-remaining? ;; Check for more jobs (post-terminate-loop)] [else ;; Time to really terminate (place-output out-ch "Place ~s is going to finish now." place-id)])) (let loop () (sync (handle-evt terminate-pch (λ (_) (place-output out-ch "Place ~s was asked to finish." place-id) (post-terminate-loop))) (handle-evt get-work-pch (λ (data) (do-work data) (loop))))))) (place-output out-ch "created place ~s" place-id) place)) (define (stop-places places) (for ([a-place (in-list places)]) (place-channel-put a-place 'terminate)) (for ([a-place (in-list places)]) (place-wait a-place))) (define (main [messages '(the quick brown fox jumped over the lazy dogs)]) (define-values (send-work-pch get-work-pch) (place-channel)) (define out-ch (make-out-ch)) (define places (start-places get-work-pch out-ch)) (for ([message messages]) (place-channel-put send-work-pch message)) (place-output out-ch "main has no more work data") (stop-places places)) (module* main #f (main)) On Sun, Jan 14, 2018 at 12:58 PM, Zelphir Kaltstahl < [email protected]> wrote: > Thank you again for your code. Some new concepts in it for me. > For example I did not know about events in Racket at all before. > I am currently trying to understand how the code works and how work is > distributed in it. > I will try to describe how each channel is used, hopefully understanding > how this works correctly. > Please tell me whether this is correct or some aspect is wrong. > > ## The `terminate-pch` ## > > I am not sure I understand when the `terminate-pch` is actually used. > I see that in the `stop-places` procedure some `'terminate` is put on the > channels' places, but that is a symbol and does not mention the > `terminate-pch`. > I have a guess: Is it implicitly that channel, because the place was > created with the `terminate-pch`? > In that case I understand the `terminate-pch`. It would also mean, that it > does not matter what I send on the `terminate-pch`, because the only action > for the child place is to terminate, no matter what it receives, as > indicated by the `_` (underscore) in the lambda in the `handle-evt` part. > > > ## `send-work-pch` ## > > This is the channel on which work is sent to the child places. > It is also the channel, on which the results from the child places are > received. > In the following part of the code the `send-work-pch` is used to check if > a message has been responded to and count the responses, until they are > equal to the number of sent messages. > > ~~~ > (unless (= done num-messages) > (place-output out-ch "message-counter: ~s" done) > (sync send-work-pch) > (loop (add1 done)))) > ~~~ > > ## `get-work-pch` ## > > This channel is exclusively used by the child places to receive work which > was put on the `sent-work-pch` by the main place. > > I only send the `get-work-pch` to the child places, because it is the > counterpart of the `send-work-pch`. This way they will be able to access > the work I am sending through the `send-work-pch` on their `get-work-pch`. > > ## The channel in general ## > > However there is a comment in your code at: > > ~~~ > (define-values (send-work-pch get-work-pch) > ;; this must not be at the module level, or each place will > ;; get its own version > (place-channel)) > ~~~ > > I thought I want a separate channel for each place to give it work once it > has none. > How is it decided, which place gets what work, if I do not have a separate > channel for each place? > Is the idea, that every child place can "take" work from the shared > `get-work-pch`? (circumventing the "Hey place, are you still busy?" problem) > > ## Other things ## > > What happens if there is no response to a message on the channel? (just > hypothetically, maybe if a server crashes or whatever) > Maybe on a single machine this is not a real issue though. > And maybe if something breaks it would not be as big a deal for single > machine programs, because you could simply run them again and hope for it > to finish, or if the machine breaks … well, it is broken and cannot run the > program anyway. > Maybe a place would always at least acknowledge that it is done, by > sending some symbol `'done` or something, so that the counting is correct. > (But what if a place somehow fails to do that? I would be stuck in counting > responses forever, I think.) > > Could I get away without counting messages? > I have a feeling (maybe unreasonable?) that this counting might be > problematic with a lot of work being distributed on many places. > Probably because it is a single point multiple processes have influence. > > I have the following idea: > Maybe I could however ignore unanswered messages, by only looking at the > work remaining and letting the child places send the main place the 'done > signal and when all places are 'done and I don't have any more work to > distribute, I am done. > The downside with this idea would be, that I would not know, if a child > place has somehow skipped or missed a message. > Oh and maybe if a place failed sending the 'done message, I would still be > stuck waiting for that place to be 'done. Hmmm. > > > On Sunday, January 14, 2018 at 3:46:01 PM UTC+1, Philip McGrath wrote: >> >> Below I have re-worked the example program you sent yesterday to send all >> the work to an explicitly-created place-channel, from which all of the >> worker places can get work. >> >> I also saw your question about sending procedures to places. The module >> web-server/lang/serial-lambda provides the serial-lambda form, but creates >> procedures that can be serialized using racket/serialize (as long as the >> contents of the closure are serializable): http://docs.racket-lang.org/we >> b-server-internal/closure.html Once serialized, these can be passed >> through place-channels, and they are far more robust than using eval. In >> general, the documentation for place-message-allowed? specifies what kind >> of messages can be passed through place-channels: >> http://docs.racket-lang.org/reference/places.html?q=place-me >> ssage-allowed%3F#%28def._%28%28lib._racket%2Fplace..rkt%29. >> _place-message-allowed~3f%29%29 >> >> #lang racket >> >> (define (place-output out-ch . args) >> (place-channel-put out-ch args)) >> >> (define (make-out-ch) >> (define-values (in-ch out-ch) >> (place-channel)) >> (thread (λ () >> (let loop () >> (apply printf (sync in-ch)) >> (newline) >> (loop)))) >> out-ch) >> >> (define (start-places get-work-pch out-ch) >> (for/list ([place-id (processor-count)]) >> (define place >> (place/context terminate-pch >> (place-output out-ch "place starting") >> (let loop () >> (sync (handle-evt >> terminate-pch >> (λ (_) >> (place-output out-ch >> "Place ~s is going to finish now." >> place-id))) >> (handle-evt >> get-work-pch >> (λ (data) >> (place-output out-ch >> "Place ~s got the following work: ~a." >> place-id >> data) >> (place-channel-put get-work-pch place-id) >> (loop))))))) >> (place-output out-ch "created place ~s" place-id) >> place)) >> >> (define (stop-places places) >> ;; simply terminate all places >> (for ([a-place (in-list places)]) >> (place-channel-put a-place 'terminate)) >> (for ([a-place (in-list places)]) >> (place-wait a-place))) >> >> (define (main [messages '(the quick brown fox jumped over the lazy dogs)]) >> (define-values (send-work-pch get-work-pch) >> ;; this must not be at the module level, or each place will >> ;; get its own version >> (place-channel)) >> (define out-ch >> (make-out-ch)) >> (define places >> (start-places get-work-pch out-ch)) >> (sleep 1) >> (define num-messages >> (for/fold ([n 0]) >> ([message messages]) >> (place-channel-put send-work-pch message) >> (add1 n))) >> (place-output out-ch "main has no more work data") >> (let loop ([done 0]) >> (unless (= done num-messages) >> (sync send-work-pch) >> (loop (add1 done)))) >> (place-output out-ch "all work reports finished") >> (stop-places places)) >> >> (module* main #f >> (main)) >> >> -Philip >> >> On Sun, Jan 14, 2018 at 8:18 AM, Zelphir Kaltstahl <[email protected] >> > wrote: >> >>> I also found the following post on StackOverflow: >>> https://stackoverflow.com/a/19095650/1829329 >>> >>> There the following timeout using construct is used: >>> >>> ~~~ >>> >>> (let loop ((i 0) (res '())) ; response loop >>> (if (= i l) >>> (reverse res) >>> (let ((response (sync/timeout 10 wch-c))) ; get answer with >>> timeout (place-channel-get blocks!) >>> (loop >>> (+ i 1) >>> (if response (cons response res) res)))))) >>> >>> ~~~ >>> >>> But it is not perfect, because the timeout time is wasted and could >>> already be used to distribute more work. >>> >>> >>> On Sunday, January 14, 2018 at 3:15:23 PM UTC+1, Zelphir Kaltstahl wrote: >>>> >>>> I have multiple places on which I would like to distribute some "work". >>>> I don't have a clear idea yet what that might be, so it is currently >>>> something abstract, which I don't want to make many assumptions about. >>>> Preferably I would like it to be so generic, that it could be anything. >>>> This is conceptually a "pool of places" I think. Such a pool of places is >>>> nice to have, because one does not always know the best way to parallelize >>>> things, unless it is a very simple case, where computation of all work >>>> items takes the same amount of time. With a pool which distributes the work >>>> itself, always giving the idle places work to do, it is like filling >>>> multiple containers with sand, rather than big stones which would leave >>>> gaps. >>>> >>>> I would like to "listen" on the channels of the places, until there is >>>> a message and then react on the message. >>>> The docs >>>> <https://docs.racket-lang.org/reference/places.html?q=place*#%28def._%28%28lib._racket%2Fplace..rkt%29._place-channel-get%29%29> >>>> say that `place-channel-get` is blocking. I also found the asynchronous >>>> buffered channels docs >>>> <https://docs.racket-lang.org/reference/async-channel.html>, but do >>>> not know if or how I could use them with places, or whether they are only >>>> for threads. >>>> The place-channel-get being blocking means that if I use that my work >>>> distribution would be blocked until I get some message from that place on >>>> its channel. >>>> The progress of computation would depend on that one checked channel >>>> giving me a message. >>>> >>>> I was thinking about putting such checks in threads, one for each >>>> channel, and then re-creating any thread that returns with some message >>>> from its place to send more work and get more messages, except if the work >>>> has all been distributed or the place has terminated. >>>> Somehow I would need to let the threads call a procedure to let them >>>> "signal", that they returned. >>>> Maybe 1 procedure for getting more work on the place and 1 procedure >>>> for signaling termination of the place. >>>> Otherwise I'd be doing busy wait again checking the threads for >>>> results, as I would have checked the channels without threads. >>>> >>>> But then I probably run into the next problem: How to manage the total >>>> available amount of work in a thread-safe manner? >>>> >>>> Is this even a feasible approach? >>>> Has there been work done that enables arbitrary computation (determined >>>> at run time, not compile time) in places distributing work to the places >>>> whenever they are no longer working on something? >>>> >>>> >>>> >>> -- >>> You received this message because you are subscribed to the Google >>> Groups "Racket Users" group. >>> To unsubscribe from this group and stop receiving emails from it, send >>> an email to [email protected]. >>> For more options, visit https://groups.google.com/d/optout. >>> >> >> > -- > You received this message because you are subscribed to the Google Groups > "Racket Users" group. > To unsubscribe from this group and stop receiving emails from it, send an > email to [email protected]. > For more options, visit https://groups.google.com/d/optout. > -- You received this message because you are subscribed to the Google Groups "Racket Users" group. To unsubscribe from this group and stop receiving emails from it, send an email to [email protected]. For more options, visit https://groups.google.com/d/optout.
