Part 8 - Synchronization problems (2)

TDA384_8

In this part we’ll cover the classical problems that occur when dealing with synchronization.

But within this paradigm, we don’t encounter the same problems as when using semaphores. Mutual exclusion is not an issue, since we never share any resource, the big problem today will be synchronization and coordination.

We’ll see that many problems have the solution of a server-client architecture.

Barriers

Let’s quickly recap what our barriers should be able to do:

1
-module(barrier).
2

3
% Initialize barrier for ‘Expected’ processes
4
init(Expected) ->
5
    % TODO
6

7
% Block at ‘Barrier’ until all processes have reached it
8
wait(Barrier) ->
9
    % TODO

Since we are talking about processes, it’s natural to have a process for the barrier itself. This process keeps track of what other processes that has arrived at the barrier point.

When a new process arrives at the barrier, it sends a arrived message to the barrier process. When the list of all arrived processes is complete, the barrier process sends a continue message to everyone.

After notifying all other processes, the barrier processes itself, goes back to its initial state.

So we need to implement the barrier’s event loop as a server function:

1
barrier(Arrived, Expected, PidRefs)

Let’s implement this:

1
% event loop of barrier for ‘Expected’ processes
2
% Arrived: number of processes arrived so far
3
% PidRefs: list of {Pid, Ref} of processes arrived so far
4

5
% All processes arrived notify all waiting processes:
6
barrier(Arrived, Expected, PidRefs) when Arrived =:= Expected ->
7
    [To ! {continue, Ref} || {To, Ref} <- PidRefs],
8

9
    % Reset barrier
10
    barrier(0, Expected, []);
11

12
% Still waiting for some processes
13
barrier(Arrived, Expected, PidRefs) ->
14
    receive
15
        {arrived, From, Ref} ->
16
            % one more arrived: add {From, Ref} to PidRefs list:
17
            barrier(Arrived + 1, Expected, [{From, Ref}|PidRefs])
18
end.

Now for the wait function:

1
% Block at ‘Barrier’ until all processes have reached it
2
wait(Barrier) ->
3
    % Notify barrier of arrival
4
    Ref = make _ ref(),
5

6
    % Wait for signal to continue
7
    Barrier ! {arrived, self(), Ref},
8

9
    receive {continue, Ref} -> through end.

And finally, the init function, simple:

1
% Initialize barrier for ‘Expected’ processes
2
init(Expected) ->
3
    spawn(fun () -> barrier(0, Expected, []) end).

Resource allocator

Let’s recap the problem, an *allocator grants users, exclusive access to a number of resources.

Users asynchronously request and release resources back. The allocator ensures exclusive access to a single user, and keeps tracks of the number of available resources.

So our module would look like:

1
-module(allocator).
2

3
% Register allocator with list of Resources
4
init(Resources) ->
5
    % TODO
6

7
% Get N resources from allocator
8
request(N) ->
9
    % TODO
10

11
% Release Resources to allocator
12
release(Resources) ->
13
    % TODO

The user would perform something like:

1
user() ->
2
    % How many resources are needed?
3
    N = howMany(),
4

5
    % Get resources from allocator
6
    Resources = allocator:request(N),
7

8
    % Do something with resources
9
    use(Resources),
10

11
    % Release resources
12
    allocator:release(Resources),
13

14
    user().

Again, in the message-passing world, using a server-client architecture often solves the problem.

We dedicate a process to the allocator, which keeps track of list of resources.

When a process requests for some resources that are available, the allocator sends a granted message. Then accordingly removes those resources from the list.

When a process releases some resources, the allocator sends a released, and then adds the resources to the list.

If requests exceed the availability, the fall into our built-in mailbox. The allocator process will resolve this as soon as they pattern-match again (resources available again).

1
allocator(Resources) ->
2
    % Count how many resources are available
3
    Available = length(Resources),
4
    receive
5

6
        % Serve requests if enough resources are available
7
        {request, From, Ref, N} when N =< Available ->
8

9
            % Granted ++ Remaining =:= Resources
10
            % Length(Granted) =:= N
11
            {Granted, Remaining} = lists:split(N, Resources),
12

13
            % Send resources to requesting process
14
            From ! {granted, Ref, Granted},
15

16
            % Continue with Remaining resources
17
            allocator(Remaining);
18

19

20
        % Serve releases
21
        {releases, From, Ref, Released} ->
22
            % Notify releasing process
23
            From ! {released, Ref},
24

25
            % Continue with previous and released resources
26
            allocator(Resources ++ Released)

The request function:

1
% Get N resources from allocator, gets blocked if not available
2
request(N) ->
3
    Ref = make_ref(),
4
    allocator ! {request, self(), Ref, N},
5
    recieve {granted, Ref, Granted} -> Granted end.
6

7
% Release Resources to allocator
8
release(Resources) ->
9
    Ref = make_ref(),
10
    allocator ! {release, self(), Ref, Resources},
11
    recieve {released, Ref} -> released end.

Producer-consumer

Recap; Implement a buffer such that:

Producers and consumers access the buffer atomically
Consumers block when the buffer is empty
Producers block when the buffer is full (bounded buffer variant)

1
-module(buffer).
2

3
% Initialize buffer with size Bound
4
init_buffer(Bound) ->
5
    % TODO
6

7
% Put Item in Buffer; Block if full
8
put(Buffer, Item) ->
9
    % TODO
10

11
% Get Item from Buffer; Block if empty
12
get(Buffer) ->
13
    % TODO

The producer and buffer:

1
producer(Buffer) ->
2
    Item = produce(),
3
    buffer:put(Buffer, Item),
4
    producer(Buffer).
5

6

7
consumer(Buffer) ->
8
    Item = buffer:get(Buffer),
9
    % Do something with Item
10

11
    consume(Item),
12
    consumer(Buffer).

At this point you pretty much can see the pattern here that arises:

1
buffer(Content, Count, Bound) ->
2
    receive
3

4
    % Serve gets when buffer not empty
5
    {get, From, Ref} when Count > 0 ->
6
        % Match first item
7
        [First | Rest] = Content,
8

9
        % Send it out
10
        From ! {item, Ref, First},
11

12
        % Remove it from buffer
13
        buffer(Rest, Count-1, Bound);
14

15
    % Serve puts when buffer not full
16
    {put, From, Ref, Item} when Count < Bound ->
17

18
        % Send ack
19
        From ! {done, Ref},
20

21
        % Add item to end
22
        buffer(Content ++ [Item], Count + 1, Bound)
23
end.

In this solution, both a bounded and unbounded will work - due to Erlang’s order between numbers and atoms!

Now for get and put:

1
% Get item from ‘Buffer’; block if empty
2
get(Buffer) ->
3
    Ref = make_ref(),
4
    Buffer ! {get, self(), Ref},
5
    receive {item, Ref, Item} -> Item end.
6

7
% Put ‘Item’ in ‘Buffer’; block if full
8
put(Buffer, Item) ->
9
    Ref = make_ref(),
10
    Buffer ! {put, self(), Ref, Item},
11
    receive {done, Ref} -> done end.

Readers-writers

1
-module(board).
2

3
% Register board with Name
4
init(Name) ->
5
    % TODO
6

7
% Get read access to Board
8
begin_read(Board) ->
9
    % TODO
10

11
% Release read access to Board
12
end_read(Board) ->
13
    % TODO
14

15
% Get write access to Board
16
begin_write(Board) ->
17
    % TODO
18

19
% Release write access to Board
20
end_write(Board) ->
21
    % TODO

Our first naive server function would be:

1
% ‘Readers’ active readers and ‘Writers’ active writers
2
board_row(Readers, Writers) ->
3
receive
4
    {begin_read, From, Ref} when Writers =:= 0 ->
5
        From ! {ok_ to_ read, Ref},
6
        board_row(Readers+1, Writers);
7

8
    {begin_write, From, Ref} when (Writers =:= 0) and (Readers =:= 0) ->
9
        From ! {ok_ to_ write, Ref},
10
        board_row(Readers, Writers+1);
11

12
    {end_read, From, Ref} -> From ! {ok, Ref},
13
        board_row(Readers-1, Writers);
14

15
    {end_write, From, Ref} -> From ! {ok, Ref},
16
        board_row(Readers, Writers-1)
17
end.

Just as our naive solution when using semaphores, this doesn’t prevent starvation due to this version prioritizes readers.

The solution based on two monitors is a approach here, but it’s quite cumbersome for a message-passing program.

We instead implement two macro states:

Empty - no readers or writers
Readers - Readers but no writers

The initial board is in empty state, then:

When board is in state emtpy:
- Read requests - served immediately, then switches to readers state.
- Write requests - served immediately and synchronously, wait until writing ends, then go into empty state.

When board is in state readers:
- Read requests - served immediately and stays in readers.
- Write requests - served as soon as possible, board waits until all reading ends, then request is served. Back to empty state.

For this we’ll need two server functions, empty_board and readers_board:

1
% Board with no readers and no writers
2
empty_board() ->
3
    receive
4

5
    % Serve read request
6
    {begin_read, From, Ref} ->
7

8
        % Notify reader
9
        From ! {ok_to_read, Ref},
10

11
        % Board has one reader
12
        readers_board(1);
13

14
    % Serve write request synchronously
15
    {begin_write, From, Ref} ->
16
        % Notify writer
17
        From ! {ok_to_write, Ref},
18

19
        % Wait for writer to finish
20
        Receive
21
            {end_write, _From, _Ref} ->
22
                % Board is empty again
23
                empty_board()
24
        end
25
end.
26

27
% Board with no readers (and no writers)
28
readers_ board(0) -> empty_ board();
29

30
% Board with ‘Readers’ active readers
31
% (and no writers)
32
readers_board(Readers) ->
33
    receive
34

35
        % Serve write request
36
        {begin_write, From, Ref} ->
37
            % Wait until all ‘Readers’ have finished
38
            [receive {end_read, _From, _Ref} -> end_read end || _ <- lists:seq(1, Readers)],
39

40
            % Notify writer
41
            From ! {ok_to_write, Ref},
42

43
            % Wait for writer to finish
44
            receive
45
                {end_write, _From, _Ref} -> empty_board()
46
            end;
47

48
        % Serve read request
49
        {begin_read, From, Ref} ->
50
            % Notify reader
51
            From ! {ok _ to _ read, Ref},
52

53
            % Board has one more reader
54
            readers _ board(Readers+1);
55

56
        % Serve end read
57
        {end_read, From, Ref} ->
58

59
            % Board has one less reader
60
            readers_board(Readers-1)
61
end.

Dining Philosophers

1
-module(philosophers).
2

3
% Set up table of N philosophers
4
init(N) ->
5
    % TODO
6

7
% Philosopher picks up Fork
8
get_fork(Fork) ->
9
    % TODO
10

11
% Philosopher releases Fork
12
put_fork(Fork) ->
13
    % TODO

We could explore the solutions we did based on locking and breaking symmetry - but there is a solution which better fits into the message-passing paradigm

We have a waiter (process) who supervises access to the table. So each philosopher asks for permission to sit at the table before picking up both forks.

So, as long as the waiter allows strictly fewer philosopher than the total number of forks to sit around the table, deadlock and starvation are avoided.

Waiter interface:

1
% Ask Waiter to be seated; may wait
2
sit(Waiter) ->
3
    % TODO
4

5
% Ssk Waiter to leave
6
leave(Waiter) ->
7
    % TODO

Our server function:

1
waiter(Eating, Seats) ->
2
    receive
3

4
    % Serve as long as seats are available
5
    {sit, From, Ref} when Eating < Seats ->
6
        From ! {ok_to_sit, Ref},
7

8
        % One more eating
9
        waiter(Eating+1, Seats);
10

11
    % Can leave at any time
12
    {leave, From, Ref} ->
13
        From ! {ok_to_leave, Ref},
14

15
        % One less eating
16
        waiter(Eating-1, Seats)
17
end.

And sit and leave:

1
% ask Waiter to be seated; may wait
2
sit(Waiter) ->
3
    Ref = make _ ref(),
4
    Waiter ! {sit, self(), Ref},
5
    receive {ok_to_sit, Ref} -> ok end.
6

7
% ask Waiter to leave
8
leave(Waiter) ->
9
    Ref = make _ ref(),
10
    Waiter ! {leave, self(), Ref},
11
    receive {ok_to_leave, Ref} -> ok end.

Now, each fork is also a process, which keeps track of whether the for is free or not.

Server function:

1
% Fork not held by anyone
2
fork() ->
3
    receive
4
        {get, From, Ref} ->
5
            From ! {ack, Ref},
6

7
            % Fork held
8
            fork(From)
9
end.
10

11
% a fork held by Owner
12
fork(Owner) ->
13
    receive
14
        {put, Owner, _ Ref} ->
15
            % Fork not held
16
            fork()
17
end.

and the get and put for the forks:

1
% Pick up Fork; block until available
2
get_fork(Fork) ->
3
    Ref = make _ ref(),
4
    Fork ! {get, self(), Ref},
5
    receive {ack, Ref} -> ack end.
6

7
% Put down Fork
8
put_fork(Fork) ->
9
    Ref = make _ ref(),
10
    Fork ! {put, self(), Ref}.

And finally, the init function for the whole problem:

1
% Set up table of ‘N’ philosophers
2
init(N) ->
3
    % Spawn waiter process
4
    Waiter = spawn(fun () -> waiter(0, N-1) end),
5

6
    % [1, 2, ..., N]
7
    Ids = lists:seq(1,N),
8

9
    % Spawn fork processes
10
    Forks = [spawn(fun fork/0) || _ <- Ids],
11

12
    % Spawn philosopher processes
13
    [spawn(fun () ->
14
        Left = lists:nth(I, Forks),
15

16
        % 1-based indexes
17
        Right = lists:nth(1+(I rem N), Forks),
18
        philosopher(#forks{left=Left, right=Right}, Waiter)
19
    end) || I <- Ids].