Optimize error handling & improve non-blocking performance (#37)

* Optimize error handling & provide fast path for non-blocking send

runtime.LockOSThread() has a cost associated with it, as well as each
cgo call.
Also provide a fastpath for non-blocking mode: try to send first and
only start the epoller when send fails, this reduces a cgo call in the
happy case (most likely the default case).

Author: Gijs Peskens

read.go

@@ -0,0 +1,67 @@
+package srtgo
+
+/*
+#cgo LDFLAGS: -lsrt
+#include <srt/srt.h>
+
+int srt_recvmsg2_wrapped(SRTSOCKET u, char* buf, int len, SRT_MSGCTRL *mctrl, int *srterror, int *syserror)
+{
+	int ret = srt_recvmsg2(u, buf, len, mctrl);
+	if (ret < 0) {
+		*srterror = srt_getlasterror(syserror);
+	}
+	return ret;
+}
+
+*/
+import "C"
+import (
+	"errors"
+	"fmt"
+	"runtime"
+	"syscall"
+	"unsafe"
+)
+
+func srtRecvMsg2Impl(u C.SRTSOCKET, buf []byte, msgctrl *C.SRT_MSGCTRL) (n int, err error) {
+	srterr := C.int(0)
+	syserr := C.int(0)
+	n = int(C.srt_recvmsg2_wrapped(u, (*C.char)(unsafe.Pointer(&buf[0])), C.int(len(buf)), msgctrl, &srterr, &syserr))
+	if n < 0 {
+		srterror := SRTErrno(srterr)
+		if syserr < 0 {
+			srterror.wrapSysErr(syscall.Errno(syserr))
+		}
+		err = srterror
+	}
+	return
+}
+
+// Read data from the SRT socket
+func (s SrtSocket) Read(b []byte) (n int, err error) {
+	//Fastpath
+	n, err = srtRecvMsg2Impl(s.socket, b, nil)
+
+	if err != nil {
+		if errors.Is(err, error(EAsyncRCV)) {
+			runtime.LockOSThread()
+			defer runtime.UnlockOSThread()
+			timeoutMs := C.int64_t(s.pollTimeout)
+			fds := [1]C.SRT_EPOLL_EVENT{}
+			len := C.int(1)
+			res := C.srt_epoll_uwait(s.epollIn, &fds[0], len, timeoutMs)
+			if res == 0 {
+				return 0, &SrtEpollTimeout{}
+			}
+			if res == SRT_ERROR {
+				return 0, fmt.Errorf("error in read:epoll %w", srtGetAndClearError())
+			}
+			if fds[0].events&C.SRT_EPOLL_ERR > 0 {
+				return 0, srtGetAndClearError()
+			}
+			//Read again, now that we are ready
+			n, err = srtRecvMsg2Impl(s.socket, b, nil)
+		}
+	}
+	return
+}

srtgo.go

@@ -292,62 +292,6 @@ func (s SrtSocket) Connect() error {
 	return nil
 }
 
-// Read data from the SRT socket
-func (s SrtSocket) Read(b []byte) (n int, err error) {
-	runtime.LockOSThread()
-	defer runtime.UnlockOSThread()
-	if !s.blocking {
-		timeoutMs := C.int64_t(s.pollTimeout)
-		fds := [1]C.SRT_EPOLL_EVENT{}
-		len := C.int(1)
-		res := C.srt_epoll_uwait(s.epollIn, &fds[0], len, timeoutMs)
-		if res == 0 {
-			return 0, &SrtEpollTimeout{}
-		}
-		if res == SRT_ERROR {
-			return 0, fmt.Errorf("error in read:epoll %w", srtGetAndClearError())
-		}
-		if fds[0].events&C.SRT_EPOLL_ERR > 0 {
-			return 0, srtGetAndClearError()
-		}
-	}
-
-	res := C.srt_recvmsg2(s.socket, (*C.char)(unsafe.Pointer(&b[0])), C.int(len(b)), nil)
-	if res == SRT_ERROR {
-		return 0, fmt.Errorf("error in read:srt_recvmsg2 %w", srtGetAndClearError())
-	}
-
-	return int(res), nil
-}
-
-// Write data to the SRT socket
-func (s SrtSocket) Write(b []byte) (n int, err error) {
-	runtime.LockOSThread()
-	defer runtime.UnlockOSThread()
-	if !s.blocking {
-		timeoutMs := C.int64_t(s.pollTimeout)
-		fds := [1]C.SRT_EPOLL_EVENT{}
-		len := C.int(1)
-		res := C.srt_epoll_uwait(s.epollOut, &fds[0], len, timeoutMs)
-		if res == 0 {
-			return 0, &SrtEpollTimeout{}
-		}
-		if res == SRT_ERROR {
-			return 0, fmt.Errorf("error in write:epoll %w", srtGetAndClearError())
-		}
-		if fds[0].events&C.SRT_EPOLL_ERR > 0 {
-			return 0, &SrtSocketClosed{}
-		}
-	}
-
-	res := C.srt_sendmsg2(s.socket, (*C.char)(unsafe.Pointer(&b[0])), C.int(len(b)), nil)
-	if res == SRT_ERROR {
-		return 0, fmt.Errorf("error in write:srt_sendmsg2 %w", srtGetAndClearError())
-	}
-
-	return int(res), nil
-}
-
 // Stats - Retrieve stats from the SRT socket
 func (s SrtSocket) Stats() (*SrtStats, error) {
 	runtime.LockOSThread()

write.go

@@ -0,0 +1,67 @@
+package srtgo
+
+/*
+#cgo LDFLAGS: -lsrt
+#include <srt/srt.h>
+
+int srt_sendmsg2_wrapped(SRTSOCKET u, const char* buf, int len, SRT_MSGCTRL *mctrl, int *srterror, int *syserror)
+{
+	int ret = srt_sendmsg2(u, buf, len, mctrl);
+	if (ret < 0) {
+		*srterror = srt_getlasterror(syserror);
+	}
+	return ret;
+}
+
+*/
+import "C"
+import (
+	"errors"
+	"fmt"
+	"runtime"
+	"syscall"
+	"unsafe"
+)
+
+func srtSendMsg2Impl(u C.SRTSOCKET, buf []byte, msgctrl *C.SRT_MSGCTRL) (n int, err error) {
+	srterr := C.int(0)
+	syserr := C.int(0)
+	n = int(C.srt_sendmsg2_wrapped(u, (*C.char)(unsafe.Pointer(&buf[0])), C.int(len(buf)), msgctrl, &srterr, &syserr))
+	if n < 0 {
+		srterror := SRTErrno(srterr)
+		if syserr < 0 {
+			srterror.wrapSysErr(syscall.Errno(syserr))
+		}
+		err = srterror
+	}
+	return
+}
+
+// Write data to the SRT socket
+func (s SrtSocket) Write(b []byte) (n int, err error) {
+
+	//Fastpath:
+	n, err = srtSendMsg2Impl(s.socket, b, nil)
+
+	if err != nil {
+		if errors.Is(err, error(EAsyncSND)) && !s.blocking {
+			runtime.LockOSThread()
+			defer runtime.UnlockOSThread()
+			timeoutMs := C.int64_t(s.pollTimeout)
+			fds := [1]C.SRT_EPOLL_EVENT{}
+			len := C.int(1)
+			res := C.srt_epoll_uwait(s.epollOut, &fds[0], len, timeoutMs)
+			if res == 0 {
+				return 0, &SrtEpollTimeout{}
+			}
+			if res == SRT_ERROR {
+				return 0, fmt.Errorf("error in write:epoll %w", srtGetAndClearError())
+			}
+			if fds[0].events&C.SRT_EPOLL_ERR > 0 {
+				return 0, &SrtSocketClosed{}
+			}
+			n, err = srtSendMsg2Impl(s.socket, b, nil)
+		}
+	}
+	return
+}