MXL Fabrics tests

internal/adapter/mxlfabrics/doc.go

@@ -0,0 +1,2 @@
+// Package mxlfabrics provides MXL Fabrics ingress adapters
+package mxlfabrics

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,
+			)
+		}
+	}
+}

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,
+	)
+}

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++
+			}
+		}
+	}
+}