save

diskcache/diskcache.go

@@ -28,7 +28,7 @@ const (
 	dataHeaderLen = 4
 
 	// EOFHint labels a file's end.
-	EOFHint = uint32(0xdeadbeef)
+	EOFHint = uint32(0xdeadbeef) // Deprecated
 )
 
 // Generic diskcache errors.

diskcache/get.go

@@ -15,13 +15,11 @@ import (
 // Fn is the handler to eat cache from diskcache.
 type Fn func([]byte) error
 
-func (c *DiskCache) skipBadFile() error {
-	defer func() {
-		droppedBatchVec.WithLabelValues(c.path, reasonBadDataFile).Inc()
-	}()
-
-	if err := c.removeCurrentReadingFile(); err != nil {
-		return fmt.Errorf("removeCurrentReadingFile: %w", err)
+func (c *DiskCache) switchNextFile() error {
+	if c.curReadfile != "" {
+		if err := c.removeCurrentReadingFile(); err != nil {
+			return fmt.Errorf("removeCurrentReadingFile: %w", err)
+		}
 	}
 
 	// clear .pos
@@ -32,15 +30,15 @@ func (c *DiskCache) skipBadFile() error {
 	}
 
 	// reopen next file to read
-	if err := c.switchNextFile(); err != nil {
-		return err
-	}
+	return c.doSwitchNextFile()
+}
 
-	// swith to next file: to skip the bad file.
-	if err := c.switchNextFile(); err != nil {
-		return fmt.Errorf("switch to next file on bad file: %w", err)
-	}
-	return nil
+func (c *DiskCache) skipBadFile() error {
+	defer func() {
+		droppedBatchVec.WithLabelValues(c.path, reasonBadDataFile).Inc()
+	}()
+
+	return c.switchNextFile()
 }
 
 // Get fetch new data from disk cache, then passing to fn
@@ -96,33 +94,25 @@ retry:
 
 	hdr := make([]byte, dataHeaderLen)
 	if n, err = c.rfd.Read(hdr); err != nil || n != dataHeaderLen {
+		//
+		// On bad datafile, just ignore and delete the file.
+		//
 		if err = c.skipBadFile(); err != nil {
 			return err
 		}
 
-		goto retry // read next new file
+		goto retry // read next new file to save another Get() calling.
 	}
 
+	// how many bytes of current data?
 	nbytes = int(binary.LittleEndian.Uint32(hdr[0:]))
 
 	if uint32(nbytes) == EOFHint { // EOF
-		if err = c.removeCurrentReadingFile(); err != nil {
-			return fmt.Errorf("removeCurrentReadingFile: %w", err)
-		}
-
-		// clear .pos
-		if !c.noPos {
-			if err = c.pos.reset(); err != nil {
-				return err
-			}
-		}
-
-		// reopen next file to read
-		if err = c.switchNextFile(); err != nil {
-			return err
+		if err := c.switchNextFile(); err != nil {
+			return fmt.Errorf("switchNextFile: %w", err)
 		}
 
-		goto retry // read next new file
+		goto retry // read next new file to save another Get() calling.
 	}
 
 	databuf := make([]byte, nbytes)

diskcache/get_test.go

@@ -23,34 +23,39 @@ func TestDropInvalidDataFile(t *T.T) {
 		c, err := Open(WithPath(p))
 		require.NoError(t, err)
 
-		// put some data
+		// put some data and rotate 10 datafiles
 		data := make([]byte, 100)
-		assert.NoError(t, c.Put(data))
-		assert.NoError(t, c.rotate())
-
-		// rotate 2 data file
-		assert.NoError(t, c.Put(data))
-		assert.NoError(t, c.rotate())
+		for i := 0; i < 10; i++ {
+			assert.NoError(t, c.Put(data))
+			assert.NoError(t, c.rotate())
 
-		// distroy 1st data file
-		assert.NoError(t, os.Truncate(c.dataFiles[0], 0))
+			// destroy the datafile
+			if i%2 == 0 {
+				assert.NoError(t, os.Truncate(c.dataFiles[i], 0))
+			}
+		}
 
-		t.Logf("%q", c.dataFiles)
+		assert.Len(t, c.datafiles, 10)
 
-		assert.NoError(t, c.Get(func(get []byte) error {
-			// switch to 2nd file
-			assert.Len(t, c.dataFiles, 1)
-			assert.Equal(t, data, get)
+		for {
+			err := c.Get(func(get []byte) error {
+				// switch to 2nd file
+				assert.Equal(t, data, get)
+				return nil
+			})
 
-			return nil
-		}))
+			if err != nil {
+				require.ErrorIs(t, err, ErrEOF)
+				break
+			}
+		}
 
 		reg := prometheus.NewRegistry()
 		register(reg)
 		mfs, err := reg.Gather()
 		require.NoError(t, err)
 
-		assert.Equalf(t, float64(1),
+		assert.Equalf(t, float64(5),
 			metrics.GetMetricOnLabels(mfs,
 				"diskcache_dropped_total",
 				c.path,

diskcache/options.go

@@ -69,9 +69,11 @@ func WithBatchSize(size int64) CacheOption {
 }
 
 // WithMaxDataSize set max single data size, default 32MB.
+// NOTE: the max size of single data is 0xdeadbeef(~ 3.47GB), if size
+// larger than 0xdeadbeef, reset to default value(32MB).
 func WithMaxDataSize(size int32) CacheOption {
 	return func(c *DiskCache) {
-		if size > 0 {
+		if size > 0 && uint32(size) < EOFHint /* about 3.47GB */ {
 			c.maxDataSize = size
 		}
 	}

diskcache/rotate.go

@@ -103,14 +103,14 @@ func (c *DiskCache) removeCurrentReadingFile() error {
 		c.rfd = nil
 	}
 
-	if fi, err := os.Stat(c.curReadfile); err == nil {
+	if fi, err := os.Stat(c.curReadfile); err == nil { // file exist
 		if fi.Size() > dataHeaderLen {
 			c.size -= (fi.Size() - dataHeaderLen) // EOF bytes do not counted in size
 		}
-	}
 
-	if err := os.Remove(c.curReadfile); err != nil {
-		return fmt.Errorf("removeCurrentReadingFile: %s: %w", c.curReadfile, err)
+		if err := os.Remove(c.curReadfile); err != nil {
+			return fmt.Errorf("removeCurrentReadingFile: %q: %w", c.curReadfile, err)
+		}
 	}
 
 	if len(c.dataFiles) > 0 {

diskcache/switch.go

@@ -55,7 +55,7 @@ func (c *DiskCache) loadUnfinishedFile() error {
 }
 
 // open next read file.
-func (c *DiskCache) switchNextFile() error {
+func (c *DiskCache) doSwitchNextFile() error {
 	c.rwlock.Lock()
 	defer c.rwlock.Unlock()