🎨 WebDAV/S3 data sync and backup support configurable concurrent requ…

…ests siyuan-note/siyuan#12798
siyuan-note · Oct 16, 2024 · 5fc0357 · 5fc0357
1 parent 7d61fd4
commit 5fc0357
Show file tree

Hide file tree

Showing 5 changed files with 57 additions and 20 deletions.
diff --git a/cloud/cloud.go b/cloud/cloud.go
@@ -47,23 +47,25 @@ type Conf struct {
 
 // ConfS3 用于描述 S3 对象存储协议所需配置。
 type ConfS3 struct {
-	Endpoint      string // 服务端点
-	AccessKey     string // Access Key
-	SecretKey     string // Secret Key
-	Region        string // 存储区域
-	Bucket        string // 存储空间
-	PathStyle     bool   // 是否使用路径风格寻址
-	SkipTlsVerify bool   //  是否跳过 TLS 验证
-	Timeout       int    // 超时时间，单位：秒
+	Endpoint       string // 服务端点
+	AccessKey      string // Access Key
+	SecretKey      string // Secret Key
+	Region         string // 存储区域
+	Bucket         string // 存储空间
+	PathStyle      bool   // 是否使用路径风格寻址
+	SkipTlsVerify  bool   //  是否跳过 TLS 验证
+	Timeout        int    // 超时时间，单位：秒
+	ConcurrentReqs int    // 并发请求数
 }
 
 // ConfWebDAV 用于描述 WebDAV 协议所需配置。
 type ConfWebDAV struct {
-	Endpoint      string // 服务端点
-	Username      string // 用户名
-	Password      string // 密码
-	SkipTlsVerify bool   // 是否跳过 TLS 验证
-	Timeout       int    // 超时时间，单位：秒
+	Endpoint       string // 服务端点
+	Username       string // 用户名
+	Password       string // 密码
+	SkipTlsVerify  bool   // 是否跳过 TLS 验证
+	Timeout        int    // 超时时间，单位：秒
+	ConcurrentReqs int    // 并发请求数
 }
 
 // Cloud 描述了云端存储服务，接入云端存储服务时需要实现该接口。
@@ -116,6 +118,9 @@ type Cloud interface {
 
 	// GetIndex 用于获取索引。
 	GetIndex(id string) (index *entity.Index, err error)
+
+	// GetConcurrentReqs 用于获取配置的并发请求数。
+	GetConcurrentReqs() int
 }
 
 // Traffic 描述了流量信息。
@@ -250,6 +255,10 @@ func (baseCloud *BaseCloud) GetIndex(id string) (index *entity.Index, err error)
 	return
 }
 
+func (baseCloud *BaseCloud) GetConcurrentReqs() int {
+	return 8
+}
+
 func (baseCloud *BaseCloud) GetConf() *Conf {
 	return baseCloud.Conf
 }

diff --git a/cloud/s3.go b/cloud/s3.go
@@ -262,6 +262,17 @@ func (s3 *S3) GetIndex(id string) (index *entity.Index, err error) {
 	return
 }
 
+func (s3 *S3) GetConcurrentReqs() (ret int) {
+	ret = s3.S3.ConcurrentReqs
+	if 1 > ret {
+		ret = 8
+	}
+	if 16 < ret {
+		ret = 16
+	}
+	return
+}
+
 func (s3 *S3) ListObjects(pathPrefix string) (ret map[string]*entity.ObjectInfo, err error) {
 	ret = map[string]*entity.ObjectInfo{}
 	svc := s3.getService()
@@ -417,8 +428,14 @@ func (s3 *S3) getNotFound(keys []string) (ret []string, err error) {
 	if 1 > len(keys) {
 		return
 	}
+
+	poolSize := s3.GetConcurrentReqs()
+	if poolSize > len(keys) {
+		poolSize = len(keys)
+	}
+
 	waitGroup := &sync.WaitGroup{}
-	p, _ := ants.NewPoolWithFunc(8, func(arg interface{}) {
+	p, _ := ants.NewPoolWithFunc(poolSize, func(arg interface{}) {
 		defer waitGroup.Done()
 		key := arg.(string)
 		info, statErr := s3.statFile(key)

diff --git a/cloud/webdav.go b/cloud/webdav.go
@@ -237,6 +237,17 @@ func (webdav *WebDAV) GetIndex(id string) (index *entity.Index, err error) {
 	return
 }
 
+func (webdav *WebDAV) GetConcurrentReqs() (ret int) {
+	ret = webdav.Conf.WebDAV.ConcurrentReqs
+	if 1 > ret {
+		ret = 1
+	}
+	if 16 < ret {
+		ret = 16
+	}
+	return
+}
+
 func (webdav *WebDAV) ListObjects(pathPrefix string) (ret map[string]*entity.ObjectInfo, err error) {
 	ret = map[string]*entity.ObjectInfo{}
 

diff --git a/repo.go b/repo.go
@@ -496,7 +496,7 @@ func (repo *Repo) OpenFile(file *entity.File) (ret []byte, err error) {
 func (repo *Repo) removeCloudObjects(objects []string) (err error) {
 	waitGroup := &sync.WaitGroup{}
 	var removeErr error
-	poolSize := 8
+	poolSize := repo.cloud.GetConcurrentReqs()
 	if poolSize > len(objects) {
 		poolSize = len(objects)
 	}

diff --git a/sync.go b/sync.go
@@ -754,7 +754,7 @@ func (repo *Repo) downloadCloudChunksPut(chunkIDs []string, context map[string]i
 
 	waitGroup := &sync.WaitGroup{}
 	var downloadErr error
-	poolSize := 8
+	poolSize := repo.cloud.GetConcurrentReqs()
 	if poolSize > len(chunkIDs) {
 		poolSize = len(chunkIDs)
 	}
@@ -814,7 +814,7 @@ func (repo *Repo) downloadCloudFilesPut(fileIDs []string, context map[string]int
 	lock := &sync.Mutex{}
 	waitGroup := &sync.WaitGroup{}
 	var downloadErr error
-	poolSize := 8
+	poolSize := repo.cloud.GetConcurrentReqs()
 	if poolSize > len(fileIDs) {
 		poolSize = len(fileIDs)
 	}
@@ -977,7 +977,7 @@ func (repo *Repo) uploadCloudMissingObjects(trafficStat *TrafficStat, context ma
 
 	waitGroup := &sync.WaitGroup{}
 	var uploadErr error
-	poolSize := 8
+	poolSize := repo.cloud.GetConcurrentReqs()
 	if poolSize > len(missingObjects) {
 		poolSize = len(missingObjects)
 	}
@@ -1180,7 +1180,7 @@ func (repo *Repo) uploadFiles(upsertFiles []*entity.File, context map[string]int
 
 	waitGroup := &sync.WaitGroup{}
 	var uploadErr error
-	poolSize := 8
+	poolSize := repo.cloud.GetConcurrentReqs()
 	if poolSize > len(upsertFiles) {
 		poolSize = len(upsertFiles)
 	}
@@ -1234,7 +1234,7 @@ func (repo *Repo) uploadChunks(upsertChunkIDs []string, context map[string]inter
 
 	waitGroup := &sync.WaitGroup{}
 	var uploadErr error
-	poolSize := 8
+	poolSize := repo.cloud.GetConcurrentReqs()
 	if poolSize > len(upsertChunkIDs) {
 		poolSize = len(upsertChunkIDs)
 	}