lib9p

client.go (10157B)

---

 package client
 
 import (
 	"context"
 	"errors"
 	"fmt"
 	"io"
 	"log"
 	"sync"
 
 	"git.mtkn.jp/lib9p"
 )
 
 // Client is a client side of the 9P conversation.
 type Client struct {
 	// Msize is the maximum message size in length
 	msize uint32
 	// MSizeLock is the mutex used when msize is to be changed.
 	mSizeLock *sync.Mutex
 
 	// Uname is used to communicate with a server.
 	uname string
 
 	// RPool is the set of all outstanding requests.
 	rPool *reqPool
 
 	// Txc is used to send a reqest to the multiplexer goroutine
 	txc chan<- *req
 
 	// Wg is the WaitGroup of all goroutines evoked by this client and its
 	// descendants.
 	wg *sync.WaitGroup
 
 	// Done is closed when the context passed to NewClient is canceled.
 	// This is used to notify transact() that the client is already
 	// canceled and the transaction should also be canceled.
 	done <-chan struct{}
 }
 
 // NewClient creates a Client and prepare to transact with a server via r and w.
 // It runs several goroutines to handle requests.
 // And the returned client should be stopped afterwords by cancelling ctx
 // and then closing r and w.
 func NewClient(ctx context.Context, mSize uint32, uname string, r io.Reader, w io.Writer) *Client {
 	c := &Client{
 		msize:     mSize,
 		mSizeLock: new(sync.Mutex),
 		uname:     uname,
 		rPool:     newReqPool(),
 		wg:        new(sync.WaitGroup),
 		done:      ctx.Done(),
 	}
 	c.txc = c.runMultiplexer(ctx, r, w)
 	return c
 }
 
 // mSize returns the maximum message size of the Client.
 func (c *Client) mSize() uint32 {
 	c.mSizeLock.Lock()
 	defer c.mSizeLock.Unlock()
 	return c.msize
 }
 
 // setMSize changes the maximum message size of the Client.
 func (c *Client) setMSize(mSize uint32) {
 	c.mSizeLock.Lock()
 	defer c.mSizeLock.Unlock()
 	c.msize = mSize
 }
 
 // RunMultiplexer runs two goroutines,
 // one for recieving Rmsg and another for sending Tmsg.
 // The goroutine for Tmsg recieves *req from the returned channel,
 // and sends the lib9p.Msg to w.
 // The goroutine for Rmsg reads lib9p.Msg from r and sends it to rxc of the
 // corresponding request.
 func (c *Client) runMultiplexer(ctx context.Context, r io.Reader, w io.Writer) chan<- *req {
 	c.wg.Add(2)
 	txc := make(chan *req)
 	// Rmsg
 	go func() {
 		defer c.wg.Done()
 		for {
 			select {
 			case <-ctx.Done():
 				return
 			default:
 			}
 			msg, err := lib9p.RecvMsg(r)
 			if err != nil {
 				c.rPool.cancelAll(fmt.Errorf("recv: %v", err))
 				continue // TODO: should return?
 			}
 			rq, ok := c.rPool.lookup(msg.GetTag())
 			if !ok {
 				log.Printf("mux: unknown tag for msg: %v", msg)
 				continue // TODO: how to recover?
 			}
 			c.rPool.delete(msg.GetTag())
 			if tflush, ok := rq.tmsg.(*lib9p.TFlush); ok {
 				if _, ok := msg.(*lib9p.RFlush); !ok {
 					rq.errc <- fmt.Errorf("mux: response to Tflush is not Rflush")
 				}
 				if oldreq, ok := c.rPool.lookup(tflush.Oldtag); ok {
 					oldreq.errc <- errors.New("request flushed")
 					c.rPool.delete(tflush.Oldtag)
 				}
 			}
 			rq.rmsg = msg
 			select {
 			case <-ctx.Done():
 			case rq.rxc <- rq:
 			}
 		}
 	}()
 	// Tmsg
 	go func() {
 		defer c.wg.Done()
 		for {
 			select {
 			case <-ctx.Done():
 				return
 			case r, ok := <-txc:
 				if !ok {
 					return
 				}
 				if _, ok := c.rPool.lookup(r.tag); ok {
 					r.errc <- fmt.Errorf("mux: %w: %d", lib9p.ErrDupTag, r.tag)
 					continue
 				}
 				c.rPool.add(r)
 				if err := lib9p.SendMsg(r.tmsg, w); err != nil {
 					r.errc <- fmt.Errorf("send: %v", err)
 				}
 			}
 		}
 	}()
 	return txc
 }
 
 // Transact sends 9P tmsg to the multiplexer and recieves
 // the reply.
 func (c *Client) transact(tmsg lib9p.Msg) (lib9p.Msg, error) {
 	r := newReq(tmsg)
 	select {
 	case c.txc <- r:
 	case <-c.done:
 		return nil, errors.New("client stopped")
 	}
 	select {
 	case r := <-r.rxc:
 		return r.rmsg, r.err
 	case err := <-r.errc: // Client side error.
 		return nil, err
 	case <-c.done:
 		return nil, errors.New("client stopped")
 	}
 }
 
 // Version sends Tversion message to the server and returns the resulting
 // data of Rversion or non nil error if any.
 // This function and other Tmessage functions don't have a context.Contex
 // as their argument.
 // The caller can call *Client.Flush to cancel a pending request if the
 // connection to the server is helthy.
 // And even if the connection has some problem sending/recieving, there
 // is no way to cancel blocking reads/writes. In this case, the caller
 // can close the connection.
 func (c *Client) Version(tag uint16, msize uint32, version string) (uint32, string, error) {
 	tmsg := &lib9p.TVersion{Tag: tag, Msize: msize, Version: version}
 	rmsg, err := c.transact(tmsg)
 	if err != nil {
 		return 0, "", fmt.Errorf("transact: %w", err)
 	}
 	switch rmsg := rmsg.(type) {
 	case *lib9p.RVersion:
 		return rmsg.Msize, rmsg.Version, nil
 	case *lib9p.RError:
 		return 0, "", rmsg.Ename
 	default:
 		return 0, "", fmt.Errorf("invalid reply: %v", rmsg)
 	}
 }
 
 func (c *Client) Auth(tag uint16, afid uint32, uname, aname string) (lib9p.Qid, error) {
 	tmsg := &lib9p.TAuth{Tag: tag, Afid: afid, Uname: uname}
 	rmsg, err := c.transact(tmsg)
 	if err != nil {
 		return lib9p.Qid{}, fmt.Errorf("transact: %w", err)
 	}
 	switch rmsg := rmsg.(type) {
 	case *lib9p.RAuth:
 		return rmsg.Aqid, nil
 	case *lib9p.RError:
 		return lib9p.Qid{}, rmsg.Ename
 	default:
 		return lib9p.Qid{}, fmt.Errorf("invalid reply: %v", rmsg)
 	}
 }
 
 func (c *Client) Attach(tag uint16, fid, afid uint32, uname, aname string) (lib9p.Qid, error) {
 	tmsg := &lib9p.TAttach{Tag: tag, Fid: fid, Afid: afid, Uname: uname, Aname: aname}
 	rmsg, err := c.transact(tmsg)
 	if err != nil {
 		return lib9p.Qid{}, fmt.Errorf("transact: %w", err)
 	}
 	switch rmsg := rmsg.(type) {
 	case *lib9p.RAttach:
 		return rmsg.Qid, nil
 	case *lib9p.RError:
 		return lib9p.Qid{}, rmsg.Ename
 	default:
 		return lib9p.Qid{}, fmt.Errorf("invalid reply: %v", rmsg)
 	}
 }
 
 func (c *Client) Flush(tag, oldtag uint16) error {
 	tmsg := &lib9p.TFlush{Tag: tag, Oldtag: oldtag}
 	rmsg, err := c.transact(tmsg)
 	if err != nil {
 		return fmt.Errorf("transact: %w", err)
 	}
 	switch rmsg := rmsg.(type) {
 	case *lib9p.RFlush:
 		return nil
 	case *lib9p.RError:
 		return rmsg.Ename
 	default:
 		return fmt.Errorf("invalid reply: %v", rmsg)
 	}
 }
 func (c *Client) Walk(tag uint16, fid, newfid uint32, wnames []string) (wqid []lib9p.Qid, err error) {
 	tmsg := &lib9p.TWalk{Tag: tag, Fid: fid, Newfid: newfid, Wnames: wnames}
 	rmsg, err := c.transact(tmsg)
 	if err != nil {
 		return nil, fmt.Errorf("transact: %w", err)
 	}
 	switch rmsg := rmsg.(type) {
 	case *lib9p.RWalk:
 		return rmsg.Qids, nil
 	case *lib9p.RError:
 		return nil, rmsg.Ename
 	default:
 		return nil, fmt.Errorf("invalid reply: %v", rmsg)
 	}
 }
 func (c *Client) Open(tag uint16, fid uint32, mode lib9p.OpenMode) (qid lib9p.Qid, iounit uint32, err error) {
 	tmsg := &lib9p.TOpen{Tag: tag, Fid: fid, Mode: mode}
 	rmsg, err := c.transact(tmsg)
 	if err != nil {
 		return lib9p.Qid{}, 0, fmt.Errorf("transact: %w", err)
 	}
 	switch rmsg := rmsg.(type) {
 	case *lib9p.ROpen:
 		return rmsg.Qid, rmsg.Iounit, nil
 	case *lib9p.RError:
 		return lib9p.Qid{}, 0, rmsg.Ename
 	default:
 		return lib9p.Qid{}, 0, fmt.Errorf("invalid reply: %v", rmsg)
 	}
 }
 func (c *Client) Create(tag uint16, fid uint32, name string, perm lib9p.FileMode, mode lib9p.OpenMode) (lib9p.Qid, uint32, error) {
 	tmsg := &lib9p.TCreate{Tag: tag, Fid: fid, Name: name, Perm: perm, Mode: mode}
 	rmsg, err := c.transact(tmsg)
 	if err != nil {
 		return lib9p.Qid{}, 0, fmt.Errorf("transact: %w", err)
 	}
 	switch rmsg := rmsg.(type) {
 	case *lib9p.RCreate:
 		return rmsg.Qid, rmsg.Iounit, nil
 	case *lib9p.RError:
 		return lib9p.Qid{}, 0, rmsg.Ename
 	default:
 		return lib9p.Qid{}, 0, fmt.Errorf("invalid reply: %v", rmsg)
 	}
 }
 
 // Read doesn't return io.EOF at the end of file,
 // but returns empty data and nil error.
 func (c *Client) Read(tag uint16, fid uint32, offset uint64, count uint32) (data []byte, err error) {
 	tmsg := &lib9p.TRead{Tag: tag, Fid: fid, Offset: offset, Count: count}
 	rmsg, err := c.transact(tmsg)
 	if err != nil {
 		return nil, fmt.Errorf("transact: %w", err)
 	}
 	switch rmsg := rmsg.(type) {
 	case *lib9p.RRead:
 		return rmsg.Data, nil
 	case *lib9p.RError:
 		return nil, rmsg.Ename
 	default:
 		return nil, fmt.Errorf("invalid reply: %v", rmsg)
 	}
 }
 func (c *Client) Write(tag uint16, fid uint32, offset uint64, count uint32, data []byte) (uint32, error) {
 	tmsg := &lib9p.TWrite{Tag: tag, Fid: fid, Offset: offset, Count: count, Data: data}
 	rmsg, err := c.transact(tmsg)
 	if err != nil {
 		return 0, fmt.Errorf("transact: %w", err)
 	}
 	switch rmsg := rmsg.(type) {
 	case *lib9p.RWrite:
 		return rmsg.Count, nil
 	case *lib9p.RError:
 		return 0, rmsg.Ename
 	default:
 		return 0, fmt.Errorf("invalid reply: %v", rmsg)
 	}
 }
 func (c *Client) Clunk(tag uint16, fid uint32) error {
 	tmsg := &lib9p.TClunk{Tag: tag, Fid: fid}
 	rmsg, err := c.transact(tmsg)
 	if err != nil {
 		return fmt.Errorf("transact: %w", err)
 	}
 	switch rmsg := rmsg.(type) {
 	case *lib9p.RClunk:
 		return nil
 	case *lib9p.RError:
 		return rmsg.Ename
 	default:
 		return fmt.Errorf("invalid reply: %v", rmsg)
 	}
 }
 func (c *Client) Remove(tag uint16, fid uint32) error {
 	tmsg := &lib9p.TRemove{Tag: tag, Fid: fid}
 	rmsg, err := c.transact(tmsg)
 	if err != nil {
 		return fmt.Errorf("transact: %w", err)
 	}
 	switch rmsg := rmsg.(type) {
 	case *lib9p.RRemove:
 		return nil
 	case *lib9p.RError:
 		return rmsg.Ename
 	default:
 		return fmt.Errorf("invalid reply: %v", rmsg)
 	}
 }
 func (c *Client) Stat(tag uint16, fid uint32) (*lib9p.Stat, error) {
 	tmsg := &lib9p.TStat{Tag: tag, Fid: fid}
 	rmsg, err := c.transact(tmsg)
 	if err != nil {
 		return nil, fmt.Errorf("transact: %w", err)
 	}
 	switch rmsg := rmsg.(type) {
 	case *lib9p.RStat:
 		return rmsg.Stat, nil
 	case *lib9p.RError:
 		return nil, rmsg.Ename
 	default:
 		return nil, fmt.Errorf("invalid reply: %v", rmsg)
 	}
 }
 func (c *Client) Wstat(tag uint16, fid uint32, stat *lib9p.Stat) error {
 	tmsg := &lib9p.TWstat{Tag: tag, Fid: fid, Stat: stat}
 	rmsg, err := c.transact(tmsg)
 	if err != nil {
 		return fmt.Errorf("transact: %w", err)
 	}
 	switch rmsg := rmsg.(type) {
 	case *lib9p.RWstat:
 		return nil
 	case *lib9p.RError:
 		return rmsg.Ename
 	default:
 		return fmt.Errorf("invalid reply: %v", rmsg)
 	}
 }