void init(MemoryBuffer &data, MemoryBuffer &slaveData)
 {
     appendOutputLinked(this);   // adding 'me' to outputs array
     helper = static_cast <IHThorSortArg *> (queryHelper());
     IHThorAlgorithm * algo = helper?(static_cast<IHThorAlgorithm *>(helper->selectInterface(TAIalgorithm_1))):NULL;
     unstable = (algo&&algo->getAlgorithmFlags()&TAFunstable);
 }

 GroupAggregateSlaveActivity(CGraphElementBase *_container) 
     : CSlaveActivity(_container)
 { 
     helper = static_cast <IHThorAggregateArg *> (queryHelper());
     setRequireInitData(false);
     appendOutputLinked(this);
 }

 BaseChooseSetsActivity(CGraphElementBase *_container) : CSlaveActivity(_container)
 {
     helper = static_cast <IHThorChooseSetsArg *> (queryHelper());
     done = false;
     tallies = NULL;
     appendOutputLinked(this);
 }

CDiskReadSlaveActivityBase::CDiskReadSlaveActivityBase(CGraphElementBase *_container) : CSlaveActivity(_container)
{
    helper = (IHThorDiskReadBaseArg *)queryHelper();
    crcCheckCompressed = 0 != container.queryJob().getWorkUnitValueInt("crcCheckCompressed", 0);
    gotMeta = false;
    checkFileCrc = !globals->getPropBool("Debug/@fileCrcDisabled");
}

CWriteMasterBase::CWriteMasterBase(CMasterGraphElement *info) : CMasterActivity(info)
{
    publishReplicatedDone = !globals->getPropBool("@replicateAsync", true);
    replicateProgress.setown(new ProgressInfo);
    diskHelperBase = (IHThorDiskWriteArg *)queryHelper();
    targetOffset = 0;
}

void CDiskWriteSlaveActivityBase::init(MemoryBuffer &data, MemoryBuffer &slaveData)
{
    diskHelperBase = static_cast <IHThorDiskWriteArg *> (queryHelper());

    data.read(logicalFilename);
    dlfn.set(logicalFilename);
    if (diskHelperBase->getFlags() & TDXtemporary)
        data.read(usageCount);
    if (diskHelperBase->getFlags() & TDWextend)
    {
        assertex(!"TBD extended CRC broken");
        unsigned crc;
        if (partDesc->getCrc(crc))
            fileCRC.reset(~crc);
    }
    partDesc.setown(deserializePartFileDescriptor(data));
    if (dlfn.isExternal())
    {
        mpTag = container.queryJob().deserializeMPTag(data);
        if (dlfn.isQuery() && (0 != container.queryJob().getWorkUnitValueInt("rfsqueryparallel", 0)))
            rfsQueryParallel = true;
    }
    if (0 != (diskHelperBase->getFlags() & TDXgrouped))
        grouped = true;
    compress = partDesc->queryOwner().isCompressed();
    void *ekey;
    size32_t ekeylen;
    diskHelperBase->getEncryptKey(ekeylen,ekey);
    if (ekeylen!=0) {
        ecomp.setown(createAESCompressor256(ekeylen,ekey));
        memset(ekey,0,ekeylen);
        free(ekey);
        compress = true;
    }
}

 void serializeSlaveData(MemoryBuffer &dst, unsigned slave)
 {
     IHThorSpillArg *helper = (IHThorSpillArg *)queryHelper();
     IPartDescriptor *partDesc = fileDesc->queryPart(slave);
     partDesc->serialize(dst);
     dst.append(helper->getTempUsageCount());
 }

 void init()
 {
     IHThorSpillArg *helper = (IHThorSpillArg *)queryHelper();
     IArrayOf<IGroup> groups;
     OwnedRoxieString fname(helper->getFileName());
     fillClusterArray(container.queryJob(), fname, clusters, groups);
     fileDesc.setown(queryThorFileManager().create(container.queryJob(), fname, clusters, groups, true, TDWnoreplicate+TDXtemporary));
     IPropertyTree &props = fileDesc->queryProperties();
     bool blockCompressed=false;
     void *ekey;
     size32_t ekeylen;
     helper->getEncryptKey(ekeylen,ekey);
     if (ekeylen)
     {
         memset(ekey,0,ekeylen);
         free(ekey);
         props.setPropBool("@encrypted", true);      
         blockCompressed = true;
     }
     else if (0 != (helper->getFlags() & TDWnewcompress) || 0 != (helper->getFlags() & TDXcompress))
         blockCompressed = true;
     if (blockCompressed)
         props.setPropBool("@blockCompressed", true);        
     if (0 != (helper->getFlags() & TDXgrouped))
         fileDesc->queryProperties().setPropBool("@grouped", true);
     props.setProp("@kind", "flat");
     if (container.queryOwner().queryOwner() && (!container.queryOwner().isGlobal())) // I am in a child query
     { // do early, because this will be local act. and will not come back to master until end of owning graph.
         container.queryTempHandler()->registerFile(fname, container.queryOwner().queryGraphId(), helper->getTempUsageCount(), TDXtemporary & helper->getFlags(), getDiskOutputKind(helper->getFlags()), &clusters);
         queryThorFileManager().publish(container.queryJob(), fname, true, *fileDesc);
     }
 }

    void setInitialCounter(rowcount_t prevRecCount)
    {
        if (numerator)
        {
            IHThorEnthArg * helper = static_cast <IHThorEnthArg *> (queryHelper());
            if(denominator == 0) denominator = 1;

            counter = (rowcount_t)((helper->getSampleNumber()-1) * greatestCommonDivisor(numerator, denominator));
            if (counter >= denominator)
                counter %= denominator;
            rowcount_t maxBatch = (RCMAX - denominator) / numerator;
            while (prevRecCount > 0)
            {
                rowcount_t next = prevRecCount;
                if (next > maxBatch) next = maxBatch;
                counter = (counter + next * numerator) % denominator;
                prevRecCount -= next;
            }
        }
        else
            abortSoon = true;
#if THOR_TRACE_LEVEL >= 5
        ActPrintLog("ENTH: init - Numerator = %"RCPF"d, Denominator = %"RCPF"d", numerator, denominator);   
        ActPrintLog("%s: Initial value of counter %"RCPF"d", actStr.str(), counter);
#endif
    }

 explicit CProjecStrandProcessor(CThorStrandedActivity &parent, IEngineRowStream *inputStream, unsigned outputId)
     : CThorStrandProcessor(parent, inputStream, outputId)
 {
     helper = static_cast <IHThorProjectArg *> (queryHelper());
     Owned<IRowInterfaces> rowIf = parent.getRowInterfaces();
     allocator.setown(parent.getRowAllocator(rowIf->queryRowMetaData(), (roxiemem::RoxieHeapFlags)(roxiemem::RHFpacked|roxiemem::RHFunique)));
 }

 virtual void done()
 {
     CDiskReadMasterVF::done();
     IHThorDiskReadBaseArg *helper = (IHThorDiskReadBaseArg *)queryHelper();
     if (0 != (helper->getFlags() & TDXtemporary) && !container.queryJob().queryUseCheckpoints())
         container.queryTempHandler()->deregisterFile(fileName, fileDesc->queryProperties().getPropBool("@pausefile"));
 }

    void done()
    {
        IPropertyTree &props = fileDesc->queryProperties();
        props.setPropBool("@csv", true);
        props.setProp("@format", "utf8n");
        IHThorCsvWriteArg *helper=(IHThorCsvWriteArg *)queryHelper();
        ICsvParameters *csvParameters = helper->queryCsvParameters();
        StringBuffer separator;
        OwnedRoxieString rs(csvParameters->getSeparator(0));
        const char *s = rs;
        while (s && *s)
        {
            if (',' == *s)
                separator.append("\\,");
            else
                separator.append(*s);
            ++s;
        }
        props.setProp("@csvSeparate", separator.str());
        props.setProp("@csvQuote", rs.setown(csvParameters->getQuote(0)));
        props.setProp("@csvTerminate", rs.setown(csvParameters->getTerminator(0)));
        props.setProp("@csvEscape", rs.setown(csvParameters->getEscape(0)));

        CWriteMasterBase::done(); // will publish
    }

 virtual void validateFile(IDistributedFile *file)
 {
     IHThorDiskReadBaseArg *helper = (IHThorDiskReadBaseArg *)queryHelper();
     bool codeGenGrouped = 0 != (TDXgrouped & helper->getFlags());
     bool isGrouped = fileDesc->isGrouped();
     if (isGrouped != codeGenGrouped)
     {
         Owned<IException> e = MakeActivityWarning(&container, TE_GroupMismatch, "DFS and code generated group info. differs: DFS(%s), CodeGen(%s), using DFS info", isGrouped?"grouped":"ungrouped", codeGenGrouped?"grouped":"ungrouped");
         queryJobChannel().fireException(e);
     }
     IOutputMetaData *recordSize = helper->queryDiskRecordSize()->querySerializedDiskMeta();
     if (recordSize->isFixedSize()) // fixed size
     {
         if (0 != fileDesc->queryProperties().getPropInt("@recordSize"))
         {
             size32_t rSz = fileDesc->queryProperties().getPropInt("@recordSize");
             if (isGrouped)
                 rSz--; // eog byte not to be included in this test.
             if (rSz != recordSize->getMinRecordSize())
                 throw MakeThorException(TE_RecordSizeMismatch, "Published record size %d for file %s, does not match coded record size %d", rSz, fileName.get(), recordSize->getMinRecordSize());
         }
         if (!fileDesc->isCompressed() && (TDXcompress & helper->getFlags()))
         {
             size32_t rSz = recordSize->getMinRecordSize();
             if (isGrouped) rSz++;
             if (rSz >= MIN_ROWCOMPRESS_RECSIZE)
             {
                 Owned<IException> e = MakeActivityWarning(&container, TE_CompressionMismatch, "Ignoring compression attribute on file '%s', which is not published as compressed in DFS", fileName.get());
                 queryJobChannel().fireException(e);
             }
         }
     }
     if (0 == (TDRnocrccheck & helper->getFlags()))
         checkFormatCrc(this, file, helper->getFormatCrc(), false);
 }

    void init()
    {
        helper = (IHThorKeyDiffArg *)queryHelper();
        OwnedRoxieString origName(helper->getOriginalName());
        OwnedRoxieString updatedName(helper->getUpdatedName());
        originalIndexFile.setown(queryThorFileManager().lookup(container.queryJob(), origName));
        newIndexFile.setown(queryThorFileManager().lookup(container.queryJob(), updatedName));
        if (originalIndexFile->numParts() != newIndexFile->numParts())
            throw MakeActivityException(this, TE_KeyDiffIndexSizeMismatch, "Index %s and %s differ in width", origName.get(), updatedName.get());
        if (originalIndexFile->querySuperFile() || newIndexFile->querySuperFile())
            throw MakeActivityException(this, 0, "Diffing super files not supported");  

        width = originalIndexFile->numParts();

        originalDesc.setown(originalIndexFile->getFileDescriptor());
        newIndexDesc.setown(newIndexFile->getFileDescriptor());
        Owned<IPartDescriptor> tlkDesc = originalDesc->getPart(originalDesc->numParts()-1);
        const char *kind = tlkDesc->queryProperties().queryProp("@kind");
        local = NULL == kind || 0 != stricmp("topLevelKey", kind);

        if (!local)
            width--; // 1 part == No n distributed / Monolithic key
        if (width > container.queryJob().querySlaves())
            throw MakeActivityException(this, 0, "Unsupported: keydiff(%s, %s) - Cannot diff a key that's wider(%d) than the target cluster size(%d)", originalIndexFile->queryLogicalName(), newIndexFile->queryLogicalName(), width, container.queryJob().querySlaves());

        queryThorFileManager().noteFileRead(container.queryJob(), originalIndexFile);
        queryThorFileManager().noteFileRead(container.queryJob(), newIndexFile);

        IArrayOf<IGroup> groups;
        OwnedRoxieString outputName(helper->getOutputName());
        fillClusterArray(container.queryJob(), outputName, clusters, groups);
        patchDesc.setown(queryThorFileManager().create(container.queryJob(), outputName, clusters, groups, 0 != (KDPoverwrite & helper->getFlags()), 0, !local, width));
    }

    virtual void init()
    {
        CSortBaseActivityMaster::init();
        IHThorSortArg *helper = (IHThorSortArg *)queryHelper();
        IHThorAlgorithm *algo = static_cast<IHThorAlgorithm *>(helper->selectInterface(TAIalgorithm_1));
        OwnedRoxieString algoname(algo->getAlgorithm());
        unsigned flags = algo->getAlgorithmFlags();
        if (algoname && (0 != stricmp(algoname, "quicksort")))
        {
            Owned<IException> e = MakeActivityException(this, 0, "Ignoring, unsupported sort order algorithm '%s'", algoname.get());
            reportExceptionToWorkunit(container.queryJob().queryWorkUnit(), e);
        }
        OwnedRoxieString cosortlogname(helper->getSortedFilename());
        if (cosortlogname&&*cosortlogname)
        {
            Owned<IDistributedFile> coSortFile = queryThorFileManager().lookup(container.queryJob(), cosortlogname);
            addReadFile(coSortFile);
            Owned<IFileDescriptor> fileDesc = coSortFile->getFileDescriptor();
            unsigned o;
            for (o=0; o<fileDesc->numParts(); o++)
            {
                Owned<IPartDescriptor> partDesc = fileDesc->getPart(o);
                if (cosortfilenames.length())
                    cosortfilenames.append("|");

                // JCSMORE - picking the primary here, means no automatic use of backup copy, could use RMF's possibly.
                getPartFilename(*partDesc, 0, cosortfilenames);
            }
        }
    }

 CCsvReadActivityMaster(CMasterGraphElement *info) : CDiskReadMasterBase(info)
 {
     helper = (IHThorCsvReadArg *)queryHelper();
     headerLines = helper->queryCsvParameters()->queryHeaderLen();
     if (headerLines)
         mpTag = container.queryJob().allocateMPTag();
 }

    virtual void init()
    {
        HashDistributeMasterBase::init();

        // JCSMORE should common up some with indexread
        IHThorKeyedDistributeArg *helper = (IHThorKeyedDistributeArg *)queryHelper();

        StringBuffer scoped;
        OwnedRoxieString indexFileName(helper->getIndexFileName());
        queryThorFileManager().addScope(container.queryJob(), indexFileName, scoped);
        file.setown(queryThorFileManager().lookup(container.queryJob(), indexFileName));
        if (!file)
            throw MakeActivityException(this, 0, "KeyedDistribute: Failed to find key: %s", scoped.str());
        if (0 == file->numParts())
            throw MakeActivityException(this, 0, "KeyedDistribute: Can't distribute based on an empty key: %s", scoped.str());

        checkFormatCrc(this, file, helper->getFormatCrc(), true);
        Owned<IFileDescriptor> fileDesc = file->getFileDescriptor();
        Owned<IPartDescriptor> tlkDesc = fileDesc->getPart(fileDesc->numParts()-1);
        if (!tlkDesc->queryProperties().hasProp("@kind") || 0 != stricmp("topLevelKey", tlkDesc->queryProperties().queryProp("@kind")))
            throw MakeActivityException(this, 0, "Cannot distribute using a non-distributed key: '%s'", scoped.str());
        unsigned location;
        OwnedIFile iFile;
        StringBuffer filePath;
        if (!getBestFilePart(this, *tlkDesc, iFile, location, filePath))
            throw MakeThorException(TE_FileNotFound, "Top level key part does not exist, for key: %s", file->queryLogicalName());
        OwnedIFileIO iFileIO = iFile->open(IFOread);
        assertex(iFileIO);

        tlkMb.append(iFileIO->size());
        ::read(iFileIO, 0, (size32_t)iFileIO->size(), tlkMb);

        queryThorFileManager().noteFileRead(container.queryJob(), file);
    }

    void process()
    {
        ActPrintLog("ReDistributeActivityMaster::process");
        HashDistributeMasterBase::process();        
        IHThorHashDistributeArg *helper = (IHThorHashDistributeArg *)queryHelper(); 
        unsigned n = container.queryJob().querySlaves();
        MemoryAttr ma;
        offset_t *sizes = (offset_t *)ma.allocate(sizeof(offset_t)*n);
        unsigned i;
        try {
            for (i=0;i<n;i++) {
                if (abortSoon)
                    return;
                CMessageBuffer mb;
#ifdef _TRACE
                ActPrintLog("ReDistribute process, Receiving on tag %d",statstag);
#endif
                rank_t sender;
                if (!receiveMsg(mb, RANK_ALL, statstag, &sender)||abortSoon) 
                    return;
#ifdef _TRACE
                ActPrintLog("ReDistribute process, Received size from %d",sender);
#endif
                sender--;
                assertex((unsigned)sender<n);
                mb.read(sizes[sender]);
            }
            ActPrintLog("ReDistributeActivityMaster::process sizes got");
            for (i=0;i<n;i++) {
                CMessageBuffer mb;
                mb.append(n*sizeof(offset_t),sizes);
#ifdef _TRACE
                ActPrintLog("ReDistribute process, Replying to node %d tag %d",i+1,statstag);
#endif
                if (!container.queryJob().queryJobComm().send(mb, (rank_t)i+1, statstag))
                    return;
            }
            // check if any max skew broken
            double maxskew = helper->getTargetSkew();
            if (maxskew>helper->getSkew()) {
                offset_t tot = 0;
                for (i=0;i<n;i++) 
                    tot += sizes[i];
                offset_t avg = tot/n;
                for (i=0;i<n;i++) {
                    double r = ((double)sizes[i]-(double)avg)/(double)avg;
                    if ((r>=maxskew)||(-r>maxskew)) {
                        throw MakeActivityException(this, TE_DistributeFailedSkewExceeded, "DISTRIBUTE maximum skew exceeded (node %d has %"I64F"d, average is %"I64F"d)",i+1,sizes[i],avg);
                    }               
                }
            }

        }
        catch (IException *e) {
            ActPrintLog(e,"ReDistribute");
            throw;
        }
        ActPrintLog("ReDistributeActivityMaster::process exit");
    }

    void init(MemoryBuffer &data, MemoryBuffer &slaveData)
    {
        CDiskWriteSlaveActivity::init(data, slaveData);
        helper = static_cast <IHThorCsvWriteArg *> (queryHelper());

        singleHF = 0 != (ICsvParameters::singleHeaderFooter & helper->queryCsvParameters()->getFlags());
        csvOutput.init(helper->queryCsvParameters(), 0 != container.queryJob().getWorkUnitValueInt("oldCSVoutputFormat", 0));
    }

 virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
 {
     helper = static_cast <IHThorCatchArg *> (queryHelper());
     kindStr.append(activityKindStr(container.getKind()));
     kindStr.toUpperCase();
     eos = false;
     appendOutputLinked(this);
 }