mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-12 01:07:35 +02:00
Two separate failures reported on 32-bit builds (void-linux 4.21): - weed/server: errorStreamImpl.count (and the same pattern in slowStream plus local totalEventsSent/totalSends) was a bare int64 sitting after smaller fields, so on 386/ARMv7/mips32 it landed at a 4-byte-aligned offset and atomic.AddInt64 panicked with "unaligned 64-bit atomic operation". Switched the counters to atomic.Int64, which Go guarantees is 8-byte aligned on every architecture. - weed/plugin/worker/iceberg: three equality-delete tests fail on 32-bit because the upstream github.com/apache/iceberg-go declares manifestEntry.EqualityIDs as *[]int while the Iceberg Avro schema defines equality_ids as long, and hamba/avro refuses to map Go int onto Avro long when int is 32-bit. Not fixable in seaweedfs, so guard the affected tests with a t.Skip() when unsafe.Sizeof(int) < 8 until the upstream type is changed to []int32/[]int64.
This commit is contained in:
1 parent
0a5c22b57e
commit
141413ad76
3 files changed
+62
-23
No files matched your search
@@ -2668,6 +2668,7 @@ func TestCompactDataFilesWithPositionDeletes(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestCompactDataFilesWithEqualityDeletes(t *testing.T) {
|
||||
skipIfEqualityDeleteAvroUnsupported(t)
|
||||
fs, client := startFakeFiler(t)
|
||||
|
||||
setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
|
||||
@@ -2820,6 +2821,7 @@ func TestCompactDataFilesApplyDeletesDisabled(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestCompactDataFilesWithMixedDeletes(t *testing.T) {
|
||||
skipIfEqualityDeleteAvroUnsupported(t)
|
||||
fs, client := startFakeFiler(t)
|
||||
|
||||
setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
|
||||
@@ -3553,6 +3555,7 @@ func TestRewritePositionDeleteFilesPreservesUnsupportedMultiTargetDeletes(t *tes
|
||||
}
|
||||
|
||||
func TestRewritePositionDeleteFilesRebuildsMixedDeleteManifests(t *testing.T) {
|
||||
skipIfEqualityDeleteAvroUnsupported(t)
|
||||
fs, client := startFakeFiler(t)
|
||||
|
||||
setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
package iceberg
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"unsafe"
|
||||
)
|
||||
|
||||
// skipIfEqualityDeleteAvroUnsupported skips the test when running on a 32-bit
|
||||
// architecture. The upstream github.com/apache/iceberg-go library declares
|
||||
// manifestEntry.EqualityIDs as *[]int, but the Iceberg Avro schema declares
|
||||
// equality_ids as a long array. hamba/avro rejects []int for Avro long when
|
||||
// int is 32-bit (386, arm, mips), causing these tests to fail through no
|
||||
// fault of seaweedfs. Tracked upstream; remove this guard once iceberg-go
|
||||
// switches the field to []int32/[]int64.
|
||||
func skipIfEqualityDeleteAvroUnsupported(t *testing.T) {
|
||||
t.Helper()
|
||||
if unsafe.Sizeof(int(0)) < 8 {
|
||||
t.Skip("equality-delete Avro decoding is broken on 32-bit platforms " +
|
||||
"because github.com/apache/iceberg-go declares EqualityIDs as *[]int " +
|
||||
"but the Iceberg schema requires Avro long")
|
||||
}
|
||||
}
|
||||
@@ -14,16 +14,19 @@ import (
|
||||
|
||||
// slowStream simulates a gRPC stream with configurable per-Send latency.
|
||||
// It counts individual events including those packed inside batches.
|
||||
// Atomic counters use atomic.Int64 so they stay 8-byte aligned on 32-bit
|
||||
// architectures (386, ARMv7, mips32) where a bare int64 struct field is
|
||||
// only 4-byte aligned and panics under atomic.AddInt64.
|
||||
type slowStream struct {
|
||||
sends atomic.Int64 // number of stream.Send() calls
|
||||
eventsSent atomic.Int64 // total events (1 + len(Events) per Send)
|
||||
sendDelay time.Duration
|
||||
sends int64 // number of stream.Send() calls
|
||||
eventsSent int64 // total events (1 + len(Events) per Send)
|
||||
}
|
||||
|
||||
func (s *slowStream) Send(msg *filer_pb.SubscribeMetadataResponse) error {
|
||||
time.Sleep(s.sendDelay)
|
||||
atomic.AddInt64(&s.sends, 1)
|
||||
atomic.AddInt64(&s.eventsSent, 1+int64(len(msg.Events)))
|
||||
s.sends.Add(1)
|
||||
s.eventsSent.Add(1 + int64(len(msg.Events)))
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -114,9 +117,9 @@ func TestPipelinedSenderThroughput(t *testing.T) {
|
||||
}
|
||||
elapsed := time.Since(start)
|
||||
|
||||
directRate = float64(stream.eventsSent) / elapsed.Seconds()
|
||||
directRate = float64(stream.eventsSent.Load()) / elapsed.Seconds()
|
||||
t.Logf("direct: %d events %4d sends %v %6.0f events/sec",
|
||||
stream.eventsSent, stream.sends, elapsed.Round(time.Millisecond), directRate)
|
||||
stream.eventsSent.Load(), stream.sends.Load(), elapsed.Round(time.Millisecond), directRate)
|
||||
})
|
||||
|
||||
// --- Pipelined + batched (new behavior): file reads overlap with batched sends ---
|
||||
@@ -139,9 +142,9 @@ func TestPipelinedSenderThroughput(t *testing.T) {
|
||||
}
|
||||
elapsed := time.Since(start)
|
||||
|
||||
batchedRate = float64(stream.eventsSent) / elapsed.Seconds()
|
||||
batchedRate = float64(stream.eventsSent.Load()) / elapsed.Seconds()
|
||||
t.Logf("pipelined+batch: %d events %4d sends %v %6.0f events/sec",
|
||||
stream.eventsSent, stream.sends, elapsed.Round(time.Millisecond), batchedRate)
|
||||
stream.eventsSent.Load(), stream.sends.Load(), elapsed.Round(time.Millisecond), batchedRate)
|
||||
})
|
||||
|
||||
if directRate > 0 {
|
||||
@@ -216,11 +219,13 @@ func TestBatchingAdaptive(t *testing.T) {
|
||||
}
|
||||
sender.Close()
|
||||
|
||||
sends := stream.sends.Load()
|
||||
events := stream.eventsSent.Load()
|
||||
t.Logf("old events: %d events in %d sends (avg batch size: %.1f)",
|
||||
stream.eventsSent, stream.sends, float64(stream.eventsSent)/float64(stream.sends))
|
||||
events, sends, float64(events)/float64(sends))
|
||||
|
||||
if stream.sends >= int64(numEvents) {
|
||||
t.Errorf("expected batching to reduce sends below %d, got %d", numEvents, stream.sends)
|
||||
if sends >= int64(numEvents) {
|
||||
t.Errorf("expected batching to reduce sends below %d, got %d", numEvents, sends)
|
||||
}
|
||||
})
|
||||
|
||||
@@ -233,24 +238,29 @@ func TestBatchingAdaptive(t *testing.T) {
|
||||
}
|
||||
sender.Close()
|
||||
|
||||
sends := stream.sends.Load()
|
||||
events := stream.eventsSent.Load()
|
||||
t.Logf("recent events: %d events in %d sends (avg batch size: %.1f)",
|
||||
stream.eventsSent, stream.sends, float64(stream.eventsSent)/float64(stream.sends))
|
||||
events, sends, float64(events)/float64(sends))
|
||||
|
||||
if stream.sends != int64(numEvents) {
|
||||
t.Errorf("expected 1:1 sends for recent events, got %d sends for %d events", stream.sends, numEvents)
|
||||
if sends != int64(numEvents) {
|
||||
t.Errorf("expected 1:1 sends for recent events, got %d sends for %d events", sends, numEvents)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// errorStreamImpl is a metadataStreamSender that returns an error after N sends.
|
||||
// count uses atomic.Int64 so it stays 8-byte aligned on 32-bit architectures
|
||||
// (386, ARMv7, mips32) where a bare int64 struct field after smaller fields
|
||||
// is only 4-byte aligned and panics under atomic.AddInt64.
|
||||
type errorStreamImpl struct {
|
||||
count atomic.Int64
|
||||
failAfter int
|
||||
err error
|
||||
count int64
|
||||
}
|
||||
|
||||
func (s *errorStreamImpl) Send(msg *filer_pb.SubscribeMetadataResponse) error {
|
||||
n := atomic.AddInt64(&s.count, 1)
|
||||
n := s.count.Add(1)
|
||||
if int(n) > s.failAfter {
|
||||
return s.err
|
||||
}
|
||||
@@ -352,8 +362,8 @@ func TestPipelinedSingleVsParallelStreams(t *testing.T) {
|
||||
err = closeErr
|
||||
}
|
||||
elapsed = time.Since(start)
|
||||
eventsSent = atomic.LoadInt64(&stream.eventsSent)
|
||||
sends = atomic.LoadInt64(&stream.sends)
|
||||
eventsSent = stream.eventsSent.Load()
|
||||
sends = stream.sends.Load()
|
||||
return
|
||||
}
|
||||
|
||||
@@ -370,7 +380,10 @@ func TestPipelinedSingleVsParallelStreams(t *testing.T) {
|
||||
|
||||
var parallelRate float64
|
||||
t.Run("10_pipelined_streams", func(t *testing.T) {
|
||||
var totalEventsSent, totalSends int64
|
||||
// atomic.Int64 guarantees 8-byte alignment on 32-bit architectures where
|
||||
// a local int64 variable's address is only 4-byte aligned and atomic
|
||||
// 64-bit operations panic with "unaligned 64-bit atomic operation".
|
||||
var totalEventsSent, totalSends atomic.Int64
|
||||
var wg sync.WaitGroup
|
||||
|
||||
start := time.Now()
|
||||
@@ -379,16 +392,17 @@ func TestPipelinedSingleVsParallelStreams(t *testing.T) {
|
||||
go func(files []logFile) {
|
||||
defer wg.Done()
|
||||
eventsSent, sends, _, _ := simulatePipeline(files)
|
||||
atomic.AddInt64(&totalEventsSent, eventsSent)
|
||||
atomic.AddInt64(&totalSends, sends)
|
||||
totalEventsSent.Add(eventsSent)
|
||||
totalSends.Add(sends)
|
||||
}(partitions[d])
|
||||
}
|
||||
wg.Wait()
|
||||
elapsed := time.Since(start)
|
||||
|
||||
parallelRate = float64(totalEventsSent) / elapsed.Seconds()
|
||||
totalEvents := totalEventsSent.Load()
|
||||
parallelRate = float64(totalEvents) / elapsed.Seconds()
|
||||
t.Logf("%d streams: %5d events %4d sends %v %7.0f events/sec",
|
||||
numDirs, totalEventsSent, totalSends, elapsed.Round(time.Millisecond), parallelRate)
|
||||
numDirs, totalEvents, totalSends.Load(), elapsed.Round(time.Millisecond), parallelRate)
|
||||
})
|
||||
|
||||
if singleRate > 0 && parallelRate > 0 {
|
||||
|
||||
Reference in new issue
Block a user