CGroupElasticMemoryController extends Thread { @Override public void run() { ExecutorService executor = null; try { setCGroupParameters(); ProcessBuilder oomListener = new ProcessBuilder(); oomListener.command(oomListenerPath, yarnCGroupPath); synchronized (this) { if (!stopped) { process = oomListener.start(); } else { resetCGroupParameters(); LOG.info("Listener stopped before starting"); return; } } LOG.info(String.format("Listening on %s with %s", yarnCGroupPath, oomListenerPath)); executor = Executors.newFixedThreadPool(2); Future<String> errorListener = executor.submit( () -> IOUtils.toString(process.getErrorStream(), Charset.defaultCharset())); InputStream events = process.getInputStream(); byte[] event = new byte[8]; int read; while ((read = events.read(event)) == event.length) { resolveOOM(executor); } if (read != -1) { LOG.warn(String.format("Characters returned from event hander: %d", read)); } int exitCode = process.waitFor(); String error = errorListener.get(); process = null; LOG.info(String.format("OOM listener exited %d %s", exitCode, error)); } catch (OOMNotResolvedException ex) { throw new YarnRuntimeException("Could not resolve OOM", ex); } catch (Exception ex) { synchronized (this) { if (!stopped) { LOG.warn("OOM Listener exiting.", ex); } } } finally { if (process != null && process.isAlive()) { process.destroyForcibly(); } if (executor != null) { try { executor.awaitTermination(6, TimeUnit.SECONDS); } catch (InterruptedException e) { LOG.warn("Exiting without processing all OOM events."); } executor.shutdown(); } resetCGroupParameters(); } } @VisibleForTesting  CGroupElasticMemoryController(Configuration conf,
                                       Context context,
                                       CGroupsHandler cgroups,
                                       boolean controlPhysicalMemory,
                                       boolean controlVirtualMemory,
                                       long limit,
                                       Runnable oomHandlerOverride);  CGroupElasticMemoryController(Configuration conf,
                                       Context context,
                                       CGroupsHandler cgroups,
                                       boolean controlPhysicalMemory,
                                       boolean controlVirtualMemory,
                                       long limit); synchronized void stopListening(); static boolean isAvailable(); @Override void run();  }
@Test(timeout = 20000) public void testMultipleOOMEvents() throws Exception { conf.set(YarnConfiguration.NM_ELASTIC_MEMORY_CONTROL_OOM_LISTENER_PATH, script.getAbsolutePath()); try { FileUtils.writeStringToFile(script, "#!/bin/bash\nprintf oomevent;printf oomevent;\n", Charset.defaultCharset(), false); assertTrue("Could not set executable", script.setExecutable(true)); CGroupsHandler cgroups = mock(CGroupsHandler.class); when(cgroups.getPathForCGroup(any(), any())).thenReturn(""); when(cgroups.getCGroupParam(any(), any(), any())) .thenReturn("under_oom 0"); Runnable handler = mock(Runnable.class); doNothing().when(handler).run(); CGroupElasticMemoryController controller = new CGroupElasticMemoryController( conf, null, cgroups, true, false, 10000, handler ); controller.run(); verify(handler, times(2)).run(); } finally { assertTrue(String.format("Could not clean up script %s", script.getAbsolutePath()), script.delete()); } }
@Test(timeout = 20000, expected = YarnRuntimeException.class) public void testInfiniteOOM() throws Exception { conf.set(YarnConfiguration.NM_ELASTIC_MEMORY_CONTROL_OOM_LISTENER_PATH, script.getAbsolutePath()); Runnable handler = mock(Runnable.class); try { FileUtils.writeStringToFile(script, "#!/bin/bash\nprintf oomevent;sleep 1000;\n", Charset.defaultCharset(), false); assertTrue("Could not set executable", script.setExecutable(true)); CGroupsHandler cgroups = mock(CGroupsHandler.class); when(cgroups.getPathForCGroup(any(), any())).thenReturn(""); when(cgroups.getCGroupParam(any(), any(), any())) .thenReturn("under_oom 1"); doNothing().when(handler).run(); CGroupElasticMemoryController controller = new CGroupElasticMemoryController( conf, null, cgroups, true, false, 10000, handler ); controller.run(); } finally { verify(handler, times(1)).run(); assertTrue(String.format("Could not clean up script %s", script.getAbsolutePath()), script.delete()); } }
@Test(timeout = 20000, expected = YarnRuntimeException.class) public void testNothingToKill() throws Exception { conf.set(YarnConfiguration.NM_ELASTIC_MEMORY_CONTROL_OOM_LISTENER_PATH, script.getAbsolutePath()); Runnable handler = mock(Runnable.class); try { FileUtils.writeStringToFile(script, "#!/bin/bash\nprintf oomevent;sleep 1000;\n", Charset.defaultCharset(), false); assertTrue("Could not set executable", script.setExecutable(true)); CGroupsHandler cgroups = mock(CGroupsHandler.class); when(cgroups.getPathForCGroup(any(), any())).thenReturn(""); when(cgroups.getCGroupParam(any(), any(), any())) .thenReturn("under_oom 1"); doThrow(new YarnRuntimeException("Expected")).when(handler).run(); CGroupElasticMemoryController controller = new CGroupElasticMemoryController( conf, null, cgroups, true, false, 10000, handler ); controller.run(); } finally { verify(handler, times(1)).run(); assertTrue(String.format("Could not clean up script %s", script.getAbsolutePath()), script.delete()); } }