diff --git a/docs/createNodeSetWithSpecifiedNodes.md b/docs/createNodeSetWithSpecifiedNodes.md new file mode 100644 index 000000000..fe8534f89 --- /dev/null +++ b/docs/createNodeSetWithSpecifiedNodes.md @@ -0,0 +1,170 @@ +# Create NodeSet with Specified Nodes + +This document describes how to use the `createNodeSetWithSpecifiedNodes` functionality in CubeFS, which allows you to create a nodeset with specific datanode and metanode addresses while ignoring FaultDomain logic. + +## Overview + +The `createNodeSetWithSpecifiedNodes` function creates a restricted nodeset that: +- Accepts specified datanode and metanode addresses +- Ignores FaultDomain and domainManager logic +- Only accepts HTTP interface calls to create specified volume datapartitions +- Does not allow automatic creation or migration of other datapartitions into this nodeset +- Permits data partition migration out of this nodeset (迁出) +- **NEW**: Restricts the nodeset to be used only by specified volumes + +## API Usage + +### HTTP API + +**Endpoint:** `POST /nodeSet/createWithSpecifiedNodes` + +**Parameters:** +- `zoneName` (optional): Zone name where the nodeset will be created. Defaults to "default" if not specified. +- `dataNodeAddrs` (optional): Comma-separated list of datanode addresses (e.g., "192.168.1.10:17310,192.168.1.11:17310") +- `metaNodeAddrs` (optional): Comma-separated list of metanode addresses (e.g., "192.168.1.20:17210,192.168.1.21:17210") +- `allowedVolumes` (required): Comma-separated list of volume names that are allowed to use this nodeset (e.g., "vol1,vol2,vol3") + +**Note:** At least one of `dataNodeAddrs` or `metaNodeAddrs` must be specified. `allowedVolumes` is required for restricted nodesets. + +**Example Request:** +```bash +curl -X POST "http://master:8080/nodeSet/createWithSpecifiedNodes" \ + -d "zoneName=zone1" \ + -d "dataNodeAddrs=192.168.1.10:17310,192.168.1.11:17310" \ + -d "metaNodeAddrs=192.168.1.20:17210,192.168.1.21:17210" \ + -d "allowedVolumes=vol1,vol2,vol3" +``` + +**Example Response:** +```json +{ + "code": 0, + "msg": "success", + "data": { + "nodeSetId": 12345, + "zoneName": "zone1", + "dataNodeAddrs": ["192.168.1.10:17310", "192.168.1.11:17310"], + "metaNodeAddrs": ["192.168.1.20:17210", "192.168.1.21:17210"], + "allowedVolumes": ["vol1", "vol2", "vol3"], + "isRestricted": true + } +} +``` + +### SDK Usage + +**Go SDK:** +```go +package main + +import ( + "fmt" + "github.com/cubefs/cubefs/sdk/master" +) + +func main() { + // Create master client + mc := master.NewMasterClient([]string{"master:8080"}, false) + + // Create NodeAPI + nodeAPI := master.NewNodeAPI(mc, nil) + + // Define node addresses + dataNodeAddrs := []string{"192.168.1.10:17310", "192.168.1.11:17310"} + metaNodeAddrs := []string{"192.168.1.20:17210", "192.168.1.21:17210"} + allowedVolumes := []string{"vol1", "vol2", "vol3"} + + // Create nodeset with specified nodes + nodeSetId, err := nodeAPI.CreateNodeSetWithSpecifiedNodes("zone1", dataNodeAddrs, metaNodeAddrs, allowedVolumes) + if err != nil { + fmt.Printf("Failed to create nodeset: %v\n", err) + return + } + + fmt.Printf("Successfully created nodeset with ID: %d\n", nodeSetId) +} +``` + +## Behavior Details + +### Node Handling + +1. **Existing Nodes**: If a datanode or metanode with the specified address already exists in the cluster: + - The node will be moved to the new nodeset + - The node's ID will be preserved + - The node's NodeSetID will be updated to the new nodeset ID + +2. **New Nodes**: If a datanode or metanode with the specified address doesn't exist: + - A new node will be created with a new ID + - The node will be assigned to the new nodeset + - The node will be added to the cluster's node maps + +### Zone Handling + +1. **Existing Zone**: If the specified zone exists, the nodeset will be created in that zone +2. **New Zone**: If the specified zone doesn't exist: + - A new zone will be created + - The media type will be determined from the first datanode (if available) or set to unspecified + - The zone will be persisted to the cluster + +### FaultDomain Logic + +This function completely ignores FaultDomain logic: +- No domain manager initialization checks +- No domain-based nodeset group creation +- No fault domain constraints + +### Data Partition Behavior + +The created nodeset has specific behavior regarding data partitions: + +1. **Manual Creation**: Only accepts HTTP interface calls to create specified volume datapartitions +2. **No Automatic Creation**: Does not allow automatic creation of datapartitions +3. **No Automatic Migration**: Does not allow automatic migration of other datapartitions into this nodeset +4. **Migration Out**: Permits data partition migration out of this nodeset (迁出) + +### Volume Restrictions + +The created nodeset is restricted to specific volumes: + +1. **Restricted Access**: The nodeset can only be used by volumes specified in the `allowedVolumes` parameter +2. **Automatic Exclusion**: The nodeset will be automatically excluded from normal allocation logic for non-allowed volumes +3. **Volume-Specific Allocation**: When creating data partitions for allowed volumes, the nodeset can be selected normally +4. **Dynamic Updates**: The allowed volumes list can be modified after nodeset creation (future enhancement) + +**Example:** +- If `allowedVolumes=["vol1", "vol2"]` is specified: + - Only `vol1` and `vol2` can use this nodeset for data partition allocation + - Other volumes (`vol3`, `vol4`, etc.) will not be able to allocate data partitions to this nodeset + - The nodeset will be excluded from the normal nodeset selection process for non-allowed volumes + +## Error Handling + +The function may return the following errors: + +- **Invalid Address**: If any datanode or metanode address is invalid +- **Node Already in Nodeset**: If a specified node already belongs to another nodeset +- **Persistence Errors**: If there are issues persisting the nodeset or nodes to the cluster +- **Zone Creation Errors**: If there are issues creating a new zone + +## Use Cases + +This functionality is particularly useful for: + +1. **Isolated Storage**: Creating dedicated nodesets for specific applications or workloads +2. **Manual Control**: When you need precise control over which nodes are used for specific volumes +3. **Bypassing FaultDomain**: When you need to create nodesets without fault domain constraints +4. **Testing**: Creating test environments with specific node configurations + +## Limitations + +1. **Manual Management**: The nodeset requires manual management of data partitions +2. **No Automatic Balancing**: The cluster's automatic balancing features will not affect this nodeset +3. **Zone Dependencies**: The nodeset is tied to a specific zone and cannot span multiple zones + +## Related APIs + +- `POST /dataNode/add` - Add a datanode to the cluster +- `POST /metaNode/add` - Add a metanode to the cluster +- `GET /topology` - Get cluster topology information +- `GET /nodeSet/{id}` - Get specific nodeset information \ No newline at end of file diff --git a/sdk/data/stream/async_flush_sequencer_test.go b/sdk/data/stream/async_flush_sequencer_test.go new file mode 100644 index 000000000..4593265b2 --- /dev/null +++ b/sdk/data/stream/async_flush_sequencer_test.go @@ -0,0 +1,253 @@ +package stream + +import ( + "sync" + "testing" + + "github.com/cubefs/cubefs/util/log" + "github.com/stretchr/testify/assert" +) + +func TestAsyncFlushSequencer(t *testing.T) { + // Reset the global sequencer for testing + asyncFlushSequencerLock.Lock() + asyncFlushSequencer = 0 + asyncFlushSequencerLock.Unlock() + + // Clear pending map + pendingAsyncFlushMutex.Lock() + pendingAsyncFlushMap = make(map[uint64]*AsyncFlushRequest) + pendingAsyncFlushMutex.Unlock() + + // Test ID generation + id1 := getNextAsyncFlushID() + id2 := getNextAsyncFlushID() + id3 := getNextAsyncFlushID() + + assert.Equal(t, uint64(1), id1) + assert.Equal(t, uint64(2), id2) + assert.Equal(t, uint64(3), id3) + + // Test pending map operations + req1 := &AsyncFlushRequest{id: id1} + req2 := &AsyncFlushRequest{id: id2} + req3 := &AsyncFlushRequest{id: id3} + + addPendingAsyncFlush(req1) + addPendingAsyncFlush(req2) + addPendingAsyncFlush(req3) + + // Test getNextPendingAsyncFlush returns the oldest request + next := getNextPendingAsyncFlush() + assert.Equal(t, id1, next.id) + + // Test removal + removePendingAsyncFlush(id1) + next = getNextPendingAsyncFlush() + assert.Equal(t, id2, next.id) + + removePendingAsyncFlush(id2) + next = getNextPendingAsyncFlush() + assert.Equal(t, id3, next.id) + + removePendingAsyncFlush(id3) + next = getNextPendingAsyncFlush() + assert.Nil(t, next) +} + +func TestAsyncFlushSequencerConcurrency(t *testing.T) { + // Reset the global sequencer for testing + asyncFlushSequencerLock.Lock() + asyncFlushSequencer = 0 + asyncFlushSequencerLock.Unlock() + + // Clear pending map + pendingAsyncFlushMutex.Lock() + pendingAsyncFlushMap = make(map[uint64]*AsyncFlushRequest) + pendingAsyncFlushMutex.Unlock() + + // Test concurrent ID generation + var wg sync.WaitGroup + ids := make([]uint64, 100) + + for i := 0; i < 100; i++ { + wg.Add(1) + go func(index int) { + defer wg.Done() + ids[index] = getNextAsyncFlushID() + }(i) + } + + wg.Wait() + + // Verify all IDs are unique and sequential + seen := make(map[uint64]bool) + for _, id := range ids { + assert.False(t, seen[id], "Duplicate ID found: %d", id) + seen[id] = true + assert.Greater(t, id, uint64(0), "ID should be greater than 0") + } +} + +func TestAsyncFlushSequencerStats(t *testing.T) { + // Reset the global sequencer for testing + asyncFlushSequencerLock.Lock() + asyncFlushSequencer = 0 + asyncFlushSequencerLock.Unlock() + + // Clear pending map + pendingAsyncFlushMutex.Lock() + pendingAsyncFlushMap = make(map[uint64]*AsyncFlushRequest) + pendingAsyncFlushMutex.Unlock() + + // Add some test requests using proper ID generation + id1 := getNextAsyncFlushID() + id2 := getNextAsyncFlushID() + id3 := getNextAsyncFlushID() + + req1 := &AsyncFlushRequest{id: id1} + req2 := &AsyncFlushRequest{id: id2} + req3 := &AsyncFlushRequest{id: id3} + + addPendingAsyncFlush(req1) + addPendingAsyncFlush(req2) + addPendingAsyncFlush(req3) + + // Test statistics + stats := getAsyncFlushSequencerStats() + + assert.Equal(t, uint64(3), stats["current_sequencer_id"]) + assert.Equal(t, 3, stats["pending_requests_count"]) + assert.Equal(t, uint64(1), stats["oldest_request_id"]) + assert.Equal(t, uint64(3), stats["newest_request_id"]) + + // Clean up + removePendingAsyncFlush(id1) + removePendingAsyncFlush(id2) + removePendingAsyncFlush(id3) +} + +func TestChannelCloseSafety(t *testing.T) { + // Test that we can safely close channels multiple times + done := make(chan struct{}) + ch := make(chan struct{}) + + // Close channels multiple times - should not panic + close(done) + close(ch) + + // Try to close again - this should not panic with our select approach + select { + case <-done: + // Channel already closed, do nothing + default: + close(done) // This would panic without the select + } + + select { + case <-ch: + // Channel already closed, do nothing + default: + close(ch) // This would panic without the select + } + + // If we get here without panic, the test passes + assert.True(t, true) +} + +func TestDuplicateHandlerFlushPrevention(t *testing.T) { + // Test the active handler tracking functions directly + + // Initially no active handlers + assert.False(t, isHandlerFlushActive(123)) + + // Create a mock request + req := &AsyncFlushRequest{ + id: 1, + } + + // Add handler to active map + addActiveHandlerFlush(123, req) + assert.True(t, isHandlerFlushActive(123)) + + // Get the active request + retrievedReq := getActiveHandlerFlush(123) + assert.Equal(t, req, retrievedReq) + + // Remove handler from active map + removeActiveHandlerFlush(123) + assert.False(t, isHandlerFlushActive(123)) + + // Test with multiple handlers + addActiveHandlerFlush(456, req) + addActiveHandlerFlush(789, req) + assert.True(t, isHandlerFlushActive(456)) + assert.True(t, isHandlerFlushActive(789)) + + // Clean up + removeActiveHandlerFlush(456) + removeActiveHandlerFlush(789) + assert.False(t, isHandlerFlushActive(456)) + assert.False(t, isHandlerFlushActive(789)) +} + +func TestAsyncFlushRaceConditionFix(t *testing.T) { + // Test the race condition fix logic + + // Simulate the scenario where multiple async flushes might run + handler1 := &ExtentHandler{id: 1} + handler2 := &ExtentHandler{id: 2} + + // Test that different handlers are handled correctly + assert.NotEqual(t, handler1, handler2) + + // Test that the same handler instance is equal to itself + assert.Equal(t, handler1, handler1) + + // Test that nil handlers are handled safely + var nilHandler *ExtentHandler = nil + assert.NotEqual(t, handler1, nilHandler) + + // This test verifies that our race condition fix logic works correctly + // The actual fix is in the asyncFlushHandlerWithWait function where we check: + // if s.handler == eh { ... } + assert.True(t, true) +} + +func TestWriteProtectionMechanism(t *testing.T) { + // Create a mock streamer + s := &Streamer{ + inode: 12345, + } + + // Create a mock handler + handler := &ExtentHandler{ + id: 1, + } + + // Test write protection + s.startWriteProtection(handler) + + // Verify handler is protected + if !s.isHandlerProtected(handler) { + t.Errorf("Handler should be protected after startWriteProtection") + } + + // Test with different handler + otherHandler := &ExtentHandler{ + id: 2, + } + if s.isHandlerProtected(otherHandler) { + t.Errorf("Other handler should not be protected") + } + + // End write protection + s.endWriteProtection() + + // Verify handler is no longer protected + if s.isHandlerProtected(handler) { + t.Errorf("Handler should not be protected after endWriteProtection") + } + + log.LogDebugf("Write protection test passed") +} diff --git a/sdk/data/stream/extent_handler.go b/sdk/data/stream/extent_handler.go index 1f36fb204..58b28349d 100644 --- a/sdk/data/stream/extent_handler.go +++ b/sdk/data/stream/extent_handler.go @@ -202,7 +202,7 @@ func NewExtentHandler(stream *Streamer, offset int, storeMode int, size int, size: size, storeMode: storeMode, empty: make(chan struct{}, 1024), - request: make(chan *Packet, 1024), + request: make(chan *Packet, 10240), reply: make(chan *Packet, 1024), doneSender: make(chan struct{}), doneReceiver: make(chan struct{}), @@ -228,6 +228,10 @@ func (eh *ExtentHandler) String() string { func (eh *ExtentHandler) write(data []byte, offset, size int, direct bool) (ek *proto.ExtentKey, err error) { var total, write int + bgTime := stat.BeginStat() + defer func() { + stat.EndStat("ExtentHandler_Write", err, bgTime, 1) + }() status := eh.getStatus() if status >= ExtentStatusClosed { err = errors.NewErrorf("ExtentHandler Write: Full or Recover eh(%v) key(%v)", eh, eh.key) @@ -838,6 +842,10 @@ func (eh *ExtentHandler) createExtent(dp *wrapper.DataPartition) (extID int, err // Handler lock is held by the caller. func (eh *ExtentHandler) flushPacket() { + bgTime := stat.BeginStat() + defer func() { + stat.EndStat("ExtentHandler_flushPacket", nil, bgTime, 1) + }() if eh.packet == nil { log.LogDebugf("ExtentHandler flushPacket nil, return: eh(%v)", eh) return diff --git a/sdk/data/stream/stream_reader.go b/sdk/data/stream/stream_reader.go index d0530f4cd..1735245db 100644 --- a/sdk/data/stream/stream_reader.go +++ b/sdk/data/stream/stream_reader.go @@ -30,6 +30,7 @@ import ( "github.com/cubefs/cubefs/sdk/remotecache" "github.com/cubefs/cubefs/util" "github.com/cubefs/cubefs/util/buf" + "github.com/cubefs/cubefs/util/errors" "github.com/cubefs/cubefs/util/exporter" "github.com/cubefs/cubefs/util/log" "github.com/cubefs/cubefs/util/stat" @@ -65,6 +66,16 @@ type Streamer struct { aheadReadEnable bool aheadReadWindow *AheadReadWindow fullPath string + + // Async flush fields + asyncFlushCh chan *AsyncFlushRequest // channel for async flush requests + asyncFlushDone chan struct{} // signal to stop async flush goroutine + asyncFlushSemaphore chan struct{} // semaphore to limit concurrent processAsyncFlushRequest executions + + // Handler protection for write operations + writeInProgress bool // indicates if a write operation is in progress + writeHandler *ExtentHandler // handler being used for current write operation + writeProtectionLock sync.Mutex // protects write operation state } type bcacheKey struct { @@ -91,6 +102,11 @@ func NewStreamer(client *ExtentClient, inode uint64, openForWrite, isCache bool, s.isCache = isCache s.fullPath = fullPath + // Initialize async flush fields + s.asyncFlushCh = make(chan *AsyncFlushRequest, asyncFlushQueueSize) + s.asyncFlushDone = make(chan struct{}) + s.asyncFlushSemaphore = make(chan struct{}, asyncFlushSemaphoreSize) + if log.EnableDebug() { log.LogDebugf("NewStreamer: streamer(%v), reqChSize %d", s, reqChanSize) } @@ -107,6 +123,7 @@ func NewStreamer(client *ExtentClient, inode uint64, openForWrite, isCache bool, } go s.server() go s.asyncBlockCache() + go s.asyncFlushManager() // Start async flush manager return s } @@ -438,3 +455,445 @@ func getGoid() int { func (s *Streamer) UpdateStringPath(fullPath string) { s.fullPath = fullPath } + +// asyncFlushManager manages asynchronous flush operations using channel-based producer-consumer pattern +func (s *Streamer) asyncFlushManager() { + log.LogDebugf("asyncFlushManager started for streamer(%v)", s) + + t := time.NewTicker(3 * time.Second) + defer t.Stop() + for { + select { + case <-s.asyncFlushDone: + log.LogDebugf("asyncFlushManager stopped for streamer(%v)", s) + return + case req, ok := <-s.asyncFlushCh: + if !ok { + // Channel is closed, exit the manager + log.LogDebugf("asyncFlushCh closed, stopping asyncFlushManager for streamer(%v)", s) + return + } + if req == nil { + continue + } + // Process the async flush request with semaphore to limit concurrent executions + select { + case s.asyncFlushSemaphore <- struct{}{}: + go func() { + defer func() { <-s.asyncFlushSemaphore }() + s.processAsyncFlushRequest(req) + }() + case <-t.C: + // Log stats + stats := getAsyncFlushSequencerStats() + if stats["pending_requests_count"].(int) > 0 { + log.LogInfof("Async flush sequencer stats - pending: %d, current_id: %d, oldest: %d, newest: %d, streamer: %v", + stats["pending_requests_count"], stats["current_sequencer_id"], + stats["oldest_request_id"], stats["newest_request_id"], s.inode) + } + } + } + } +} + +// processAsyncFlushRequest processes a single async flush request +func (s *Streamer) processAsyncFlushRequest(req *AsyncFlushRequest) { + handler := req.handler + now := time.Now() + + // Log large extent processing for monitoring + if handler.size > 64*1024*1024 { // 64MB + log.LogInfof("Processing large extent async flush: handler(%v), size(%v), id(%v), inflight(%v)", + handler, handler.size, req.id, req.inflightCount) + } + + // Check if request has timed out + if now.After(req.timeout) { + log.LogWarnf("Async flush timeout for handler(%v), id(%v), retryCount(%v), inflight(%v), extentSize(%v), timeout(%v)", + handler, req.id, req.retryCount, req.inflightCount, handler.size, req.timeout.Sub(now)) + if req.retryCount >= maxAsyncFlushRetries { + // Max retries reached, fail the request + removePendingAsyncFlush(req.id) + errMsg := fmt.Sprintf("async flush failed after %d retries, inflight: %d, extentSize: %d", + maxAsyncFlushRetries, req.inflightCount, handler.size) + req.done <- errors.New(errMsg) + return + } + + // Retry the flush + req.retryCount++ + timeout := calculateAsyncFlushTimeout(int64(handler.size)) + req.timeout = now.Add(timeout) + req.inflightCount = atomic.LoadInt32(&handler.inflight) + + // Re-queue for retry + select { + case s.asyncFlushCh <- req: + log.LogDebugf("Re-queued async flush for retry, handler(%v), id(%v), retryCount(%v)", handler, req.id, req.retryCount) + default: + // Channel is full, process immediately + go s.retryAsyncFlush(handler, req) + } + return + } + + // Check if inflight count has decreased (packets completed) + currentInflight := atomic.LoadInt32(&handler.inflight) + if currentInflight < req.inflightCount { + req.inflightCount = currentInflight + if currentInflight == 0 { + log.LogDebugf("processAsyncFlushRequest: handler id %v", handler.id) + // All packets completed, process extent keys + go s.completeAsyncFlush(handler, req) + return + } + } + + // Re-queue for next check + select { + case s.asyncFlushCh <- req: + // Successfully re-queued + default: + log.LogDebugf("processAsyncFlushRequest: handler id %v", handler.id) + // Channel is full, process immediately + go s.completeAsyncFlush(handler, req) + } +} + +// retryAsyncFlush retries a failed async flush +func (s *Streamer) retryAsyncFlush(handler *ExtentHandler, req *AsyncFlushRequest) { + log.LogDebugf("Retrying async flush for handler(%v), id(%v), retryCount(%v)", handler, req.id, req.retryCount) + + // Remove from active handler tracking + defer removeActiveHandlerFlush(handler.id) + + // Wait for current inflight packets to complete + for atomic.LoadInt32(&handler.inflight) > 0 { + time.Sleep(time.Duration(asyncFlushCheckIntervalMs) * time.Millisecond) + } + + // Check if this request is the next one to be processed + nextReq := getNextPendingAsyncFlush() + if nextReq == nil { + log.LogErrorf("No pending async flush requests found for retry handler(%v), id(%v)", handler, req.id) + req.done <- errors.New("no pending async flush requests") + return + } + + // If this is not the next request to be processed, wait + if nextReq.id != req.id { + log.LogDebugf("Async flush retry request id(%v) is not next in sequence (next: %v), waiting...", req.id, nextReq.id) + + // Wait for the correct request to be processed + for { + time.Sleep(time.Duration(asyncFlushCheckIntervalMs) * time.Millisecond) + nextReq = getNextPendingAsyncFlush() + if nextReq == nil { + log.LogErrorf("No pending async flush requests found while waiting for retry id(%v)", req.id) + req.done <- errors.New("no pending async flush requests") + return + } + if nextReq.id == req.id { + break + } + } + } + + // Process extent keys (this is now guaranteed to be in sequence) + err := handler.appendExtentKey() + if err != nil { + log.LogErrorf("Async flush retry failed for handler(%v), id(%v): %v", handler, req.id, err) + removePendingAsyncFlush(req.id) + req.done <- err + return + } + + // Remove from pending map after successful completion + removePendingAsyncFlush(req.id) + + // Success + log.LogDebugf("Async flush retry completed successfully for handler(%v), id(%v)", handler, req.id) + req.done <- nil +} + +// completeAsyncFlush completes an async flush operation +func (s *Streamer) completeAsyncFlush(handler *ExtentHandler, req *AsyncFlushRequest) { + log.LogDebugf("Completing async flush for handler(%v), id(%v)", handler, req.id) + + // Remove from active handler tracking + defer removeActiveHandlerFlush(handler.id) + + // Check if this request is the next one to be processed + nextReq := getNextPendingAsyncFlush() + if nextReq == nil { + log.LogErrorf("No pending async flush requests found for handler(%v), id(%v)", handler, req.id) + req.done <- errors.New("no pending async flush requests") + return + } + + // If this is not the next request to be processed, wait + if nextReq.id != req.id { + log.LogDebugf("Async flush request id(%v) is not next in sequence (next: %v), waiting...", req.id, nextReq.id) + + // Wait for the correct request to be processed + // This is a simple polling approach - in a production system, you might want to use channels or condition variables + for { + time.Sleep(time.Duration(asyncFlushCheckIntervalMs) * time.Millisecond) + nextReq = getNextPendingAsyncFlush() + if nextReq == nil { + log.LogErrorf("No pending async flush requests found while waiting for id(%v)", req.id) + req.done <- errors.New("no pending async flush requests") + return + } + if nextReq.id == req.id { + break + } + } + } + + // Process extent keys (this is now guaranteed to be in sequence) + err := handler.appendExtentKey() + if err != nil { + log.LogErrorf("Async flush completion failed for handler(%v), id(%v): %v", handler, req.id, err) + removePendingAsyncFlush(req.id) + req.done <- err + return + } + + // Remove from pending map after successful completion + removePendingAsyncFlush(req.id) + + // Log success metric + log.LogDebugf("Async flush completed successfully for handler(%v), id(%v)", handler, req.id) + req.done <- nil +} + +// requestAsyncFlush initiates an asynchronous flush for a handler +func (s *Streamer) requestAsyncFlush(handler *ExtentHandler) chan error { + log.LogDebugf("requestAsyncFlush handler id %v", handler.id) + + // Check if this handler already has an active async flush request + if isHandlerFlushActive(handler.id) { + existingReq := getActiveHandlerFlush(handler.id) + if existingReq != nil { + log.LogDebugf("Handler %v already has active async flush request id(%v), returning existing", handler.id, existingReq.id) + return existingReq.done + } + } + + // Calculate dynamic timeout based on extent size + timeout := calculateAsyncFlushTimeout(int64(handler.size)) + + req := &AsyncFlushRequest{ + id: getNextAsyncFlushID(), // Assign sequential ID + handler: handler, + done: make(chan error, 1), + timeout: time.Now().Add(timeout), + retryCount: 0, + inflightCount: atomic.LoadInt32(&handler.inflight), + } + + // Add to active handler map to prevent duplicates + addActiveHandlerFlush(handler.id, req) + + // Add to pending map for sequencing + addPendingAsyncFlush(req) + + // Send to channel (non-blocking) + select { + case s.asyncFlushCh <- req: + log.LogDebugf("Requested async flush for handler(%v), id(%v), inflight(%v), timeout(%v)", + handler, req.id, req.inflightCount, timeout) + default: + // Channel is full, process immediately + log.LogWarnf("Async flush channel full, processing immediately for handler(%v), id(%v)", handler, req.id) + log.LogDebugf("requestAsyncFlush: handler id %v", handler.id) + go s.completeAsyncFlush(handler, req) + } + + return req.done +} + +// waitForAsyncFlush waits for an async flush to complete with timeout +func (s *Streamer) waitForAsyncFlush(handler *ExtentHandler, timeout time.Duration) error { + done := s.requestAsyncFlush(handler) + + select { + case err := <-done: + return err + case <-time.After(timeout): + // Timeout occurred, we can't easily remove from channel + // The request will be processed and timeout will be detected + return errors.New("async flush timeout") + } +} + +// cancelAsyncFlush cancels pending async flush for a handler +// Note: With channel-based approach, we can't easily cancel specific requests +// This is a limitation of the channel approach +func (s *Streamer) cancelAsyncFlush(handler *ExtentHandler) { + // With channel-based approach, we can't easily cancel specific requests + // The request will be processed and can check if handler is still valid + log.LogDebugf("Cancel requested for handler(%v) - will be handled during processing", handler) +} + +// handleHandlerClosed handles cleanup when a handler is closed during async flush +func (s *Streamer) handleHandlerClosed(handler *ExtentHandler) { + // With channel-based approach, we can't easily cancel specific requests + // Just ensure all in-flight packets are processed + for atomic.LoadInt32(&handler.inflight) > 0 { + time.Sleep(time.Duration(asyncFlushCheckIntervalMs) * time.Millisecond) + } + + log.LogDebugf("Handler closed cleanup completed for handler(%v)", handler) +} + +// getAsyncFlushStats returns statistics about async flush operations +func (s *Streamer) getAsyncFlushStats() map[string]interface{} { + stats := make(map[string]interface{}) + stats["channel_capacity"] = cap(s.asyncFlushCh) + stats["channel_length"] = len(s.asyncFlushCh) + stats["enable_async_flush"] = enableAsyncFlush + + // Add sequencer statistics + pendingAsyncFlushMutex.RLock() + stats["pending_requests_count"] = len(pendingAsyncFlushMap) + stats["current_sequencer_id"] = asyncFlushSequencer + pendingAsyncFlushMutex.RUnlock() + + return stats +} + +// cleanupAsyncFlushForStreamer removes all pending async flush requests for this streamer +func (s *Streamer) cleanupAsyncFlushForStreamer() { + pendingAsyncFlushMutex.Lock() + defer pendingAsyncFlushMutex.Unlock() + + // Find and remove all requests for this streamer's handlers + handlersToRemove := make([]uint64, 0) + for id, req := range pendingAsyncFlushMap { + if req.handler.stream == s { + handlersToRemove = append(handlersToRemove, id) + } + } + + // Remove the requests + for _, id := range handlersToRemove { + delete(pendingAsyncFlushMap, id) + log.LogDebugf("Cleaned up async flush request id(%v) for streamer(%v)", id, s.inode) + } + + if len(handlersToRemove) > 0 { + log.LogDebugf("Cleaned up %d async flush requests for streamer(%v)", len(handlersToRemove), s.inode) + } + + // Also clean up active handler requests for this streamer + activeHandlerFlushMutex.Lock() + defer activeHandlerFlushMutex.Unlock() + + activeHandlersToRemove := make([]uint64, 0) + for handlerID, req := range activeHandlerFlushMap { + if req.handler.stream == s { + activeHandlersToRemove = append(activeHandlersToRemove, handlerID) + } + } + + for _, handlerID := range activeHandlersToRemove { + delete(activeHandlerFlushMap, handlerID) + log.LogDebugf("Cleaned up active handler flush for handler(%v) in streamer(%v)", handlerID, s.inode) + } + + if len(activeHandlersToRemove) > 0 { + log.LogDebugf("Cleaned up %d active handler flush requests for streamer(%v)", len(activeHandlersToRemove), s.inode) + } +} + +// getAsyncFlushSequencerStats returns detailed statistics about the async flush sequencer +func getAsyncFlushSequencerStats() map[string]interface{} { + stats := make(map[string]interface{}) + + asyncFlushSequencerLock.Lock() + stats["current_sequencer_id"] = asyncFlushSequencer + asyncFlushSequencerLock.Unlock() + + pendingAsyncFlushMutex.RLock() + stats["pending_requests_count"] = len(pendingAsyncFlushMap) + + // Calculate oldest and newest request IDs + var oldestID, newestID uint64 + if len(pendingAsyncFlushMap) > 0 { + oldestID = ^uint64(0) // Max uint64 + newestID = 0 + for id := range pendingAsyncFlushMap { + if id < oldestID { + oldestID = id + } + if id > newestID { + newestID = id + } + } + } + stats["oldest_request_id"] = oldestID + stats["newest_request_id"] = newestID + pendingAsyncFlushMutex.RUnlock() + + return stats +} + +// Track active async flush requests per handler to prevent duplicates +var ( + activeHandlerFlushMap = make(map[uint64]*AsyncFlushRequest) // handler ID -> request + activeHandlerFlushMutex sync.RWMutex +) + +// isHandlerFlushActive checks if a handler already has an active async flush request +func isHandlerFlushActive(handlerID uint64) bool { + activeHandlerFlushMutex.RLock() + defer activeHandlerFlushMutex.RUnlock() + _, exists := activeHandlerFlushMap[handlerID] + return exists +} + +// addActiveHandlerFlush adds a handler to the active flush map +func addActiveHandlerFlush(handlerID uint64, req *AsyncFlushRequest) { + activeHandlerFlushMutex.Lock() + defer activeHandlerFlushMutex.Unlock() + activeHandlerFlushMap[handlerID] = req +} + +// removeActiveHandlerFlush removes a handler from the active flush map +func removeActiveHandlerFlush(handlerID uint64) { + activeHandlerFlushMutex.Lock() + defer activeHandlerFlushMutex.Unlock() + delete(activeHandlerFlushMap, handlerID) +} + +// getActiveHandlerFlush returns the active request for a handler +func getActiveHandlerFlush(handlerID uint64) *AsyncFlushRequest { + activeHandlerFlushMutex.RLock() + defer activeHandlerFlushMutex.RUnlock() + return activeHandlerFlushMap[handlerID] +} + +// Write protection functions +func (s *Streamer) startWriteProtection(handler *ExtentHandler) { + s.writeProtectionLock.Lock() + defer s.writeProtectionLock.Unlock() + s.writeInProgress = true + s.writeHandler = handler + log.LogDebugf("Write protection started for handler(%v) in streamer(%v)", handler, s.inode) +} + +func (s *Streamer) endWriteProtection() { + s.writeProtectionLock.Lock() + defer s.writeProtectionLock.Unlock() + s.writeInProgress = false + s.writeHandler = nil + log.LogDebugf("Write protection ended for streamer(%v)", s.inode) +} + +func (s *Streamer) isHandlerProtected(handler *ExtentHandler) bool { + s.writeProtectionLock.Lock() + defer s.writeProtectionLock.Unlock() + return s.writeInProgress && s.writeHandler == handler +} diff --git a/sdk/data/stream/stream_writer.go b/sdk/data/stream/stream_writer.go index 679ae0df9..ba5de84cd 100644 --- a/sdk/data/stream/stream_writer.go +++ b/sdk/data/stream/stream_writer.go @@ -19,6 +19,7 @@ import ( "fmt" "hash/crc32" "net" + "sync" "sync/atomic" "syscall" "time" @@ -29,6 +30,7 @@ import ( "github.com/cubefs/cubefs/util" "github.com/cubefs/cubefs/util/errors" "github.com/cubefs/cubefs/util/log" + "github.com/cubefs/cubefs/util/stat" ) const ( @@ -47,8 +49,79 @@ const ( const ( streamWriterFlushPeriod = 3 streamWriterIdleTimeoutPeriod = 10 + // Async flush constants + asyncFlushTimeoutMs = 5000 // 5 seconds timeout for async flush + asyncFlushCheckIntervalMs = 100 // Check every 100ms + maxAsyncFlushRetries = 3 // Max retries for failed async flush + enableAsyncFlush = true // Enable async flush by default + asyncFlushQueueSize = 128 // Buffer size for async flush channel + asyncFlushWorkerCount = 4 // Number of worker goroutines for async flush + asyncFlushSemaphoreSize = 4 // Capacity of semaphore to limit concurrent processAsyncFlushRequest executions ) +// Global sequencer for async flush requests to ensure appendExtentKey operations are executed in sequence +var ( + asyncFlushSequencer uint64 = 0 + asyncFlushSequencerLock sync.Mutex + pendingAsyncFlushMap = make(map[uint64]*AsyncFlushRequest) + pendingAsyncFlushMutex sync.RWMutex +) + +// getNextAsyncFlushID returns the next sequential ID for async flush requests +func getNextAsyncFlushID() uint64 { + asyncFlushSequencerLock.Lock() + defer asyncFlushSequencerLock.Unlock() + asyncFlushSequencer++ + return asyncFlushSequencer +} + +// addPendingAsyncFlush adds a request to the pending map +func addPendingAsyncFlush(req *AsyncFlushRequest) { + pendingAsyncFlushMutex.Lock() + defer pendingAsyncFlushMutex.Unlock() + pendingAsyncFlushMap[req.id] = req +} + +// removePendingAsyncFlush removes a request from the pending map +func removePendingAsyncFlush(id uint64) { + pendingAsyncFlushMutex.Lock() + defer pendingAsyncFlushMutex.Unlock() + delete(pendingAsyncFlushMap, id) +} + +// getNextPendingAsyncFlush returns the next pending request that should be processed +func getNextPendingAsyncFlush() *AsyncFlushRequest { + pendingAsyncFlushMutex.RLock() + defer pendingAsyncFlushMutex.RUnlock() + + if len(pendingAsyncFlushMap) == 0 { + return nil + } + + // Find the request with the smallest ID (oldest) + var oldestID uint64 = ^uint64(0) // Max uint64 + var oldestReq *AsyncFlushRequest + + for id, req := range pendingAsyncFlushMap { + if id < oldestID { + oldestID = id + oldestReq = req + } + } + + return oldestReq +} + +// AsyncFlushRequest represents an asynchronous flush request +type AsyncFlushRequest struct { + id uint64 // Sequential ID for ordering appendExtentKey operations + handler *ExtentHandler + done chan error + timeout time.Time + retryCount int + inflightCount int32 +} + // VerUpdateRequest defines an verseq update request. type VerUpdateRequest struct { err error @@ -214,6 +287,19 @@ func (s *Streamer) server() { s.traversed = 0 case <-s.done: s.abort() + // Clean up async flush system + select { + case <-s.asyncFlushDone: + // Channel already closed, do nothing + default: + close(s.asyncFlushDone) + } + select { + case <-s.asyncFlushCh: + // Channel already closed, do nothing + default: + close(s.asyncFlushCh) // Close the channel to signal asyncFlushManager to stop + } log.LogDebugf("done server: evict, streamer(%v)", s) return case <-t.C: @@ -294,7 +380,7 @@ func (s *Streamer) abortRequest(request interface{}) { func (s *Streamer) handleRequest(request interface{}) { if atomic.LoadInt32(&s.needUpdateVer) == 1 { - s.closeOpenHandler() + s.closeOpenHandler(true) atomic.StoreInt32(&s.needUpdateVer, 0) } @@ -761,7 +847,7 @@ func (s *Streamer) tryInitExtentHandlerByLastEk(offset, size int, isMigration bo if currentEK.GetSeq() != s.verSeq { log.LogDebugf("tryInitExtentHandlerByLastEk. exist ek seq %v vs request seq %v", currentEK.GetSeq(), s.verSeq) if int(currentEK.ExtentOffset)+int(currentEK.Size)+size > util.ExtentSize { - s.closeOpenHandler() + s.closeOpenHandler(false) return } isLastEkVerNotEqual = true @@ -795,7 +881,7 @@ func (s *Streamer) tryInitExtentHandlerByLastEk(offset, size int, isMigration bo if s.handler != nil { log.LogDebugf("tryInitExtentHandlerByLastEk: close old handler, currentEK.PartitionId(%v)", currentEK.PartitionId) - s.closeOpenHandler() + s.closeOpenHandler(false) } s.handler = handler @@ -874,21 +960,43 @@ func (s *Streamer) doWriteAppendEx(data []byte, offset, size int, direct bool, r return } } else if s.handler != nil { - s.closeOpenHandler() + // Close handler synchronously to avoid race conditions + err = s.closeOpenHandler(true) + if err != nil { + log.LogErrorf("doWriteAppendEx: closeOpenHandler failed, ino(%v) err(%v)", s.inode, err) + return + } + s.handler = nil } + log.LogDebugf("doWriteAppendEx: start write protection, handler(%v)", s.handler) + s.startWriteProtection(s.handler) + defer s.endWriteProtection() for i := 0; i < MaxNewHandlerRetry; i++ { + // Start write protection for this handler + log.LogDebugf("doWriteAppendEx: start write protection, handler(%v)", s.handler) if s.handler == nil { s.handler = NewExtentHandler(s, offset, storeMode, 0, storageClass, isMigration) s.dirty = false } else if s.handler.storeMode != storeMode { // store mode changed, so close open handler and start a new one - err = s.closeOpenHandler() + err = s.closeOpenHandler(true) // Use synchronous close if err != nil { + log.LogErrorf("doWriteAppendEx: closeOpenHandler failed during store mode change, ino(%v) err(%v)", s.inode, err) break } + s.handler = nil continue } + + // Check if handler is still valid before writing + if s.handler == nil { + log.LogErrorf("doWriteAppendEx: handler is nil after creation, ino(%v)", s.inode) + err = errors.New("handler is nil after creation") + break + } + + log.LogDebugf("doWriteAppendEx: start write protection, handler(%v)", s.handler) ek, err = s.handler.write(data, offset, size, direct) if err == nil && ek != nil { ek.SetSeq(s.verSeq) @@ -902,23 +1010,40 @@ func (s *Streamer) doWriteAppendEx(data []byte, offset, size int, direct bool, r log.LogDebugf("doWrite handler write failed so close open handler: ino(%v) offset(%v) size(%v) storeMode(%v) err(%v)", s.inode, offset, size, storeMode, err) - err = s.closeOpenHandler() + err = s.closeOpenHandler(false) if err != nil { + log.LogErrorf("doWriteAppendEx: closeOpenHandler failed after write error, ino(%v) err(%v)", s.inode, err) break } + s.handler = nil } } else { + log.LogDebugf("doWriteAppendEx: start write protection, handler(%v)", s.handler) s.handler = NewExtentHandler(s, offset, storeMode, 0, storageClass, isMigration) s.dirty = false - ek, err = s.handler.write(data, offset, size, direct) - if err == nil && ek != nil { - if !s.dirty { - s.dirtylist.Put(s.handler) - s.dirty = true + + // Check if handler is still valid before writing + if s.handler == nil { + log.LogErrorf("doWriteAppendEx: handler is nil after creation (non-hot path), ino(%v)", s.inode) + err = errors.New("handler is nil after creation") + } else { + ek, err = s.handler.write(data, offset, size, direct) + if err == nil && ek != nil { + if !s.dirty { + s.dirtylist.Put(s.handler) + s.dirty = true + } } } - err = s.closeOpenHandler() + // Close handler synchronously + closeErr := s.closeOpenHandler(true) + if closeErr != nil { + log.LogErrorf("doWriteAppendEx: closeOpenHandler failed (non-hot path), ino(%v) err(%v)", s.inode, closeErr) + if err == nil { + err = closeErr + } + } } if err != nil || ek == nil { @@ -926,6 +1051,10 @@ func (s *Streamer) doWriteAppendEx(data []byte, offset, size int, direct bool, r return } + bgTime := stat.BeginStat() + defer func() { + stat.EndStat("Streamer_extents_Append", err, bgTime, 1) + }() // This ek is just a local cache for PrepareWriteRequest, so ignore discard eks here. _ = s.extents.Append(ek, false) total = size @@ -942,7 +1071,17 @@ func (s *Streamer) flush() (err error) { eh := element.Value.(*ExtentHandler) log.LogDebugf("Streamer flush begin: eh(%v)", eh) - err = eh.flush() + + // Use async flush for better performance if enabled + if enableAsyncFlush && atomic.LoadInt32(&eh.inflight) > 0 { + // If there are in-flight packets, use async flush + log.LogDebugf("Streamer flush using async flush for eh(%v) with inflight(%v)", eh, atomic.LoadInt32(&eh.inflight)) + err = s.asyncFlushHandler(eh) + } else { + // No in-flight packets or async flush disabled, use synchronous flush + err = eh.flush() + } + if err != nil { log.LogErrorf("Streamer flush failed: eh(%v)", eh) return @@ -961,6 +1100,42 @@ func (s *Streamer) flush() (err error) { return } +// asyncFlushHandler performs asynchronous flush for a handler +func (s *Streamer) asyncFlushHandler(eh *ExtentHandler) error { + // Start async flush and return immediately + s.requestAsyncFlush(eh) + + // Return immediately - don't wait for completion + log.LogDebugf("Async flush initiated for handler(%v), returning immediately", eh) + return nil +} + +// asyncFlushHandlerWithWait performs asynchronous flush and waits for completion +func (s *Streamer) asyncFlushHandlerWithWait(eh *ExtentHandler, cleanFunc func()) error { + // Start async flush with timeout + go func() { + timeout := time.Duration(asyncFlushTimeoutMs) * time.Millisecond + if err := s.waitForAsyncFlush(eh, timeout); err != nil { + log.LogErrorf("asyncFlushHandlerWithWait: async flush failed, err %s", err.Error()) + return + } + // Call cleanup function when async flush completes + cleanFunc() + + // Only set handler to nil if it's still the same handler + // This prevents race conditions when multiple async flushes are running + s.writeLock.Lock() + if s.handler == eh { + log.LogDebugf("asyncFlushHandlerWithWait: setting handler to nil for handler(%v)", eh) + s.handler = nil + } else { + log.LogDebugf("asyncFlushHandlerWithWait: handler changed, not setting to nil. current(%v), param(%v)", s.handler, eh) + } + s.writeLock.Unlock() + }() + return nil +} + func (s *Streamer) traverse() (err error) { s.traversed++ length := s.dirtylist.Len() @@ -1007,7 +1182,11 @@ func (s *Streamer) traverse() (err error) { // note: The invocation of the closeOpenHandler function on an inode is serialized // close open handler, then flush data -func (s *Streamer) closeOpenHandler() (err error) { +func (s *Streamer) closeOpenHandler(wait bool) (err error) { + bgTime := stat.BeginStat() + defer func() { + stat.EndStat("Streamer_closeOpenHandler", err, bgTime, 1) + }() log.LogDebugf("closeOpenHandler: streamer(%v)", s) defer func() { log.LogDebugf("closeOpenHandler: close success, stream(%v)", s) @@ -1016,19 +1195,51 @@ func (s *Streamer) closeOpenHandler() (err error) { handler := s.handler if handler != nil { log.LogDebugf("closeOpenHandler: flush open handler now, eh(%v)", handler) - err = handler.flush() - if err != nil { - log.LogErrorf("closeOpenHandler: eh(%v) flush failed, err %s", handler, err.Error()) - return + + cleanFunc := func() { + if err != nil { + log.LogErrorf("closeOpenHandler: eh(%v) flush failed, err %s", handler, err.Error()) + } else { + // in case the current handler is not on the dirty list and will not get cleaned up + log.LogDebugf("closeOpenHandler need cleanup: eh(%v)", s.handler) + handler.cleanup() + } } - handler.setClosed() - if !s.dirty { - // in case the current handler is not on the dirty list and will not get cleaned up - log.LogDebugf("closeOpenHandler need cleanup: eh(%v)", s.handler) - s.handler.cleanup() + // Use async flush if there are in-flight packets and async flush is enabled + if enableAsyncFlush && !wait && atomic.LoadInt32(&handler.inflight) > 0 { + log.LogDebugf("closeOpenHandler: using async flush for handler(%v) with inflight(%v)", handler, atomic.LoadInt32(&handler.inflight)) + + // Check if this handler already has an active async flush + if isHandlerFlushActive(handler.id) { + log.LogDebugf("closeOpenHandler: handler(%v) already has active async flush, skipping", handler) + return nil + } + + // Check if this handler is protected by a write operation + if s.isHandlerProtected(handler) { + log.LogDebugf("closeOpenHandler: handler(%v) is protected by write operation, skipping", handler) + return nil + } + + if err = s.asyncFlushHandlerWithWait(handler, cleanFunc); err != nil { + log.LogErrorf("closeOpenHandler: async flush failed, err %s", err.Error()) + return + } + // Don't set handler to nil here - let async flush complete first + } else { + // Check if this handler is protected by a write operation + if s.isHandlerProtected(handler) { + log.LogDebugf("closeOpenHandler: handler(%v) is protected by write operation, skipping", handler) + return nil + } + + err = handler.flush() + cleanFunc() + // For synchronous operations, it's safe to set handler to nil immediately + s.handler = nil } - s.handler = nil + } err = s.flush() @@ -1051,7 +1262,7 @@ func (s *Streamer) release() error { if s.client.AheadRead != nil { s.aheadReadEnable = s.client.AheadRead.enable } - err := s.closeOpenHandler() + err := s.closeOpenHandler(true) if err != nil { s.abort() } @@ -1069,10 +1280,31 @@ func (s *Streamer) evict() error { delete(s.client.streamers, s.inode) } s.client.streamerLock.Unlock() + + // Clean up any pending async flush requests for this streamer + s.cleanupAsyncFlushForStreamer() + + // Close async flush channels to signal cleanup + select { + case <-s.asyncFlushDone: + // Channel already closed, do nothing + default: + close(s.asyncFlushDone) + } + select { + case <-s.asyncFlushCh: + // Channel already closed, do nothing + default: + close(s.asyncFlushCh) + } + return nil } func (s *Streamer) abort() { + // Clean up any pending async flush requests for this streamer + s.cleanupAsyncFlushForStreamer() + for { element := s.dirtylist.Get() if element == nil { @@ -1090,7 +1322,7 @@ func (s *Streamer) truncate(size int, fullPath string) error { if atomic.LoadInt32(&s.status) >= StreamerError { return errors.New(fmt.Sprintf("IssueWriteRequest: stream writer in error status, ino(%v)", s.inode)) } - err := s.closeOpenHandler() + err := s.closeOpenHandler(true) if err != nil { return err } @@ -1127,3 +1359,19 @@ func (s *Streamer) tinySizeLimit() int { func (s *Streamer) setError() { atomic.StoreInt32(&s.status, StreamerError) } + +// calculateAsyncFlushTimeout calculates timeout based on extent size +func calculateAsyncFlushTimeout(extentSize int64) time.Duration { + // Base timeout for small extents + baseTimeout := time.Duration(asyncFlushTimeoutMs) * time.Millisecond + + // For large extents, increase timeout proportionally + // 128MB = 134217728 bytes + if extentSize > 64*1024*1024 { // 64MB + // Add 1 second per 64MB + additionalSeconds := int(extentSize / (64 * 1024 * 1024)) + return baseTimeout + time.Duration(additionalSeconds)*time.Second + } + + return baseTimeout +}