Xload prefetch initial code

cmd/imports.go

@@ -42,4 +42,5 @@ import (
 	_ "github.com/Azure/azure-storage-fuse/v2/component/file_cache"
 	_ "github.com/Azure/azure-storage-fuse/v2/component/libfuse"
 	_ "github.com/Azure/azure-storage-fuse/v2/component/loopback"
+	_ "github.com/Azure/azure-storage-fuse/v2/component/xload"
 )

common/types.go

@@ -308,6 +308,11 @@ func NewUUID() (u uuid) {
 	return
 }
 
+// returns block id of given length
+func GetBlockID(len int64) string {
+	return base64.StdEncoding.EncodeToString(NewUUIDWithLength(len))
+}
+
 func GetIdLength(id string) int64 {
 	existingBlockId, _ := base64.StdEncoding.DecodeString(id)
 	return int64(len(existingBlockId))

component/azstorage/block_blob.go

@@ -36,7 +36,6 @@ package azstorage
 import (
 	"bytes"
 	"context"
-	"encoding/base64"
 	"errors"
 	"fmt"
 	"math"
@@ -996,7 +995,7 @@ func (bb *BlockBlob) GetFileBlockOffsets(name string) (*common.BlockOffsetList,
 }
 
 func (bb *BlockBlob) createBlock(blockIdLength, startIndex, size int64) *common.Block {
-	newBlockId := base64.StdEncoding.EncodeToString(common.NewUUIDWithLength(blockIdLength))
+	newBlockId := common.GetBlockID(blockIdLength)
 	newBlock := &common.Block{
 		Id:         newBlockId,
 		StartIndex: startIndex,
@@ -1080,14 +1079,14 @@ func (bb *BlockBlob) TruncateFile(name string, size int64) error {
 			blobClient := bb.Container.NewBlockBlobClient(blobName)
 
 			blkList := make([]string, 0)
-			id := base64.StdEncoding.EncodeToString(common.NewUUIDWithLength(16))
+			id := common.GetBlockID(16)
 
 			for i := 0; size > 0; i++ {
 				if i == 0 || size < blkSize {
 					// Only first and last block we upload and rest all we replicate with the first block itself
 					if size < blkSize {
 						blkSize = size
-						id = base64.StdEncoding.EncodeToString(common.NewUUIDWithLength(16))
+						id = common.GetBlockID(16)
 					}
 					data := make([]byte, blkSize)
 

component/block_cache/block_cache.go

@@ -36,7 +36,6 @@ package block_cache
 import (
 	"container/list"
 	"context"
-	"encoding/base64"
 	"fmt"
 	"io"
 	"math"
@@ -1315,8 +1314,7 @@ func shouldCommitAndDownload(blockID int64, handle *handlemap.Handle) (bool, boo
 
 // lineupUpload : Create a work item and schedule the upload
 func (bc *BlockCache) lineupUpload(handle *handlemap.Handle, block *Block, listMap map[int64]*blockInfo) {
-
-	id := base64.StdEncoding.EncodeToString(common.NewUUIDWithLength(16))
+	id := common.GetBlockID(16)
 	listMap[block.id] = &blockInfo{
 		id:        id,
 		committed: false,
@@ -1584,7 +1582,7 @@ func (bc *BlockCache) stageZeroBlock(handle *handlemap.Handle, tryCnt int) (stri
 		return "", fmt.Errorf("3 attempts to upload zero block have failed for %v=>%v", handle.ID, handle.Path)
 	}
 
-	id := base64.StdEncoding.EncodeToString(common.NewUUIDWithLength(16))
+	id := common.GetBlockID(16)
 
 	log.Debug("BlockCache::stageZeroBlock : Staging zero block for %v=>%v, try = %v", handle.ID, handle.Path, tryCnt)
 	err := bc.NextComponent().StageData(internal.StageDataOptions{

component/xload/block.go

@@ -0,0 +1,96 @@
+/*
+    _____           _____   _____   ____          ______  _____  ------
+   |     |  |      |     | |     | |     |     | |       |            |
+   |     |  |      |     | |     | |     |     | |       |            |
+   | --- |  |      |     | |-----| |---- |     | |-----| |-----  ------
+   |     |  |      |     | |     | |     |     |       | |       |
+   | ____|  |_____ | ____| | ____| |     |_____|  _____| |_____  |_____
+
+
+   Licensed under the MIT License <http://opensource.org/licenses/MIT>.
+
+   Copyright © 2020-2024 Microsoft Corporation. All rights reserved.
+   Author : <[email protected]>
+
+   Permission is hereby granted, free of charge, to any person obtaining a copy
+   of this software and associated documentation files (the "Software"), to deal
+   in the Software without restriction, including without limitation the rights
+   to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+   copies of the Software, and to permit persons to whom the Software is
+   furnished to do so, subject to the following conditions:
+
+   The above copyright notice and this permission notice shall be included in all
+   copies or substantial portions of the Software.
+
+   THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+   IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+   FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+   AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+   LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+   OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+   SOFTWARE
+*/
+
+package xload
+
+import (
+	"fmt"
+	"syscall"
+)
+
+// Block is a memory mapped buffer with its state to hold data
+type Block struct {
+	index  int    // Index of the block in the pool
+	offset int64  // Start offset of the data this block holds
+	length int64  // Length of data that this block holds
+	id     string // ID to represent this block in the blob
+	data   []byte // Data this block holds
+}
+
+// AllocateBlock creates a new memory mapped buffer for the given size
+func AllocateBlock(size uint64) (*Block, error) {
+	if size == 0 {
+		return nil, fmt.Errorf("invalid size")
+	}
+
+	prot, flags := syscall.PROT_READ|syscall.PROT_WRITE, syscall.MAP_ANON|syscall.MAP_PRIVATE
+	addr, err := syscall.Mmap(-1, 0, int(size), prot, flags)
+
+	if err != nil {
+		return nil, fmt.Errorf("mmap error: %v", err)
+	}
+
+	block := &Block{
+		index:  0,
+		data:   addr,
+		offset: 0,
+		length: 0,
+		id:     "",
+	}
+
+	return block, nil
+}
+
+// Delete cleans up the memory mapped buffer
+func (b *Block) Delete() error {
+	if b.data == nil {
+		return fmt.Errorf("invalid buffer")
+	}
+
+	err := syscall.Munmap(b.data)
+	b.data = nil
+	if err != nil {
+		// if we get here, there is likely memory corruption.
+		return fmt.Errorf("munmap error: %v", err)
+	}
+
+	return nil
+}
+
+// Clear the old data of this block
+func (b *Block) ReUse() {
+	b.id = ""
+	b.index = 0
+	b.offset = 0
+	b.length = 0
+}

component/xload/blockpool.go

@@ -0,0 +1,118 @@
+/*
+    _____           _____   _____   ____          ______  _____  ------
+   |     |  |      |     | |     | |     |     | |       |            |
+   |     |  |      |     | |     | |     |     | |       |            |
+   | --- |  |      |     | |-----| |---- |     | |-----| |-----  ------
+   |     |  |      |     | |     | |     |     |       | |       |
+   | ____|  |_____ | ____| | ____| |     |_____|  _____| |_____  |_____
+
+
+   Licensed under the MIT License <http://opensource.org/licenses/MIT>.
+
+   Copyright © 2020-2024 Microsoft Corporation. All rights reserved.
+   Author : <[email protected]>
+
+   Permission is hereby granted, free of charge, to any person obtaining a copy
+   of this software and associated documentation files (the "Software"), to deal
+   in the Software without restriction, including without limitation the rights
+   to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+   copies of the Software, and to permit persons to whom the Software is
+   furnished to do so, subject to the following conditions:
+
+   The above copyright notice and this permission notice shall be included in all
+   copies or substantial portions of the Software.
+
+   THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+   IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+   FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+   AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+   LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+   OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+   SOFTWARE
+*/
+
+package xload
+
+import "github.com/Azure/azure-storage-fuse/v2/common/log"
+
+const _1MB uint64 = (1024 * 1024)
+
+// BlockPool is a pool of Blocks
+type BlockPool struct {
+	// Channel holding free blocks
+	blocksCh chan *Block
+
+	// Size of each block this pool holds
+	blockSize uint64
+
+	// Number of block that this pool can handle at max
+	maxBlocks uint32
+}
+
+// NewBlockPool allocates a new pool of blocks
+func NewBlockPool(blockSize uint64, blockCount uint32) *BlockPool {
+	// Ignore if config is invalid
+	if blockSize == 0 {
+		log.Err("blockpool::NewBlockPool : blockSize : %v", blockSize)
+		return nil
+	}
+
+	pool := &BlockPool{
+		blocksCh:  make(chan *Block, blockCount),
+		maxBlocks: uint32(blockCount),
+		blockSize: blockSize,
+	}
+
+	// Preallocate all blocks so that during runtime we do not spend CPU cycles on this
+	for i := (uint32)(0); i < blockCount; i++ {
+		b, err := AllocateBlock(blockSize)
+		if err != nil {
+			return nil
+		}
+
+		pool.blocksCh <- b
+	}
+
+	return pool
+}
+
+// Terminate ends the block pool life
+func (pool *BlockPool) Terminate() {
+	close(pool.blocksCh)
+
+	// Release back the memory allocated to each block
+	for {
+		b := <-pool.blocksCh
+		if b == nil {
+			break
+		}
+		_ = b.Delete()
+	}
+}
+
+// Usage provides % usage of this block pool
+func (pool *BlockPool) Usage() uint32 {
+	return ((pool.maxBlocks - (uint32)(len(pool.blocksCh))) * 100) / pool.maxBlocks
+}
+
+// Get a Block from the pool, return back if nothing is available
+func (pool *BlockPool) Get() *Block {
+	// getting a block from pool will be a blocking operation if the pool is empty
+	b := <-pool.blocksCh
+
+	// Mark the buffer ready for reuse now
+	if b != nil {
+		b.ReUse()
+	}
+	return b
+}
+
+// Release back the Block to the pool
+func (pool *BlockPool) Release(b *Block) {
+	select {
+	case pool.blocksCh <- b:
+		break
+	default:
+		_ = b.Delete()
+	}
+}