CPS-2146 Using Java Streams to reduce memory consumption in CPS and NCMP

References

CPS-2146 - Getting issue details... STATUS

Purpose

The long-term goal of this Study & Implementation Proposal is gain precise control over memory consumption in CPS and NCMP, going from linear O(N) to constant O(1) space complexity.

The immediate objective is to fix Out Of Memory Errors encountered in NCMP while performing CM handles searches.

Summary of Problem

CPS and NCMP receive queries (CPS path queries and NCMP CM handle queries respectively) for which it will not be known how much data will be returned. Since internal and external APIs return Collections, these Collections containing arbitrary amounts of data are being held in memory. Additionally, these Collections undergo many transformations. Each collection cannot be garbage collected until fully processed/transformed.

In many cases, such as CM handle search, NCMP may be viewed as simply transforming lists of data. Here is an example flow in NCMP showing getting all CM handle IDs, GET http://ncmp-1/ncmpInventory/v1/ch/cmHandles?dmi-plugin-identifier=http://dmi-1

Proposed Solution

It is proposed to create an end-to-end streaming solution, from Persistence layer to Controller. A Proof of Concept will be constructed to document challenges and investigate performance characteristics.

An important observation as to why this solution will achieve O(1) constant space complexity is that the streams will not be terminated until they leave the Rest controller.

Streaming all the way

This will require adding Stream versions of CPS Core read operations, e.g.

Stream<DataNode> queryDataNodesAsStream(String dataspaceName, String anchorName, String cpsPath, FetchDescendantsOption fetchDescendantsOption);

This Stream will implement pagination when fetching data from the FragmentRepository. (This may be implemented in a variety of ways, using Spring JpaRepository Pageable interface, or alternately Spring Data supports streaming from a repository directly, but this needs to be investigated for suitability.) The below example shows the Stream<FragmentEntity> using pagination internally to control memory usage:

Here is some source code showing how the streams API would be used:

// In CPS core
Stream<DataNode> queryDataNodesAsStream(String dataspaceName, String anchorName, String cpsPath, FetchDescendantsOption fetchDescendantsOption) {
	return fragmentRepository.streamByAnchorAndCpsPath(getAnchor(dataspaceName, anchorName), cpsPath)
			.map(fragment -> fetchDescendants(fragment, fetchDescendantsOption))
			.map(fragment -> convertToDataNode(fragment));
}

// In NCMP
private YangModelCmHandle getAnyReadyCmHandleByModuleSetTag(final String moduleSetTag) {
    return cmHandleQueries.queryNcmpRegistryByCpsPath("/dmi-registry/cm-handles[@module-set-tag='" + moduleSetTag + "']", DIRECT_CHILDREN_ONLY)
			.map(YangDataConverter::convertCmHandleToYangModel)
            .filter(cmHandle -> cmHandle.getCompositeState().getCmHandleState() == CmHandleState.READY)
            .findFirst()
            .orElse(null);
}

CPS and NCMP Rest APIs

Instead of returning Collections from Rest APIs, a Stream may be returned, reducing memory pressure on the server.

Current Rest APIs return ResponseEntity using Lists. This means the whole structure must be held in memory before returning response.

    @Override
    public ResponseEntity<List<String>> searchCmHandleIds(final CmHandleQueryParameters cmHandleQueryParameters) {
        final CmHandleQueryServiceParameters cmHandleQueryServiceParameters = ncmpRestInputMapper.toCmHandleQueryServiceParameters(cmHandleQueryParameters);
        final Collection<String> cmHandleIds = networkCmProxyDataService.executeCmHandleIdSearchForInventory(cmHandleQueryServiceParameters);
        return ResponseEntity.ok(List.copyOf(cmHandleIds));
    }

It is proposed to return a Stream instead - Spring Boot supports this, and will allow returning very large results without incurring memory penalty.

    @Override
    public Stream<String> searchCmHandleIds(final CmHandleQueryParameters cmHandleQueryParameters) {
        final CmHandleQueryServiceParameters cmHandleQueryServiceParameters = ncmpRestInputMapper.toCmHandleQueryServiceParameters(cmHandleQueryParameters);
        return networkCmProxyDataService.executeCmHandleIdSearchForInventory(cmHandleQueryServiceParameters);
    }

Additional details of current memory consumption - data conversions

The read APIs in CPS Core (cps-service and cps-ri) return Collection<DataNode>:

Collection<DataNode> queryDataNodes(String dataspaceName, String anchorName, String cpsPath, FetchDescendantsOption fetchDescendantsOption);
Collection<DataNode> getDataNodes(String dataspaceName, String anchorName, String xpath, FetchDescendantsOption fetchDescendantsOption);
Collection<DataNode> getDataNodesForMultipleXpaths(String dataspaceName, String anchorName, Collection<String> xpaths, FetchDescendantsOption fetchDescendantsOption);

Additionally, internal APIs in CPS Reference Implementation (cps-ri) use List<FragmentEntity>, e.g.

List<FragmentEntity> findByAnchorAndCpsPath(AnchorEntity anchorEntity, CpsPathQuery cpsPathQuery);

When a CPS path query is run, this will result in a List<FragmentEntity> which needs to be converted to a Collection<DataNode>. Thus, the Fragment Entities cannot be garbage collected until the list is converted to Data Nodes. This doubles the memory usage.

Additionally, NCMP uses CPS path queries, e.g. to find CM handles in a given state. NCMP will then convert Collection<DataNode> to Collection<YangModelCmHandle>. Again, the Collection<DataNode> cannot be garbage collected until fully converted to YangModelCmHandles. This again results in doubling of memory usage.

Similar applies when converting to NcmpServiceCmHandle.

NCMP also contains many queries where only partial results are needed, making a Streams API ideal.

Additionally, all Rest APIs returning query results return Lists. Spring framework allows returning Streams, eliminating memory overhead.