chunked file works now

weed/filesys/dirty_pages_temp_file.go

@@ -12,22 +12,21 @@ import (
 )
 
 type TempFileDirtyPages struct {
-	f                *File
-	tf               *os.File
-	writtenIntervals *page_writer.WrittenContinuousIntervals
-	writeWaitGroup   sync.WaitGroup
-	pageAddLock      sync.Mutex
-	chunkAddLock     sync.Mutex
-	lastErr          error
-	collection       string
-	replication      string
+	f              *File
+	writeWaitGroup sync.WaitGroup
+	pageAddLock    sync.Mutex
+	chunkAddLock   sync.Mutex
+	lastErr        error
+	collection     string
+	replication    string
+	chunkedFile    *page_writer.ChunkedFileWriter
 }
 
-func newTempFileDirtyPages(file *File) *TempFileDirtyPages {
+func newTempFileDirtyPages(file *File, chunkSize int64) *TempFileDirtyPages {
 
 	tempFile := &TempFileDirtyPages{
-		f:                file,
-		writtenIntervals: &page_writer.WrittenContinuousIntervals{},
+		f:           file,
+		chunkedFile: page_writer.NewChunkedFileWriter(file.wfs.option.getTempFilePageDir(), chunkSize),
 	}
 
 	return tempFile
@@ -38,28 +37,8 @@ func (pages *TempFileDirtyPages) AddPage(offset int64, data []byte) {
 	pages.pageAddLock.Lock()
 	defer pages.pageAddLock.Unlock()
 
-	if pages.tf == nil {
-		tf, err := os.CreateTemp(pages.f.wfs.option.getTempFilePageDir(), "")
-		if err != nil {
-			glog.Errorf("create temp file: %v", err)
-			pages.lastErr = err
-			return
-		}
-		pages.tf = tf
-		pages.writtenIntervals.TempFile = tf
-		pages.writtenIntervals.LastOffset = 0
-	}
-
-	writtenOffset := pages.writtenIntervals.LastOffset
-	dataSize := int64(len(data))
-
-	// glog.V(4).Infof("%s AddPage %v at %d [%d,%d)", pages.f.fullpath(), pages.tf.Name(), writtenOffset, offset, offset+dataSize)
-
-	if _, err := pages.tf.WriteAt(data, writtenOffset); err != nil {
+	if _, err := pages.chunkedFile.WriteAt(data, offset); err != nil {
 		pages.lastErr = err
-	} else {
-		pages.writtenIntervals.AddInterval(writtenOffset, len(data), offset)
-		pages.writtenIntervals.LastOffset += dataSize
 	}
 
 	// pages.writtenIntervals.debug()
@@ -68,54 +47,38 @@ func (pages *TempFileDirtyPages) AddPage(offset int64, data []byte) {
 }
 
 func (pages *TempFileDirtyPages) FlushData() error {
-
-	pages.saveExistingPagesToStorage()
+	pages.saveChunkedFileToStorage()
 	pages.writeWaitGroup.Wait()
 	if pages.lastErr != nil {
 		return fmt.Errorf("flush data: %v", pages.lastErr)
 	}
-	pages.pageAddLock.Lock()
-	defer pages.pageAddLock.Unlock()
-	if pages.tf != nil {
-
-		pages.writtenIntervals.TempFile = nil
-		pages.writtenIntervals.Lists = nil
-
-		pages.tf.Close()
-		os.Remove(pages.tf.Name())
-		pages.tf = nil
-	}
 	return nil
 }
 
-func (pages *TempFileDirtyPages) saveExistingPagesToStorage() {
+func (pages *TempFileDirtyPages) ReadDirtyDataAt(data []byte, startOffset int64) (maxStop int64) {
+	return pages.chunkedFile.ReadDataAt(data, startOffset)
+}
 
-	pageSize := pages.f.wfs.option.ChunkSizeLimit
+func (pages *TempFileDirtyPages) GetStorageOptions() (collection, replication string) {
+	return pages.collection, pages.replication
+}
 
-	// glog.V(4).Infof("%v saveExistingPagesToStorage %d lists", pages.f.Name, len(pages.writtenIntervals.lists))
+func (pages *TempFileDirtyPages) saveChunkedFileToStorage() {
 
-	for _, list := range pages.writtenIntervals.Lists {
-		listStopOffset := list.Offset() + list.Size()
-		for uploadedOffset := int64(0); uploadedOffset < listStopOffset; uploadedOffset += pageSize {
-			start, stop := max(list.Offset(), uploadedOffset), min(listStopOffset, uploadedOffset+pageSize)
-			if start >= stop {
-				continue
-			}
-			// glog.V(4).Infof("uploading %v [%d,%d) %d/%d", pages.f.Name, start, stop, i, len(pages.writtenIntervals.lists))
-			pages.saveToStorage(list.ToReader(start, stop), start, stop-start)
-		}
-	}
+	pages.chunkedFile.ProcessEachInterval(func(file *os.File, logicChunkIndex int, interval *page_writer.PageChunkWrittenInterval) {
+		reader := page_writer.NewFileIntervalReader(pages.chunkedFile, logicChunkIndex, interval)
+		pages.saveChunkedFileIntevalToStorage(reader, int64(logicChunkIndex)*pages.chunkedFile.ChunkSize, interval.Size())
+	})
 
 }
 
-func (pages *TempFileDirtyPages) saveToStorage(reader io.Reader, offset int64, size int64) {
+func (pages *TempFileDirtyPages) saveChunkedFileIntevalToStorage(reader io.Reader, offset int64, size int64) {
 
 	mtime := time.Now().UnixNano()
 	pages.writeWaitGroup.Add(1)
 	writer := func() {
 		defer pages.writeWaitGroup.Done()
 
-		reader = io.LimitReader(reader, size)
 		chunk, collection, replication, err := pages.f.wfs.saveDataAsChunk(pages.f.fullpath())(reader, pages.f.Name, offset)
 		if err != nil {
 			glog.V(0).Infof("%s saveToStorage [%d,%d): %v", pages.f.fullpath(), offset, offset+size, err)
@@ -135,12 +98,9 @@ func (pages *TempFileDirtyPages) saveToStorage(reader io.Reader, offset int64, s
 	} else {
 		go writer()
 	}
+
 }
 
-func (pages *TempFileDirtyPages) ReadDirtyDataAt(data []byte, startOffset int64) (maxStop int64) {
-	return pages.writtenIntervals.ReadDataAt(data, startOffset)
-}
-
-func (pages *TempFileDirtyPages) GetStorageOptions() (collection, replication string) {
-	return pages.collection, pages.replication
+func (pages TempFileDirtyPages) Destroy() {
+	pages.chunkedFile.Destroy()
 }