get rid os pipe in favor of sockets

Author: ssgreg

.gitignore

@@ -3,6 +3,7 @@
 *.dll
 *.so
 *.dylib
+debug.test
 
 # Test binary, build with `go test -c`
 *.test

app.go

@@ -89,13 +89,18 @@ CatchSignals:
 				a.shutdown()
 			// Fork/Exec a child and shutdown.
 			case syscall.SIGUSR2:
-				_, cp, err := forkExec(e.activeFiles())
+				_, f, err := forkExec(e.activeFiles())
 				if err != nil {
 					logger.Printf("ZeroDT: failed to forkExec: '%v'", err)
 					continue CatchSignals
 				}
+				m, err := ListenSocket(f)
+				if err != nil {
+					logger.Printf("ZeroDT: failed to listen communication socket: '%v'", err)
+					continue CatchSignals
+				}
 				// Nothing to do with errors.
-				protocolActAsParent(cp, a.WaitForChildTimeout, func() {
+				protocolActAsParent(m, a.WaitForChildTimeout, func() {
 					a.shutdown()
 				})
 			}
@@ -108,7 +113,7 @@ CatchSignals:
 
 // Serve TODO
 func (a *App) Serve() error {
-	inherited, cp, err := inherit()
+	inherited, m, err := inherit()
 	if err != nil {
 		logger.Printf("ZeroDT: failed to inherit listeners with: '%v'", err)
 		return err
@@ -152,8 +157,8 @@ func (a *App) Serve() error {
 	// Wait for all listeners to start listening.
 	startWG.Wait()
 
-	if cp != nil {
-		protocolActAsChild(cp, a.WaitForParentTimeout)
+	if m != nil {
+		protocolActAsChild(m, a.WaitForParentTimeout)
 	}
 
 	// Allow serverse's goroutines to start serving
@@ -172,7 +177,7 @@ func (a *App) Serve() error {
 
 // forkExec starts another process of youself and passes active
 // listeners to a child to perform socket activation.
-func forkExec(files []*os.File) (int, *PipeJSONMessenger, error) {
+func forkExec(files []*os.File) (int, *os.File, error) {
 	// Get the path name for the executable that started the current process.
 	path, err := os.Executable()
 	if err != nil {
@@ -185,15 +190,14 @@ func forkExec(files []*os.File) (int, *PipeJSONMessenger, error) {
 		return -1, nil, err
 	}
 
-	cr, cw, err := os.Pipe()
+	fds, err := syscall.Socketpair(syscall.AF_UNIX, syscall.SOCK_STREAM, 0)
 	if err != nil {
 		return -1, nil, err
 	}
-	pr, pw, err := os.Pipe()
-	if err != nil {
-		return -1, nil, err
-	}
-	files = append(files, cr, pw)
+	f0 := os.NewFile(uintptr(fds[0]), "s|0")
+	f1 := os.NewFile(uintptr(fds[1]), "s|0")
+
+	files = append(files, f1)
 
 	// Start the original executable with the original working directory.
 	process, err := os.StartProcess(path, os.Args, &os.ProcAttr{
@@ -205,7 +209,7 @@ func forkExec(files []*os.File) (int, *PipeJSONMessenger, error) {
 		return -1, nil, err
 	}
 
-	return process.Pid, ListenPipe(pr, cw), nil
+	return process.Pid, f0, nil
 }
 
 // formatInherited prints info about inherited listeners to a string.
@@ -221,7 +225,7 @@ func formatInherited(e *exchange) string {
 	return result
 }
 
-func protocolActAsParent(m *PipeJSONMessenger, timeout time.Duration, shutdownFn func()) error {
+func protocolActAsParent(m *StreamMessenger, timeout time.Duration, shutdownFn func()) error {
 	defer m.Close()
 	// Set a timeout for the whole dialog.
 	m.SetDeadline(time.Now().Add(timeout))
@@ -248,7 +252,7 @@ func protocolActAsParent(m *PipeJSONMessenger, timeout time.Duration, shutdownFn
 	return nil
 }
 
-func protocolActAsChild(m *PipeJSONMessenger, timeout time.Duration) {
+func protocolActAsChild(m *StreamMessenger, timeout time.Duration) {
 	defer m.Close()
 	// Child->Parent, ready message
 	logger.Printf("ZeroDT: sending ready to a parent...")

examples/test/main.go

@@ -2,14 +2,12 @@ package main
 
 import (
 	"fmt"
-	"net"
 	"net/http"
 	"os"
 	"ssgreg/zerodt"
 	"time"
 
 	"github.com/Sirupsen/logrus"
-	"github.com/gorilla/mux"
 )
 
 func sleep(w http.ResponseWriter, r *http.Request) {
@@ -24,30 +22,23 @@ func sleep(w http.ResponseWriter, r *http.Request) {
 	logrus.Printf("Handled message %d!", os.Getpid())
 }
 
-type childReadyMsg struct {
-	Pid  int `json:"pid"`
-	Data [65520]byte
-	Str  int
-	Str1 string
-}
-
 func main() {
 	zerodt.SetLogger(logrus.StandardLogger())
 
-	r := mux.NewRouter()
-	r.Path("/sleep").Methods("GET").HandlerFunc(sleep)
+	// r := mux.NewRouter()
+	// r.Path("/sleep").Methods("GET").HandlerFunc(sleep)
 
 	// a := zerodt.NewApp(&http.Server{Addr: "127.0.0.1:8081", Handler: r}, &http.Server{Addr: "127.0.0.1:8082", Handler: r})
 	// a.Serve()
 
-	srv := http.Server{Addr: "127.0.0.1:8081", Handler: r}
+	// srv := http.Server{Addr: "127.0.0.1:8081", Handler: r}
 
-	addr, _ := net.ResolveTCPAddr("tcp", "127.0.0.1:8081")
-	l, _ := net.ListenTCP("tcp", addr)
+	// addr, _ := net.ResolveTCPAddr("tcp", "127.0.0.1:8081")
+	// l, _ := net.ListenTCP("tcp", addr)
 
 	// err := srv.Shutdown(context.Background())
 	// fmt.Println(err)
-	srv.Serve(l)
+	// srv.Serve(l)
 
 	// var wg sync.WaitGroup
 	// wg.Add(2)

file_listener.go

@@ -47,7 +47,7 @@ type fileListenerPair struct {
 // inherit returns all inherited listeners with
 // duplicated file descriptors wrapped in os.File.
 // Can be called only once.
-func inherit() ([]*fileListenerPair, *PipeJSONMessenger, error) {
+func inherit() ([]*fileListenerPair, *StreamMessenger, error) {
 	// Are there some listeners to inherit?
 	fds, err := listenFds()
 	if err != nil {
@@ -61,13 +61,13 @@ func inherit() ([]*fileListenerPair, *PipeJSONMessenger, error) {
 	return pairs, cp, nil
 }
 
-func inheritWithFDS(fds []int) ([]*fileListenerPair, *PipeJSONMessenger, error) {
+func inheritWithFDS(fds []int) ([]*fileListenerPair, *StreamMessenger, error) {
 	m, err := listenPipeWithFDS(fds)
 	if err != nil {
 		return nil, nil, err
 	}
 	if m != nil {
-		fds = fds[0 : len(fds)-2]
+		fds = fds[0 : len(fds)-1]
 	}
 	// Start to listen them.
 	pairs := make([]*fileListenerPair, len(fds))
@@ -85,23 +85,20 @@ func inheritWithFDS(fds []int) ([]*fileListenerPair, *PipeJSONMessenger, error)
 	return pairs, m, nil
 }
 
-func listenPipeWithFDS(fds []int) (*PipeJSONMessenger, error) {
+func listenPipeWithFDS(fds []int) (*StreamMessenger, error) {
 	count := len(fds)
-	if count > 1 {
+	if count > 0 {
 		ok, err := isSocketTCP(fds[count-1])
 		if err != nil {
 			return nil, err
 		}
 		if !ok {
-			ok, err := isSocketTCP(fds[count-2])
+			s := os.NewFile(uintptr(fds[count-1]), "s|0")
+			m, err := ListenSocket(s)
 			if err != nil {
 				return nil, err
 			}
-			if !ok {
-				r := os.NewFile(uintptr(fds[count-2]), "|0")
-				w := os.NewFile(uintptr(fds[count-1]), "|1")
-				return ListenPipe(r, w), nil
-			}
+			return m, nil
 		}
 	}
 	return nil, nil
@@ -152,6 +149,15 @@ func isSocketTCP(fd int) (bool, error) {
 	if socketType != syscall.SOCK_STREAM {
 		return false, nil
 	}
+	lsa, err := syscall.Getsockname(fd)
+	if err != nil {
+		return false, err
+	}
+
+	switch lsa.(type) {
+	case *syscall.SockaddrUnix:
+		return false, nil
+	}
 	return true, nil
 }