pserver etcd client #2559

jacquesqiao · 2017-06-22T04:23:29Z

helinwang · 2017-06-22T05:01:52Z

go/pserver/cclient/cclient.go

+	if etcd_addr == "" {
+		return C.PSERVER_ERROR
+	}
+	etcd_addrs := strings.Split(etcd_addr, ",")


Sorry I should have mentioned this earlier, the cgo code are supposed to be just a thin layer. Because it's just a interface from C to Go. If it contains too much logic, then when we want to use Pserver from Go to Go, we need to duplicate the logic. Another reason is Go's test does not work well with cgo, it's supposed to be tested from C side. It's much easier to write test for Go.

You can create a new file in pserver package. E.g., etcd.go, and implement this interface: https://github.com/PaddlePaddle/Paddle/blob/develop/go/pserver/client.go#L26
And this interface:
https://github.com/PaddlePaddle/Paddle/blob/develop/go/pserver/client.go#L15

So cgo just need to do:

e := pserver.NewEtcd(addr) // we can call it NewEtcd, or some other name. c := pserver.NewClient(e, len(as), e) return add(c)

Thank you for the comment! I will add a separate module that implement these interfaces.

typhoonzero · 2017-06-22T04:51:28Z

go/pserver/cclient/cclient.go

+		return err
+	}
+	retryTimes := defaultRetryTimes
+	for retryTimes < 0 {


Do we need to try infinitely because we don't know how long it will take til the etcd key is written.

Ok, I will optimize this strategy

typhoonzero · 2017-06-22T05:10:03Z

go/pserver/cclient/cclient.go

+			time.Sleep(time.Second)
+			continue
+		}
+		ps_addr := string(kvs[0].Value)


From design doc: https://github.com/PaddlePaddle/Paddle/tree/develop/doc/design/cluster_train#trainer-process-1

ps_addr is not fetched from one etcd key. Addresses are read from /ps/<id> which stands for IP address for each pserver, so ps_addr need to concat all these addresses.

And each trainer must "watch" pserver nodes hand handle events of pserver creation and loss.

Done the first part, but I have not implemented the watcher because now the pserver client cannot change to new pserver when running, I have added a TODO and will implment this function in the following pr. Thank you!

Yancey1989 · 2017-06-22T15:14:52Z

go/pserver/cclient/cclient.go

+	})
+	if err != nil {
+		log.Errorln(err)
+		return err


The return type is C.paddle_pserver_client, I think the function would panic if client connect etcd failed.

… cclient-etcd

typhoonzero · 2017-06-26T03:22:28Z

go/pserver/etcd_lister.go

+}
+
+// read ps desired number from etcd.
+func(p pserverEtcdLister) desired() int {


p pserverEtcdLister -> p *pserverEtcdLister same as func(p pserverEtcdLister) List() []Server

typhoonzero · 2017-06-26T03:24:28Z

go/pserver/etcd_lister.go

+}
+
+// read ps desired number from etcd.
+func(p pserverEtcdLister) desired() int {


Can put https://github.com/PaddlePaddle/Paddle/blob/develop/go/pserver/service.go#L106 as a function and call it directly.

typhoonzero · 2017-06-26T03:27:03Z

go/pserver/service.go

@@ -21,6 +21,10 @@ type ElementType int
 const (
 	AlreadyInitialized = "pserver already initialized"
 	Uninitialized      = "pserver not fully initialized"
+	// PsDesired is etcd path for store desired pserver count
+	DefaultPsDesiredPath = "/ps_desired"


Just a note: etcdv3 uses only key-values, thought the key looks like a path.

typhoonzero · 2017-06-26T03:30:02Z

go/pserver/etcd_lister.go

+	}
+}
+
+func(p pserverEtcdLister) List() []Server {


In #2551 , trainers need to discover pservers and keep watch pservers using etcd's watch. I think you need to discuss with @Yancey1989 to see where wo implement the "etcd watcher"

Thanks for reminding! I will discuss with your today~~

helinwang · 2017-06-26T23:40:52Z

go/master/etcd_client.go

@@ -98,7 +98,7 @@ func (e *EtcdClient) Save(state []byte) error {
 			// We lost the master lock and can not acquire
 			// it back, it means some other master is
 			// already started. We don't want cluster
-			// managment system to kill the master server
+			// management system to kill the master server


I have so many typos :p. Thanks!

helinwang · 2017-06-26T23:45:37Z

go/pserver/cclient/cclient.go

-func paddle_new_etcd_pserver_client(etcd_addr *C.char) C.paddle_pserver_client {
-	// TODO(helin): fault tolerant pserver client using etcd.
-	panic("not implemented.")
+func paddle_new_etcd_pserver_client(etcd_addr *C.char, selected int) C.paddle_pserver_client {


We need to use etcd lock to decide which trainer is selected to initialize the parameter servers. This probably should done in the pserver package (not here in cgo) by implementing the pserver.Selector interface. I think in etcd_client.go mentioned in #2559 (comment) could be a good place to hold such code.

Don't need to do in the PR, but can you add a TODO, and it would be awesome if you can send a follow up PR for it?

Please see more detail here: https://github.com/PaddlePaddle/Paddle/blob/develop/doc/design/cluster_train/pserver_client.md#trainer-selection

I add a TODO to optimize this part and will learn how to use etcd lock to sync status.

helinwang · 2017-06-26T23:54:58Z

go/pserver/client_test.go

+	ClientTest(t, c1)
+}
+
+func TestEtcdClient(t *testing.T) {


TestEtcdClient is not a self-contained test, it requires an etcd server to be running when the test is running. So the CI (after we setup cd go && go test ./...) will not run correctly if etcd is not running on 127.0.0.1:2379.

Maybe for know let's disable the test be renaming TestEtcdClient to EtcdClient, and user can enable manually when necessary.

We can explore embedding etcd later: #2504

helinwang · 2017-06-27T00:01:30Z

go/pserver/etcd_lister.go

+)
+
+const (
+	DefaultEtcdTimeout time.Duration = time.Second * time.Duration(5)


5 * time.Second is fine. Constants in Go is very well designed: https://blog.golang.org/constants

helinwang · 2017-06-27T00:20:00Z

go/pserver/etcd_lister.go

+	servers := make([]Server, psDesired)
+	for {
+		ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
+		for i := 0; i < psDesired; i++ {


This can be batched into a single transaction, here is an related example: https://github.com/PaddlePaddle/Paddle/blob/develop/go/master/etcd_client.go#L121
Something like:

resp, err := e.client.Txn(ctx).If().Then(get0, get1, get2, get3).Commit()

Feel free to improve it later, not a merge blocker.

helinwang · 2017-06-27T00:27:08Z

go/pserver/etcd_lister.go

+			// TODO(Longfei) check the ps address
+			if  psAddr == "" {
+				cancel()
+				log.Infof("Get psKey = %s,  psAddr is null illegal", psKey, psAddr)


Only one "%s", but two variable. go vet is a good tool for detecting this kind of error.

ok, done, thanx~

helinwang · 2017-06-27T00:31:01Z

go/pserver/etcd_lister.go

+			kvs := resp.Kvs
+			if len(kvs) == 0 {
+				log.Infof("Waiting for ps addr registered ...")
+				time.Sleep(p.timeout)


cancel() need to be called.

helinwang · 2017-06-27T00:35:04Z

go/pserver/etcd_lister.go

+	DefaultEtcdTimeout time.Duration = time.Second * time.Duration(5)
+)
+
+type pserverEtcdLister struct {


The name could be more concise if it's something like etcdLister, the package is already pserver, people read it in their minds as: pserver etcd lister. If it's named as pserverEtcdLister, it's long, and does not read good: pserver pserver etcd lister.

This type does more than just Lister (tells desired number of pservers as well), maybe it's best to name it as EtcdClient (if rename, please change file name as well.)

done, thanks for this very valuable suggestion!

helinwang · 2017-06-27T00:41:45Z

go/pserver/etcd_lister.go

+	return servers
+}
+
+func NewEtcdAddrLister(endpoints string) (Lister, int) {


It's less abstraction (interface is an abstraction) if the return type is not an interface:PserverEtcdLister (I think it's a good idea to make pserverEtcdLister public, perhaps rename to EtcdClient, and desired public), than interface Lister.

The benefit of no abstraction unless necessary is not very obvious in the beginning, maybe the article below could explain more. I will give a not-so-good-example: At least it's easier to jump to the definition when not returning an interface (editor jumps to the function definition rather than interface definition).

A good general rule in Go would be: accept interfaces, return structs

… cclient-etcd

typhoonzero · 2017-06-30T12:32:42Z

go/pserver/cclient/cclient.go

-	// TODO(helin): fault tolerant pserver client using etcd.
-	panic("not implemented.")
+func paddle_new_etcd_pserver_client(etcd_addr *C.char, selected int) C.paddle_pserver_client {
+	addr := C.GoString(etcd_addr)


etcd_addr => etcd_endpoints. We always use etcd_endpoints to represent comma separated etcd addresses.

helinwang · 2017-07-01T00:41:50Z

go/pserver/client_test.go

+}
+
+//TODO(Qiao: tmperary disable etcdClient test for dependency of etcd)
+func EtcdClient(t *testing.T) {


Please make sure to disable it before merging into develop branch. We need to make sure test always runs correctly, since other developers need to run test as well.

helinwang · 2017-07-01T00:42:41Z

go/pserver/etcd_cclient.go

+)
+
+const (
+	DefaultEtcdTimeout time.Duration = 5 * time.Second


Just FYI DefaultEtcdTimeout = 5 * time.Second could work as well.

helinwang · 2017-07-01T00:45:10Z

go/pserver/etcd_cclient.go

+	DefaultEtcdTimeout time.Duration = 5 * time.Second
+)
+
+type EtcdCClient interface {


I know it's hard to choose a name, since EtcdClient is already used ;)
However, EtcdCClient may not be a good name, because C Client means it's for C. But here is not really related to C (Go can use it as well). Maybe ClientEtcd?

helinwang · 2017-07-01T00:48:29Z

go/pserver/etcd_cclient.go

+// TODO(Longfei)
+// 1. add watcher to watch the change state of pservers)
+// 1. add etcd lock)
+type EtcdCClientImpl struct {


It's a very common pattern in Java: create interface, and write the implementation, this is due to Java has explicit interface. It's not required for Go: Go has implicit interface. Please see: https://medium.com/@cep21/preemptive-interface-anti-pattern-in-go-54c18ac0668a Unnecessary interface only adds complication to the program.

I think we need to stick to "accept interface, return struct" unless there is good reason not to.
In this case we need to remove type EtcdCClient interface, rename EtcdCClientImpl to EtcdCClient, and make func NewEtcdCClient(endpoints string) (*EtcdCClient, error) return pointer to the struct.

helinwang · 2017-07-01T00:57:39Z

go/pserver/etcd_cclient.go

+		for i := 0; i < psDesired; i++ {
+			psKey := PsPath + strconv.Itoa(i)
+			log.Debugf("checking %s", psKey)
+			resp, err := p.client.Get(ctx, psKey)


是不是改成

ctx, cancel := context.WithTimeout(context.Background(), p.timeout) resp, err := p.client.Get(ctx, psKey) cancel() if err != nil { ...

更清楚些？

这个地方还不能这么改，因为是在一个循环里边，只能最后cancel。否则下次循环到这里就失败了

@jacquesqiao 下一个循环貌似又创建了一个新的ctx?

for ... { ctx, cancel := context.WithTimeout(context.Background(), p.timeout) resp, err := p.client.Get(ctx, psKey) cancel() }

… cclient-etcd

typhoonzero · 2017-07-03T07:39:02Z

go/pserver/etcd_client.go

@@ -64,11 +71,11 @@ func (e *EtcdClient) Register() (int, error) {
 		log.Debugf("inited client to %s", e.etcdEndpoints)
 		break
 	}
-	// init /ps_desired using transaction, for multiple pservers may want to write
+	// init /ps/desired using transaction, for multiple pservers may want to write


We use ps_desired in design doc: https://github.com/PaddlePaddle/Paddle/tree/develop/doc/design/cluster_train#parameter-server-process-1

Comment is not updated

typhoonzero · 2017-07-03T07:40:50Z