Commit d4384ff7 authored by Rob Pike's avatar Rob Pike

netchan: use gob DecodeValue to eliminate the need for a pointer value

in Import and Export.

R=rsc
CC=golang-dev
https://golang.org/cl/1707047
parent 12a4d843
...@@ -8,6 +8,7 @@ import ( ...@@ -8,6 +8,7 @@ import (
"gob" "gob"
"net" "net"
"os" "os"
"reflect"
"sync" "sync"
) )
...@@ -62,9 +63,9 @@ func newEncDec(conn net.Conn) *encDec { ...@@ -62,9 +63,9 @@ func newEncDec(conn net.Conn) *encDec {
} }
// Decode an item from the connection. // Decode an item from the connection.
func (ed *encDec) decode(e interface{}) os.Error { func (ed *encDec) decode(value reflect.Value) os.Error {
ed.decLock.Lock() ed.decLock.Lock()
err := ed.dec.Decode(e) err := ed.dec.DecodeValue(value)
if err != nil { if err != nil {
// TODO: tear down connection? // TODO: tear down connection?
} }
......
...@@ -36,7 +36,6 @@ import ( ...@@ -36,7 +36,6 @@ import (
type exportChan struct { type exportChan struct {
ch *reflect.ChanValue ch *reflect.ChanValue
dir Dir dir Dir
ptr *reflect.PtrValue // a pointer value we can point at each new received item
} }
// An Exporter allows a set of channels to be published on a single // An Exporter allows a set of channels to be published on a single
...@@ -101,17 +100,19 @@ func (client *expClient) getChan(hdr *header, dir Dir) *exportChan { ...@@ -101,17 +100,19 @@ func (client *expClient) getChan(hdr *header, dir Dir) *exportChan {
// while (client Send) requests are handled as data arrives from the client. // while (client Send) requests are handled as data arrives from the client.
func (client *expClient) run() { func (client *expClient) run() {
hdr := new(header) hdr := new(header)
hdrValue := reflect.NewValue(hdr)
req := new(request) req := new(request)
reqValue := reflect.NewValue(req)
error := new(error) error := new(error)
for { for {
if err := client.decode(hdr); err != nil { if err := client.decode(hdrValue); err != nil {
log.Stderr("error decoding client header:", err) log.Stderr("error decoding client header:", err)
// TODO: tear down connection // TODO: tear down connection
return return
} }
switch hdr.payloadType { switch hdr.payloadType {
case payRequest: case payRequest:
if err := client.decode(req); err != nil { if err := client.decode(reqValue); err != nil {
log.Stderr("error decoding client request:", err) log.Stderr("error decoding client request:", err)
// TODO: tear down connection // TODO: tear down connection
return return
...@@ -169,9 +170,8 @@ func (client *expClient) serveSend(hdr header) { ...@@ -169,9 +170,8 @@ func (client *expClient) serveSend(hdr header) {
return return
} }
// Create a new value for each received item. // Create a new value for each received item.
val := reflect.MakeZero(ech.ptr.Type().(*reflect.PtrType).Elem()) val := reflect.MakeZero(ech.ch.Type().(*reflect.ChanType).Elem())
ech.ptr.PointTo(val) if err := client.decode(val); err != nil {
if err := client.decode(ech.ptr.Interface()); err != nil {
log.Stderr("exporter value decode:", err) log.Stderr("exporter value decode:", err)
return return
} }
...@@ -224,9 +224,7 @@ func checkChan(chT interface{}, dir Dir) (*reflect.ChanValue, os.Error) { ...@@ -224,9 +224,7 @@ func checkChan(chT interface{}, dir Dir) (*reflect.ChanValue, os.Error) {
// channel type. // channel type.
// Despite the literal signature, the effective signature is // Despite the literal signature, the effective signature is
// Export(name string, chT chan T, dir Dir) // Export(name string, chT chan T, dir Dir)
// where T must be a struct, pointer to struct, etc. func (exp *Exporter) Export(name string, chT interface{}, dir Dir) os.Error {
// TODO: fix reflection so we can eliminate the need for pT.
func (exp *Exporter) Export(name string, chT interface{}, dir Dir, pT interface{}) os.Error {
ch, err := checkChan(chT, dir) ch, err := checkChan(chT, dir)
if err != nil { if err != nil {
return err return err
...@@ -237,7 +235,6 @@ func (exp *Exporter) Export(name string, chT interface{}, dir Dir, pT interface{ ...@@ -237,7 +235,6 @@ func (exp *Exporter) Export(name string, chT interface{}, dir Dir, pT interface{
if present { if present {
return os.ErrorString("channel name already being exported:" + name) return os.ErrorString("channel name already being exported:" + name)
} }
ptr := reflect.MakeZero(reflect.Typeof(pT)).(*reflect.PtrValue) exp.chans[name] = &exportChan{ch, dir}
exp.chans[name] = &exportChan{ch, dir, ptr}
return nil return nil
} }
...@@ -19,7 +19,6 @@ import ( ...@@ -19,7 +19,6 @@ import (
type importChan struct { type importChan struct {
ch *reflect.ChanValue ch *reflect.ChanValue
dir Dir dir Dir
ptr *reflect.PtrValue // a pointer value we can point at each new received item
} }
// An Importer allows a set of channels to be imported from a single // An Importer allows a set of channels to be imported from a single
...@@ -67,9 +66,11 @@ func (imp *Importer) shutdown() { ...@@ -67,9 +66,11 @@ func (imp *Importer) shutdown() {
func (imp *Importer) run() { func (imp *Importer) run() {
// Loop on responses; requests are sent by ImportNValues() // Loop on responses; requests are sent by ImportNValues()
hdr := new(header) hdr := new(header)
hdrValue := reflect.NewValue(hdr)
err := new(error) err := new(error)
errValue := reflect.NewValue(err)
for { for {
if e := imp.decode(hdr); e != nil { if e := imp.decode(hdrValue); e != nil {
log.Stderr("importer header:", e) log.Stderr("importer header:", e)
imp.shutdown() imp.shutdown()
return return
...@@ -78,7 +79,7 @@ func (imp *Importer) run() { ...@@ -78,7 +79,7 @@ func (imp *Importer) run() {
case payData: case payData:
// done lower in loop // done lower in loop
case payError: case payError:
if e := imp.decode(err); e != nil { if e := imp.decode(errValue); e != nil {
log.Stderr("importer error:", e) log.Stderr("importer error:", e)
return return
} }
...@@ -103,20 +104,19 @@ func (imp *Importer) run() { ...@@ -103,20 +104,19 @@ func (imp *Importer) run() {
return return
} }
// Create a new value for each received item. // Create a new value for each received item.
val := reflect.MakeZero(ich.ptr.Type().(*reflect.PtrType).Elem()) value := reflect.MakeZero(ich.ch.Type().(*reflect.ChanType).Elem())
ich.ptr.PointTo(val) if e := imp.decode(value); e != nil {
if e := imp.decode(ich.ptr.Interface()); e != nil {
log.Stderr("importer value decode:", e) log.Stderr("importer value decode:", e)
return return
} }
ich.ch.Send(val) ich.ch.Send(value)
} }
} }
// Import imports a channel of the given type and specified direction. // Import imports a channel of the given type and specified direction.
// It is equivalent to ImportNValues with a count of 0, meaning unbounded. // It is equivalent to ImportNValues with a count of 0, meaning unbounded.
func (imp *Importer) Import(name string, chT interface{}, dir Dir, pT interface{}) os.Error { func (imp *Importer) Import(name string, chT interface{}, dir Dir) os.Error {
return imp.ImportNValues(name, chT, dir, pT, 0) return imp.ImportNValues(name, chT, dir, 0)
} }
// ImportNValues imports a channel of the given type and specified direction // ImportNValues imports a channel of the given type and specified direction
...@@ -125,36 +125,26 @@ func (imp *Importer) Import(name string, chT interface{}, dir Dir, pT interface{ ...@@ -125,36 +125,26 @@ func (imp *Importer) Import(name string, chT interface{}, dir Dir, pT interface{
// the remote site's channel is provided in the call and may be of arbitrary // the remote site's channel is provided in the call and may be of arbitrary
// channel type. // channel type.
// Despite the literal signature, the effective signature is // Despite the literal signature, the effective signature is
// ImportNValues(name string, chT chan T, dir Dir, pT T, n int) os.Error // ImportNValues(name string, chT chan T, dir Dir, n int) os.Error
// where T must be a struct, pointer to struct, etc. pT may be more indirect
// than the value type of the channel (e.g. chan T, pT *T) but it must be a
// pointer.
// Example usage: // Example usage:
// imp, err := NewImporter("tcp", "netchanserver.mydomain.com:1234") // imp, err := NewImporter("tcp", "netchanserver.mydomain.com:1234")
// if err != nil { log.Exit(err) } // if err != nil { log.Exit(err) }
// ch := make(chan myType) // ch := make(chan myType)
// err := imp.ImportNValues("name", ch, Recv, new(myType), 1) // err := imp.ImportNValues("name", ch, Recv, 1)
// if err != nil { log.Exit(err) } // if err != nil { log.Exit(err) }
// fmt.Printf("%+v\n", <-ch) // fmt.Printf("%+v\n", <-ch)
// TODO: fix reflection so we can eliminate the need for pT. func (imp *Importer) ImportNValues(name string, chT interface{}, dir Dir, n int) os.Error {
func (imp *Importer) ImportNValues(name string, chT interface{}, dir Dir, pT interface{}, n int) os.Error {
ch, err := checkChan(chT, dir) ch, err := checkChan(chT, dir)
if err != nil { if err != nil {
return err return err
} }
// Make sure pT is a pointer (to a pointer...) to a struct.
rt := reflect.Typeof(pT)
if _, ok := rt.(*reflect.PtrType); !ok {
return os.ErrorString("not a pointer:" + rt.String())
}
imp.chanLock.Lock() imp.chanLock.Lock()
defer imp.chanLock.Unlock() defer imp.chanLock.Unlock()
_, present := imp.chans[name] _, present := imp.chans[name]
if present { if present {
return os.ErrorString("channel name already being imported:" + name) return os.ErrorString("channel name already being imported:" + name)
} }
ptr := reflect.MakeZero(reflect.Typeof(pT)).(*reflect.PtrValue) imp.chans[name] = &importChan{ch, dir}
imp.chans[name] = &importChan{ch, dir, ptr}
// Tell the other side about this channel. // Tell the other side about this channel.
hdr := new(header) hdr := new(header)
hdr.name = name hdr.name = name
......
...@@ -11,7 +11,7 @@ const closeCount = 5 // number of items when sender closes early ...@@ -11,7 +11,7 @@ const closeCount = 5 // number of items when sender closes early
func exportSend(exp *Exporter, n int, t *testing.T) { func exportSend(exp *Exporter, n int, t *testing.T) {
ch := make(chan int) ch := make(chan int)
err := exp.Export("exportedSend", ch, Send, new(int)) err := exp.Export("exportedSend", ch, Send)
if err != nil { if err != nil {
t.Fatal("exportSend:", err) t.Fatal("exportSend:", err)
} }
...@@ -23,7 +23,7 @@ func exportSend(exp *Exporter, n int, t *testing.T) { ...@@ -23,7 +23,7 @@ func exportSend(exp *Exporter, n int, t *testing.T) {
func exportReceive(exp *Exporter, t *testing.T) { func exportReceive(exp *Exporter, t *testing.T) {
ch := make(chan int) ch := make(chan int)
err := exp.Export("exportedRecv", ch, Recv, new(int)) err := exp.Export("exportedRecv", ch, Recv)
if err != nil { if err != nil {
t.Fatal("exportReceive:", err) t.Fatal("exportReceive:", err)
} }
...@@ -37,7 +37,7 @@ func exportReceive(exp *Exporter, t *testing.T) { ...@@ -37,7 +37,7 @@ func exportReceive(exp *Exporter, t *testing.T) {
func importReceive(imp *Importer, t *testing.T) { func importReceive(imp *Importer, t *testing.T) {
ch := make(chan int) ch := make(chan int)
err := imp.ImportNValues("exportedSend", ch, Recv, new(int), count) err := imp.ImportNValues("exportedSend", ch, Recv, count)
if err != nil { if err != nil {
t.Fatal("importReceive:", err) t.Fatal("importReceive:", err)
} }
...@@ -57,7 +57,7 @@ func importReceive(imp *Importer, t *testing.T) { ...@@ -57,7 +57,7 @@ func importReceive(imp *Importer, t *testing.T) {
func importSend(imp *Importer, t *testing.T) { func importSend(imp *Importer, t *testing.T) {
ch := make(chan int) ch := make(chan int)
err := imp.ImportNValues("exportedRecv", ch, Send, new(int), count) err := imp.ImportNValues("exportedRecv", ch, Send, count)
if err != nil { if err != nil {
t.Fatal("importSend:", err) t.Fatal("importSend:", err)
} }
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment