尝试修复无法上传, #897

baidupcs/expires/cachemap/cachemap.go

@@ -16,15 +16,14 @@ type (
 )
 
 func (cm *CacheOpMap) LazyInitCachePoolOp(op string) CacheUnit {
-	cacheItf, ok := cm.cachePool.Load(op)
-	if !ok {
-		cache := &cacheUnit{}
-		cm.cachePool.Store(op, cache)
-		return cache
-	}
+	cacheItf, _ := cm.cachePool.LoadOrStore(op, &cacheUnit{})
 	return cacheItf.(CacheUnit)
 }
 
+func (cm *CacheOpMap) RemoveCachePoolOp(op string) {
+	cm.cachePool.Delete(op)
+}
+
 // ClearInvalidate 清除已过期的数据(一般用不到)
 func (cm *CacheOpMap) ClearInvalidate() {
 	cm.cachePool.Range(func(_, cacheItf interface{}) bool {
@@ -41,5 +40,4 @@ func (cm *CacheOpMap) ClearInvalidate() {
 
 // PrintAll 输出所有缓冲项目
 func (cm *CacheOpMap) PrintAll() {
-
 }

baidupcs/expires/cachemap/cachemap_test.go

@@ -4,6 +4,8 @@ import (
 	"fmt"
 	"github.com/iikira/BaiduPCS-Go/baidupcs/expires"
 	"github.com/iikira/BaiduPCS-Go/baidupcs/expires/cachemap"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 )
@@ -13,12 +15,13 @@ func TestCacheMapDataExpires(t *testing.T) {
 	cache := cm.LazyInitCachePoolOp("op")
 	cache.Store("key_1", expires.NewDataExpires("value_1", 1*time.Second))
 
-	time.Sleep(1 * time.Second)
+	time.Sleep(2 * time.Second)
 	data, ok := cache.Load("key_1")
-	if !ok {
+	if ok {
+		fmt.Printf("data: %s\n", data.Data())
+		// 超时仍能读取到数据, 失败
 		t.FailNow()
 	}
-	fmt.Printf("data: %s\n", data.Data())
 }
 
 func TestCacheOperation(t *testing.T) {
@@ -36,3 +39,42 @@ func TestCacheOperation(t *testing.T) {
 	}
 	fmt.Printf("data: %s\n", data.Data())
 }
+
+func TestCacheOperation_LockKey(t *testing.T) {
+	cm := cachemap.CacheOpMap{}
+	wg := sync.WaitGroup{}
+	wg.Add(5000)
+
+	var (
+		execTimes1 int32 = 0 // 执行次数1
+		execTimes2 int32 = 0 // 执行次数2
+	)
+
+	for i := 0; i < 5000; i++ {
+		go func(i int) {
+			defer wg.Done()
+			cm.CacheOperation("op", "key_1", func() expires.DataExpires {
+				time.Sleep(50 * time.Microsecond) // 一些耗时的操作
+				atomic.AddInt32(&execTimes1, 1)
+				return expires.NewDataExpires(fmt.Sprintf("value_1: %d", i), 10*time.Second)
+			})
+
+			cm.CacheOperation("op", "key_2", func() expires.DataExpires {
+				time.Sleep(50 * time.Microsecond) // 一些耗时的操作
+				atomic.AddInt32(&execTimes2, 1)
+				return expires.NewDataExpires(fmt.Sprintf("value_2: %d", i), 10*time.Second)
+			})
+		}(i)
+	}
+	wg.Wait()
+
+	// 执行次数应为1
+	if execTimes1 != 1 {
+		fmt.Printf("execTimes1: %d\n", execTimes1)
+		t.FailNow()
+	}
+	if execTimes2 != 1 {
+		fmt.Printf("execTimes2: %d\n", execTimes2)
+		t.FailNow()
+	}
+}

baidupcs/expires/cachemap/cacheunit.go

@@ -12,15 +12,19 @@ type (
 		LoadOrStore(key interface{}, value expires.DataExpires) (actual expires.DataExpires, loaded bool)
 		Range(f func(key interface{}, value expires.DataExpires) bool)
 		Store(key interface{}, value expires.DataExpires)
+		LockKey(key interface{})
+		UnlockKey(key interface{})
 	}
 
 	cacheUnit struct {
-		unit sync.Map
+		unit   sync.Map
+		keyMap sync.Map
 	}
 )
 
 func (cu *cacheUnit) Delete(key interface{}) {
 	cu.unit.Delete(key)
+	cu.keyMap.Delete(key)
 }
 
 func (cu *cacheUnit) Load(key interface{}) (value expires.DataExpires, ok bool) {
@@ -63,3 +67,15 @@ func (cu *cacheUnit) Store(key interface{}, value expires.DataExpires) {
 	}
 	cu.unit.Store(key, value)
 }
+
+func (cu *cacheUnit) LockKey(key interface{}) {
+	muItf, _ := cu.keyMap.LoadOrStore(key, &sync.Mutex{})
+	mu := muItf.(*sync.Mutex)
+	mu.Lock()
+}
+
+func (cu *cacheUnit) UnlockKey(key interface{}) {
+	muItf, _ := cu.keyMap.LoadOrStore(key, &sync.Mutex{})
+	mu := muItf.(*sync.Mutex)
+	mu.Unlock()
+}

baidupcs/expires/cachemap/utils.go

@@ -4,11 +4,19 @@ import (
 	"github.com/iikira/BaiduPCS-Go/baidupcs/expires"
 )
 
-func (cm *CacheOpMap) CacheOperation(op string, key interface{}, opFunc func() expires.DataExpires) (data expires.DataExpires) {
+type (
+	OpFunc          func() expires.DataExpires
+	OpFuncWithError func() (expires.DataExpires, error)
+)
+
+func (cm *CacheOpMap) CacheOperation(op string, key interface{}, opFunc OpFunc) (data expires.DataExpires) {
 	var (
 		cache = cm.LazyInitCachePoolOp(op)
 		ok    bool
 	)
+
+	cache.LockKey(key)
+	defer cache.UnlockKey(key)
 	data, ok = cache.Load(key)
 	if !ok {
 		data = opFunc()
@@ -21,17 +29,24 @@ func (cm *CacheOpMap) CacheOperation(op string, key interface{}, opFunc func() e
 	return
 }
 
-func (cm *CacheOpMap) CacheOperationWithError(op string, key interface{}, opFunc func() (expires.DataExpires, error)) (data expires.DataExpires, err error) {
+func (cm *CacheOpMap) CacheOperationWithError(op string, key interface{}, opFunc OpFuncWithError) (data expires.DataExpires, err error) {
 	var (
 		cache = cm.LazyInitCachePoolOp(op)
 		ok    bool
 	)
+
+	cache.LockKey(key)
+	defer cache.UnlockKey(key)
 	data, ok = cache.Load(key)
 	if !ok {
 		data, err = opFunc()
 		if err != nil {
 			return
 		}
+		if data == nil {
+			// 数据为空时也不存
+			return
+		}
 		cache.Store(key, data)
 	}
 

baidupcs/expires/expires.go

@@ -21,8 +21,10 @@ type (
 )
 
 // StripMono strip monotonic clocks
-//go:linkname StripMono time.(*Time).stripMono
-func StripMono(t *time.Time)
+// go:linkname StripMono time.(*Time).stripMono
+func StripMono(t *time.Time) {
+	t.Round(0)
+}
 
 func NewExpires(dur time.Duration) Expires {
 	t := time.Now().Add(dur)

baidupcs/prepare.go

@@ -407,8 +407,17 @@ func (pcs *BaiduPCS) PrepareUploadTmpFile(uploadFunc UploadFunc) (dataReadCloser
 }
 
 // PrepareUploadCreateSuperFile 分片上传—合并分片文件, 只返回服务器响应数据和错误信息
-func (pcs *BaiduPCS) PrepareUploadCreateSuperFile(targetPath string, blockList ...string) (dataReadCloser io.ReadCloser, pcsError pcserror.Error) {
+func (pcs *BaiduPCS) PrepareUploadCreateSuperFile(checkDir bool, targetPath string, blockList ...string) (dataReadCloser io.ReadCloser, pcsError pcserror.Error) {
 	pcs.lazyInit()
+
+	if checkDir {
+		// 检查是否为目录
+		pcsError = pcs.checkIsdir(OperationUploadCreateSuperFile, targetPath)
+		if pcsError != nil {
+			return nil, pcsError
+		}
+	}
+
 	bl := BlockListJSON{
 		BlockList: blockList,
 	}
@@ -420,7 +429,7 @@ func (pcs *BaiduPCS) PrepareUploadCreateSuperFile(targetPath string, blockList .
 
 	pcsURL := pcs.generatePCSURL("file", "createsuperfile", map[string]string{
 		"path":  targetPath,
-		"ondup": "newcopy",
+		"ondup": "overwrite",
 	})
 	baiduPCSVerbose.Infof("%s URL: %s\n", OperationUploadCreateSuperFile, pcsURL)
 

baidupcs/upload.go

@@ -178,8 +178,8 @@ func (pcs *BaiduPCS) UploadTmpFile(uploadFunc UploadFunc) (md5 string, pcsError
 }
 
 // UploadCreateSuperFile 分片上传—合并分片文件
-func (pcs *BaiduPCS) UploadCreateSuperFile(targetPath string, blockList ...string) (pcsError pcserror.Error) {
-	dataReadCloser, pcsError := pcs.PrepareUploadCreateSuperFile(targetPath, blockList...)
+func (pcs *BaiduPCS) UploadCreateSuperFile(checkDir bool, targetPath string, blockList ...string) (pcsError pcserror.Error) {
+	dataReadCloser, pcsError := pcs.PrepareUploadCreateSuperFile(checkDir, targetPath, blockList...)
 	if pcsError != nil {
 		return pcsError
 	}

go.mod

@@ -3,6 +3,7 @@ module github.com/iikira/BaiduPCS-Go
 go 1.12
 
 require (
+	github.com/GeertJohan/go.incremental v1.0.0
 	github.com/GeertJohan/go.rice v0.0.0-20170420135705-c02ca9a983da // indirect
 	github.com/astaxie/beego v1.10.1 // indirect
 	github.com/bitly/go-simplejson v0.5.0 // indirect

go.sum

@@ -1,4 +1,5 @@
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+github.com/GeertJohan/go.incremental v1.0.0 h1:7AH+pY1XUgQE4Y1HcXYaMqAI0m9yrFqo/jt0CW30vsg=
 github.com/GeertJohan/go.incremental v1.0.0/go.mod h1:6fAjUhbVuX1KcMD3c8TEgVUqmo4seqhv0i0kdATSkM0=
 github.com/GeertJohan/go.rice v0.0.0-20170420135705-c02ca9a983da h1:UVU3a9pRUyLdnBtn60WjRl0s4SEyJc2ChCY56OAR6wI=
 github.com/GeertJohan/go.rice v0.0.0-20170420135705-c02ca9a983da/go.mod h1:DgrzXonpdQbfN3uYaGz1EG4Sbhyum/MMIn6Cphlh2bw=