Design Doc: CSP in PaddlePaddle Fluid¶

Motivation¶

Concurrent programming is important for deep learning. Few example applications are:

The main thread keeps reading the next mini-batch while another thread uses the GPU for computing.
The main thread performs the computation while another thread uploads the local gradients from each trainer to the parameter server.

Most DL systems, including TensorFlow, Caffe2, and MxNet, can asynchronously execute operators in a graph. However, Fluid doesn’t have the concept of a graph at all, as the design goal of Fluid is that of a programming language.

Concurrent Programming Models¶

There were many concurrent programming models, implemented in various forms:

Since Fluid was designed to be a programming language, we would like to implement CSP in Fluid.

CSP v.s. Actor Model¶

A well-known implementation of Actor Model is the Erlang programming language. In Actor Model, processes could send messages to another process and receive messages from another process given the process IDs. We can find the three ingredients, process with ID, send, and recv, in MPI too. Indeed, we can rewrite Erlang programs in Python + MPI with possibly fewer lines of code. Our concern with Actor Model is that it doesn’t seem reasonable to implement process management in a programming language’s runtime library; instead, it should be the operating systems’ responsibility to manage processes and libraries like MPI for send/recv.

CSP in Fluid¶

Fluid has two fundamental control-flows: if-else and while. If we are to implement CSP, we need the following:

a new data type: channel and operators send and recv,
goroutine or thread, and
a new control-flow: select.

We also need Python wrappers for the above components.

The type channel is conceptually the blocking queue. In Go, its implemented is a blocking circular queue, which supports send and recv.

The select operation has been in OS kernels long before Go language. All Unix kernels implement system calls poll and select. They monitor multiple file descriptors to see if I/O is possible on any of them. This takes O(N) time. Since Linux 2.6, a new system call, epoll, can do the same in O(1) time. In BSD systems, there is a similar system call kqueue. Go’s Linux implementation uses epoll.

It might be a good idea to implement Fluid’s select using epoll too. In this design doc, we start from the O(N) way so that we could focus on Python binding and the syntax.

Type Channel¶

Fluid supports many data types:

Tensor,
Row-sparse Tensor
LoD Tensor,
Tensor array, etc

Each data type is registered in the framework.proto as an enum value. To add a new type channel, we need to add a new type enum.

To expose a C++ type to Python, we need to edit the pybind.cc file. Here is an example how we expose C++ class LoDTensor.

Syntax Design¶

Create Channel¶

In Go, we create a channel by specifying the element type and buffer size:

ch  := make(chan int)       // a channel without buffer
ch1 := make(chan int, 100)  // a channel that can buffer 100 ints.

In Fluid, we should be able to do the same:

ch  = fluid.make_channel(dtype=INT)
ch1 = fluid.make_channel(dtype=INT, 100)

In addition to that, we want channels that can hold more complex element types, e.g., Tensors of float16:

ch = fluid.make_channel(dtype=Tensor, etype=float16)

or Tensors of Tensors of float16 etc.

The point here is that we need a consistent way to compose types, like in C++ we can have Tensor<Tensor<...<float16>...> >.

Send and Recv¶

Go’s CSP implementation depends on data type channel. There are two types of channels:

The unblocked channel, or buffered channel, is a blocking queue with a non-zero sized buffer. The sending to buffered channel blocks if the buffer is full, and the receive operation blocks if the buffer is empty.
blocked channel, or unbuffered channel, is a blocking queue with no buffer. Both sending and receiving block with unbuffered channels.

There are four types of actions with a channel:

Create a channel

ch := make(chan int) // this is an unbuffered channel
ch := make(chan int, 100) // this is a buffered channel of 100 ints.

Send
```
ch <- 111
```
Recv
```
y, ok <- ch
```
Close
```
close(ch)
```
Please be aware that a closed channel is not a nil channel, which is var ch chan int.

There are some axioms with channels:

A send to a nil channel blocks forever
A receive from a nil channel blocks forever
A send to a closed channel panics
A receive from a closed channel returns the residual values and then zeros.

In Fluid, we have buffered channels and unbuffered channels

The following program illustrates the Python syntax for accessing Fluid buffers.

import fluid

buffer_size = 10
ch = fluid.make_channel(dtype=INT, buffer_size)

# Now write three elements to the channel
with fluid.while(steps=buffer_size):
  fluid.send(ch, step)
  fluid.close_channel(ch)
  
with fluid.while(steps=buffer_size):
  fluid.print(fluid.recv(ch))

The following example shows that to avoid the always-blocking behavior of unbuffered channels, we need to use Fluid’s goroutines.

import fluid

ch = fluid.make_channel(dtype=INT)

with fluid.go():
  fluid.send(ch)

y = fluid.recv(ch)

fluid.close_channel(ch)

Select¶

In Go, the select statement lets a goroutine wait on multiple communication operations. A select blocks until one of its cases can run, then it executes that case. It chooses one at random if multiple are ready.

ch1  := make(chan int)       
ch2  := make(chan int, 100)

x := 0

for {
    select {
    case ch1 <- x:
      x := x + 1
    case y <- ch2:
      fmt.Println("Received on channel")
    default:
      fmt.Println("Default")
    }
  }