From 1f345e350dfff2246c39ea12c415de1de81d2b7b Mon Sep 17 00:00:00 2001 From: Chao Wang Date: Fri, 25 Oct 2024 14:45:52 -0700 Subject: [PATCH] WIP --- restore/go.mod | 3 + restore/pkg/restore/restore.go | 340 ++++++++++++++++++++++++++++ restore/pkg/restore/restore_test.go | 10 + 3 files changed, 353 insertions(+) create mode 100644 restore/go.mod create mode 100644 restore/pkg/restore/restore.go create mode 100644 restore/pkg/restore/restore_test.go diff --git a/restore/go.mod b/restore/go.mod new file mode 100644 index 00000000..0ea2b07c --- /dev/null +++ b/restore/go.mod @@ -0,0 +1,3 @@ +module restore + +go 1.21.11 diff --git a/restore/pkg/restore/restore.go b/restore/pkg/restore/restore.go new file mode 100644 index 00000000..19768ab7 --- /dev/null +++ b/restore/pkg/restore/restore.go @@ -0,0 +1,340 @@ +package restore + +/* +#cgo CFLAGS: -I${SRCDIR}/../../../utils/src -I${SRCDIR}/../../../kmod/src +#cgo LDFLAGS: -L${SRCDIR}/../../../utils/src -l:scoutfs_parallel_restore.a -lm + +#include +#include +#include +#include +#include "sparse.h" +#include "util.h" +#include "format.h" +#include "parallel_restore.h" + +// If there are any type conflicts, you might need to add: +// #include "kernel_types.h" +*/ +import "C" +import ( + "errors" + "fmt" + "sync" + "syscall" + "unsafe" +) + +const batchSize = 1000 +const bufSize = 4096 + +type WorkerWriter struct { + writer *C.struct_scoutfs_parallel_restore_writer + progressCh chan *ScoutfsParallelWriterProgress + fileCreated int64 + devFd int + buf []byte + wg *sync.WaitGroup +} + +type MasterWriter struct { + writer *C.struct_scoutfs_parallel_restore_writer + progressCh chan *ScoutfsParallelWriterProgress + workers []*WorkerWriter + wg sync.WaitGroup + slice *C.struct_scoutfs_parallel_restore_slice // Add slice field + progressWg sync.WaitGroup + devFd int + super *C.struct_scoutfs_super_block +} + +type ScoutfsParallelWriterProgress struct { + Progress *C.struct_scoutfs_parallel_restore_progress + Slice *C.struct_scoutfs_parallel_restore_slice +} + +func (m *MasterWriter) aggregateProgress() { + defer m.progressWg.Done() + for progress := range m.progressCh { + ret := C.scoutfs_parallel_restore_add_progress(m.writer, progress.Progress) + if ret != 0 { + // Handle error appropriately, e.g., log it + fmt.Printf("Failed to add progress, error code: %d\n", ret) + } + if progress.Slice != nil { + ret = C.scoutfs_parallel_restore_add_slice(m.writer, progress.Slice) + C.free(unsafe.Pointer(progress.Slice)) + if ret != 0 { + // Handle error appropriately, e.g., log it + fmt.Printf("Failed to add slice, error code: %d\n", ret) + } + } + // Free the C-allocated progress structures + C.free(unsafe.Pointer(progress.Progress)) + } +} + +func (m *MasterWriter) Destroy() { + m.wg.Wait() + close(m.progressCh) + m.progressWg.Wait() + + if m.slice != nil { + C.free(unsafe.Pointer(m.slice)) // Free slice on error + } + if m.super != nil { + C.free(unsafe.Pointer(m.super)) // Free superblock on error + } + if m.devFd != 0 { + syscall.Close(m.devFd) + } + // Destroy master writer + C.scoutfs_parallel_restore_destroy_writer(&m.writer) +} + +func NewWriters(path string, numWriters int) (*MasterWriter, []*WorkerWriter, error) { + if numWriters <= 1 { + return nil, nil, errors.New("number of writers must be positive") + } + + devFd, err := syscall.Open(path, syscall.O_DIRECT|syscall.O_RDWR|syscall.O_EXCL, 0) + if err != nil { + return nil, nil, fmt.Errorf("failed to open metadata device '%s': %v", path, err) + } + + var masterWriter MasterWriter + masterWriter.progressCh = make(chan *ScoutfsParallelWriterProgress, numWriters*2) + masterWriter.workers = make([]*WorkerWriter, numWriters) + masterWriter.devFd = devFd + + masterWriter.super = (*C.struct_scoutfs_super_block)(C.malloc(C.size_t(unsafe.Sizeof(*masterWriter.super)))) + if masterWriter.super == nil { + masterWriter.Destroy() + return nil, nil, errors.New("failed to allocate memory for superblock") + } + + // Read the superblock from devFd. + superOffset := C.SCOUTFS_SUPER_BLKNO << C.SCOUTFS_BLOCK_SM_SHIFT + count, err := syscall.Pread(devFd, (*[1 << 30]byte)(unsafe.Pointer(masterWriter.super))[:C.SCOUTFS_BLOCK_SM_SIZE], int64(superOffset)) + if err != nil { + masterWriter.Destroy() + return nil, nil, fmt.Errorf("failed to read superblock: %v", err) + } + if count != int(C.SCOUTFS_BLOCK_SM_SIZE) { + masterWriter.Destroy() + return nil, nil, fmt.Errorf("failed to read superblock, bytes read: %d", count) + } + + // Check if the superblock is valid. + if C.le64_to_cpu(masterWriter.super.flags)&C.SCOUTFS_FLAG_IS_META_BDEV == 0 { + masterWriter.Destroy() + return nil, nil, errors.New("superblock is not a metadata device") + } + + var ret C.int + // Create master writer + ret = C.scoutfs_parallel_restore_create_writer(&masterWriter.writer) + if ret != 0 { + masterWriter.Destroy() + return nil, nil, errors.New("failed to create master writer") + } + + ret = C.scoutfs_parallel_restore_import_super(masterWriter.writer, masterWriter.super, C.int(devFd)) + if ret != 0 { + masterWriter.Destroy() + return nil, nil, fmt.Errorf("failed to import superblock, error code: %d", ret) + } + + // Initialize slices for each worker + masterWriter.slice = (*C.struct_scoutfs_parallel_restore_slice)(C.malloc(C.size_t(numWriters) * + C.size_t(unsafe.Sizeof(C.struct_scoutfs_parallel_restore_slice{})))) + if masterWriter.slice == nil { + masterWriter.Destroy() + return nil, nil, errors.New("failed to allocate slices") + } + + ret = C.scoutfs_parallel_restore_init_slices(masterWriter.writer, + masterWriter.slice, + C.int(numWriters)) + if ret != 0 { + masterWriter.Destroy() + return nil, nil, errors.New("failed to initialize slices") + } + + ret = C.scoutfs_parallel_restore_add_slice(masterWriter.writer, masterWriter.slice) + if ret != 0 { + masterWriter.Destroy() + return nil, nil, errors.New("failed to add slice to master writer") + } + + // Create worker writers + for i := 1; i < numWriters; i++ { + worker := &WorkerWriter{ + progressCh: masterWriter.progressCh, + buf: make([]byte, bufSize), + } + ret = C.scoutfs_parallel_restore_create_writer(&worker.writer) + if ret != 0 { + masterWriter.Destroy() + return nil, nil, errors.New("failed to create worker writer") + } + + masterWriter.wg.Add(1) + + // Use each slice for the corresponding worker + slice := (*C.struct_scoutfs_parallel_restore_slice)(unsafe.Pointer(uintptr(unsafe.Pointer(masterWriter.slice)) + + uintptr(i)*unsafe.Sizeof(C.struct_scoutfs_parallel_restore_slice{}))) + ret = C.scoutfs_parallel_restore_add_slice(worker.writer, slice) + if ret != 0 { + C.scoutfs_parallel_restore_destroy_writer(&worker.writer) + masterWriter.Destroy() + return nil, nil, errors.New("failed to add slice to worker writer") + } + + masterWriter.workers[i] = worker + } + go masterWriter.aggregateProgress() + + return &masterWriter, masterWriter.workers, nil + +} + +func (w *WorkerWriter) getProgress(withSlice bool) (*ScoutfsParallelWriterProgress, error) { + progress := (*C.struct_scoutfs_parallel_restore_progress)( + C.malloc(C.size_t(unsafe.Sizeof(C.struct_scoutfs_parallel_restore_progress{}))), + ) + if progress == nil { + return nil, errors.New("failed to allocate memory for progress") + } + + // Fetch the current progress from the C library + ret := C.scoutfs_parallel_restore_get_progress(w.writer, progress) + if ret != 0 { + C.free(unsafe.Pointer(progress)) + return nil, fmt.Errorf("failed to get progress, error code: %d", ret) + } + + var slice *C.struct_scoutfs_parallel_restore_slice + if withSlice { + slice = (*C.struct_scoutfs_parallel_restore_slice)( + C.malloc(C.size_t(unsafe.Sizeof(C.struct_scoutfs_parallel_restore_slice{}))), + ) + if slice == nil { + C.free(unsafe.Pointer(progress)) + return nil, errors.New("failed to allocate memory for slice") + } + + // Optionally fetch the slice information + ret = C.scoutfs_parallel_restore_get_slice(w.writer, slice) + if ret != 0 { + C.free(unsafe.Pointer(progress)) + C.free(unsafe.Pointer(slice)) + return nil, fmt.Errorf("failed to get slice, error code: %d", ret) + } + } + + return &ScoutfsParallelWriterProgress{ + Progress: progress, + Slice: slice, + }, nil +} + +// writeBufs writes data from the buffer to the device file descriptor. +// It uses scoutfs_parallel_restore_write_buf to get data and pwrite to write it. +func (w *WorkerWriter) writeBufs() (int64, error) { + var totalWritten int64 + var count int64 + var off int64 + var ret C.int + + // Allocate memory for off and count + offPtr := (*C.off_t)(unsafe.Pointer(&off)) + countPtr := (*C.size_t)(unsafe.Pointer(&count)) + + for { + ret = C.scoutfs_parallel_restore_write_buf(w.writer, unsafe.Pointer(&w.buf[0]), + C.size_t(len(w.buf)), offPtr, countPtr) + + if ret != 0 { + return totalWritten, fmt.Errorf("failed to write buffer: error code %d", ret) + } + + if count > 0 { + n, err := syscall.Pwrite(w.devFd, w.buf[:count], off) + if err != nil { + return totalWritten, fmt.Errorf("pwrite failed: %v", err) + } + if n != int(count) { + return totalWritten, fmt.Errorf("pwrite wrote %d bytes; expected %d", n, count) + } + totalWritten += int64(n) + } + + if count == 0 { + break + } + } + + return totalWritten, nil +} + +func (w *WorkerWriter) InsertEntry(entry *C.struct_scoutfs_parallel_restore_entry) error { + // Add the entry using the C library + ret := C.scoutfs_parallel_restore_add_entry(w.writer, entry) + if ret != 0 { + return fmt.Errorf("failed to add entry, error code: %d", ret) + } + + // Increment the fileCreated counter + w.fileCreated++ + if w.fileCreated >= batchSize { + _, err := w.writeBufs() + if err != nil { + return fmt.Errorf("error writing buffers: %v", err) + } + // Allocate memory for progress and slice structures + progress, err := w.getProgress(false) + if err != nil { + return err + } + // Send the progress update to the shared progress channel + w.progressCh <- progress + // Reset the fileCreated counter + w.fileCreated = 0 + } + + return nil +} + +func (w *WorkerWriter) InsertXattr(xattr *C.struct_scoutfs_parallel_restore_xattr) error { + ret := C.scoutfs_parallel_restore_add_xattr(w.writer, xattr) + if ret != 0 { + return fmt.Errorf("failed to add xattr, error code: %d", ret) + } + return nil +} + +func (w *WorkerWriter) InsertInode(inode *C.struct_scoutfs_parallel_restore_inode) error { + ret := C.scoutfs_parallel_restore_add_inode(w.writer, inode) + if ret != 0 { + return fmt.Errorf("failed to add inode, error code: %d", ret) + } + return nil +} + +// should only be called once +func (w *WorkerWriter) Destroy() error { + defer w.wg.Done() + // Send final progress if there are remaining entries + if w.fileCreated > 0 { + progress := &ScoutfsParallelWriterProgress{ + Progress: (*C.struct_scoutfs_parallel_restore_progress)(C.malloc(C.size_t(unsafe.Sizeof(C.struct_scoutfs_parallel_restore_progress{})))), + Slice: (*C.struct_scoutfs_parallel_restore_slice)(C.malloc(C.size_t(unsafe.Sizeof(C.struct_scoutfs_parallel_restore_slice{})))), + } + w.progressCh <- progress + w.fileCreated = 0 + } + + C.scoutfs_parallel_restore_destroy_writer(&w.writer) + return nil +} diff --git a/restore/pkg/restore/restore_test.go b/restore/pkg/restore/restore_test.go new file mode 100644 index 00000000..9231b543 --- /dev/null +++ b/restore/pkg/restore/restore_test.go @@ -0,0 +1,10 @@ +package restore + +import "testing" + +func TestNewWriters(t *testing.T) { + _, _, err := NewWriters("/tmp", 2) + if err != nil { + t.Fatalf("failed to create master writer: %v", err) + } +}