func startP2PServer() {
	p2pListener, err := net.Listen("tcp", ":35800")
	if err != nil {
		log.Fatal(err)
	}

	p2psl, err = tcpNew(p2pListener)
	if err != nil {
		log.Fatal(err)
	}

	P2pRouter := httprouter.New()
	P2pRouter.GET("/", sayhello)
	P2pRouter.GET("/pull/:repo/:dataitem/:tag", p2p_pull)
	P2pRouter.GET("/health", p2pHealthyCheckHandler)

	p2pserver := http.Server{Handler: P2pRouter}

	wg.Add(1)
	defer wg.Done()

	log.Info("p2p server start")
	p2pserver.Serve(p2psl)
	log.Info("p2p server stop")

}

func commToServer(method, path string, buffer []byte, w http.ResponseWriter) (body []byte, err error) {
	//Trace()
	s := log.Info("daemon: connecting to", DefaultServer+path)
	logq.LogPutqueue(s)
	req, err := http.NewRequest(strings.ToUpper(method), DefaultServer+path, bytes.NewBuffer(buffer))
	if len(loginAuthStr) > 0 {
		req.Header.Set("Authorization", loginAuthStr)
	}

	//req.Header.Set("User", "admin")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		log.Error(err)
		d := ds.Result{Code: cmd.ErrorServiceUnavailable, Msg: err.Error()}
		body, e := json.Marshal(d)
		if e != nil {
			log.Error(e)
			return body, e
		}
		w.WriteHeader(http.StatusServiceUnavailable)
		w.Write(body)
		return body, err
	}
	defer resp.Body.Close()

	w.WriteHeader(resp.StatusCode)
	body, err = ioutil.ReadAll(resp.Body)
	w.Write(body)
	log.Info(resp.StatusCode, string(body))
	return
}

func delTagsForDelItem(reponame, itemname string) error {
	log.Println("Begin to remove tags for remove item from db")
	sqlrpdmid := fmt.Sprintf(`SELECT RPDMID FROM DH_DP_RPDM_MAP WHERE REPOSITORY='%s' AND DATAITEM='%s' AND STATUS='A';`, reponame, itemname)

	row, err := g_ds.QueryRow(sqlrpdmid)
	if err != nil {
		l := log.Error("select rpdmid from DH_DP_RPDM_MAP error:", err)
		logq.LogPutqueue(l)
		return err
	}
	var rpdmId int
	row.Scan(&rpdmId)
	if rpdmId == 0 {
		log.Debug(reponame, itemname, "not exist.")
		return nil
	}
	sqldeltag := fmt.Sprintf(`UPDATE DH_RPDM_TAG_MAP SET STATUS='N' WHERE RPDMID=%d`, rpdmId)
	_, err = g_ds.Update(sqldeltag)
	log.Info("sqldeltag", sqldeltag)
	if err != nil {
		l := log.Error("delete tag error:", err)
		logq.LogPutqueue(l)
		return err
	}

	return nil
}

func (s3 *s3driver) CheckDataAndGetSize(dpconn, itemlocation, fileName string) (exist bool, size int64, err error) {
	bucket := getAwsInfoFromDpconn(dpconn)

	destFullPathFileName := bucket + "/" + itemlocation + "/" + fileName
	log.Info(destFullPathFileName)

	AWS_REGION = Env("AWS_REGION", false)

	svc := s3aws.New(session.New(&aws.Config{Region: aws.String(AWS_REGION)}))
	result, err := svc.ListObjects(&s3aws.ListObjectsInput{Bucket: aws.String(bucket),
		Prefix: aws.String(itemlocation + "/" + fileName)})
	if err != nil {
		log.Error("Failed to list objects", err)
		return exist, size, err
	}

	exist = false
	for _, v := range result.Contents {
		log.Infof("Tag:%s, key:%s, size:%v\n", aws.StringValue(v.ETag), aws.StringValue(v.Key), aws.Int64Value(v.Size))
		if aws.StringValue(v.Key) == fileName {
			size = aws.Int64Value(v.Size)
			exist = true
		}
	}

	return
}

func (hdfs *hdfsdriver) StoreFile(status, filename, dpconn, dp, itemlocation, destfile string) string {

	log.Infof("Begin to upload %v to %v\n", filename, dp)

	client, err := getClient(dpconn)
	if err != nil {
		log.Error("Failed to get a client", err)
		status = "put to hdfs err"
		return status
	}
	defer client.Close()

	err = client.MkdirAll("/"+itemlocation, 1777)
	if err != nil {
		log.Error("Failed to mkdirall in hdfs", err)
		status = "put to hdfs err"
		return status
	}

	hdfsfile := "/" + itemlocation + "/" + destfile
	err = client.CopyToRemote(filename, hdfsfile)
	if err != nil {
		log.Error("Failed to CopyToRemote", err)
		status = "put to hdfs err"
		return status
	}

	status = "put to hdfs ok"
	log.Info("Successfully uploaded to", itemlocation, "in hdfs")
	return status
}

func (hdfs *hdfsdriver) CheckDataAndGetSize(dpconn, itemlocation, fileName string) (exist bool, size int64, err error) {

	destFullPathFileName := "/" + itemlocation + "/" + fileName
	log.Info(destFullPathFileName)

	exist = false

	client, err := getClient(dpconn)
	if err != nil {
		log.Error("Failed to get a client", err)
		return
	}
	defer client.Close()

	fileinfo, _ := client.Stat(destFullPathFileName)

	if fileinfo != nil {
		exist = true
		cs, _ := client.GetContentSummary(destFullPathFileName)
		size = cs.Size()
	} else {
		err = errors.New("文件不存在")
		return
	}

	return
}

func UpdateStatMsgTagadded(ID, Status int) (err error) {

	log.Info("update MSG_TAGADDED status")
	sql := fmt.Sprintf(`UPDATE MSG_TAGADDED SET STATUS=%d 
		WHERE ID=%d;`, Status, ID)
	_, err = g_ds.Update(sql)
	if err != nil {
		l := log.Error(err)
		logq.LogPutqueue(l)
		return
	}
	return
}

func getDaemonid() (id string) {
	log.Println("TODO get daemonid from db.")
	s := `SELECT DAEMONID FROM DH_DAEMON;`
	row, e := g_ds.QueryRow(s)
	if e != nil {
		l := log.Error(s, "error.", e)
		logq.LogPutqueue(l)
		return
	}
	row.Scan(&id)
	log.Info("daemon id is", id)
	return id
}

func testserver() {
	var tsl = new(StoppabletcpListener)

	Listener, err := net.Listen("tcp", ":35888")
	if err != nil {
		log.Fatal(err)
	}

	tsl, err = tcpNew(Listener)
	if err != nil {
		log.Fatal(err)
	}

	tRouter := httprouter.New()
	tRouter.GET("/", sayhello)

	http.Handle("/", tRouter)

	server := http.Server{Handler: tRouter}

	log.Info("p2p server start")
	server.Serve(tsl)
	log.Info("p2p server stop")
}

func epPostHandler(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	reqBody, _ := ioutil.ReadAll(r.Body)
	ep := cmd.FormatEp{}
	if err := json.Unmarshal(reqBody, &ep); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	EntryPoint = ep.Ep
	saveEntryPoint(EntryPoint)

	msg := ds.MsgResp{Msg: "OK. your entrypoint is: " + EntryPoint + "\n" + "The entrypoint is setted successfully.  You can check whether it is available in one minute."}
	log.Info(msg.Msg)
	resp, _ := json.Marshal(&msg)
	fmt.Fprintln(w, string(resp))
	return
}

func commToServerGetRsp(method, path string, buffer []byte) (resp *http.Response, err error) {

	s := log.Info("daemon: connecting to", DefaultServer+path)
	logq.LogPutqueue(s)
	req, err := http.NewRequest(strings.ToUpper(method), DefaultServer+path, bytes.NewBuffer(buffer))
	if len(loginAuthStr) > 0 {
		req.Header.Set("Authorization", loginAuthStr)
	}

	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		log.Error(err)
		return resp, err
	}

	return resp, nil
}

func PullTagAutomatic() {
	for {
		time.Sleep(30 * time.Second)

		//log.Debug("AutomaticPullList.Len()", AutomaticPullList.Len())
		var Tags map[int]string
		for e := AutomaticPullList.Front(); e != nil; e = e.Next() {
			v := e.Value.(ds.DsPull)
			log.Info("PullTagAutomatic begin", v.Repository, v.Dataitem)
			Tags = GetTagFromMsgTagadded(v.Repository, v.Dataitem, NOTREAD)

			log.Println("Tags ", Tags)
			go PullItemAutomatic(Tags, v)

		}
	}
}

func (s3 *s3driver) StoreFile(status, filename, dpconn, dp, itemlocation, destfile string) string {
	bucket := getAwsInfoFromDpconn(dpconn)

	//AWS_SECRET_ACCESS_KEY = Env("AWS_SECRET_ACCESS_KEY", false)
	//AWS_ACCESS_KEY_ID = Env("AWS_ACCESS_KEY_ID", false)
	AWS_REGION = Env("AWS_REGION", false)

	file, err := os.Open(filename)
	if err != nil {
		l := log.Error("Failed to open file", err)
		logq.LogPutqueue(l)
		status = "put to s3 err"
		return status
	}

	log.Infof("Begin to upload %v to %v\n", filename, dp)

	// Not required, but you could zip the file before uploading it
	// using io.Pipe read/writer to stream gzip'd file contents.
	reader, writer := io.Pipe()
	go func() {
		gw := gzip.NewWriter(writer)
		io.Copy(gw, file)

		file.Close()
		gw.Close()
		writer.Close()

		//updateJobQueueStatus(jobid, "puttos3ok")
	}()
	uploader := s3manager.NewUploader(session.New(&aws.Config{Region: aws.String(AWS_REGION)}))
	//uploader := s3manager.NewUploader(session.New(aws.NewConfig()))
	result, err := uploader.Upload(&s3manager.UploadInput{
		Body:   reader,
		Bucket: aws.String(bucket),
		Key:    aws.String( /*dp + "/" + */ itemlocation + "/" + destfile + ".gz"),
	})
	if err != nil {
		log.Error("Failed to upload", err)
		status = "put to s3 err"
		return status
	}
	status = "put to s3 ok"
	log.Info("Successfully uploaded to", result.Location)
	return status
}

func CheckHealthClock() {
	log.Debug("--------->BEGIN")

	checkHealth(&Errortagsmap)

	timer := time.NewTicker(10 * time.Minute)
	for {
		select {
		case <-timer.C:
			now := time.Now()
			if now.Hour()%6 == 0 {
				log.Info("Time:", now)
				checkHealth(&Errortagsmap)
			}
		}
	}
	log.Debug("---------->END")
}

func AutomaticPullRmqueue(p ds.DsPull) (exist bool) {
	exist = false
	pullmu.Lock()
	defer pullmu.Unlock()

	var next *list.Element
	for e := AutomaticPullList.Front(); e != nil; e = next {
		v := e.Value.(ds.DsPull)
		if v.Repository == p.Repository && v.Dataitem == p.Dataitem {
			exist = true
			AutomaticPullList.Remove(e)
			log.Info(v, "removed from the queue.")
			break
		} else {
			next = e.Next()
		}
	}
	return
}

func (s3 *s3driver) GetFileTobeSend(dpconn, dpname, itemlocation, tagdetail string) (filepathname string) {

	bucket := getAwsInfoFromDpconn(dpconn)

	e := os.MkdirAll(gDpPath+"/"+bucket+"/"+itemlocation, 0777)
	if e != nil {
		log.Error(e)
		return
	}

	filepathname = gDpPath + "/" + bucket + "/" + itemlocation + "/" + tagdetail

	if true == isFileExists(filepathname) {
		return
	}

	//AWS_SECRET_ACCESS_KEY = Env("AWS_SECRET_ACCESS_KEY", false)
	//AWS_ACCESS_KEY_ID = Env("AWS_ACCESS_KEY_ID", false)
	AWS_REGION = Env("AWS_REGION", false)
	file, err := os.Create(filepathname)
	if err != nil {
		log.Error("Failed to create file", err)
		return ""
	}
	defer file.Close()

	downloader := s3manager.NewDownloader(session.New(&aws.Config{Region: aws.String(AWS_REGION)}))
	numBytes, err := downloader.Download(file,
		&s3aws.GetObjectInput{
			Bucket: aws.String(bucket),
			Key:    aws.String( /*dpname + "/" + */ itemlocation + "/" + tagdetail),
		})
	if err != nil {
		log.Info("Failed to download file.", err)
		os.Remove(filepathname)
		return
	}

	log.Println("Downloaded file", file.Name(), numBytes, "bytes")

	return
}

func checkHealth(errorTagsMap *map[string]string) {

	localfiles := make([]string, 0)
	alllocalfiles := make([]string, 0)
	localfilepath := GetLocalfilePath()

	for _, path := range localfilepath {
		localfiles = ScanLocalFile(path)
		for _, localfile := range localfiles {
			alllocalfiles = append(alllocalfiles, localfile)
		}
	}

	log.Info(alllocalfiles)
	var tagDetails map[string]string
	tagDetails = make(map[string]string)

	err := GetAllTagDetails(&tagDetails)
	if err != nil {
		log.Error(err)
	}

	var i int
	for file, tag := range tagDetails {
		for i = 0; i < len(alllocalfiles); i++ {
			//log.Info("--------->tag:", tag)
			//log.Info("--------->tagfile:",file)
			//log.Info("--------->localfile:",localfiles[i])
			if file == alllocalfiles[i] {
				break
			}
		}
		if i >= len(alllocalfiles) {
			(*errorTagsMap)[file] = tag
		}
	}

	//for errortagfile, errortag := range *errorTagsMap {
	//	log.Info("------->errortag:", errortag, "-------->", errortagfile)
	//}
}

func judgeRepoOrItemExist(repository, dataitem string) (exist bool, msg string, err error) {

	path := "/api/repositories/" + repository + "/" + dataitem

	exist = false

	resp, err := commToServerGetRsp("get", path, nil)
	if err != nil {
		log.Error(err)
		//HttpNoData(w, http.StatusInternalServerError, cmd.ErrorServiceUnavailable, err.Error())
		return
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusBadRequest {
		err = errors.New("unkown error")
		return
	}
	result := ds.Response{}
	respbody, _ := ioutil.ReadAll(resp.Body)
	unmarshalerr := json.Unmarshal(respbody, &result)
	if unmarshalerr != nil {
		log.Error(unmarshalerr)
		//HttpNoData(w, http.StatusInternalServerError, cmd.ErrorUnmarshal, "error while unmarshal respBody")
		return
	}
	log.Info(string(respbody))

	if resp.StatusCode == http.StatusBadRequest && result.Code == cmd.ServerErrResultCode1009 {
		//HttpNoData(w, http.StatusBadRequest, cmd.RepoOrItemNotExist, result.Msg)
		msg = result.Msg
		return
	} else if resp.StatusCode == http.StatusOK && result.Code == cmd.ServerErrResultCodeOk {
		exist = true
		msg = result.Msg
		return
	}

	return
}

func judgeTagExist(repository, dataitem, tag string) (exist bool, msg string, err error) {

	path := "/api/repositories/" + repository + "/" + dataitem + "/" + tag

	exist = false

	resp, err := commToServerGetRsp("get", path, nil)
	defer resp.Body.Close()
	if err != nil {
		log.Error(err)
		return
	}

	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusBadRequest {
		err = errors.New("unkown error")
		return
	}

	result := ds.Response{}
	respbody, _ := ioutil.ReadAll(resp.Body)
	unmarshalerr := json.Unmarshal(respbody, &result)
	if unmarshalerr != nil {
		log.Error(unmarshalerr)
		return
	}
	log.Info(string(respbody))

	if resp.StatusCode == http.StatusBadRequest && result.Code == cmd.ServerErrResultCode1009 {
		msg = result.Msg
		return
	} else if resp.StatusCode == http.StatusOK && result.Code == cmd.ServerErrResultCodeOk {
		exist = true
		msg = result.Msg
		return
	}

	return
}

func repoDelTagHandler(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {

	if len(loginAuthStr) == 0 {
		HttpNoData(w, http.StatusUnauthorized, cmd.ErrorServiceUnavailable, " ")
		return
	}

	repository := ps.ByName("repo")
	dataitem := ps.ByName("item")
	tag := ps.ByName("tag")

	reqBody, _ := ioutil.ReadAll(r.Body)

	if strings.Contains(tag, "*") {
		tagsname, err := getBatchDelTagsName(repository, dataitem, tag)
		if err != nil {
			log.Error(err)
			HttpNoData(w, http.StatusInternalServerError, cmd.ErrorSqlExec, "error while delete tag.")
			return
		}
		if len(tagsname) == 0 {
			log.Println("没有匹配的tag")
			HttpNoData(w, http.StatusInternalServerError, cmd.ErrorSqlExec, "No match tag.")
			return
		}
		successflag := true
		for _, tagname := range tagsname {
			if successflag {
				_, err := delTag(repository, dataitem, tagname)
				if err != nil {
					log.Error(err)
					HttpNoData(w, http.StatusInternalServerError, cmd.ErrorSqlExec, "error while delete tag")
					return
				}

				path := "/api/repositories/" + repository + "/" + dataitem + "/" + tagname
				resp, err := commToServerGetRsp("delete", path, reqBody)
				if err != nil {
					log.Error(err)
					HttpNoData(w, resp.StatusCode, cmd.ErrorServiceUnavailable, "commToServer error")
					return
				}
				defer resp.Body.Close()

				result := ds.Response{}

				respbody, err := ioutil.ReadAll(resp.Body)

				unmarshalerr := json.Unmarshal(respbody, &result)
				if unmarshalerr != nil {
					log.Error(unmarshalerr)
					HttpNoData(w, http.StatusInternalServerError, cmd.ErrorUnmarshal, "error while unmarshal respBody")
					return
				}
				if resp.StatusCode == http.StatusOK && result.Code == 0 {
					continue
				} else if resp.StatusCode == http.StatusOK && result.Code != 0 {
					HttpNoData(w, resp.StatusCode, result.Code, result.Msg)
					log.Info("Error :", result.Msg, "ResultCode:", result.Code, "HttpCode :", resp.StatusCode)
					successflag = false
					break
				} else {
					HttpNoData(w, resp.StatusCode, result.Code, result.Msg)
					log.Info("Error :", result.Msg, "ResultCode:", result.Code, "HttpCode :", resp.StatusCode)
					successflag = false
					break
				}
			}

		}
		if successflag {
			log.Info("批量删除tag成功")
			HttpNoData(w, http.StatusOK, cmd.ResultOK, "ok.")
		}
	} else {
		tagid, err := delTag(repository, dataitem, tag)
		if err != nil {
			log.Error(err)
			HttpNoData(w, http.StatusInternalServerError, cmd.ErrorSqlExec, "error while delete tag")
			return
		}

		reqBody, _ := ioutil.ReadAll(r.Body)
		path := "/api/repositories/" + repository + "/" + dataitem + "/" + tag
		resp, err := commToServerGetRsp("delete", path, reqBody)
		if err != nil {
			log.Error(err)
			HttpNoData(w, resp.StatusCode, cmd.ErrorServiceUnavailable, "commToServer error")
			return
		}
		defer resp.Body.Close()

		result := ds.Response{}

		respbody, err := ioutil.ReadAll(resp.Body)

		unmarshalerr := json.Unmarshal(respbody, &result)
		if unmarshalerr != nil {
			log.Error(unmarshalerr)
			HttpNoData(w, http.StatusInternalServerError, cmd.ErrorUnmarshal, "error while unmarshal respBody")
//.........这里部分代码省略.........