mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-02 02:00:56 +00:00
fix(master): add some log to locate migrate mp problem. #1000295064
Signed-off-by: zhumingze <zhumingze@oppo.com>
(cherry picked from commit f6359bf1e8)
This commit is contained in:
zhumingze
chihe
parent
1803b18700
commit
ff1f157f20
@ -22,6 +22,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/cubefs/cubefs/proto"
|
||||
"github.com/cubefs/cubefs/util/auditlog"
|
||||
"github.com/cubefs/cubefs/util/errors"
|
||||
"github.com/cubefs/cubefs/util/log"
|
||||
)
|
||||
@ -105,6 +106,7 @@ func (c *Cluster) migrateMetaPartition(srcAddr, targetAddr string, mp *MetaParti
|
||||
finalHosts []string
|
||||
oldHosts []string
|
||||
zones []string
|
||||
auditMsg string
|
||||
)
|
||||
|
||||
log.LogWarnf("action[migrateMetaPartition],volName[%v], migrate from src[%s] to target[%s],partitionID[%v] begin",
|
||||
@ -174,6 +176,10 @@ func (c *Cluster) migrateMetaPartition(srcAddr, targetAddr string, mp *MetaParti
|
||||
}
|
||||
}
|
||||
|
||||
auditMsg = fmt.Sprintf("volName[%v] partitionID[%v] hosts[%v] srcAddr[%v] choose targetAddr[%v]",
|
||||
mp.volName, mp.PartitionID, mp.Hosts, srcAddr, newPeers)
|
||||
auditlog.LogMasterOp("migrateMetaPartition", auditMsg, err)
|
||||
|
||||
finalHosts = make([]string, 0, len(oldHosts))
|
||||
for _, host := range oldHosts {
|
||||
if host != srcAddr {
|
||||
@ -539,7 +545,7 @@ func (c *Cluster) addMetaReplica(partition *MetaPartition, addr string) (err err
|
||||
partition.Lock()
|
||||
defer partition.Unlock()
|
||||
if contains(partition.Hosts, addr) {
|
||||
err = fmt.Errorf("vol[%v],mp[%v] has contains host[%v]", partition.volName, partition.PartitionID, addr)
|
||||
err = fmt.Errorf("vol[%v],mp[%v] hosts[%v] has contains host[%v]", partition.volName, partition.PartitionID, partition.Hosts, addr)
|
||||
return
|
||||
}
|
||||
metaNode, err := c.metaNode(addr)
|
||||
|
||||
@ -1,117 +1,12 @@
|
||||
package stream
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/cubefs/cubefs/util/log"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestAsyncFlushSequencer(t *testing.T) {
|
||||
// Create a test streamer to test local sequencer
|
||||
client := &ExtentClient{}
|
||||
streamer := NewStreamer(client, 1, true, false, "/test")
|
||||
|
||||
// Test ID generation
|
||||
id1 := streamer.getNextAsyncFlushID()
|
||||
id2 := streamer.getNextAsyncFlushID()
|
||||
id3 := streamer.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}
|
||||
|
||||
streamer.addPendingAsyncFlush(1, req1)
|
||||
streamer.addPendingAsyncFlush(2, req2)
|
||||
streamer.addPendingAsyncFlush(3, req3)
|
||||
|
||||
// Test getNextPendingAsyncFlush returns the oldest request
|
||||
next := streamer.getNextPendingAsyncFlush()
|
||||
assert.Equal(t, uint64(1), next.handler.id)
|
||||
|
||||
// Test removal
|
||||
streamer.removePendingAsyncFlush(1)
|
||||
next = streamer.getNextPendingAsyncFlush()
|
||||
assert.Equal(t, uint64(2), next.handler.id)
|
||||
|
||||
streamer.removePendingAsyncFlush(2)
|
||||
next = streamer.getNextPendingAsyncFlush()
|
||||
assert.Equal(t, uint64(3), next.handler.id)
|
||||
|
||||
streamer.removePendingAsyncFlush(3)
|
||||
next = streamer.getNextPendingAsyncFlush()
|
||||
assert.Nil(t, next)
|
||||
}
|
||||
|
||||
func TestAsyncFlushSequencerConcurrency(t *testing.T) {
|
||||
// Create a test streamer to test local sequencer
|
||||
client := &ExtentClient{}
|
||||
streamer := NewStreamer(client, 1, true, false, "/test")
|
||||
|
||||
// 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] = streamer.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) {
|
||||
// Create a test streamer to test local sequencer
|
||||
client := &ExtentClient{}
|
||||
streamer := NewStreamer(client, 1, true, false, "/test")
|
||||
|
||||
// Test stats with no requests
|
||||
stats := streamer.getAsyncFlushStats()
|
||||
assert.Equal(t, uint64(0), stats["current_sequencer_id"])
|
||||
assert.Equal(t, 0, stats["pending_requests_count"])
|
||||
|
||||
// Add some requests
|
||||
id1 := streamer.getNextAsyncFlushID()
|
||||
id2 := streamer.getNextAsyncFlushID()
|
||||
id3 := streamer.getNextAsyncFlushID()
|
||||
|
||||
req1 := &AsyncFlushRequest{id: id1}
|
||||
req2 := &AsyncFlushRequest{id: id2}
|
||||
req3 := &AsyncFlushRequest{id: id3}
|
||||
|
||||
streamer.addPendingAsyncFlush(1, req1)
|
||||
streamer.addPendingAsyncFlush(2, req2)
|
||||
streamer.addPendingAsyncFlush(3, req3)
|
||||
|
||||
// Test statistics
|
||||
stats = streamer.getAsyncFlushStats()
|
||||
|
||||
assert.Equal(t, uint64(3), stats["current_sequencer_id"])
|
||||
assert.Equal(t, 3, stats["pending_requests_count"])
|
||||
|
||||
// Clean up
|
||||
streamer.removePendingAsyncFlush(1)
|
||||
streamer.removePendingAsyncFlush(2)
|
||||
streamer.removePendingAsyncFlush(3)
|
||||
}
|
||||
|
||||
func TestChannelCloseSafety(t *testing.T) {
|
||||
// Test that we can safely close channels multiple times
|
||||
done := make(chan struct{})
|
||||
@ -140,44 +35,6 @@ func TestChannelCloseSafety(t *testing.T) {
|
||||
assert.True(t, true)
|
||||
}
|
||||
|
||||
func TestDuplicateHandlerFlushPrevention(t *testing.T) {
|
||||
// Test the active handler tracking functions directly
|
||||
client := &ExtentClient{}
|
||||
streamer := NewStreamer(client, 1, true, false, "/test")
|
||||
|
||||
// Initially no active handlers
|
||||
assert.False(t, streamer.isHandlerFlushActive(123))
|
||||
|
||||
// Create a mock request
|
||||
req := &AsyncFlushRequest{
|
||||
id: 1,
|
||||
}
|
||||
|
||||
// Add handler to pending map
|
||||
streamer.addPendingAsyncFlush(123, req)
|
||||
assert.True(t, streamer.isHandlerFlushActive(123))
|
||||
|
||||
// Get the active request
|
||||
retrievedReq := streamer.getActiveHandlerFlush(123)
|
||||
assert.Equal(t, req, retrievedReq)
|
||||
|
||||
// Remove handler from pending map
|
||||
streamer.removePendingAsyncFlush(123)
|
||||
assert.False(t, streamer.isHandlerFlushActive(123))
|
||||
|
||||
// Test with multiple handlers
|
||||
streamer.addPendingAsyncFlush(456, req)
|
||||
streamer.addPendingAsyncFlush(789, req)
|
||||
assert.True(t, streamer.isHandlerFlushActive(456))
|
||||
assert.True(t, streamer.isHandlerFlushActive(789))
|
||||
|
||||
// Clean up
|
||||
streamer.removePendingAsyncFlush(456)
|
||||
streamer.removePendingAsyncFlush(789)
|
||||
assert.False(t, streamer.isHandlerFlushActive(456))
|
||||
assert.False(t, streamer.isHandlerFlushActive(789))
|
||||
}
|
||||
|
||||
func TestAsyncFlushRaceConditionFix(t *testing.T) {
|
||||
// Test the race condition fix logic
|
||||
|
||||
|
||||
@ -795,40 +795,6 @@ func (s *Streamer) waitForAsyncFlush(handler *ExtentHandler, timeout time.Durati
|
||||
}
|
||||
}
|
||||
|
||||
// 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"] = s.client.enableAsyncFlush
|
||||
|
||||
// Add sequencer statistics
|
||||
stats["pending_requests"] = s.getPendingRequests()
|
||||
stats["current_sequencer_id"] = atomic.LoadUint64(&s.asyncFlushSequencer)
|
||||
|
||||
return stats
|
||||
}
|
||||
|
||||
// getAsyncFlushSequencerStats returns detailed statistics about the async flush sequencer
|
||||
// Note: This function is deprecated as sequencer is now per-streamer
|
||||
func (s *Streamer) getAsyncFlushSequencerStats() map[string]interface{} {
|
||||
|
||||
@ -53,7 +53,6 @@ const (
|
||||
asyncFlushCheckIntervalMs = 100 // Check every 100ms
|
||||
maxAsyncFlushRetries = 3 // Max retries for failed async flush
|
||||
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
|
||||
)
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
