Started on silo protocol/transport

pkg/storage/modules/from_protocol.go

@@ -0,0 +1,68 @@
+package modules
+
+import (
+	"github.com/loopholelabs/silo/pkg/storage"
+	"github.com/loopholelabs/silo/pkg/storage/protocol"
+)
+
+type FromProtocol struct {
+	dev      uint32
+	prov     storage.StorageProvider
+	protocol protocol.Protocol
+}
+
+func NewFromProtocol(dev uint32, prov storage.StorageProvider, protocol protocol.Protocol) *FromProtocol {
+	return &FromProtocol{
+		dev:      dev,
+		prov:     prov,
+		protocol: protocol,
+	}
+}
+
+// Handle any ReadAt commands, and send to provider
+func (fp *FromProtocol) HandleReadAt() error {
+	for {
+		id, data, err := fp.protocol.WaitForCommand(fp.dev, protocol.COMMAND_READ_AT)
+		if err != nil {
+			return err
+		}
+		offset, length, err := protocol.DecodeReadAt(data)
+		if err != nil {
+			return err
+		}
+		buff := make([]byte, length)
+		n, err := fp.prov.ReadAt(buff, offset)
+		rar := &protocol.ReadAtResponse{
+			Bytes: n,
+			Error: err,
+			Data:  buff,
+		}
+		_, err = fp.protocol.SendPacket(fp.dev, id, protocol.EncodeReadAtResponse(rar))
+		if err != nil {
+			return err
+		}
+	}
+}
+
+// Handle any WriteAt commands, and send to provider
+func (fp *FromProtocol) HandleWriteAt() error {
+	for {
+		id, data, err := fp.protocol.WaitForCommand(fp.dev, protocol.COMMAND_WRITE_AT)
+		if err != nil {
+			return err
+		}
+		offset, data, err := protocol.DecodeWriteAt(data)
+		if err != nil {
+			return err
+		}
+		n, err := fp.prov.WriteAt(data, offset)
+		war := &protocol.WriteAtResponse{
+			Bytes: n,
+			Error: err,
+		}
+		_, err = fp.protocol.SendPacket(fp.dev, id, protocol.EncodeWriteAtResponse(war))
+		if err != nil {
+			return err
+		}
+	}
+}

pkg/storage/modules/to_protocol.go

@@ -0,0 +1,71 @@
+package modules
+
+import (
+	"github.com/loopholelabs/silo/pkg/storage/protocol"
+)
+
+type ToProtocol struct {
+	size     uint64
+	dev      uint32
+	protocol protocol.Protocol
+}
+
+func NewToProtocol(size uint64, deviceID uint32, p protocol.Protocol) *ToProtocol {
+	return &ToProtocol{
+		size:     size,
+		dev:      deviceID,
+		protocol: p,
+	}
+}
+
+func (i *ToProtocol) ReadAt(buffer []byte, offset int64) (int, error) {
+	b := protocol.EncodeReadAt(offset, int32(len(buffer)))
+	id, err := i.protocol.SendPacket(i.dev, protocol.ID_PICK_ANY, b)
+	if err != nil {
+		return 0, err
+	}
+	// Wait for the response...
+	r, err := i.protocol.WaitForPacket(i.dev, id)
+	if err != nil {
+		return 0, err
+	}
+
+	// Decode the response and use it...
+	rp, err := protocol.DecodeReadAtResponse(r)
+	if err != nil {
+		return 0, err
+	}
+
+	copy(buffer, rp.Data)
+
+	return rp.Bytes, rp.Error
+}
+
+func (i *ToProtocol) WriteAt(buffer []byte, offset int64) (int, error) {
+	b := protocol.EncodeWriteAt(offset, buffer)
+	id, err := i.protocol.SendPacket(i.dev, protocol.ID_PICK_ANY, b)
+	if err != nil {
+		return 0, err
+	}
+	// Wait for the response...
+	r, err := i.protocol.WaitForPacket(i.dev, id)
+	if err != nil {
+		return 0, err
+	}
+
+	rp, err := protocol.DecodeWriteAtResponse(r)
+	if err != nil {
+		return 0, err
+	}
+
+	return rp.Bytes, rp.Error
+}
+
+func (i *ToProtocol) Flush() error {
+	// TODO...
+	return nil
+}
+
+func (i *ToProtocol) Size() uint64 {
+	return i.size
+}

pkg/storage/protocol/protocol.go

@@ -0,0 +1,22 @@
+package protocol
+
+const COMMAND_READ_AT = byte(1)
+const COMMAND_READ_AT_RESPONSE = byte(2)
+const COMMAND_READ_AT_RESPONSE_ERR = byte(3)
+
+const COMMAND_WRITE_AT = byte(0x10)
+const COMMAND_WRITE_AT_RESPONSE = byte(0x11)
+const COMMAND_WRITE_AT_RESPONSE_ERR = byte(0x12)
+
+const ID_PICK_ANY = 0
+
+type Protocol interface {
+	// Send a packet (Returns a transaction id)
+	SendPacket(dev uint32, id uint32, data []byte) (uint32, error)
+
+	// Wait for a response packet (Given specific transaction id)
+	WaitForPacket(dev uint32, id uint32) ([]byte, error)
+
+	// Wait for a specific command
+	WaitForCommand(dev uint32, cmd byte) (uint32, []byte, error)
+}

pkg/storage/protocol/protocol_test.go

@@ -0,0 +1,113 @@
+package protocol
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestReadAt(t *testing.T) {
+
+	b := EncodeReadAt(12345, 10)
+
+	off, length, err := DecodeReadAt(b)
+	assert.NoError(t, err)
+	assert.Equal(t, int64(12345), off)
+	assert.Equal(t, int32(10), length)
+
+	// Make sure we can't decode silly things
+	_, _, err = DecodeReadAt(nil)
+	assert.Error(t, err)
+
+	_, _, err = DecodeReadAt([]byte{
+		99,
+	})
+	assert.Error(t, err)
+
+}
+
+func TestReadAtResponse(t *testing.T) {
+	rar := &ReadAtResponse{
+		Bytes: 10,
+		Error: nil,
+		Data:  []byte{1, 2, 3, 4, 5},
+	}
+
+	b := EncodeReadAtResponse(rar)
+
+	rar2, err := DecodeReadAtResponse(b)
+	assert.NoError(t, err)
+	assert.Equal(t, rar.Bytes, rar2.Bytes)
+	assert.Equal(t, rar.Data, rar2.Data)
+	assert.Equal(t, rar.Error, rar2.Error)
+
+	// Make sure we can't decode silly things
+	_, err = DecodeReadAtResponse(nil)
+	assert.Error(t, err)
+
+	_, err = DecodeReadAtResponse([]byte{
+		99,
+	})
+	assert.Error(t, err)
+
+	// Test encoding error
+	be := EncodeReadAtResponse(&ReadAtResponse{Error: errors.New("Something")})
+
+	rare, err := DecodeReadAtResponse(be)
+	assert.NoError(t, err)
+	assert.Error(t, rare.Error)
+
+}
+
+func TestWriteAt(t *testing.T) {
+
+	b := EncodeWriteAt(12345, []byte{1, 2, 3, 4, 5})
+
+	off, data, err := DecodeWriteAt(b)
+
+	assert.NoError(t, err)
+	assert.Equal(t, int64(12345), off)
+	assert.Equal(t, []byte{1, 2, 3, 4, 5}, data)
+
+	// Make sure we can't decode silly things
+	_, _, err = DecodeWriteAt(nil)
+	assert.Error(t, err)
+
+	_, _, err = DecodeWriteAt([]byte{
+		99,
+	})
+	assert.Error(t, err)
+
+}
+
+func TestWriteAtResponse(t *testing.T) {
+	war := &WriteAtResponse{
+		Bytes: 10,
+		Error: nil,
+	}
+
+	b := EncodeWriteAtResponse(war)
+
+	war2, err := DecodeWriteAtResponse(b)
+	assert.NoError(t, err)
+	assert.Equal(t, war.Bytes, war2.Bytes)
+	assert.Equal(t, war.Error, war2.Error)
+
+	// Make sure we can't decode silly things
+	_, err = DecodeWriteAtResponse(nil)
+	assert.Error(t, err)
+
+	_, err = DecodeWriteAtResponse([]byte{
+		99,
+	})
+	assert.Error(t, err)
+
+	// Test encoding error
+	be := EncodeWriteAtResponse(&WriteAtResponse{Error: errors.New("Something")})
+
+	rare, err := DecodeWriteAtResponse(be)
+	assert.NoError(t, err)
+	assert.Error(t, rare.Error)
+
+}

pkg/storage/protocol/read_at.go

@@ -0,0 +1,68 @@
+package protocol
+
+import (
+	"encoding/binary"
+	"errors"
+)
+
+func EncodeReadAt(offset int64, length int32) []byte {
+	buff := make([]byte, 1+8+4)
+	buff[0] = COMMAND_READ_AT
+	binary.LittleEndian.PutUint64(buff[1:], uint64(offset))
+	binary.LittleEndian.PutUint32(buff[9:], uint32(length))
+	return buff
+}
+
+func DecodeReadAt(buff []byte) (int64, int32, error) {
+	if buff == nil || len(buff) < 13 || buff[0] != COMMAND_READ_AT {
+		return 0, 0, errors.New("Invalid packet")
+	}
+	off := int64(binary.LittleEndian.Uint64(buff[1:]))
+	length := int32(binary.LittleEndian.Uint32(buff[9:]))
+	return off, length, nil
+}
+
+type ReadAtResponse struct {
+	Bytes int
+	Data  []byte
+	Error error
+}
+
+func EncodeReadAtResponse(rar *ReadAtResponse) []byte {
+	if rar.Error != nil {
+		buff := make([]byte, 1)
+		buff[0] = COMMAND_READ_AT_RESPONSE_ERR
+		return buff
+	} else {
+		buff := make([]byte, 1+4+len(rar.Data))
+		buff[0] = COMMAND_READ_AT_RESPONSE
+		binary.LittleEndian.PutUint32(buff[1:], uint32(rar.Bytes))
+		copy(buff[5:], rar.Data)
+		return buff
+	}
+}
+
+func DecodeReadAtResponse(buff []byte) (*ReadAtResponse, error) {
+	if buff == nil {
+		return nil, errors.New("Invalid packet")
+	}
+
+	if buff[0] == COMMAND_READ_AT_RESPONSE_ERR {
+		return &ReadAtResponse{
+			Error: errors.New("Remote error"),
+			Bytes: 0,
+			Data:  make([]byte, 0),
+		}, nil
+	} else if buff[0] == COMMAND_READ_AT_RESPONSE {
+		if len(buff) < 5 {
+			return nil, errors.New("Invalid packet")
+		}
+		return &ReadAtResponse{
+			Error: nil,
+			Bytes: int(binary.LittleEndian.Uint32(buff[1:])),
+			Data:  buff[5:],
+		}, nil
+	}
+
+	return nil, errors.New("Unknown packet")
+}