src/pkg/old/netchan/import.go - go - 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 (
 	"errors"
 	"io"
 	"log"
 	"net"
 	"reflect"
 	"sync"
 	"time"
 )

 // Import

 // impLog is a logging convenience function.  The first argument must be a string.
 func impLog(args ...interface{}) {
 	args[0] = "netchan import: " + args[0].(string)
 	log.Print(args...)
 }

 // An Importer allows a set of channels to be imported from a single
 // remote machine/network port.  A machine may have multiple
 // importers, even from the same machine/network port.
 type Importer struct {
 	*encDec
 	chanLock sync.Mutex // protects access to channel map
 	names    map[string]*netChan
 	chans    map[int]*netChan
 	errors   chan error
 	maxId    int
 	mu       sync.Mutex // protects remaining fields
 	unacked  int64      // number of unacknowledged sends.
 	seqLock  sync.Mutex // guarantees messages are in sequence, only locked under mu
 }

 // NewImporter creates a new Importer object to import a set of channels
 // from the given connection. The Exporter must be available and serving when
 // the Importer is created.
 func NewImporter(conn io.ReadWriter) *Importer {
 	imp := new(Importer)
 	imp.encDec = newEncDec(conn)
 	imp.chans = make(map[int]*netChan)
 	imp.names = make(map[string]*netChan)
 	imp.errors = make(chan error, 10)
 	imp.unacked = 0
 	go imp.run()
 	return imp
 }

 // Import imports a set of channels from the given network and address.
 func Import(network, remoteaddr string) (*Importer, error) {
 	conn, err := net.Dial(network, remoteaddr)
 	if err != nil {
 		return nil, err
 	}
 	return NewImporter(conn), nil
 }

 // shutdown closes all channels for which we are receiving data from the remote side.
 func (imp *Importer) shutdown() {
 	imp.chanLock.Lock()
 	for _, ich := range imp.chans {
 		if ich.dir == Recv {
 			ich.close()
 		}
 	}
 	imp.chanLock.Unlock()
 }

 // Handle the data from a single imported data stream, which will
 // have the form
 //	(response, data)*
 // The response identifies by name which channel is transmitting data.
 func (imp *Importer) run() {
 	// Loop on responses; requests are sent by ImportNValues()
 	hdr := new(header)
 	hdrValue := reflect.ValueOf(hdr)
 	ackHdr := new(header)
 	err := new(error_)
 	errValue := reflect.ValueOf(err)
 	for {
 		*hdr = header{}
 		if e := imp.decode(hdrValue); e != nil {
 			if e != io.EOF {
 				impLog("header:", e)
 				imp.shutdown()
 			}
 			return
 		}
 		switch hdr.PayloadType {
 		case payData:
 			// done lower in loop
 		case payError:
 			if e := imp.decode(errValue); e != nil {
 				impLog("error:", e)
 				return
 			}
 			if err.Error != "" {
 				impLog("response error:", err.Error)
 				select {
 				case imp.errors <- errors.New(err.Error):
 					continue // errors are not acknowledged
 				default:
 					imp.shutdown()
 					return
 				}
 			}
 		case payClosed:
 			nch := imp.getChan(hdr.Id, false)
 			if nch != nil {
 				nch.close()
 			}
 			continue // closes are not acknowledged.
 		case payAckSend:
 			// we can receive spurious acks if the channel is
 			// hung up, so we ask getChan to ignore any errors.
 			nch := imp.getChan(hdr.Id, true)
 			if nch != nil {
 				nch.acked()
 				imp.mu.Lock()
 				imp.unacked--
 				imp.mu.Unlock()
 			}
 			continue
 		default:
 			impLog("unexpected payload type:", hdr.PayloadType)
 			return
 		}
 		nch := imp.getChan(hdr.Id, false)
 		if nch == nil {
 			continue
 		}
 		if nch.dir != Recv {
 			impLog("cannot happen: receive from non-Recv channel")
 			return
 		}
 		// Acknowledge receipt
 		ackHdr.Id = hdr.Id
 		ackHdr.SeqNum = hdr.SeqNum
 		imp.encode(ackHdr, payAck, nil)
 		// Create a new value for each received item.
 		value := reflect.New(nch.ch.Type().Elem()).Elem()
 		if e := imp.decode(value); e != nil {
 			impLog("importer value decode:", e)
 			return
 		}
 		nch.send(value)
 	}
 }

 func (imp *Importer) getChan(id int, errOk bool) *netChan {
 	imp.chanLock.Lock()
 	ich := imp.chans[id]
 	imp.chanLock.Unlock()
 	if ich == nil {
 		if !errOk {
 			impLog("unknown id in netchan request: ", id)
 		}
 		return nil
 	}
 	return ich
 }

 // Errors returns a channel from which transmission and protocol errors
 // can be read. Clients of the importer are not required to read the error
 // channel for correct execution. However, if too many errors occur
 // without being read from the error channel, the importer will shut down.
 func (imp *Importer) Errors() chan error {
 	return imp.errors
 }

 // Import imports a channel of the given type, size and specified direction.
 // It is equivalent to ImportNValues with a count of -1, meaning unbounded.
 func (imp *Importer) Import(name string, chT interface{}, dir Dir, size int) error {
 	return imp.ImportNValues(name, chT, dir, size, -1)
 }

 // ImportNValues imports a channel of the given type and specified
 // direction and then receives or transmits up to n values on that
 // channel.  A value of n==-1 implies an unbounded number of values.  The
 // channel will have buffer space for size values, or 1 value if size < 1.
 // The channel to be bound to the remote site's channel is provided
 // in the call and may be of arbitrary channel type.
 // Despite the literal signature, the effective signature is
 //	ImportNValues(name string, chT chan T, dir Dir, size, n int) error
 // Example usage:
 //	imp, err := NewImporter("tcp", "netchanserver.mydomain.com:1234")
 //	if err != nil { log.Fatal(err) }
 //	ch := make(chan myType)
 //	err = imp.ImportNValues("name", ch, Recv, 1, 1)
 //	if err != nil { log.Fatal(err) }
 //	fmt.Printf("%+v\n", <-ch)
 func (imp *Importer) ImportNValues(name string, chT interface{}, dir Dir, size, n int) error {
 	ch, err := checkChan(chT, dir)
 	if err != nil {
 		return err
 	}
 	imp.chanLock.Lock()
 	defer imp.chanLock.Unlock()
 	_, present := imp.names[name]
 	if present {
 		return errors.New("channel name already being imported:" + name)
 	}
 	if size < 1 {
 		size = 1
 	}
 	id := imp.maxId
 	imp.maxId++
 	nch := newNetChan(name, id, &chanDir{ch, dir}, imp.encDec, size, int64(n))
 	imp.names[name] = nch
 	imp.chans[id] = nch
 	// Tell the other side about this channel.
 	hdr := &header{Id: id}
 	req := &request{Name: name, Count: int64(n), Dir: dir, Size: size}
 	if err = imp.encode(hdr, payRequest, req); err != nil {
 		impLog("request encode:", err)
 		return err
 	}
 	if dir == Send {
 		go func() {
 			for i := 0; n == -1 || i < n; i++ {
 				val, ok := nch.recv()
 				if !ok {
 					if err = imp.encode(hdr, payClosed, nil); err != nil {
 						impLog("error encoding client closed message:", err)
 					}
 					return
 				}
 				// We hold the lock during transmission to guarantee messages are
 				// sent in order.
 				imp.mu.Lock()
 				imp.unacked++
 				imp.seqLock.Lock()
 				imp.mu.Unlock()
 				if err = imp.encode(hdr, payData, val.Interface()); err != nil {
 					impLog("error encoding client send:", err)
 					return
 				}
 				imp.seqLock.Unlock()
 			}
 		}()
 	}
 	return nil
 }

 // Hangup disassociates the named channel from the Importer and closes
 // the channel.  Messages in flight for the channel may be dropped.
 func (imp *Importer) Hangup(name string) error {
 	imp.chanLock.Lock()
 	defer imp.chanLock.Unlock()
 	nc := imp.names[name]
 	if nc == nil {
 		return errors.New("netchan import: hangup: no such channel: " + name)
 	}
 	delete(imp.names, name)
 	delete(imp.chans, nc.id)
 	nc.close()
 	return nil
 }

 func (imp *Importer) unackedCount() int64 {
 	imp.mu.Lock()
 	n := imp.unacked
 	imp.mu.Unlock()
 	return n
 }

 // Drain waits until all messages sent from this exporter/importer, including
 // those not yet sent to any server and possibly including those sent while
 // Drain was executing, have been received by the exporter.  In short, it
 // waits until all the importer's messages have been received.
 // If the timeout (measured in nanoseconds) is positive and Drain takes
 // longer than that to complete, an error is returned.
 func (imp *Importer) Drain(timeout int64) error {
 	deadline := time.Now().Add(time.Duration(timeout))
 	for imp.unackedCount() > 0 {
 		if timeout > 0 && time.Now().After(deadline) {
 			return errors.New("timeout")
 		}
 		time.Sleep(100 * 1e6)
 	}
 	return nil
 }
	// 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 (
	"errors"
	"io"
	"log"
	"net"
	"reflect"
	"sync"
	"time"
	)

	// Import

	// impLog is a logging convenience function. The first argument must be a string.
	func impLog(args ...interface{}) {
	args[0] = "netchan import: " + args[0].(string)
	log.Print(args...)
	}

	// An Importer allows a set of channels to be imported from a single
	// remote machine/network port. A machine may have multiple
	// importers, even from the same machine/network port.
	type Importer struct {
	*encDec
	chanLock sync.Mutex // protects access to channel map
	names map[string]*netChan
	chans map[int]*netChan
	errors chan error
	maxId int
	mu sync.Mutex // protects remaining fields
	unacked int64 // number of unacknowledged sends.
	seqLock sync.Mutex // guarantees messages are in sequence, only locked under mu
	}

	// NewImporter creates a new Importer object to import a set of channels
	// from the given connection. The Exporter must be available and serving when
	// the Importer is created.
	func NewImporter(conn io.ReadWriter) *Importer {
	imp := new(Importer)
	imp.encDec = newEncDec(conn)
	imp.chans = make(map[int]*netChan)
	imp.names = make(map[string]*netChan)
	imp.errors = make(chan error, 10)
	imp.unacked = 0
	go imp.run()
	return imp
	}

	// Import imports a set of channels from the given network and address.
	func Import(network, remoteaddr string) (*Importer, error) {
	conn, err := net.Dial(network, remoteaddr)
	if err != nil {
	return nil, err
	}
	return NewImporter(conn), nil
	}

	// shutdown closes all channels for which we are receiving data from the remote side.
	func (imp *Importer) shutdown() {
	imp.chanLock.Lock()
	for _, ich := range imp.chans {
	if ich.dir == Recv {
	ich.close()
	}
	}
	imp.chanLock.Unlock()
	}

	// Handle the data from a single imported data stream, which will
	// have the form
	// (response, data)*
	// The response identifies by name which channel is transmitting data.
	func (imp *Importer) run() {
	// Loop on responses; requests are sent by ImportNValues()
	hdr := new(header)
	hdrValue := reflect.ValueOf(hdr)
	ackHdr := new(header)
	err := new(error_)
	errValue := reflect.ValueOf(err)
	for {
	*hdr = header{}
	if e := imp.decode(hdrValue); e != nil {
	if e != io.EOF {
	impLog("header:", e)
	imp.shutdown()
	}
	return
	}
	switch hdr.PayloadType {
	case payData:
	// done lower in loop
	case payError:
	if e := imp.decode(errValue); e != nil {
	impLog("error:", e)
	return
	}
	if err.Error != "" {
	impLog("response error:", err.Error)
	select {
	case imp.errors <- errors.New(err.Error):
	continue // errors are not acknowledged
	default:
	imp.shutdown()
	return
	}
	}
	case payClosed:
	nch := imp.getChan(hdr.Id, false)
	if nch != nil {
	nch.close()
	}
	continue // closes are not acknowledged.
	case payAckSend:
	// we can receive spurious acks if the channel is
	// hung up, so we ask getChan to ignore any errors.
	nch := imp.getChan(hdr.Id, true)
	if nch != nil {
	nch.acked()
	imp.mu.Lock()
	imp.unacked--
	imp.mu.Unlock()
	}
	continue
	default:
	impLog("unexpected payload type:", hdr.PayloadType)
	return
	}
	nch := imp.getChan(hdr.Id, false)
	if nch == nil {
	continue
	}
	if nch.dir != Recv {
	impLog("cannot happen: receive from non-Recv channel")
	return
	}
	// Acknowledge receipt
	ackHdr.Id = hdr.Id
	ackHdr.SeqNum = hdr.SeqNum
	imp.encode(ackHdr, payAck, nil)
	// Create a new value for each received item.
	value := reflect.New(nch.ch.Type().Elem()).Elem()
	if e := imp.decode(value); e != nil {
	impLog("importer value decode:", e)
	return
	}
	nch.send(value)
	}
	}

	func (imp Importer) getChan(id int, errOk bool) netChan {
	imp.chanLock.Lock()
	ich := imp.chans[id]
	imp.chanLock.Unlock()
	if ich == nil {
	if !errOk {
	impLog("unknown id in netchan request: ", id)
	}
	return nil
	}
	return ich
	}

	// Errors returns a channel from which transmission and protocol errors
	// can be read. Clients of the importer are not required to read the error
	// channel for correct execution. However, if too many errors occur
	// without being read from the error channel, the importer will shut down.
	func (imp *Importer) Errors() chan error {
	return imp.errors
	}

	// Import imports a channel of the given type, size and specified direction.
	// It is equivalent to ImportNValues with a count of -1, meaning unbounded.
	func (imp *Importer) Import(name string, chT interface{}, dir Dir, size int) error {
	return imp.ImportNValues(name, chT, dir, size, -1)
	}

	// ImportNValues imports a channel of the given type and specified
	// direction and then receives or transmits up to n values on that
	// channel. A value of n==-1 implies an unbounded number of values. The
	// channel will have buffer space for size values, or 1 value if size < 1.
	// The channel to be bound to the remote site's channel is provided
	// in the call and may be of arbitrary channel type.
	// Despite the literal signature, the effective signature is
	// ImportNValues(name string, chT chan T, dir Dir, size, n int) error
	// Example usage:
	// imp, err := NewImporter("tcp", "netchanserver.mydomain.com:1234")
	// if err != nil { log.Fatal(err) }
	// ch := make(chan myType)
	// err = imp.ImportNValues("name", ch, Recv, 1, 1)
	// if err != nil { log.Fatal(err) }
	// fmt.Printf("%+v\n", <-ch)
	func (imp *Importer) ImportNValues(name string, chT interface{}, dir Dir, size, n int) error {
	ch, err := checkChan(chT, dir)
	if err != nil {
	return err
	}
	imp.chanLock.Lock()
	defer imp.chanLock.Unlock()
	_, present := imp.names[name]
	if present {
	return errors.New("channel name already being imported:" + name)
	}
	if size < 1 {
	size = 1
	}
	id := imp.maxId
	imp.maxId++
	nch := newNetChan(name, id, &chanDir{ch, dir}, imp.encDec, size, int64(n))
	imp.names[name] = nch
	imp.chans[id] = nch
	// Tell the other side about this channel.
	hdr := &header{Id: id}
	req := &request{Name: name, Count: int64(n), Dir: dir, Size: size}
	if err = imp.encode(hdr, payRequest, req); err != nil {
	impLog("request encode:", err)
	return err
	}
	if dir == Send {
	go func() {
	for i := 0; n == -1 \|\| i < n; i++ {
	val, ok := nch.recv()
	if !ok {
	if err = imp.encode(hdr, payClosed, nil); err != nil {
	impLog("error encoding client closed message:", err)
	}
	return
	}
	// We hold the lock during transmission to guarantee messages are
	// sent in order.
	imp.mu.Lock()
	imp.unacked++
	imp.seqLock.Lock()
	imp.mu.Unlock()
	if err = imp.encode(hdr, payData, val.Interface()); err != nil {
	impLog("error encoding client send:", err)
	return
	}
	imp.seqLock.Unlock()
	}
	}()
	}
	return nil
	}

	// Hangup disassociates the named channel from the Importer and closes
	// the channel. Messages in flight for the channel may be dropped.
	func (imp *Importer) Hangup(name string) error {
	imp.chanLock.Lock()
	defer imp.chanLock.Unlock()
	nc := imp.names[name]
	if nc == nil {
	return errors.New("netchan import: hangup: no such channel: " + name)
	}
	delete(imp.names, name)
	delete(imp.chans, nc.id)
	nc.close()
	return nil
	}

	func (imp *Importer) unackedCount() int64 {
	imp.mu.Lock()
	n := imp.unacked
	imp.mu.Unlock()
	return n
	}

	// Drain waits until all messages sent from this exporter/importer, including
	// those not yet sent to any server and possibly including those sent while
	// Drain was executing, have been received by the exporter. In short, it
	// waits until all the importer's messages have been received.
	// If the timeout (measured in nanoseconds) is positive and Drain takes
	// longer than that to complete, an error is returned.
	func (imp *Importer) Drain(timeout int64) error {
	deadline := time.Now().Add(time.Duration(timeout))
	for imp.unackedCount() > 0 {
	if timeout > 0 && time.Now().After(deadline) {
	return errors.New("timeout")
	}
	time.Sleep(100 * 1e6)
	}
	return nil
	}