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