Concurrency

• We will use concurrency concepts repeatedly

• Recap: Why is concurrency important/useful?
• Allows safe/multiplexed access to shared resources
• i.e. safe operation with multiple independent entities
• Single core CPUs to wide area distributed systems
• Why is sharing resources useful? Real life example?

With (some at least!)

Concurrency Control …or not! 4
Concurrency is key in DS

• Today, we start with threads on a single node

…. we will expand them to multiple machines
• Key assumption: ignore independent failures
• Code Concurrency: Terms (Ref: Dijkstra ‘65,’68)
• Critical Section: piece of code accessing a shared resource,
usually variables or data structures
• Race Condition: Multiple threads of execution enter CS at the
same time, update shared resource, leading to undesirable outcome
• Indeterminate Program: One or more Race Conditions, output of
program depending on ordering, non deterministic
• So, what is the solution?
• Mutual Exclusion!

5
Achieving Mutual Exclusion

• Mutual Exclusion: guarantee that only a single

thread/process enters a CS, avoiding races
• Desired Properties of Mutual Exclusion
• Mutual Exclusion (Correctness): single process in CS at one time
• Progress (Efficiency): Processes don’t wait for available
resources, or no spin-locks => no wasted resources
• Bounded Waiting (Fairness): No process waits for ever for a
resource, i.e. a notion of fairness

• Trivial solution if we didn’t care about fairness! What is it?

• Can you give an example of concurrent access to files in an OS
which does not lead to concurrency problems or need mutex?

6
Achieving Mutual Exclusion

• Mutual Exclusion: Usually need some H/W support

• Test-and-Set instruction (atomic testing/setting of a value)
TS (<memloc>) // Do this before entering CS
{ if <memloc>==1 Acquire_Mutex(<mutex>)
{ {
<memloc>=0; while(!TS(<mutex>))
return 1; }
}
else
{CS: Critical Section of Code}
return 0;
}
// After Exiting CS
Note: Atomic/Non-interruptable Release_Mutex(<mutex>)
{
We can use Test-and-Set to <mutex> = 1
implement Mutex }

• Build more complex primitives around this H/W

7
Classical Model of Concurrency

• Threads running within an address space

• Private/Shared state => primitive to access shared state
• E.g. Semaphores: Integer variable ‘x’ with 2 operations

x.P():
while (x == 0) wait;
x–
// binary Semaphore = Mutex
x.V(): x=1; Unlocked, Resource Available
x++ x=0; Locked, Must wait for resource

Note: `P’ and `V’ operations are

Atomic, not interruptable

What is a Binary Semaphore then?

8
Semaphones to Create a FIFO

• Idea: Want to create (thread safe) FIFO queue

• Similar to GO channel, without a capacity limit
b.Init():
Initialize values

b.Insert(x)
Insert item into queue

b.Remove()
Block until queue not empty (if necessary)
Return element at head of queue

b.Flush():
Clear queue

// Assume we have a sequential implementation of a FIFO buffer

// b represents a structure with fields
sb: Sequential buffer implementation
mutex: Mutual exclusion lock

• What do we need to do next? Wrap with Mutex 9

Thread Safe FIFO Queue

b.Init(): .
b.sb = NewBuf() .
b.mutex = 1 .
.
b.Insert(x): b.Remove():
b.mutex.lock() retry:
b.sb.Insert(x) b.mutex.lock()
b.mutex.unlock() if !(b.sb.len() > 0) {
b.mutex.unlock()
b.Remove(): goto retry
b.mutex.lock() }
x = b.sb.Remove() .
b.mutex.unlock() .
return x .

b.Flush(): Potential Fix. Does this do it?

b.mutex.lock()
b.sb.Flush()
b.mutex.unlock()
No, not really. This is a spin-lock. Wastes resources and
Is this correct? unclear if a Insert(x) will ever make progress.
This is inefficient and can be a potential LIVELOCK.
No, what if Remove is called
and the buffer is empty? 10
Thread Safe FIFO Queue
• Use semaphore “items” (Bryant and O’hallaron)
b.Init():
b.sb = NewBuf()
.
b.mutex = 1 .
b.items = 0 .
b.Remove():
b.Insert(x): b.mutex.lock()
b.mutex.lock() b.items.P()
b.sb.Insert(x)
x = b.sb.Remove()
b.mutex.unlock()
b.items.V() b.mutex.unlock()
return x
b.Remove(): .
b.items.P() .
b.mutex.lock() .
x = b.sb.Remove()
b.mutex.unlock()
return x OK, now we are good!
b.Flush(): No, not really. We avoid the race condition but prone
b.mutex.lock() to a DEADLOCK. Can reach point where no one is able
b.sb.Flush() to proceed. How?
b.items = 0
b.mutex.unlock() Lets say you call Remove when buffer is empty.
Remove gets lock. Somewhere else, want to Insert,
This fixes it. but can't get past lock.
No, what if Flush is called after
b.items.P() and the buffer is empty? Hard to fix, need different approach. 11
Buffer using ConditionVars
• Lets use Condition Variables (cvars)
• cvars provide a sync point, one thread suspended until
activated by another. (more efficient way to wait than
spin lock )
• cvar always associated with mutex
• Wait() and Signal() operations defined with cvars
cvar.Wait():
Must be called after locking mutex.
Atomically: release mutex & suspend operation

When resume, lock mutex (but maybe not right away)

cvar.Signal():
If no thread suspended, then NO-OP
Wake up (at least) one suspended thread.
(Typically do within scope of mutex, but not required)

12
Buffer using ConditionVars
b.Init():
b.sb = NewBuf()
b.mutex = 1
b.cvar = NewCond(b.mutex)

b.Insert(x):
b.mutex.lock()
b.sb.Insert(x)
b.sb.Signal() // Signal() If no thread suspended, then NO-OP
b.mutex.unlock() Wake up (at least) one suspended thread.
(Typically do within scope of mutex, but not required)
b.Remove():
b.mutex.lock()
if b.sb.Empty() {
b.cvar.wait() // wait() Note that lock is first released & then retaken
} Atomically: release mutex & suspend operation
x = b.sb.Remove() When resume, lock mutex (but maybe not right away)
b.mutex.unlock()
return x

b.Flush():
b.mutex.lock()
b.sb.Flush()
b.mutex.unlock()

13
Buffer using ConditionVars
b.Init(): Cvar.wait() // 3 steps
b.sb = NewBuf()
b.mutex = 1 Atomically {release lock + suspend operation)
b.cvar = NewCond(b.mutex) .
.
b.Insert(x): .
b.mutex.lock() Resume Execution
b.sb.Insert(x) .
b.sb.Signal() // point of vulnerability, someone can flush here
b.mutex.unlock() .
Lock Mutex
b.Remove():
b.mutex.lock()
if b.sb.Empty() {
b.cvar.wait() What a signal means:
} Mesa semantics (looser) vs Hoare Semantics (tighter)
x = b.sb.Remove()
b.mutex.unlock()
return x

b.Flush():
b.mutex.lock()
b.sb.Flush()
b.mutex.unlock()

Still a small problem.

14
Buffer using ConditionVars
b.Init(): // What happens
b.sb = NewBuf() Lock
b.mutex = 1 if !sb.empty() goto ready
b.cvar = NewCond(b.mutex) Unlock
wait for signal
b.Insert(x): Lock
b.mutex.lock()
if !sb.empty() goto ready
b.sb.Insert(x) Unlock
b.sb.Signal() wait for signal
b.mutex.unlock() Lock
. . .
b.Remove(): ready: Can safely assume have lock & that buffer nonempty
b.mutex.lock()
while b.sb.Empty() {
b.cvar.wait() // wait() Note that lock is first released & then retaken
} Atomically: release mutex & suspend operation
x = b.sb.Remove() When resume, lock mutex (but maybe not right away)
b.mutex.unlock()
return x

b.Flush(): Simple Rule: With Mesa semantics, use while loops

b.mutex.lock() to recheck the condition. Always safe to do so.
b.sb.Flush()
b.mutex.unlock()
Additional examples for Cvars in the Remzi Ref.
Change “if” to While

15