Snippet: Elements to Thread Allocation Without Mappings

Prerequisites

Below snippets assume data source and data target of the TI process are cubes with common parallel enabled dimensions, other dimensions might differ.

The source pDimPar and equivalent target dimension have to have total rollup containing all leaf elements in their structure, the name of the total rollup root element must be Total <dimension name>.

You should define a process parameter allowing entry of a filter expression for each parallel enabled dimension.

Prolog (after parameter tests)

# Elements to allocate to threads - always will be used even in case when parallelism is not required
If( sParFilter @= '' );
  # MDX that is used by default to drive parallelization set when no filter is specified
  sDimParMDX = Expand( 'TM1FILTERBYLEVEL(DESCENDANTS([%pDimPar%].[Total %pDimPar%]), 0)' );
  sDimClrMDX = Expand( 'TM1FILTERBYLEVEL(DESCENDANTS([%sDimParClr%].[Total %sDimParClr%]), 0)' );
Else;
  # We need to provide MDX out of Bedrock style filter depending on the threadID
  sFind = sParFilter;
  nTokens = 0;
  While( sFind @<> '' );
    nToken = SCAN( cElementDelim, sFind );
    nToken = IF( nToken = 0, LONG( sFind ) + 1, nToken );
    sFind = DELET( sFind, 1, nToken );
    nTokens = nTokens + 1;
  End;
  If( nTokens < nMaxThreads & nThreadID = 0 );
    nMaxThreads = nTokens;
    pMaxThreads = NumberToString( nMaxThreads );
    LogOutput( 'WARN', Expand( 'Number of elements in parallel filter (%pMaxThreads%) is smaller than required number of threads. Number of threads is set to %pMaxThreads%.' ));
  EndIf;

  sElemMDX = '';
  sElemMDXClr = '';
  sFind = sParFilter;
  nTokens = 0;
  nElem = 0;
  While( sFind @<> '' );
    nToken = SCAN( cElementDelim, sFind );
    nToken = IF( nToken = 0, LONG( sFind ) + 1, nToken );
    sToken = TRIM( SUBST( sFind, 1, nToken - 1 ));
    nTokens = nTokens + 1;
    nThread = IF( MOD( nTokens, nMaxThreads ) = 0, nMaxThreads, MOD( nTokens, nMaxThreads )) - 1;
    If( nThread = nThreadID );
      IF( nLogOutput <> 0 );
        LogOutput('Info', Expand('Thread %pThreadID% allocating [%sToken%] from parallel filter %pDimPar% for further expansion for data source.'));
      EndIf;
      sElemMDX = Expand( '%sElemMDX%{TM1FILTERBYLEVEL(DESCENDANTS([%pDimPar%].[%sToken%], 0, SELF_AND_AFTER), 0)}+' );
      IF( nLogOutput <> 0 );
        LogOutput('Info', Expand('Thread %pThreadID% allocating [%sToken%] from parallel filter %sDimParClr% for further expansion for target data clear.'));
      EndIf;
      sElemMDXClr = Expand( '%sElemMDXClr%{TM1FILTERBYLEVEL(DESCENDANTS([%sDimParClr%].[%sToken%], 0, SELF_AND_AFTER), 0)}+' );
      nElem = nElem + 1;
    EndIf;
    sFind = DELET( sFind, 1, nToken );
  End;
  sElemMDX = SUBST( sElemMDX, 1, LONG( sElemMDX ) - 1);
  sElemMDXClr = SUBST( sElemMDXClr, 1, LONG( sElemMDXClr ) - 1);
  If( nElem = 1 );
    sDimParMDX = sElemMDX;
    sDimClrMDX = sElemMDXClr;
  Else;
    sDimParMDX = Expand( '{%sElemMDX%}');
    sDimClrMDX = Expand( '{%sElemMDXClr%}');
  EndIf;
EndIf;

sSubLen = cSubSrc | '_LEN';
ExecuteProcess('}bedrock.hier.sub.create.bymdx',
  'pDim', pDimPar,
  'pSub', sSubLen,
  'pMDXExpr', sDimParMDX,
  'pConvertToStatic', 1,
  'pTemp', cTemp
);
nElements = SubsetGetSize( pDimPar, sSubLen );

sSubClrLen = cSubSrc | '_CLR_LEN';
ExecuteProcess('}bedrock.hier.sub.create.bymdx',
  'pDim', sDimParClr,
  'pSub', sSubClrLen,
  'pMDXExpr', sDimClrMDX,
  'pConvertToStatic', 1,
  'pTemp', cTemp
);
nElementsClr = SubsetGetSize( sDimParClr, sSubClrLen );

Prolog (after spawning child threads)

If ( nMaxThreads > 1 & nThreadID >= 0 );
  # Allocate working space to threads - indices start and step
  If( sParFilter @= '' );
    # We need to shuffle elements from one common heap
    If ( nMaxThreads > 1 );
      If( cThreadMonitoringEnabled <> 0 );
        nLowerIndex = nThreadID;
        nThreadStep = nThreads - 1;
      Else;
        nLowerIndex = nThreadID + 1;
        nThreadStep = nThreads;        
      EndIf;
    Else;
       nLowerIndex = 1;
       nThreadStep = 1;
    EndIf;
  Else;
    # Elements are already assigned per parallel filter, need to allocate one by one
    nLowerIndex = 1;
    nThreadStep = 1;
  EndIf;
Else;
  # Regular processing - single thread
  nThreadStep = 1;
  nLowerIndex = 1;
EndIf;

Prolog (when creating source and target views)

If( nLowerIndex <= nElementsClr & ( nMaxThreads =1 % nMaxThreads > 1 & ( cThreadMonitoringEnabled <> 0 & nThreadID <> 0 % cThreadMonitoringEnabled = 0)));
  
  # Allocate working space within parallelized dimension
  SubsetCreate( pDimPar, cParSubSrc, cTemp );
  SubsetCreate( sDimParClr, cSubClr, cTemp );
  
  # Allocate working space to threads - shuffle elements
  n = nLowerIndex;
  nSubsetItem = 1;
  While ( n <= nElements );
    sElem = SubsetGetElementName( pDimPar, sSubLen, n );
    If( nLogOutput > 1 );
      LogOutput( 'Info', Expand( 'Thread %pThreadID% allocating %sElem%.' ));
    EndIf;
    SubsetElementInsert( pDimPar, cParSubSrc, sElem, nSubsetItem );
    nSubsetItem = nSubsetItem + 1;
    n = n + nThreadStep;
  End;
  
  n = nLowerIndex;
  nSubsetItem = 1;
  While ( n <= nElementsClr );
    sElem = SubsetGetElementName( sDimParClr, sSubClrLen, n );
    If( nLogOutput > 1 );
      LogOutput( 'Info', Expand( 'Thread %pThreadID% allocating %sElem% for ViewZeroOut.' ));
    EndIf;
    SubsetElementInsert( sDimParClr, cSubClr, sElem, nSubsetItem );
    nSubsetItem = nSubsetItem + 1;
    n = n + nThreadStep;
  End;
  
  If( nLogOutput <> 0 );
    sElem = NumberToString( SubsetGetSize( pDimPar, cParSubSrc ));
    sElemClr = NumberToString( SubsetGetSize( sDimParClr, cSubClr ));
    LogOutput( 'Info', Expand( 'Thread %pThreadID% allocated %sElem% elements from %pDimPar% for data source and %sElemClr% elements from %sDimParClr% for target data clear.' ));
  EndIf;
  
  #Region Prepare Source Data Filters
  #TODO: Repeat for every dimension that is allowed for parallel processing, define process parameter for each
  sDimName = 'DIMENSION_NAME_1';
  sPar = pDIMENSION_NAME_1;
  If( TRIM( sPar ) @<> '' & pDimPar @<> sDimName );
    sFilter = sFilter | cDimensionDelim | sDimName | cElementStartDelim | sPar;
  EndIf;
  #EndRegion Prepare Source Data Filters
  
  #Region Create Data Source
  #EndRegion Create Data Source
  
  # Elements in parallel dimenison as per thread allocation were already present in the subset - we need to exclude elements that were provided by Bedrock by default as we have omitted the parallel dimension from filter - these follow after all elements generated for parallel processing
  SubsetDeleteAllElements( pDimPar, cSubSrc );
  nElem = 1;
  nElems = SubsetGetSize( pDimPar, cParSubSrc );
  While( nElem <= nElems );
    sElem = SubsetGetElementName( pDimPar, cParSubSrc, nElem );
    SubsetElementInsert( pDimPar, cSubSrc, sElem, nElem );
    nElem = nElem + 1;
  End;
  
  #Region Create Data Target
  #EndRegion Create Data Target
  
  SubsetDeleteAllElements( sDim, cViewClr );
  nElem = 1;
  nElems = SubsetGetSize( sDim, cSubSrc );
  While( nElem <= nElems );
    sElem = SubsetGetElementName( sDim, cSubSrc, nElem );
    sElem = SUBST( sElem, 1, LONG( sElem ) - LONG( cRuleSuffix ) );
    SubsetElementInsert( sDim, cViewClr, sElem, nElem );
    nElem = nElem + 1;
  End;
  
  #Region Clear Data Target
  #EndRegion Clear Data Target
  
  #Region Assign Data Source to TI
  #EndRegion Assign Data Source to TI
Else;
  DataSourceType = 'NULL';
EndIf;

Apliqo FPM Documentation

Snippet: Elements to Thread Allocation Without Mappings

Related content