Subversion Repositories openrisc

// 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 * time.Millisecond)

        }

        return nil

}