j6: account bytes for ret-classified syscalls

9 files changed, 411 insertions, 3 deletions

diff --git a/cmd/ioworkload/scenario_retbytes.go b/cmd/ioworkload/scenario_retbytes.go
new file mode 100644
index 0000000..fa0b677
--- /dev/null
+++ b/cmd/ioworkload/scenario_retbytes.go
@@ -0,0 +1,243 @@
+package main
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"runtime"
+	"syscall"
+	"unsafe"
+)
+
+const (
+	sysProcessVMReadv  = 310
+	sysProcessVMWritev = 311
+	retbytesPayloadLen = 18
+)
+
+// retbytesPhaseA exercises byte-classified syscalls that use generic ret_event exits.
+func retbytesPhaseA() error {
+	if err := retbytesSocketIO(); err != nil {
+		return err
+	}
+	if err := retbytesSendfile(); err != nil {
+		return err
+	}
+	if err := retbytesSplice(); err != nil {
+		return err
+	}
+	if err := retbytesTee(); err != nil {
+		return err
+	}
+	return retbytesProcessVM()
+}
+
+func retbytesSocketIO() error {
+	fds, err := syscall.Socketpair(syscall.AF_UNIX, syscall.SOCK_STREAM, 0)
+	if err != nil {
+		return fmt.Errorf("socketpair: %w", err)
+	}
+	defer syscall.Close(fds[0])
+	defer syscall.Close(fds[1])
+
+	payload := []byte("phase-a-send-recv!")
+	if err := syscall.Sendto(fds[0], payload, 0, nil); err != nil {
+		return fmt.Errorf("sendto: %w", err)
+	}
+	buf := make([]byte, len(payload))
+	n, _, err := syscall.Recvfrom(fds[1], buf, 0)
+	if err != nil {
+		return fmt.Errorf("recvfrom: %w", err)
+	}
+	if n != len(payload) {
+		return fmt.Errorf("recvfrom read %d bytes, want %d", n, len(payload))
+	}
+
+	if n, err := syscall.SendmsgN(fds[0], payload, nil, nil, 0); err != nil {
+		return fmt.Errorf("sendmsg: %w", err)
+	} else if n != len(payload) {
+		return fmt.Errorf("sendmsg wrote %d bytes, want %d", n, len(payload))
+	}
+	n, _, _, _, err = syscall.Recvmsg(fds[1], buf, nil, 0)
+	if err != nil {
+		return fmt.Errorf("recvmsg: %w", err)
+	}
+	if n != len(payload) {
+		return fmt.Errorf("recvmsg read %d bytes, want %d", n, len(payload))
+	}
+	return nil
+}
+
+func retbytesSendfile() error {
+	dir, cleanup, err := makeTempDir("retbytes-sendfile")
+	if err != nil {
+		return err
+	}
+	defer cleanup()
+
+	src, dst, err := openTransferFiles(dir, "sendfilesrc.txt", "sendfiledst.txt")
+	if err != nil {
+		return err
+	}
+	defer syscall.Close(src)
+	defer syscall.Close(dst)
+
+	n, err := syscall.Sendfile(dst, src, nil, retbytesPayloadLen)
+	if err != nil {
+		return fmt.Errorf("sendfile: %w", err)
+	}
+	if n != retbytesPayloadLen {
+		return fmt.Errorf("sendfile copied %d bytes, want %d", n, retbytesPayloadLen)
+	}
+	return nil
+}
+
+func retbytesSplice() error {
+	dir, cleanup, err := makeTempDir("retbytes-splice")
+	if err != nil {
+		return err
+	}
+	defer cleanup()
+
+	src, err := openPayloadFile(filepath.Join(dir, "splicesrc.txt"))
+	if err != nil {
+		return err
+	}
+	defer syscall.Close(src)
+
+	pipe := make([]int, 2)
+	if err := syscall.Pipe2(pipe, syscall.O_CLOEXEC); err != nil {
+		return fmt.Errorf("pipe2: %w", err)
+	}
+	defer syscall.Close(pipe[0])
+	defer syscall.Close(pipe[1])
+
+	n, err := syscall.Splice(src, nil, pipe[1], nil, retbytesPayloadLen, 0)
+	if err != nil {
+		return fmt.Errorf("splice: %w", err)
+	}
+	if n != retbytesPayloadLen {
+		return fmt.Errorf("splice copied %d bytes, want %d", n, retbytesPayloadLen)
+	}
+	return nil
+}
+
+func retbytesTee() error {
+	pipeA := make([]int, 2)
+	if err := syscall.Pipe2(pipeA, syscall.O_CLOEXEC); err != nil {
+		return fmt.Errorf("pipe2 source: %w", err)
+	}
+	defer syscall.Close(pipeA[0])
+	defer syscall.Close(pipeA[1])
+
+	pipeB := make([]int, 2)
+	if err := syscall.Pipe2(pipeB, syscall.O_CLOEXEC); err != nil {
+		return fmt.Errorf("pipe2 dest: %w", err)
+	}
+	defer syscall.Close(pipeB[0])
+	defer syscall.Close(pipeB[1])
+
+	payload := []byte("phase-a-tee-bytes!")
+	if _, err := syscall.Write(pipeA[1], payload); err != nil {
+		return fmt.Errorf("write pipe: %w", err)
+	}
+	n, err := syscall.Tee(pipeA[0], pipeB[1], len(payload), 0)
+	if err != nil {
+		return fmt.Errorf("tee: %w", err)
+	}
+	if n != int64(len(payload)) {
+		return fmt.Errorf("tee copied %d bytes, want %d", n, len(payload))
+	}
+	return nil
+}
+
+func retbytesProcessVM() error {
+	src := []byte("phase-a-process-vm")
+	dst := make([]byte, len(src))
+	if n, err := processVMWritev(os.Getpid(), dst, src); err != nil {
+		return err
+	} else if n != len(src) {
+		return fmt.Errorf("process_vm_writev wrote %d bytes, want %d", n, len(src))
+	}
+
+	readBuf := make([]byte, len(dst))
+	if n, err := processVMReadv(os.Getpid(), readBuf, dst); err != nil {
+		return err
+	} else if n != len(dst) {
+		return fmt.Errorf("process_vm_readv read %d bytes, want %d", n, len(dst))
+	}
+	runtime.KeepAlive(src)
+	runtime.KeepAlive(dst)
+	runtime.KeepAlive(readBuf)
+	return nil
+}
+
+func openTransferFiles(dir, srcName, dstName string) (int, int, error) {
+	src, err := openPayloadFile(filepath.Join(dir, srcName))
+	if err != nil {
+		return 0, 0, err
+	}
+	dstPath := filepath.Join(dir, dstName)
+	dst, err := syscall.Open(dstPath, syscall.O_RDWR|syscall.O_CREAT|syscall.O_TRUNC, 0o644)
+	if err != nil {
+		syscall.Close(src)
+		return 0, 0, fmt.Errorf("open destination: %w", err)
+	}
+	return src, dst, nil
+}
+
+func openPayloadFile(path string) (int, error) {
+	fd, err := syscall.Open(path, syscall.O_RDWR|syscall.O_CREAT|syscall.O_TRUNC, 0o644)
+	if err != nil {
+		return 0, fmt.Errorf("open payload: %w", err)
+	}
+	if _, err := syscall.Write(fd, []byte("phase-a-ret-bytes!")); err != nil {
+		syscall.Close(fd)
+		return 0, fmt.Errorf("write payload: %w", err)
+	}
+	if _, err := syscall.Seek(fd, 0, 0); err != nil {
+		syscall.Close(fd)
+		return 0, fmt.Errorf("seek payload: %w", err)
+	}
+	return fd, nil
+}
+
+func processVMReadv(pid int, local, remote []byte) (int, error) {
+	localIov := syscall.Iovec{Base: &local[0], Len: uint64(len(local))}
+	remoteIov := syscall.Iovec{Base: &remote[0], Len: uint64(len(remote))}
+	n, _, errno := syscall.Syscall6(
+		sysProcessVMReadv,
+		uintptr(pid),
+		uintptr(unsafe.Pointer(&localIov)),
+		1,
+		uintptr(unsafe.Pointer(&remoteIov)),
+		1,
+		0,
+	)
+	runtime.KeepAlive(local)
+	runtime.KeepAlive(remote)
+	if errno != 0 {
+		return 0, fmt.Errorf("process_vm_readv: %w", errno)
+	}
+	return int(n), nil
+}
+
+func processVMWritev(pid int, remote, local []byte) (int, error) {
+	localIov := syscall.Iovec{Base: &local[0], Len: uint64(len(local))}
+	remoteIov := syscall.Iovec{Base: &remote[0], Len: uint64(len(remote))}
+	n, _, errno := syscall.Syscall6(
+		sysProcessVMWritev,
+		uintptr(pid),
+		uintptr(unsafe.Pointer(&localIov)),
+		1,
+		uintptr(unsafe.Pointer(&remoteIov)),
+		1,
+		0,
+	)
+	runtime.KeepAlive(local)
+	runtime.KeepAlive(remote)
+	if errno != 0 {
+		return 0, fmt.Errorf("process_vm_writev: %w", errno)
+	}
+	return int(n), nil
+}
diff --git a/cmd/ioworkload/scenarios.go b/cmd/ioworkload/scenarios.go
index 6910314..fc06825 100644
--- a/cmd/ioworkload/scenarios.go
+++ b/cmd/ioworkload/scenarios.go
@@ -24,6 +24,7 @@ var scenarios = map[string]func() error{
 	"readwrite-rdonly-write":      readwriteRdonlyWrite,
 	"readwrite-pread-invalid":     readwritePreadInvalid,
 	"readwrite-pwrite-invalid":    readwritePwriteInvalid,
+	"retbytes-phase-a":            retbytesPhaseA,
 	"close-basic":                 closeBasic,
 	"close-range":                 closeRange,
 	"close-invalid-fd":            closeInvalidFd,
diff --git a/integrationtests/readwrite_test.go b/integrationtests/readwrite_test.go
index 4e8cbef..69f60ea 100644
--- a/integrationtests/readwrite_test.go
+++ b/integrationtests/readwrite_test.go
@@ -228,3 +228,22 @@ func assertEventBytesReasonable(t *testing.T, result TestResult, exp ExpectedEve
 		t.Fatalf("expected event not found while asserting byte range: %+v", exp)
 	}
 }
+
+func assertEventDurationPositive(t *testing.T, result TestResult, exp ExpectedEvent) {
+	t.Helper()
+	var matched bool
+	var totalDuration uint64
+	for _, rec := range result.Records {
+		if !matchesExpectation(rec, exp) {
+			continue
+		}
+		matched = true
+		totalDuration += rec.Cnt.Duration
+	}
+	if !matched {
+		t.Fatalf("expected event not found while asserting duration: %+v", exp)
+	}
+	if totalDuration == 0 {
+		t.Fatalf("duration for %+v is zero", exp)
+	}
+}
diff --git a/integrationtests/retbytes_test.go b/integrationtests/retbytes_test.go
new file mode 100644
index 0000000..2e2ea1d
--- /dev/null
+++ b/integrationtests/retbytes_test.go
@@ -0,0 +1,35 @@
+package integrationtests
+
+import "testing"
+
+func TestRetbytesPhaseA(t *testing.T) {
+	const payloadLen = uint64(18)
+
+	result, _ := runScenarioResult(t, "retbytes-phase-a", []ExpectedEvent{
+		{Tracepoint: "enter_sendto", Comm: "ioworkload", MinCount: 1},
+		{Tracepoint: "enter_recvfrom", Comm: "ioworkload", MinCount: 1},
+		{Tracepoint: "enter_sendmsg", Comm: "ioworkload", MinCount: 1},
+		{Tracepoint: "enter_recvmsg", Comm: "ioworkload", MinCount: 1},
+		{Tracepoint: "enter_sendfile64", Comm: "ioworkload", MinCount: 1},
+		{Tracepoint: "enter_splice", Comm: "ioworkload", MinCount: 1},
+		{Tracepoint: "enter_tee", Comm: "ioworkload", MinCount: 1},
+		{Tracepoint: "enter_process_vm_writev", Comm: "ioworkload", MinCount: 1},
+		{Tracepoint: "enter_process_vm_readv", Comm: "ioworkload", MinCount: 1},
+	})
+
+	for _, tracepoint := range []string{
+		"enter_sendto",
+		"enter_recvfrom",
+		"enter_sendmsg",
+		"enter_recvmsg",
+		"enter_sendfile64",
+		"enter_splice",
+		"enter_tee",
+		"enter_process_vm_writev",
+		"enter_process_vm_readv",
+	} {
+		exp := ExpectedEvent{Tracepoint: tracepoint, Comm: "ioworkload"}
+		assertEventBytesAtLeast(t, result, exp, payloadLen)
+		assertEventDurationPositive(t, result, exp)
+	}
+}
diff --git a/internal/eventloop_bytes_test.go b/internal/eventloop_bytes_test.go
index ed7f7af..a7c25ef 100644
--- a/internal/eventloop_bytes_test.go
+++ b/internal/eventloop_bytes_test.go
@@ -3,6 +3,7 @@ package internal
 import (
 	"testing"
 
+	"ior/internal/event"
 	"ior/internal/types"
 )
 
@@ -71,3 +72,20 @@ func TestBytesFromRet(t *testing.T) {
 		})
 	}
 }
+
+func TestApplyRetBytesForNullEnterRetExitPair(t *testing.T) {
+	pair := &event.Pair{
+		EnterEv: &types.NullEvent{TraceId: types.SYS_ENTER_SPLICE},
+		ExitEv: &types.RetEvent{
+			TraceId: types.SYS_EXIT_SPLICE,
+			Ret:     4096,
+			RetType: types.TRANSFER_CLASSIFIED,
+		},
+	}
+
+	applyRetBytes(pair)
+
+	if pair.Bytes != 4096 {
+		t.Fatalf("pair.Bytes = %d, want 4096", pair.Bytes)
+	}
+}
diff --git a/internal/eventloop_exit.go b/internal/eventloop_exit.go
index 79c1b5b..a0cc675 100644
--- a/internal/eventloop_exit.go
+++ b/internal/eventloop_exit.go
@@ -109,9 +109,6 @@ func (e *eventLoop) handleFdExit(ep *event.Pair, fdEv *types.FdEvent) bool {
 	if ok := e.applyFdTransferOp(ep, fdEv); !ok {
 		return false
 	}
-	if retEv, ok := ep.ExitEv.(*types.RetEvent); ok {
-		ep.Bytes = bytesFromRet(retEv)
-	}
 	return true
 }
 
@@ -315,6 +312,14 @@ func (e *eventLoop) recyclePair(ep *event.Pair, warning string) {
 	ep.Recycle()
 }
 
+func applyRetBytes(ep *event.Pair) {
+	retEv, ok := ep.ExitEv.(*types.RetEvent)
+	if !ok {
+		return
+	}
+	ep.Bytes = bytesFromRet(retEv)
+}
+
 // dropMalformedRawEvent records a warning when a raw BPF event cannot be
 // decoded, keeping the error visible without crashing the event loop.
 func (e *eventLoop) dropMalformedRawEvent(evType types.EventType, raw []byte) {
diff --git a/internal/eventloop_runtime.go b/internal/eventloop_runtime.go
index 74571c8..c285507 100644
--- a/internal/eventloop_runtime.go
+++ b/internal/eventloop_runtime.go
@@ -314,6 +314,7 @@ func (e *eventLoop) tracepointExited(exitEv event.Event, ch chan<- *event.Pair)
 	if !e.handleTracepointExit(ep) {
 		return
 	}
+	applyRetBytes(ep)
 	tid := ep.EnterEv.GetTid()
 	ep.CalculateDurations(e.pairs.prevTime(tid))
 	e.pairs.setPrevTime(tid, ep.ExitEv.GetTime())
diff --git a/internal/generate/classify_test.go b/internal/generate/classify_test.go
index f02f7de..4dd216e 100644
--- a/internal/generate/classify_test.go
+++ b/internal/generate/classify_test.go
@@ -338,6 +338,67 @@ func TestClassifySyscallPairEmitsAllFamilies(t *testing.T) {
 	}
 }
 
+func TestClassifyPhaseAByteSyscallPairsAccepted(t *testing.T) {
+	tests := []struct {
+		name          string
+		enterKindText string
+		retText       string
+	}{
+		{"recvfrom", "struct fd_event", "READ_CLASSIFIED"},
+		{"recvmsg", "struct fd_event", "READ_CLASSIFIED"},
+		{"sendto", "struct fd_event", "WRITE_CLASSIFIED"},
+		{"sendmsg", "struct fd_event", "WRITE_CLASSIFIED"},
+		{"sendfile64", "struct null_event", "TRANSFER_CLASSIFIED"},
+		{"splice", "struct null_event", "TRANSFER_CLASSIFIED"},
+		{"tee", "struct null_event", "TRANSFER_CLASSIFIED"},
+		{"process_vm_readv", "struct null_event", "READ_CLASSIFIED"},
+		{"process_vm_writev", "struct null_event", "WRITE_CLASSIFIED"},
+	}
+
+	for i, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			formats := phaseAFormats(tt.name, 9000+i*2)
+			output := GenerateTracepointsC(formats)
+			if strings.Contains(output, "Ignoring") || strings.Contains(output, "Skipping") {
+				t.Fatalf("syscall %s was not accepted:\n%s", tt.name, output)
+			}
+			if !strings.Contains(output, "/// sys_enter_"+tt.name+" is a "+tt.enterKindText) {
+				t.Fatalf("sys_enter_%s did not use %s:\n%s", tt.name, tt.enterKindText, output)
+			}
+			if !strings.Contains(output, "/// sys_exit_"+tt.name+" is a struct ret_event ("+tt.retText+")") {
+				t.Fatalf("sys_exit_%s did not use %s:\n%s", tt.name, tt.retText, output)
+			}
+		})
+	}
+}
+
+func phaseAFormats(name string, enterID int) []Format {
+	enterFields := []Field{
+		{Type: "long", Name: "__syscall_nr"},
+	}
+	if name == "sendto" || name == "recvfrom" || name == "sendmsg" || name == "recvmsg" {
+		enterFields = append(enterFields, Field{Type: "int", Name: "fd"})
+	}
+
+	return []Format{
+		{
+			Name:           "sys_enter_" + name,
+			ID:             enterID,
+			Family:         ClassifySyscallFamily("sys_enter_" + name),
+			ExternalFields: enterFields,
+		},
+		{
+			Name:   "sys_exit_" + name,
+			ID:     enterID - 1,
+			Family: ClassifySyscallFamily("sys_exit_" + name),
+			ExternalFields: []Field{
+				{Type: "long", Name: "__syscall_nr"},
+				{Type: "long", Name: "ret"},
+			},
+		},
+	}
+}
+
 func TestClassifyFormatNoExternalFields(t *testing.T) {
 	f := &Format{
 		Name:           "sys_enter_test",
diff --git a/internal/generate/retclassify_test.go b/internal/generate/retclassify_test.go
index 3152005..9a75a15 100644
--- a/internal/generate/retclassify_test.go
+++ b/internal/generate/retclassify_test.go
@@ -57,3 +57,28 @@ func TestClassifyRetCaseInsensitive(t *testing.T) {
 		t.Errorf("ClassifyRet(sys_exit_READ) = %q, want READ_CLASSIFIED", got)
 	}
 }
+
+func TestPhaseAByteClassifiedSyscallsUseExistingRetClassifications(t *testing.T) {
+	tests := []struct {
+		name string
+		want RetClassification
+	}{
+		{"recvfrom", ReadClassified},
+		{"recvmsg", ReadClassified},
+		{"sendto", WriteClassified},
+		{"sendmsg", WriteClassified},
+		{"sendfile64", TransferClassified},
+		{"splice", TransferClassified},
+		{"tee", TransferClassified},
+		{"process_vm_readv", ReadClassified},
+		{"process_vm_writev", WriteClassified},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := ClassifyRet("sys_exit_" + tt.name); got != tt.want {
+				t.Fatalf("ClassifyRet(sys_exit_%s) = %q, want %q", tt.name, got, tt.want)
+			}
+		})
+	}
+}