Merge branch 'main' into feat-obj-ctx

Author: davies

pkg/meta/utils.go

@@ -370,11 +370,7 @@ func (m *baseMeta) GetSummary(ctx Context, inode Ino, summary *Summary, recursiv
 		return st
 	}
 	if attr.Typ != TypeDirectory {
-		if attr.Typ == TypeDirectory {
-			summary.Dirs++
-		} else {
-			summary.Files++
-		}
+		summary.Files++
 		summary.Size += uint64(align4K(attr.Length))
 		if attr.Typ == TypeFile {
 			summary.Length += attr.Length

pkg/object/interface.go

@@ -94,7 +94,7 @@ type ObjectStorage interface {
 	Head(ctx context.Context, key string) (Object, error)
 	// List returns a list of objects using ListObjectV2.
 	List(ctx context.Context, prefix, startAfter, token, delimiter string, limit int64, followLink bool) ([]Object, bool, string, error)
-	// ListAll returns all the objects as an channel.
+	// ListAll returns all the objects as a channel.
 	ListAll(ctx context.Context, prefix, marker string, followLink bool) (<-chan Object, error)
 
 	// CreateMultipartUpload starts to upload a large object part by part.
@@ -105,7 +105,7 @@ type ObjectStorage interface {
 	UploadPartCopy(ctx context.Context, key string, uploadID string, num int, srcKey string, off, size int64) (*Part, error)
 	// AbortUpload abort a multipart upload.
 	AbortUpload(ctx context.Context, key string, uploadID string)
-	// CompleteUpload finish an multipart upload.
+	// CompleteUpload finish a multipart upload.
 	CompleteUpload(ctx context.Context, key string, uploadID string, parts []*Part) error
 	// ListUploads lists existing multipart uploads.
 	ListUploads(ctx context.Context, marker string) ([]*PendingPart, string, error)

pkg/object/minio.go

@@ -35,11 +35,17 @@ import (
 
 type minio struct {
 	s3client
-	endpoint string
 }
 
 func (m *minio) String() string {
-	return fmt.Sprintf("minio://%s/%s/", m.endpoint, m.s3client.bucket)
+	if m.s3.Options().BaseEndpoint != nil {
+		endpoint := *m.s3.Options().BaseEndpoint
+		if idx := strings.Index(endpoint, "://"); idx >= 0 {
+			endpoint = endpoint[idx+3:]
+		}
+		return fmt.Sprintf("minio://%s/%s/", endpoint, m.bucket)
+	}
+	return fmt.Sprintf("minio://%s/", m.bucket)
 }
 
 func (m *minio) Limits() Limits {
@@ -103,7 +109,7 @@ func newMinio(endpoint, accessKey, secretKey, token string) (ObjectStorage, erro
 		bucket = bucket[len("minio/"):]
 	}
 	bucket = strings.Split(bucket, "/")[0]
-	return &minio{s3client{bucket: bucket, s3: client, region: region}, endpoint}, nil
+	return &minio{s3client{bucket: bucket, s3: client, region: region}}, nil
 }
 
 func init() {

pkg/object/oss.go

@@ -83,7 +83,7 @@ func (o *ossClient) Head(ctx context.Context, key string) (Object, error) {
 	})
 	if err != nil {
 		var svcErr *oss.ServiceError
-		if errors.As(err, &svcErr); svcErr.StatusCode == http.StatusNotFound {
+		if errors.As(err, &svcErr) && svcErr.StatusCode == http.StatusNotFound {
 			err = os.ErrNotExist
 		}
 		return nil, err

pkg/vfs/reader.go

@@ -51,7 +51,7 @@ const (
 
 const readSessions = 2
 
-var readBufferUsed int64
+var readBufferUsed atomic.Int64
 
 type sstate uint8
 
@@ -267,7 +267,7 @@ func (s *sliceReader) delete() {
 	} else {
 		s.file.last = s.prev
 	}
-	atomic.AddInt64(&readBufferUsed, -int64(cap(s.page.Data)))
+	readBufferUsed.Add(-int64(cap(s.page.Data)))
 	s.page.Release()
 }
 
@@ -325,7 +325,7 @@ func (f *fileReader) newSlice(block *frange) *sliceReader {
 	*(f.last) = s
 	f.last = &(s.next)
 	go s.run()
-	atomic.AddInt64(&readBufferUsed, int64(cap(s.page.Data)))
+	readBufferUsed.Add(int64(cap(s.page.Data)))
 	return s
 }
 
@@ -417,7 +417,7 @@ func (f *fileReader) checkReadahead(block *frange) int {
 	ses := &f.sessions[idx]
 	seqdata := ses.total
 	readahead := ses.readahead
-	used := uint64(atomic.LoadInt64(&readBufferUsed))
+	used := uint64(readBufferUsed.Load())
 	if readahead == 0 && f.r.blockSize <= f.r.readAheadMax && (block.off == 0 || seqdata > block.len) { // begin with read-ahead turned on
 		ses.readahead = f.r.blockSize
 	} else if readahead < f.r.readAheadMax && seqdata >= readahead && f.r.readAheadTotal > used+readahead*4 {
@@ -482,7 +482,7 @@ func (f *fileReader) releaseIdleBuffer() {
 	defer f.Unlock()
 	now := time.Now()
 	var idle = time.Minute
-	used := atomic.LoadInt64(&readBufferUsed)
+	used := readBufferUsed.Load()
 	if used > int64(f.r.readAheadTotal) {
 		idle /= time.Duration(used / int64(f.r.readAheadTotal))
 	}
@@ -538,7 +538,7 @@ func (f *fileReader) readAhead(block *frange) {
 		}
 		return true
 	})
-	if block.len > 0 && block.off < f.length && uint64(atomic.LoadInt64(&readBufferUsed)) < f.r.readAheadTotal {
+	if block.len > 0 && block.off < f.length && uint64(readBufferUsed.Load()) < f.r.readAheadTotal {
 		if block.len < f.r.blockSize {
 			block.len += f.r.blockSize - block.end()%f.r.blockSize // align to end of a block
 		}
@@ -724,7 +724,7 @@ func NewDataReader(conf *Config, m meta.Meta, store chunk.ChunkStore) DataReader
 }
 
 func (r *dataReader) readBufferUsed() int64 {
-	used := atomic.LoadInt64(&readBufferUsed)
+	used := readBufferUsed.Load()
 	return used
 }
 

pkg/vfs/writer.go

@@ -183,7 +183,7 @@ func (c *chunkWriter) commitThread() {
 	f := c.file
 	defer f.w.free(f)
 	f.Lock()
-	defer f.Unlock()
+
 	// the slices should be committed in the order that are created
 	for len(c.slices) > 0 {
 		s := c.slices[0]
@@ -218,6 +218,7 @@ func (c *chunkWriter) commitThread() {
 		c.slices = c.slices[1:]
 	}
 	f.freeChunk(c)
+	f.Unlock()
 }
 
 type fileWriter struct {

sdk/python/juicefs/juicefs/juicefs.py

@@ -35,7 +35,12 @@
 XATTR_REPLACE = 2
 
 def check_error(r, fn, args):
-    if r < 0:
+    if fn.__name__ == "jfs_init" and r == 0:
+        name = args[0].decode()
+        e = OSError(f'JuiceFS initialized failed for {name}')
+        e.errno = 1
+        raise e
+    elif r < 0:
         formatted_args = []
         for arg in args[2:]:
             if isinstance(arg, (bytes, bytearray)) and len(arg) > 1024:
@@ -350,7 +355,7 @@ def removexattr(self, path, name):
         self.lib.jfs_removeXattr(c_int64(_tid()), c_int64(self.h), _bin(path), _bin(name))
 
     def clone(self, src, dst, preserve=False):
-        """Clone a file."""
+        """Clone a file or directory."""
         self.lib.jfs_clone(c_int64(_tid()), c_int64(self.h), _bin(src), _bin(dst), c_bool(preserve))
 
     def set_quota(self, path, capacity=0, inodes=0, create=False, strict=False):