old/netchan/common.go - exp - Git at Google

 // Copyright 2010 The Go Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.

 package netchan

 import (
 	"encoding/gob"
 	"errors"
 	"io"
 	"reflect"
 	"sync"
 	"time"
 )

 // The direction of a connection from the client's perspective.
 type Dir int

 const (
 	Recv Dir = iota
 	Send
 )

 func (dir Dir) String() string {
 	switch dir {
 	case Recv:
 		return "Recv"
 	case Send:
 		return "Send"
 	}
 	return "???"
 }

 // Payload types
 const (
 	payRequest = iota // request structure follows
 	payError          // error structure follows
 	payData           // user payload follows
 	payAck            // acknowledgement; no payload
 	payClosed         // channel is now closed
 	payAckSend        // payload has been delivered.
 )

 // A header is sent as a prefix to every transmission.  It will be followed by
 // a request structure, an error structure, or an arbitrary user payload structure.
 type header struct {
 	Id          int
 	PayloadType int
 	SeqNum      int64
 }

 // Sent with a header once per channel from importer to exporter to report
 // that it wants to bind to a channel with the specified direction for count
 // messages, with space for size buffered values. If count is -1, it means unlimited.
 type request struct {
 	Name  string
 	Count int64
 	Size  int
 	Dir   Dir
 }

 // Sent with a header to report an error.
 type error_ struct {
 	Error string
 }

 // Used to unify management of acknowledgements for import and export.
 type unackedCounter interface {
 	unackedCount() int64
 	ack() int64
 	seq() int64
 }

 // A channel and its direction.
 type chanDir struct {
 	ch  reflect.Value
 	dir Dir
 }

 // clientSet contains the objects and methods needed for tracking
 // clients of an exporter and draining outstanding messages.
 type clientSet struct {
 	mu      sync.Mutex // protects access to channel and client maps
 	names   map[string]*chanDir
 	clients map[unackedCounter]bool
 }

 // Mutex-protected encoder and decoder pair.
 type encDec struct {
 	decLock sync.Mutex
 	dec     *gob.Decoder
 	encLock sync.Mutex
 	enc     *gob.Encoder
 }

 func newEncDec(conn io.ReadWriter) *encDec {
 	return &encDec{
 		dec: gob.NewDecoder(conn),
 		enc: gob.NewEncoder(conn),
 	}
 }

 // Decode an item from the connection.
 func (ed *encDec) decode(value reflect.Value) error {
 	ed.decLock.Lock()
 	err := ed.dec.DecodeValue(value)
 	if err != nil {
 		// TODO: tear down connection?
 	}
 	ed.decLock.Unlock()
 	return err
 }

 // Encode a header and payload onto the connection.
 func (ed *encDec) encode(hdr *header, payloadType int, payload interface{}) error {
 	ed.encLock.Lock()
 	hdr.PayloadType = payloadType
 	err := ed.enc.Encode(hdr)
 	if err == nil {
 		if payload != nil {
 			err = ed.enc.Encode(payload)
 		}
 	}
 	if err != nil {
 		// TODO: tear down connection if there is an error?
 	}
 	ed.encLock.Unlock()
 	return err
 }

 // See the comment for Exporter.Drain.
 func (cs *clientSet) drain(timeout time.Duration) error {
 	deadline := time.Now().Add(timeout)
 	for {
 		pending := false
 		cs.mu.Lock()
 		// Any messages waiting for a client?
 		for _, chDir := range cs.names {
 			if chDir.ch.Len() > 0 {
 				pending = true
 			}
 		}
 		// Any unacknowledged messages?
 		for client := range cs.clients {
 			n := client.unackedCount()
 			if n > 0 { // Check for > rather than != just to be safe.
 				pending = true
 				break
 			}
 		}
 		cs.mu.Unlock()
 		if !pending {
 			break
 		}
 		if timeout > 0 && time.Now().After(deadline) {
 			return errors.New("timeout")
 		}
 		time.Sleep(100 * time.Millisecond)
 	}
 	return nil
 }

 // See the comment for Exporter.Sync.
 func (cs *clientSet) sync(timeout time.Duration) error {
 	deadline := time.Now().Add(timeout)
 	// seq remembers the clients and their seqNum at point of entry.
 	seq := make(map[unackedCounter]int64)
 	cs.mu.Lock()
 	for client := range cs.clients {
 		seq[client] = client.seq()
 	}
 	cs.mu.Unlock()
 	for {
 		pending := false
 		cs.mu.Lock()
 		// Any unacknowledged messages?  Look only at clients that existed
 		// when we started and are still in this client set.
 		for client := range seq {
 			if _, ok := cs.clients[client]; ok {
 				if client.ack() < seq[client] {
 					pending = true
 					break
 				}
 			}
 		}
 		cs.mu.Unlock()
 		if !pending {
 			break
 		}
 		if timeout > 0 && time.Now().After(deadline) {
 			return errors.New("timeout")
 		}
 		time.Sleep(100 * time.Millisecond)
 	}
 	return nil
 }

 // A netChan represents a channel imported or exported
 // on a single connection. Flow is controlled by the receiving
 // side by sending payAckSend messages when values
 // are delivered into the local channel.
 type netChan struct {
 	*chanDir
 	name   string
 	id     int
 	size   int // buffer size of channel.
 	closed bool

 	// sender-specific state
 	ackCh chan bool // buffered with space for all the acks we need
 	space int       // available space.

 	// receiver-specific state
 	sendCh chan reflect.Value // buffered channel of values received from other end.
 	ed     *encDec            // so that we can send acks.
 	count  int64              // number of values still to receive.
 }

 // Create a new netChan with the given name (only used for
 // messages), id, direction, buffer size, and count.
 // The connection to the other side is represented by ed.
 func newNetChan(name string, id int, ch *chanDir, ed *encDec, size int, count int64) *netChan {
 	c := &netChan{chanDir: ch, name: name, id: id, size: size, ed: ed, count: count}
 	if c.dir == Send {
 		c.ackCh = make(chan bool, size)
 		c.space = size
 	}
 	return c
 }

 // Close the channel.
 func (nch *netChan) close() {
 	if nch.closed {
 		return
 	}
 	if nch.dir == Recv {
 		if nch.sendCh != nil {
 			// If the sender goroutine is active, close the channel to it.
 			// It will close nch.ch when it can.
 			close(nch.sendCh)
 		} else {
 			nch.ch.Close()
 		}
 	} else {
 		nch.ch.Close()
 		close(nch.ackCh)
 	}
 	nch.closed = true
 }

 // Send message from remote side to local receiver.
 func (nch *netChan) send(val reflect.Value) {
 	if nch.dir != Recv {
 		panic("send on wrong direction of channel")
 	}
 	if nch.sendCh == nil {
 		// If possible, do local send directly and ack immediately.
 		if nch.ch.TrySend(val) {
 			nch.sendAck()
 			return
 		}
 		// Start sender goroutine to manage delayed delivery of values.
 		nch.sendCh = make(chan reflect.Value, nch.size)
 		go nch.sender()
 	}
 	select {
 	case nch.sendCh <- val:
 		// ok
 	default:
 		// TODO: should this be more resilient?
 		panic("netchan: remote sender sent more values than allowed")
 	}
 }

 // sendAck sends an acknowledgment that a message has left
 // the channel's buffer. If the messages remaining to be sent
 // will fit in the channel's buffer, then we don't
 // need to send an ack.
 func (nch *netChan) sendAck() {
 	if nch.count < 0 || nch.count > int64(nch.size) {
 		nch.ed.encode(&header{Id: nch.id}, payAckSend, nil)
 	}
 	if nch.count > 0 {
 		nch.count--
 	}
 }

 // The sender process forwards items from the sending queue
 // to the destination channel, acknowledging each item.
 func (nch *netChan) sender() {
 	if nch.dir != Recv {
 		panic("sender on wrong direction of channel")
 	}
 	// When Exporter.Hangup is called, the underlying channel is closed,
 	// and so we may get a "too many operations on closed channel" error
 	// if there are outstanding messages in sendCh.
 	// Make sure that this doesn't panic the whole program.
 	defer func() {
 		if r := recover(); r != nil {
 			// TODO check that r is "too many operations", otherwise re-panic.
 		}
 	}()
 	for v := range nch.sendCh {
 		nch.ch.Send(v)
 		nch.sendAck()
 	}
 	nch.ch.Close()
 }

 // Receive value from local side for sending to remote side.
 func (nch *netChan) recv() (val reflect.Value, ok bool) {
 	if nch.dir != Send {
 		panic("recv on wrong direction of channel")
 	}

 	if nch.space == 0 {
 		// Wait for buffer space.
 		<-nch.ackCh
 		nch.space++
 	}
 	nch.space--
 	return nch.ch.Recv()
 }

 // acked is called when the remote side indicates that
 // a value has been delivered.
 func (nch *netChan) acked() {
 	if nch.dir != Send {
 		panic("recv on wrong direction of channel")
 	}
 	select {
 	case nch.ackCh <- true:
 		// ok
 	default:
 		// TODO: should this be more resilient?
 		panic("netchan: remote receiver sent too many acks")
 	}
 }
	// Copyright 2010 The Go Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style
	// license that can be found in the LICENSE file.

	package netchan

	import (
	"encoding/gob"
	"errors"
	"io"
	"reflect"
	"sync"
	"time"
	)

	// The direction of a connection from the client's perspective.
	type Dir int

	const (
	Recv Dir = iota
	Send
	)

	func (dir Dir) String() string {
	switch dir {
	case Recv:
	return "Recv"
	case Send:
	return "Send"
	}
	return "???"
	}

	// Payload types
	const (
	payRequest = iota // request structure follows
	payError // error structure follows
	payData // user payload follows
	payAck // acknowledgement; no payload
	payClosed // channel is now closed
	payAckSend // payload has been delivered.
	)

	// A header is sent as a prefix to every transmission. It will be followed by
	// a request structure, an error structure, or an arbitrary user payload structure.
	type header struct {
	Id int
	PayloadType int
	SeqNum int64
	}

	// Sent with a header once per channel from importer to exporter to report
	// that it wants to bind to a channel with the specified direction for count
	// messages, with space for size buffered values. If count is -1, it means unlimited.
	type request struct {
	Name string
	Count int64
	Size int
	Dir Dir
	}

	// Sent with a header to report an error.
	type error_ struct {
	Error string
	}

	// Used to unify management of acknowledgements for import and export.
	type unackedCounter interface {
	unackedCount() int64
	ack() int64
	seq() int64
	}

	// A channel and its direction.
	type chanDir struct {
	ch reflect.Value
	dir Dir
	}

	// clientSet contains the objects and methods needed for tracking
	// clients of an exporter and draining outstanding messages.
	type clientSet struct {
	mu sync.Mutex // protects access to channel and client maps
	names map[string]*chanDir
	clients map[unackedCounter]bool
	}

	// Mutex-protected encoder and decoder pair.
	type encDec struct {
	decLock sync.Mutex
	dec *gob.Decoder
	encLock sync.Mutex
	enc *gob.Encoder
	}

	func newEncDec(conn io.ReadWriter) *encDec {
	return &encDec{
	dec: gob.NewDecoder(conn),
	enc: gob.NewEncoder(conn),
	}
	}

	// Decode an item from the connection.
	func (ed *encDec) decode(value reflect.Value) error {
	ed.decLock.Lock()
	err := ed.dec.DecodeValue(value)
	if err != nil {
	// TODO: tear down connection?
	}
	ed.decLock.Unlock()
	return err
	}

	// Encode a header and payload onto the connection.
	func (ed encDec) encode(hdr header, payloadType int, payload interface{}) error {
	ed.encLock.Lock()
	hdr.PayloadType = payloadType
	err := ed.enc.Encode(hdr)
	if err == nil {
	if payload != nil {
	err = ed.enc.Encode(payload)
	}
	}
	if err != nil {
	// TODO: tear down connection if there is an error?
	}
	ed.encLock.Unlock()
	return err
	}

	// See the comment for Exporter.Drain.
	func (cs *clientSet) drain(timeout time.Duration) error {
	deadline := time.Now().Add(timeout)
	for {
	pending := false
	cs.mu.Lock()
	// Any messages waiting for a client?
	for _, chDir := range cs.names {
	if chDir.ch.Len() > 0 {
	pending = true
	}
	}
	// Any unacknowledged messages?
	for client := range cs.clients {
	n := client.unackedCount()
	if n > 0 { // Check for > rather than != just to be safe.
	pending = true
	break
	}
	}
	cs.mu.Unlock()
	if !pending {
	break
	}
	if timeout > 0 && time.Now().After(deadline) {
	return errors.New("timeout")
	}
	time.Sleep(100 * time.Millisecond)
	}
	return nil
	}

	// See the comment for Exporter.Sync.
	func (cs *clientSet) sync(timeout time.Duration) error {
	deadline := time.Now().Add(timeout)
	// seq remembers the clients and their seqNum at point of entry.
	seq := make(map[unackedCounter]int64)
	cs.mu.Lock()
	for client := range cs.clients {
	seq[client] = client.seq()
	}
	cs.mu.Unlock()
	for {
	pending := false
	cs.mu.Lock()
	// Any unacknowledged messages? Look only at clients that existed
	// when we started and are still in this client set.
	for client := range seq {
	if _, ok := cs.clients[client]; ok {
	if client.ack() < seq[client] {
	pending = true
	break
	}
	}
	}
	cs.mu.Unlock()
	if !pending {
	break
	}
	if timeout > 0 && time.Now().After(deadline) {
	return errors.New("timeout")
	}
	time.Sleep(100 * time.Millisecond)
	}
	return nil
	}

	// A netChan represents a channel imported or exported
	// on a single connection. Flow is controlled by the receiving
	// side by sending payAckSend messages when values
	// are delivered into the local channel.
	type netChan struct {
	*chanDir
	name string
	id int
	size int // buffer size of channel.
	closed bool

	// sender-specific state
	ackCh chan bool // buffered with space for all the acks we need
	space int // available space.

	// receiver-specific state
	sendCh chan reflect.Value // buffered channel of values received from other end.
	ed *encDec // so that we can send acks.
	count int64 // number of values still to receive.
	}

	// Create a new netChan with the given name (only used for
	// messages), id, direction, buffer size, and count.
	// The connection to the other side is represented by ed.
	func newNetChan(name string, id int, ch chanDir, ed encDec, size int, count int64) *netChan {
	c := &netChan{chanDir: ch, name: name, id: id, size: size, ed: ed, count: count}
	if c.dir == Send {
	c.ackCh = make(chan bool, size)
	c.space = size
	}
	return c
	}

	// Close the channel.
	func (nch *netChan) close() {
	if nch.closed {
	return
	}
	if nch.dir == Recv {
	if nch.sendCh != nil {
	// If the sender goroutine is active, close the channel to it.
	// It will close nch.ch when it can.
	close(nch.sendCh)
	} else {
	nch.ch.Close()
	}
	} else {
	nch.ch.Close()
	close(nch.ackCh)
	}
	nch.closed = true
	}

	// Send message from remote side to local receiver.
	func (nch *netChan) send(val reflect.Value) {
	if nch.dir != Recv {
	panic("send on wrong direction of channel")
	}
	if nch.sendCh == nil {
	// If possible, do local send directly and ack immediately.
	if nch.ch.TrySend(val) {
	nch.sendAck()
	return
	}
	// Start sender goroutine to manage delayed delivery of values.
	nch.sendCh = make(chan reflect.Value, nch.size)
	go nch.sender()
	}
	select {
	case nch.sendCh <- val:
	// ok
	default:
	// TODO: should this be more resilient?
	panic("netchan: remote sender sent more values than allowed")
	}
	}

	// sendAck sends an acknowledgment that a message has left
	// the channel's buffer. If the messages remaining to be sent
	// will fit in the channel's buffer, then we don't
	// need to send an ack.
	func (nch *netChan) sendAck() {
	if nch.count < 0 \|\| nch.count > int64(nch.size) {
	nch.ed.encode(&header{Id: nch.id}, payAckSend, nil)
	}
	if nch.count > 0 {
	nch.count--
	}
	}

	// The sender process forwards items from the sending queue
	// to the destination channel, acknowledging each item.
	func (nch *netChan) sender() {
	if nch.dir != Recv {
	panic("sender on wrong direction of channel")
	}
	// When Exporter.Hangup is called, the underlying channel is closed,
	// and so we may get a "too many operations on closed channel" error
	// if there are outstanding messages in sendCh.
	// Make sure that this doesn't panic the whole program.
	defer func() {
	if r := recover(); r != nil {
	// TODO check that r is "too many operations", otherwise re-panic.
	}
	}()
	for v := range nch.sendCh {
	nch.ch.Send(v)
	nch.sendAck()
	}
	nch.ch.Close()
	}

	// Receive value from local side for sending to remote side.
	func (nch *netChan) recv() (val reflect.Value, ok bool) {
	if nch.dir != Send {
	panic("recv on wrong direction of channel")
	}

	if nch.space == 0 {
	// Wait for buffer space.
	<-nch.ackCh
	nch.space++
	}
	nch.space--
	return nch.ch.Recv()
	}

	// acked is called when the remote side indicates that
	// a value has been delivered.
	func (nch *netChan) acked() {
	if nch.dir != Send {
	panic("recv on wrong direction of channel")
	}
	select {
	case nch.ackCh <- true:
	// ok
	default:
	// TODO: should this be more resilient?
	panic("netchan: remote receiver sent too many acks")
	}
	}