void CSteppedConjunctionOptimizer::finishCandidates()
{
    rootActivity->resetEOF();
    ForEachItemIn(i1, inputs)
        inputs.item(i1).setRestriction(NULL, 0);
    ForEachItemIn(i2, joins)
        joins.item(i2).stopRestrictedJoin();
    ForEachItemIn(i3, pseudoInputs)
        pseudoInputs.item(i3).clearPending();
    if (prevEqualityRow)
        inputAllocator->releaseRow(prevEqualityRow);
    prevEqualityRow = equalityRow;
    equalityRow = NULL;
}

void CHThorNaryActivity::done()
{
    ForEachItemIn(i, expandedInputs)
        expandedInputs.item(i)->done();
    expandedInputs.kill();
    CHThorMultiInputActivity::done();
}

    virtual void flushTraceSummary()
    {
        StringBuffer logstr;
        logstr.appendf("activeReqs=").append(m_active).append(';');
        logstr.append("user=").append(queryUserId());
        if (m_peer.length())
            logstr.append('@').append(m_peer.get());
        logstr.append(';');

        if (m_hasException)
        {
            logstr.appendf("[email protected]%dms=%d;", m_exceptionTime, m_exceptionCode);
        }

        StringBuffer value;
        value.append("total=").appendulong(m_processingTime).append("ms");
        if (m_hasException || (getEspLogLevel() > LogNormal))
        {
            m_traceValues.append(value.str());

            if (m_traceValues.length())
            {
                ForEachItemIn(idx, m_traceValues)
                    logstr.append(m_traceValues.item(idx)).append(";");
                m_traceValues.kill();
            }
        }
        else
        {
            logstr.appendf("%s;", value.str());
        }

        DBGLOG("TxSummary[%s]", logstr.str());
    }

 virtual void waitForStrands()
 {
     producerStopSem.signal(threads.ordinality());
     ForEachItemIn(i, threads)
         threads.item(i).join();
     threads.kill();
 }

void CachedPasswordProvider::serialize(MemoryBuffer & out)
{
    unsigned num = passwords.ordinality();
    out.append(num);
    ForEachItemIn(idx, passwords)
        passwords.item(idx).serialize(out);
}

    virtual int processCMD()
    {
        Owned<IClientWsPackageProcess> packageProcessClient = createCmdClient(WsPackageProcess, *this);
        StringBuffer pkgInfo;
        pkgInfo.loadFile(optFileName);

        fprintf(stdout, "\n ... adding package map %s now\n\n", optFileName.sget());

        Owned<IClientAddPackageRequest> request = packageProcessClient->createAddPackageRequest();
        request->setActivate(optActivate);
        request->setInfo(pkgInfo);
        request->setTarget(optTarget);
        request->setPackageMap(optPackageMapId);
        request->setProcess(optProcess);
        request->setDaliIp(optDaliIP);
        request->setOverWrite(optOverWrite);
        request->setGlobalScope(optGlobalScope);
        request->setSourceProcess(optSourceProcess);

        Owned<IClientAddPackageResponse> resp = packageProcessClient->AddPackage(request);
        if (resp->getExceptions().ordinality())
            outputMultiExceptions(resp->getExceptions());

        StringArray &notFound = resp->getFilesNotFound();
        if (notFound.length())
        {
            fputs("\nFiles defined in package but not found in DFS:\n", stderr);
            ForEachItemIn(i, notFound)
                fprintf(stderr, "  %s\n", notFound.item(i));
            fputs("\n", stderr);
        }

        return 0;
    }

 //Wait for all active producers to complete - including calling stop on their inputs
 void waitForProducers()
 {
     producerStopSem.signal(numProducers);
     ForEachItemIn(i, threads)
         threads.item(i).join();
     threads.kill();
 }

 //Stop producers that have already been started()
 virtual void stopActiveProducers()
 {
     producerStopSem.signal(numProducers);
     ForEachItemIn(i, threads)
         threads.item(i).join();
     threads.kill();
 }

unsigned getNumUniqueExpressions(const HqlExprArray & exprs)
{
    TransformMutexBlock block;
    ExpressionStatsInfo info;
    ForEachItemIn(i, exprs)
        calcNumUniqueExpressions(&exprs.item(i),info);
    return info.count;
}

void CCassandraLogAgent::createTable(const char *dbName, const char *tableName, StringArray& columnNames, StringArray& columnTypes, const char* keys)
{
    StringBuffer fields;
    ForEachItemIn(i, columnNames)
    fields.appendf("%s %s,", columnNames.item(i), columnTypes.item(i));

    VStringBuffer createTableSt("CREATE TABLE IF NOT EXISTS %s.%s (%s PRIMARY KEY (%s));", dbName, tableName, fields.str(), keys);
    executeSimpleStatement(createTableSt.str());
}

void ScopeConsistencyChecker::checkConsistent(IHqlExpression * root, const HqlExprArray & _activeTables)
{
    ForEachItemIn(i, _activeTables)
        activeTables.append(OLINK(_activeTables.item(i)));
    if (root->isDataset())
        pushScope();
    analyse(root, 0);
    if (root->isDataset())
        popScope();
}

void CSlavePartMapping::getParts(unsigned i, IArrayOf<IPartDescriptor> &parts)
{
    if (local)
        i = 0;
    if (i>=maps.ordinality()) return;

    CSlaveMap &map = maps.item(i);
    ForEachItemIn(m, map)
        parts.append(*LINK(&map.item(m)));
}

void LogicalGraphCreator::createLogicalGraph(HqlExprArray & exprs)
{
    graph.setown(createPTree("graph"));

//  beginSubGraph(NULL, false);
    ForEachItemIn(i, exprs)
        createRootGraphActivity(&exprs.item(i));
//  endSubGraph();
    wu->createGraph("Logical", NULL, GraphTypeEcl, graph.getClear(), 0);
}

void LogicalGraphCreator::createLogicalGraph(HqlExprArray & exprs)
{
    graph.setown(createPTree("graph"));

//  beginSubGraph(NULL, false);
    ForEachItemIn(i, exprs)
        createRootGraphActivity(&exprs.item(i));
//  endSubGraph();

    Owned<IWUGraph> wug = wu->updateGraph("Logical");
    wug->setXGMMLTree(graph.getClear());
    wug->setType(GraphTypeEcl);
}

void CDiskReadMasterBase::serializeSlaveData(MemoryBuffer &dst, unsigned slave)
{
    IHThorDiskReadBaseArg *helper = (IHThorDiskReadBaseArg *) queryHelper();
    dst.append(helper->getFileName());
    dst.append(subfileLogicalFilenames.ordinality());
    if (subfileLogicalFilenames.ordinality())
    {
        ForEachItemIn(s, subfileLogicalFilenames)
            dst.append(subfileLogicalFilenames.item(s));
    }
    if (mapping)
        mapping->serializeMap(slave, dst);
    else
        CSlavePartMapping::serializeNullMap(dst);
}

 static StringBuffer &appendJSONExceptions(StringBuffer &s, IMultiException *e, const char *objname="Exceptions", const char *arrayName = "Exception")
 {
     if (!e)
         return s;
     if (objname && *objname)
         appendJSONName(s, objname).append('{');
     if (arrayName && *arrayName)
         appendJSONName(s, arrayName).append('[');
     ForEachItemIn(i, *e)
         appendJSONException(s, &e->item(i), NULL, NULL);
     if (arrayName && *arrayName)
         s.append(']');
     if (objname && *objname)
         s.append('}');
     return s;
 }

void CWSESPControlEx::cleanSessions(bool allSessions, const char* _id, const char* _userID, const char* _fromIP)
{
    StringBuffer searchPath;
    setSessionXPath(allSessions, _id, _userID, _fromIP, searchPath);

    Owned<IRemoteConnection> globalLock = querySDSConnectionForESPSession(RTM_LOCK_WRITE, SESSION_SDS_LOCK_TIMEOUT);
    Owned<IPropertyTreeIterator> iter = globalLock->queryRoot()->getElements("*");
    ForEach(*iter)
    {
        IArrayOf<IPropertyTree> toRemove;
        Owned<IPropertyTreeIterator> iter1 = iter->query().getElements(searchPath.str());
        ForEach(*iter1)
            toRemove.append(*LINK(&iter1->query()));

        ForEachItemIn(i, toRemove)
            iter->query().removeTree(&toRemove.item(i));
    }
}

void CSteppedConjunctionOptimizer::beforeProcessing()
{
    //NB: This function is only called once, after we have decided it is worth processing.
    assertex(!eof);     // just check it isn't called more than once
    assertex(numInputs);

    bool hasDistance = (helper.getJoinFlags() & IHThorNWayMergeJoinArg::MJFhasdistance) != 0;
    for (unsigned i3 = 0; i3 < numInputs; i3++)
    {
        OrderedInput & next = *new OrderedInput(inputs.item(i3), i3, hasDistance);
        orderedInputs.append(next);
        if (next.canOptimizeOrder())
            numOptimizeInputs++;
    }
    //Sort so that inputs are ordered (priority-inputs, optimizable, non-optimizable)
    orderedInputs.sort(compareInitialInputOrder);

    //If only a single re-orderable input, treat it as unorderable.
    if (numOptimizeInputs == 1)
    {
        assertex(orderedInputs.item(numPriorityInputs).canOptimizeOrder());
        orderedInputs.item(numPriorityInputs).stopOptimizeOrder();
        numOptimizeInputs = 0;
    }

    maxOptimizeInput = numPriorityInputs + numOptimizeInputs;

    associateRemoteInputs(orderedInputs, numPriorityInputs);
    
    //MORE: If some inputs have known priority, and other remote inputs don't, then we could consider
    //      connecting the unknown inputs to the last known inputs.
    ForEachItemIn(i4, joins)
        joins.item(i4).markRestrictedJoin(numEqualFields);

    assertex(helper.getJoinFlags() & IHThorNWayMergeJoinArg::MJFhasclearlow);       // Don't support (very) old workunits that don't define this..
    if (helper.getJoinFlags() & IHThorNWayMergeJoinArg::MJFhasclearlow)
    {
        RtlDynamicRowBuilder rowBuilder(inputAllocator);
        size32_t size = helper.createLowInputRow(rowBuilder);
        lowestSeekRow = rowBuilder.finalizeRowClear(size);
    }
}

int CEclDirectSoapBindingEx::sendRunEclExForm(IEspContext &context, CHttpRequest* request, CHttpResponse* response)
{
    StringBuffer xml;
    xml.append("<RunEclEx clientVersion='").append(context.getClientVersion()).append("'>");
    appendXMLTag(xml, "UseEclRepository", (supportRepository) ? "Yes" : "No");
    appendXMLTag(xml, "Redirect", (redirect) ? "Yes" : "No");
    appendXMLTag(xml, "IncludeResults", (redirect) ? "No" : "Yes");
    ForEachItemIn(i, clusters)
        appendXMLTag(xml, "Cluster", clusters.item(i));
    xml.append("</RunEclEx>");

    StringBuffer xslt(getCFD());
    xslt.append("./smc_xslt/run_ecl.xslt");

    StringBuffer html;
    xsltTransform(xml.str(), xslt.str(), NULL, html);
    response->setContent(html.str());
    response->setContentType(HTTP_TYPE_TEXT_HTML_UTF8);
    response->send();

    return 0;
}

    CKeyedJoinMaster(CMasterGraphElement *info) : CMasterActivity(info)
    {
        helper = (IHThorKeyedJoinArg *) queryHelper();
        //GH->JCS a bit wasteful creating this array each time.
        progressKinds.append(StNumIndexSeeks);
        progressKinds.append(StNumIndexScans);
        progressKinds.append(StNumIndexAccepted);
        progressKinds.append(StNumPostFiltered);
        progressKinds.append(StNumPreFiltered);

        if (helper->diskAccessRequired())
        {
            progressKinds.append(StNumDiskSeeks);
            progressKinds.append(StNumDiskAccepted);
            progressKinds.append(StNumDiskRejected);
        }
        ForEachItemIn(l, progressKinds)
            progressInfoArr.append(*new ProgressInfo);
        localKey = false;
        numTags = 0;
        tags[0] = tags[1] = tags[2] = tags[3] = TAG_NULL;
        reInit = 0 != (helper->getFetchFlags() & (FFvarfilename|FFdynamicfilename));
        remoteDataFiles = false;
    }