diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 61b989dc698798eca932516e558c63f107ef2754..efb4dcb2dfbc63bb6905961b054cdef860cf4573 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -21,10 +21,10 @@ sha: 28c0ea8a67a3e2dbbf4822ef44e85b63a0080a29 hooks: - id: clang-formater -- repo: https://github.com/dnephin/pre-commit-golang - sha: e4693a4c282b4fc878eda172a929f7a6508e7d16 +- repo: https://github.com/PaddlePaddle/pre-commit-golang + sha: 16398aeccf263adaf53b2495eed0406347d76281 hooks: - id: go-fmt - files: (.*\.go) - - id: go-lint - files: (.*\.go) + types: [go] + - id: gometalinter + types: [go] diff --git a/.travis.yml b/.travis.yml index 2cf7666fb5d0c47034676864a52c3d3dbce19683..376c693602b56fe719decfeb41c217497e143e12 100644 --- a/.travis.yml +++ b/.travis.yml @@ -41,6 +41,8 @@ before_install: - pip install rarfile - curl https://glide.sh/get | bash - eval "$(GIMME_GO_VERSION=1.8.3 gimme)" + - go get -u github.com/alecthomas/gometalinter + - gometalinter --install - | function timeout() { perl -e 'alarm shift; exec @ARGV' "$@"; } script: diff --git a/go/master/c/client.go b/go/master/c/client.go index 31f431197454c2ec6a25624d37b60876d99f0087..2cbe164c7b406b189f44ec850796203f24779205 100644 --- a/go/master/c/client.go +++ b/go/master/c/client.go @@ -23,7 +23,6 @@ import ( log "github.com/sirupsen/logrus" ) -var nullPtr = unsafe.Pointer(uintptr(0)) var mu sync.Mutex var handleMap = make(map[C.paddle_master_client]*master.Client) var curHandle C.paddle_master_client @@ -114,13 +113,13 @@ func paddle_next_record(client C.paddle_master_client, record **C.uchar) C.int { if err != nil { // Error // TODO: return the type of error? - *record = (*C.uchar)(nullPtr) + *record = (*C.uchar)(nil) return -1 } if len(r) == 0 { // Empty record - *record = (*C.uchar)(nullPtr) + *record = (*C.uchar)(nil) return 0 } diff --git a/go/master/client.go b/go/master/client.go index de883bf4b9a3de8d6d6e35e8e808dcf7ba54cb46..90b99470978d21480eb2d8097e7dec217b9524eb 100644 --- a/go/master/client.go +++ b/go/master/client.go @@ -69,7 +69,10 @@ func (c *Client) getRecords() { // We treat a task as finished whenever the last data // instance of the task is read. This is not exactly // correct, but a reasonable approximation. - c.taskFinished(t.Meta.ID) + err = c.taskFinished(t.Meta.ID) + if err != nil { + log.Errorln(err) + } } } diff --git a/go/master/client_internal_test.go b/go/master/client_internal_test.go index 49263474c8fe2410ffb6db93a9647f5ab066b06b..70dc09bf9461142ff6498355a5858ba9a1320510 100644 --- a/go/master/client_internal_test.go +++ b/go/master/client_internal_test.go @@ -66,11 +66,21 @@ func TestGetFinishTask(t *testing.T) { for i := 0; i < totalTask*chunkPerTask; i++ { w := recordio.NewWriter(f, -1, -1) - w.Write(nil) + _, err = w.Write(nil) + if err != nil { + panic(err) + } + // call Close to force RecordIO writing a chunk. - w.Close() + err = w.Close() + if err != nil { + panic(err) + } + } + err = f.Close() + if err != nil { + panic(err) } - f.Close() // Manually intialize client to avoid calling c.getRecords() c := &Client{} @@ -79,7 +89,11 @@ func TestGetFinishTask(t *testing.T) { ch := make(chan string, 1) ch <- addr go c.monitorMaster(ch) - c.SetDataset([]string{path}) + err = c.SetDataset([]string{path}) + if err != nil { + panic(err) + } + checkOnePass := func(i int) { var tasks []Task for idx := 0; idx < totalTask; idx++ { diff --git a/go/master/client_test.go b/go/master/client_test.go index 6666d3860c412daa8711fbfa2d729a261b3fd887..bc92dc5ac973d62434b71e09705143ac8fbbd2fa 100644 --- a/go/master/client_test.go +++ b/go/master/client_test.go @@ -57,14 +57,30 @@ func TestNextRecord(t *testing.T) { w := recordio.NewWriter(f, -1, -1) for i := 0; i < total; i++ { - w.Write([]byte{byte(i)}) + _, err = w.Write([]byte{byte(i)}) + if err != nil { + panic(err) + } + } + + err = w.Close() + if err != nil { + panic(err) + } + + err = f.Close() + if err != nil { + panic(err) } - w.Close() - f.Close() + curAddr := make(chan string, 1) curAddr <- fmt.Sprintf(":%d", p) c := master.NewClient(curAddr, 10) - c.SetDataset([]string{path}) + err = c.SetDataset([]string{path}) + if err != nil { + panic(err) + } + for pass := 0; pass < 50; pass++ { received := make(map[byte]bool) for i := 0; i < total; i++ { diff --git a/go/master/etcd_client.go b/go/master/etcd_client.go index 04c1394e963d1eb541b80b91407fb55b0d1e1f2a..69dc6a8268748ad9a72eb10fc2789982f565d291 100644 --- a/go/master/etcd_client.go +++ b/go/master/etcd_client.go @@ -30,7 +30,7 @@ type EtcdClient struct { // NewEtcdClient creates a new EtcdClient. func NewEtcdClient(endpoints []string, addr string, lockPath, addrPath, statePath string, ttlSec int) (*EtcdClient, error) { log.Debugf("Connecting to etcd at %v", endpoints) - // TODO(helin): gracefully shutdown etcd store. Becuase etcd + // TODO(helin): gracefully shutdown etcd store. Because etcd // store holds a etcd lock, even though the lock will expire // when the lease timeout, we need to implement graceful // shutdown to release the lock. @@ -60,7 +60,7 @@ func NewEtcdClient(endpoints []string, addr string, lockPath, addrPath, statePat } log.Debugf("Successfully acquired lock at %s.", lockPath) - put := clientv3.OpPut(addrPath, string(addr)) + put := clientv3.OpPut(addrPath, addr) resp, err := cli.Txn(context.Background()).If(lock.IsOwner()).Then(put).Commit() if err != nil { return nil, err diff --git a/go/master/inmem_store.go b/go/master/inmem_store.go index bcd549b20e46381783bad11caa08cb7f4ba40add..57e75dc4e01b4bafa8153bcc7fbc82a9eb2b08f5 100644 --- a/go/master/inmem_store.go +++ b/go/master/inmem_store.go @@ -4,7 +4,7 @@ import "sync" // InMemStore is an in memory implementation of Store interface. // -// It does not tolerate the fault that casues the program to crash. +// It does not tolerate the fault that causes the program to crash. type InMemStore struct { mu sync.Mutex buf []byte diff --git a/go/master/service.go b/go/master/service.go index 9cef2270ce6a51425e40b9281f93f2f9c9981329..262735f421ad7ae04050e9264a177ee4c46e68d0 100644 --- a/go/master/service.go +++ b/go/master/service.go @@ -160,7 +160,7 @@ func (s *Service) recover() (bool, error) { // snapshot *must* be called with s.mu being held. func (s *Service) snapshot() error { - // TOOD(helin): etcd request has a size limit, so the snapshot + // TODO(helin): etcd request has a size limit, so the snapshot // size is limited by the max request size. We should either // divide the snapshot into smaller chunks and save under // different keys, or configure the request size to be big @@ -289,7 +289,6 @@ func (s *Service) processFailedTask(t taskEntry, epoch int) { log.Warningf("Task %v failed %d times, discard.", t.Task, t.NumFailure) s.taskQueues.Todo = append(s.taskQueues.Todo, t) - return } func (s *Service) checkTimeoutFunc(taskID int, epoch int) func() { diff --git a/go/pserver/client/c/cclient.go b/go/pserver/client/c/cclient.go index 7ddaceb7ed33db32e19a191402100a0c0efa241a..718b4304c80791b4d8a8816f256c8fa93e0b1ead 100644 --- a/go/pserver/client/c/cclient.go +++ b/go/pserver/client/c/cclient.go @@ -34,7 +34,6 @@ import ( log "github.com/sirupsen/logrus" ) -var nullPtr = unsafe.Pointer(uintptr(0)) var mu sync.Mutex var handleMap = make(map[C.paddle_pserver_client]*client.Client) var curHandle C.paddle_pserver_client @@ -63,7 +62,7 @@ func remove(client C.paddle_pserver_client) *client.Client { } func cArrayToSlice(p unsafe.Pointer, len int) []byte { - if p == nullPtr { + if p == nil { return nil } @@ -101,11 +100,11 @@ func paddle_new_pserver_client(addrs *C.char, selected int) C.paddle_pserver_cli } //export paddle_new_etcd_pserver_client -func paddle_new_etcd_pserver_client(etcd_endpoints *C.char, selected int) C.paddle_pserver_client { +func paddle_new_etcd_pserver_client(etcdEndpoints *C.char, selected int) C.paddle_pserver_client { // TODO(Longfei: use etcd lock to decide which trainer to initialize the parameters) - addr := C.GoString(etcd_endpoints) - etcd_client := client.NewEtcd(addr) - c := client.NewClient(etcd_client, etcd_client.Desired(), selector(selected != 0)) + addr := C.GoString(etcdEndpoints) + etcdClient := client.NewEtcd(addr) + c := client.NewClient(etcdClient, etcdClient.Desired(), selector(selected != 0)) return add(c) } @@ -124,20 +123,20 @@ func paddle_begin_init_params(client C.paddle_pserver_client) C.int { } //export paddle_init_param -func paddle_init_param(client C.paddle_pserver_client, param C.paddle_parameter, param_config unsafe.Pointer, config_len C.int) C.int { +func paddle_init_param(client C.paddle_pserver_client, param C.paddle_parameter, paramConfig unsafe.Pointer, configLen C.int) C.int { et := pserver.ElementType(param.element_type) name := C.GoString(param.name) content := cArrayToSlice(unsafe.Pointer(param.content), int(param.content_len)) pc := pserver.ParameterWithConfig{ Param: pserver.Parameter{Name: name, ElementType: et, Content: content}, - Config: cArrayToSlice(param_config, int(config_len)), + Config: cArrayToSlice(paramConfig, int(configLen)), } c := get(client) err := c.InitParam(pc) if err != nil { if err.Error() == pserver.AlreadyInitialized { - log.Warningf("parameter %s already initialized, treat paddle_init_param as sucessful.", name) + log.Warningf("parameter %s already initialized, treat paddle_init_param as successful.", name) return C.PSERVER_OK } log.Errorln(err) @@ -153,7 +152,7 @@ func paddle_finish_init_params(client C.paddle_pserver_client) C.int { err := c.FinishInitParams() if err != nil { if err.Error() == pserver.AlreadyInitialized { - log.Warningln("parameters already initialized, treat paddle_finish_init_params as sucessful.") + log.Warningln("parameters already initialized, treat paddle_finish_init_params as successful.") return C.PSERVER_OK } @@ -223,12 +222,12 @@ func paddle_get_params(client C.paddle_pserver_client, dst **C.paddle_parameter, p := ps[i] param := *(**C.paddle_parameter)(unsafe.Pointer((uintptr(unsafe.Pointer(dst)) + uintptr(i)*unsafe.Sizeof(*dst)))) - if unsafe.Pointer(param) == nullPtr { + if unsafe.Pointer(param) == nil { log.Errorln("must pre-allocate parameter.") return C.PSERVER_ERROR } - if unsafe.Pointer(param.content) != nullPtr { + if unsafe.Pointer(param.content) != nil { if int(param.content_len) != len(p.Content) { log.Errorf("the pre-allocated content len does not match parameter content len. Pre-allocated len: %d, returned len: %d", param.content_len, len(p.Content)) return C.PSERVER_ERROR diff --git a/go/pserver/client/client.go b/go/pserver/client/client.go index aa8bfe30c26fcc0875ad479ecd562700ccefa5a3..b4a45e1c21056550ef9264746bcf58a8abb369a1 100644 --- a/go/pserver/client/client.go +++ b/go/pserver/client/client.go @@ -233,7 +233,7 @@ func (c *Client) Save(path string) error { func strHash(s string) uint32 { h := fnv.New32a() - h.Write([]byte(s)) + _, _ = h.Write([]byte(s)) return h.Sum32() } diff --git a/go/pserver/client/client_test.go b/go/pserver/client/client_test.go index aab91556b4b91fab6de66322987eabe24f1b0f70..5c89882a297323034be2875a6d4cb71d715eb0c2 100644 --- a/go/pserver/client/client_test.go +++ b/go/pserver/client/client_test.go @@ -79,15 +79,33 @@ func initEtcdClient() { log.Errorf("err %v", err) } ctx, cancel := context.WithTimeout(context.Background(), timeout) - client.Delete(ctx, pserver.PsDesired) - client.Delete(ctx, pserver.PsPath) - client.Put(ctx, pserver.PsDesired, strconv.Itoa(numPserver)) + _, err = client.Delete(ctx, pserver.PsDesired) + if err != nil { + panic(err) + } + + _, err = client.Delete(ctx, pserver.PsPath) + if err != nil { + panic(err) + } + + _, err = client.Put(ctx, pserver.PsDesired, strconv.Itoa(numPserver)) + if err != nil { + panic(err) + } + ports := initClient() for i := 0; i < numPserver; i++ { - client.Put(ctx, pserver.PsPath+strconv.Itoa(i), ":"+strconv.Itoa(ports[i])) + _, err = client.Put(ctx, pserver.PsPath+strconv.Itoa(i), ":"+strconv.Itoa(ports[i])) + if err != nil { + panic(err) + } } cancel() - client.Close() + err = client.Close() + if err != nil { + panic(err) + } } type selector bool diff --git a/go/pserver/client/etcd_client.go b/go/pserver/client/etcd_client.go index 8eb2a4f4511fc7139a55a2cd47ad73a82137b260..953065b427ed52d39f1253ea94d485b765ea5dc2 100644 --- a/go/pserver/client/etcd_client.go +++ b/go/pserver/client/etcd_client.go @@ -12,8 +12,7 @@ import ( ) const ( - // DefaultEtcdTimeout is the default etcd timeout - DefaultEtcdTimeout time.Duration = 5 * time.Second + defaultEtcdTimeout time.Duration = 5 * time.Second ) // EtcdClient is used by pserver client that is a part of trainer process. @@ -48,7 +47,7 @@ func (p *EtcdClient) Desired() int { psDesired, err = strconv.Atoi(string(resp.Kvs[0].Value)) if err != nil { - log.Errorf("psDesired %s invalid %v", psDesired, err) + log.Errorf("psDesired %d invalid %v", psDesired, err) time.Sleep(p.timeout) continue } @@ -67,12 +66,12 @@ func (p *EtcdClient) List() []Server { for { for i := 0; i < psDesired; i++ { ctx, cancel := context.WithTimeout(context.Background(), p.timeout) + cancel() psKey := pserver.PsPath + strconv.Itoa(i) log.Debugf("checking %s", psKey) resp, err := p.client.Get(ctx, psKey) - cancel() if err != nil { - log.Infof("Get psKey=%s error, %v", psKey, err) + log.Infof("Get psKey= %s error, %v", psKey, err) time.Sleep(p.timeout) continue } @@ -107,11 +106,11 @@ func NewEtcd(endpoints string) *EtcdClient { for { cli, err = clientv3.New(clientv3.Config{ Endpoints: ep, - DialTimeout: DefaultEtcdTimeout, + DialTimeout: defaultEtcdTimeout, }) if err != nil { log.Errorf("Init etcd connection failed: %v", err) - time.Sleep(DefaultEtcdTimeout) + time.Sleep(defaultEtcdTimeout) continue } break @@ -119,7 +118,7 @@ func NewEtcd(endpoints string) *EtcdClient { log.Infof("Connected to etcd: %s\n", endpoints) client := &EtcdClient{ client: cli, - timeout: DefaultEtcdTimeout, + timeout: defaultEtcdTimeout, endpoints: ep, } return client diff --git a/go/pserver/etcd_client.go b/go/pserver/etcd_client.go index 66af4fa0b483f1caea385df61e54d871072a0375..e70e826975b26db302a6799e9171cff970153aac 100644 --- a/go/pserver/etcd_client.go +++ b/go/pserver/etcd_client.go @@ -177,10 +177,10 @@ func (e *EtcdClient) registerPserverEtcd(ctx context.Context, port int) (int, er break } } - if registered == true { + if registered { return nil } - return errors.New("not registerd, may due to already have enough pservers") + return errors.New("not registered, may due to already have enough pservers") }, concurrency.WithAbortContext(ctx), concurrency.WithIsolation(concurrency.RepeatableReads)) if err != nil { @@ -211,8 +211,5 @@ func (e *EtcdClient) PutKey(key string, value []byte, timeout time.Duration) err ctx, cancel := context.WithTimeout(context.Background(), timeout) _, err := e.etcdClient.Put(ctx, key, string(value)) cancel() - if err != nil { - return err - } - return nil + return err } diff --git a/go/pserver/optimizer.go b/go/pserver/optimizer.go index d6b7fafd59c0453b9f40019166d01ebdc9775117..151a3f80332b0e62767586f9f769c839ba19ce1e 100644 --- a/go/pserver/optimizer.go +++ b/go/pserver/optimizer.go @@ -14,8 +14,6 @@ import ( log "github.com/sirupsen/logrus" ) -var nullPtr = unsafe.Pointer(uintptr(0)) - type optimizer struct { opt *C.struct_paddle_optimizer elementType ElementType @@ -23,7 +21,7 @@ type optimizer struct { } func cArrayToSlice(p unsafe.Pointer, len int) []byte { - if p == nullPtr { + if p == nil { return nil } @@ -92,8 +90,8 @@ func (o *optimizer) UpdateParameter(g Gradient) error { } func (o *optimizer) Cleanup() { - if unsafe.Pointer(o.opt) != nullPtr { + if unsafe.Pointer(o.opt) != nil { C.paddle_release_optimizer(o.opt) - o.opt = (*C.struct_paddle_optimizer)(nullPtr) + o.opt = (*C.struct_paddle_optimizer)(nil) } } diff --git a/go/pserver/service.go b/go/pserver/service.go index fec2ec61dc67756439d9fa478788d1f155876538..c723959d6b87524762e2f874bb5e4d5bd567cd00 100644 --- a/go/pserver/service.go +++ b/go/pserver/service.go @@ -211,7 +211,7 @@ func (s *Service) GetParam(name string, parameter *Parameter) error { // learning optimization methods are stochastic in // nature. This race condition is allowed deliberately // to save the program from making a copy of the - // paramter content. + // parameter content. parameter.Name = name parameter.ElementType = opt.elementType parameter.Content = opt.GetWeights() @@ -219,7 +219,7 @@ func (s *Service) GetParam(name string, parameter *Parameter) error { } // pserver save checkpoint -func (s *Service) doCheckpoint() error { +func (s *Service) doCheckpoint() (err error) { <-s.initialized s.mu.Lock() defer s.mu.Unlock() @@ -237,9 +237,9 @@ func (s *Service) doCheckpoint() error { } var buf bytes.Buffer encoder := gob.NewEncoder(&buf) - err := encoder.Encode(cp) + err = encoder.Encode(cp) if err != nil { - return err + return } cpMeta := checkpointMeta{} @@ -248,10 +248,14 @@ func (s *Service) doCheckpoint() error { h := md5.New() cpMeta.MD5 = hex.EncodeToString(h.Sum(buf.Bytes())) - cpMetajson, _ := json.Marshal(cpMeta) + cpMetajson, err := json.Marshal(cpMeta) + if err != nil { + return + } + err = s.client.PutKey(filepath.Join(PsCheckpoint, strconv.Itoa(s.idx)), cpMetajson, 3*time.Second) if err != nil { - return err + return } if _, err = os.Stat(cpMeta.UUID); os.IsNotExist(err) { log.Info("checkpoint does not exists.") @@ -264,15 +268,32 @@ func (s *Service) doCheckpoint() error { } } f, err := os.Create(cpMeta.UUID) - defer f.Close() if err != nil { - return err + return } + + defer func() { + closeErr := f.Close() + if closeErr != nil { + if err != nil { + log.Errorln(closeErr) + } else { + // Set closeErr as return value. + err = closeErr + } + } + }() + writer := bufio.NewWriter(f) _, err = writer.Write(buf.Bytes()) - writer.Flush() if err != nil { - return err + return } - return nil + + err = writer.Flush() + if err != nil { + return + } + + return } diff --git a/paddle/scripts/travis/check_style.sh b/paddle/scripts/travis/check_style.sh index 4754bdd4c80de9700d92b0e33ecfdfc582f42813..8049aeb7b00870220e59c981addf6d70a66877c7 100755 --- a/paddle/scripts/travis/check_style.sh +++ b/paddle/scripts/travis/check_style.sh @@ -13,6 +13,11 @@ export PATH=/usr/bin:$PATH pre-commit install clang-format --version +# set up go environment for running gometalinter +mkdir -p $GOPATH/src/github.com/PaddlePaddle/ +ln -sf $TRAVIS_BUILD_DIR $GOPATH/src/github.com/PaddlePaddle/Paddle +cd $GOPATH/src/github.com/PaddlePaddle/Paddle/go; glide install; cd - + if ! pre-commit run -a ; then git diff --exit-code fi