big-data/hadoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

YARN-7688. Miscellaneous Improvements To ProcfsBasedProcessTree. Contributed by BELUGA BEHR.

Browse Source
This commit is contained in:
Miklos Szegedi 2018-01-02 17:02:31 -08:00
parent 4ad39ec308
commit 626b5103d4
2 changed files with 55 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/ProcfsBasedProcessTree.java
View File

@ -20,21 +20,30 @@
import java.io.BufferedReader;
import java.io.File;
import java.io.FileFilter;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.InputStreamReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.math.BigInteger;
import java.nio.charset.Charset;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.commons.io.IOUtils;
import org.apache.commons.io.filefilter.AndFileFilter;
import org.apache.commons.io.filefilter.DirectoryFileFilter;
import org.apache.commons.io.filefilter.RegexFileFilter;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
@ -85,8 +94,9 @@ private MemInfo(String name) {
}
public static MemInfo getMemInfoByName(String name) {
String searchName = StringUtils.trimToNull(name);
for (MemInfo info : MemInfo.values()) {
if (info.name.trim().equalsIgnoreCase(name.trim())) {
if (info.name.trim().equalsIgnoreCase(searchName)) {
return info;
}
}
@ -170,7 +180,7 @@ public static boolean isAvailable() {
return false;
}
} catch (SecurityException se) {
LOG.warn("Failed to get Operating System name. " + se);
LOG.warn("Failed to get Operating System name.", se);
return false;
}
return true;
@ -214,12 +224,12 @@ public void updateProcessTree() {
// Add each process to its parent.
for (Map.Entry<String, ProcessInfo> entry : allProcessInfo.entrySet()) {
String pID = entry.getKey();
if (!pID.equals("1")) {
if (!"1".equals(pID)) {
ProcessInfo pInfo = entry.getValue();
String ppid = pInfo.getPpid();
// If parent is init and process is not session leader,
// attach to sessionID
if (ppid.equals("1")) {
if ("1".equals(ppid)) {
String sid = pInfo.getSessionId().toString();
if (!pID.equals(sid)) {
ppid = sid;
@ -233,8 +243,8 @@ public void updateProcessTree() {
}
// now start constructing the process-tree
LinkedList<ProcessInfo> pInfoQueue = new LinkedList<ProcessInfo>();
pInfoQueue.addAll(me.getChildren());
List<ProcessInfo> children = me.getChildren();
Queue<ProcessInfo> pInfoQueue = new ArrayDeque<ProcessInfo>(children);
while (!pInfoQueue.isEmpty()) {
ProcessInfo pInfo = pInfoQueue.remove();
if (!processTree.containsKey(pInfo.getPid())) {
@ -254,10 +264,8 @@ public void updateProcessTree() {
}
}
if (LOG.isDebugEnabled()) {
// Log.debug the ProcfsBasedProcessTree
LOG.debug(this.toString());
}
LOG.debug(this);
if (smapsEnabled) {
// Update smaps info
processSMAPTree.clear();
@ -296,9 +304,7 @@ public static boolean checkPidPgrpidForMatch(String _pid, String procfs) {
"\t|- %s %s %d %d %s %d %d %d %d %s%n";
public List<String> getCurrentProcessIDs() {
List<String> currentPIDs = new ArrayList<String>();
currentPIDs.addAll(processTree.keySet());
return currentPIDs;
return Collections.unmodifiableList(new ArrayList<>(processTree.keySet()));
}
/**
@ -327,18 +333,17 @@ public String getProcessTreeDump() {
@Override
public long getVirtualMemorySize(int olderThanAge) {
long total = UNAVAILABLE;
long total = 0L;
boolean isAvailable = false;
for (ProcessInfo p : processTree.values()) {
if (p != null) {
if (total == UNAVAILABLE ) {
total = 0;
}
isAvailable = true;
if (p.getAge() > olderThanAge) {
total += p.getVmem();
}
}
}
return total;
return isAvailable ? total : UNAVAILABLE;
}
@Override
@ -352,11 +357,11 @@ public long getRssMemorySize(int olderThanAge) {
boolean isAvailable = false;
long totalPages = 0;
for (ProcessInfo p : processTree.values()) {
if ((p != null) ) {
if (p != null) {
isAvailable = true;
if (p.getAge() > olderThanAge) {
totalPages += p.getRssmemPage();
}
isAvailable = true;
}
}
return isAvailable ? totalPages * PAGE_SIZE : UNAVAILABLE; // convert # pages to byte
@ -405,9 +410,7 @@ private long getSmapBasedRssMemorySize(int olderThanAge) {
}
}
}
if (LOG.isDebugEnabled()) {
LOG.debug(procMemInfo.toString());
}
LOG.debug(procMemInfo);
}
}
}
@ -427,9 +430,9 @@ public long getCumulativeCpuTime() {
boolean isAvailable = false;
for (ProcessInfo p : processTree.values()) {
if (p != null) {
incJiffies += p.getDtime();
// data is available
isAvailable = true;
incJiffies += p.getDtime();
}
}
if (isAvailable) {
@ -481,21 +484,17 @@ private static String getValidPID(String pid) {
* Get the list of all processes in the system.
*/
private List<String> getProcessList() {
List<String> processList = new ArrayList<String>();
String[] processDirs = (new File(procfsDir)).list();
if (processDirs != null) {
for (String dir : processDirs) {
Matcher m = numberPattern.matcher(dir);
if (!m.matches()) {
continue;
}
try {
if ((new File(procfsDir, dir)).isDirectory()) {
processList.add(dir);
}
} catch (SecurityException s) {
// skip this process
}
List<String> processList = Collections.emptyList();
FileFilter procListFileFilter = new AndFileFilter(
DirectoryFileFilter.INSTANCE, new RegexFileFilter(numberPattern));
File dir = new File(procfsDir);
File[] processDirs = dir.listFiles(procListFileFilter);
if (ArrayUtils.isNotEmpty(processDirs)) {
processList = new ArrayList<String>(processDirs.length);
for (File processDir : processDirs) {
processList.add(processDir.getName());
}
}
return processList;
@ -547,7 +546,7 @@ private static ProcessInfo constructProcessInfo(ProcessInfo pinfo,
ret = null;
}
} catch (IOException io) {
LOG.warn("Error reading the stream " + io);
LOG.warn("Error reading the stream", io);
ret = null;
} finally {
// Close the streams
@ -556,10 +555,10 @@ private static ProcessInfo constructProcessInfo(ProcessInfo pinfo,
try {
in.close();
} catch (IOException i) {
LOG.warn("Error closing the stream " + in);
LOG.warn("Error closing the stream", i);
}
} catch (IOException i) {
LOG.warn("Error closing the stream " + fReader);
LOG.warn("Error closing the stream", i);
}
}
@ -729,14 +728,14 @@ public String getCmdLine(String procfsDir) {
ret = "N/A";
} else {
ret = ret.replace('\0', ' '); // Replace each null char with a space
if (ret.equals("")) {
if (ret.isEmpty()) {
// The cmdline might be empty because the process is swapped out or
// is a zombie.
ret = "N/A";
}
}
} catch (IOException io) {
LOG.warn("Error reading the stream " + io);
LOG.warn("Error reading the stream", io);
ret = "N/A";
} finally {
// Close the streams
@ -745,10 +744,10 @@ public String getCmdLine(String procfsDir) {
try {
in.close();
} catch (IOException i) {
LOG.warn("Error closing the stream " + in);
LOG.warn("Error closing the stream", i);
}
} catch (IOException i) {
LOG.warn("Error closing the stream " + fReader);
LOG.warn("Error closing the stream", i);
}
}
@ -805,11 +804,11 @@ private static void constructProcessSMAPInfo(ProcessTreeSmapMemInfo pInfo,
}
}
} catch (FileNotFoundException f) {
LOG.error(f.getMessage());
LOG.error(f);
} catch (IOException e) {
LOG.error(e.getMessage());
LOG.error(e);
} catch (Throwable t) {
LOG.error(t.getMessage());
LOG.error(t);
} finally {
IOUtils.closeQuietly(in);
}
@ -839,7 +838,7 @@ public String toString() {
StringBuilder sb = new StringBuilder();
for (ProcessSmapMemoryInfo info : memoryInfoList) {
sb.append("\n");
sb.append(info.toString());
sb.append(info);
}
return sb.toString();
}

15
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/util/TestProcfsBasedProcessTree.java
View File

@ -118,7 +118,6 @@ public void setup() throws IOException {
}
@Test(timeout = 30000)
@SuppressWarnings("deprecation")
public void testProcessTree() throws Exception {
try {
Assert.assertTrue(ProcfsBasedProcessTree.isAvailable());
@ -163,7 +162,7 @@ public void testProcessTree() throws Exception {
LOG.info("Root process pid: " + pid);
ProcfsBasedProcessTree p = createProcessTree(pid);
p.updateProcessTree(); // initialize
LOG.info("ProcessTree: " + p.toString());
LOG.info("ProcessTree: " + p);
File leaf = new File(lowestDescendant);
// wait till lowest descendant process of Rougue Task starts execution
@ -176,7 +175,7 @@ public void testProcessTree() throws Exception {
}
p.updateProcessTree(); // reconstruct
LOG.info("ProcessTree: " + p.toString());
LOG.info("ProcessTree: " + p);
// Verify the orphaned pid is In process tree
String lostpid = getPidFromPidFile(lostDescendant);
@ -395,7 +394,6 @@ public void createMemoryMappingInfo(ProcessTreeSmapMemInfo[] procMemInfo) {
* files.
*/
@Test(timeout = 30000)
@SuppressWarnings("deprecation")
public void testCpuAndMemoryForProcessTree() throws IOException {
// test processes
@ -908,13 +906,8 @@ public static void setupPidDirs(File procfsRootDir, String[] pids)
throws IOException {
for (String pid : pids) {
File pidDir = new File(procfsRootDir, pid);
pidDir.mkdir();
if (!pidDir.exists()) {
throw new IOException("couldn't make process directory under "
+ "fake procfs");
} else {
LOG.info("created pid dir");
}
FileUtils.forceMkdir(pidDir);
LOG.info("created pid dir: " + pidDir);
}
}