ftp: close idle connections after --ftp-idle-timeout (1m by default)

This fixes a problem where ftp backends live on forever when using
the rc and use more and more connections.

backend/ftp/ftp.go

@@ -93,6 +93,17 @@ to an encrypted one. Cannot be used in combination with implicit FTP.`,
 			Help:     "Disable using MLSD even if server advertises support",
 			Default:  false,
 			Advanced: true,
+		}, {
+			Name:    "idle_timeout",
+			Default: fs.Duration(60 * time.Second),
+			Help: `Max time before closing idle connections
+
+If no connections have been returned to the connection pool in the time
+given, rclone will empty the connection pool.
+
+Set to 0 to keep connections indefinitely.
+`,
+			Advanced: true,
 		}, {
 			Name:     config.ConfigEncoding,
 			Help:     config.ConfigEncodingHelp,
@@ -120,6 +131,7 @@ type Options struct {
 	SkipVerifyTLSCert bool                 `config:"no_check_certificate"`
 	DisableEPSV       bool                 `config:"disable_epsv"`
 	DisableMLSD       bool                 `config:"disable_mlsd"`
+	IdleTimeout       fs.Duration          `config:"idle_timeout"`
 	Enc               encoder.MultiEncoder `config:"encoding"`
 }
 
@@ -136,6 +148,7 @@ type Fs struct {
 	dialAddr string
 	poolMu   sync.Mutex
 	pool     []*ftp.ServerConn
+	drain    *time.Timer // used to drain the pool when we stop using the connections
 	tokens   *pacer.TokenDispenser
 	tlsConf  *tls.Config
 }
@@ -322,6 +335,9 @@ func (f *Fs) putFtpConnection(pc **ftp.ServerConn, err error) {
 	}
 	f.poolMu.Lock()
 	f.pool = append(f.pool, c)
+	if f.opt.IdleTimeout > 0 {
+		f.drain.Reset(time.Duration(f.opt.IdleTimeout)) // nudge on the pool emptying timer
+	}
 	f.poolMu.Unlock()
 }
 
@@ -329,6 +345,12 @@ func (f *Fs) putFtpConnection(pc **ftp.ServerConn, err error) {
 func (f *Fs) drainPool(ctx context.Context) (err error) {
 	f.poolMu.Lock()
 	defer f.poolMu.Unlock()
+	if f.opt.IdleTimeout > 0 {
+		f.drain.Stop()
+	}
+	if len(f.pool) != 0 {
+		fs.Debugf(f, "closing %d unused connections", len(f.pool))
+	}
 	for i, c := range f.pool {
 		if cErr := c.Quit(); cErr != nil {
 			err = cErr
@@ -393,6 +415,10 @@ func NewFs(ctx context.Context, name, root string, m configmap.Mapper) (ff fs.Fs
 	f.features = (&fs.Features{
 		CanHaveEmptyDirectories: true,
 	}).Fill(ctx, f)
+	// set the pool drainer timer going
+	if f.opt.IdleTimeout > 0 {
+		f.drain = time.AfterFunc(time.Duration(opt.IdleTimeout), func() { _ = f.drainPool(ctx) })
+	}
 	// Make a connection and pool it to return errors early
 	c, err := f.getFtpConnection(ctx)
 	if err != nil {