testing with examples

Author: tjungblu

go.mod

@@ -1,6 +1,7 @@
 module github.com/thomasjungblut/go-sstables
 
 require (
+	github.com/godzie44/go-uring v0.0.0-20220131163417-b7645a723f7b
 	github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db
 	github.com/libp2p/go-buffer-pool v0.0.2
 	github.com/ncw/directio v1.0.5
@@ -12,8 +13,10 @@ require (
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/libp2p/go-sockaddr v0.1.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/steakknife/hamming v0.0.0-20180906055917-c99c65617cd3 // indirect
+	golang.org/x/sys v0.0.0-20210921065528-437939a70204 // indirect
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c // indirect
 )

go.sum

@@ -1,13 +1,17 @@
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/godzie44/go-uring v0.0.0-20220131163417-b7645a723f7b h1:H31NTxP3FAzF0mVbqRCJA04XKxoAQD85Ug+zZ761duo=
+github.com/godzie44/go-uring v0.0.0-20220131163417-b7645a723f7b/go.mod h1:ermjEDUoT/fS+3Ona5Vd6t6mZkw1eHp99ILO5jGRBkM=
 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db h1:woRePGFeVFfLKN/pOkfl+p/TAqKOfFu+7KPlMVpok/w=
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/libp2p/go-buffer-pool v0.0.2 h1:QNK2iAFa8gjAe1SPz6mHSMuCcjs+X1wlHzeOSqcmlfs=
 github.com/libp2p/go-buffer-pool v0.0.2/go.mod h1:MvaB6xw5vOrDl8rYZGLFdKAuk/hRoRZd1Vi32+RXyFM=
+github.com/libp2p/go-sockaddr v0.1.1 h1:yD80l2ZOdGksnOyHrhxDdTDFrf7Oy+v3FMVArIRgZxQ=
+github.com/libp2p/go-sockaddr v0.1.1/go.mod h1:syPvOmNs24S3dFVGJA1/mrqdeijPxLV2Le3BRLKd68k=
 github.com/ncw/directio v1.0.5 h1:JSUBhdjEvVaJvOoyPAbcW0fnd0tvRXD76wEfZ1KcQz4=
 github.com/ncw/directio v1.0.5/go.mod h1:rX/pKEYkOXBGOggmcyJeJGloCkleSvphPx2eV3t6ROk=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -21,6 +25,9 @@ github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5Cc
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 golang.org/x/exp v0.0.0-20181210123644-7d6377eee41f h1:wJ3O7VtAmBlW5LFzggOI2U6CBWIkG+/IYf4Q/VGJdaA=
 golang.org/x/exp v0.0.0-20181210123644-7d6377eee41f/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
+golang.org/x/sys v0.0.0-20190228124157-a34e9553db1e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20210921065528-437939a70204 h1:JJhkWtBuTQKyz2bd5WG9H8iUsJRU3En/KRfN8B2RnDs=
+golang.org/x/sys v0.0.0-20210921065528-437939a70204/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

recordio/README.md

@@ -210,3 +210,14 @@ available, err := recordio.IsDirectIOAvailable()
 
 In this package the DirectIO support comes through a library called [ncw/directio](https://github.com/ncw/directio), which has good support across Linux, macOS and Windows under a single interface. The caveats of each platform, for example the buffer/block sizes, need to still be taken into account.  
 Another caveat is that the block alignment causes to write a certain amount of waste. Let's imagine you have blocks of 1024 bytes and only want to write 1025 bytes, with DirectIO enabled you will end up with a file of size 2048 (2 blocks) instead of a file with only 1025 bytes with DirectIO disabled. The DirectIO file will be padded with zeroes towards the end and the in-library readers honor this format and not assume a corrupted file format. 
+
+
+## io_uring (experimental)
+
+Since version 5.x the linux kernel supports a new asynchronous approach to execute syscalls. In a few words, io_uring is a shared ring buffer between the kernel and user space which allows queueing syscalls and later retrieve their results.
+
+You can read more about io_uring at [https://kernel.dk/io_uring.pdf](https://kernel.dk/io_uring.pdf).
+
+
+
+

recordio/async_writer.go

@@ -0,0 +1,110 @@
+package recordio
+
+import (
+	"github.com/godzie44/go-uring/uring"
+	"os"
+	"sync/atomic"
+)
+
+// AsyncWriter takes an uring and executes all writes asynchronously. There are only two barriers: flush and close.
+// Those barriers will ensure all previous writes have succeeded.
+type AsyncWriter struct {
+	ringSize      int32
+	submittedSQEs int32
+	ring          *uring.Ring
+
+	file   *os.File
+	offset uint64
+}
+
+// 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)
+		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)
+	if err != nil {
+		return 0, err
+	}
+
+	atomic.AddInt32(&w.submittedSQEs, 1)
+	atomic.AddUint64(&w.offset, uint64(len(p)))
+
+	return len(p), nil
+}
+
+func (w *AsyncWriter) Flush() error {
+	for w.submittedSQEs >= 0 {
+		// wait for at least one event to free from the queue
+		cqe, err := w.ring.SubmitAndWaitCQEvents(1)
+		if err != nil {
+			return err
+		}
+
+		atomic.AddInt32(&w.submittedSQEs, -1)
+		w.ring.SeenCQE(cqe)
+	}
+
+	return nil
+}
+
+func (w *AsyncWriter) Size() int {
+	return 0
+}
+
+func (w *AsyncWriter) Close() error {
+	err := w.Flush()
+	if err != nil {
+		return err
+	}
+
+	err = w.ring.UnRegisterFiles()
+	if err != nil {
+		return err
+	}
+
+	err = w.ring.Close()
+	if err != nil {
+		return err
+	}
+
+	return w.file.Close()
+}
+
+func NewAsyncWriter(filePath string, numRingEntries uint32, opts ...uring.SetupOption) (WriteCloserFlusher, error) {
+	ring, err := uring.New(numRingEntries, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	writeFile, err := os.OpenFile(filePath, os.O_WRONLY|os.O_CREATE, 0666)
+	if err != nil {
+		return nil, err
+	}
+
+	err = ring.RegisterFiles([]int{int(writeFile.Fd())})
+	if err != nil {
+		return nil, err
+	}
+
+	writer := &AsyncWriter{
+		ringSize: int32(numRingEntries),
+		file:     writeFile,
+		ring:     ring,
+	}
+
+	return writer, nil
+}

recordio/async_writer_test.go

@@ -0,0 +1,59 @@
+package recordio
+
+import (
+	"fmt"
+	"github.com/godzie44/go-uring/uring"
+	"github.com/stretchr/testify/require"
+	"io/ioutil"
+	"os"
+	"testing"
+)
+
+func TestAsyncWriter_HappyPath(t *testing.T) {
+	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)
+	require.NoError(t, err)
+
+	for i := 0; i < 10000; i++ {
+		_, err = writer.Write(randomRecordOfSize(1024))
+		require.NoError(t, err)
+	}
+
+	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)
+
+}