bugfix:1.modify preload report.

2.fix nullpoint error in hot volume hadoop

Signed-off-by: chihe <chi.he@oppo.com>

libsdk/libsdk.go

@@ -60,6 +60,7 @@ import "C"
 import (
 	"context"
 	"fmt"
+	"github.com/cubefs/cubefs/util/buf"
 	"io"
 	syslog "log"
 	"os"
@@ -973,7 +974,6 @@ func (c *client) absPath(path string) string {
 
 func (c *client) start() (err error) {
 	var masters = strings.Split(c.masterAddr, ",")
-
 	if c.logDir != "" {
 		if c.logLevel == "" {
 			c.logLevel = "WARN"
@@ -992,7 +992,6 @@ func (c *client) start() (err error) {
 	if err = c.loadConfFromMaster(masters); err != nil {
 		return
 	}
-
 	if err = c.checkPermission(); err != nil {
 		err = errors.NewErrorf("check permission failed: %v", err)
 		syslog.Println(err)
@@ -1005,7 +1004,6 @@ func (c *client) start() (err error) {
 	if c.enableBcache {
 		c.bc = bcache.NewBcacheClient()
 	}
-
 	var ebsc *blobstore.BlobStoreClient
 	if ebsc, err = blobstore.NewEbsClient(access.Config{
 		ConnMode: access.NoLimitConnMode,
@@ -1019,7 +1017,6 @@ func (c *client) start() (err error) {
 	}); err != nil {
 		return
 	}
-
 	var mw *meta.MetaWrapper
 	if mw, err = meta.NewMetaWrapper(&meta.MetaConfig{
 		Volume:        c.volName,
@@ -1030,7 +1027,6 @@ func (c *client) start() (err error) {
 		log.LogErrorf("newClient NewMetaWrapper failed(%v)", err)
 		return err
 	}
-
 	var ec *stream.ExtentClient
 	if ec, err = stream.NewExtentClient(&stream.ExtentConfig{
 		Volume:            c.volName,
@@ -1293,6 +1289,7 @@ func (c *client) loadConfFromMaster(masters []string) (err error) {
 	c.ebsEndpoint = clusterInfo.EbsAddr
 	c.servicePath = clusterInfo.ServicePath
 	c.cluster = clusterInfo.Cluster
+	buf.InitCachePool(c.ebsBlockSize)
 	return
 }
 

preload/preload.go

@@ -84,7 +84,9 @@ func main() {
 	action := cfg.GetString("action")
 	if action == "preload" {
 		if err := cli.PreloadDir(cfg.GetString("target"), int(replicaNum), uint64(ttl), cfg.GetString("zones")); err != nil {
-			fmt.Println(err)
+			total, succeed := cli.GetPreloadResult()
+			fmt.Printf("Preload failed:%v\n",err)
+			fmt.Printf("Result: total[%v], succeed[%v]\n", total, succeed)
 		} else {
 			fmt.Println("Preload succeed")
 		}

preload/sdk/preloadsdk.go

@@ -515,11 +515,11 @@ func (c *PreLoadClient) preloadFile() error {
 	}
 	close(jobs)
 	wg.Wait()
-	log.LogInfof("preloadFile end:total %v, succeed %v", c.preloadFileNumSucceed, c.preloadFileNumTotal)
+	log.LogInfof("preloadFile end:total %v, succeed %v", c.preloadFileNumTotal, c.preloadFileNumSucceed)
 	if c.preloadFileNumTotal == c.preloadFileNumSucceed {
 		return nil
 	} else if c.preloadFileNumSucceed < c.preloadFileNumTotal {
-		return errors.New("Preload partitionly succeed")
+		return errors.New("Preload partially succeed")
 	} else {
 		return errors.New("Preload failed")
 	}
@@ -563,5 +563,8 @@ func (c *PreLoadClient) PreloadDir(target string, count int, ttl uint64, zones s
 	log.LogDebugf("Sleep end")
 	//Step3.2  preload the file
 	return c.preloadFile()
-
+}
+
+func (c *PreLoadClient) GetPreloadResult()(int64 , int64){
+	return c.preloadFileNumTotal, c.preloadFileNumSucceed
 }

sdk/data/wrapper/wrapper.go

@@ -197,15 +197,15 @@ func (w *Wrapper) updateDataPartitionByRsp(isInit bool, DataPartitions []*proto.
 			continue
 		}
 		dp := convert(partition)
-		//do not insert preload dp in cold vol
-		if proto.IsCold(w.volType) && proto.IsPreLoadDp(dp.PartitionType) {
-			continue
-		}
 		if w.followerRead && w.nearRead {
 			dp.NearHosts = w.sortHostsByDistance(dp.Hosts)
 		}
 		log.LogInfof("updateDataPartition: dp(%v)", dp)
 		w.replaceOrInsertPartition(dp)
+		//do not insert preload dp in cold vol
+		if proto.IsCold(w.volType) && proto.IsPreLoadDp(dp.PartitionType) {
+			continue
+		}
 		if dp.Status == proto.ReadWrite {
 			dp.MetricsRefresh()
 			rwPartitionGroups = append(rwPartitionGroups, dp)

util/buf/cache_pool.go

@@ -31,6 +31,9 @@ type FileCachePool struct {
 
 
 func InitCachePool(blockSize int) {
+	if blockSize == 0 {
+		return
+	}
 	CachePool = &FileCachePool{}
 	cacheTotalLimit = int64((4 * util.GB) / blockSize)
 	CachePool.pool = newWriterCachePool(blockSize)