tlogclient

README

TLOG Client

Tlog client is asynchronous tlog client library. It is currently not goroutine safe.

API

The main APIs are : New, Send and Recv:

• New to create new client instance.
• Send to send transaction to server.
• Recv to get the channel to receive the server
• Close make this client invalid. It is user responsibility to call this func.

response.

usage example

package main

import (
	"flag"
	"io"
	"sync"
	"time"

	"github.com/zero-os/0-Disk/log"
	"github.com/zero-os/0-Disk/tlog"
	"github.com/zero-os/0-Disk/tlog/schema"
	client "github.com/zero-os/0-Disk/tlog/tlogclient"
)

var (
	numFlush  int
	printResp bool
)

func main() {
	flag.IntVar(&numFlush, "num_flush", 40, "number of flush")
	flag.BoolVar(&printResp, "print_resp", false, "print response")

	flag.Parse()

	const (
		vdiskID       = "1234567890"
		firstSequence = 0
		dataLen       = 4096
	)

	client, err := client.New("127.0.0.1:11211", vdiskID, firstSequence)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	data := make([]byte, dataLen)
	for i := 0; i < dataLen; i++ {
		data[i] = 'a'
	}
	data[0] = 'b'
	data[1] = 'c'

	numLogsToSend := 25 * numFlush
	logsToSend := map[uint64]struct{}{}
	for i := 0; i < numLogsToSend; i++ {
		logsToSend[uint64(i)] = struct{}{}
	}

	var wg sync.WaitGroup

	// start the response receiver goroutine
	// wait for all sequences to be flushed
	wg.Add(1)
	go func() {
		defer wg.Done()

		respChan := client.Recv()
		for {
			r := <-respChan
			if r.Err != nil {
				if r.Err == io.EOF {
					continue
				}
				log.Fatalf("resp error:%v", r.Err)
			}
			resp := r.Resp
			if printResp {
				log.Infof("status=%v, seqs=%v\n", resp.Status, resp.Sequences)
			}

			if resp.Status == tlog.BlockStatusFlushOK {
				for _, seq := range resp.Sequences {
					delete(logsToSend, seq)
				}
				if len(logsToSend) == 0 {
					return
				}
			}
		}
	}()

	// send the data
	for i := 0; i < numLogsToSend; i++ {
		seq := uint64(i)
		err := client.Send(schema.OpWrite, seq, seq*dataLen, uint64(time.Now().Unix()), data, uint64(len(data)))
		if err != nil {
			log.Fatalf("send failed at seq=%v, err= %v", seq, err)
			return
		}
	}

	wg.Wait() //wait until we receive all the response
}

Documentation

Index

• Variables
• type Client
  • func New(addr, vdiskID string, firstSequence uint64, resetFirstSeq bool) (*Client, error)
  • func (c *Client) Close() error
  • func (c *Client) ForceFlushAtSeq(seq uint64) error
  • func (c *Client) Recv() <-chan *Result
  • func (c *Client) Send(op uint8, seq, offset, timestamp uint64, data []byte, size uint64) error
  • func (c *Client) WaitNbdSlaveSync() error
• type Response
• type Result

Variables

var (

	// ErrClientClosed returned when client do something when
	// it already closed
	ErrClientClosed = errors.New("client already closed")

	// ErrWaitSlaveSyncTimeout returned when nbd slave sync couldn't be finished
	ErrWaitSlaveSyncTimeout = errors.New("wait nbd slave sync timed out")
)

Types

type Client

type Client struct {
	// contains filtered or unexported fields
}

Client defines a Tlog Client. This client is not thread/goroutine safe.

func New

func New(addr, vdiskID string, firstSequence uint64, resetFirstSeq bool) (*Client, error)

New creates a new tlog client for a vdisk with 'addr' is the tlogserver address. 'firstSequence' is the first sequence number this client is going to send. Set 'resetFirstSeq' to true to force reset the vdisk first/expected sequence. The client is not goroutine safe.

func (*Client) Close

func (c *Client) Close() error

Close the open connection, making this client invalid. It is user responsibility to call this function.

func (*Client) ForceFlushAtSeq

func (c *Client) ForceFlushAtSeq(seq uint64) error

ForceFlushAtSeq force flush at given sequence

func (*Client) Recv

func (c *Client) Recv() <-chan *Result

Recv get channel of responses and errors (Result)

func (*Client) Send

func (c *Client) Send(op uint8, seq, offset, timestamp uint64,
	data []byte, size uint64) error

Send sends the transaction tlog to server. It returns error in these cases: - failed to encode the capnp. - failed to recover from broken network connection.

func (*Client) WaitNbdSlaveSync

func (c *Client) WaitNbdSlaveSync() error

WaitNbdSlaveSync commands tlog server to wait for nbd slave to be fully synced

type Response

type Response struct {
	Status    tlog.BlockStatus // status of the call
	Sequences []uint64         // flushed sequences number (optional)
}

Response defines a response from tlog server

type Result

type Result struct {
	Resp *Response
	Err  error
}

Result defines a struct that contains response and error from tlog.