Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Rewrite the Send/Recv part of csp.md #8164

Merged
merged 3 commits into from
Feb 6, 2018
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
110 changes: 59 additions & 51 deletions doc/design/csp.md
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ The type *channel* is conceptually the blocking queue. In Go, its implemented i

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 we could focus on Python binding and the syntax.
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

Expand Down Expand Up @@ -87,79 +87,87 @@ The point here is that we need a consistent way to compose types, like in C++ we

### Send and Recv

In Go, we first create a channel as explained in the section above and then perform read and write operations on top of the channels.
Go's CSP implementation depends on data type *channel*. There are two types of channels:

```go
ch1 := make(chan int)
ch2 := make(chan int, 100)
```
1. 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.
1. blocked channel, or unbuffered channel, is a blocking queue with no buffer. Both sending and receiving block with unbuffered channels.

To write (or perform a `Send` operation) the value of a variable `x`, to channel `ch1` above, we perform the following:
There are four types of actions with a channel:

```go
ch1 <- x
fmt.Println("Written to the channel")
```
Now to read (or perform a `Recv` operation) the value stored in `ch2` into a variable `y`, we perform the following:
1. Create a channel

```go
y <- ch2
fmt.Println("Received from channel")
```
```go
ch := make(chan int) // this is an unbuffered channel
ch := make(chan int, 100) // this is a buffered channel of 100 ints.
```

In Fluid, we should be able to perform the above operations on the channel objects as well. As of now, we support two different kinds of channels : [Buffered Channel](https://github.com/PaddlePaddle/Paddle/blob/develop/paddle/framework/details/buffered_channel.h) and [UnBuffered Channel](https://github.com/PaddlePaddle/Paddle/blob/develop/paddle/framework/details/unbuffered_channel.h)
1. Send

Send and Receive can be performed as following on a buffered channel:
```go
ch <- 111
```

```python
import threading
1. Recv

def send_to_channel(channel, num_time=1):
for i in xrange(num_time):
channel.send(i)
```go
y, ok <- ch
```

# Create a buffered channel of capacity 10
buffer_size = 10;
ch = fluid.make_channel(dtype=INT, buffer_size)
1. Close

# Now write three elements to the channel
thread = threading.Thread(target=send_to_channel, args=(ch, 3, ))
thread.daemon = True
thread.start()
```go
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](https://dave.cheney.net/2014/03/19/channel-axioms):

# Read all the data from the channel
for i in xrange(3):
y = ch.recv()
1. A send to a nil channel blocks forever

# Done receiving , now close the channel
ch.close()
```
1. A receive from a nil channel blocks forever

1. A send to a closed channel panics

1. A receive from a closed channel returns the residual values and then zeros.

The send and receive operations will be similar for unbuffered channel as well, except for the fact that there is no buffer in an unbuffered channel, so the operations are completely synchronized. For example:
In Fluid, we have [buffered channels](https://github.com/PaddlePaddle/Paddle/blob/develop/paddle/framework/details/buffered_channel.h) and [unbuffered channels](https://github.com/PaddlePaddle/Paddle/blob/develop/paddle/framework/details/unbuffered_channel.h)

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

```python
import threading
import fluid

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

def send_to_channel(channel, data):
channel.send(data)
# 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.

```python
import fluid

# Create an unbuffered channel
ch = fluid.make_channel(dtype=INT)

# Writes and Reads are synchronous otherwise the calls will block.
thread = threading.Thread(target=send_to_channel, args=(ch, 10, ))
thread.daemon = True
thread.start()
with fluid.go():
fluid.send(ch)

y = ch.recv()
y = fluid.recv(ch)

# Done receiving , now close the channel
ch.close()
fluid.close_channel(ch)
```

### Select

In Go, the `select` statement lets a goroutine wait on multiple communication operations. A `select` blocks untill one of its cases can run, then it executes that case. It chooses one at random if multiple are ready.
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.

```go

Expand Down Expand Up @@ -202,9 +210,9 @@ with sel.default():

In the above code snippet, `X` and `Y` are variables. Now let us look at each of these statements one by one.

- `sel.case(ch1, 'w', X)` : This specifies that we are writing to `ch1` and we want to write the integer in variable `X` to the channel. The character `w` is used here to make the syntax familar to write syntax in Python I/O.
- `sel.case(ch1, 'w', X)` : This specifies that we are writing to `ch1` and we want to write the integer in variable `X` to the channel. The character `w` is used here to make the syntax familiar to write syntax in Python I/O.

- `sel.case(ch2, 'r', Y)` : This specifies that we would like to read the result from `ch2` into variable `Y`. The character `r` is used here to make the syntax familar to read syntax in Python I/O.
- `sel.case(ch2, 'r', Y)` : This specifies that we would like to read the result from `ch2` into variable `Y`. The character `r` is used here to make the syntax familiar to read syntax in Python I/O.

- `sel.default()` : This is equivalent to the default in Go `select`. If none of the channels are ready for read or write, then the fluid code in the default block will be executed.

Expand Down