1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
|
package graceful
import (
"crypto/tls"
"log"
"net"
"net/http"
"os"
"os/signal"
"sync"
"syscall"
"time"
)
// Server wraps an http.Server with graceful connection handling.
// It may be used directly in the same way as http.Server, or may
// be constructed with the global functions in this package.
//
// Example:
// srv := &graceful.Server{
// Timeout: 5 * time.Second,
// Server: &http.Server{Addr: ":1234", Handler: handler},
// }
// srv.ListenAndServe()
type Server struct {
*http.Server
// Timeout is the duration to allow outstanding requests to survive
// before forcefully terminating them.
Timeout time.Duration
// Limit the number of outstanding requests
ListenLimit int
// TCPKeepAlive sets the TCP keep-alive timeouts on accepted
// connections. It prunes dead TCP connections ( e.g. closing
// laptop mid-download)
TCPKeepAlive time.Duration
// ConnState specifies an optional callback function that is
// called when a client connection changes state. This is a proxy
// to the underlying http.Server's ConnState, and the original
// must not be set directly.
ConnState func(net.Conn, http.ConnState)
// BeforeShutdown is an optional callback function that is called
// before the listener is closed. Returns true if shutdown is allowed
BeforeShutdown func() bool
// ShutdownInitiated is an optional callback function that is called
// when shutdown is initiated. It can be used to notify the client
// side of long lived connections (e.g. websockets) to reconnect.
ShutdownInitiated func()
// NoSignalHandling prevents graceful from automatically shutting down
// on SIGINT and SIGTERM. If set to true, you must shut down the server
// manually with Stop().
NoSignalHandling bool
// Logger used to notify of errors on startup and on stop.
Logger *log.Logger
// LogFunc can be assigned with a logging function of your choice, allowing
// you to use whatever logging approach you would like
LogFunc func(format string, args ...interface{})
// Interrupted is true if the server is handling a SIGINT or SIGTERM
// signal and is thus shutting down.
Interrupted bool
// interrupt signals the listener to stop serving connections,
// and the server to shut down.
interrupt chan os.Signal
// stopLock is used to protect against concurrent calls to Stop
stopLock sync.Mutex
// stopChan is the channel on which callers may block while waiting for
// the server to stop.
stopChan chan struct{}
// chanLock is used to protect access to the various channel constructors.
chanLock sync.RWMutex
// connections holds all connections managed by graceful
connections map[net.Conn]struct{}
// idleConnections holds all idle connections managed by graceful
idleConnections map[net.Conn]struct{}
}
// Run serves the http.Handler with graceful shutdown enabled.
//
// timeout is the duration to wait until killing active requests and stopping the server.
// If timeout is 0, the server never times out. It waits for all active requests to finish.
func Run(addr string, timeout time.Duration, n http.Handler) {
srv := &Server{
Timeout: timeout,
TCPKeepAlive: 3 * time.Minute,
Server: &http.Server{Addr: addr, Handler: n},
// Logger: DefaultLogger(),
}
if err := srv.ListenAndServe(); err != nil {
if opErr, ok := err.(*net.OpError); !ok || (ok && opErr.Op != "accept") {
srv.logf("%s", err)
os.Exit(1)
}
}
}
// RunWithErr is an alternative version of Run function which can return error.
//
// Unlike Run this version will not exit the program if an error is encountered but will
// return it instead.
func RunWithErr(addr string, timeout time.Duration, n http.Handler) error {
srv := &Server{
Timeout: timeout,
TCPKeepAlive: 3 * time.Minute,
Server: &http.Server{Addr: addr, Handler: n},
Logger: DefaultLogger(),
}
return srv.ListenAndServe()
}
// ListenAndServe is equivalent to http.Server.ListenAndServe with graceful shutdown enabled.
//
// timeout is the duration to wait until killing active requests and stopping the server.
// If timeout is 0, the server never times out. It waits for all active requests to finish.
func ListenAndServe(server *http.Server, timeout time.Duration) error {
srv := &Server{Timeout: timeout, Server: server, Logger: DefaultLogger()}
return srv.ListenAndServe()
}
// ListenAndServe is equivalent to http.Server.ListenAndServe with graceful shutdown enabled.
func (srv *Server) ListenAndServe() error {
// Create the listener so we can control their lifetime
addr := srv.Addr
if addr == "" {
addr = ":http"
}
conn, err := srv.newTCPListener(addr)
if err != nil {
return err
}
return srv.Serve(conn)
}
// ListenAndServeTLS is equivalent to http.Server.ListenAndServeTLS with graceful shutdown enabled.
//
// timeout is the duration to wait until killing active requests and stopping the server.
// If timeout is 0, the server never times out. It waits for all active requests to finish.
func ListenAndServeTLS(server *http.Server, certFile, keyFile string, timeout time.Duration) error {
srv := &Server{Timeout: timeout, Server: server, Logger: DefaultLogger()}
return srv.ListenAndServeTLS(certFile, keyFile)
}
// ListenTLS is a convenience method that creates an https listener using the
// provided cert and key files. Use this method if you need access to the
// listener object directly. When ready, pass it to the Serve method.
func (srv *Server) ListenTLS(certFile, keyFile string) (net.Listener, error) {
// Create the listener ourselves so we can control its lifetime
addr := srv.Addr
if addr == "" {
addr = ":https"
}
config := &tls.Config{}
if srv.TLSConfig != nil {
*config = *srv.TLSConfig
}
var err error
config.Certificates = make([]tls.Certificate, 1)
config.Certificates[0], err = tls.LoadX509KeyPair(certFile, keyFile)
if err != nil {
return nil, err
}
// Enable http2
enableHTTP2ForTLSConfig(config)
conn, err := srv.newTCPListener(addr)
if err != nil {
return nil, err
}
srv.TLSConfig = config
tlsListener := tls.NewListener(conn, config)
return tlsListener, nil
}
// Enable HTTP2ForTLSConfig explicitly enables http/2 for a TLS Config. This is due to changes in Go 1.7 where
// http servers are no longer automatically configured to enable http/2 if the server's TLSConfig is set.
// See https://github.com/golang/go/issues/15908
func enableHTTP2ForTLSConfig(t *tls.Config) {
if TLSConfigHasHTTP2Enabled(t) {
return
}
t.NextProtos = append(t.NextProtos, "h2")
}
// TLSConfigHasHTTP2Enabled checks to see if a given TLS Config has http2 enabled.
func TLSConfigHasHTTP2Enabled(t *tls.Config) bool {
for _, value := range t.NextProtos {
if value == "h2" {
return true
}
}
return false
}
// ListenAndServeTLS is equivalent to http.Server.ListenAndServeTLS with graceful shutdown enabled.
func (srv *Server) ListenAndServeTLS(certFile, keyFile string) error {
l, err := srv.ListenTLS(certFile, keyFile)
if err != nil {
return err
}
return srv.Serve(l)
}
// ListenAndServeTLSConfig can be used with an existing TLS config and is equivalent to
// http.Server.ListenAndServeTLS with graceful shutdown enabled,
func (srv *Server) ListenAndServeTLSConfig(config *tls.Config) error {
addr := srv.Addr
if addr == "" {
addr = ":https"
}
conn, err := srv.newTCPListener(addr)
if err != nil {
return err
}
srv.TLSConfig = config
tlsListener := tls.NewListener(conn, config)
return srv.Serve(tlsListener)
}
// Serve is equivalent to http.Server.Serve with graceful shutdown enabled.
//
// timeout is the duration to wait until killing active requests and stopping the server.
// If timeout is 0, the server never times out. It waits for all active requests to finish.
func Serve(server *http.Server, l net.Listener, timeout time.Duration) error {
srv := &Server{Timeout: timeout, Server: server, Logger: DefaultLogger()}
return srv.Serve(l)
}
// Serve is equivalent to http.Server.Serve with graceful shutdown enabled.
func (srv *Server) Serve(listener net.Listener) error {
if srv.ListenLimit != 0 {
listener = LimitListener(listener, srv.ListenLimit)
}
// Make our stopchan
srv.StopChan()
// Track connection state
add := make(chan net.Conn)
idle := make(chan net.Conn)
active := make(chan net.Conn)
remove := make(chan net.Conn)
srv.Server.ConnState = func(conn net.Conn, state http.ConnState) {
switch state {
case http.StateNew:
add <- conn
case http.StateActive:
active <- conn
case http.StateIdle:
idle <- conn
case http.StateClosed, http.StateHijacked:
remove <- conn
}
srv.stopLock.Lock()
defer srv.stopLock.Unlock()
if srv.ConnState != nil {
srv.ConnState(conn, state)
}
}
// Manage open connections
shutdown := make(chan chan struct{})
kill := make(chan struct{})
go srv.manageConnections(add, idle, active, remove, shutdown, kill)
interrupt := srv.interruptChan()
// Set up the interrupt handler
if !srv.NoSignalHandling {
signal.Notify(interrupt, syscall.SIGINT, syscall.SIGTERM)
}
quitting := make(chan struct{})
go srv.handleInterrupt(interrupt, quitting, listener)
// Serve with graceful listener.
// Execution blocks here until listener.Close() is called, above.
err := srv.Server.Serve(listener)
if err != nil {
// If the underlying listening is closed, Serve returns an error
// complaining about listening on a closed socket. This is expected, so
// let's ignore the error if we are the ones who explicitly closed the
// socket.
select {
case <-quitting:
err = nil
default:
}
}
srv.shutdown(shutdown, kill)
return err
}
// Stop instructs the type to halt operations and close
// the stop channel when it is finished.
//
// timeout is grace period for which to wait before shutting
// down the server. The timeout value passed here will override the
// timeout given when constructing the server, as this is an explicit
// command to stop the server.
func (srv *Server) Stop(timeout time.Duration) {
srv.stopLock.Lock()
defer srv.stopLock.Unlock()
srv.Timeout = timeout
interrupt := srv.interruptChan()
interrupt <- syscall.SIGINT
}
// StopChan gets the stop channel which will block until
// stopping has completed, at which point it is closed.
// Callers should never close the stop channel.
func (srv *Server) StopChan() <-chan struct{} {
srv.chanLock.Lock()
defer srv.chanLock.Unlock()
if srv.stopChan == nil {
srv.stopChan = make(chan struct{})
}
return srv.stopChan
}
// DefaultLogger returns the logger used by Run, RunWithErr, ListenAndServe, ListenAndServeTLS and Serve.
// The logger outputs to STDERR by default.
func DefaultLogger() *log.Logger {
return log.New(os.Stderr, "[graceful] ", 0)
}
func (srv *Server) manageConnections(add, idle, active, remove chan net.Conn, shutdown chan chan struct{}, kill chan struct{}) {
var done chan struct{}
srv.connections = map[net.Conn]struct{}{}
srv.idleConnections = map[net.Conn]struct{}{}
for {
select {
case conn := <-add:
srv.connections[conn] = struct{}{}
case conn := <-idle:
srv.idleConnections[conn] = struct{}{}
case conn := <-active:
delete(srv.idleConnections, conn)
case conn := <-remove:
delete(srv.connections, conn)
delete(srv.idleConnections, conn)
if done != nil && len(srv.connections) == 0 {
done <- struct{}{}
return
}
case done = <-shutdown:
if len(srv.connections) == 0 && len(srv.idleConnections) == 0 {
done <- struct{}{}
return
}
// a shutdown request has been received. if we have open idle
// connections, we must close all of them now. this prevents idle
// connections from holding the server open while waiting for them to
// hit their idle timeout.
for k := range srv.idleConnections {
if err := k.Close(); err != nil {
srv.logf("[ERROR] %s", err)
}
}
case <-kill:
srv.stopLock.Lock()
defer srv.stopLock.Unlock()
srv.Server.ConnState = nil
for k := range srv.connections {
if err := k.Close(); err != nil {
srv.logf("[ERROR] %s", err)
}
}
return
}
}
}
func (srv *Server) interruptChan() chan os.Signal {
srv.chanLock.Lock()
defer srv.chanLock.Unlock()
if srv.interrupt == nil {
srv.interrupt = make(chan os.Signal, 1)
}
return srv.interrupt
}
func (srv *Server) handleInterrupt(interrupt chan os.Signal, quitting chan struct{}, listener net.Listener) {
for _ = range interrupt {
if srv.Interrupted {
srv.logf("already shutting down")
continue
}
srv.logf("shutdown initiated")
srv.Interrupted = true
if srv.BeforeShutdown != nil {
if !srv.BeforeShutdown() {
srv.Interrupted = false
continue
}
}
close(quitting)
srv.SetKeepAlivesEnabled(false)
if err := listener.Close(); err != nil {
srv.logf("[ERROR] %s", err)
}
if srv.ShutdownInitiated != nil {
srv.ShutdownInitiated()
}
}
}
func (srv *Server) logf(format string, args ...interface{}) {
if srv.LogFunc != nil {
srv.LogFunc(format, args...)
} else if srv.Logger != nil {
srv.Logger.Printf(format, args...)
}
}
func (srv *Server) shutdown(shutdown chan chan struct{}, kill chan struct{}) {
// Request done notification
done := make(chan struct{})
shutdown <- done
if srv.Timeout > 0 {
select {
case <-done:
case <-time.After(srv.Timeout):
close(kill)
}
} else {
<-done
}
// Close the stopChan to wake up any blocked goroutines.
srv.chanLock.Lock()
if srv.stopChan != nil {
close(srv.stopChan)
}
srv.chanLock.Unlock()
}
func (srv *Server) newTCPListener(addr string) (net.Listener, error) {
conn, err := net.Listen("tcp", addr)
if err != nil {
return conn, err
}
if srv.TCPKeepAlive != 0 {
conn = keepAliveListener{conn, srv.TCPKeepAlive}
}
return conn, nil
}
|