basis/concurrency/distributed/distributed.factor

   1 ! Copyright (C) 2005 Chris Double. All Rights Reserved.
   2 ! See http://factorcode.org/license.txt for BSD license.
   3 USING: serialize sequences concurrency.messaging threads io
   4 io.servers.connection io.encodings.binary assocs init
   5 arrays namespaces kernel accessors ;
   6 FROM: io.sockets => host-name <inet> with-client ;
   7 IN: concurrency.distributed
   8
   9 <PRIVATE
  10
  11 : registered-processes ( -- hash )
  12    \ registered-processes get-global ;
  13
  14 PRIVATE>
  15
  16 : register-process ( name process -- )
  17     swap registered-processes set-at ;
  18
  19 : unregister-process ( name -- )
  20     registered-processes delete-at ;
  21
  22 : get-process ( name -- process )
  23     dup registered-processes at [ ] [ thread ] ?if ;
  24
  25 SYMBOL: local-node
  26
  27 : handle-node-client ( -- )
  28     deserialize
  29     [ first2 get-process send ] [ stop-this-server ] if* ;
  30
  31 : <node-server> ( addrspec -- threaded-server )
  32     binary <threaded-server>
  33         swap >>insecure
  34         "concurrency.distributed" >>name
  35         [ handle-node-client ] >>handler ;
  36
  37 : (start-node) ( addrspec addrspec -- )
  38     local-node set-global <node-server> start-server* ;
  39
  40 : start-node ( port -- )
  41     host-name over <inet> (start-node) ;
  42
  43 TUPLE: remote-process id node ;
  44
  45 C: <remote-process> remote-process
  46
  47 : send-remote-message ( message node -- )
  48     binary [ serialize ] with-client ;
  49
  50 M: remote-process send ( message thread -- )
  51     [ id>> 2array ] [ node>> ] bi
  52     send-remote-message ;
  53
  54 M: thread (serialize) ( obj -- )
  55     id>> local-node get-global <remote-process>
  56     (serialize) ;
  57
  58 : stop-node ( node -- )
  59     f swap send-remote-message ;
  60
  61 [
  62     H{ } clone \ registered-processes set-global
  63 ] "remote-thread-registry" add-init-hook
  64
  65