FpgaResourceHandlerImpl implements ResourceHandler { @Override public List<PrivilegedOperation> preStart(Container container) throws ResourceHandlerException { List<PrivilegedOperation> ret = new ArrayList<>(); String containerIdStr = container.getContainerId().toString(); Resource requestedResource = container.getResource(); cGroupsHandler.createCGroup(CGroupsHandler.CGroupController.DEVICES, containerIdStr); long deviceCount = requestedResource.getResourceValue(FPGA_URI); LOG.info(containerIdStr + " requested " + deviceCount + " Intel FPGA(s)"); String ipFilePath = null; try { final String requestedIPID = getRequestedIPID(container); String localizedIPIDHash = null; ipFilePath = vendorPlugin.retrieveIPfilePath( requestedIPID, container.getWorkDir(), container.getResourceSet().getLocalizedResources()); if (ipFilePath != null) { try (FileInputStream fis = new FileInputStream(ipFilePath)) { localizedIPIDHash = DigestUtils.sha256Hex(fis); } catch (IOException e) { throw new ResourceHandlerException("Could not calculate SHA-256", e); } } FpgaResourceAllocator.FpgaAllocation allocation = allocator.assignFpga( vendorPlugin.getFpgaType(), deviceCount, container, localizedIPIDHash); LOG.info("FpgaAllocation:" + allocation); PrivilegedOperation privilegedOperation = new PrivilegedOperation(PrivilegedOperation.OperationType.FPGA, Arrays.asList(CONTAINER_ID_CLI_OPTION, containerIdStr)); if (!allocation.getDenied().isEmpty()) { List<Integer> denied = new ArrayList<>(); allocation.getDenied().forEach(device -> denied.add(device.getMinor())); privilegedOperation.appendArgs(Arrays.asList(EXCLUDED_FPGAS_CLI_OPTION, StringUtils.join(",", denied))); } privilegedOperationExecutor.executePrivilegedOperation( privilegedOperation, true); if (deviceCount > 0) { ipFilePath = vendorPlugin.retrieveIPfilePath( getRequestedIPID(container), container.getWorkDir(), container.getResourceSet().getLocalizedResources()); if (ipFilePath == null) { LOG.warn("FPGA plugin failed to downloaded IP, please check the" + " value of environment viable: " + REQUEST_FPGA_IP_ID_KEY + " if you want YARN to program the device"); } else { LOG.info("IP file path:" + ipFilePath); List<FpgaDevice> allowed = allocation.getAllowed(); String majorMinorNumber; for (int i = 0; i < allowed.size(); i++) { FpgaDevice device = allowed.get(i); majorMinorNumber = device.getMajor() + ":" + device.getMinor(); String currentHash = allowed.get(i).getAocxHash(); if (currentHash != null && currentHash.equalsIgnoreCase(localizedIPIDHash)) { LOG.info("IP already in device \"" + allowed.get(i).getAliasDevName() + "," + majorMinorNumber + "\", skip reprogramming"); continue; } if (vendorPlugin.configureIP(ipFilePath, device)) { allocator.updateFpga(containerIdStr, allowed.get(i), requestedIPID, localizedIPIDHash); } } } } } catch (ResourceHandlerException re) { allocator.cleanupAssignFpgas(containerIdStr); cGroupsHandler.deleteCGroup(CGroupsHandler.CGroupController.DEVICES, containerIdStr); throw re; } catch (PrivilegedOperationException e) { allocator.cleanupAssignFpgas(containerIdStr); cGroupsHandler.deleteCGroup(CGroupsHandler.CGroupController.DEVICES, containerIdStr); LOG.warn("Could not update cgroup for container", e); throw new ResourceHandlerException(e); } ret.add(new PrivilegedOperation( PrivilegedOperation.OperationType.ADD_PID_TO_CGROUP, PrivilegedOperation.CGROUP_ARG_PREFIX + cGroupsHandler.getPathForCGroupTasks( CGroupsHandler.CGroupController.DEVICES, containerIdStr))); return ret; } @VisibleForTesting  FpgaResourceHandlerImpl(Context nmContext,
      CGroupsHandler cGroupsHandler,
      PrivilegedOperationExecutor privilegedOperationExecutor,
      AbstractFpgaVendorPlugin plugin,
      FpgaDiscoverer fpgaDiscoverer); String getRequestedIPID(Container container); @Override List<PrivilegedOperation> bootstrap(Configuration configuration); @Override List<PrivilegedOperation> preStart(Container container); @Override List<PrivilegedOperation> reacquireContainer(ContainerId containerId); @Override List<PrivilegedOperation> updateContainer(Container container); @Override List<PrivilegedOperation> postComplete(ContainerId containerId); @Override List<PrivilegedOperation> teardown(); @Override String toString(); static final String EXCLUDED_FPGAS_CLI_OPTION; static final String CONTAINER_ID_CLI_OPTION; }
@Test public void testSha256CalculationFails() throws ResourceHandlerException { expected.expect(ResourceHandlerException.class); expected.expectMessage("Could not calculate SHA-256"); dummyAocx.delete(); fpgaResourceHandler.preStart(mockContainer(0, 1, "GEMM")); }