declarative-concurrency

What execution order do we mean when we say threads execute simultaneously? There are two aspects to concurrency: interleaving among steps, and overlapping of two computation steps.

All computation steps form a global order called a partial order. This is called the causal order.

Example:

Causal Order:

      ------>---->
      ^ a      b
      | 
----->------>----->
  I     J      K           


Valid Execution Orders:
I a b J K
I a J b K
I a J K b
I J a b K
I J a K b
I J K a b

There is an execution state in which there is a choice of what thread to execute next.

In a declarative model, the nondeterminism is not visible to the programmer. This is due to:

(1) The single assignment store (2) Any operation has to wait until all its variables are bound. If we allow operations which the programmer can use to determine whether to wait or proceed, then the nondeterminism will be visible.

Browsing sometimes exhibits nondeterministic output - does this violate the above statement?

The Mozart system has a thread scheduler, which decides which thread to run next. A thread can be in the following states: (1) Ready/Runnable : (2) Suspended : e.g. blocked on an unbound variable (3) Terminated

Fair Thread Scheduler: Every runnable thread will eventually execute.

An interesting observation: fairness is related to modularity. Each thread's effect can be studied independent of the existence of other threads (is this possible in fair, nondeclarative models?)

---

The initial execution state is

Semantic Stack	SAS
{[<s>,0]}	0

Note that the semantic stack is a multiset.

At each step, one runnable semantic stack ST is selected from MST, leaving MST'.

MST = {ST} \multisetunion MST'.

One computation step is done in ST according to the semantics.

(ST, s) ——> (ST', s')

The choice of which stack to pick is up to the scheduler. The scheduler ensures fairness.

What happens when the currently executing statement blocks? In the sequential model, the interpreter would just hang. In the concurrent model, the interpreter has to try and run other stacks…

If there are no runnable stacks, and if every stack is terminated, then the computation terminates. If there are no runnable stacks, but at least one stack is suspended, then the computation blocks.

The syntax of a thread statement: thread <s> end

Step: Create a new stack! What should be its contents?

Suppose the multiset is MST. if the selected stack ST is of the form

[thread <s> end, E]     % Stack Top
ST'                     % Rest of stack

then the new multi set is:

[<s>, E]  \multisetunion     ST'          \multisetunion    MST'

A terminated semantic stack can be deallocated.

A blocked semantic stack: Reclaimed if its activation condition depends on an unreachable variable. In this case, the thread will never be ready again, so execution is unchanged by removing this thread.

local B in
  thread B=true end
  if B then true end
end

When the local statement has executed, the semantic multistack looks thus:

Stack	Store
([thread…end, if…end])	{b}

After executing the thread statement, we get

Stack	Store
([b=true], [if b=true then true end]),	{b}

Now, there are two stacks. The second is blocked since its activation depends on b, which is unbound. So the scheduler picks the other thread. After executing b=tre, the stack is now

Stack	Store
([],[if b=true then true end])	{b=true}

Now the empty stack is removed.

Stack	Store
([if b=true then true end])	{b=true}

The scheduler then executes the single thread containing the if statement.

---

---

fun {Double L} case L of nil then nil else H|T then 2*H | {Double T} end end Y = {Double X}

---

If X keeps growing, then so does Y. But if X does not change, then Y stops growing and the program remains unchanged.

If the inputs do not change, the program stops executing any further. This is called partial termination.

A set of store bindings: constraint.

For each variable x and constraint c, values (x,c) is the set of all possible values x can have, given that c holds.

Two constraints c1 and c2 are logically equivalent if (1) They contain the same variables and (2) For each variable x, values(x, c1) = values(x,c2)

For example,

x = rec(a b) and a = d is

equivalent to

x = rec(d b) and a = d

A program is said to be declaratively concurrent if for all inputs, the program does the following. All the executions either

(1) fail to terminate (2) terminate and are logically equivalent.

If a declarative concurrent program results in a failure for a given set of inputs, then all possible executions with that set will fail.

For example, the bindings in

thread X=1 end
thread Y=X+1 end
thread X=Y end

will eventually conflict. The program terminates.

One way to handle this is using exceptions. But, with these, the executions are no longer declarative, since the store after different execution sequences could be different. For example, Y might be bound to 1 or to 2 before the execution fails in the example above.

Hiding the nondeterminism in case of failure: This is up to the programmer.

declare X Y
local XTrial YTrial StatusX StatusY StatusV in
  thread
    try
      XTrial = 1
      StatusX = success
    catch _ then StatusX=error end
  end
  thread
    try
      YTrial = XTrial+1
      StatusX = success
    catch _ then StatusX=error end
  end
  thread
    try
      XTrial = YTrial
      StatusX = success
    catch _ then StatusV=error end
  end
  if StatusX==error orelse 
     StatusY==error orelse 
     StatusV==error then
    X=1
    Y=1
  else 
    X=XTrial
    Y=YTrial
  end 
end
% Rest of the code using X and Y

---

Allowing each process to run for one step can increase the overload on the scheduler, or in other cases, hog the processor for too long executing an intensive step. The Mozart scheduler uses a preemptive scheduling based on timers.

Allowing one whole computation step leads to a deterministic scheduler. Running the threads multiple times preempts the threads exactly at the same steps.

The timer based-approach is hardware-implemented, hence more efficient. This is however, nondeterministic, and depends on external events in the system as well.

Either the thread itself, or the scheduler can keep track of the time a thread has run. The second is easier to implement.

A short time slice -

• responsive system.
• large overhead for thread switches.

Large number of threads with long timeslices (~10 ms)? == if the threads are interdependent, then long timeslices are ok. == if the threads are independent, long timeslices are not ok. You need a hard-real-time OS.

Provides more control over how processor time is shared between threads. Three priority levels in Mozart: high, medium, low. Each has a guaranteed lower bound on processor time share. 100:10:1

When a thread creates a child thread, then child thread inherits the parent's priority. This is important for threading high-priority applications.

Threads are cooperative in completing a global task. On the other hand, you can have a competitive model where entities compete to complete a task. Usually those are OS processes.

Competitive Concurrency in Mozart - distributed computation model and Remote module. The distributed model is network-transparent.

---

Recall the list of random bits in Homework 1. The function which produced the stream of random bits is like a producer, and the function which computed running averages is like the consumer. This shows how to form a basic producer/consumer style program in the declarative model with threads.

You can also have multiple consumers without affecting the execution in any way.

local Xs S1 S2 in
    thread Xs={GenerateRandom} end
    thread S1={Average Count1} end
    thread S2={Average Count2} end
end

We can put many other stream objects between a producer and a consumer. These are called **Transducers**. A sequence of stream objects each of which feeds the next is called a **pipeline**. The simplest stream is a filter.

%-----------
% Producer
%-----------
fun {Generate N Limit}
   if N<Limit
   then N|{Generate N+1 Limit}
   end
end
%------------
% Consumer
%------------
fun {Sum Xs Accumulator}
   case Xs
   of X|Xr then {Sum Xr X+Accumulator}
   [] nil then Accumulator
   end
end
%-----------
% A filter in between producer and consumer
%-----------
local Xs Ys S in
   thread Xs={Generate 0 1000000} end
   thread Ys={Filter Xs 
                     fun {$ X} X mod 2\=0 end} end % Take odd numbers
   thread S={Sum Ys 0} end
end

               fun {$ X} X mod 2 \= 0 end
                          .
                          .
                          V
+----------+         +----------+       +----------+
|          |         |          |       |          |
|          | Xs      |          | Ys    |          |
| Producer |-------->| Filter   |------>| Consumer |
|          |         |          |       |          |
|          |         |          |       |          |
+----------+         +----------+       +----------+

%---------------
% Returns an infinite stream of integers starting from N
%---------------
fun {IntsFrom N}
    N|{IntsFrom N+1}
end
%--------------------
% Returns a function which checks whether its argument is a multiple
% of N
%--------------------
fun {NonMultiple N}
   fun {$ M}
       M mod N \= 0
   end
end
%--------------------
% Sieve of Eratosthenes
%--------------------
fun {Sieve Xs}
    case Xs
    of X|Xr then
        thread
           Ys = {Filter Xr {NonMultiple X}} % Remove multiples of X
        end
        X|{Sieve Ys}
    [] nil then nil
    end
end
%---------------------
% Example Execution
%---------------------
local Xs Ys in
   thread Xs={IntsFrom 2} end
   thread Ys={Sieve Xs} end
   {Browse Ys}
end

% Think about why the usage code was not simply
% local Ys in
%    Ys = {Sieve {IntsFrom 2}} % Eager version
% end

As an abstraction, we can introduce a new concept called a Stream Object. A stream object is a recursive procedure, which executes in its own thread, communicates with other stream objects via input and output streams, and finally, maintains a state.

proc {StreamObject InStream CurrentState ?OutStream}
   case InStream
   of InMsg|InStreamTail then
       NextState OutMsg OutStreamTail
   in
       % NexStateProcedure is the transition procedure
       {NextStateProcedure InMsg CurrentState OutMsg NextState}
       OutStream = OutMsg|OutStreamTail % Why is this done before recursing?
       {StreamObject InStreamTail NextState OutStreamTail}
   end
end

---

We can create new threads using a thread statement. In a system with threading, the two important operations that are usually supported are thread creation, and thread join. Joining is the process whereby the creator of the new thread waits until the newly created thread is destroyed, before it continues.

The basic idea is for the creator thread to Wait for the binding of an unbound variable, which is bound by the newly created thread just before it terminates. The procedure {Wait X} blocks until X is bound, and gets unblocked when X is bound.

local X in
   thread {P1} X=unit end
   {Wait X} % Creator thread blocks until X is bound
end

This is easily extended to multiple processes.

local X1 X2 X3 in
   thread {P1} X1=unit end
   thread {P2} X2=X1   end
   thread {P3} X3=X2   end
   {Wait X3} % blocks until X3 is bound,
             % which happens only when all three new threads terminate.
end

We can abstract this into a procedure {Barrier Procs}, which takes in a list Procs of procedures. This executes each procedure in the list in its own thread, and blocks until all the threads have terminated.

The procedure merely generalizes the methodology described above.

proc {Barrier Procs}
   %------------------------
   % Takes a finite list of procedures, and a variable L used for 
   % synchronizing.
   %
   % Runs each process in its own thread,
   % creates a new variable M, binds it to L, and passes M recursively
   % to run other procedures. 
   %
   % Return value is L if there is no procedure to execute,
   % and the result of the recursive call otherwise.
   %------------------------
   fun {BarrierLoop Procs L}
      case Procs
      of  Proc|Procr then
          M
      in
          thread {Proc} M=L end
          {BarrierLoop Procr M}
      [] nil then L
   end


   S = {BarrierLoop Procs unit}
in
   {Wait S}
end

%------------- Example Usage-------------------
{Barrier [ proc {$} X = 1 end
           proc {$} {Browse X} end]}

---