net.go

package buddystore

import (
	"encoding/json"
	"fmt"
	"net"
	"sync"
	"sync/atomic"
	"time"

	"github.com/golang/glog"
)

/*
TCPTransport provides a TCP based Chord transport layer. This allows Chord
to be implemented over a network, instead of only using the LocalTransport. It is
meant to be a simple implementation, optimizing for simplicity instead of performance.
Messages are sent with a header frame, followed by a body frame. All data is encoded
using the GOB format for simplicity.

Internally, there is 1 Goroutine listening for inbound connections, 1 Goroutine PER
inbound connection.
*/
type TCPTransport struct {
	sock     *net.TCPListener
	timeout  time.Duration
	maxIdle  time.Duration
	lock     sync.RWMutex
	local    map[string]*localRPC
	inbound  map[*net.TCPConn]struct{}
	poolLock sync.Mutex
	pool     map[string][]*tcpOutConn
	shutdown int32

	// Implements:
	Transport
}

var _ Transport = new(TCPTransport)

type tcpOutConn struct {
	host   string
	sock   *net.TCPConn
	header tcpHeader
	enc    *json.Encoder
	dec    *json.Decoder
	used   time.Time
}

const (
	tcpPing = iota
	tcpListReq
	tcpGetPredReq
	tcpGetPredListReq
	tcpNotifyReq
	tcpFindSucReq
	tcpClearPredReq
	tcpSkipSucReq
	tcpGet
	tcpSet
	tcpList
	tcpBulkSet
	tcpSyncKeys
	tcpMissingKeys
	tcpPurgeVersions
	tcpRLockReq
	tcpWLockReq
	tcpCommitWLockReq
	tcpAbortWLockReq
	tcpInvalidateRLockReq
	tcpVersionMapUpdate
	tcpJoinRingReq
)

type tcpHeader struct {
	ReqType int
}

type tcpRequest interface {
}

type TCPResponseBody interface {
}

type TCPResponse interface {
	Error() error
	SetError(error)
}

type TCPResponseImpl struct {
	Err string `json:"err,string,omitempty"`

	// Implements:
	TCPResponse `json:"-"`
}

func (t *TCPResponseImpl) Error() error {
	if t.Err != "" {
		var nerr BuddyStoreError
		json.Unmarshal([]byte(t.Err), &nerr)
		// fmt.Println("TCPResponseImpl: ", t.Err)
		return nerr
	}
	return nil
}

func (t *TCPResponseImpl) SetError(err error) {
	if err != nil {
		var nerr BuddyStoreError
		var ok bool

		if nerr, ok = err.(BuddyStoreError); ok {
		} else {
			nerr = PermanentError(err.Error())
		}

		e, _ := json.Marshal(nerr)
		t.Err = string(e)
		// fmt.Println("Creating error: ", t.Err)
	}
}

var _ TCPResponse = new(tcpBodyVnodeListError)

// Potential body types
type tcpBodyError struct {
	Dummy bool

	// Extends:
	TCPResponseImpl
}
type tcpBodyString struct {
	S string
}
type tcpBodyVnode struct {
	Vn *Vnode
}
type tcpBodyTwoVnode struct {
	Target *Vnode
	Vn     *Vnode
}
type tcpBodyFindSuc struct {
	Target *Vnode
	Num    int
	Key    []byte
}

type tcpBodyVnodeError struct {
	Vnode *Vnode

	// Extends:
	TCPResponseImpl
}

type tcpBodyVnodeListError struct {
	Vnodes []*Vnode

	// Extends:
	TCPResponseImpl
}

type tcpBodyBoolError struct {
	B bool

	// Extends:
	TCPResponseImpl
}

// Creates a new TCP transport on the given listen address with the
// configured timeout duration.
func InitTCPTransport(listen string, timeout time.Duration) (*TCPTransport, error) {
	// Try to start the listener
	sock, err := net.Listen("tcp", listen)
	if err != nil {
		return nil, err
	}

	// allocate maps
	local := make(map[string]*localRPC)
	inbound := make(map[*net.TCPConn]struct{})
	pool := make(map[string][]*tcpOutConn)

	// Maximum age of a connection
	maxIdle := time.Duration(100 * time.Second)

	// Setup the transport
	tcp := &TCPTransport{sock: sock.(*net.TCPListener),
		timeout: timeout,
		maxIdle: maxIdle,
		local:   local,
		inbound: inbound,
		pool:    pool}

	// Listen for connections
	go tcp.listen()

	// Reap old connections
	go tcp.reapOld()

	// Done
	return tcp, nil
}

// Checks for a local vnode
func (t *TCPTransport) get(vn *Vnode) (VnodeRPC, bool) {
	key := vn.String()
	t.lock.RLock()
	defer t.lock.RUnlock()
	w, ok := t.local[key]
	if ok {
		return w.obj, ok
	} else {
		return nil, ok
	}
}

// Gets an outbound connection to a host
func (t *TCPTransport) getConn(host string) (*tcpOutConn, error) {
	// Check if we have a conn cached
	var out *tcpOutConn
	t.poolLock.Lock()
	if atomic.LoadInt32(&t.shutdown) == 1 {
		t.poolLock.Unlock()
		return nil, fmt.Errorf("TCP transport is shutdown")
	}

	list, ok := t.pool[host]
	if ok && len(list) > 0 {
		out = list[len(list)-1]
		list = list[:len(list)-1]
		t.pool[host] = list
	}
	t.poolLock.Unlock()

	if out == nil {
		// Try to establish a connection
		conn, err := net.DialTimeout("tcp", host, t.timeout)
		if err != nil {
			return nil, err
		}

		// Setup the socket
		sock := conn.(*net.TCPConn)
		t.setupConn(sock)

		out = &tcpOutConn{host: host, sock: sock}
	}

	out.enc = json.NewEncoder(out.sock)
	out.dec = json.NewDecoder(out.sock)
	out.used = time.Now()

	// Wrap the sock
	return out, nil
}

// Returns an outbound TCP connection to the pool
func (t *TCPTransport) returnConn(o *tcpOutConn) {
	// Update the last used time
	o.used = time.Now()

	// Push back into the pool
	t.poolLock.Lock()
	defer t.poolLock.Unlock()
	if atomic.LoadInt32(&t.shutdown) == 1 {
		o.sock.Close()
		return
	}
	list, _ := t.pool[o.host]
	t.pool[o.host] = append(list, o)
}

// Setup a connection
func (t *TCPTransport) setupConn(c *net.TCPConn) {
	c.SetNoDelay(true)
	c.SetKeepAlive(true)
}

func (t *TCPTransport) networkCall(host string, tcpReqType int, req tcpRequest, resp TCPResponse) error {
	// Get a conn
	out, err := t.getConn(host)
	if err != nil {
		return err
	}

	// Response channels
	respChan := make(chan bool, 1)
	errChan := make(chan error, 1)

	go func() {
		out.header.ReqType = tcpReqType
		if err := out.enc.Encode(&out.header); err != nil {
			errChan <- err
			return
		}
		if err := out.enc.Encode(req); err != nil {
			errChan <- err
			return
		}

		// Read in the response
		if err := out.dec.Decode(resp); err != nil {
			errChan <- err
		}

		// glog.Infof("Resp: %s", resp.Error())

		// Return the connection
		t.returnConn(out)
		if resp.Error() == nil {
			respChan <- true
		} else {
			errChan <- resp.Error()
		}
	}()

	select {
	case <-time.After(t.timeout):
		return fmt.Errorf("Command timed out!")
	case err := <-errChan:
		return err
	case <-respChan:
		return nil
	}
}

// Gets a list of the vnodes on the box
func (t *TCPTransport) ListVnodes(host string) ([]*Vnode, error) {
	resp := tcpBodyVnodeListError{}
	err := t.networkCall(host, tcpListReq, tcpBodyString{S: host}, &resp)

	if err != nil {
		return nil, err
	} else {
		return resp.Vnodes, nil
	}
}

// Ping a Vnode, check for liveness
func (t *TCPTransport) Ping(vn *Vnode) (bool, error) {
	resp := tcpBodyBoolError{}
	err := t.networkCall(vn.Host, tcpPing, tcpBodyVnode{Vn: vn}, &resp)

	if err != nil {
		return false, err
	} else {
		return resp.B, nil
	}
}

// Request a nodes predecessor
func (t *TCPTransport) GetPredecessor(vn *Vnode) (*Vnode, error) {
	resp := tcpBodyVnodeError{}
	err := t.networkCall(vn.Host, tcpGetPredReq, tcpBodyVnode{Vn: vn}, &resp)

	if err != nil {
		return nil, err
	} else {
		return resp.Vnode, nil
	}
}

// Notify our successor of ourselves
func (t *TCPTransport) Notify(target, self *Vnode) ([]*Vnode, error) {
	resp := tcpBodyVnodeListError{}
	err := t.networkCall(target.Host, tcpNotifyReq, tcpBodyTwoVnode{Target: target, Vn: self}, &resp)

	if err != nil {
		return nil, err
	} else {
		return resp.Vnodes, nil
	}
}

// Find a successor
func (t *TCPTransport) FindSuccessors(vn *Vnode, n int, k []byte) ([]*Vnode, error) {
	resp := new(tcpBodyVnodeListError)
	vn.fullLock.RLock()
	str := vn.Host
	req := tcpBodyFindSuc{Target: vn, Num: n, Key: k}
	vn.fullLock.RUnlock()

	err := t.networkCall(str, tcpFindSucReq, req, resp)

	if err != nil {
		return nil, err
	} else {
		return resp.Vnodes, nil
	}
}

// Request a nodes predecessor list
func (t *TCPTransport) GetPredecessorList(vn *Vnode) ([]*Vnode, error) {
	resp := tcpBodyVnodeListError{}
	err := t.networkCall(vn.Host, tcpGetPredListReq, tcpBodyVnode{Vn: vn}, &resp)

	if err != nil {
		return nil, err
	} else {
		return resp.Vnodes, nil
	}
}

/* Transport operation that gets the value of a given key - This operation is additional to what is there in the interface already
 */

func (t *TCPTransport) Get(target *Vnode, key string, version uint) ([]byte, error) {
	resp := tcpBodyRespValue{}
	err := t.networkCall(target.Host, tcpGet, tcpBodyGet{Vnode: target, Key: key, Version: version}, &resp)

	if err != nil {
		return nil, err
	} else {
		return resp.Value, nil
	}
}

/* Transport operation that sets the value of a given key - This operation is additional to what is there in the interface already
 */

func (t *TCPTransport) Set(target *Vnode, key string, version uint, value []byte) error {
	resp := tcpBodyError{}
	err := t.networkCall(target.Host, tcpSet, tcpBodySet{Vnode: target, Key: key, Version: version, Value: value}, &resp)

	if err != nil {
		return err
	} else {
		return nil
	}
}

/* Transport operation that lists the keys for a particular ring - This operation is additional to what is there in the interface already
 */

func (t *TCPTransport) List(target *Vnode) ([]string, error) {
	resp := tcpBodyRespKeys{}
	err := t.networkCall(target.Host, tcpList, tcpBodyList{Vnode: target}, &resp)

	if err != nil {
		return nil, err
	} else {
		return resp.Keys, nil
	}
}

func (t *TCPTransport) BulkSet(target *Vnode, key string, valLst []KVStoreValue) error {
	resp := tcpBodyError{}
	err := t.networkCall(target.Host, tcpBulkSet, tcpBodyBulkSet{Vnode: target, Key: key, ValueLst: valLst}, &resp)

	if err != nil {
		return err
	} else {
		return nil
	}
}

func (t *TCPTransport) SyncKeys(target *Vnode, ownerVn *Vnode, key string, ver []uint) error {
	resp := tcpBodyError{}
	err := t.networkCall(target.Host, tcpSyncKeys, tcpBodySyncKeys{Vnode: target, OwnerVn: ownerVn, Key: key, Version: ver}, &resp)

	if err != nil {
		return err
	} else {
		return nil
	}
}

func (t *TCPTransport) MissingKeys(target *Vnode, replVn *Vnode, key string, ver []uint) error {
	resp := tcpBodyError{}
	err := t.networkCall(target.Host, tcpMissingKeys, tcpBodyMissingKeys{Vnode: target, ReplVn: replVn, Key: key, Version: ver}, &resp)

	if err != nil {
		return err
	} else {
		return nil
	}
}

func (t *TCPTransport) PurgeVersions(target *Vnode, key string, maxVersion uint) error {
	resp := tcpBodyError{}
	err := t.networkCall(target.Host, tcpPurgeVersions, tcpBodyPurgeVersions{Vnode: target, Key: key, MaxVersion: maxVersion}, &resp)

	if err != nil {
		return err
	} else {
		return nil
	}
}

func (t *TCPTransport) JoinRing(target *Vnode, ringId string, joiner *Vnode) ([]*Vnode, error) {
	resp := tcpBodyJoinRingResp{}
	err := t.networkCall(target.Host, tcpJoinRingReq, tcpBodyJoinRingReq{Target: target, RingId: ringId, Joiner: joiner}, &resp)

	if err != nil {
		return nil, err
	} else {
		return resp.Vnodes, nil
	}
}

// Clears a predecessor if it matches a given vnode. Used to leave.
func (t *TCPTransport) ClearPredecessor(target, self *Vnode) error {
	resp := tcpBodyError{}
	err := t.networkCall(target.Host, tcpClearPredReq, tcpBodyTwoVnode{Target: target, Vn: self}, &resp)

	if err != nil {
		return err
	} else {
		return nil
	}
}

// Instructs a node to skip a given successor. Used to leave.
func (t *TCPTransport) SkipSuccessor(target, self *Vnode) error {
	resp := tcpBodyError{}
	err := t.networkCall(target.Host, tcpSkipSucReq, tcpBodyTwoVnode{Target: target, Vn: self}, &resp)

	if err != nil {
		return err
	} else {
		return nil
	}
}

// Register for an RPC callbacks
func (t *TCPTransport) Register(v *Vnode, o VnodeRPC) {
	key := v.String()
	t.lock.Lock()
	t.local[key] = &localRPC{v, o}
	t.lock.Unlock()
}

// Shutdown the TCP transport
func (t *TCPTransport) Shutdown() {
	atomic.StoreInt32(&t.shutdown, 1)
	t.sock.Close()

	// Close all the inbound connections
	t.lock.RLock()
	for conn := range t.inbound {
		conn.Close()
	}
	t.lock.RUnlock()

	// Close all the outbound
	t.poolLock.Lock()
	for _, conns := range t.pool {
		for _, out := range conns {
			out.sock.Close()
		}
	}
	t.pool = nil
	t.poolLock.Unlock()
}

// Closes old outbound connections
func (t *TCPTransport) reapOld() {
	for {
		if atomic.LoadInt32(&t.shutdown) == 1 {
			return
		}
		time.Sleep(30 * time.Second)
		t.reapOnce()
	}
}

func (t *TCPTransport) reapOnce() {
	t.poolLock.Lock()
	defer t.poolLock.Unlock()
	for host, conns := range t.pool {
		max := len(conns)
		for i := 0; i < max; i++ {
			if time.Since(conns[i].used) > t.maxIdle {
				conns[i].sock.Close()
				conns[i], conns[max-1] = conns[max-1], nil
				max--
				i--
			}
		}
		// Trim any idle conns
		t.pool[host] = conns[:max]
	}
}

// Listens for inbound connections
func (t *TCPTransport) listen() {
	for {
		conn, err := t.sock.AcceptTCP()
		if err != nil {
			if atomic.LoadInt32(&t.shutdown) == 0 {
				glog.Errorf("Error accepting TCP connection! %s", err)
				continue
			} else {
				return
			}
		}

		// Setup the conn
		t.setupConn(conn)

		// Register the inbound conn
		t.lock.Lock()
		t.inbound[conn] = struct{}{}
		t.lock.Unlock()

		// Start handler
		go t.handleConn(conn)
	}
}

/*
RLock tranasport layer implementation
Param Vnode : The destination Vnode i.e. the Lock Manager
Param key : The key for which the read lock should be obtained
*/
func (t *TCPTransport) RLock(target *Vnode, key string, nodeID string, opsLogEntry *OpsLogEntry) (string, uint, uint64, error) {
	resp := tcpBodyLMRLockResp{}
	for k, _ := range t.local {
		nodeID = k
		break //  Think of a better way to get the local nodeID
	}
	err := t.networkCall(target.Host, tcpRLockReq, tcpBodyLMRLockReq{Vn: target, Key: key, SenderID: nodeID, SenderAddr: t.sock.Addr().String(), OpsLogEntryPrimary: opsLogEntry}, &resp)

	if err != nil {
		return "", 0, 0, resp.Error()
	} else {
		return resp.LockId, resp.Version, resp.CommitPoint, nil
	}
}

/*
WLock transport layer implementation
Param Vnode : The destination Vnode i.e. the Vnode with the Lock Manager
Param key : The key for which the write lock should be obtained
Param version : The version of the key
Param timeout : Requested Timeout value.
Param NodeID : NodeID of the requesting node
*/
func (t *TCPTransport) WLock(target *Vnode, key string, version uint, timeout uint, nodeID string, opsLogEntry *OpsLogEntry) (string, uint, uint, uint64, error) {
	resp := tcpBodyLMWLockResp{}
	err := t.networkCall(target.Host, tcpWLockReq, tcpBodyLMWLockReq{Vn: target, Key: key, Version: version, Timeout: timeout, SenderID: nodeID, OpsLogEntryPrimary: opsLogEntry}, &resp)

	if err != nil {
		return "", 0, 0, 0, err
	} else {
		return resp.LockId, resp.Version, resp.Timeout, resp.CommitPoint, nil
	}
}

/*
CommitWLock transport layer implementation
Param Vnode : The destination Vnode i.e. the Lock Manager
Param key : The key for which the read lock should be obtained
Param version : The version of the key to be committed
*/
func (t *TCPTransport) CommitWLock(target *Vnode, key string, version uint, nodeID string, opsLogEntry *OpsLogEntry) (uint64, error) {
	resp := tcpBodyLMCommitWLockResp{}
	body := tcpBodyLMCommitWLockReq{Vn: target, Key: key, Version: version, SenderID: nodeID, OpsLogEntryPrimary: opsLogEntry}
	err := t.networkCall(target.Host, tcpCommitWLockReq, body, &resp)

	if err != nil {
		return 0, resp.Error()
	} else {
		return resp.CommitPoint, nil
	}
}

/*
InvalidateRLock transport layer implementation
Param Vnode : The destination Vnode i.e. the Client where the RLock should be invalidated
Param lockID : The exact lock to be invalidated
*/
func (t *TCPTransport) InvalidateRLock(target *Vnode, lockID string) error {
	// Get a conn
	out, err := t.getConn(target.Host)
	if err != nil {
		return err
	}

	respChan := make(chan bool, 1)
	errChan := make(chan error, 1)

	resp := tcpBodyLMInvalidateRLockResp{}
	go func() {
		// Send a list command
		out.header.ReqType = tcpInvalidateRLockReq
		body := tcpBodyLMInvalidateRLockReq{Vn: target, LockID: lockID}
		if err := out.enc.Encode(&out.header); err != nil {
			errChan <- err
			return
		}
		if err := out.enc.Encode(&body); err != nil {
			errChan <- err
			return
		}

		// Read in the response
		if err := out.dec.Decode(&resp); err != nil {
			errChan <- err
			return
		}

		// Return the connection
		t.returnConn(out)
		if resp.Error() == nil {
			respChan <- true
		} else {
			errChan <- resp.Error()
		}
	}()

	select {
	case <-time.After(t.timeout):
		return fmt.Errorf("Command timed out!")
	case _ = <-errChan:
		return resp.Error()
	case <-respChan:
		return resp.Error()
	}
}

/*
AbortWLock transport layer implementation
Param Vnode : The destination Vnode i.e. the Lock Manager
Param key : The key for which the read lock should be obtained
*/
func (t *TCPTransport) AbortWLock(target *Vnode, key string, version uint, nodeID string, opsLogEntry *OpsLogEntry) (uint64, error) {
	resp := tcpBodyLMAbortWLockResp{}
	body := tcpBodyLMAbortWLockReq{Vn: target, Key: key, Version: version, SenderID: nodeID, OpsLogEntryPrimary: opsLogEntry}
	err := t.networkCall(target.Host, tcpAbortWLockReq, body, &resp)

	if err != nil {
		return 0, resp.Error()
	} else {
		return resp.CommitPoint, nil
	}
}

// Handles inbound TCP connections
func (t *TCPTransport) handleConn(conn *net.TCPConn) {
	// Defer the cleanup
	defer func() {
		t.lock.Lock()
		delete(t.inbound, conn)
		t.lock.Unlock()
		conn.Close()
	}()

	dec := json.NewDecoder(conn)
	enc := json.NewEncoder(conn)
	header := tcpHeader{}
	var sendResp TCPResponse
	for {
		// Get the header
		if err := dec.Decode(&header); err != nil {
			if atomic.LoadInt32(&t.shutdown) == 0 && err.Error() != "EOF" {
				glog.Errorf("Failed to decode TCP header! Got %s", err)
			}
			return
		}

		// Read in the body and process request
		switch header.ReqType {
		case tcpPing:
			body := tcpBodyVnode{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			_, ok := t.get(body.Vn)
			if ok {
				sendResp = &tcpBodyBoolError{B: ok}
			} else {
				sendResp = &tcpBodyBoolError{B: ok}
				sendResp.SetError(fmt.Errorf("Target VN not found! Target %s:%s", body.Vn.Host, body.Vn.String()))
			}

		case tcpListReq:
			body := tcpBodyString{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate all the local clients
			res := make([]*Vnode, 0, len(t.local))

			// Build list
			t.lock.RLock()
			for _, v := range t.local {
				res = append(res, v.vnode)
			}
			t.lock.RUnlock()

			// Make response
			sendResp = &tcpBodyVnodeListError{Vnodes: trimSlice(res)}

		case tcpGetPredReq:
			body := tcpBodyVnode{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, ok := t.get(body.Vn)
			resp := tcpBodyVnodeError{}
			sendResp = &resp
			if ok {
				node, err := obj.GetPredecessor()
				resp.Vnode = node
				resp.SetError(err)
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Vn.Host, body.Vn.String()))
			}

		case tcpNotifyReq:
			body := tcpBodyTwoVnode{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, ok := t.get(body.Target)
			resp := tcpBodyVnodeListError{}
			sendResp = &resp
			if ok {
				nodes, err := obj.Notify(body.Vn)
				resp.Vnodes = trimSlice(nodes)
				resp.SetError(err)
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Target.Host, body.Target.String()))
			}

		case tcpFindSucReq:
			body := tcpBodyFindSuc{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, ok := t.get(body.Target)
			resp := tcpBodyVnodeListError{}
			sendResp = &resp
			if ok {
				nodes, err := obj.FindSuccessors(body.Num, body.Key)
				resp.Vnodes = trimSlice(nodes)
				resp.SetError(err)
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Target.Host, body.Target.String()))
			}

		case tcpClearPredReq:
			body := tcpBodyTwoVnode{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, ok := t.get(body.Target)
			resp := tcpBodyError{}
			sendResp = &resp
			if ok {
				resp.SetError(obj.ClearPredecessor(body.Vn))
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Target.Host, body.Target.String()))
			}

		case tcpSkipSucReq:
			body := tcpBodyTwoVnode{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, ok := t.get(body.Target)
			resp := tcpBodyError{}
			sendResp = &resp
			if ok {
				resp.SetError(obj.SkipSuccessor(body.Vn))
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Target.Host, body.Target.String()))
			}

		case tcpGetPredListReq:
			body := tcpBodyVnode{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, ok := t.get(body.Vn)
			resp := tcpBodyVnodeListError{}
			sendResp = &resp
			if ok {
				nodes, err := obj.GetPredecessorList()
				resp.Vnodes = nodes
				resp.SetError(err)
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Vn.Host, body.Vn.String()))
			}

		case tcpGet:
			body := tcpBodyGet{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, ok := t.get(body.Vnode)
			resp := tcpBodyRespValue{}
			sendResp = &resp
			if ok {
				value, err := obj.Get(body.Key, body.Version)
				resp.Value = value
				resp.SetError(err)
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Vnode.Host, body.Vnode.String()))
			}

		case tcpSet:
			body := tcpBodySet{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, ok := t.get(body.Vnode)
			resp := tcpBodyError{}
			sendResp = &resp
			if ok {
				err := obj.Set(body.Key, body.Version, body.Value)

				resp.SetError(err)
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Vnode.Host, body.Vnode.String()))
			}

		case tcpList:
			body := tcpBodyList{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, ok := t.get(body.Vnode)
			resp := tcpBodyRespKeys{}
			sendResp = &resp
			if ok {
				keys, err := obj.List()
				resp.Keys = keys
				resp.SetError(err)
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Vnode.Host, body.Vnode.String()))
			}

		case tcpBulkSet:
			body := tcpBodyBulkSet{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, ok := t.get(body.Vnode)
			resp := tcpBodyError{}
			sendResp = &resp
			if ok {
				err := obj.BulkSet(body.Key, body.ValueLst)

				resp.SetError(err)
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Vnode.Host, body.Vnode.String()))
			}

		case tcpSyncKeys:
			body := tcpBodySyncKeys{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, ok := t.get(body.Vnode)
			resp := tcpBodyError{}
			sendResp = &resp
			if ok {
				err := obj.SyncKeys(body.OwnerVn, body.Key, body.Version)
				resp.SetError(err)
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Vnode.Host, body.Vnode.String()))
			}

		case tcpMissingKeys:
			body := tcpBodyMissingKeys{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, ok := t.get(body.Vnode)
			resp := tcpBodyError{}
			sendResp = &resp
			if ok {
				err := obj.MissingKeys(body.ReplVn, body.Key, body.Version)
				resp.SetError(err)
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Vnode.Host, body.Vnode.String()))
			}

		case tcpPurgeVersions:
			body := tcpBodyPurgeVersions{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, ok := t.get(body.Vnode)
			resp := tcpBodyError{}
			sendResp = &resp
			if ok {
				err := obj.PurgeVersions(body.Key, body.MaxVersion)
				resp.SetError(err)
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Vnode.Host, body.Vnode.String()))
			}

		case tcpJoinRingReq:
			body := tcpBodyJoinRingReq{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, ok := t.get(body.Target)
			resp := tcpBodyJoinRingResp{}
			sendResp = &resp
			if ok {
				vnodes, err := obj.JoinRing(body.RingId, body.Joiner)
				resp.Vnodes = vnodes
				resp.SetError(err)
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Target.Host, body.Target.String()))
			}

		case tcpRLockReq:
			body := tcpBodyLMRLockReq{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, ok := t.get(body.Vn)
			resp := tcpBodyLMRLockResp{}
			sendResp = &resp
			if ok {
				lockId, version, cp, err :=
					obj.RLock(body.Key, body.SenderID, body.SenderAddr, body.OpsLogEntryPrimary)

				resp.SetError(err)
				resp.LockId = lockId
				resp.Version = version
				resp.CommitPoint = cp
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Vn.Host, body.Vn.String()))
			}

		case tcpWLockReq:
			body := tcpBodyLMWLockReq{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, ok := t.get(body.Vn)
			resp := tcpBodyLMWLockResp{}
			sendResp = &resp
			if ok {
				lockId, version, timeout, commitPoint, err :=
					obj.WLock(body.Key, body.Version, body.Timeout, body.SenderID, body.OpsLogEntryPrimary)

				resp.SetError(err)
				resp.LockId = lockId
				resp.Version = version
				resp.Timeout = timeout
				resp.CommitPoint = commitPoint
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Vn.Host, body.Vn.String()))
			}

		case tcpCommitWLockReq:
			body := tcpBodyLMCommitWLockReq{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, ok := t.get(body.Vn)
			resp := tcpBodyLMCommitWLockResp{}
			sendResp = &resp
			if ok {
				cp, err := obj.CommitWLock(body.Key, body.Version, body.SenderID, body.OpsLogEntryPrimary)
				resp.CommitPoint = cp
				resp.SetError(err)
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Vn.Host, body.Vn.String()))
			}

		case tcpAbortWLockReq:
			body := tcpBodyLMAbortWLockReq{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, ok := t.get(body.Vn)
			resp := tcpBodyLMAbortWLockResp{}
			sendResp = &resp
			if ok {
				cp, err := obj.AbortWLock(body.Key, body.Version, body.SenderID, body.OpsLogEntryPrimary)
				resp.CommitPoint = cp
				resp.SetError(err)
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Vn.Host, body.Vn.String()))
			}

		case tcpInvalidateRLockReq:
			body := tcpBodyLMInvalidateRLockReq{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, _ := t.get(body.Vn)
			resp := tcpBodyLMInvalidateRLockResp{}
			sendResp = &resp
			if obj != nil {
				_ = obj.InvalidateRLock(body.LockID)
				resp.SetError(nil) // Change it to incorporate the error from the client
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Vn.Host, body.Vn.String()))
			}

		case tcpVersionMapUpdate:
			body := tcpVersionMapUpdateReq{}
			if err := dec.Decode(&body); err != nil {
				glog.Errorf("Failed to decode TCP body! Got %s", err)
				return
			}

			// Generate a response
			obj, _ := t.get(body.Vn)
			resp := tcpVersionMapUpdateResp{}
			sendResp = &resp
			if obj != nil {
				obj.UpdateVersionMap(body.VersionMap)
				resp.SetError(nil)
			} else {
				resp.SetError(fmt.Errorf("Target VN not found! Target %s:%s",
					body.Vn.Host, body.Vn.String()))
			}

		default:
			glog.Errorf("Unknown request type! Got %d", header.ReqType)
			return
		}

		// Send the response
		if err := enc.Encode(sendResp); err != nil {
			glog.Errorf("Failed to send TCP body! Got %s", err)
			return
		}
	}
}

// Trims the slice to remove nil elements
func trimSlice(vn []*Vnode) []*Vnode {
	if vn == nil {
		return vn
	}

	// Find a non-nil index
	idx := len(vn) - 1

	for vn[idx] == nil {
		idx--
	}
	return vn[:idx+1]
}