%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

%%% Reference: Chapter 11, "Concepts, Techniques,
%%% and Models of Computer Programming"

%%%%% 1. Open distribution - sharing records
% First process
declare
X=the_novel(text:"It was a dark and stormy night."
			author:"E.G.E. Bulwer-Lytton"
			year:1803)
{Browse {Connection.offerUnlimited X}}

% Second process
declare
X2={Connection.take 'x-ozticket://127.0.0.1:9000:RFeiky:Mt/v:w:w:m:VXD1hv'}
{Browse X2}

{Browse X}

%%%%%% 2. Sharing functions
% First process
declare
fun {MyEncoder X} (X*4449+1234) mod 33667 end
{Browse {Connection.offerUnlimited MyEncoder}}

% Second process
declare
E2={Connection.take 'x-ozticket://127.0.0.1:9000:RFeiky:Mt/w:w:w:m:7vECay'}

{Browse {E2 10000}}  % Call the function


%%%%% 3. Sharing dataflow variables
% First process
declare X in
{Browse {Connection.offerUnlimited X}}

% Second process
declare X in
X={Connection.take '...Xs ticket...'}
{Browse X*X}

X=100

declare S in
{Browse {Connection.offerUnlimited S}}
{Browse S}

S=1|2|3|4|_

%%%%% 4. Sharing state (cells)
% First process
declare
C={NewCell 0}
{Browse {Connection.offerUnlimited C}}

C:=10001

for I in 1..100 do C:=@C+1 end

{Browse @C}


% Second process
declare
C1={Connection.take '...Cs ticket...'}


%%%%% 5. Sharing objects
% First process
declare
class Coder
   attr seed
   meth init(S) seed:=S end
   meth get(X)
      X=@seed
      seed:=(@seed*1234+4449)mod 33667
   end
end
C={New Coder init(100)}
{Browse {Connection.offerUnlimited C}}
{Browse {C get($)}}

% Second process
declare
C2={Connection.take '...Cs ticket...'}
{Browse {C2 get($)}}


%%%%% 5a. Distributed lexical scoping

%%%%% 6. Ticket distribution
% Making tickets available through a file
% Define Offer and Take, to be used from now on
declare
proc {Offer X FN}
   {Pickle.save {Connection.offerUnlimited X} FN}
end
fun {Take FN}
   {Connection.take {Pickle.load FN}}
end


%%%%% 7. Stream communication
% First process
declare Xs Sum in
{Offer Xs tickfile}
fun {Sum Xs A}
   case Xs of X|Xr then {Sum Xr A+X}
   [] nil then A end
end
{Browse {Sum Xs 0}}

local Xs Sum in
   {Offer Xs tickfile}
   fun {Sum Xs A}
	  case Xs of X|Xr then {Sum Xr A+X}
	  [] nil then A end
   end
   {Browse {Sum Xs 0}}
end

% Second process
declare Xs Generate in
Xs={Take tickfile}
fun {Generate N Limit}
   if N<Limit then N|{Generate N+1 Limit}
   else nil end
end
Xs={Generate 0 150000}


%%%%% 8. An open server
% First process
declare S P in
{NewPort S P}
thread
   for X in S do {Browse X} end
end
{Offer P tickfile}

% Second process
declare
P={Take tickfile}
thread {Send P 100} {Send P 200} end
thread {Send P foo} {Send P bar} end

%%%%% 9. Asynchronous objects and dataflow
% First process
declare
C={NewAsync Coder init(100)}
{Offer C tickfile}

% Second process
C1={Take tickfile}
X={C1 get($)}
Y={C1 get($)}
Z={C1 get($)}
...
% use X, Y, and Z as usual
S=X+Y+Z
{Browse S}

% Definition of NewAsync
declare
fun {NewAsync Class Init}
   S P={NewPort S}
   Obj={New Class Init}
in
   thread
      for M in S do
		 try {Obj M} catch _ then skip end
      end
   end
   proc {$ M} {Send P M} end
end


%%%%% 10. Stationary and mobile objects
% First process
declare
C1={NewStat Coder init(100)}
{Offer C1 tick1}
C2={New Coder init(100)}
{Offer C2 tick2}
% also possible: {Offer C1#C2 tick}

% Second process
declare
C1={Take tick1}
{Browse start}
for I in 1..100000 do {C1 get(_)} end
{Browse done}

{OS.chDir "/Users/petervanroy/Courses/LINF2345"}

declare
C2={Take tick2}
{Browse start}
for I in 1..100000 do {C2 get(_)} end
{Browse done}


%%%%% 10a. Distributed lexical scoping
declare
fun {Ran} {C1 get($)} end

% Distributing Ran will keep the reference to C1!

%%%%% 11. A compute server
% First process
declare
class ComputeServer
   meth init skip end
   meth run(P) {P} end
end
C={NewStat ComputeServer init}
{Offer C tickfile}

%  Compute server with functors
declare
class ComputeServerR
   meth init skip end
   meth run(F) {Module.apply [F] _} end
   meth runModule(F M)
	  {Module.apply [F] [M]}
   end
end
CR={NewStat ComputeServerR init}
{Offer CR tickR}

% Second process
declare
C={Take tickfile}
declare
fun {Fibo N}
   if N<2 then 1 else {Fibo N-1}+{Fibo N-2} end
end

local F in
   {C run(proc {$} F={Fibo 30} end)}
   {Browse F}
end

local F in
   F={Fibo 30}
   {Browse F}
end                               


%%%%% 12. Resource awareness & closed distribution
declare
functor F
import OS
export time:T
define
   % Server (port + thread):
   S P={NewPort S}
   thread
      for X in S do {OS.time X} end
   end
   % Query function:
   fun {T} {Send P} end
end


%%%%% 13. The Remote module
declare
R={New Remote.manager init(fork:sh)}
fun {CS F} {R apply(F $)} end

declare
M={CS F}

{Browse M}
{Browse {M.time}}

declare
T={R apply(F $)}.time
{Browse {T}} % Call T, display locally

% asynchronous version of query
fun {T} X in {Send P X} {Wait X} X end


%%%%% 14. A compute server that handles resources
% Execute a statement remotely through a functor.
R={New Remote.manager init}

fun {Fact N} if N==0 then 1
			 else N*{Fact N-1} end end
F=functor export x:X define X={Fact 30} end
M={R apply(F $)}
{Browse M.x}


%%%%% 15. A dynamically-upgradable compute server
declare
proc {NewUpgradableStat Class Init ?Upg ?Srv}
   Obj={New Class Init}
   C={NewCell Obj}
in
   Srv={MakeStat
		proc {$ M} {{Access C} M} end}
   Upg={MakeStat
		proc {$ Class2#Init2}
		   {Assign C {New Class2 Init2}} end}
end
declare Srv Upg in
Srv={NewUpgradableStat ComputeServer init Upg}

declare
class CComputeServer from ComputeServer
   meth run(P Prio<=medium)
	  thread
		 {Thread.setThisPriority Prio}
		 ComputeServer,run(P)
	  end
   end
end
{Upg CComputeServer#init}

local F in
   {Srv run(proc {$} F={Fibo 20} end)}
   {Browse F}
end

%%%%% 16. Programming with distributed objects

% Explain programming techniques


%%%%% 17. Fault tolerance

% Explain difference between fault resistance
% & active fault tolerance


%%%%% 18. Disconnected operation

% Each site executes this on startup:
{Fault.defaultEnable [permFail] _}


%%%%% 19. Detecting a problem

% Get a reference to the port:
X={Take tickfile}

% Signal as soon as a problem is detected:
{Fault.installWatcher X [tempFail permFail]
 proc {$ _ _}
	{Browse X#" has problems!  Don\'t use it."}
 end}


%%%%% 20. Active fault tolerance


%%%%% 21. Persistence


%%%%% 22. Debugging


%%%%% 23. Building applications


%%%%% 24. Practical tips


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

%%% Second part: transactional object store

% Example: fault tolerant counter

%%%% First file: starting up 
functor 
import Pickle Connection GlobalStore Module Browser(browse:Browse)
define
   class Counter
      attr i
      meth init i<-0 end
      meth inc i<-@i+1 end
      meth get(X) @i=X end
   end

   NewObj
   BLS={GlobalStore.newStore}
   LS={BLS newLocal(Module user1 NewObj $ _)}

   C={NewObj Counter init}
   CN={LS saveobj(C $)}
   {Pickle.save {Connection.offerUnlimited LS#CN} 'ctr.tick'}
end

%%%% Second file: adding a user
functor
import Pickle Connection QTk Module Browser(browse:Browse)
define
%%%% Make local user interface:
   A H
   W={QTk.build
      td(title:"Counter"
		 button(text:nil action:A handle:H glue:nswe))}
   {W show}

%%%% Get object reference:
   BLS#CN={Connection.take {Pickle.load 'ctr.tick'}}
   LS={BLS newLocal(Module userN _ $ _)}
   C={LS loadobj(CN $)}

%%%% Keep ticket active (recovery action):
   {LS setnotifyrecovery(
		  proc {$}
			 {Pickle.save {Connection.offerUnlimited LS#CN} 'ctr.tick'}
		  end)}

%%%% Connect UI with object:
   % 1. Do increment + update myself
   proc {A} TId X in
      {LS trans(proc {$ X1} {C inc} {C get(X1)} end X TId)}
      if {LS checktrans(TId $ _)}==commit then {H set(X)} end
   end
   % 2. Update other guys
   {LS setnotifyupdate(
		  proc {$ C} X in {C get(X)} {H set(X)} end)}
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
declare
fun {NewStat Class Init}
   {MakeStat {New Class Init}}
end

fun {MakeStat PO}
   S P={NewPort S}
   N={NewName}
in 
   thread 
	  for M#R in S do
		 thread 
			try {PO M} R=N catch X then R=X end 
		 end 
	  end
   end 
   proc {$ M}
      R in 
	  {Send P M#R}
	  if R==N then skip else raise R end end 
   end 
end

fun {NewActive Class Init}
   S P={NewPort S}
   Obj={New Class Init}
in
   thread
	  for M in S do
		 try {Obj M} catch _ then skip end
	  end
   end
   proc {$ M} {Send P M} end
end

NewAsync=NewActive

fun {NewAsyncExc Class Init}
   S P={NewPort S}
   Obj={New Class Init}
in
   thread
	  for M#X in S do
		 try {Obj M} X=unit catch Y then X=Y end
	  end
   end
   proc {$ M X} {Send P M#X} end
end

proc {Offer X FN}
   {Pickle.save {Connection.offerUnlimited X} FN}
end

fun {Take FN}
   {Connection.take {Pickle.load FN}}
end