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(addrs []string, vdiskID string, firstSequence uint64, resetFirstSeq bool) (*Client, error)
  • func (c *Client) ChangeServerAddrs(addrs []string)
  • 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")

	// ErrFlushFailed returned when client failed to do flush
	ErrFlushFailed = errors.New("tlogserver failed to flush")
)

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(addrs []string, vdiskID string, firstSequence uint64, resetFirstSeq bool) (*Client, error)

New creates a new tlog client for a vdisk with 'addrs' is the tlogserver addresses. Client is going to use first address and then move to next addresses if the first address is failed. '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) ChangeServerAddrs

func (c *Client) ChangeServerAddrs(addrs []string)

ChangeServerAddrs change server addresses to the given addresses addrs

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 NOTE : this func doesn't have retry logic, user need to add it

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.