replace non-blocking send, receive syntax with select
R=golang-dev, nigeltao, niemeyer, r
CC=golang-dev
https://golang.org/cl/4079053
diff --git a/src/pkg/compress/flate/deflate_test.go b/src/pkg/compress/flate/deflate_test.go
index 3db9556..68dcd7b 100644
--- a/src/pkg/compress/flate/deflate_test.go
+++ b/src/pkg/compress/flate/deflate_test.go
@@ -116,9 +116,16 @@
panic("unreachable")
}
+func (b *syncBuffer) signal() {
+ select {
+ case b.ready <- true:
+ default:
+ }
+}
+
func (b *syncBuffer) Write(p []byte) (n int, err os.Error) {
n, err = b.buf.Write(p)
- _ = b.ready <- true
+ b.signal()
return
}
@@ -128,12 +135,12 @@
func (b *syncBuffer) ReadMode() {
b.mu.Unlock()
- _ = b.ready <- true
+ b.signal()
}
func (b *syncBuffer) Close() os.Error {
b.closed = true
- _ = b.ready <- true
+ b.signal()
return nil
}
diff --git a/src/pkg/exp/draw/x11/conn.go b/src/pkg/exp/draw/x11/conn.go
index da21815..e28fb217 100644
--- a/src/pkg/exp/draw/x11/conn.go
+++ b/src/pkg/exp/draw/x11/conn.go
@@ -122,10 +122,13 @@
func (c *conn) Screen() draw.Image { return c.img }
func (c *conn) FlushImage() {
- // We do the send (the <- operator) in an expression context, rather than in
- // a statement context, so that it does not block, and fails if the buffered
- // channel is full (in which case there already is a flush request pending).
- _ = c.flush <- false
+ select {
+ case c.flush <- false:
+ // Flush notification sent.
+ default:
+ // Could not send.
+ // Flush notification must be pending already.
+ }
}
func (c *conn) Close() os.Error {
diff --git a/src/pkg/fmt/print.go b/src/pkg/fmt/print.go
index 96029a8..d6dc8eb 100644
--- a/src/pkg/fmt/print.go
+++ b/src/pkg/fmt/print.go
@@ -74,15 +74,42 @@
fmt fmt
}
-// A leaky bucket of reusable pp structures.
-var ppFree = make(chan *pp, 100)
+// A cache holds a set of reusable objects.
+// The buffered channel holds the currently available objects.
+// If more are needed, the cache creates them by calling new.
+type cache struct {
+ saved chan interface{}
+ new func() interface{}
+}
-// Allocate a new pp struct. Probably can grab the previous one from ppFree.
-func newPrinter() *pp {
- p, ok := <-ppFree
- if !ok {
- p = new(pp)
+func (c *cache) put(x interface{}) {
+ select {
+ case c.saved <- x:
+ // saved in cache
+ default:
+ // discard
}
+}
+
+func (c *cache) get() interface{} {
+ select {
+ case x := <-c.saved:
+ return x // reused from cache
+ default:
+ return c.new()
+ }
+ panic("not reached")
+}
+
+func newCache(f func() interface{}) *cache {
+ return &cache{make(chan interface{}, 100), f}
+}
+
+var ppFree = newCache(func() interface{} { return new(pp) })
+
+// Allocate a new pp struct or grab a cached one.
+func newPrinter() *pp {
+ p := ppFree.get().(*pp)
p.fmt.init(&p.buf)
return p
}
@@ -94,7 +121,7 @@
return
}
p.buf.Reset()
- _ = ppFree <- p
+ ppFree.put(p)
}
func (p *pp) Width() (wid int, ok bool) { return p.fmt.wid, p.fmt.widPresent }
diff --git a/src/pkg/fmt/scan.go b/src/pkg/fmt/scan.go
index ebbb171..a408c42 100644
--- a/src/pkg/fmt/scan.go
+++ b/src/pkg/fmt/scan.go
@@ -303,15 +303,11 @@
}
-// A leaky bucket of reusable ss structures.
-var ssFree = make(chan *ss, 100)
+var ssFree = newCache(func() interface{} { return new(ss) })
-// Allocate a new ss struct. Probably can grab the previous one from ssFree.
+// Allocate a new ss struct or grab a cached one.
func newScanState(r io.Reader, nlIsSpace bool) *ss {
- s, ok := <-ssFree
- if !ok {
- s = new(ss)
- }
+ s := ssFree.get().(*ss)
if rr, ok := r.(readRuner); ok {
s.rr = rr
} else {
@@ -333,7 +329,7 @@
}
s.buf.Reset()
s.rr = nil
- _ = ssFree <- s
+ ssFree.put(s)
}
// skipSpace skips spaces and maybe newlines.
diff --git a/src/pkg/net/fd.go b/src/pkg/net/fd.go
index 896178f..2ba9296 100644
--- a/src/pkg/net/fd.go
+++ b/src/pkg/net/fd.go
@@ -220,11 +220,16 @@
nn, _ = s.pr.Read(scratch[0:])
}
// Read from channels
- for fd, ok := <-s.cr; ok; fd, ok = <-s.cr {
- s.AddFD(fd, 'r')
- }
- for fd, ok := <-s.cw; ok; fd, ok = <-s.cw {
- s.AddFD(fd, 'w')
+ Update:
+ for {
+ select {
+ case fd := <-s.cr:
+ s.AddFD(fd, 'r')
+ case fd := <-s.cw:
+ s.AddFD(fd, 'w')
+ default:
+ break Update
+ }
}
} else {
netfd := s.LookupFD(fd, mode)
diff --git a/src/pkg/net/server_test.go b/src/pkg/net/server_test.go
index 543227f..3dda500 100644
--- a/src/pkg/net/server_test.go
+++ b/src/pkg/net/server_test.go
@@ -140,13 +140,16 @@
listening <- c.LocalAddr().String()
c.SetReadTimeout(10e6) // 10ms
var buf [1000]byte
+Run:
for {
n, addr, err := c.ReadFrom(buf[0:])
if e, ok := err.(Error); ok && e.Timeout() {
- if done <- 1 {
- break
+ select {
+ case done <- 1:
+ break Run
+ default:
+ continue Run
}
- continue
}
if err != nil {
break
diff --git a/src/pkg/netchan/common.go b/src/pkg/netchan/common.go
index 56c0b25..6c08548 100644
--- a/src/pkg/netchan/common.go
+++ b/src/pkg/netchan/common.go
@@ -256,7 +256,10 @@
nch.sendCh = make(chan reflect.Value, nch.size)
go nch.sender()
}
- if ok := nch.sendCh <- val; !ok {
+ select {
+ case nch.sendCh <- val:
+ // ok
+ default:
// TODO: should this be more resilient?
panic("netchan: remote sender sent more values than allowed")
}
@@ -318,8 +321,11 @@
if nch.dir != Send {
panic("recv on wrong direction of channel")
}
- if ok := nch.ackCh <- true; !ok {
- panic("netchan: remote receiver sent too many acks")
+ select {
+ case nch.ackCh <- true:
+ // ok
+ default:
// TODO: should this be more resilient?
+ panic("netchan: remote receiver sent too many acks")
}
}
diff --git a/src/pkg/netchan/import.go b/src/pkg/netchan/import.go
index a694fb4..d1e9bbd 100644
--- a/src/pkg/netchan/import.go
+++ b/src/pkg/netchan/import.go
@@ -91,11 +91,13 @@
}
if err.Error != "" {
impLog("response error:", err.Error)
- if sent := imp.errors <- os.ErrorString(err.Error); !sent {
+ select {
+ case imp.errors <- os.ErrorString(err.Error):
+ continue // errors are not acknowledged
+ default:
imp.shutdown()
return
}
- continue // errors are not acknowledged.
}
case payClosed:
nch := imp.getChan(hdr.Id, false)
diff --git a/src/pkg/os/inotify/inotify_linux.go b/src/pkg/os/inotify/inotify_linux.go
index 1e74c7f..9d7a074 100644
--- a/src/pkg/os/inotify/inotify_linux.go
+++ b/src/pkg/os/inotify/inotify_linux.go
@@ -153,7 +153,11 @@
for {
n, errno = syscall.Read(w.fd, buf[0:])
// See if there is a message on the "done" channel
- _, done := <-w.done
+ var done bool
+ select {
+ case done = <-w.done:
+ default:
+ }
// If EOF or a "done" message is received
if n == 0 || done {
diff --git a/src/pkg/path/path_test.go b/src/pkg/path/path_test.go
index 6b4be07..ab0b48a 100644
--- a/src/pkg/path/path_test.go
+++ b/src/pkg/path/path_test.go
@@ -256,8 +256,11 @@
// 2) handle errors, expect none
errors := make(chan os.Error, 64)
Walk(tree.name, v, errors)
- if err, ok := <-errors; ok {
+ select {
+ case err := <-errors:
t.Errorf("no error expected, found: %s", err)
+ default:
+ // ok
}
checkMarks(t)
@@ -276,14 +279,21 @@
errors = make(chan os.Error, 64)
os.Chmod(Join(tree.name, tree.entries[1].name), 0)
Walk(tree.name, v, errors)
+ Loop:
for i := 1; i <= 2; i++ {
- if _, ok := <-errors; !ok {
+ select {
+ case <-errors:
+ // ok
+ default:
t.Errorf("%d. error expected, none found", i)
- break
+ break Loop
}
}
- if err, ok := <-errors; ok {
+ select {
+ case err := <-errors:
t.Errorf("only two errors expected, found 3rd: %v", err)
+ default:
+ // ok
}
// the inaccessible subtrees were marked manually
checkMarks(t)
diff --git a/src/pkg/rpc/client.go b/src/pkg/rpc/client.go
index 601c497..6f028c1 100644
--- a/src/pkg/rpc/client.go
+++ b/src/pkg/rpc/client.go
@@ -58,7 +58,7 @@
if client.shutdown != nil {
c.Error = client.shutdown
client.mutex.Unlock()
- _ = c.Done <- c // do not block
+ c.done()
return
}
c.seq = client.seq
@@ -102,16 +102,14 @@
// Empty strings should turn into nil os.Errors
c.Error = nil
}
- // We don't want to block here. It is the caller's responsibility to make
- // sure the channel has enough buffer space. See comment in Go().
- _ = c.Done <- c // do not block
+ c.done()
}
// Terminate pending calls.
client.mutex.Lock()
client.shutdown = err
for _, call := range client.pending {
call.Error = err
- _ = call.Done <- call // do not block
+ call.done()
}
client.mutex.Unlock()
if err != os.EOF || !client.closing {
@@ -119,6 +117,16 @@
}
}
+func (call *Call) done() {
+ select {
+ case call.Done <- call:
+ // ok
+ default:
+ // We don't want to block here. It is the caller's responsibility to make
+ // sure the channel has enough buffer space. See comment in Go().
+ }
+}
+
// NewClient returns a new Client to handle requests to the
// set of services at the other end of the connection.
func NewClient(conn io.ReadWriteCloser) *Client {
@@ -233,7 +241,7 @@
c.Done = done
if client.shutdown != nil {
c.Error = client.shutdown
- _ = c.Done <- c // do not block
+ c.done()
return c
}
client.send(c)
diff --git a/src/pkg/rpc/server_test.go b/src/pkg/rpc/server_test.go
index 355d51c..67b8762 100644
--- a/src/pkg/rpc/server_test.go
+++ b/src/pkg/rpc/server_test.go
@@ -364,14 +364,12 @@
testSendDeadlock(client)
done <- true
}()
- for i := 0; i < 50; i++ {
- time.Sleep(100 * 1e6)
- _, ok := <-done
- if ok {
- return
- }
+ select {
+ case <-done:
+ return
+ case <-time.After(5e9):
+ t.Fatal("deadlock")
}
- t.Fatal("deadlock")
}
func testSendDeadlock(client *Client) {
diff --git a/src/pkg/time/sleep_test.go b/src/pkg/time/sleep_test.go
index 4007db5..8bf599c 100644
--- a/src/pkg/time/sleep_test.go
+++ b/src/pkg/time/sleep_test.go
@@ -120,10 +120,12 @@
t.Fatalf("failed to stop event 1")
}
<-c2
- _, ok0 := <-t0.C
- _, ok1 := <-c1
- if ok0 || ok1 {
- t.Fatalf("events were not stopped")
+ select {
+ case <-t0.C:
+ t.Fatalf("event 0 was not stopped")
+ case <-c1:
+ t.Fatalf("event 1 was not stopped")
+ default:
}
if t1.Stop() {
t.Fatalf("Stop returned true twice")
diff --git a/src/pkg/time/tick.go b/src/pkg/time/tick.go
index ddd7272..6c21bf1 100644
--- a/src/pkg/time/tick.go
+++ b/src/pkg/time/tick.go
@@ -22,8 +22,12 @@
// Stop turns off a ticker. After Stop, no more ticks will be sent.
func (t *Ticker) Stop() {
- // Make it non-blocking so multiple Stops don't block.
- _ = t.shutdown <- true
+ select {
+ case t.shutdown <- true:
+ // ok
+ default:
+ // Stop in progress already
+ }
}
// Tick is a convenience wrapper for NewTicker providing access to the ticking
@@ -106,7 +110,8 @@
// that need it and determining the next wake time.
// TODO(r): list should be sorted in time order.
for t := tickers; t != nil; t = t.next {
- if _, ok := <-t.shutdown; ok {
+ select {
+ case <-t.shutdown:
// Ticker is done; remove it from list.
if prev == nil {
tickers = t.next
@@ -114,6 +119,7 @@
prev.next = t.next
}
continue
+ default:
}
if t.nextTick <= now {
if len(t.c) == 0 {
diff --git a/src/pkg/time/tick_test.go b/src/pkg/time/tick_test.go
index f2304a1..4dcb639 100644
--- a/src/pkg/time/tick_test.go
+++ b/src/pkg/time/tick_test.go
@@ -29,9 +29,11 @@
}
// Now test that the ticker stopped
Sleep(2 * Delta)
- _, received := <-ticker.C
- if received {
+ select {
+ case <-ticker.C:
t.Fatal("Ticker did not shut down")
+ default:
+ // ok
}
}
