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 (
        "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")
        }
}