From 8b581e1ff1d72713f8c84ee1cf0f164f52b0a603 Mon Sep 17 00:00:00 2001
From: Chris Mar <cmar@meraki.com>
Date: Tue, 26 Nov 2013 22:00:53 -0800
Subject: [PATCH] mcmd: close FDs when finished so we don't run out

Close file descriptors (FDs) for the pipes between the MultipleCmd
parent and its children to prevent the parent from exceeding the default
per-process Linux limit of 1024 FDs.

The pipes created for communication between the parent process and each
subprocess are unidirectional. Close the unused ends of the pipes in the
parent and child as soon as a child is spawned.

In the parent, keep track of the other FDs for each subprocesses, and
close the FDs for a subproccess when it's reaped.

MultipleCmd now requires only (3 * maxflight) open FDs instead of
(6 * total number of commands). Given a per-process FD limit of 1024,
the theoretical limit for maxflight is 341, versus the previous limit of
170.
---
 mssh/lib/mcmd.rb | 18 +++++++++++++++---
 1 file changed, 15 insertions(+), 3 deletions(-)

diff --git a/mssh/lib/mcmd.rb b/mssh/lib/mcmd.rb
index 32bbbc0..780f674 100755
--- a/mssh/lib/mcmd.rb
+++ b/mssh/lib/mcmd.rb
@@ -14,6 +14,7 @@ def initialize
     # these are re-initialized after every run
     @subproc_by_pid = Hash.new
     @subproc_by_fd = Hash.new
+    @fds_by_pid = Hash.new
     @processed_commands = []
     # end items which are re-initialized
 
@@ -50,16 +51,19 @@ def add_subprocess(cmd)
     pid = fork
     if not pid.nil?
       # parent
+      # close the ends of the pipes we don't need
+      stdin_rd.close
+      stdout_wr.close
+      stderr_wr.close
       # for mapping to subproc by pid
       subproc.pid = pid
       @subproc_by_pid[pid] = subproc
       # for mapping to subproc by i/o handle (returned from select)
-      @subproc_by_fd[stdin_rd] = subproc
       @subproc_by_fd[stdin_wr] = subproc
       @subproc_by_fd[stdout_rd] = subproc
-      @subproc_by_fd[stdout_wr] = subproc
       @subproc_by_fd[stderr_rd] = subproc
-      @subproc_by_fd[stderr_wr] = subproc
+      # for mapping to parent and child fds by pid (to be closed later)
+      @fds_by_pid[pid] = [stdin_wr, stdout_rd, stderr_rd]
       
       self.yield_startcmd.call(subproc) unless self.yield_startcmd.nil?
     else
@@ -68,6 +72,10 @@ def add_subprocess(cmd)
       STDIN.reopen(stdin_rd)
       STDOUT.reopen(stdout_wr)
       STDERR.reopen(stderr_wr)
+      # close the ends of the pipes we don't need
+      stdin_wr.close
+      stdout_rd.close
+      stderr_rd.close
       noshell_exec(cmd)
       raise "can't be reached!!. exec failed!!"
     end
@@ -246,6 +254,10 @@ def wait
     end
     just_reaped.each do |p|
       self.yield_wait.call(p) unless self.yield_wait.nil?
+      # close the parent and child ends of the pipes for this subproc
+      puts "just reaped subproc #{p.pid}. closing fds #{@fds_by_pid[p.pid].map(&:fileno).join(',')}." if self.debug
+      @fds_by_pid[p.pid].each { |fd| fd.close rescue true }
+      @fds_by_pid.delete(p.pid)
     end
   end