Commit a8bec8ce authored by Han-Wen Nienhuys's avatar Han-Wen Nienhuys

splice: use syscall directly for splice I/O

This avoids involving the go 1.9 poller in I/O, which causes hangs, as
the poller does not understand the fcntl(O_NONBLOCK) tweak that we
executed on the pipe's file descriptors.

Fixes #164
parent c878ca45
...@@ -6,11 +6,11 @@ package splice ...@@ -6,11 +6,11 @@ package splice
import ( import (
"fmt" "fmt"
"os" "syscall"
) )
type Pair struct { type Pair struct {
r, w *os.File r, w int
size int size int
} }
...@@ -30,7 +30,7 @@ func (p *Pair) Grow(n int) error { ...@@ -30,7 +30,7 @@ func (p *Pair) Grow(n int) error {
return fmt.Errorf("splice: want %d bytes, max pipe size %d", n, maxPipeSize) return fmt.Errorf("splice: want %d bytes, max pipe size %d", n, maxPipeSize)
} }
newsize, errNo := fcntl(p.r.Fd(), F_SETPIPE_SZ, n) newsize, errNo := fcntl(uintptr(p.r), F_SETPIPE_SZ, n)
if errNo != 0 { if errNo != 0 {
return fmt.Errorf("splice: fcntl returned %v", errNo) return fmt.Errorf("splice: fcntl returned %v", errNo)
} }
...@@ -43,8 +43,8 @@ func (p *Pair) Cap() int { ...@@ -43,8 +43,8 @@ func (p *Pair) Cap() int {
} }
func (p *Pair) Close() error { func (p *Pair) Close() error {
err1 := p.r.Close() err1 := syscall.Close(p.r)
err2 := p.w.Close() err2 := syscall.Close(p.w)
if err1 != nil { if err1 != nil {
return err1 return err1
} }
...@@ -52,17 +52,17 @@ func (p *Pair) Close() error { ...@@ -52,17 +52,17 @@ func (p *Pair) Close() error {
} }
func (p *Pair) Read(d []byte) (n int, err error) { func (p *Pair) Read(d []byte) (n int, err error) {
return p.r.Read(d) return syscall.Read(p.r, d)
} }
func (p *Pair) ReadFd() uintptr { func (p *Pair) Write(d []byte) (n int, err error) {
return p.r.Fd() return syscall.Write(p.w, d)
} }
func (p *Pair) WriteFd() uintptr { func (p *Pair) ReadFd() uintptr {
return p.w.Fd() return uintptr(p.r)
} }
func (p *Pair) Write(d []byte) (n int, err error) { func (p *Pair) WriteFd() uintptr {
return p.w.Write(d) return uintptr(p.w)
} }
...@@ -11,7 +11,7 @@ import ( ...@@ -11,7 +11,7 @@ import (
) )
func (p *Pair) LoadFromAt(fd uintptr, sz int, off int64) (int, error) { func (p *Pair) LoadFromAt(fd uintptr, sz int, off int64) (int, error) {
n, err := syscall.Splice(int(fd), &off, int(p.w.Fd()), nil, sz, 0) n, err := syscall.Splice(int(fd), &off, p.w, nil, sz, 0)
return int(n), err return int(n), err
} }
...@@ -21,7 +21,7 @@ func (p *Pair) LoadFrom(fd uintptr, sz int) (int, error) { ...@@ -21,7 +21,7 @@ func (p *Pair) LoadFrom(fd uintptr, sz int) (int, error) {
sz, p.size) sz, p.size)
} }
n, err := syscall.Splice(int(fd), nil, int(p.w.Fd()), nil, sz, 0) n, err := syscall.Splice(int(fd), nil, p.w, nil, sz, 0)
if err != nil { if err != nil {
err = os.NewSyscallError("Splice load from", err) err = os.NewSyscallError("Splice load from", err)
} }
...@@ -29,7 +29,7 @@ func (p *Pair) LoadFrom(fd uintptr, sz int) (int, error) { ...@@ -29,7 +29,7 @@ func (p *Pair) LoadFrom(fd uintptr, sz int) (int, error) {
} }
func (p *Pair) WriteTo(fd uintptr, n int) (int, error) { func (p *Pair) WriteTo(fd uintptr, n int) (int, error) {
m, err := syscall.Splice(int(p.r.Fd()), nil, int(fd), nil, int(n), 0) m, err := syscall.Splice(p.r, nil, int(fd), nil, int(n), 0)
if err != nil { if err != nil {
err = os.NewSyscallError("Splice write", err) err = os.NewSyscallError("Splice write", err)
} }
......
...@@ -5,8 +5,8 @@ ...@@ -5,8 +5,8 @@
package splice package splice
import ( import (
"io"
"sync" "sync"
"syscall"
) )
var splicePool *pairPool var splicePool *pairPool
...@@ -95,10 +95,15 @@ func (me *pairPool) get() (p *Pair, err error) { ...@@ -95,10 +95,15 @@ func (me *pairPool) get() (p *Pair, err error) {
var discardBuffer [32 * 1024]byte var discardBuffer [32 * 1024]byte
func DiscardAll(r io.Reader) { func discardAll(fd int) {
buf := discardBuffer[:] buf := discardBuffer[:]
r := 0
for { for {
n, _ := r.Read(buf) n, _ := syscall.Read(fd, buf)
if n > 0 {
r += n
}
if n < len(buf) { if n < len(buf) {
break break
} }
...@@ -106,7 +111,7 @@ func DiscardAll(r io.Reader) { ...@@ -106,7 +111,7 @@ func DiscardAll(r io.Reader) {
} }
func (me *pairPool) done(p *Pair) { func (me *pairPool) done(p *Pair) {
DiscardAll(p.r) discardAll(p.r)
me.Lock() me.Lock()
me.usedCount-- me.usedCount--
......
...@@ -61,30 +61,27 @@ func fcntl(fd uintptr, cmd int, arg int) (val int, errno syscall.Errno) { ...@@ -61,30 +61,27 @@ func fcntl(fd uintptr, cmd int, arg int) (val int, errno syscall.Errno) {
const F_SETPIPE_SZ = 1031 const F_SETPIPE_SZ = 1031
const F_GETPIPE_SZ = 1032 const F_GETPIPE_SZ = 1032
func osPipe() (int, int, error) {
var fds [2]int
err := syscall.Pipe2(fds[:], syscall.O_NONBLOCK)
return fds[0], fds[1], err
}
func newSplicePair() (p *Pair, err error) { func newSplicePair() (p *Pair, err error) {
p = &Pair{} p = &Pair{}
p.r, p.w, err = os.Pipe() p.r, p.w, err = osPipe()
if err != nil { if err != nil {
return nil, err return nil, err
} }
var errNo syscall.Errno
errNo := syscall.Errno(0) p.size, errNo = fcntl(uintptr(p.r), F_GETPIPE_SZ, 0)
for _, f := range []*os.File{p.r, p.w} { if err == syscall.EINVAL {
_, errNo = fcntl(f.Fd(), syscall.F_SETFL, syscall.O_NONBLOCK)
if errNo != 0 {
p.Close()
return nil, os.NewSyscallError("fcntl setfl", errNo)
}
}
p.size, errNo = fcntl(p.r.Fd(), F_GETPIPE_SZ, 0)
if errNo == syscall.EINVAL {
p.size = DefaultPipeSize p.size = DefaultPipeSize
return p, nil return p, nil
} }
if errNo != 0 { if errNo != 0 {
p.Close() p.Close()
return nil, os.NewSyscallError("fcntl getsize", errNo) return nil, fmt.Errorf("fcntl getsize: %v", errNo)
} }
return p, nil return p, nil
} }
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