basic support for writing

Author: tjungblu

recordio/README.md

@@ -218,6 +218,4 @@ Since version 5.x the linux kernel supports a new asynchronous approach to execu
 
 You can read more about io_uring at [https://kernel.dk/io_uring.pdf](https://kernel.dk/io_uring.pdf).
 
-
-
-
+In case you have SELinux enabled, you might hit "permission denied" errors when initializing the uring try to test with permissive mode enabled temporarily. 

recordio/async_writer.go

@@ -1,3 +1,5 @@
+//go:build linux
+
 package recordio
 
 import (
@@ -17,22 +19,13 @@ type AsyncWriter struct {
 	offset uint64
 }
 
-// TODO(thomas): not thread-safe yet
+// TODO(thomas): not thread-safe (yet)
 func (w *AsyncWriter) Write(p []byte) (int, error) {
 	for w.submittedSQEs >= w.ringSize {
-		// wait for at least one event to free from the queue
-		cqe, err := w.ring.SubmitAndWaitCQEvents(1)
+		err := w.submitAwaitOne()
 		if err != nil {
 			return 0, err
 		}
-
-		err = cqe.Error()
-		if err != nil {
-			return 0, err
-		}
-
-		atomic.AddInt32(&w.submittedSQEs, -1)
-		w.ring.SeenCQE(cqe)
 	}
 
 	err := w.ring.QueueSQE(uring.Write(w.file.Fd(), p, w.offset), 0, 0)
@@ -47,15 +40,30 @@ func (w *AsyncWriter) Write(p []byte) (int, error) {
 }
 
 func (w *AsyncWriter) Flush() error {
-	for w.submittedSQEs >= 0 {
+	for w.submittedSQEs > 0 {
 		// wait for at least one event to free from the queue
-		cqe, err := w.ring.SubmitAndWaitCQEvents(1)
+		err := w.submitAwaitOne()
 		if err != nil {
 			return err
 		}
+	}
+
+	return nil
+}
 
-		atomic.AddInt32(&w.submittedSQEs, -1)
-		w.ring.SeenCQE(cqe)
+func (w *AsyncWriter) submitAwaitOne() error {
+	// TODO(thomas): most likely there are more CQ events waiting, we should try to drain them optimistically to avoid overflowing memory buffers
+	cqe, err := w.ring.SubmitAndWaitCQEvents(1)
+	if err != nil {
+		return err
+	}
+
+	atomic.AddInt32(&w.submittedSQEs, -1)
+	w.ring.SeenCQE(cqe)
+
+	err = cqe.Error()
+	if err != nil {
+		return err
 	}
 
 	return nil

recordio/async_writer_test.go

@@ -1,18 +1,24 @@
+//go:build linux
+
 package recordio
 
 import (
-	"fmt"
-	"github.com/godzie44/go-uring/uring"
 	"github.com/stretchr/testify/require"
+	"github.com/thomasjungblut/go-sstables/recordio/iouring"
 	"io/ioutil"
-	"os"
 	"testing"
 )
 
 func TestAsyncWriter_HappyPath(t *testing.T) {
+	ok, err := iouring.IsIOUringAvailable()
+	require.NoError(t, err)
+	if !ok {
+		t.Skip("iouring not available here")
+		return
+	}
+
 	temp, err := ioutil.TempFile("", "TestAsyncWriter_HappyPath")
 	require.NoError(t, err)
-	require.NoError(t, temp.Close())
 	defer closeCleanFile(t, temp)
 
 	writer, err := NewAsyncWriter(temp.Name(), 4)
@@ -25,35 +31,3 @@ func TestAsyncWriter_HappyPath(t *testing.T) {
 
 	require.NoError(t, writer.Close())
 }
-
-func TestExample(t *testing.T) {
-	ring, err := uring.New(8)
-	require.NoError(t, err)
-	defer ring.Close()
-
-	// open file and init read buffers
-	file, err := os.Open("./go.mod")
-	require.NoError(t, err)
-	stat, _ := file.Stat()
-	buff := make([]byte, stat.Size())
-
-	// add Read operation to SQ queue
-	err = ring.QueueSQE(uring.Read(file.Fd(), buff, 0), 0, 0)
-	require.NoError(t, err)
-
-	// submit all SQ new entries
-	_, err = ring.Submit()
-	require.NoError(t, err)
-
-	// wait until data is reading into buffer
-	cqe, err := ring.WaitCQEvents(1)
-	require.NoError(t, err)
-
-	require.NoError(t, cqe.Error()) //check read error
-
-	fmt.Printf("read %d bytes, read result: \n%s", cqe.Res, string(buff))
-
-	// dequeue CQ
-	ring.SeenCQE(cqe)
-
-}

recordio/direct_io.go

@@ -1,9 +1,11 @@
 package recordio
 
 import (
+	"errors"
 	"github.com/ncw/directio"
 	"io/ioutil"
 	"os"
+	"syscall"
 )
 
 type DirectIOFactory struct {
@@ -51,7 +53,11 @@ func IsDirectIOAvailable() (available bool, err error) {
 	tmpFile, err = directio.OpenFile(tmpFile.Name(), os.O_WRONLY|os.O_CREATE, 0666)
 	if err != nil {
 		// this syscall specifically signals that DirectIO is not supported
-		// if errors.Is(err, syscall.EINVAL)
+		if errors.Is(err, syscall.EINVAL) {
+			available = false
+			err = nil
+			return
+		}
 		return
 	}
 

recordio/file_writer_test.go

@@ -178,6 +178,13 @@ func TestWriterCrashCreatesValidHeader(t *testing.T) {
 }
 
 func TestWriterCrashCreatesNoValidHeaderWithDirectIO(t *testing.T) {
+	ok, err := IsDirectIOAvailable()
+	require.NoError(t, err)
+	if !ok {
+		t.Skip("directio not available here")
+		return
+	}
+
 	tmpFile, err := ioutil.TempFile("", "recordio_CrashCreatesValidHeaderDirectIO")
 	require.Nil(t, err)
 	defer closeCleanFile(t, tmpFile)
@@ -195,6 +202,13 @@ func TestWriterCrashCreatesNoValidHeaderWithDirectIO(t *testing.T) {
 }
 
 func TestWriterNotAllowsSyncsWithDirectIO(t *testing.T) {
+	ok, err := IsDirectIOAvailable()
+	require.NoError(t, err)
+	if !ok {
+		t.Skip("directio not available here")
+		return
+	}
+
 	tmpFile, err := ioutil.TempFile("", "recordio_WriterNotAllowsSyncsWithDirectIO")
 	require.Nil(t, err)
 	defer closeCleanFile(t, tmpFile)

recordio/iouring/iouring.go

@@ -0,0 +1,17 @@
+//go:build linux
+
+package iouring
+
+import "github.com/godzie44/go-uring/uring"
+
+// IsIOUringAvailable tests whether io_uring is supported by the kernel.
+// It will return (true, nil) if that's the case, if it's not available it will be (false, nil).
+// Any other error will be indicated by the error (either true/false).
+func IsIOUringAvailable() (available bool, err error) {
+	ring, err := uring.New(1)
+	defer func() {
+		err = ring.Close()
+	}()
+
+	return err == nil, err
+}

recordio/iouring/iouring_test.go

@@ -0,0 +1,15 @@
+package iouring
+
+import (
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func TestIsIOUringAvailable_HappyPath(t *testing.T) {
+	ok, err := IsIOUringAvailable()
+	require.NoError(t, err)
+	if !ok {
+		t.Skip("iouring not available here")
+		return
+	}
+}