feat: add write-through disk caching & dynamic config (#43)

* add pct

* cleanup

* one more check

* move helpers down

* wip

* add dynamic configuration endpoint

* example region

* only override config if no err

* fix cache ids

* update readme

* remove word

* wip

* push store content benchmark

---------

Co-authored-by: Luke Lombardi <[email protected]>

Author: luke-lombardi

README.md

@@ -1,4 +1,10 @@
-# blobcache
+# Blobcache
 
 ## Overview
-A very simple in-memory cache used as a content-addressed cache. Exposes a GRPC server that can be embedded directly in a golang application.
+Blobcache is a distributed, in-memory cache used to cache objects near the workloads that need them. It does not require an external metadata server
+for each region, making it easy to spin up in various regions without performance penalties.
+
+## Features
+- **Tiered Storage**: Fast and efficient caching mechanism - both in-memory and on disk
+- **Content-Addressed**: Stores data based on content hash, so workloads sharing the same data (i.e. model weights), can benefit from the distributed cache.
+- **GRPC Interface**: Easily embeddable in Golang applications for seamless integration

e2e/throughput/main.go

@@ -53,10 +53,14 @@ func main() {
 	hashBytes := sha256.Sum256(b)
 	fileHash := hex.EncodeToString(hashBytes[:])
 
-	hash, err := storeFile(client, filePath, fileHash)
+	var totalStoreTime float64
+
+	hash, storeElapsedTime, err := storeFile(client, filePath, fileHash)
 	if err != nil {
 		log.Fatalf("Failed to store file: %v\n", err)
 	}
+	log.Printf("StoreContent elapsed time: %f seconds\n", storeElapsedTime)
+	totalStoreTime += storeElapsedTime
 
 	var totalStreamResult, totalGetContentResult TestResult
 	for i := 0; i < totalIterations; i++ {
@@ -83,10 +87,12 @@ func main() {
 		log.Printf("TestGetContent - %v\n", getContentResult)
 	}
 
-	GenerateReport(totalStreamResult, totalGetContentResult, len(b), totalIterations)
+	GenerateReport(totalStreamResult, totalGetContentResult, totalStoreTime, len(b), totalIterations)
 }
 
-func storeFile(client *blobcache.BlobCacheClient, filePath string, fileHash string) (string, error) {
+func storeFile(client *blobcache.BlobCacheClient, filePath string, fileHash string) (string, float64, error) {
+	startTime := time.Now() // Start timing
+
 	chunks := make(chan []byte)
 	go func() {
 		file, err := os.Open(filePath)
@@ -116,9 +122,11 @@ func storeFile(client *blobcache.BlobCacheClient, filePath string, fileHash stri
 
 	hash, err := client.StoreContent(chunks, fileHash)
 	if err != nil {
-		return "", err
+		return "", 0, err
 	}
-	return hash, nil
+
+	elapsedTime := time.Since(startTime).Seconds() // Calculate elapsed time
+	return hash, elapsedTime, nil
 }
 
 func TestGetContentStream(client *blobcache.BlobCacheClient, hash string, fileSize int, expectedHash string) (TestResult, error) {
@@ -212,13 +220,17 @@ func TestGetContent(client *blobcache.BlobCacheClient, hash string, fileSize int
 	return TestResult{ElapsedTime: elapsedTime, ContentCheckPassed: contentCheckPassed}, nil
 }
 
-func GenerateReport(streamResult, contentResult TestResult, fileSize, iterations int) {
+func GenerateReport(streamResult, contentResult TestResult, totalStoreTime float64, fileSize, iterations int) {
 	averageTimeStream := streamResult.ElapsedTime / float64(iterations)
 	averageTimeContent := contentResult.ElapsedTime / float64(iterations)
+	averageTimeStore := totalStoreTime / float64(iterations)
 	totalBytesReadMB := float64(fileSize*iterations) / (1024 * 1024)
 	mbPerSecondStream := totalBytesReadMB / streamResult.ElapsedTime
 	mbPerSecondContent := totalBytesReadMB / contentResult.ElapsedTime
 
+	log.Printf("================================================")
+	log.Printf("Total time for StoreContent: %f seconds\n", totalStoreTime)
+	log.Printf("Average time per iteration for StoreContent: %f seconds\n", averageTimeStore)
 	log.Printf("Total time for GetContentStream: %f seconds\n", streamResult.ElapsedTime)
 	log.Printf("Average time per iteration for GetContentStream: %f seconds\n", averageTimeStream)
 	log.Printf("Total time for GetContent: %f seconds\n", contentResult.ElapsedTime)

pkg/config.default.yaml

@@ -1,5 +1,7 @@
 server:
   mode: slave
+  diskCacheDir: /blobcache-data
+  diskCacheMaxUsagePct: 0.95
   token:
   rttThresholdMilliseconds: 100
   hostStorageCapacityThresholdPct: 0.95
@@ -10,6 +12,12 @@ server:
     redisAddr:
     redisPasswd:
     redisTLSEnabled: false
+  regions:
+    us-east-1:
+      token:
+      rttThresholdMilliseconds: 40
+      hostStorageCapacityThresholdPct: 0.5
+      maxCachePct: 50
 client:
   token:
   nTopHosts: 3

pkg/coordinator_local.go

@@ -8,6 +8,7 @@ type CoordinatorClient interface {
 	AddHostToIndex(ctx context.Context, locality string, host *BlobCacheHost) error
 	SetHostKeepAlive(ctx context.Context, locality string, host *BlobCacheHost) error
 	GetAvailableHosts(ctx context.Context, locality string) ([]*BlobCacheHost, error)
+	GetRegionConfig(ctx context.Context, locality string) (BlobCacheServerConfig, error)
 	SetClientLock(ctx context.Context, hash string, host string) error
 	RemoveClientLock(ctx context.Context, hash string, host string) error
 	SetStoreFromContentLock(ctx context.Context, locality string, sourcePath string) error
@@ -35,6 +36,10 @@ func NewCoordinatorClientLocal(globalConfig BlobCacheGlobalConfig, serverConfig
 	return &CoordinatorClientLocal{globalConfig: globalConfig, serverConfig: serverConfig, metadata: metadata}, nil
 }
 
+func (c *CoordinatorClientLocal) GetRegionConfig(ctx context.Context, locality string) (BlobCacheServerConfig, error) {
+	return c.serverConfig, nil
+}
+
 func (c *CoordinatorClientLocal) AddHostToIndex(ctx context.Context, locality string, host *BlobCacheHost) error {
 	return c.metadata.AddHostToIndex(ctx, locality, host)
 }

pkg/coordinator_remote.go

@@ -132,6 +132,20 @@ func (c *CoordinatorClientRemote) GetAvailableHosts(ctx context.Context, localit
 	return hosts, nil
 }
 
+func (c *CoordinatorClientRemote) GetRegionConfig(ctx context.Context, locality string) (BlobCacheServerConfig, error) {
+	response, err := c.client.GetRegionConfig(ctx, &proto.GetRegionConfigRequest{Locality: locality})
+	if err != nil {
+		return BlobCacheServerConfig{}, err
+	}
+
+	if !response.Ok {
+		return BlobCacheServerConfig{}, errors.New("failed to get region config")
+	}
+
+	cfg := BlobCacheServerConfigFromProto(response.ServerConfig)
+	return cfg, nil
+}
+
 func (c *CoordinatorClientRemote) SetClientLock(ctx context.Context, hash string, hostId string) error {
 	r, err := c.client.SetClientLock(ctx, &proto.SetClientLockRequest{Hash: hash, HostId: hostId})
 	if err != nil {

pkg/server.go

@@ -50,7 +50,6 @@ type CacheService struct {
 }
 
 func NewCacheService(ctx context.Context, cfg BlobCacheConfig, locality string) (*CacheService, error) {
-	hostId := fmt.Sprintf("%s-%s", BlobCacheHostPrefix, uuid.New().String()[:6])
 	currentHost := &BlobCacheHost{
 		RTT: 0,
 	}
@@ -62,12 +61,29 @@ func NewCacheService(ctx context.Context, cfg BlobCacheConfig, locality string)
 		coordinator, err = NewCoordinatorClientLocal(cfg.Global, cfg.Server)
 	default:
 		coordinator, err = NewCoordinatorClientRemote(cfg.Global, cfg.Client.Token)
+		if err != nil {
+			return nil, err
+		}
+
+		regionConfig, err := coordinator.GetRegionConfig(ctx, locality)
+		if err != nil {
+			Logger.Infof("No region-specific config found for locality %s, using current config", locality)
+		} else {
+			cfg.Server = regionConfig
+		}
 	}
 	if err != nil {
 		return nil, err
 	}
 
-	Logger.Infof("Server started in %s mode", cfg.Server.Mode)
+	// Create the disk cache directory if it doesn't exist
+	err = os.MkdirAll(cfg.Server.DiskCacheDir, 0755)
+	if err != nil {
+		return nil, err
+	}
+
+	hostId := getHostId(cfg.Server)
+	Logger.Infof("Server<%s> started in %s mode", hostId, cfg.Server.Mode)
 
 	publicIpAddr, _ := GetPublicIpAddr()
 	if publicIpAddr != "" {
@@ -116,6 +132,20 @@ func NewCacheService(ctx context.Context, cfg BlobCacheConfig, locality string)
 	return cs, nil
 }
 
+func getHostId(serverConfig BlobCacheServerConfig) string {
+	filePath := filepath.Join(serverConfig.DiskCacheDir, "HOST_ID")
+
+	hostId := ""
+	if content, err := os.ReadFile(filePath); err == nil {
+		hostId = strings.TrimSpace(string(content))
+	} else {
+		hostId = fmt.Sprintf("%s-%s", BlobCacheHostPrefix, uuid.New().String()[:6])
+		os.WriteFile(filePath, []byte(hostId), 0644)
+	}
+
+	return hostId
+}
+
 func (cs *CacheService) HostKeepAlive() {
 	err := cs.coordinator.SetHostKeepAlive(cs.ctx, cs.locality, cs.cas.currentHost)
 	if err != nil {

pkg/server_coordinator.go

@@ -23,6 +23,17 @@ func (cs *CacheService) GetAvailableHosts(ctx context.Context, req *proto.GetAva
 	return &proto.GetAvailableHostsResponse{Hosts: protoHosts, Ok: true}, nil
 }
 
+func (cs *CacheService) GetRegionConfig(ctx context.Context, req *proto.GetRegionConfigRequest) (*proto.GetRegionConfigResponse, error) {
+	Logger.Infof("GetRegionConfig[ACK] - [%s]", req.Locality)
+
+	config, ok := cs.serverConfig.Regions[req.Locality]
+	if !ok {
+		return &proto.GetRegionConfigResponse{Ok: false, ServerConfig: nil}, nil
+	}
+
+	return &proto.GetRegionConfigResponse{Ok: true, ServerConfig: config.ServerConfig.ToProto()}, nil
+}
+
 func (cs *CacheService) SetClientLock(ctx context.Context, req *proto.SetClientLockRequest) (*proto.SetClientLockResponse, error) {
 	Logger.Debugf("SetClientLock[ACK] - [%s]", req.Hash)