diff --git a/internal/adapter/mxlfabrics/doc.go b/internal/adapter/mxlfabrics/doc.go new file mode 100644 index 0000000..551b1ef --- /dev/null +++ b/internal/adapter/mxlfabrics/doc.go @@ -0,0 +1,2 @@ +// Package mxlfabrics provides MXL Fabrics ingress adapters +package mxlfabrics diff --git a/internal/adapter/mxlfabrics/grain_transfer_integration_test.go b/internal/adapter/mxlfabrics/grain_transfer_integration_test.go new file mode 100644 index 0000000..af2f489 --- /dev/null +++ b/internal/adapter/mxlfabrics/grain_transfer_integration_test.go @@ -0,0 +1,341 @@ +//go:build mxl_integration + +package mxlfabrics_test + +import ( + "errors" + "os" + "sync" + "testing" + "time" + + "github.com/qvest-digital/go-mxl/fabrics" + "github.com/qvest-digital/go-mxl/mxl" +) + +const testVideoFlowID = "5fbec3b1-1b0f-417d-9059-8b94a47197ed" + +const testVideoFlow = `{ + "description": "MXL Player Fabrics SHM integration test", + "id": "5fbec3b1-1b0f-417d-9059-8b94a47197ed", + "format": "urn:x-nmos:format:video", + "label": "Fabrics SHM test video", + "tags": { + "urn:x-nmos:tag:grouphint/v1.0": [ + "mxl-player-fabrics-test:Video" + ] + }, + "parents": [], + "media_type": "video/v210", + "grain_rate": { + "numerator": 25, + "denominator": 1 + }, + "frame_width": 1920, + "frame_height": 1080, + "interlace_mode": "progressive", + "colorspace": "BT709", + "components": [ + { + "name": "Y", + "width": 1920, + "height": 1080, + "bit_depth": 10 + }, + { + "name": "Cb", + "width": 960, + "height": 1080, + "bit_depth": 10 + }, + { + "name": "Cr", + "width": 960, + "height": 1080, + "bit_depth": 10 + } + ] +}` + +func newTestDomain(t *testing.T) *mxl.Instance { + t.Helper() + + domain, err := os.MkdirTemp("/dev/shm", "mxl-player-fabrics-*") + if err != nil { + t.Fatalf("create temporary MXL domain: %v", err) + } + t.Cleanup(func() { + if err := os.RemoveAll(domain); err != nil { + t.Errorf("remove temporary MXL domain: %v", err) + } + }) + + instance, err := mxl.NewInstance(domain, "") + if err != nil { + t.Fatalf("create MXL instance: %v", err) + } + t.Cleanup(func() { + if err := instance.Close(); err != nil { + t.Errorf("close MXL instance: %v", err) + } + }) + + return instance +} + +func TestGrainTransferSHM(t *testing.T) { + sourceInstance := newTestDomain(t) + targetInstance := newTestDomain(t) + + sourceWriter, _, err := sourceInstance.NewWriter(testVideoFlow) + if err != nil { + t.Fatalf("create source writer: %v", err) + } + t.Cleanup(func() { + if err := sourceWriter.Close(); err != nil { + t.Errorf("close source writer: %v", err) + } + }) + + sourceReader, err := sourceInstance.NewReader(testVideoFlowID) + if err != nil { + t.Fatalf("create source reader: %v", err) + } + t.Cleanup(func() { + if err := sourceReader.Close(); err != nil { + t.Errorf("close source reader: %v", err) + } + }) + + targetWriter, _, err := targetInstance.NewWriter(testVideoFlow) + if err != nil { + t.Fatalf("create target writer: %v", err) + } + t.Cleanup(func() { + if err := targetWriter.Close(); err != nil { + t.Errorf("close target writer: %v", err) + } + }) + + sourceFabrics, err := fabrics.NewInstance(sourceInstance) + if err != nil { + t.Fatalf("create source Fabrics instance: %v", err) + } + t.Cleanup(func() { + if err := sourceFabrics.Close(); err != nil { + t.Errorf("close source Fabrics instance: %v", err) + } + }) + + targetFabrics, err := fabrics.NewInstance(targetInstance) + if err != nil { + t.Fatalf("create target Fabrics instance: %v", err) + } + t.Cleanup(func() { + if err := targetFabrics.Close(); err != nil { + t.Errorf("close target Fabrics instance: %v", err) + } + }) + + target, err := targetFabrics.NewTarget() + if err != nil { + t.Fatalf("create target: %v", err) + } + t.Cleanup(func() { + if err := target.Close(); err != nil { + t.Errorf("close target: %v", err) + } + }) + + targetInfo, err := target.Setup(fabrics.TargetConfig{ + Interface: requireSHMInterface(t, targetFabrics), + Writer: targetWriter, + }) + if err != nil { + t.Fatalf("set up target: %v", err) + } + t.Cleanup(func() { + if err := targetInfo.Close(); err != nil { + t.Errorf("close target info: %v", err) + } + }) + + initiator, err := sourceFabrics.NewInitiator() + if err != nil { + t.Fatalf("create initiator: %v", err) + } + t.Cleanup(func() { + if err := initiator.Close(); err != nil { + t.Errorf("close initiator: %v", err) + } + }) + + if err := initiator.Setup(fabrics.InitiatorConfig{ + Interface: requireSHMInterface(t, sourceFabrics), + Reader: sourceReader, + }); err != nil { + t.Fatalf("set up initiator: %v", err) + } + + if err := initiator.AddTarget(targetInfo); err != nil { + t.Fatalf("add target: %v", err) + } + + connectDeadline := time.Now().Add(5 * time.Second) + + for { + if time.Now().After(connectDeadline) { + t.Fatal("SHM initiator did not become ready") + } + + // The target may also need to advance its endpoint state. + _, err := target.ReadGrainNonBlocking() + if err != nil && !errors.Is(err, fabrics.ErrNotReady) { + t.Fatalf("progress target during setup: %v", err) + } + + err = initiator.MakeProgressNonBlocking() + if err == nil { + break + } + if !errors.Is(err, fabrics.ErrNotReady) { + t.Fatalf("progress initiator during setup: %v", err) + } + + time.Sleep(time.Millisecond) + } + + index := mxl.CurrentIndex(sourceWriter.Config().Common.GrainRate) + + sourceGrain, err := sourceWriter.OpenGrain(index) + if err != nil { + t.Fatalf("open source grain: %v", err) + } + + for offset := range sourceGrain.Payload { + sourceGrain.Payload[offset] = byte((uint64(offset) + index) & 0xff) + } + + totalSlices := sourceGrain.TotalSlices + if err := sourceGrain.Commit(totalSlices, 0); err != nil { + t.Fatalf("commit source grain: %v", err) + } + + deadline := time.Now().Add(5 * time.Second) + received := make(chan uint64, 1) + targetErrors := make(chan error, 1) + stopTarget := make(chan struct{}) + var targetWG sync.WaitGroup + + targetWG.Add(1) + go func() { + defer targetWG.Done() + + for { + select { + case <-stopTarget: + return + default: + } + + receivedIndex, err := target.ReadGrainNonBlocking() + switch { + case err == nil: + received <- receivedIndex + return + case errors.Is(err, fabrics.ErrNotReady): + time.Sleep(time.Millisecond) + default: + targetErrors <- err + return + } + } + }() + defer func() { + close(stopTarget) + targetWG.Wait() + }() + + // The SHM provider can apply backpressure while the peer progresses its + // endpoint. ErrNotReady means the write was not queued, so it is safe to + // make progress and retry the transfer. + for { + err := initiator.TransferGrain(index, 0, totalSlices) + if err == nil { + break + } + if !errors.Is(err, fabrics.ErrNotReady) { + t.Fatalf("transfer grain: %v", err) + } + if time.Now().After(deadline) { + t.Fatal("timed out enqueueing SHM grain transfer") + } + + err = initiator.MakeProgressNonBlocking() + if err != nil && !errors.Is(err, fabrics.ErrNotReady) { + t.Fatalf("make initiator progress while enqueueing: %v", err) + } + + time.Sleep(time.Millisecond) + } + + var receivedIndex uint64 + + for { + if time.Now().After(deadline) { + t.Fatal("SHM grain transfer timed out") + } + + err := initiator.MakeProgressNonBlocking() + if err != nil && !errors.Is(err, fabrics.ErrNotReady) { + t.Fatalf("make initiator progress: %v", err) + } + + select { + case receivedIndex = <-received: + goto receivedGrain + case err := <-targetErrors: + t.Fatalf("read target completion: %v", err) + default: + time.Sleep(time.Millisecond) + } + } + +receivedGrain: + if receivedIndex != index { + t.Fatalf( + "received grain index = %d, want %d", + receivedIndex, + index, + ) + } + + // Fabrics writes directly into the target writer's mapped grain memory. + // Open the completed slot to inspect those bytes without relying on the + // local reader head, which the transfer itself does not advance. + receivedGrain, err := targetWriter.OpenGrain(index) + if err != nil { + t.Fatalf("open transferred grain: %v", err) + } + defer func() { + if err := receivedGrain.Cancel(); err != nil { + t.Errorf("cancel transferred grain: %v", err) + } + }() + + if len(receivedGrain.Payload) == 0 { + t.Fatal("transferred grain has an empty payload") + } + + for offset := 0; offset < len(receivedGrain.Payload); offset += 4096 { + want := byte((uint64(offset) + index) & 0xff) + if got := receivedGrain.Payload[offset]; got != want { + t.Fatalf( + "payload[%d] = %d, want %d", + offset, + got, + want, + ) + } + } +} diff --git a/internal/adapter/mxlfabrics/interface_integration_test.go b/internal/adapter/mxlfabrics/interface_integration_test.go new file mode 100644 index 0000000..3f0af96 --- /dev/null +++ b/internal/adapter/mxlfabrics/interface_integration_test.go @@ -0,0 +1,89 @@ +//go:build mxl_integration + +package mxlfabrics_test + +import ( + "os" + "testing" + + "github.com/qvest-digital/go-mxl/fabrics" + "github.com/qvest-digital/go-mxl/mxl" +) + +func requireSHMInterface( + t *testing.T, + instance *fabrics.Instance, +) fabrics.InterfaceConfig { + t.Helper() + + interfaces, err := instance.Interfaces(&fabrics.InterfaceConfig{ + Provider: fabrics.ProviderSHM, + }) + if err != nil { + t.Fatalf("enumerate SHM interfaces: %v", err) + } + + for _, iface := range interfaces { + if iface.Provider != fabrics.ProviderSHM { + continue + } + if iface.Caps.Flags&fabrics.InterfaceCapRemoteWrite == 0 { + continue + } + + // Allow libfabric to assign a unique local SHM endpoint during + // Target.Setup or Initiator.Setup. The enumerated service is + // descriptive and should not be reused by both endpoints. + iface.Address.Service = "" + return iface + } + + t.Fatal("no remote-write-capable SHM Fabrics interface found") + return fabrics.InterfaceConfig{} +} + +func TestInterfacesIncludesSHM(t *testing.T) { + if _, err := os.Stat("/dev/shm"); err != nil { + t.Skipf("/dev/shm is unavailable: %v", err) + } + + domain, err := os.MkdirTemp("/dev/shm", "mxl-player-fabrics-*") + if err != nil { + t.Fatalf("create temporary MXL domain: %v", err) + } + t.Cleanup(func() { + if err := os.RemoveAll(domain); err != nil { + t.Errorf("remove temporary MXL domain: %v", err) + } + }) + + instance, err := mxl.NewInstance(domain, "") + if err != nil { + t.Fatalf("create MXL instance: %v", err) + } + t.Cleanup(func() { + if err := instance.Close(); err != nil { + t.Errorf("close MXL instance: %v", err) + } + }) + + fabricInstance, err := fabrics.NewInstance(instance) + if err != nil { + t.Fatalf("create Fabrics instance: %v", err) + } + t.Cleanup(func() { + if err := fabricInstance.Close(); err != nil { + t.Errorf("close Fabrics instance: %v", err) + } + }) + + iface := requireSHMInterface(t, fabricInstance) + + t.Logf( + "SHM interface: node=%q flags=%#x max-message-size=%d attr=%q", + iface.Address.Node, + iface.Caps.Flags, + iface.Caps.MaxMessageSize, + iface.Attr, + ) +} diff --git a/internal/adapter/mxlfabrics/sample_transfer_integration_test.go b/internal/adapter/mxlfabrics/sample_transfer_integration_test.go new file mode 100644 index 0000000..ab35a4a --- /dev/null +++ b/internal/adapter/mxlfabrics/sample_transfer_integration_test.go @@ -0,0 +1,293 @@ +//go:build mxl_integration + +package mxlfabrics_test + +import ( + "errors" + "sync" + "testing" + "time" + + "github.com/qvest-digital/go-mxl/fabrics" + "github.com/qvest-digital/go-mxl/mxl" +) + +const testAudioFlowID = "b3bb5be7-9fe9-4324-a5bb-4c70e1084449" + +const testAudioFlow = `{ + "description": "MXL Player Fabrics SHM audio integration test", + "id": "b3bb5be7-9fe9-4324-a5bb-4c70e1084449", + "format": "urn:x-nmos:format:audio", + "label": "Fabrics SHM test audio", + "tags": { + "urn:x-nmos:tag:grouphint/v1.0": [ + "mxl-player-fabrics-test:Audio" + ] + }, + "parents": [], + "media_type": "audio/float32", + "sample_rate": { + "numerator": 48000 + }, + "channel_count": 2, + "bit_depth": 32 +}` + +type sampleCompletion struct { + head uint64 + count int +} + +func TestSampleTransferSHM(t *testing.T) { + sourceInstance := newTestDomain(t) + targetInstance := newTestDomain(t) + + sourceWriter, _, err := sourceInstance.NewWriter(testAudioFlow) + if err != nil { + t.Fatalf("create source writer: %v", err) + } + t.Cleanup(func() { + if err := sourceWriter.Close(); err != nil { + t.Errorf("close source writer: %v", err) + } + }) + + sourceReader, err := sourceInstance.NewReader(testAudioFlowID) + if err != nil { + t.Fatalf("create source reader: %v", err) + } + t.Cleanup(func() { + if err := sourceReader.Close(); err != nil { + t.Errorf("close source reader: %v", err) + } + }) + + targetWriter, _, err := targetInstance.NewWriter(testAudioFlow) + if err != nil { + t.Fatalf("create target writer: %v", err) + } + t.Cleanup(func() { + if err := targetWriter.Close(); err != nil { + t.Errorf("close target writer: %v", err) + } + }) + + sourceFabrics, err := fabrics.NewInstance(sourceInstance) + if err != nil { + t.Fatalf("create source Fabrics instance: %v", err) + } + t.Cleanup(func() { + if err := sourceFabrics.Close(); err != nil { + t.Errorf("close source Fabrics instance: %v", err) + } + }) + + targetFabrics, err := fabrics.NewInstance(targetInstance) + if err != nil { + t.Fatalf("create target Fabrics instance: %v", err) + } + t.Cleanup(func() { + if err := targetFabrics.Close(); err != nil { + t.Errorf("close target Fabrics instance: %v", err) + } + }) + + target, err := targetFabrics.NewTarget() + if err != nil { + t.Fatalf("create target: %v", err) + } + t.Cleanup(func() { + if err := target.Close(); err != nil { + t.Errorf("close target: %v", err) + } + }) + + targetInfo, err := target.Setup(fabrics.TargetConfig{ + Interface: requireSHMInterface(t, targetFabrics), + Writer: targetWriter, + }) + if err != nil { + t.Fatalf("set up target: %v", err) + } + t.Cleanup(func() { + if err := targetInfo.Close(); err != nil { + t.Errorf("close target info: %v", err) + } + }) + + initiator, err := sourceFabrics.NewInitiator() + if err != nil { + t.Fatalf("create initiator: %v", err) + } + t.Cleanup(func() { + if err := initiator.Close(); err != nil { + t.Errorf("close initiator: %v", err) + } + }) + + if err := initiator.Setup(fabrics.InitiatorConfig{ + Interface: requireSHMInterface(t, sourceFabrics), + Reader: sourceReader, + }); err != nil { + t.Fatalf("set up initiator: %v", err) + } + + if err := initiator.AddTarget(targetInfo); err != nil { + t.Fatalf("add target: %v", err) + } + + connectDeadline := time.Now().Add(5 * time.Second) + for { + if time.Now().After(connectDeadline) { + t.Fatal("SHM initiator did not become ready") + } + + _, _, err := target.ReadSamplesNonBlocking() + if err != nil && !errors.Is(err, fabrics.ErrNotReady) { + t.Fatalf("progress target during setup: %v", err) + } + + err = initiator.MakeProgressNonBlocking() + if err == nil { + break + } + if !errors.Is(err, fabrics.ErrNotReady) { + t.Fatalf("progress initiator during setup: %v", err) + } + + time.Sleep(time.Millisecond) + } + + const sampleCount = 480 + head := mxl.CurrentIndex(sourceWriter.Config().Common.GrainRate) + + sourceSamples, err := sourceWriter.OpenSamples(head, sampleCount) + if err != nil { + t.Fatalf("open source samples: %v", err) + } + for channel := uint64(0); channel < sourceSamples.ChannelCount; channel++ { + first, second, err := sourceSamples.ChannelFragments(channel) + if err != nil { + t.Fatalf("get source channel %d fragments: %v", channel, err) + } + var position uint64 + for _, fragment := range [][]byte{first, second} { + for offset := range fragment { + fragment[offset] = byte((position + head + channel) & 0xff) + position++ + } + } + } + if err := sourceSamples.Commit(); err != nil { + t.Fatalf("commit source samples: %v", err) + } + + deadline := time.Now().Add(5 * time.Second) + received := make(chan sampleCompletion, 1) + targetErrors := make(chan error, 1) + stopTarget := make(chan struct{}) + var targetWG sync.WaitGroup + + targetWG.Add(1) + go func() { + defer targetWG.Done() + for { + select { + case <-stopTarget: + return + default: + } + + receivedHead, receivedCount, err := target.ReadSamplesNonBlocking() + switch { + case err == nil: + received <- sampleCompletion{head: receivedHead, count: receivedCount} + return + case errors.Is(err, fabrics.ErrNotReady): + time.Sleep(time.Millisecond) + default: + targetErrors <- err + return + } + } + }() + defer func() { + close(stopTarget) + targetWG.Wait() + }() + + for { + err := initiator.TransferSamples(head, sampleCount) + if err == nil { + break + } + if !errors.Is(err, fabrics.ErrNotReady) { + t.Fatalf("transfer samples: %v", err) + } + if time.Now().After(deadline) { + t.Fatal("timed out enqueueing SHM sample transfer") + } + + err = initiator.MakeProgressNonBlocking() + if err != nil && !errors.Is(err, fabrics.ErrNotReady) { + t.Fatalf("make initiator progress while enqueueing: %v", err) + } + time.Sleep(time.Millisecond) + } + + var completion sampleCompletion + for { + if time.Now().After(deadline) { + t.Fatal("SHM sample transfer timed out") + } + + err := initiator.MakeProgressNonBlocking() + if err != nil && !errors.Is(err, fabrics.ErrNotReady) { + t.Fatalf("make initiator progress: %v", err) + } + + select { + case completion = <-received: + goto receivedSamples + case err := <-targetErrors: + t.Fatalf("read target completion: %v", err) + default: + time.Sleep(time.Millisecond) + } + } + +receivedSamples: + if completion.head != head { + t.Fatalf("received sample head = %d, want %d", completion.head, head) + } + if completion.count != sampleCount { + t.Fatalf("received sample count = %d, want %d", completion.count, sampleCount) + } + + targetSamples, err := targetWriter.OpenSamples(head, sampleCount) + if err != nil { + t.Fatalf("open transferred samples: %v", err) + } + defer func() { + if err := targetSamples.Cancel(); err != nil { + t.Errorf("cancel transferred samples: %v", err) + } + }() + + for channel := uint64(0); channel < targetSamples.ChannelCount; channel++ { + first, second, err := targetSamples.ChannelFragments(channel) + if err != nil { + t.Fatalf("get target channel %d fragments: %v", channel, err) + } + var position uint64 + for _, fragment := range [][]byte{first, second} { + for _, got := range fragment { + want := byte((position + head + channel) & 0xff) + if got != want { + t.Fatalf("channel %d sample byte %d = %d, want %d", channel, position, got, want) + } + position++ + } + } + } +}