Implement fluid API using python with guard.

[typhoonzero]

• change current implement to listen_and_serv, send, recv op implementation.
• implement python API with guard for listen_and_serv
• build sample program using python with statement APIs.
• update document about this.

According to https://github.com/PaddlePaddle/Paddle/blob/develop/doc/design/concurrent_programming.md#the-worker-program we need to implement similar API looks like:

Server side:

loss = define_model()
server = fluid.listen_and_serv()
with server.do():
    opt = fluid.optimizer.Adam()
    opt.minimize(loss)

Worker side:

loss = define_model()
params, grads = fluid.append_backward(loss)
splited = layers.split(params, grads)
with fluid.parallel_for(len(splited)) as iter:
    layers.send(splited["grad"][iter.idx])
with fluid.parallel_for(len(splited)) as iter:
    layers.recv(splited["param"][iter.idx])
layers.concat(splited["param"])

If we are using CSP model, the server side may look like:

loss = define_model()
params, grads = fluid.append_backward(loss)
param_ch = fluid.make_chan()
param_recved_ch = fluid.make_chan()
grad_ch = fluid.make_chan()
layers.split_to_chan(params, param_ch)
layers.split_to_chan(grads, grad_ch)

with fluid.go():
    layers.send(grad_ch)
with fluid.go():
    updated_param = layers.recv(param_ch)
    param_recved_ch.push(updated_param)
layers.concat(param_recved_ch)

[Yancey1989]

For the trainer side, I think the order of send/recv is:

with fluid.parallel_for(len(splited)) as iter:
    layers.send(splited["grad"][iter.idx])
with fluid.parallel_for(len(splited)) as iter:
    layers.recv(splited["param"][iter.idx])

We need to execute Recv after all variables are sent.

And on the other hand, I saw #7706 also list the send/recv Op, shall we execute the Send/Recv Op in a goroutine?

[helinwang]

"If we are using CSP model, the server side may look like:", do you mean the worker side?

I think layers.send(grad_ch) should be something like: layers.send(grad_ch.recv()), in this way the send is still sending a variable, not a channel.

layers.recv(param_ch) could be param_ch.send(layers.recv())

[helinwang]

Btw, is the Python code for illustration only? I don't think we should expose send/recv OP to the user.

[typhoonzero]

I think layers.send(grad_ch) should be something like: layers.send(grad_ch.recv()), in this way the send is still sending a variable, not a channel.

Good point, thank you.

Btw, is the Python code for illustration only? I don't think we should expose send/recv OP to the user.

I think we should expose send/recv/listen_and_serv as layers to users, so that fluid can be a "real" programming language.

For example, @dzhwinter talked yestoday that we may need a single server to merge all the trainers evaluation, this could be done by using these ops as layers.

[helinwang]

I see, thanks. Does the user really care to merge all the trainers evaluation (maybe trainer-id==0's local evaluation is suffice), and write the Python code manually? I am a little worried about the Python binding part becomes something that no one actually uses, but just my 2 cents.

[typhoonzero]

People may need to define their own distributed network, like in #7671. And, yes, we'd like users to use transpilers for simplicity in most cases. If we have those layers, we can also use them in the transpiler to simplify the transpiler's implementation.