<commit_before>package tlogclient

import (
	"bufio"
	"context"
	"errors"
	"fmt"
	"io"
	"net"
	"sync"
	"time"

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

const (
	sendRetryNum     = 5
	sendSleepTime    = 500 * time.Millisecond // sleep duration before retrying the Send
	resendTimeoutDur = 2 * time.Second        // duration to wait before re-send the tlog.
	readTimeout      = 2 * time.Second
)

var (
	errMaxSendRetry = errors.New("max send retry reached")

	// 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")
)

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

// Result defines a struct that contains
// response and error from tlog.
type Result struct {
	Resp *Response
	Err  error
}

// Client defines a Tlog Client.
// This client is not thread/goroutine safe.
type Client struct {
	servers         []string
	vdiskID         string
	conn            *net.TCPConn
	bw              writerFlusher
	rd              io.Reader // reader of this client
	blockBuffer     *blockbuffer.Buffer
	capnpSegmentBuf []byte

	// write lock, to protect against parallel Send
	// which is not goroutine safe yet
	wLock sync.Mutex

	// read lock, to protect it from race condition
	// caused by 'handshake' and recvOne
	rLock      sync.Mutex
	ctx        context.Context
	cancelFunc context.CancelFunc

	lastConnected time.Time
	stopped       bool

	waitSlaveSyncCond *sync.Cond

	serverAddrLock sync.Mutex
}

// 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 New(servers []string, vdiskID string, firstSequence uint64, resetFirstSeq bool) (*Client, error) {
	ctx, cancelFunc := context.WithCancel(context.Background())
	client := &Client{
		servers:           servers,
		vdiskID:           vdiskID,
		blockBuffer:       blockbuffer.NewBuffer(resendTimeoutDur),
		ctx:               ctx,
		cancelFunc:        cancelFunc,
		waitSlaveSyncCond: sync.NewCond(&sync.Mutex{}),
	}

	if err := client.connect(firstSequence, resetFirstSeq); err != nil {
		return nil, err
	}

	go client.resender()
	return client, nil
}

// ChangeServerAddresses change servers to the given servers
func (c *Client) ChangeServerAddresses(servers []string) {
	c.serverAddrLock.Lock()
	c.serverAddrLock.Unlock()

	if len(servers) == 0 {
		return
	}
	log.Infof("tlogclient vdisk '%v' change server addrs to '%v'", c.vdiskID, servers)

	// reconnect client if current server not exist in the
	// new addresses
	curExist := func() bool {
		for _, server := range servers {
			if c.servers[0] == server {
				return true
			}
		}
		return false
	}()
	c.servers = servers

	if !curExist {
		log.Infof("tlogclient vdisk '%v' current server is not in new address, close it", c.vdiskID)
		c.conn.Close()
	}

}

// shift server address move current active client to the back
// and use 2nd entry as main server
func (c *Client) shiftServer() {
	c.serverAddrLock.Lock()
	c.serverAddrLock.Unlock()

	c.servers = append(c.servers[1:], c.servers[0])
}

func (c *Client) curServerAddress() string {
	c.serverAddrLock.Lock()
	c.serverAddrLock.Unlock()

	return c.servers[0]
}

// reconnect to server
// it must be called under wLock
func (c *Client) reconnect(closedTime time.Time) error {<commit_msg>fix mutex unlock typos in tlog client code<commit_after>package tlogclient

import (
	"bufio"
	"context"
	"errors"
	"fmt"
	"io"
	"net"
	"sync"
	"time"

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

const (
	sendRetryNum     = 5
	sendSleepTime    = 500 * time.Millisecond // sleep duration before retrying the Send
	resendTimeoutDur = 2 * time.Second        // duration to wait before re-send the tlog.
	readTimeout      = 2 * time.Second
)

var (
	errMaxSendRetry = errors.New("max send retry reached")

	// 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")
)

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

// Result defines a struct that contains
// response and error from tlog.
type Result struct {
	Resp *Response
	Err  error
}

// Client defines a Tlog Client.
// This client is not thread/goroutine safe.
type Client struct {
	servers         []string
	vdiskID         string
	conn            *net.TCPConn
	bw              writerFlusher
	rd              io.Reader // reader of this client
	blockBuffer     *blockbuffer.Buffer
	capnpSegmentBuf []byte

	// write lock, to protect against parallel Send
	// which is not goroutine safe yet
	wLock sync.Mutex

	// read lock, to protect it from race condition
	// caused by 'handshake' and recvOne
	rLock      sync.Mutex
	ctx        context.Context
	cancelFunc context.CancelFunc

	lastConnected time.Time
	stopped       bool

	waitSlaveSyncCond *sync.Cond

	serverAddrLock sync.Mutex
}

// 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 New(servers []string, vdiskID string, firstSequence uint64, resetFirstSeq bool) (*Client, error) {
	ctx, cancelFunc := context.WithCancel(context.Background())
	client := &Client{
		servers:           servers,
		vdiskID:           vdiskID,
		blockBuffer:       blockbuffer.NewBuffer(resendTimeoutDur),
		ctx:               ctx,
		cancelFunc:        cancelFunc,
		waitSlaveSyncCond: sync.NewCond(&sync.Mutex{}),
	}

	if err := client.connect(firstSequence, resetFirstSeq); err != nil {
		return nil, err
	}

	go client.resender()
	return client, nil
}

// ChangeServerAddresses change servers to the given servers
func (c *Client) ChangeServerAddresses(servers []string) {
	c.serverAddrLock.Lock()
	defer c.serverAddrLock.Unlock()

	if len(servers) == 0 {
		return
	}
	log.Infof("tlogclient vdisk '%v' change server addrs to '%v'", c.vdiskID, servers)

	// reconnect client if current server not exist in the
	// new addresses
	curExist := func() bool {
		for _, server := range servers {
			if c.servers[0] == server {
				return true
			}
		}
		return false
	}()
	c.servers = servers

	if !curExist {
		log.Infof("tlogclient vdisk '%v' current server is not in new address, close it", c.vdiskID)
		c.conn.Close()
	}

}

// shift server address move current active client to the back
// and use 2nd entry as main server
func (c *Client) shiftServer() {
	c.serverAddrLock.Lock()
	defer c.serverAddrLock.Unlock()

	c.servers = append(c.servers[1:], c.servers[0])
}

func (c *Client) curServerAddress() string {
	c.serverAddrLock.Lock()
	defer c.serverAddrLock.Unlock()

	return c.servers[0]
}

// reconnect to server
// it must be called under wLock
func (c *Client) reconnect(closedTime time.Time) error {
{'fraction_non_alphanumeric': 0.09830549212834996, 'fraction_numerical': 0.0050474702559788484, 'mean_word_length': 3.3007751937984495, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '892077', 'n_tokens_mistral': 2686, 'n_tokens_neox': 2617, 'n_words': 870}