source: doc/theses/colby_parsons_MMAth/text/channels.tex @ 6ece306

% ======================================================================
% ======================================================================
\chapter{Channels}\label{s:channels}
% ======================================================================
% ======================================================================

Most modern concurrent programming languages do not subscribe to just one style of communication among threads and provide features that support multiple approaches.
Channels are a concurrent-language feature used to perform \Newterm{message-passing concurrency}: a model of concurrency where threads communicate by sending data as messages (mostly non\-blocking) and synchronizing by receiving sent messages (blocking).
This model is an alternative to shared-memory concurrency, where threads communicate directly by changing shared state.

Channels were first introduced by Kahn~\cite{Kahn74} and extended by Hoare~\cite{Hoare78} (CSP).
Both papers present a pseudo (unimplemented) concurrent language where processes communicate using input/output channels to send data.
Both languages are highly restrictive.
Kahn's language restricts a reading process to only wait for data on a single channel at a time and different writing processes cannot send data on the same channel.
Hoare's language restricts both the sender and receiver to explicitly name the process that is the destination of a channel send or the source of a channel receive.
These channel semantics remove the ability to have an anonymous sender or receiver.
Additionally all channel operations in CSP are synchronous (no buffering).
Advanced channels as a programming language feature has been popularized in recent years by the language Go~\cite{Go}, which encourages the use of channels as its fundamental concurrent feature.
It was the popularity of Go channels that lead to their implemention in \CFA.
Neither Go nor \CFA channels have the restrictions of the early channel-based concurrent systems.

\section{Producer-Consumer Problem}
A channel is an abstraction for a shared-memory buffer, which turns the implementation of a channel into the producer-consumer problem.
The producer-consumer problem, also known as the bounded-buffer problem, was introduced by Dijkstra~\cite[\S~4.1]{Dijkstra65}.
In the problem, threads interact with a buffer in two ways: producing threads insert values into the buffer and consuming threads remove values from the buffer.
In general, a buffer needs protection to ensure a producer only inserts into a non-full buffer and a consumer only removes from a non-empty buffer (synchronization).
As well, a buffer needs protection from concurrent access by multiple producers or consumers attempting to insert or remove simultaneously (MX).

\section{Channel Size}\label{s:ChannelSize}
Channels come in three flavours of buffers:
\begin{enumerate}
\item
Zero sized implies the communication is synchronous, \ie the producer must wait for the consumer to arrive or vice versa for a value to be communicated.
\item
Fixed sized (bounded) implies the communication is asynchronous, \ie the producer can proceed up to the buffer size and vice versa for the consumer with respect to removal.
\item
Infinite sized (unbounded) implies the communication is asynchronous, \ie the producer never waits but the consumer waits when the buffer is empty.
Since memory is finite, all unbounded buffers are ultimately bounded;
this restriction must be part of its implementation.
\end{enumerate}

In general, the order values are processed by the consumer does not affect the correctness of the producer-consumer problem.
For example, the buffer can be LIFO, FIFO, or prioritized with respect to insertion and removal.
However, like MX, a buffer should ensure every value is eventually removed after some reasonable bounded time (no long-term starvation).
The simplest way to prevent starvation is to implement the buffer as a queue, either with a cyclic array or linked nodes.

\section{First-Come First-Served}
As pointed out, a bounded buffer requires MX among multiple producers or consumers.
This MX should be fair among threads, independent of the FIFO buffer being fair among values.
Fairness among threads is called \gls{fcfs} and was defined by Lamport~\cite[p.~454]{Lamport74}.
\gls{fcfs} is defined in relation to a doorway~\cite[p.~330]{Lamport86II}, which is the point at which an ordering among threads can be established.
Given this doorway, a CS is said to be \gls{fcfs}, if threads access the shared resource in the order they proceed through the doorway.
A consequence of \gls{fcfs} execution is the elimination of \Newterm{barging}, where barging means a thread arrives at a CS with waiting threads, and the MX protecting the CS allows the arriving thread to enter the CS ahead of one or more of the waiting threads.

\gls{fcfs} is a fairness property that prevents unequal access to the shared resource and prevents starvation, however it comes at a cost.
Implementing an algorithm with \gls{fcfs} can lead to \Newterm{double blocking}, where arriving threads block outside the doorway waiting for a thread in the lock entry-protocol and inside the doorway waiting for a thread in the CS.
An analogue is boarding an airplane: first you wait to get through security to the departure gates (short term), and then wait again at the departure gate for the airplane (long term).
As such, algorithms that are not \gls{fcfs} (barging) can be more performant by skipping the wait for the CS and entering directly;
however, this performance gain comes by introducing unfairness with possible starvation for waiting threads.

\section{Channel Implementation}
Currently, only the Go programming language provides user-level threading where the primary communication mechanism is channels.
Experiments were conducted that varied the producer-consumer problem algorithm and lock type used inside the channel.
With the exception of non-\gls{fcfs} or non-FIFO algorithms, no algorithm or lock usage in the channel implementation was found to be consistently more performant that Go's choice of algorithm and lock implementation.
Performance of channels can be improved by sharding the underlying buffer \cite{Dice11}. 
In doing so the FIFO property is lost, which is undesireable for user-facing channels.
Therefore, the low-level channel implementation in \CFA is largely copied from the Go implementation, but adapted to the \CFA type and runtime systems.
As such the research contributions added by \CFA's channel implementation lie in the realm of safety and productivity features.

The Go channel implementation utilitizes cooperation between threads to achieve good performance~\cite{go:chan}.
The cooperation between threads only occurs when producers or consumers need to block due to the buffer being full or empty.
In these cases the blocking thread stores their relevant data in a shared location and the signalling thread will complete their operation before waking them.
This helps improve performance in a few ways.
First, each thread interacting with the channel with only acquire and release the internal channel lock exactly once.
This decreases contention on the internal lock, as only entering threads will compete for the lock since signalled threads never reacquire the lock.
The other advantage of the cooperation approach is that it eliminates the potential bottleneck of waiting for signalled threads.
The property of acquiring/releasing the lock only once can be achieved without cooperation by \Newterm{baton passing} the lock.
Baton passing is when one thread acquires a lock but does not release it, and instead signals a thread inside the critical section conceptually "passing" the mutual exclusion to the signalled thread.
While baton passing is useful in some algorithms, it results in worse performance than the cooperation approach in channel implementations since all entering threads then need to wait for the blocked thread to reach the front of the ready queue and run before other operations on the channel can proceed.

In this work, all channel sizes \see{Sections~\ref{s:ChannelSize}} are implemented with bounded buffers.
However, only non-zero-sized buffers are analysed because of their complexity and higher usage.

\section{Safety and Productivity}
Channels in \CFA come with safety and productivity features to aid users.
The features include the following.

\begin{itemize}
\item Toggle-able statistic collection on channel behaviour that count channel and blocking operations.
Tracking blocking operations helps illustrate usage for tuning the channel size, where the aim is to reduce blocking.

\item Deadlock detection on channel deallocation.
If threads are blocked inside a channel when it terminates, this case is detected and the user is informed, as this can cause a deadlock.

\item A @flush@ routine that delivers copies of an element to all waiting consumers, flushing the buffer.
Programmers use this mechanism to broadcast a sentinel value to multiple consumers.
Additionally, the @flush@ routine is more performant then looping around the @insert@ operation since it can deliver the elements without having to reacquire mutual exclusion for each element sent.
\end{itemize}

\subsection{Toggle-able Statistics}
As discussed, a channel is a concurrent layer over a bounded buffer.
To achieve efficient buffering users should aim for as few blocking operations on a channel as possible.
Often to achieve this users may change the buffer size, shard a channel into multiple channels, or tweak the number of producer and consumer threads.
Fo users to be able to make informed decisions when tuning channel usage, toggle-able channel statistics are provided.
The statistics are toggled at compile time via the @CHAN_STATS@ macro to ensure that they are entirely elided when not used.
When statistics are turned on, four counters are maintained per channel, two for producers and two for consumers.
The two counters per type of operation track the number of blocking operations and total operations.
In the channel destructor the counters are printed out aggregated and also per type of operation.
An example use case of the counters follows.
A user is buffering information between producer and consumer threads and wants to analyze channel performance.
Via the statistics they see that producers block for a large percentage of their operations while consumers do not block often.
They then can use this information to adjust their number of producers/consumers or channel size to achieve a larger percentage of non-blocking producer operations, thus increasing their channel throughput.

\subsection{Deadlock Detection}
The deadlock detection in the \CFA channels is fairly basic.
It only detects the case where threads are blocked on the channel during deallocation.
This case is guaranteed to deadlock since the list holding the blocked thread is internal to the channel and will be deallocated.
If a user maintained a separate reference to a thread and unparked it outside the channel they could avoid the deadlock, but would run into other runtime errors since the thread would access channel data after waking that is now deallocated.
More robust deadlock detection surrounding channel usage would have to be implemented separate from the channel implementation since it would require knowledge about the threading system and other channel/thread state.

\subsection{Program Shutdown}
Terminating concurrent programs is often one of the most difficult parts of writing concurrent code, particularly if graceful termination is needed.
The difficulty of graceful termination often arises from the usage of synchronization primitives that need to be handled carefully during shutdown.
It is easy to deadlock during termination if threads are left behind on synchronization primitives.
Additionally, most synchronization primitives are prone to \gls{toctou} issues where there is race between one thread checking the state of a concurrent object and another thread changing the state.
\gls{toctou} issues with synchronization primitives often involve a race between one thread checking the primitive for blocked threads and another thread blocking on it.
Channels are a particularly hard synchronization primitive to terminate since both sending and receiving to/from a channel can block.
Thus, improperly handled \gls{toctou} issues with channels often result in deadlocks as threads trying to perform the termination may end up unexpectedly blocking in their attempt to help other threads exit the system.

\paragraph{Go channels} provide a set of tools to help with concurrent shutdown~\cite{go:chan}.
Channels in Go have a @close@ operation and a \Go{select} statement that both can be used to help threads terminate.
The \Go{select} statement is discussed in \ref{s:waituntil}, where \CFA's @waituntil@ statement is compared with the Go \Go{select} statement.

The @close@ operation on a channel in Go changes the state of the channel.
When a channel is closed, sends to the channel panic along with additional calls to @close@.
Receives are handled differently. 
Receivers (consumers) never block on a closed channel and continue to remove elements from the channel.
Once a channel is empty, receivers can continue to remove elements, but receive the zero-value version of the element type.
To avoid unwanted zero-value elements, Go provides the ability to iterate over a closed channel to remove the remaining elements.
These Go design choices enforce a specific interaction style with channels during termination: careful thought is needed to ensure additional @close@ calls do not occur and no sends occur after a channel is closed.
These design choices fit Go's paradigm of error management, where users are expected to explicitly check for errors, rather than letting errors occur and catching them.
If errors need to occur in Go, return codes are used to pass error information up call levels.
Note, panics in Go can be caught, but it is not the idiomatic way to write Go programs.

While Go's channel closing semantics are powerful enough to perform any concurrent termination needed by a program, their lack of ease of use leaves much to be desired.
Since both closing and sending panic once a channel is closed, a user often has to synchronize the senders (producers) before the channel can be closed to avoid panics.
However, in doing so it renders the @close@ operation nearly useless, as the only utilities it provides are the ability to ensure receivers no longer block on the channel and receive zero-valued elements.
This functionality is only useful if the zero-typed element is recognized as a sentinel value, but if another sentinel value is necessary, then @close@ only provides the non-blocking feature.
To avoid \gls{toctou} issues during shutdown, a busy wait with a \Go{select} statement is often used to add or remove elements from a channel.
Due to Go's asymmetric approach to channel shutdown, separate synchronization between producers and consumers of a channel has to occur during shutdown.

\paragraph{\CFA channels} have access to an extensive exception handling mechanism~\cite{Beach21}.
As such \CFA uses an exception-based approach to channel shutdown that is symmetric for both producers and consumers, and supports graceful shutdown.

Exceptions in \CFA support both termination and resumption.
\Newterm{Termination exception}s perform a dynamic call that unwinds the stack preventing the exception handler from returning to the raise point, such as in \CC, Python and Java.
\Newterm{Resumption exception}s perform a dynamic call that does not unwind the stack allowing the exception handler to return to the raise point.
In \CFA, if a resumption exception is not handled, it is reraised as a termination exception.
This mechanism is used to create a flexible and robust termination system for channels.

When a channel in \CFA is closed, all subsequent calls to the channel raise a resumption exception at the caller.
If the resumption is handled, the caller attempts to complete the channel operation.
However, if channel operation would block, a termination exception is thrown.
If the resumption is not handled, the exception is rethrown as a termination.
These termination exceptions allow for non-local transfer that is used to great effect to eagerly and gracefully shut down a thread.
When a channel is closed, if there are any blocked producers or consumers inside the channel, they are woken up and also have a resumption thrown at them.
The resumption exception, @channel_closed@, has a couple fields to aid in handling the exception.
The exception contains a pointer to the channel it was thrown from, and a pointer to an element.
In exceptions thrown from remove the element pointer will be null.
In the case of insert the element pointer points to the element that the thread attempted to insert.
This element pointer allows the handler to know which operation failed and also allows the element to not be lost on a failed insert since it can be moved elsewhere in the handler.
Furthermore, due to \CFA's powerful exception system, this data can be used to choose handlers based which channel and operation failed.
Exception handlers in \CFA have an optional predicate after the exception type which can be used to optionally trigger or skip handlers based on the content of an exception.
It is worth mentioning that the approach of exceptions for termination may incur a larger performance cost during termination that the approach used in Go.
This should not be an issue, since termination is rarely an fast-path of an application and ensuring that termination can be implemented correctly with ease is the aim of the exception approach.

\section{\CFA / Go channel Examples}
To highlight the differences between \CFA's and Go's close semantics, three examples will be presented.
The first example is a simple shutdown case, where there are producer threads and consumer threads operating on a channel for a fixed duration.
Once the duration ends, producers and consumers terminate without worrying about any leftover values in the channel.
The second example extends the first example by requiring the channel to be empty upon shutdown.
Both the first and second example are shown in Figure~\ref{f:ChannelTermination}.


First the Go solutions to these examples shown in Figure~\ref{l:go_chan_term} are discussed.
Since some of the elements being passed through the channel are zero-valued, closing the channel in Go does not aid in communicating shutdown.
Instead, a different mechanism to communicate with the consumers and producers needs to be used.
This use of an additional flag or communication method is common in Go channel shutdown code, since to avoid panics on a channel, the shutdown of a channel often has to be communicated with threads before it occurs.
In this example, a flag is used to communicate with producers and another flag is used for consumers.
Producers and consumers need separate avenues of communication both so that producers terminate before the channel is closed to avoid panicking, and to avoid the case where all the consumers terminate first, which can result in a deadlock for producers if the channel is full.
The producer flag is set first, then after producers terminate the consumer flag is set and the channel is closed.
In the second example where all values need to be consumed, the main thread iterates over the closed channel to process any remaining values.


In the \CFA solutions in Figure~\ref{l:cfa_chan_term}, shutdown is communicated directly to both producers and consumers via the @close@ call.
In the first example where all values do not need to be consumed, both producers and consumers do not handle the resumption and finish once they receive the termination exception.
The second \CFA example where all values must be consumed highlights how resumption is used with channel shutdown.
The @Producer@ thread-main knows to stop producing when the @insert@ call on a closed channel raises exception @channel_closed@.
The @Consumer@ thread-main knows to stop consuming after all elements of a closed channel are removed and the call to @remove@ would block.
Hence, the consumer knows the moment the channel closes because a resumption exception is raised, caught, and ignored, and then control returns to @remove@ to return another item from the buffer.
Only when the buffer is drained and the call to @remove@ would block, a termination exception is raised to stop consuming.
The \CFA semantics allow users to communicate channel shutdown directly through the channel, without having to share extra state between threads.
Additionally, when the channel needs to be drained, \CFA provides users with easy options for processing the leftover channel values in the main thread or in the consumer threads.
If one wishes to consume the leftover values in the consumer threads in Go, extra synchronization between the main thread and the consumer threads is needed.

\begin{figure}
\centering

\begin{lrbox}{\myboxA}
\begin{cfa}[aboveskip=0pt,belowskip=0pt]
channel( size_t ) Channel{ ChannelSize };

thread Consumer {};
void main( Consumer & this ) {
    try {
        for ( ;; )
            remove( Channel );
    @} catchResume( channel_closed * ) { @
    // handled resume => consume from chan
    } catch( channel_closed * ) {
        // empty or unhandled resume
    } 
}

thread Producer {};
void main( Producer & this ) {
    size_t count = 0;
    try {
        for ( ;; )
            insert( Channel, count++ );
    } catch ( channel_closed * ) {
        // unhandled resume or full
    } 
}

int main( int argc, char * argv[] ) {
    Consumer c[Consumers];
    Producer p[Producers];
    sleep(Duration`s);
    close( Channel );
    return 0;
}
\end{cfa}
\end{lrbox}

\begin{lrbox}{\myboxB}
\begin{cfa}[aboveskip=0pt,belowskip=0pt]
var cons_done, prod_done bool = false, false;
var prodJoin chan int = make(chan int, Producers)
var consJoin chan int = make(chan int, Consumers)

func consumer( channel chan uint64 ) {
    for {
        if cons_done { break }
        <-channel
    }
    consJoin <- 0 // synch with main thd
}

func producer( channel chan uint64 ) {
    var count uint64 = 0
    for {
        if prod_done { break }
        channel <- count++
    }
    prodJoin <- 0 // synch with main thd
}

func main() {
    channel = make(chan uint64, ChannelSize)
    for j := 0; j < Consumers; j++ {
        go consumer( channel )
    }
    for j := 0; j < Producers; j++ {
        go producer( channel )
    }
    time.Sleep(time.Second * Duration)
    prod_done = true
    for j := 0; j < Producers ; j++ {
        <-prodJoin // wait for prods
    }
    cons_done = true
    close(channel) // ensure no cons deadlock
    @for elem := range channel { @
        // process leftover values
    @}@
    for j := 0; j < Consumers; j++{
        <-consJoin // wait for cons
    }
}
\end{cfa}
\end{lrbox}

\subfloat[\CFA style]{\label{l:cfa_chan_term}\usebox\myboxA}
\hspace*{3pt}
\vrule
\hspace*{3pt}
\subfloat[Go style]{\label{l:go_chan_term}\usebox\myboxB}
\caption{Channel Termination Examples 1 and 2. Code specific to example 2 is highlighted.}
\label{f:ChannelTermination}
\end{figure}

The final shutdown example uses channels to implement a barrier.
It is shown in Figure~\ref{f:ChannelBarrierTermination}.
The problem of implementing a barrier is chosen since threads are both producers and consumers on the barrier-internal channels, which removes the ability to easily synchronize producers before consumers during shutdown.
As such, while the shutdown details will be discussed with this problem in mind, they are also applicable to other problems taht have individual threads both producing and consuming from channels.
Both of these examples are implemented using \CFA syntax so that they can be easily compared.
Figure~\ref{l:cfa_chan_bar} uses \CFA-style channel close semantics and Figure~\ref{l:go_chan_bar} uses Go-style close semantics.
In this example it is infeasible to use the Go @close@ call since all threads are both potentially producers and consumers, causing panics on close to be unavoidable without complex synchronization.
As such in Figure~\ref{l:go_chan_bar} to implement a flush routine for the buffer, a sentinel value of @-1@ has to be used to indicate to threads that they need to leave the barrier.
This sentinel value has to be checked at two points.
Furthermore, an additional flag @done@ is needed to communicate to threads once they have left the barrier that they are done.

In the \CFA version~\ref{l:cfa_chan_bar}, the barrier shutdown results in an exception being thrown at threads operating on it, which informs the threads that they must terminate.
This avoids the need to use a separate communication method other than the barrier, and avoids extra conditional checks on the fast path of the barrier implementation.
Also note that in the Go version~\ref{l:go_chan_bar}, the size of the barrier channels has to be larger than in the \CFA version to ensure that the main thread does not block when attempting to clear the barrier.

\begin{figure}
\centering

\begin{lrbox}{\myboxA}
\begin{cfa}[aboveskip=0pt,belowskip=0pt]
struct barrier {
        channel( int ) barWait, entryWait;
        int size;
};
void ?{}( barrier & this, int size ) with(this) {
        barWait{size};   entryWait{size};
        this.size = size;
        for ( i; size )
                insert( entryWait, i );
}
void wait( barrier & this ) with(this) {
        int ticket = remove( entryWait );

        if ( ticket == size - 1 ) {
                for ( i; size - 1 )
                        insert( barWait, i );
                return;
        }
        ticket = remove( barWait );

        if ( size == 1 || ticket == size - 2 ) { // last ?
                for ( i; size )
                        insert( entryWait, i );
        }
}
void flush(barrier & this) with(this) {
        @close( barWait );   close( entryWait );@
}
enum { Threads = 4 };
barrier b{Threads};

thread Thread {};
void main( Thread & this ) {
        @try {@
                for ()
                        wait( b );
        @} catch ( channel_closed * ) {}@
}
int main() {
        Thread t[Threads];
        sleep(10`s);

        flush( b );
} // wait for threads to terminate
\end{cfa}
\end{lrbox}

\begin{lrbox}{\myboxB}
\begin{cfa}[aboveskip=0pt,belowskip=0pt]
struct barrier {
        channel( int ) barWait, entryWait;
        int size;
};
void ?{}( barrier & this, int size ) with(this) {
        barWait{size + 1};   entryWait{size + 1};
        this.size = size;
        for ( i; size )
                insert( entryWait, i );
}
void wait( barrier & this ) with(this) {
        int ticket = remove( entryWait );
        @if ( ticket == -1 ) { insert( entryWait, -1 ); return; }@
        if ( ticket == size - 1 ) {
                for ( i; size - 1 )
                        insert( barWait, i );
                return;
        }
        ticket = remove( barWait );
        @if ( ticket == -1 ) { insert( barWait, -1 ); return; }@
        if ( size == 1 || ticket == size - 2 ) { // last ?
                for ( i; size )
                        insert( entryWait, i );
        }
}
void flush(barrier & this) with(this) {
        @insert( entryWait, -1 );   insert( barWait, -1 );@
}
enum { Threads = 4 };
barrier b{Threads};
@bool done = false;@
thread Thread {};
void main( Thread & this ) {
        for () {
          @if ( done ) break;@
                wait( b );
        }
}
int main() {
        Thread t[Threads];
        sleep(10`s);
        done = true;
        flush( b );
} // wait for threads to terminate
\end{cfa}
\end{lrbox}

\subfloat[\CFA style]{\label{l:cfa_chan_bar}\usebox\myboxA}
\hspace*{3pt}
\vrule
\hspace*{3pt}
\subfloat[Go style]{\label{l:go_chan_bar}\usebox\myboxB}
\caption{Channel Barrier Termination}
\label{f:ChannelBarrierTermination}
\end{figure}

\section{Performance}

Given that the base implementation of the \CFA channels is very similar to the Go implementation, this section aims to show the performance of the two implementations are comparable.
The microbenchmark for the channel comparison is similar to Figure~\ref{f:ChannelTermination}, where the number of threads and processors is set from the command line.
The processors are divided equally between producers and consumers, with one producer or consumer owning each core.
The number of cores is varied to measure how throughput scales.

The results of the benchmark are shown in Figure~\ref{f:chanPerf}.
The performance of Go and \CFA channels on this microbenchmark is comparable.
Note, the performance should decline as the number of cores increases as the channel operations occur in a critical section, so increasing cores results in higher contention with no increase in parallelism.

\begin{figure}
        \centering
        \subfloat[AMD \CFA Channel Benchmark]{
                \resizebox{0.5\textwidth}{!}{\input{figures/nasus_Channel_Contention.pgf}}
                \label{f:chanAMD}
        }
        \subfloat[Intel \CFA Channel Benchmark]{
                \resizebox{0.5\textwidth}{!}{\input{figures/pyke_Channel_Contention.pgf}}
                \label{f:chanIntel}
        }
        \caption{The channel contention benchmark comparing \CFA and Go channel throughput (higher is better).}
        \label{f:chanPerf}
\end{figure}

% Local Variables: %
% tab-width: 4 %
% End: %