Modelling Complex Software Systems

Concurrency

menu_book

Concurrency

What?

Why?

What makes it hard?

Concurrent Language Paradigms

Speed Dependence

Arbitrary interleaving

Concurrent Programming Abstraction

Arbitrary Interleaving

Atomicity

Correctness

Java Threads

Creation

Two ways to create:

States

Alive thread is in one of these states:

Java thread states

Primitives

More states

Additional states:

Interruption

Mutual Exclusion (Mutex)

class P extends Thread {
  while (true) {
    non_critical_P();
    pre_protocol_P();
    critical_P();
    post_protocol_P();
  }
}

class Q extends Thread {
  while (true) {
    non_critical_Q();
    pre_protocol_Q();
    critical_Q();
    post_protocol_Q();
  }
}

Properties of mutex solution

Also desirable:

Assumptions

Attempt 1

Attempt 2

Attempt 3

Attempt 4

Livelock: processes are still moving, but critical section is unable to be completed

Attempt 5: Dekker’s Algorithm

while (true) {
  non_critical_P();
  p = 1;
  // repeat while Q has flag raised
  while (q != 0) {
    // if it is Q's turn
    if (turn == 2) {
        // lower flag
        p = 0;
        // wait until its P's turn
        while (turn == 2);
        // raise p's flag
        p = 1;
    }
  }
  critical_P();
  turn = 2;
  p = 0;
}

while (true) {
  non_critical_Q();
  q = 1;
  // repeat while P has flag raised
  while (p != 0) {
    // if it is Q's turn
    if (turn == 1) {
        // lower flag
        q = 0;
        // wait until its Q's turn
        while (turn == 1);
        // raise Q's flag
        q = 1;
    }
  }
  critical_Q();
  turn = 1;
  q = 0;
}

Peterson’s Mutex Algorithm

static int turn = 1;
static int p = 0;
static int q = 0;

while (true) {
  non_critical_P();
  p = 1;
  turn = 2;
  // give Q a turn.  wait till it is complete
  while (q && turn == 2);
  critical_p();
  p = 0;
}

Java: Monitors and synchronisation

Synchronised methods

synchronized void increment() { ... }

Synchronized object

class SynchedObject extends Thread {
  Counter c;

  public SynchedObject(Counter c) { this.c = c; }

  public void run() {
    for (int i = 0; i < 5; i++) {
      synchronized(c) {
        c.increment();
      }
    }
  }
}

Monitors

class MonitorAccount extends Account {
    public synchronized void withdraw(int amount) {
        while (balance < amount) {
            // withdrawal cannot proceed.  get thread to wait until balance updates
            try {
                wait();
            } catch (InterruptedException e) {}
        }
        super.withdraw(amount);
    }

    public synchronized void deposit(int amount) {
        super.deposit(amount);
        // after deposit, notify all threads waiting for updated balance
        notifyAll();
    }
}

Lightweight monitors

Implementation

Volatile variables

Process states

Blocked Java States

Synchronisation constructs

Level of abstraction Construct
High Monitor
  Semaphore
Low Protocol variables

Java: Semaphores

Analogy: Hotel with $k$ rooms

Operations

S.wait():

if S.v > 0
  # provide permit
  S.v--
else 
  # add process p to wait set 
  S.w = union(S.W, p)
  p.state = blocked

S.signal():

if S.W == {}
  # empty wait set, so keep the permit
  S.v++
else 
  # hand out permit to someone in the wait set
  choose q from S.W
  # remove q from wait set
  S.W = S.W \ {q}
  q.state = runnable

Binary Semaphore: Mutex

Solution of Mutex Problem

binary semaphore S = (1, {});

Process P loop:
p1:  non_critical_p();
p2:  S.wait();
p3:  critical_p();
p4:  S.signal();

Process Q loop:
q1: non_critical_q();
q2: S.wait();
q3: critical_q();
q4: S.signal();

State Diagrams

Semaphore mutual exclusion

Controlling execution order

integer array A
binary semaphore S1 = (0, {})
binary semaphore S2 = (0, {})

p1: sort low half
p2: S1.signal()
p3: 

q1: sort high half
q2: S2.signal()
q3:

# wait for both semaphores to become available
m1: S1.wait()
m2: S2.wait()
m3: merge halves

Strong semaphores

Bounded Buffer Problem

buffer = empty queue;
// no permits available for removal from queue
semaphore notEmpty = (0, {});
// n permits available for adding to queue
semaphore notFull = (n, {});

Producer
idem d
loop
  # produce items
  p1: d = produce();
  # wait for buffer to have space
  p2: notFull.wait();
  p3: buffer.put(d);
  # indicate data has been put onto buffer
  p4: notEmpty.signal();

Consumer
item d
loop
  # wait until the buffer has items to consume
  q1: notEmpty.wait();
  q2: d = buffer.take();
  # indicate item taken from buffer
  q3: notFull.signal();
  q4: consume(d);

Java semaphores

Java Thread states in detail

Java thread states

Peterson’s mutex algorithm

static int turn = 1;
static int p = 0;
static int q = 0;

while (true) {
p1:  non_critical_P();
p2:  p = 1;
p3:  turn = 2;
     // give Q a turn.  wait till it is complete
p4:  while (q && turn == 2);
p5:  critical_p();
p6:  p = 0;
}

while (true) {
q1:  non_critical_q();
q2:  q = 1;
q3:  turn = 1;
     // give P a turn.  wait till it is comqlete
q4:  while (p && turn == 2);
q5:  critical_q();
q6:  q = 0;
}

Formal modelling with FSP

Advantages of formal modelling

LTS

FSP

Concepts

Action prefix operator ->

If x is an action and P a process, then x -> P describes a process that first engages in action x and then behaves as described by P

Subprocesses ,

PROCESS = SUBPROCESS,
SUBPROCESS = (action1 -> SUBPROCESS2),
SUBPROCESS 2 = (action2 -> SUBPROCESS).

Choice |

Non-deterministic choice

Indexed Processes

BUFFER = (in[i:0..3] -> out[i] -> BUFFER).

Constants and Ranges

const N = 3
range T = 0..N

BUFF = (in[i:T] -> STORE[i]),
STORE[i:T] = (out[i] -> BUFF).

Guarded actions

COUNT(N=3) = COUNT[0],
COUNT[i:0..N] = ( when (i<N) inc -> COUNT[i+1]
                | when (i>0) dec -> COUNT[i-1]
                ).

STOP process

Concurrency in FSP

Parallel composition ||

Parallel composition rules

Shared Actions

Relabelling actions

Process Labelling

Client-Server example

CLIENT = (call -> wait -> continue -> CLIENT).
SERVER = (request -> service -> reply -> CLIENT).

||N_CLIENT_SERVER(N=2) = 
    (  forall[i:1..N] (c[i]:CLIENT) 
    || {c[i..N]}::(SERVER/{call/request, wait/reply})
    ).

Variable hiding

FSP Synchronisation

Coffman Conditions

4 necessary and sufficient conditions. All must occur for deadlock to happen

  1. serially reusable resources: processes must share some reusable resources between themselves under mutual exclusion
  2. incremental acquisition: processes hold on to allocated resources while waiting for other resources
  3. no preemption: once a process has acquired a resource, it can only release it voluntarily, i.e. it cannot be preempted/forced to release it
  4. wait-for cycle: a cycle exists in which each process holds a resource which its successor is waiting for

LTSA Deadlock

Monitors: FSP vs Java

public synchronized void act() throws InterruptedException {
    while (!cond) wait();
    // modify monitor data
    notifyAll();
}

Bounded buffers using monitors

// bounded buffer using monitor

// buffer size 
const N = 4
range U = 0..N

BUFFER = BUFF[0],
BUFF[i:U] = ( when (i < N) put -> BUFF[i+1]
      			| when (i > 0) get -> BUFF[i-1]
		      	).

PRODUCER = (put -> PRODUCER).
CONSUMER = (get -> CONSUMER).

||BOUNDED_BUFFER = (PRODUCER || CONSUMER || BUFFER).

Bounded buffers using semaphores

// bounded buffer using semaphores

// buffer size 
const N = 4
range U = 0..N

// up/signal: return permit
// down/wait: block until permit acquired
SEMAPHORE(X=N) = SEMAPHORE[X],
SEMAPHORE[i:U] = 
	( when (i < N) signal -> SEMAPHORE[i+1]
	| when (i > 0) wait -> SEMAPHORE[i-1]
	).

BUFFER =
	// given a put, one empty token is removed, and one full token is acquired
	( empty.wait -> put -> full.signal -> BUFFER
	// given a get, one full token is removed, and one empty token is acquired
	| full.wait -> get -> empty.signal -> BUFFER
	).

PRODUCER = (put -> PRODUCER).
CONSUMER = (get -> CONSUMER).

// empty: semaphore that blocks when buffer is empty
// full: semaphore that blocks when buffer is full
||BOUNDED_BUFFER = (empty:SEMAPHORE(N) || full:SEMAPHORE(0) 
					|| PRODUCER || CONSUMER || BUFFER).

BUFFER = (put -> BUFFER | get -> BUFFER).
PRODUCER = (empty.wait -> put -> full.signal -> PRODUCER).
CONSUMER = (full.wait -> get -> empty.signal -> CONSUMER).

Dining Philosophers problem

Philosophers 1: Deadlock

const N = 5

PHILOSOPHER = (think -> left.get -> right.get -> eat -> left.release -> right.release -> PHILOSOPHER).

FORK = (get -> release -> FORK).

||DINING_PHILOSOPHERS = 
	(  forall[i:0..N-1] p[i]:PHILOSOPHER
	|| forall[i:0..N-1] {p[i].left,p[((i-1)+N)%5].right}::FORK
	).

Philosophers 2

const N = 5

PHILOSOPHER(I=0) = 
	// even philosopher: left fork first
	( when (I%2 == 0) think -> left.get -> right.get -> eat -> left.release -> right.release -> PHILOSOPHER
	// odd philosopher: right fork first
	| when (I%2 == 1) right.get -> left.get -> eat -> left.release -> right.release -> PHILOSOPHER
	). 

FORK = (get -> release -> FORK).

||DINING_PHILOSOPHERS = 
	(  forall[i:0..N-1] p[i]:PHILOSOPHER(i)
	|| forall[i:0..N-1] {p[i].left,p[((i-1)+N)%5].right}::FORK
	).

Checking Safety in FSP

Counter

const N = 4
range T = 0..N

VAR = VAR[0],
// variable can be read/written to 
VAR[u:T] = (read[u] -> VAR[u] | write[v:T] -> VAR[v]).

CTR = ( read[x:T] -> 
        ( when (x<N) increment -> write[x+1] -> CTR
        | when (x==N) end -> END
        )
      )+{read[T], write[T]}.

// create a shared counter
||SHARED_COUNTER = ({a,b}:CTR || {a,b}::VAR).

Alphabet Extensions

Checking for interference

INTERFERENCE = (a.write[v:T] -> b.write[v] -> ERROR).

||SHARED_COUNTER = ({a,b}:CTR || {a,b}::VAR || INTERFERENCE).

ERROR is a predefined process signalling an error in the model, causing deadlock.

a.read.0
a.increment
b.read.0
a.write.1
b.increment
b.write.1

Mutual Exclusion

LOCK = (acquire -> release -> LOCK).

CTR = ( acquire -> read[x:T] -> 
        ( when (x<N) increment -> write[x+1] -> release -> CTR
        | when (x==N) release -> END
        )
      )+{read[T], write[T]}.


||LOCKED_SHAREDCOUNTER = ({a,b}:CTR || {a,b}::(LOCK||VAR)).

Safety and Liveness Properties

Error States

Safety Properties

ACTUATOR = (command -> ACT),
ACT = (respond -> ACTUATOR | command -> ACTUATOR).

property SAFE_ACTUATOR = (command -> respond -> SAFE_ACTUATOR).

||CHECK_ACTUATOR = (ACTUATOR || SAFE_ACTUATOR).

Safety property: interference

property NO_INTERFERENCE = ({a,b}.write[v:T] -> (when (v<N) {a,b}.write[v+1] -> NO_INTERFERENCE)).

||SHARED_COUNTER = ({a,b}:CTR || {a,b}::VAR || NO_INTERFERENCE).

Without lock

const N = 4
range T = 0..N

VAR = VAR[0],
VAR[u:T] = (read[u] -> VAR[u] | write[v:T] -> VAR[v]).

CTR = (read[x:T] -> ( when (x<N) increment -> write[x+1] -> CTR
					| when (x == N) end -> END
				))+{read[T], write[T]}.

property NO_INTERFERENCE = ({a,b}.write[v:T] -> (when (v<N) {a,b}.write[v+1] -> NO_INTERFERENCE)).

||SHARED_COUNTER = ({a,b}:CTR || {a,b}::VAR || NO_INTERFERENCE).

With lock

const N = 4
range T = 0..N

VAR = VAR[0],
VAR[u:T] = (read[u] -> VAR[u] | write[v:T] -> VAR[v]).

LOCK = (acquire -> release -> LOCK).
CTR = ( acquire -> read[x:T] -> 
        ( when (x<N) increment -> write[x+1] -> release -> CTR
        | when (x==N) release -> END
        )
      )+{read[T], write[T]}.

property NO_INTERFERENCE = ({a,b}.write[v:T] -> (when (v<N) {a,b}.write[v+1] -> NO_INTERFERENCE)).

||LOCKED_SHAREDCOUNTER = ({a,b}:CTR || {a,b}::(LOCK||VAR) || NO_INTERFERENCE).

Safety: Mutual Exclusion with Semaphores

// example of mutual exclusion with 10 processes attempting to access critical region, using binary semaphore

// number of loops
const M = 10

// up/signal: return permit
// down/wait: block until permit acquired
SEMAPHORE(X=1) = SEMAPHORE[X],
SEMAPHORE[i:0..X] = 
	( when (i < X) signal -> SEMAPHORE[i+1]
	| when (i > 0) wait -> SEMAPHORE[i-1]
	).

LOOP = (mutex.wait -> enter -> exit -> mutex.signal -> LOOP).

// check safety property: mutual exclusion, when a process enters critical
// region, the same process must exit critical region
property MUTEX = (p[i:1..M].enter -> p[i].exit -> MUTEX).

// compose M loops with binary semaphore
||M_LOOPS = (  p[1..M]:LOOP 
            || {p[1..M]}::mutex:SEMAPHORE(1)
            || MUTEX
            ).

Checking liveness in FSP

Toin coss and Fair Choice

COIN = (toss -> heads -> COIN | toss -> tails -> COIN).

Progress Properties in FSP

progress HEADS = {heads}
progress TAILS = {tails}

Trick coin

// choose between fair coin and trick coin
TWOCOIN = (pick -> COIN | pick -> TRICK),
COIN = (toss -> heads -> COIN | toss -> tails -> COIN),
// trick coin only returns heads
TRICK = (toss -> heads -> TRICK).

Progress Analysis

Terminal Sets

Strongly connected components

Actions occurring finitely often

LTSA Output

Readers/writers problem

// readers-writers problem

// set of actions is useful for extending alphabet
set Actions = {acquireRead, releaseRead, acquireWrite, releaseWrite}

READER = (acquireRead -> examine -> releaseRead -> READER)+Actions.
WRITER = (acquireWrite -> modify -> releaseWrite -> WRITER)+Actions.

const False = 0
const True = 1
range Bool = False..True

const NReaders = 5
const NWriters = 1

LOCK = LOCK[0][False],
LOCK[i:0..NReaders][b:Bool] = 
	( when (i == 0 && !b) acquireWrite -> LOCK[i][True]
	| when (b) releaseWrite -> LOCK[i][False]
    | when (i > 0) releaseRead -> LOCK[i-1][b]
	| when (i < NReaders && !b) acquireRead -> LOCK[i+1][b]
	).

||READERS_WRITERS = (  r[1..NReaders]:READER 
                    || w[1..NWriters]:WRITER 
                    || {r[1..NReaders],w[1..NWriters]}::LOCK).

Readers/writers safety property

property SAFE_RW = ( acquireRead -> READING[1] | acquireWrite -> WRITING),
// record the number of readers holding a read lock
READING[i:1..NReaders] = (acquireRead -> READING[i+1]
						 | when (i>1) releaseRead -> READING[i-1]
						 | when (i == 1) releaseRead -> SAFE_RW
						 ),
// block until the writer releases
WRITING = (releaseWrite -> SAFE_RW).

||READERS_WRITERS = (  r[1..NReaders]:READER 
				    || w[1..NWriters]:WRITER 
					|| {r[1..NReaders],w[1..NWriters]}::LOCK
					|| {r[1..NReaders],w[1..NWriters]}::SAFE_RW
					).

Readers/writers progress property

progress WRITE[i:1..NWriters] = {w[i].acquireWrite}
progress READ[i:1..NReaders] = {r[i].acquireRead}

Progress in a stressed system

FSP Action Priority

P = (a -> b -> P | c -> d -> P).
// a is higher priority than all other actions
||HIGH = P<<{a}.

is equivalent to: HIGH = (a -> b -> HIGH)

Readers-Writers Action priority

||RW_PROGRESS = READERS_WRITERS>>{r[1..NReaders].releaseRead, w[1..NWriters].releaseWrite}.

Improved readers-writers

LOCK = LOCK[0][False][0][False],
LOCK[i:0..NReaders][writing:Bool][nWaiting:0..NWriters][readerTurn:Bool] = 
	( when (!writing && (nWaiting == 0 || readerTurn)) 	
		acquireRead -> LOCK[i+1][writing][nWaiting][readerTurn]
    | releaseRead -> LOCK[i-1][writing][nWaiting][False]
	| when (i == 0 && !writing) 
		acquireWrite -> LOCK[i][True][nWaiting-1][readerTurn]
	| requestWrite -> LOCK[i][writing][nWaiting+1][readerTurn]
	| releaseWrite -> LOCK[i][False][nWaiting][True]
	).

Temporal Logic

Linear Temporal Logic (LTL)

Logic for Actions

FSP Fluents

FSP Indexed Fluents

fluent GREEN[i:1..2] = <{green[i]}, {yellow[i], red[i]}> initially 1

FSP Fluent Expressions

Temporal Logic: Always and Eventually

Always

always

Eventually

eventually

Example

fluent GREEN = <{green}, {yellow,red}> initially 1
fluent YELLOW = <{yellow}, {green, red}> initially 0
fluent RED = <{red}, {yellow, green}> initially 0

// the light is always green, yellow, or red
assert ALWAYS_A_COLOUR = [](GREEN || YELLOW || RED)

// the light will eventually become red
assert EVENTUALLY_RED = <>RED

Safety and Liveness

A = (a -> b -> END | c -> b -> END)

Combining temporal operators

eventually always

always eventually

Temporal logic laws

Temporal Logic Expressiveness

Until operator

until

Next Operator

Mutual Exclusion revisited

// binary semaphore
const K = 1
// 3 processes
const N = 3

LOOP = (mutex.down -> enter -> exit -> mutex.up -> LOOP).

SEMAPHORE = SEMAPHORE[K],
SEMAPHORE[i:0..K] = ( when (i < K) up -> SEMAPHORE[i+1]
                    | when (i > 0) down -> SEMAPHORE[i-1]
                    ).

||N_LOOPS = (p[1..N]:LOOP || {p[1..N]}::mutex:SEMAPHORE).

// fluent that holds when a process is in its critical section
fluent IN_CRITICAL[i:1..N] = <{p[i].enter}, {p[i].exit}>

// safety: mutual exclusion: only one thread in critical section at a time
// its always the case that there do not exist 2 processes in critical section at the same instant
assert MUTUAL_EXCLUSION = []!(exists[i:1..N-1] (IN_CRITICAL[i] && IN_CRITICAL[i+1..N]))

// liveness: progress, eventually all processes get to enter their critical section
// for all processes, it is eventually the case that it enters its critical section
assert EVENTUALLY_ENTER = forall[i:1..N] <>p[i].enter

Other mutex properties

// no process enters critical section before locking mutex
// it's the case that a process must not be in critical section until mutex acquired
assert NO_ENTER_BEFORE_MUTEX = forall[i:1..N] (!IN_CRITICAL[i] U p[i].mutex.down)

// when a process locks the mutex, it will be in its critical section at the next tick
assert LOCKED_OUT = forall[i:1..N] (p[i].mutex.down -> X IN_CRITICAL[i])

// when a thread enters its critical section, it must eventually exit
// for all processes, its always the case that a process entering its critical section must eventually exit
assert MUST_EXIT = forall[i:1..N] [](IN_CRITICAL[i] -> <>!IN_CRITICAL[i])

Checking all properties

Real world model checking

Concurrent Programming Languages

Concurrency via Shared Memory

Concurrency via message passing

Message passing

Message Protocols - Synchronous message: sender waits

sender waits

Message Protocols - Synchronous message: receiver waits

receiver waits

Asynchronous Communication

Message Protocols - Asynchronous message

async

Message Protocols - Simulating asynchronous messages

simulate async

Message Protocols - Simulating synchronous messages via async channel

simulate sync

Analogies

Addressing

Data Flow

Edit this page.