Paste: pipes

USING: io io.files io.timeouts io.launcher concurrency.messaging threads printf random kernel continuations calendar prettyprint math math.ranges math.functions locals io.encodings.ascii sequences shuffle parser debugger math.parser namespaces ;
IN: ftest

DEFER: generate
DEFER: 2exec>files

SYMBOL: default-stdout inline

: . ( val -- )
  [
    default-stdout get-global
    output-stream set
!    global [ pprint nl flush ] bind
    pprint nl flush
  ] with-scope ;

: send-noerr ( thread msg -- )
  "send-noerr" .
  [ send ] [ 3drop ] recover ;

: read-loop ( controller -- )
  "read-loop" .
  readln dup
  [
    over f send-noerr
    write nl
    read-loop
  ] [
    drop t send-noerr
  ] if
;

! USE: destructors
! : with-streams ( input output quot -- )
!     "with-streams" .
!     [ [ "with-streams*" . with-streams* "after with-streams*" . ] 3curry ]
!     [ [ "with-streams cleanup-a" . drop dispose dispose ] 3curry ] 3bi [ "with-streams cleanup-e" . ] cleanup ;


: enter-read-loop ( controller in out-file -- )
  "enter-read-loop" .
  ascii <file-writer> "after file-writer" .
  [
    [ dup read-loop ] with-streams
  ] [ print-error B 2drop "enter-read-loop - err handler" . ] recover
  t swap send-noerr ;
  

: enter-generate ( w-control out1 out2 -- )
  "enter-generate" .
  [ generate ] [ print-error 2drop ] recover
  t send-noerr ;
  
: w-control-loop ( controller generator -- )
  "w-control-loop" .
  receive "w-control-loop received" . [
    "preparing to stop w-c-l" .
    stop
    t send-noerr
  ] [
    "w-c-l resume" .
    #! resume can throw
    dup [ resume w-control-loop ]
        [ 2drop t send-noerr  ] recover
  ] if ;

: enter-write-loop ( out1 out2 -- )
#!  [ generate ] cocreate write-loop ;
  "enter-write-loop" .
  receive [
    self -rot [ enter-generate ] 3curry "ftest-generator" spawn
    w-control-loop
    "done w-control-loop" .
  ] unless ;

  
: wait ( -- )
    100 milliseconds receive-timeout
    [ wait ] unless
;

: controller-loop ( writer -- )
!  "controller-loop" .
  [ wait t "controller-loop finished" . ] [ drop f ] recover
  
  [ drop ]
  [ f over send-noerr
    controller-loop ] if
;


: read-commands ( -- cmd1 cmd2 )
  "read-commands" .
  "cmds" ascii file-lines dup first swap second ;

: similar? ( file1 file2 -- similar? )
  "similar?" .
  [ "diff %s %s" sprintf try-process t ] 2curry [ drop f ] recover ;

: pass ( cmd1 cmd2 i -- )
  "pass" .
  3dup 0 = [ 4drop drop ] [
    2exec>files similar?
    [ 1 - pass ] [ 3drop ] if
  ] if ;

: main ( -- )
  output-stream get default-stdout set-global
  [ read-commands 100 pass ] "main" spawn drop ;


#! : create-process ( cmd -- )  ascii <process-stream> ;
  
: start-reader ( in out-file -- thread )
  "start-reader" .
  self -rot [ enter-read-loop ] 3curry "reader" spawn ;

: out-file-names ( -- file1 file2 )
  "out-file-names" .
  1 2 [ "/tmp/diffproc/%d" sprintf ] bi@ ;
  
: start-readers ( in1 in2 -- thread1 thread2 )
  "start-readers" .
  out-file-names swapd [ start-reader ] 2bi@ ;

#! : create-writer ( out1 out2 -- )
#!  [ enter-write-loop ] 2curry spawn ;


DEFER: pb
: open-proc-streams ( cmd1 cmd2 -- stream1 stream2 )
  "open-proc-streams" .
!  [ ascii <process-stream> ] bi@ ;
  ascii <process-stream>
  swap ascii <process-stream>
  dup dup "12" swap stream-write stream-nl stream-flush  B
  dup dup "12" swap stream-write stream-nl stream-flush
!    tuck stream-write dup stream-nl stream-flush B
!    tuck stream-write dup stream-nl stream-flush 
;

: create-threads ( cmd1 cmd2 -- reader1 reader2 writer )
  "create-threads" .
    open-proc-streams
    2dup [ enter-write-loop ] 2curry "w-control" spawn
    -rot start-readers rot
;


: await-termination ( -- )
  "await-termination" .
  receive receive 2drop ;


: terminate-threads ( reader1 reader2 writer -- )
  "terminate-threads" .
  [ t send-noerr ] tri@ await-termination ;


: 2exec>files ( cmd1 cmd2 -- file1 file2 )
  "2exec>files" .
  create-threads "created threads" .
  dup controller-loop "done controller loop" .
  terminate-threads "terminating threads" .
  out-file-names
;


#! ##### BUSINESS LOGIC #####

: random-int ( -- val ) 4096 random 2048 - ;

: random-real ( -- val ) 0.0 4096.0 [a,b] random 2048 - ;

: floor1 ( n -- n ) 10 * floor 10 / ;


: pb ( val out1 out2 -- )
  "pb" .
  
  rot number>string -rot
  pick swap [ tuck stream-write dup stream-nl stream-flush ] 2bi@

  "generator suspending" .
  [ drop ] "suspending" suspend drop ;

:: pbl ( out1 out2 -- quot )
  "pbl" .
  [| val | val out1 out2 pb ] ;


:: three ( lambda -- )
  "three" .
  lambda [ random-int swap keep ] dup bi
  "three-ifs" .
  2dup [ 0 <= ] bi@ or [
    2dup [ 0 <= ] bi@ and [
      2drop lambda three
    ] [
      random-real lambda keep
#!      0 <= -rot [ 0 <= ] bi@ and or [ lambda three ] when
      2nip 0 <= [ lambda three ] when
    ] if
  ] [
    2drop
  ] if
;

: interact-menu-mort ( lambda -- )
  "interact-menu-mort" .
  dup 10 random swap keep
  dup 1 >= swap 8 < and
    [ interact-menu-mort ] [ drop "POST_MORTEM" . ] if
;
  
: interact-mort ( lambda -- )
  "interact-mort" .
  
  dup random-real swap keep
  "print 1 done" .
  0 <= [ interact-mort ] [ dup three interact-menu-mort ] if
;


:: outval ( lambda quot -- )
  "outval" .
  quot call lambda keep
  "outval after keep" .
  0 <= [ lambda quot outval ] when ;

  
:: interact-menu-invest ( lambda -- )
  "interact-menu-invest" .
  9 random lambda keep
  dup 1 >= swap 7 < and
  [ lambda interact-menu-invest ] [ "AFTER INVEST" . ] if ;

  
: interact-invest ( lambda -- )
  "interact-invest" .
  dup [ random-int ] [ outval ] dup 2bi
  dup [ random-real ] outval
#!  dup [ random-int ] outval
#!  "interact-invest-1" .
#!  dup [ random-int ] outval
#!  "interact-invest-2" .
#!  dup [ random-real ] outval
#!  "interact-invest-3" .

  interact-menu-invest
;


DEFER: generate2
DEFER: gen1
DEFER: gen2
DEFER: gen3
#!  : yield *coyield drop ;

: generate ( out1 out2 -- )
  "generate" .
  2dup pbl
  gen1 ;

: gen1 ( out1 out2 lambda -- )
  5 random 1 -
  gen2 ;

: gen2 ( out1 out2 lambda num -- )
  swap
  gen3 ;

: gen3 ( out1 out2 num lambda -- )
  keep
  generate2 ;

: generate2 ( out1 out2 num -- )
  "generate2" .
  dup dup 0 <= swap 2 > or [
    "recursing into generate, discarding %d" sprintf . generate
  ] [
    "entering interaction" .
    -rot pbl swap
    1 = [ interact-mort ] [ interact-invest ] if
  ] if
;

MAIN: main