303 lines
No EOL
7.7 KiB
Markdown
303 lines
No EOL
7.7 KiB
Markdown
**Object:** entity with an implementation (hidden) and an interface (visible), made up of a set of operations and a specification of the behavior.
|
||
**Concurrent:** if the object can be accessed by different processes.
|
||
|
||
**Semaphore:** is a shared counter S accessed via primitives $up$ and $down$ s.t.:
|
||
- is initialized at s0 >= 0
|
||
- it is alwayz >= 0
|
||
- *up* atomically increases S
|
||
- *down* atomically decreases S if it is not 0, otherwise the calling process is blocked and waits.
|
||
|
||
Main use: **prevent busy waiting**: suspend processes that cannot perform *down*.
|
||
|
||
**Strong semaphore:** if uses a FIFO policy for blocking/unblocking, otherwise it's **weak**.
|
||
**Binary semaphore:** if it is at most 1 (also *up* are blocking).
|
||
|
||
A semaphore needs two underlying objects:
|
||
- a counter initialized at s0 that can also become negative (see the implementation below to understand)
|
||
- a data structure (typically a queue), initially empty, to store suspended processes.
|
||
|
||
#### Ideal implementation
|
||
```
|
||
S.down() :=
|
||
S.counter--
|
||
if S.counter < 0 then
|
||
enter into S.queue
|
||
SUSPEND
|
||
return
|
||
|
||
S.up() :=
|
||
S.counter++
|
||
if S.counter <= 0 then
|
||
activate a proc from S.queue
|
||
return
|
||
```
|
||
>[!note] note
|
||
>if S.counter ≥ 0, then this is the value of the semaphore; otherwise, S.counter tells you how many processes are suspended in S
|
||
>
|
||
>all operations are in MUTEX
|
||
|
||
|
||
#### Actual implementation
|
||
```
|
||
Let t be a test&set register initialized at 0
|
||
|
||
S.down() :=
|
||
Disable interrupts
|
||
wait S.t.test&set() = 0
|
||
S.counter--
|
||
if S.counter < 0 then
|
||
enter into S.queue
|
||
S.t <- 0
|
||
Enable interrupts
|
||
SUSPEND
|
||
else
|
||
S.t <- 0
|
||
Enable interrupts
|
||
return
|
||
|
||
S.up() :=
|
||
Disable interrupts
|
||
wait S.t.test&set() = 0
|
||
S.counter++
|
||
if S.counter <= 0 then
|
||
activate a proc from S.queue
|
||
S.t <- 0
|
||
Enable interrupts
|
||
return
|
||
```
|
||
Same as before but we use test&set to actually ensure MUTEX (of course we could have used any other hardware MUTEX implementation seen so far).
|
||
|
||
#### (Single) Producer/Consumer
|
||
It's a shared FIFO buffer of size k.
|
||
- `BUF[0,…,k-1]`: generic registers (not even safe) accessed in MUTEX
|
||
- `IN/OUT` : two variables pointing to locations in `BUF` to (circularly) insert/remove items, both initialized at 0
|
||
- `FREE/BUSY`: two semaphores that count thew number of free/busy cells of `BUF`, initialized at k and 0 respectively.
|
||
|
||
```
|
||
B.produce(v) :=
|
||
FREE.down() # blocking if FREE becomes negative
|
||
BUF[IN] <- v
|
||
IN <- (IN+1) mod k # mod k since it is circular
|
||
BUSY.up() # "there is something to consume"
|
||
return
|
||
|
||
B.consume() :=
|
||
BUSY.down() # it blocks if there is nothing to consume
|
||
tmp <- BUF[OUT]
|
||
OUT <- (OUT+1) mod k
|
||
FREE.up()
|
||
return tmp
|
||
```
|
||
|
||
> [!note] Downside
|
||
> Reading from / writing into the buffer can be very expensive!
|
||
|
||
#### (Multiple) Producers/Consumers
|
||
**Accessing BUF in MUTEX slows down the implementation**, we would like to have the possibility to read and write from different cells **in parallel**.
|
||
- we use two arrays FULL and EMPTY of atomic boolean registers, initialized at ff (all zeros) and tt (all ones), respectively
|
||
- we have two extra semaphores SP and SC, both initialized at 1
|
||
|
||
```
|
||
B.produce(v) :=
|
||
FREE.down()
|
||
SP.down()
|
||
while not EMPTY[IN] do
|
||
IN <- (IN+1) mod k
|
||
i <- IN
|
||
EMPTY[IN] <- ff
|
||
SP.up()
|
||
BUF[i] <- v
|
||
FULL[i] <- tt
|
||
BUSY.up()
|
||
return
|
||
|
||
B.consume() :=
|
||
BUSY.down()
|
||
SC.down()
|
||
while not FULL[OUT] do
|
||
OUT <- (OUT+1) mod k
|
||
o <- OUT
|
||
FULL[OUT] <- ff
|
||
SC.up()
|
||
tmp <- BUF[o]
|
||
EMPTY[o] <- tt
|
||
FREE.up()
|
||
return tmp
|
||
```
|
||
Thanks to the semaphores, we are sure that while loops will not go on forever! The loops starts only if there is at least a FREE / BUSY cell.
|
||
|
||
#### (Multiple) Producers/Consumers - Wrong solution
|
||
**EXERCISE** this example is wrong
|
||
```
|
||
B.produce(v) :=
|
||
FREE.down()
|
||
SP.down()
|
||
i <- IN
|
||
IN <- (IN + 1) mod k
|
||
EMPTY[in] <- ff
|
||
SP.up()
|
||
BUF[i] <- v
|
||
FULL[i] <- tt
|
||
BUSY.up()
|
||
return
|
||
|
||
B.consume() :=
|
||
BUSY.down()
|
||
SC.down()
|
||
o <- OUT
|
||
OUT <- (OUT+1) mod k
|
||
FULL[OUT] <- ff
|
||
SC.up()
|
||
tmp <- BUF[o]
|
||
EMPTY[o] <- tt
|
||
FREE.up()
|
||
return tmp
|
||
```
|
||
okay, so the difference seems to be just the fact that there are no while loops to find the first free location, or the first location with data.
|
||
|
||
Let's imagine to have quick producers and a slow consumer:
|
||
producer A:
|
||
- writes at BUF[0], so IN becomes 1
|
||
- writes at BUF[1], so IN becomes 2
|
||
- ...
|
||
- writes at BUF[n], so IN becomes 0
|
||
|
||
**Now there are no free slots.**
|
||
|
||
consumer A:
|
||
- starts reading BUF[0], now OUT is 1
|
||
- it is a sloooooooooow consumer
|
||
|
||
consumer B:
|
||
- finds OUT at 1, sets OUT at 2 and starts reading
|
||
- finishes reading (before consumer A)
|
||
- sets EMPTY[1] <- tt
|
||
- calls `FREE.up()`
|
||
|
||
producer A:
|
||
- as consumer B called `FREE.up()`, the producer can finally produce
|
||
- remember, IN is set to 0!
|
||
- so producer A will write at `BUF[0]`
|
||
- but wait! Consumer B is still reading there
|
||
- **Producer A doesn't give a fuck.**
|
||
![]()
|
||
*don't be like Producer A, be more like Bob, who always scans EMPTY before!*
|
||
|
||
So the issue here is that producers just assume that IN is the first available slot. But it its not necessarily the case.
|
||
#### The Readers/Writers problem
|
||
- Several processes want to access a file
|
||
- Readers may simultaneously access the file
|
||
- At most one writer at a time
|
||
- Reads and writes are mutually exclusive
|
||
|
||
>[!note] Remark
|
||
>this generalizes the MUTEX problem (MUTEX = RW with only writers)
|
||
|
||
The read/write operations on the file will all have the following shape:
|
||
```
|
||
conc_read() :=
|
||
begin_read()
|
||
read()
|
||
end_read()
|
||
|
||
conc_write() :=
|
||
begin_write()
|
||
write()
|
||
end_write()
|
||
```
|
||
|
||
##### Weak priority to Readers
|
||
- If a reader arrives during a read, it can surpass possible writers already suspended (risk of starvation for the writes)
|
||
- When a writer terminates, it activates the first suspended process, irrispectively of whether it is a reader or a writer (so, the priority to readers is said «weak»)
|
||
|
||
```
|
||
GLOB_MUTEX and R_MUTEX semaphores init. at 1
|
||
R a shared register init. at 0
|
||
|
||
begin_read() :=
|
||
R_MUTEX.down()
|
||
R++ # currently active readers
|
||
if R = 1 then GLOB_MUTEX.down()
|
||
R_MUTEX.up()
|
||
return
|
||
|
||
end_read() :=
|
||
R_MUTEX.down()
|
||
R--
|
||
if R = 0 then GLOB_MUTEX.up()
|
||
R_MUTEX.up()
|
||
return
|
||
|
||
begin_write() :=
|
||
GLOB_MUTEX.down()
|
||
return
|
||
|
||
end_write() :=
|
||
GLOB_MUTEX.up()
|
||
return
|
||
```
|
||
If readers keeps arriving, they surpass writers. But when a writer terminates it activates the first suspended process, which can be a reader or a writer.
|
||
##### Strong priority to Readers
|
||
When a writer terminates, it activates the first reader, if there is any, or the first writer, otherwise.
|
||
|
||
```
|
||
GLOB_MUTEX and R_MUTEX and W_MUTEX semaphores init. at 1
|
||
R a shared register init. at 0
|
||
|
||
begin_read() := end_read() :=
|
||
come prima come prima
|
||
|
||
begin_write() :=
|
||
W_MUTEX.down()
|
||
GLOB_MUTEX.down()
|
||
return
|
||
|
||
end_write() :=
|
||
GLOB_MUTEX.up()
|
||
W_MUTEX.up()
|
||
return
|
||
```
|
||
This is prioritizing readers as `GLOB_MUTEX.up()` is called before `W_MUTEX.up()`, this is stronger that what it is done before.
|
||
##### Weak priority to Writers
|
||
```
|
||
GLOB_MUTEX, PRIO_MUTEX (to prioritize the writers), R_MUTEX and W_MUTEX semaphores init. at 1
|
||
R and W shared registers init. at 0
|
||
|
||
begin_read() :=
|
||
PRIO_MUTEX.down()
|
||
R_MUTEX.down()
|
||
R++
|
||
if R = 1 then
|
||
GLOB_MUTEX.down()
|
||
R_MUTEX.up()
|
||
PRIO_MUTEX.up()
|
||
return
|
||
|
||
end_read() := # like weak priority to readers
|
||
R_MUTEX.down()
|
||
R--
|
||
if R = 0 then
|
||
GLOB_MUTEX.up()
|
||
R_MUTEX.up()
|
||
return
|
||
|
||
def begin_write() :=
|
||
W_MUTEX.down()
|
||
W++
|
||
if W = 1 then
|
||
PRIO_MUTEX.down()
|
||
W_MUTEX.up()
|
||
GLOB_MUTEX.down()
|
||
return
|
||
|
||
def end_write() :=
|
||
GLOB_MUTEX.up()
|
||
W_MUTEX.down()
|
||
W--
|
||
if W = 0 then
|
||
PRIO_MUTEX.up()
|
||
W_MUTEX.up()
|
||
return
|
||
```
|
||
This is prioritizing writers as if there are writers waiting, they will be waiting at `GLOB_MUTEX.down()`. This semaphore is upped before `PRIO_MUTEX` which is the one that blocks readers.
|
||
But writers won't be able to writer until there are no readers, if they keep coming, they will block the writers as `GLOB_MUTEX` will never be upped. |