added RPC/IPC support

1 files changed, 699 insertions, 0 deletions

diff --git a/rpc/comms/ipc_windows.go b/rpc/comms/ipc_windows.go
new file mode 100644
index 000000000..ff9015d03
--- /dev/null
+++ b/rpc/comms/ipc_windows.go
@@ -0,0 +1,699 @@
+// +build windows

+

+package comms

+

+import (

+    "fmt"

+	"io"

+	"net"

+    "os"

+	"sync"

+	"syscall"

+	"time"

+    "unsafe"

+    

+    "github.com/ethereum/go-ethereum/logger"

+	"github.com/ethereum/go-ethereum/logger/glog"

+	"github.com/ethereum/go-ethereum/rpc/api"

+	"github.com/ethereum/go-ethereum/rpc/codec"

+	"github.com/ethereum/go-ethereum/rpc/shared"

+)

+

+var (

+	modkernel32 = syscall.NewLazyDLL("kernel32.dll")

+

+	procCreateNamedPipeW    = modkernel32.NewProc("CreateNamedPipeW")

+	procConnectNamedPipe    = modkernel32.NewProc("ConnectNamedPipe")

+	procDisconnectNamedPipe = modkernel32.NewProc("DisconnectNamedPipe")

+	procWaitNamedPipeW      = modkernel32.NewProc("WaitNamedPipeW")

+	procCreateEventW        = modkernel32.NewProc("CreateEventW")

+	procGetOverlappedResult = modkernel32.NewProc("GetOverlappedResult")

+	procCancelIoEx          = modkernel32.NewProc("CancelIoEx")

+)

+

+func createNamedPipe(name *uint16, openMode uint32, pipeMode uint32, maxInstances uint32, outBufSize uint32, inBufSize uint32, defaultTimeout uint32, sa *syscall.SecurityAttributes) (handle syscall.Handle, err error) {

+	r0, _, e1 := syscall.Syscall9(procCreateNamedPipeW.Addr(), 8, uintptr(unsafe.Pointer(name)), uintptr(openMode), uintptr(pipeMode), uintptr(maxInstances), uintptr(outBufSize), uintptr(inBufSize), uintptr(defaultTimeout), uintptr(unsafe.Pointer(sa)), 0)

+	handle = syscall.Handle(r0)

+	if handle == syscall.InvalidHandle {

+		if e1 != 0 {

+			err = error(e1)

+		} else {

+			err = syscall.EINVAL

+		}

+	}

+	return

+}

+

+func cancelIoEx(handle syscall.Handle, overlapped *syscall.Overlapped) (err error) {

+	r1, _, e1 := syscall.Syscall(procCancelIoEx.Addr(), 2, uintptr(handle), uintptr(unsafe.Pointer(overlapped)), 0)

+	if r1 == 0 {

+		if e1 != 0 {

+			err = error(e1)

+		} else {

+			err = syscall.EINVAL

+		}

+	}

+	return

+}

+

+func connectNamedPipe(handle syscall.Handle, overlapped *syscall.Overlapped) (err error) {

+	r1, _, e1 := syscall.Syscall(procConnectNamedPipe.Addr(), 2, uintptr(handle), uintptr(unsafe.Pointer(overlapped)), 0)

+	if r1 == 0 {

+		if e1 != 0 {

+			err = error(e1)

+		} else {

+			err = syscall.EINVAL

+		}

+	}

+	return

+}

+

+func disconnectNamedPipe(handle syscall.Handle) (err error) {

+	r1, _, e1 := syscall.Syscall(procDisconnectNamedPipe.Addr(), 1, uintptr(handle), 0, 0)

+	if r1 == 0 {

+		if e1 != 0 {

+			err = error(e1)

+		} else {

+			err = syscall.EINVAL

+		}

+	}

+	return

+}

+

+func waitNamedPipe(name *uint16, timeout uint32) (err error) {

+	r1, _, e1 := syscall.Syscall(procWaitNamedPipeW.Addr(), 2, uintptr(unsafe.Pointer(name)), uintptr(timeout), 0)

+	if r1 == 0 {

+		if e1 != 0 {

+			err = error(e1)

+		} else {

+			err = syscall.EINVAL

+		}

+	}

+	return

+}

+

+func createEvent(sa *syscall.SecurityAttributes, manualReset bool, initialState bool, name *uint16) (handle syscall.Handle, err error) {

+	var _p0 uint32

+	if manualReset {

+		_p0 = 1

+	} else {

+		_p0 = 0

+	}

+	var _p1 uint32

+	if initialState {

+		_p1 = 1

+	} else {

+		_p1 = 0

+	}

+	r0, _, e1 := syscall.Syscall6(procCreateEventW.Addr(), 4, uintptr(unsafe.Pointer(sa)), uintptr(_p0), uintptr(_p1), uintptr(unsafe.Pointer(name)), 0, 0)

+	handle = syscall.Handle(r0)

+	if handle == syscall.InvalidHandle {

+		if e1 != 0 {

+			err = error(e1)

+		} else {

+			err = syscall.EINVAL

+		}

+	}

+	return

+}

+

+func getOverlappedResult(handle syscall.Handle, overlapped *syscall.Overlapped, transferred *uint32, wait bool) (err error) {

+	var _p0 uint32

+	if wait {

+		_p0 = 1

+	} else {

+		_p0 = 0

+	}

+	r1, _, e1 := syscall.Syscall6(procGetOverlappedResult.Addr(), 4, uintptr(handle), uintptr(unsafe.Pointer(overlapped)), uintptr(unsafe.Pointer(transferred)), uintptr(_p0), 0, 0)

+	if r1 == 0 {

+		if e1 != 0 {

+			err = error(e1)

+		} else {

+			err = syscall.EINVAL

+		}

+	}

+	return

+}

+

+

+const (

+	// openMode

+	pipe_access_duplex   = 0x3

+	pipe_access_inbound  = 0x1

+	pipe_access_outbound = 0x2

+

+	// openMode write flags

+	file_flag_first_pipe_instance = 0x00080000

+	file_flag_write_through       = 0x80000000

+	file_flag_overlapped          = 0x40000000

+

+	// openMode ACL flags

+	write_dac              = 0x00040000

+	write_owner            = 0x00080000

+	access_system_security = 0x01000000

+

+	// pipeMode

+	pipe_type_byte    = 0x0

+	pipe_type_message = 0x4

+

+	// pipeMode read mode flags

+	pipe_readmode_byte    = 0x0

+	pipe_readmode_message = 0x2

+

+	// pipeMode wait mode flags

+	pipe_wait   = 0x0

+	pipe_nowait = 0x1

+

+	// pipeMode remote-client mode flags

+	pipe_accept_remote_clients = 0x0

+	pipe_reject_remote_clients = 0x8

+

+	pipe_unlimited_instances = 255

+

+	nmpwait_wait_forever = 0xFFFFFFFF

+

+	// the two not-an-errors below occur if a client connects to the pipe between

+	// the server's CreateNamedPipe and ConnectNamedPipe calls.

+	error_no_data        syscall.Errno = 0xE8

+	error_pipe_connected syscall.Errno = 0x217

+	error_pipe_busy      syscall.Errno = 0xE7

+	error_sem_timeout    syscall.Errno = 0x79

+

+	error_bad_pathname syscall.Errno = 0xA1

+	error_invalid_name syscall.Errno = 0x7B

+

+	error_io_incomplete syscall.Errno = 0x3e4

+)

+

+var _ net.Conn = (*PipeConn)(nil)

+var _ net.Listener = (*PipeListener)(nil)

+

+// ErrClosed is the error returned by PipeListener.Accept when Close is called

+// on the PipeListener.

+var ErrClosed = PipeError{"Pipe has been closed.", false}

+

+// PipeError is an error related to a call to a pipe

+type PipeError struct {

+	msg     string

+	timeout bool

+}

+

+// Error implements the error interface

+func (e PipeError) Error() string {

+	return e.msg

+}

+

+// Timeout implements net.AddrError.Timeout()

+func (e PipeError) Timeout() bool {

+	return e.timeout

+}

+

+// Temporary implements net.AddrError.Temporary()

+func (e PipeError) Temporary() bool {

+	return false

+}

+

+// Dial connects to a named pipe with the given address. If the specified pipe is not available,

+// it will wait indefinitely for the pipe to become available.

+//

+// The address must be of the form \\.\\pipe\<name> for local pipes and \\<computer>\pipe\<name>

+// for remote pipes.

+//

+// Dial will return a PipeError if you pass in a badly formatted pipe name.

+//

+// Examples:

+//   // local pipe

+//   conn, err := Dial(`\\.\pipe\mypipename`)

+//

+//   // remote pipe

+//   conn, err := Dial(`\\othercomp\pipe\mypipename`)

+func Dial(address string) (*PipeConn, error) {

+	for {

+		conn, err := dial(address, nmpwait_wait_forever)

+		if err == nil {

+			return conn, nil

+		}

+		if isPipeNotReady(err) {

+			<-time.After(100 * time.Millisecond)

+			continue

+		}

+		return nil, err

+	}

+}

+

+// DialTimeout acts like Dial, but will time out after the duration of timeout

+func DialTimeout(address string, timeout time.Duration) (*PipeConn, error) {

+	deadline := time.Now().Add(timeout)

+

+	now := time.Now()

+	for now.Before(deadline) {

+		millis := uint32(deadline.Sub(now) / time.Millisecond)

+		conn, err := dial(address, millis)

+		if err == nil {

+			return conn, nil

+		}

+		if err == error_sem_timeout {

+			// This is WaitNamedPipe's timeout error, so we know we're done

+			return nil, PipeError{fmt.Sprintf(

+				"Timed out waiting for pipe '%s' to come available", address), true}

+		}

+		if isPipeNotReady(err) {

+			left := deadline.Sub(time.Now())

+			retry := 100 * time.Millisecond

+			if left > retry {

+				<-time.After(retry)

+			} else {

+				<-time.After(left - time.Millisecond)

+			}

+			now = time.Now()

+			continue

+		}

+		return nil, err

+	}

+	return nil, PipeError{fmt.Sprintf(

+		"Timed out waiting for pipe '%s' to come available", address), true}

+}

+

+// isPipeNotReady checks the error to see if it indicates the pipe is not ready

+func isPipeNotReady(err error) bool {

+	// Pipe Busy means another client just grabbed the open pipe end,

+	// and the server hasn't made a new one yet.

+	// File Not Found means the server hasn't created the pipe yet.

+	// Neither is a fatal error.

+

+	return err == syscall.ERROR_FILE_NOT_FOUND || err == error_pipe_busy

+}

+

+// newOverlapped creates a structure used to track asynchronous

+// I/O requests that have been issued.

+func newOverlapped() (*syscall.Overlapped, error) {

+	event, err := createEvent(nil, true, true, nil)

+	if err != nil {

+		return nil, err

+	}

+	return &syscall.Overlapped{HEvent: event}, nil

+}

+

+// waitForCompletion waits for an asynchronous I/O request referred to by overlapped to complete.

+// This function returns the number of bytes transferred by the operation and an error code if

+// applicable (nil otherwise).

+func waitForCompletion(handle syscall.Handle, overlapped *syscall.Overlapped) (uint32, error) {

+	_, err := syscall.WaitForSingleObject(overlapped.HEvent, syscall.INFINITE)

+	if err != nil {

+		return 0, err

+	}

+	var transferred uint32

+	err = getOverlappedResult(handle, overlapped, &transferred, true)

+	return transferred, err

+}

+

+// dial is a helper to initiate a connection to a named pipe that has been started by a server.

+// The timeout is only enforced if the pipe server has already created the pipe, otherwise

+// this function will return immediately.

+func dial(address string, timeout uint32) (*PipeConn, error) {

+	name, err := syscall.UTF16PtrFromString(string(address))

+	if err != nil {

+		return nil, err

+	}

+	// If at least one instance of the pipe has been created, this function

+	// will wait timeout milliseconds for it to become available.

+	// It will return immediately regardless of timeout, if no instances

+	// of the named pipe have been created yet.

+	// If this returns with no error, there is a pipe available.

+	if err := waitNamedPipe(name, timeout); err != nil {

+		if err == error_bad_pathname {

+			// badly formatted pipe name

+			return nil, badAddr(address)

+		}

+		return nil, err

+	}

+	pathp, err := syscall.UTF16PtrFromString(address)

+	if err != nil {

+		return nil, err

+	}

+	handle, err := syscall.CreateFile(pathp, syscall.GENERIC_READ|syscall.GENERIC_WRITE,

+		uint32(syscall.FILE_SHARE_READ|syscall.FILE_SHARE_WRITE), nil, syscall.OPEN_EXISTING,

+		syscall.FILE_FLAG_OVERLAPPED, 0)

+	if err != nil {

+		return nil, err

+	}

+	return &PipeConn{handle: handle, addr: PipeAddr(address)}, nil

+}

+

+// Listen returns a new PipeListener that will listen on a pipe with the given

+// address. The address must be of the form \\.\pipe\<name>

+//

+// Listen will return a PipeError for an incorrectly formatted pipe name.

+func Listen(address string) (*PipeListener, error) {

+	handle, err := createPipe(address, true)

+	if err == error_invalid_name {

+		return nil, badAddr(address)

+	}

+	if err != nil {

+		return nil, err

+	}

+	return &PipeListener{

+		addr:   PipeAddr(address),

+		handle: handle,

+	}, nil

+}

+

+// PipeListener is a named pipe listener. Clients should typically

+// use variables of type net.Listener instead of assuming named pipe.

+type PipeListener struct {

+	addr   PipeAddr

+	handle syscall.Handle

+	closed bool

+

+	// acceptHandle contains the current handle waiting for

+	// an incoming connection or nil.

+	acceptHandle syscall.Handle

+	// acceptOverlapped is set before waiting on a connection.

+	// If not waiting, it is nil.

+	acceptOverlapped *syscall.Overlapped

+	// acceptMutex protects the handle and overlapped structure.

+	acceptMutex sync.Mutex

+}

+

+// Accept implements the Accept method in the net.Listener interface; it

+// waits for the next call and returns a generic net.Conn.

+func (l *PipeListener) Accept() (net.Conn, error) {

+	c, err := l.AcceptPipe()

+	for err == error_no_data {

+		// Ignore clients that connect and immediately disconnect.

+		c, err = l.AcceptPipe()

+	}

+	if err != nil {

+		return nil, err

+	}

+	return c, nil

+}

+

+// AcceptPipe accepts the next incoming call and returns the new connection.

+// It might return an error if a client connected and immediately cancelled

+// the connection.

+func (l *PipeListener) AcceptPipe() (*PipeConn, error) {

+	if l == nil || l.addr == "" || l.closed {

+		return nil, syscall.EINVAL

+	}

+

+	// the first time we call accept, the handle will have been created by the Listen

+	// call. This is to prevent race conditions where the client thinks the server

+	// isn't listening because it hasn't actually called create yet. After the first time, we'll

+	// have to create a new handle each time

+	handle := l.handle

+	if handle == 0 {

+		var err error

+		handle, err = createPipe(string(l.addr), false)

+		if err != nil {

+			return nil, err

+		}

+	} else {

+		l.handle = 0

+	}

+

+	overlapped, err := newOverlapped()

+	if err != nil {

+		return nil, err

+	}

+	defer syscall.CloseHandle(overlapped.HEvent)

+	if err := connectNamedPipe(handle, overlapped); err != nil && err != error_pipe_connected {

+		if err == error_io_incomplete || err == syscall.ERROR_IO_PENDING {

+			l.acceptMutex.Lock()

+			l.acceptOverlapped = overlapped

+			l.acceptHandle = handle

+			l.acceptMutex.Unlock()

+			defer func() {

+				l.acceptMutex.Lock()

+				l.acceptOverlapped = nil

+				l.acceptHandle = 0

+				l.acceptMutex.Unlock()

+			}()

+

+			_, err = waitForCompletion(handle, overlapped)

+		}

+		if err == syscall.ERROR_OPERATION_ABORTED {

+			// Return error compatible to net.Listener.Accept() in case the

+			// listener was closed.

+			return nil, ErrClosed

+		}

+		if err != nil {

+			return nil, err

+		}

+	}

+	return &PipeConn{handle: handle, addr: l.addr}, nil

+}

+

+// Close stops listening on the address.

+// Already Accepted connections are not closed.

+func (l *PipeListener) Close() error {

+	if l.closed {

+		return nil

+	}

+	l.closed = true

+	if l.handle != 0 {

+		err := disconnectNamedPipe(l.handle)

+		if err != nil {

+			return err

+		}

+		err = syscall.CloseHandle(l.handle)

+		if err != nil {

+			return err

+		}

+		l.handle = 0

+	}

+	l.acceptMutex.Lock()

+	defer l.acceptMutex.Unlock()

+	if l.acceptOverlapped != nil && l.acceptHandle != 0 {

+		// Cancel the pending IO. This call does not block, so it is safe

+		// to hold onto the mutex above.

+		if err := cancelIoEx(l.acceptHandle, l.acceptOverlapped); err != nil {

+			return err

+		}

+		err := syscall.CloseHandle(l.acceptOverlapped.HEvent)

+		if err != nil {

+			return err

+		}

+		l.acceptOverlapped.HEvent = 0

+		err = syscall.CloseHandle(l.acceptHandle)

+		if err != nil {

+			return err

+		}

+		l.acceptHandle = 0

+	}

+	return nil

+}

+

+// Addr returns the listener's network address, a PipeAddr.

+func (l *PipeListener) Addr() net.Addr { return l.addr }

+

+// PipeConn is the implementation of the net.Conn interface for named pipe connections.

+type PipeConn struct {

+	handle syscall.Handle

+	addr   PipeAddr

+

+	// these aren't actually used yet

+	readDeadline  *time.Time

+	writeDeadline *time.Time

+}

+

+type iodata struct {

+	n   uint32

+	err error

+}

+

+// completeRequest looks at iodata to see if a request is pending. If so, it waits for it to either complete or to

+// abort due to hitting the specified deadline. Deadline may be set to nil to wait forever. If no request is pending,

+// the content of iodata is returned.

+func (c *PipeConn) completeRequest(data iodata, deadline *time.Time, overlapped *syscall.Overlapped) (int, error) {

+	if data.err == error_io_incomplete || data.err == syscall.ERROR_IO_PENDING {

+		var timer <-chan time.Time

+		if deadline != nil {

+			if timeDiff := deadline.Sub(time.Now()); timeDiff > 0 {

+				timer = time.After(timeDiff)

+			}

+		}

+		done := make(chan iodata)

+		go func() {

+			n, err := waitForCompletion(c.handle, overlapped)

+			done <- iodata{n, err}

+		}()

+		select {

+		case data = <-done:

+		case <-timer:

+			syscall.CancelIoEx(c.handle, overlapped)

+			data = iodata{0, timeout(c.addr.String())}

+		}

+	}

+	// Windows will produce ERROR_BROKEN_PIPE upon closing

+	// a handle on the other end of a connection. Go RPC

+	// expects an io.EOF error in this case.

+	if data.err == syscall.ERROR_BROKEN_PIPE {

+		data.err = io.EOF

+	}

+	return int(data.n), data.err

+}

+

+// Read implements the net.Conn Read method.

+func (c *PipeConn) Read(b []byte) (int, error) {

+	// Use ReadFile() rather than Read() because the latter

+	// contains a workaround that eats ERROR_BROKEN_PIPE.

+	overlapped, err := newOverlapped()

+	if err != nil {

+		return 0, err

+	}

+	defer syscall.CloseHandle(overlapped.HEvent)

+	var n uint32

+	err = syscall.ReadFile(c.handle, b, &n, overlapped)

+	return c.completeRequest(iodata{n, err}, c.readDeadline, overlapped)

+}

+

+// Write implements the net.Conn Write method.

+func (c *PipeConn) Write(b []byte) (int, error) {

+	overlapped, err := newOverlapped()

+	if err != nil {

+		return 0, err

+	}

+	defer syscall.CloseHandle(overlapped.HEvent)

+	var n uint32

+	err = syscall.WriteFile(c.handle, b, &n, overlapped)

+	return c.completeRequest(iodata{n, err}, c.writeDeadline, overlapped)

+}

+

+// Close closes the connection.

+func (c *PipeConn) Close() error {

+	return syscall.CloseHandle(c.handle)

+}

+

+// LocalAddr returns the local network address.

+func (c *PipeConn) LocalAddr() net.Addr {

+	return c.addr

+}

+

+// RemoteAddr returns the remote network address.

+func (c *PipeConn) RemoteAddr() net.Addr {

+	// not sure what to do here, we don't have remote addr....

+	return c.addr

+}

+

+// SetDeadline implements the net.Conn SetDeadline method.

+// Note that timeouts are only supported on Windows Vista/Server 2008 and above

+func (c *PipeConn) SetDeadline(t time.Time) error {

+	c.SetReadDeadline(t)

+	c.SetWriteDeadline(t)

+	return nil

+}

+

+// SetReadDeadline implements the net.Conn SetReadDeadline method.

+// Note that timeouts are only supported on Windows Vista/Server 2008 and above

+func (c *PipeConn) SetReadDeadline(t time.Time) error {

+	c.readDeadline = &t

+	return nil

+}

+

+// SetWriteDeadline implements the net.Conn SetWriteDeadline method.

+// Note that timeouts are only supported on Windows Vista/Server 2008 and above

+func (c *PipeConn) SetWriteDeadline(t time.Time) error {

+	c.writeDeadline = &t

+	return nil

+}

+

+// PipeAddr represents the address of a named pipe.

+type PipeAddr string

+

+// Network returns the address's network name, "pipe".

+func (a PipeAddr) Network() string { return "pipe" }

+

+// String returns the address of the pipe

+func (a PipeAddr) String() string {

+	return string(a)

+}

+

+// createPipe is a helper function to make sure we always create pipes

+// with the same arguments, since subsequent calls to create pipe need

+// to use the same arguments as the first one. If first is set, fail

+// if the pipe already exists.

+func createPipe(address string, first bool) (syscall.Handle, error) {

+	n, err := syscall.UTF16PtrFromString(address)

+	if err != nil {

+		return 0, err

+	}

+	mode := uint32(pipe_access_duplex | syscall.FILE_FLAG_OVERLAPPED)

+	if first {

+		mode |= file_flag_first_pipe_instance

+	}

+	return createNamedPipe(n,

+		mode,

+		pipe_type_byte,

+		pipe_unlimited_instances,

+		512, 512, 0, nil)

+}

+

+func badAddr(addr string) PipeError {

+	return PipeError{fmt.Sprintf("Invalid pipe address '%s'.", addr), false}

+}

+func timeout(addr string) PipeError {

+	return PipeError{fmt.Sprintf("Pipe IO timed out waiting for '%s'", addr), true}

+}

+

+

+

+func newIpcClient(cfg IpcConfig, codec codec.Codec) (*ipcClient, error) {

+    c, err := Dial(cfg.Endpoint)

+	if err != nil {

+		return nil, err

+	}

+

+	return &ipcClient{codec.New(c)}, nil

+}

+

+func startIpc(cfg IpcConfig, codec codec.Codec, api api.Ethereum) error {

+    os.Remove(cfg.Endpoint) // in case it still exists from a previous run

+    

+    l, err := Listen(cfg.Endpoint)

+    if err != nil {

+        return err

+    }

+    os.Chmod(cfg.Endpoint, 0600)

+	

+    go func() {

+		for {

+			conn, err := l.Accept()

+			if err != nil {

+				glog.V(logger.Error).Infof("Error accepting ipc connection - %v\n", err)

+				continue

+			}

+

+			go func(conn net.Conn) {

+				codec := codec.New(conn)

+

+				for {

+					req, err := codec.ReadRequest()

+					if err == io.EOF {

+						codec.Close()

+						return

+					} else if err != nil {

+						glog.V(logger.Error).Infof("IPC recv err - %v\n", err)

+						codec.Close()

+						return

+					}

+

+					var rpcResponse interface{}

+					res, err := api.Execute(req)

+

+					rpcResponse = shared.NewRpcResponse(req.Id, req.Jsonrpc, res, err)

+					err = codec.WriteResponse(rpcResponse)

+					if err != nil {

+						glog.V(logger.Error).Infof("IPC send err - %v\n", err)

+						codec.Close()

+						return

+					}

+				}

+			}(conn)

+		}

+    }()

+    

+    glog.V(logger.Info).Infof("IPC service started (%s)\n", cfg.Endpoint)

+    

+    return nil

+}