libgo/go/net/pipe.go

   1 // Copyright 2010 The Go Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style
   3 // license that can be found in the LICENSE file.
   4
   5 package net
   6
   7 import (
   8         "io"
   9         "sync"
  10         "time"
  11 )
  12
  13 // pipeDeadline is an abstraction for handling timeouts.
  14 type pipeDeadline struct {
  15         mu     sync.Mutex // Guards timer and cancel
  16         timer  *time.Timer
  17         cancel chan struct{} // Must be non-nil
  18 }
  19
  20 func makePipeDeadline() pipeDeadline {
  21         return pipeDeadline{cancel: make(chan struct{})}
  22 }
  23
  24 // set sets the point in time when the deadline will time out.
  25 // A timeout event is signaled by closing the channel returned by waiter.
  26 // Once a timeout has occurred, the deadline can be refreshed by specifying a
  27 // t value in the future.
  28 //
  29 // A zero value for t prevents timeout.
  30 func (d *pipeDeadline) set(t time.Time) {
  31         d.mu.Lock()
  32         defer d.mu.Unlock()
  33
  34         if d.timer != nil && !d.timer.Stop() {
  35                 <-d.cancel // Wait for the timer callback to finish and close cancel
  36         }
  37         d.timer = nil
  38
  39         // Time is zero, then there is no deadline.
  40         closed := isClosedChan(d.cancel)
  41         if t.IsZero() {
  42                 if closed {
  43                         d.cancel = make(chan struct{})
  44                 }
  45                 return
  46         }
  47
  48         // Time in the future, setup a timer to cancel in the future.
  49         if dur := time.Until(t); dur > 0 {
  50                 if closed {
  51                         d.cancel = make(chan struct{})
  52                 }
  53                 d.timer = time.AfterFunc(dur, func() {
  54                         close(d.cancel)
  55                 })
  56                 return
  57         }
  58
  59         // Time in the past, so close immediately.
  60         if !closed {
  61                 close(d.cancel)
  62         }
  63 }
  64
  65 // wait returns a channel that is closed when the deadline is exceeded.
  66 func (d *pipeDeadline) wait() chan struct{} {
  67         d.mu.Lock()
  68         defer d.mu.Unlock()
  69         return d.cancel
  70 }
  71
  72 func isClosedChan(c <-chan struct{}) bool {
  73         select {
  74         case <-c:
  75                 return true
  76         default:
  77                 return false
  78         }
  79 }
  80
  81 type timeoutError struct{}
  82
  83 func (timeoutError) Error() string   { return "deadline exceeded" }
  84 func (timeoutError) Timeout() bool   { return true }
  85 func (timeoutError) Temporary() bool { return true }
  86
  87 type pipeAddr struct{}
  88
  89 func (pipeAddr) Network() string { return "pipe" }
  90 func (pipeAddr) String() string  { return "pipe" }
  91
  92 type pipe struct {
  93         wrMu sync.Mutex // Serialize Write operations
  94
  95         // Used by local Read to interact with remote Write.
  96         // Successful receive on rdRx is always followed by send on rdTx.
  97         rdRx <-chan []byte
  98         rdTx chan<- int
  99
 100         // Used by local Write to interact with remote Read.
 101         // Successful send on wrTx is always followed by receive on wrRx.
 102         wrTx chan<- []byte
 103         wrRx <-chan int
 104
 105         once       sync.Once // Protects closing localDone
 106         localDone  chan struct{}
 107         remoteDone <-chan struct{}
 108
 109         readDeadline  pipeDeadline
 110         writeDeadline pipeDeadline
 111 }
 112
 113 // Pipe creates a synchronous, in-memory, full duplex
 114 // network connection; both ends implement the Conn interface.
 115 // Reads on one end are matched with writes on the other,
 116 // copying data directly between the two; there is no internal
 117 // buffering.
 118 func Pipe() (Conn, Conn) {
 119         cb1 := make(chan []byte)
 120         cb2 := make(chan []byte)
 121         cn1 := make(chan int)
 122         cn2 := make(chan int)
 123         done1 := make(chan struct{})
 124         done2 := make(chan struct{})
 125
 126         p1 := &pipe{
 127                 rdRx: cb1, rdTx: cn1,
 128                 wrTx: cb2, wrRx: cn2,
 129                 localDone: done1, remoteDone: done2,
 130                 readDeadline:  makePipeDeadline(),
 131                 writeDeadline: makePipeDeadline(),
 132         }
 133         p2 := &pipe{
 134                 rdRx: cb2, rdTx: cn2,
 135                 wrTx: cb1, wrRx: cn1,
 136                 localDone: done2, remoteDone: done1,
 137                 readDeadline:  makePipeDeadline(),
 138                 writeDeadline: makePipeDeadline(),
 139         }
 140         return p1, p2
 141 }
 142
 143 func (*pipe) LocalAddr() Addr  { return pipeAddr{} }
 144 func (*pipe) RemoteAddr() Addr { return pipeAddr{} }
 145
 146 func (p *pipe) Read(b []byte) (int, error) {
 147         n, err := p.read(b)
 148         if err != nil && err != io.EOF && err != io.ErrClosedPipe {
 149                 err = &OpError{Op: "read", Net: "pipe", Err: err}
 150         }
 151         return n, err
 152 }
 153
 154 func (p *pipe) read(b []byte) (n int, err error) {
 155         switch {
 156         case isClosedChan(p.localDone):
 157                 return 0, io.ErrClosedPipe
 158         case isClosedChan(p.remoteDone):
 159                 return 0, io.EOF
 160         case isClosedChan(p.readDeadline.wait()):
 161                 return 0, timeoutError{}
 162         }
 163
 164         select {
 165         case bw := <-p.rdRx:
 166                 nr := copy(b, bw)
 167                 p.rdTx <- nr
 168                 return nr, nil
 169         case <-p.localDone:
 170                 return 0, io.ErrClosedPipe
 171         case <-p.remoteDone:
 172                 return 0, io.EOF
 173         case <-p.readDeadline.wait():
 174                 return 0, timeoutError{}
 175         }
 176 }
 177
 178 func (p *pipe) Write(b []byte) (int, error) {
 179         n, err := p.write(b)
 180         if err != nil && err != io.ErrClosedPipe {
 181                 err = &OpError{Op: "write", Net: "pipe", Err: err}
 182         }
 183         return n, err
 184 }
 185
 186 func (p *pipe) write(b []byte) (n int, err error) {
 187         switch {
 188         case isClosedChan(p.localDone):
 189                 return 0, io.ErrClosedPipe
 190         case isClosedChan(p.remoteDone):
 191                 return 0, io.ErrClosedPipe
 192         case isClosedChan(p.writeDeadline.wait()):
 193                 return 0, timeoutError{}
 194         }
 195
 196         p.wrMu.Lock() // Ensure entirety of b is written together
 197         defer p.wrMu.Unlock()
 198         for once := true; once || len(b) > 0; once = false {
 199                 select {
 200                 case p.wrTx <- b:
 201                         nw := <-p.wrRx
 202                         b = b[nw:]
 203                         n += nw
 204                 case <-p.localDone:
 205                         return n, io.ErrClosedPipe
 206                 case <-p.remoteDone:
 207                         return n, io.ErrClosedPipe
 208                 case <-p.writeDeadline.wait():
 209                         return n, timeoutError{}
 210                 }
 211         }
 212         return n, nil
 213 }
 214
 215 func (p *pipe) SetDeadline(t time.Time) error {
 216         if isClosedChan(p.localDone) || isClosedChan(p.remoteDone) {
 217                 return io.ErrClosedPipe
 218         }
 219         p.readDeadline.set(t)
 220         p.writeDeadline.set(t)
 221         return nil
 222 }
 223
 224 func (p *pipe) SetReadDeadline(t time.Time) error {
 225         if isClosedChan(p.localDone) || isClosedChan(p.remoteDone) {
 226                 return io.ErrClosedPipe
 227         }
 228         p.readDeadline.set(t)
 229         return nil
 230 }
 231
 232 func (p *pipe) SetWriteDeadline(t time.Time) error {
 233         if isClosedChan(p.localDone) || isClosedChan(p.remoteDone) {
 234                 return io.ErrClosedPipe
 235         }
 236         p.writeDeadline.set(t)
 237         return nil
 238 }
 239
 240 func (p *pipe) Close() error {
 241         p.once.Do(func() { close(p.localDone) })
 242         return nil
 243 }