Trainer library discover master by etcd #2551

Yancey1989 · 2017-06-21T16:03:41Z

helinwang

Great PR! Could you move the code to https://github.com/PaddlePaddle/Paddle/blob/develop/go/master/c/client.go#L58 ? We already have function paddle_new_master_client which takes master address rather than etcd address, we can just add a new function paddle_new_etcd_master_client. Sorry that I have not mentioned to you earlier!

typhoonzero

Related to #2559?

I think this PR will implement in the master client with paddle_new_etcd_master_client

--Yancey1989

Yancey1989 · 2017-06-22T15:05:40Z

Updated with @helinwang 's comment.

helinwang · 2017-06-22T22:49:56Z

go/master/client.go

+
+		mAddr := kvs[0].Value
+		log.Debugf("Fetched master address: %s\n", mAddr)
+		return NewClient(MasterAddresser(mAddr), bufSize)


Could you take a look at this piece of code: https://github.com/PaddlePaddle/Paddle/blob/develop/go/master/client.go#L80

The idea is the client takes a Addresser and query it to know if it have changed.
This current implementation is create a "constant" Addresser that returns the same address every time. The thing is master server could be killed and restarts with a different address.

So I think a better approach could be this etcd query function be the implementation of the Addresser interface. So the master server will know the updated address when calling Address.

Another detail is when implementing the Addresser, it don't need to create a client in every call to Address, it should reuse the same client.

Done, move GET operator to Address method.

typhoonzero · 2017-06-23T08:38:34Z

go/master/client.go

@@ -20,6 +25,51 @@ type Client struct {
 	ch   chan []byte
 }

+// MasterAddresser provide master address


Comment function name must be the same to function name masterAddresser

Done.
Delete MasterAddresser.

typhoonzero · 2017-06-23T09:15:28Z

go/master/client.go

+				time.Sleep(m.timeout)
+				continue
+			}
+			// reconnect to etcd server


It seem that etcd have a some kind of retry method for grpc but don't know how to use yet, may be help here https://github.com/coreos/etcd/blob/master/clientv3/retry.go#L28

typhoonzero · 2017-06-23T09:18:07Z

go/master/client.go

+func (m masterAddresser) Address() string {
+	for {
+		ctx, cancel := context.WithTimeout(context.Background(), m.timeout)
+		resp, err := m.client.Get(ctx, masterAddrPath)


Clients should use watch to retrieve notification about master node events. masterMonitor should also change to handle these events.

Clients should use watch to retrieve notification about master node events
I think it's a good idea, it will reduce the count for GET operator to etcd.

masterMonitor should also change to handle these events

I do not think masterMonitor will handle the events, we can handle these in a new method named masterAddresser.init(), and masterAddresser.Addresser() only return the latest master address.

I originally thought client can poll etcd to get the latest master address. But I think @typhoonzero 's idea is better.
Because Addresser is an polling interface, to get notification about the change rather than polling to get the change. We need to change the interface. One possible solution could be:
Change:

func NewClient(addr Addresser, bufSize int)

To:

func NewClient(addrCh <-chan string, bufSize int)

And maybe change moniterMaster to:

func (c *Client) monitorMaster(addrCh <-chan string) { lastMaster := "" for curMaster := range addrCh { // connect to the new address once address changed. if curMaster != lastMaster { if curMaster == "" { err := c.conn.Close() if err != nil { log.Errorln(err) } } else { err := c.conn.Connect(curMaster) if err != nil { log.Errorln(err) // connect to addr failed, set // to last known addr in order // to retry next time. curMaster = lastMaster } } } lastMaster = curMaster } }

Done.
I think we also need an interface with etcd operator, so that we can do some unit test easier, I'll add this feature in the next PR.

@Yancey1989 Thanks!
We can use interface to make unit test easier. For example, the master server is taking an interface Store as the parameter: https://github.com/PaddlePaddle/Paddle/blob/develop/go/master/service.go#L94.
The definition of store is:

type Store interface { Save([]byte) error Load() ([]byte, error) }

We have an implementation of Store using etcd. But to test the master server, it's easy, we don't to run a etcd cluster. We can just write a stub interface for Store.
However this only tests master service. Does not test the etcd Store implementation. But that's a good start.

jacquesqiao · 2017-06-26T23:55:41Z

go/.gitignore

@@ -0,0 +1,4 @@
+CMakeCache.txt


do we still need this files? Since the build result is now moved to build dir

Thanks, done.

jacquesqiao · 2017-06-27T00:11:10Z

go/master/client.go


 // Client is the client of the master server.
 type Client struct {
 	conn *connection.Conn
 	ch   chan []byte
 }

+// EtcdClient is the client of


complete this comment

Done.
Move to etcd_client.go

helinwang · 2017-06-27T01:27:17Z

go/master/c/client.go

+//export paddle_new_etcd_master_client
+func paddle_new_etcd_master_client(etcdEndpoints *C.char, bufSize int) C.paddle_master_client {
+	p := C.GoString(etcdEndpoints)
+	c := master.NewEtcdClient(addresser(p), bufSize)


Maybe just c := master.NewEtcdClient(p), bufSize) (remove addresser)

helinwang · 2017-06-27T01:28:04Z

go/master/client.go

 		if curMaster != lastMaster {
 			if curMaster == "" {
 				err := c.conn.Close()
+				fmt.Printf("close conn error: %s", err)


Please use log.Errorf instead of fmt.Printf.

helinwang · 2017-06-27T01:32:25Z

go/master/client.go

+	})
+	if err != nil {
+		log.Errorf("Init etcd connection failed: %v", err)
+		panic(err)


Please return error here instead of panic, and don't need to log error. Caller will get the error and log it.

helinwang · 2017-06-27T01:32:53Z

go/master/client.go

+	}
+	etcdClient.ch = make(chan string)
+	c := NewClient(etcdClient.ch, bufSize)
+	//go etcdClient.monitorMasterAddr()


No commented out code please.

helinwang · 2017-06-27T01:33:47Z

go/master/client.go


 // Client is the client of the master server.
 type Client struct {
 	conn *connection.Conn
 	ch   chan []byte
 }

+// EtcdClient is the client of
+type EtcdClient struct {


Perhaps put EtcdClient into etcd_client.go.

helinwang · 2017-06-27T01:42:35Z

go/master/client.go

+		e.ch <- mAddr
+		break
+	}
+	fmt.Println("init master addr finished.")


Please use log rather than fmt.

helinwang · 2017-06-27T01:44:15Z

go/master/client.go

+	fmt.Println("init master addr finished.")
+}
+func (e *EtcdClient) monitorMasterAddr() {
+	rch := e.client.Watch(context.Background(), masterAddrPath)


第一次调用watch的时候能返回现有的值吗，如果可以的话，就不需要initMasterAddr这个函数了。

看起来是不行的，还是得先Get再watch。
initMasterAddr 改为了 Get这样更通用一些。

helinwang · 2017-06-27T01:47:57Z

go/master/c/client.go

+func paddle_new_etcd_master_client(etcdEndpoints *C.char, bufSize int) C.paddle_master_client {
+	p := C.GoString(etcdEndpoints)
+	c := master.NewEtcdClient(addresser(p), bufSize)
+	return add(c)


感觉把NewEtcdClient里面的c := NewClient(etcdClient.ch, bufSize)放在这里比较好。这样master.EtcdClient就和master.Client不耦合了。EtcdClient负责和etcd交互，Client只读chan string里的新地址。这样unit test也比较好写。

Done.
直接用了etcd_clint.go中的EtcdClient,因为想要复用NewEtcdClient所以将原有初始化时写入master地址的逻辑独立到一个Put函数中。

…_client

helinwang · 2017-06-27T21:10:14Z

go/master/c/client.go

 }

 //export paddle_new_master_client
 func paddle_new_master_client(addr *C.char, bufSize int) C.paddle_master_client {
 	a := C.GoString(addr)
-	c := master.NewClient(addresser(a), bufSize)
+	ch := make(chan string)


Better use:

ch := make(chan string, 1) ch <- a c := master.NewClient(ch, bufSize)

Otherwise this function will not return unless master.Client read the value. We don't want to add unnecessary requirement.

Good comment, thanks @helinwang !!
Done.

helinwang · 2017-06-27T21:45:22Z

go/master/client_internal_test.go

@@ -82,9 +75,11 @@ func TestGetFinishTask(t *testing.T) {
 	// Manually intialize client to avoid calling c.getRecords()
 	c := &Client{}
 	c.conn = connection.New()
-	go c.monitorMaster(TestAddresser(fmt.Sprintf(":%d", p)))
+	addr := fmt.Sprintf(":%d", p)
+	ch := make(chan string)


Maybe ch := make(chan string, 1) is better, so ch <- addr does not need to wait for channel to be read.

helinwang · 2017-06-27T21:54:40Z

go/master/client_test.go

@@ -63,10 +74,10 @@ func TestNextRecord(t *testing.T) {
 	}
 	w.Close()
 	f.Close()
-
-	c := master.NewClient(master.TestAddresser(fmt.Sprintf(":%d", p)), 10)
+	curAddr := make(chan string)


Maybe curAddr := make(chan string, 1)

helinwang · 2017-06-27T22:02:46Z

go/master/client.go

 	go c.getRecords()
 	return c
 }

+// NewEtcdMasterClient creates a new Client by etcd
+func NewEtcdMasterClient(db DatabaseOperator, bufSize int) *Client {


db is only used here to create ch, which is used as input for NewClient. this is a sign that maybe we don't need the function NewEtcdMasterClient.
I think we need to follow rule: less code is better.
Here is a possible way to do it:

func paddle_new_master_client(...) { p := C.GoString(etcdEndpoints) e, err := master.NewEtcdClientWithoutLock(strings.Split(p, ",")) if err != nil { panic(err) } ch := make(chan string, 1) ch <- string(db.WaitMasterReady(DefaultAddrPath, 3)) go db.WatchWithKey(DefaultAddrPath, ch) c := master.NewClient(ch, bufSize) return add(c) }

Done. Delete function NewEtcdMasterClient

helinwang · 2017-06-27T22:07:33Z

go/master/etcd_client.go

+
+// EtcdClientWithoutLock is the etcd client the master users
+// for service discovery
+type EtcdClientWithoutLock struct {


EtcdClientWithoutLock is a very confusing name, it's hard to figure out what does "WithoutLock" means.
Considering the two function are only few lines, maybe just remove this type, and add two helper functions in go/master/etcd_client.go:

func GetKey(c *clientv3.Client, key string) string { // ... } func WatchKey(c *clientv3.Client, key string, ch chan<- string) { // ... }

So paddle_new_etcd_master_client could be changed to:

func paddle_new_etcd_master_client(...) { p := C.GoString(etcdEndpoints) cli, err := clientv3.New(...) if err != nil { panic(err) } ch := make(chan string, 1) ch <- master.GetKey(cli, DefaultAddrPath) go master.WatchKey(cli, DefaultAddrPath, ch) c := master.NewClient(ch, bufSize) return add(c) }

Just an idea, feel free to modify!

Done.
WaitMasterReady => GetKey, and I also think this function does not need to be blocked, if the master address is empty, the function monitorMaster will not connect to master.

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

go/master/etcd_client.go

@@ -142,3 +172,38 @@ func (e *EtcdClient) Load() ([]byte, error) {
 	state := kvs[0].Value
 	return state, nil
 }
+
+// WaitMasterReady will wait for master is ready
+func (e *EtcdClientWithoutLock) WaitMasterReady(key string, timeout int) []byte {


WaitMasterReady is very long, maybe Getkey would be sufficient?

helinwang · 2017-06-27T22:08:42Z

go/master/etcd_client.go

+}
+
+// WatchWithKey watch the specify key and send to valChan
+func (e *EtcdClientWithoutLock) WatchWithKey(key string, valChan chan string) {


WatchWithKey is very long, maybe WatchKey would be sufficient?

helinwang · 2017-06-27T22:09:12Z

go/master/etcd_client.go

+}
+
+// WatchWithKey watch the specify key and send to valChan
+func (e *EtcdClientWithoutLock) WatchWithKey(key string, valChan chan string) {


valChan chan string can be changed to valChan chan<- string

helinwang · 2017-06-27T22:10:05Z

go/master/etcd_client.go

@@ -18,6 +18,19 @@ const (
 	DefaultAddrPath = "/master/addr"
 )

+// DatabaseOperator is an interface fo database operator,
+// it's useful for unittest
+type DatabaseOperator interface {


Please see my other comment on "removing NewEtcdMasterClient". Maybe we don't need this interface anymore.

…_client

helinwang

LGTM++ after the doc string has been fixed.

helinwang · 2017-06-28T21:05:03Z

go/master/etcd_client.go

@@ -27,7 +27,7 @@ type EtcdClient struct {
 	lock      *concurrency.Mutex
 }

-// NewEtcdClient creates a new EtcdClient.
+// NewEtcdClient creates a new EtcdClient


Subsequent sentences and/or paragraphs can give more details. Sentences should be properly punctuated.

https://github.com/golang/go/wiki/Comments

The comment should be a sentence (ends with ".").

helinwang · 2017-06-28T21:05:09Z

go/master/etcd_client.go

@@ -142,3 +142,31 @@ func (e *EtcdClient) Load() ([]byte, error) {
 	state := kvs[0].Value
 	return state, nil
 }
+
+// GetKey gets the value by the specify key


Subsequent sentences and/or paragraphs can give more details. Sentences should be properly punctuated.

https://github.com/golang/go/wiki/Comments

The comment should be a sentence (ends with ".").

helinwang · 2017-06-28T21:05:22Z

go/master/etcd_client.go

+	return string(v), nil
+}
+
+// WatchKey watches the specify key and send to valChan if there is some event


Subsequent sentences and/or paragraphs can give more details. Sentences should be properly punctuated.

https://github.com/golang/go/wiki/Comments

The comment should be a sentence (ends with ".").

add trainer library

bff8d8d

Yancey1989 requested review from helinwang, jacquesqiao, dzhwinter and typhoonzero June 21, 2017 16:03

modifty file name

d80bd05

helinwang requested changes Jun 21, 2017

View reviewed changes

helinwang mentioned this pull request Jun 21, 2017

add local recordio reader interface #2549

Merged

typhoonzero reviewed Jun 22, 2017

View reviewed changes

move trainer to master client

475f77d

update

c1cfcdc

helinwang reviewed Jun 22, 2017

View reviewed changes

update

3c88fd2

typhoonzero reviewed Jun 23, 2017

View reviewed changes

typhoonzero mentioned this pull request Jun 26, 2017

pserver etcd client #2559

Merged

modify monitor master to receive a chan

cfeac09

jacquesqiao reviewed Jun 26, 2017

View reviewed changes

jacquesqiao reviewed Jun 27, 2017

View reviewed changes

helinwang requested changes Jun 27, 2017

View reviewed changes

Yancey1989 added 5 commits June 27, 2017 15:18

update

ff3324c

Merge branch 'develop' of github.com:PaddlePaddle/Paddle into trainer…

83e7e45

…_client

use etcd client from etcd_client.go

77cbc55

update

f691bf6

update

e6492ff

helinwang requested changes Jun 27, 2017

View reviewed changes

Yancey1989 added 4 commits June 28, 2017 11:55

remove etcd client without lock

ca6a114

Merge branch 'develop' of github.com:PaddlePaddle/Paddle into trainer…

f5ed683

…_client

update

0792515

update

d4c06ed

update the comment

3b02067

helinwang approved these changes Jun 28, 2017

View reviewed changes

update commonts

0d8a0ed

Yancey1989 merged commit 9af8d86 into PaddlePaddle:develop Jun 29, 2017

Yancey1989 deleted the trainer_client branch June 29, 2017 05:20

Trainer library discover master by etcd #2551

Trainer library discover master by etcd #2551

Conversation

Yancey1989 commented Jun 21, 2017

helinwang left a comment

Choose a reason for hiding this comment

typhoonzero left a comment • edited by Yancey1989 Loading

Choose a reason for hiding this comment

Yancey1989 commented Jun 22, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

helinwang Jun 23, 2017 • edited Loading

Choose a reason for hiding this comment

Yancey1989 Jun 26, 2017 • edited Loading

Choose a reason for hiding this comment

helinwang Jun 26, 2017 • edited Loading

Choose a reason for hiding this comment

jacquesqiao Jun 26, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Yancey1989 Jun 27, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

helinwang Jun 27, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

helinwang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

typhoonzero left a comment •

edited by Yancey1989

Loading

helinwang Jun 23, 2017 •

edited

Loading

Yancey1989 Jun 26, 2017 •

edited

Loading

helinwang Jun 26, 2017 •

edited

Loading

jacquesqiao Jun 26, 2017 •

edited

Loading

Yancey1989 Jun 27, 2017 •

edited

Loading

helinwang Jun 27, 2017 •

edited

Loading