0ct0pu5/moby
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Merge pull request #18241 from LK4D4/remove_listenbuffer

Browse source 
Remove listenbuffer
This commit is contained in:
Phil Estes 2015-11-30 14:37:14 -05:00
commit 2a354790a6
8 changed files with 8 additions and 175 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
api/server/server.go
View file

@ -39,7 +39,6 @@ type Config struct {
// Server contains instance details for the server
type Server struct {
cfg *Config
start chan struct{}
servers []*HTTPServer
routers []router.Router
}
@ -54,8 +53,7 @@ type Addr struct {
// It allocates resources which will be needed for ServeAPI(ports, unix-sockets).
func New(cfg *Config) (*Server, error) {
s := &Server{
cfg: cfg,
start: make(chan struct{}),
cfg: cfg,
}
for _, addr := range cfg.Addrs {
srv, err := s.newServer(addr.Proto, addr.Addr)
@ -132,7 +130,7 @@ func (s *Server) initTCPSocket(addr string) (l net.Listener, err error) {
if s.cfg.TLSConfig == nil || s.cfg.TLSConfig.ClientAuth != tls.RequireAndVerifyClientCert {
logrus.Warn("/!\\ DON'T BIND ON ANY IP ADDRESS WITHOUT setting -tlsverify IF YOU DON'T KNOW WHAT YOU'RE DOING /!\\")
}
if l, err = sockets.NewTCPSocket(addr, s.cfg.TLSConfig, s.start); err != nil {
if l, err = sockets.NewTCPSocket(addr, s.cfg.TLSConfig); err != nil {
return nil, err
}
if err := allocateDaemonPort(addr); err != nil {
@ -202,15 +200,3 @@ func (s *Server) CreateMux() *mux.Router {
return m
}
// AcceptConnections allows clients to connect to the API server.
// Referenced Daemon is notified about this server, and waits for the
// daemon acknowledgement before the incoming connections are accepted.
func (s *Server) AcceptConnections() {
// close the lock so the listeners start accepting connections
select {
case <-s.start:
default:
close(s.start)
}
}

2
api/server/server_unix.go
View file

@ -36,7 +36,7 @@ func (s *Server) newServer(proto, addr string) ([]*HTTPServer, error) {
}
ls = append(ls, l)
case "unix":
l, err := sockets.NewUnixSocket(addr, s.cfg.SocketGroup, s.start)
l, err := sockets.NewUnixSocket(addr, s.cfg.SocketGroup)
if err != nil {
return nil, err
}

4
docker/daemon.go
View file

@ -268,10 +268,8 @@ func (cli *DaemonCli) CmdDaemon(args ...string) error {
}
})
// after the daemon is done setting up we can tell the api to start
// accepting connections with specified daemon
// after the daemon is done setting up we can notify systemd api
notifySystem()
api.AcceptConnections()
// Daemon is fully initialized and handling API traffic
// Wait for serve API to complete

27
pkg/listenbuffer/README.md
View file

@ -1,27 +0,0 @@
# listenbuffer
listenbuffer uses the kernel's listening backlog functionality to queue
connections, allowing applications to start listening immediately and handle
connections later. This is signaled by closing the activation channel passed to
the constructor.
The maximum amount of queued connections depends on the configuration of your
kernel (typically called SOMAXXCON) and cannot be configured in Go with the
net package. See `src/net/sock_platform.go` in the Go tree or consult your
kernel's manual.
activator := make(chan struct{})
buffer, err := NewListenBuffer("tcp", "localhost:4000", activator)
if err != nil {
panic(err)
}
// will block until activator has been closed or is sent an event
client, err := buffer.Accept()
Somewhere else in your application once it's been booted:
close(activator)
`buffer.Accept()` will return the first client in the kernel listening queue, or
continue to block until a client connects or an error occurs.

76
pkg/listenbuffer/buffer.go
View file

@ -1,76 +0,0 @@
/*
Package listenbuffer uses the kernel's listening backlog functionality to queue
connections, allowing applications to start listening immediately and handle
connections later. This is signaled by closing the activation channel passed to
the constructor.
The maximum amount of queued connections depends on the configuration of your
kernel (typically called SOMAXXCON) and cannot be configured in Go with the
net package. See `src/net/sock_platform.go` in the Go tree or consult your
kernel's manual.
activator := make(chan struct{})
buffer, err := NewListenBuffer("tcp", "localhost:4000", activator)
if err != nil {
panic(err)
}
// will block until activator has been closed or is sent an event
client, err := buffer.Accept()
Somewhere else in your application once it's been booted:
close(activator)
`buffer.Accept()` will return the first client in the kernel listening queue, or
continue to block until a client connects or an error occurs.
*/
package listenbuffer
import "net"
// NewListenBuffer returns a net.Listener listening on addr with the protocol
// passed. The channel passed is used to activate the listenbuffer when the
// caller is ready to accept connections.
func NewListenBuffer(proto, addr string, activate <-chan struct{}) (net.Listener, error) {
wrapped, err := net.Listen(proto, addr)
if err != nil {
return nil, err
}
return &defaultListener{
wrapped: wrapped,
activate: activate,
}, nil
}
// defaultListener is the buffered wrapper around the net.Listener
type defaultListener struct {
wrapped net.Listener // The net.Listener wrapped by listenbuffer
ready bool // Whether the listenbuffer has been activated
activate <-chan struct{} // Channel to control activation of the listenbuffer
}
// Close closes the wrapped socket.
func (l *defaultListener) Close() error {
return l.wrapped.Close()
}
// Addr returns the listening address of the wrapped socket.
func (l *defaultListener) Addr() net.Addr {
return l.wrapped.Addr()
}
// Accept returns a client connection on the wrapped socket if the listen buffer
// has been activated. To active the listenbuffer the activation channel passed
// to NewListenBuffer must have been closed or sent an event.
func (l *defaultListener) Accept() (net.Conn, error) {
// if the listen has been told it is ready then we can go ahead and
// start returning connections
if l.ready {
return l.wrapped.Accept()
}
<-l.activate
l.ready = true
return l.Accept()
}

41
pkg/listenbuffer/listen_buffer_test.go
View file

@ -1,41 +0,0 @@
package listenbuffer
import (
"io/ioutil"
"net"
"testing"
)
func TestListenBufferAllowsAcceptingWhenActivated(t *testing.T) {
lock := make(chan struct{})
buffer, err := NewListenBuffer("tcp", "", lock)
if err != nil {
t.Fatal("Unable to create listen buffer: ", err)
}
go func() {
conn, err := net.Dial("tcp", buffer.Addr().String())
if err != nil {
t.Fatal("Client failed to establish connection to server: ", err)
}
conn.Write([]byte("ping"))
conn.Close()
}()
close(lock)
client, err := buffer.Accept()
if err != nil {
t.Fatal("Failed to accept client: ", err)
}
response, err := ioutil.ReadAll(client)
if err != nil {
t.Fatal("Failed to read from client: ", err)
}
if string(response) != "ping" {
t.Fatal("Expected to receive ping from client, received: ", string(response))
}
}

8
pkg/sockets/tcp_socket.go
View file

@ -7,17 +7,13 @@ import (
"net"
"net/http"
"time"
"github.com/docker/docker/pkg/listenbuffer"
)
// NewTCPSocket creates a TCP socket listener with the specified address and
// and the specified tls configuration. If TLSConfig is set, will encapsulate the
// TCP listener inside a TLS one.
// The channel passed is used to activate the listenbuffer when the caller is ready
// to accept connections.
func NewTCPSocket(addr string, tlsConfig *tls.Config, activate <-chan struct{}) (net.Listener, error) {
l, err := listenbuffer.NewListenBuffer("tcp", addr, activate)
func NewTCPSocket(addr string, tlsConfig *tls.Config) (net.Listener, error) {
l, err := net.Listen("tcp", addr)
if err != nil {
return nil, err
}

7
pkg/sockets/unix_socket.go
View file

@ -10,20 +10,17 @@ import (
"syscall"
"github.com/Sirupsen/logrus"
"github.com/docker/docker/pkg/listenbuffer"
"github.com/opencontainers/runc/libcontainer/user"
)
// NewUnixSocket creates a unix socket with the specified path and group.
// The channel passed is used to activate the listenbuffer when the caller is ready
// to accept connections.
func NewUnixSocket(path, group string, activate <-chan struct{}) (net.Listener, error) {
func NewUnixSocket(path, group string) (net.Listener, error) {
if err := syscall.Unlink(path); err != nil && !os.IsNotExist(err) {
return nil, err
}
mask := syscall.Umask(0777)
defer syscall.Umask(mask)
l, err := listenbuffer.NewListenBuffer("unix", path, activate)
l, err := net.Listen("unix", path)
if err != nil {
return nil, err
}