XXX THIS FILE IS NOT YET DONE.  FINISHED VERSION UP BY 9PM 12/12.

The final is cumulative.

It should go without saying that you should do all of the practice
problems from the midterms and make sure you understand the answers to
the midterms.  I will be particularly focusing on questions people did
not do well on for the first two midterms.  For midterm 1, these were
problems 3 and 4 (parallelization and identifying data races).  For
midterm 2, these were problems 2 and 4 (identifying deadlock and
programming Erlang servers).

New material
------------

** Map-Reduce:

- Why do all of the Mappers need to complete before we execute any
calls to the Reducer.reduce(InputKey key, Iterable<InputValue> values,
Context context) method?

  Answer: For any InputKey to a reduce call, we must ensure that we
  have all of the InputValues for that key before we invoke the reduce
  method. Since any mapper might produce a value for a particular
  InputKey, all Map tasks must be completed before we make any calls
  to a reduce method. If there are some workers still working on
  redundant computations that will be discarded, it is OK to proceed.
  Technically, you could imagine speculating that some partitions are
  complete, and invalidating those results and redoing them if you
  produced an InputKey, InputValue pair after you had already reduced
  that InputKey. But doing so is is unlikely to be useful.

- In a real Hadoop or MapReduce implementation, the workers send back
progress information to the Master (e.g., I’m now 65% done.... I’m now
66% done...). Suggest how this information might be used.

  Answer: When all tasks have been scheduled, but not completed, the
  master speculative sends out redundant requests to perform
  tasks. With progress information, we can can avoid sending out
  requests for tasks that some other node is is almost finished
  with. We might even be able to track the speed of each node, and
  predict for each node the time remaining until it completes its
  current task. With that information, we could send out for redundant
  execution the task whose estimated completion time is furthest away.
  There are several levels of additional refinement you could use for
  this.

- Do Map tasks running on workers send all of their output keys and
values to the Master node? Why or why not?

  Answer: No. The map tasks generate far too much data to funnel it
  all through the master node. The network bandwidth to the master
  would be a huge bottleneck. Instead, map tasks write their output to
  files, and only send information to the master about their progress
  and completion of the task.

- MapReduce is inspired by functional programming, and the Map and
Reduce are supposed to be “functional” or “not have side effects”,
such as changing static fields or writing output other than via
Context. Why? What can go wrong if mappers and reducers have side
effects?

  Answer: Because Map and Reduce are functional, we know that if we
  ignore the output of a Map or Reduce task, it doesn’t effect the
  computation. Thus, we don’t have to worry about any side effects
  from aborted, failed, or redundant tasks. Also (although less
  directly) means it shouldn’t mean how values get spread/grouped as
  they are sent to various tasks and workers.

- What are the algebraic properties required of mapper and reducer
functions for mapreduce to work at all?  What property is needed of
the reducer in order to make a combiner useful?

  Answer: No specific properties are required of mappers or reducers,
  but the implementation/model makes no guarantee which files are
  passed to which mappers, and no ordering is assumed of keys in the
  output of mappers, and likewise regarding the input and output of
  reducers.  The only assumption is that *all* values associated with
  a particular key will be passed, at once, to a single reducer.  This
  in itself can be slow, so a combiner can be used if reduction is
  commutative and associative.  For example, when counting word
  frequencies, since frequencies tend to follow a Zipf distribution,
  each map task will produce hundreds or thousands of records of the
  form <the, 1>. All of these counts will be sent over the network to
  a single reduce task and then added together by the Reduce function
  to produce one number.  But since addition is commutative and
  associative, we can use a combiner: on each map task, all the counts
  of words for that task will be summed before sending them to the
  reduce task.  So, for example, rather than receive 10 pairs <the,
  1>, 5 from each of two mapper tasks, the reduce task might only
  receive the pair <the, 5> twice, once from each task.

** Parallelization using divide-and-conquer strategies
   (Dr. Luchangco's lecture, plus earlier stuff)

- What are the main differences between accumulation-based
computations over lists of data, and divide-and-conquer-based
computations?  Why is the first good for sequential programming and
not for parallel, and vice versa?  As an example, think of using an
accumulator to sum all values in a list vs. using a divide-and-conquer
strategy to do it.

  Answer: An accumulation method maintains a single variable that
  contains the "result so far".  We process each element in order,
  combining it with the current accumulator.  As such we save space:
  we need only one variable to store all the intermediate results.
  Since we have to process all elements of the list one at a time
  anyway (since we are sequential) this makes good sense.  The problem
  is that doing this in a parallel setting we will have to synchronize
  on access to this accumulator.  So we could spawn a thread for each
  element of the list that does the computation acc = acc + elem, but
  the synchronization basically serializes all of the accesses.
      Divide-and-conquer is better for parallel, since it allows
  computations to take place independently, without serialization.
  For example, we could sum all the second half of a list at the same
  time as the left half.  The drawback here is that we need more
  space.  Each parallel computation needs its own accumulator, so to
  speak, and in the worst case we have nearly as many accumulators
  (actually, one less) than list elements.  So we pay a cost in space
  usage to avoid synchronization and increase parallelism.

- It is important to distinguish between spatial order and temporal
order of elements in a list being computed on.  In a sequential
algorithm, the two orders are often the same, while in a parallel
algorithm they can be different.  Explain.

  Answer: If we consider summing up the contents of a list, the order
  we process each of the elements of the list is the same as the order
  those elements appear in the list.  In other words, the temporal
  order (or "processing order") is the same as the spatial order (that
  is, the order the elements appear in the list).  In a parallel
  algorithm we could sum the lists in a different order.  For example,
  we could sum up the second half of the list in parallel with summing
  the first half.  In this case, the temporal order differs from the
  spatial order.

- A "Split list" is a datastructure that is designed to be easy to
break in roughly equal-sized chunks.  The definitions of the
operations of this list were given in Dr. Luchangco's talk.  Implement
a split-list in Erlang, in particular, the following functions:

empty() creates an empty split list
is_empty(L) returns true if L is an empty split list
singleton(X) returns a split list containing the single element X
is_singleton(L) returns true if L is a singleton split list 
get(L) returns X if L is a singleton split list containing X
concat(L1,L2) returns a split list that concatenates L1 and L2
is_split(L) returns true if L is a split node
split(L) returns a pair {L1,L2} if L is a split node that joins split
  lists L1 and L2
from_list(L) returns a split list constructed from the standard list L

  Write an Erlang program to compute the length of a split list.
  Write a sequential version of this function (call it length(L)) and
  a parallel version (par_length(L)).  For brownie points/fun, write a
  function to "balance" the split list so that each child of a split
  node has roughly equal length.  For even more brownie points, write
  a function to_list that converts a split list to a standard list, in
  parallel.

Answer:

%copy into file split.erl
-module(split).
-compile(export_all).
-compile({no_auto_import,[length/1]}).

empty() -> empty.
is_empty(empty) -> true;
is_empty(_) -> false.
    
singleton(X) -> {singleton,X}.
is_singleton({singleton,_}) -> true;
is_singleton(_) -> false.
get({singleton,X}) -> X.

concat(L1,L2) -> {split,L1,L2}.
is_split({split,_,_}) -> true;
is_split(_) -> false.
split({split,L1,L2}) -> {L1,L2}.

length(L) ->
    IE = is_empty(L),
    IT = is_singleton(L),
    IS = is_split(L),
    if IE -> 0;
       IT -> 1;
       IS -> {X,Y} = split(L),
	     length(X) + length(Y)
    end.

% This function implements a "future" of the same style we had in
% Java: it calls function F on its arguments A in a separate process,
% returning the Pid of the process.
fut_call(F,A) -> 
    Pid = self(),
    spawn(fun () -> R = apply(F,A), Pid ! {result,self(),R} end).
	
% Waits for a message from process P which contains the result of 
% spawned function call.	  
fut_wait(P) ->
    receive 
	{result,P,X} -> X
    end.

par_length(L) ->
    IE = is_empty(L),
    IT = is_singleton(L),
    IS = is_split(L),
    % two base cases are the same.  The split case
    % uses futures to compute the results of the two halves
    % in parallel
    if IE -> 0;
       IT -> 1;
       IS -> {X,Y} = split(L),
	     LX = fut_call(fun par_length/1,[X]),
	     LY = fut_call(fun par_length/1,[Y]),
	     fut_wait(LX) + fut_wait(LY)
    end.

test() ->
    E = empty(),
    true = is_empty(E),
    L = concat(concat(singleton(1),singleton(1)),singleton(2)),
    3 = length(L),
    3 = par_length(L).


- Practice implementing divide-and-conquer algorithms by implementing
a parenthesis matcher.  The idea is that you are given a buffer
consisting of a sequence of open and closed parentheses.  Your job is
to compute whether all of the parentheses match.  You can use
fork/join to do implement this using divide-and-conquer.  Start from
two files that are based on the MaxSolver we went over in class:
http://www.cs.umd.edu/class/fall2010/cmsc433/examples/MatchSolver.java
and
http://www.cs.umd.edu/class/fall2010/cmsc433/examples/MatchProblem.java
. The entrypoint to the program is MatchSolver's main() method.  You
need to fill in code bits in both classes to implement sequential
parenthesis matching, along with code to combine the results of
applying parenthesis matching to both sides of an array.

  Answer: 

//copy into MatchProblem.java
public class MatchProblem {
    private final int[] array;
    private final int start, end;
    public static class Result {
        public final int open, closed;
        public Result(int open, int closed) {
            this.open = open;
            this.closed = closed;
        }
    }
    public MatchProblem(int[] array, int start, int end) {
        int sz = array.length;
        if (start < 0 || start > sz || end <= start || end > sz)
            throw new IllegalArgumentException("Illegal problem");
        this.array = array;
        this.start = start;
        this.end = end;
    }
    public Result solveSequentially() {
        int open = 0;
        int closed = 0;
        for (int i = start; i<end; i++) {
            if (array[i] == 0) { // 0 models open-paren
                open++;
            } else if (array[i] == 1) { // 1 models closed-paren
                if (open > 0)
                    open--;
                else
                    closed++;
            }
            else
                throw new IllegalArgumentException("Illegal array contents");
        }
        return new Result(open,closed);
    }
    public int size() {
        return (end - start);
    }
    public MatchProblem subproblem(int newStart, int newEnd) {
        newStart += start;
        newEnd += start;
        if (newStart > end || newEnd <= newStart || newEnd > end)
            throw new IllegalArgumentException("Illegal subproblem");
        return new MatchProblem(array,newStart,newEnd);
    }
}

//copy into MatchSolver.java
import java.util.Random;
import jsr166y.RecursiveAction;
import jsr166y.ForkJoinPool;

public class MatchSolver extends RecursiveAction {
    private final static int THRESHOLD = 500000;
    private final MatchProblem problem;
    MatchProblem.Result result;
	
    public MatchSolver(MatchProblem prob) {
        this.problem = prob;
    }
	
    protected void compute() {
        if (problem.size() < THRESHOLD) {
            result = problem.solveSequentially();
        }
        else {
            int m = problem.size() / 2;
            MatchSolver left, right;
            left = new MatchSolver(problem.subproblem(0, m));
            right = new MatchSolver(problem.subproblem(m,problem.size()));
            invokeAll(left,right);
            int middle = left.result.open - right.result.closed;
            if (middle >= 0)
                result = 
                    new MatchProblem.Result(right.result.open+middle,left.result.closed);
            else
                result = 
                    new MatchProblem.Result(right.result.open,left.result.closed+middle);
	}
    }
	
    /**
     * Creates an array of the given size with random contents.
     */
    static private Random rand = new Random();
    public static int[] createRand(int num) {
        int[] result = new int[num];
        for(int i = 0 ; i < num; i++)
            result[i] = rand.nextInt(2);
        return result;
    }

    public static void main(String args[]) {
        int nThreads = 9;
        int arrSz = 100000000;
        int[] array = createRand(arrSz);
        MatchProblem prob = new MatchProblem(array,0,arrSz);
        long startTm = System.currentTimeMillis();
        MatchProblem.Result result = prob.solveSequentially();
        long endTm = System.currentTimeMillis();
        System.out.println("SEQ:  ("+result.open+","+result.closed+") time ="+(endTm-startTm)+" ms");
        startTm = System.currentTimeMillis();
        ForkJoinPool pool = new ForkJoinPool(nThreads);
        MatchSolver solver = new MatchSolver(prob);
        pool.invoke(solver);
        endTm = System.currentTimeMillis();
        System.out.println("FJ:  ("+result.open+","+result.closed+") time ="+(endTm-startTm)+" ms");
    }
}

** Distributed computing

- How is writing a distributed program similar to writing a parallel
program?  How is it different?  Why might you prefer a distributed
implementation to a parallel one?

  Answer: These two activities are similar in that they both involve
  the coordination of parallel activities to complete a task.  They
  differ in two ways.  First, parallel programs may use shared memory;
  distributed programs, by definition, do not share any memory.
  Second, parallel programs do not need to worry about communication
  failures, since tasks are coordinating on the same machine.
  Typically it is reasonable to assume the entire machine will fail,
  not that some part of it will.  On the other hand, a distributed
  program may have its components fail in various ways.  For example,
  a request or response message may get dropped, or a node computing
  part of a result may fail.  Thus, to be robust, a distributed
  program must be written to account for these problems.

- What is an advantage to using an RPC library compared to performing
network-oriented communications directly?  What is a potential
disadvantage?

  Answer: RPCs are much closer to the level of the rest of the
  program.  You don't need to worry about encoding your data as
  strings before you send, nor do you need to decode and multiplex at
  the receiver; the RPC mechanism takes care of all of this.  On the
  other hand, if RPCs look too much like normal function calls, the
  programmer might fail to appreciate that they can fail.  Java RMI
  tries to avoid this problem by having potentially remote calls throw
  a checked RemoteException, which forces callers to account for
  possible failure.  Another possible problem is that RPCs may be more
  expensive and heavyweight than lighter-weight custom messages.

- If one part of a distributed program sends a request to another part
and receives no response, either the remote side received the message
but did not respond (or the response was lost by the network), or the
remote side never received the request.  What is the important
distinction between these two cases?

  Answer: If the remote side did not receive a request, then a sender
  can just resend it.  On the other hand if the remote side received
  the request and changed its state, it is not OK, in general, for the
  requester to resend.  If you did, for example, you might end up
  buying two books rather than just one.

- Consider the following Erlang program, which creates two processes
that send messages back and forth.  Change this program so that
instead of running the pong process on the local node, it runs it on a
remote node called bigmac@gandalf.

-module(pingpong).
-compile(export_all).

ping(_Pid,0) -> ok;
ping(Pid,N) when N > 0 -> 
    Me = self(),
    Pid ! {ping,Me},
    receive {pong,Me} -> ping(Pid,N-1) end.

pong() ->
    receive 
        {ping,Pid} -> Pid ! {pong,Pid}, pong()
    end.

start(N) ->
    Pid = spawn(fun pong/0),
    ping(Pid,N).


  Answer: This is easy: just change 
    Pid = spawn(fun pong/0),
  to be
    Pid = spawn(gandalf@bigmac,fun pong/0),
  Of course, you have to start up the node with this name before the
  program starts.

- Consider the above program again.  Change it so that it registers a
pong_server, which is just a process running pong().  Then change the
program so that it uses the Erlang rpc module to start the pong server
on bigmac@gandalf, and then likewise uses the rpc module to remotely
call a function that invokes ping() at bigmac@gandalf.

  Answer:

% copy into file pingpong_remote.erl
% here we have changed ping() so that it doesn't take a pid, but 
% sends messages to the registered pong server.
-module(pingpong_remote).
-compile(export_all).

ping(0) -> ok;
ping(N) when N > 0 -> 
    Me = self(),
    pong_server ! {ping,Me},
    receive {pong,Me} -> ping(N-1) end.

pong() ->
    receive 
        {ping,Pid} -> Pid ! {pong,Pid}, pong()
    end.

start_pong_server() ->
    register(pong_server,spawn(fun pong/0)).

start(N) ->
    rpc:call(gandalf@bigmac,pingpong_remote,start_pong_server,[]),
    rpc:call(gandalf@bigmac,pingpong_remote,ping,[N]).

- Write a service that receives messages consisting of a function to
run and the arguments to give to that function.  (Note that you can
call apply(F,A) to run such a function and arguments.  For example, if
you did apply(fun (X,Y) -> X + Y end,[1,2]) you'd get 3.)  The service
then sends back the result.  Then use the keep_alive function
presented in class to keep this service running indefinitely.  To test
that it's working, send the service a bogus message, e.g., a function
that takes two arguments but you only give it one.  Make sure the
service properly restarts.

  Answer:  Note the only thing that's important here not specified
  above is the need to allow an rpc to timeout.  This can be used to
  detect that a failure happened.

% copy into file rpcserver.erl.  Run by calling the test() method.
-module(rpcserver).
-export([test/0]).

on_exit(Pid, Fun) ->
    spawn(fun() -> 
		  process_flag(trap_exit, true),
		  link(Pid),
		  receive
		      {'EXIT', Pid, Why} ->
			  Fun(Why)
		  end
	  end).

keep_alive(Name,Fun) ->
    Pid = spawn(Fun),
    register(Name,Pid),
    on_exit(Pid, fun(Why) -> 
			 io:format(" ~p died with: ~p~n",[Pid,Why]),
			 keep_alive(Name,Fun) end).

loop() ->
    receive
	{call,Pid,F,A} -> 
	    R = apply(F,A),
	    Pid ! {result,R},
	    loop()
    end.

start() ->
    keep_alive(rpc_service,fun loop/0),
    ok.

rpc_apply(F,A) ->
    rpc_service ! {call,self(),F,A},
    receive 
	{result,R} -> {ok,R}
    % times out after (an arbitrarily chosen) 100 ms
    after 100 -> timeout
    end.

test() ->	    
    start(),
    {ok,3} = rpc_apply(fun (X,Y) -> X + Y end,[1,2]),
    {ok,[1,2]} = rpc_apply(fun (X,Y) -> [X|Y] end,[1,[2]]),
    % next call should fail, since adding 1 and hello is not OK,
    % thus the call will time out
    timeout = rpc_apply(fun (X,Y) -> X + Y end,[1,hello]),
    % next call should succeed, since keep_alive will restart
    % the service
    {ok,3} = rpc_apply(fun (X,Y) -> X + Y end,[1,2]),
    ok.

** Dynamic Software Updating

- Name a problem that could occur with a dynamic update, if we weren't
careful, e.g., if we allowed the update to take effect at an arbitrary
time.  How does the gen_server framework's approach to code updating
help avoid such problems?

  Answer: The most apparent problem would be a change to the type
  signature of a function.  The update would take effect, changing the
  compiled code.  But if the caller were running the old version, it
  would call the updated code with the wrong number of arguments.  The
  gen_server approach to updating avoids this problem by defining a
  new module as a replacement for the old one, and this module is
  invoked via calls to the server module.  As such, it controls the
  effect of updating, so that it cannot take effect except "between"
  calls to the server.

- Extend the example for updating the Erlang name server we went over
in class so that instead of just re-using the old state after updating
the module, we apply a state transformation function instead, to
change the state to work with the new code.  Change the
new_name_server.erl module to additionally have an "undo" function.
This function can be used to undo any deletions from the dictionary.
To implement this, you want to change the state of the name server so
that it's not just a dictionary, as is the case now, but is a pair of
a dictionary (just as now) and a list of deleted items.  Undo will
restore to the dictionary the last deleted item.  Your state
transformation code will have to initialize this list (it should be
empty) and you will have to change the implementations of add,
allNames, etc. in the handle_call(...) cases to use the state
properly.

  Answer: 
	     
Here's the new server_dsu.erl module (almost the same as the original,
but for the parts handling swap_code):

-module(server_dsu). 
-export([start/1, start/2, call/2, cast/2, swap_code/3]). 
% Client-server messaging framework. 
% 
% The callback module implements the following callbacks: 
% init() -> InitialState 
% handle_call(Params, State) -> {Reply, NewState} 
% handle_cast(Params, State) -> NewState 
% Return the pid of a new server with the given callback module. 

start(Name,Module) ->
    Pid = start(Module),
    register(Name,Pid),
    ok.

start(Module) -> 
    spawn(fun() -> loop(Module, Module:init()) end). 

loop(Module, State) -> 
    receive 
        {call, {Client, Id}, Params} -> 
            {Reply, NewState} = Module:handle_call(Params, State), 
            Client ! {Id, Reply}, 
            loop(Module, NewState); 
        {cast, Params} -> 
            NewState = Module:handle_cast(Params, State), 
            loop(Module, NewState);
        {swap_code, NewModule, XForm} ->
            loop(NewModule, XForm(State))
    end. 

% Client-side function to call the server and return its reply. 
call(Server, Params) -> 
    Id = make_ref(), 
    Server ! {call, {self(), Id}, Params}, 
    receive 
        {Id, Reply} -> Reply 
    end. 

% Like call, but no reply is returned. 
cast(Server, Params) -> 
    Server ! {cast, Params}. 

% Changes the module used to handle callbacks
swap_code(Server,Module,XForm) -> 
    Server ! {swap_code, Module, XForm}.


--- Here's the new new_name_server.erl module.  It contains the new
    feature and a state transformation function at the bottom for
    transforming the old state:

-module(new_name_server).
-export([init/0, add/2, all_names/0, delete/1, undo/0, 
	 whereis/1, handle_call/2, dsu_state_xform/1]).

%% interface
all_names()      -> server_dsu:call(name_server, allNames).
add(Name, Place) -> server_dsu:call(name_server, {add, Name, Place}).
delete(Name)     -> server_dsu:call(name_server, {delete, Name}).
whereis(Name)    -> server_dsu:call(name_server, {whereis, Name}).
undo()           -> server_dsu:call(name_server, undo).
     

%% callback routines
init() -> {dict:new(),[]}.
    
handle_call({add, Name, Place}, {Dict,Dels}) -> 
    {ok, {dict:store(Name, Place, Dict), Dels}};
handle_call(allNames, {Dict,Dels}) ->
    {dict:fetch_keys(Dict), {Dict,Dels}};
handle_call({delete, Name}, {Dict,Dels}) -> 
    R = dict:find(Name,Dict),
    case R of
	error -> {error,{Dict,Dels}};
	{ok,V} -> {ok, {dict:erase(Name, Dict),[{Name,V}|Dels]}}
    end;
handle_call({whereis, Name}, {Dict,Dels}) ->
    {dict:find(Name, Dict), {Dict,Dels}};
handle_call(undo,{Dict,Dels}) ->
    case Dels of
	[] -> {error,{Dict,Dels}};
	[{Name,V}|Rest] -> 
	    {ok, {NewDict,_}} = handle_call({add,Name,V},{Dict,Dels}),
	    {ok, {NewDict,Rest}}
    end.

%% this function initializes the new server's state.  The deletion
%% list starts out as empty
dsu_state_xform(Dict) -> {Dict,[]}.

---- Here's a transcript that shows this program in action
     (name_server1 is the initial implementation provided with the
     files from the course web site):

[ server ]> erl
Eshell V5.8  (abort with ^G)
1> server_dsu:start(name_server,name_server1).
ok
2> name_server1:add(mike,"at home").
ok
3> name_server1:add(joe,"at work").
ok
4> name_server1:all_names().  % should fail, since not defined yet
** exception error: undefined function name_server1:all_names/0
5> server_dsu:swap_code(name_server,new_name_server,{new_name_server,dsu_state_xform}). % recall that {M,N} specifies function N from module M
{swap_code,new_name_server, {new_name_server,dsu_state_xform}}
6> new_name_server:all_names().
[joe,mike]
7> new_name_server:delete(mike).
ok
8> new_name_server:all_names().
[joe]
9> new_name_server:undo().
ok
10> new_name_server:all_names().
[mike,joe]
11> new_name_server:undo().
error
12>