text:
/* Copyright (C) 2014 InfiniDB, Inc.
   Copyright (C) 2019 MariaDB Corporation

   This program is free software; you can redistribute it and/or
   modify it under the terms of the GNU General Public License
   as published by the Free Software Foundation; version 2 of
   the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
   MA 02110-1301, USA. */

//  $Id: tupleannexstep.cpp 9661 2013-07-01 20:33:05Z pleblanc $

//#define NDEBUG
#include <cassert>
#include <sstream>
#include <iomanip>
#ifdef _MSC_VER
#include <unordered_set>
#else
#include <tr1/unordered_set>
#endif
using namespace std;

#include <boost/shared_ptr.hpp>
#include <boost/shared_array.hpp>
#include <boost/uuid/uuid_io.hpp>
using namespace boost;

#include "messagequeue.h"
using namespace messageqcpp;

#include "loggingid.h"
#include "errorcodes.h"
using namespace logging;

#include "calpontsystemcatalog.h"
#include "constantcolumn.h"
#include "simplecolumn.h"
using namespace execplan;

#include "rowgroup.h"
using namespace rowgroup;

#include "hasher.h"
#include "stlpoolallocator.h"
#include "threadnaming.h"
using namespace utils;

#include "querytele.h"
using namespace querytele;

#include "funcexp.h"
#include "jobstep.h"
#include "jlf_common.h"
#include "tupleconstantstep.h"
#include "limitedorderby.h"

#include "tupleannexstep.h"

#define QUEUE_RESERVE_SIZE 100000

namespace
{
struct TAHasher
{
    joblist::TupleAnnexStep* ts;
    utils::Hasher_r h;
    TAHasher(joblist::TupleAnnexStep* t) : ts(t) { }
    uint64_t operator()(const rowgroup::Row::Pointer&) const;
};
struct TAEq
{
    joblist::TupleAnnexStep* ts;
    TAEq(joblist::TupleAnnexStep* t) : ts(t) { }
    bool operator()(const rowgroup::Row::Pointer&, const rowgroup::Row::Pointer&) const;
};
//TODO:  Generalize these and put them back in utils/common/hasher.h
typedef tr1::unordered_set<rowgroup::Row::Pointer, TAHasher, TAEq,
        STLPoolAllocator<rowgroup::Row::Pointer> > DistinctMap_t;
};

inline uint64_t TAHasher::operator()(const Row::Pointer& p) const
{
    Row& row = ts->row1;
    row.setPointer(p);
    return row.hash();
}

inline bool TAEq::operator()(const Row::Pointer& d1, const Row::Pointer& d2) const
{
    Row& r1 = ts->row1, &r2 = ts->row2;
    r1.setPointer(d1);
    r2.setPointer(d2);
    return r1.equals(r2);
}

namespace joblist
{

TupleAnnexStep::TupleAnnexStep(const JobInfo& jobInfo) :
    JobStep(jobInfo),
    fInputDL(NULL),
    fOutputDL(NULL),
    fInputIterator(0),
    fOutputIterator(0),
    fRunner(0),
    fRowsProcessed(0),
    fRowsReturned(0),
    fLimitStart(0),
    fLimitCount(-1),
    fLimitHit(false),
    fEndOfResult(false),
    fDistinct(false),
    fParallelOp(false),
    fOrderBy(NULL),
    fConstant(NULL),
    fFeInstance(funcexp::FuncExp::instance()),
    fJobList(jobInfo.jobListPtr),
    fFinishedThreads(0)
{
    fExtendedInfo = "TNS: ";
    fQtc.stepParms().stepType = StepTeleStats::T_TNS;
}

TupleAnnexStep::~TupleAnnexStep()
{
    if(fParallelOp)
    {
        if(fOrderByList.size() > 0)
        {
            for(uint64_t id = 0; id < fOrderByList.size(); id++)
            {
                delete fOrderByList[id];
            }

            fOrderByList.clear();
        }

        fInputIteratorsList.clear();
        fRunnersList.clear();
    }

    if (fOrderBy)
        delete fOrderBy;

    fOrderBy = NULL;

    if (fConstant)
        delete fConstant;

    fConstant = NULL;
}

void TupleAnnexStep::setOutputRowGroup(const rowgroup::RowGroup& rg)
{
    throw runtime_error("Disabled, use initialize() to set output RowGroup.");
}

void TupleAnnexStep::initialize(const RowGroup& rgIn, const JobInfo& jobInfo)
{
    // Initialize structures used by separate workers
    uint64_t id = 1;
    fRowGroupIn = rgIn;
    fRowGroupIn.initRow(&fRowIn);
    if(fParallelOp && fOrderBy)
    {
        fOrderByList.resize(fMaxThreads+1);
        for(id = 0; id <= fMaxThreads; id++)
        {
            // *DRRTUY use SP here?
            fOrderByList[id] = new LimitedOrderBy();
            fOrderByList[id]->distinct(fDistinct);
            fOrderByList[id]->initialize(rgIn, jobInfo, false, true);
        }
    }
    else
    {
        if (fOrderBy)
        {
            fOrderBy->distinct(fDistinct);
            fOrderBy->initialize(rgIn, jobInfo);
        }
    }

    if (fConstant == NULL)
    {
        vector<uint32_t> oids, oidsIn = rgIn.getOIDs();
        vector<uint32_t> keys, keysIn = rgIn.getKeys();
        vector<uint32_t> scale, scaleIn = rgIn.getScale();
        vector<uint32_t> precision, precisionIn = rgIn.getPrecision();
        vector<CalpontSystemCatalog::ColDataType> types, typesIn = rgIn.getColTypes();
        vector<uint32_t> csNums, csNumsIn = rgIn.getCharsetNumbers();
        vector<uint32_t> pos, posIn = rgIn.getOffsets();
        size_t n = jobInfo.nonConstDelCols.size();

        // Add all columns into output RG as keys. Can we put only keys?
        oids.insert(oids.end(), oidsIn.begin(), oidsIn.begin() + n);
        keys.insert(keys.end(), keysIn.begin(), keysIn.begin() + n);
        scale.insert(scale.end(), scaleIn.begin(), scaleIn.begin() + n);
        precision.insert(precision.end(), precisionIn.begin(), precisionIn.begin() + n);
        types.insert(types.end(), typesIn.begin(), typesIn.begin() + n);
        csNums.insert(csNums.end(), csNumsIn.begin(), csNumsIn.begin() + n);
        pos.insert(pos.end(), posIn.begin(), posIn.begin() + n + 1);

        fRowGroupOut = RowGroup(oids.size(), pos, oids, keys, types, csNums, scale, precision, jobInfo.stringTableThreshold);
    }
    else
    {
        fConstant->initialize(jobInfo, &rgIn);
        fRowGroupOut = fConstant->getOutputRowGroup();
    }

    fRowGroupOut.initRow(&fRowOut);
    fRowGroupDeliver = fRowGroupOut;
}

void TupleAnnexStep::run()
{
    if (fInputJobStepAssociation.outSize() == 0)
        throw logic_error("No input data list for annex step.");

    fInputDL = fInputJobStepAssociation.outAt(0)->rowGroupDL();

    if (fInputDL == NULL)
        throw logic_error("Input is not a RowGroup data list.");

    if (fOutputJobStepAssociation.outSize() == 0)
        throw logic_error("No output data list for annex step.");

    fOutputDL = fOutputJobStepAssociation.outAt(0)->rowGroupDL();

    if (fOutputDL == NULL)
        throw logic_error("Output is not a RowGroup data list.");

    if (fDelivery == true)
    {
        fOutputIterator = fOutputDL->getIterator();
    }

    if(fParallelOp)
    {
        // Indexing begins with 1
        fRunnersList.resize(fMaxThreads);
        fInputIteratorsList.resize(fMaxThreads+1);

        // Activate stats collecting before CS spawns threads.
        if (traceOn()) dlTimes.setFirstReadTime();

        // *DRRTUY Make this block conditional
        StepTeleStats sts;
        sts.query_uuid = fQueryUuid;
        sts.step_uuid = fStepUuid;
        sts.msg_type = StepTeleStats::ST_START;
        sts.total_units_of_work = 1;
        postStepStartTele(sts);

        for(uint32_t id = 1; id <= fMaxThreads; id++)
        {
            fInputIteratorsList[id] = fInputDL->getIterator();
            fRunnersList[id-1] = jobstepThreadPool.invoke(Runner(this, id));
        }
    }
    else
    {
        fInputDL = fInputJobStepAssociation.outAt(0)->rowGroupDL();

        if (fInputDL == NULL)
            throw logic_error("Input is not a RowGroup data list.");

        fInputIterator = fInputDL->getIterator();
        fRunner = jobstepThreadPool.invoke(Runner(this));
    }

}

void TupleAnnexStep::join()
{
    if(fParallelOp)
    {
        jobstepThreadPool.join(fRunnersList);
    }
    else
    {
        if (fRunner)
        {
            jobstepThreadPool.join(fRunner);
        }
    }

}

uint32_t TupleAnnexStep::nextBand(messageqcpp::ByteStream& bs)
{
    RGData rgDataOut;
    bool more = false;
    uint32_t rowCount = 0;

    try
    {
        bs.restart();

        more = fOutputDL->next(fOutputIterator, &rgDataOut);

        if (more && !cancelled())
        {
            fRowGroupDeliver.setData(&rgDataOut);
            fRowGroupDeliver.serializeRGData(bs);
            rowCount = fRowGroupDeliver.getRowCount();
        }
        else
        {
            while (more)
                more = fOutputDL->next(fOutputIterator, &rgDataOut);

            fEndOfResult = true;
        }
    }
    catch (...)
    {
        handleException(std::current_exception(),
                        logging::ERR_IN_DELIVERY,
                        logging::ERR_ALWAYS_CRITICAL,
                        "TupleAnnexStep::nextBand()");
        while (more)
            more = fOutputDL->next(fOutputIterator, &rgDataOut);

        fEndOfResult = true;
    }

    if (fEndOfResult)
    {
        // send an empty / error band
        rgDataOut.reinit(fRowGroupDeliver, 0);
        fRowGroupDeliver.setData(&rgDataOut);
        fRowGroupDeliver.resetRowGroup(0);
        fRowGroupDeliver.setStatus(status());
        fRowGroupDeliver.serializeRGData(bs);
    }

    return rowCount;
}

void TupleAnnexStep::execute()
{
    if (fOrderBy)
        executeWithOrderBy();
    else if (fDistinct)
        executeNoOrderByWithDistinct();
    else
        executeNoOrderBy();

    StepTeleStats sts;
    sts.query_uuid = fQueryUuid;
    sts.step_uuid = fStepUuid;
    sts.msg_type = StepTeleStats::ST_SUMMARY;
    sts.total_units_of_work = sts.units_of_work_completed = 1;
    sts.rows = fRowsReturned;
    postStepSummaryTele(sts);

    if (traceOn())
    {
        if (dlTimes.FirstReadTime().tv_sec == 0)
            dlTimes.setFirstReadTime();

        dlTimes.setLastReadTime();
        dlTimes.setEndOfInputTime();
        printCalTrace();
    }
}

void TupleAnnexStep::execute(uint32_t id)
{
    if(fOrderByList[id])
        executeParallelOrderBy(id);

}

void TupleAnnexStep::executeNoOrderBy()
{
    utils::setThreadName("TASwoOrd");
    RGData rgDataIn;
    RGData rgDataOut;
    bool more = false;

    try
    {
        more = fInputDL->next(fInputIterator, &rgDataIn);

        if (traceOn()) dlTimes.setFirstReadTime();

        StepTeleStats sts;
        sts.query_uuid = fQueryUuid;
        sts.step_uuid = fStepUuid;
        sts.msg_type = StepTeleStats::ST_START;
        sts.total_units_of_work = 1;
        postStepStartTele(sts);

        while (more && !cancelled() && !fLimitHit)
        {
            fRowGroupIn.setData(&rgDataIn);
            fRowGroupIn.getRow(0, &fRowIn);
            // Get a new output rowgroup for each input rowgroup to preserve the rids
            rgDataOut.reinit(fRowGroupOut, fRowGroupIn.getRowCount());
            fRowGroupOut.setData(&rgDataOut);
            fRowGroupOut.resetRowGroup(fRowGroupIn.getBaseRid());
            fRowGroupOut.setDBRoot(fRowGroupIn.getDBRoot());
            fRowGroupOut.getRow(0, &fRowOut);

            for (uint64_t i = 0; i < fRowGroupIn.getRowCount() && !cancelled() && !fLimitHit; ++i)
            {
                // skip first limit-start rows
                if (fRowsProcessed++ < fLimitStart)
                {
                    fRowIn.nextRow();
                    continue;
                }

                if (UNLIKELY(fRowsReturned >= fLimitCount))
                {
                    fLimitHit = true;
                    fJobList->abortOnLimit((JobStep*) this);
                    continue;
                }

                if (fConstant)
                    fConstant->fillInConstants(fRowIn, fRowOut);
                else
                    copyRow(fRowIn, &fRowOut);

                fRowGroupOut.incRowCount();

                if (++fRowsReturned < fLimitCount)
                {
                    fRowOut.nextRow();
                    fRowIn.nextRow();
                }
            }

            if (fRowGroupOut.getRowCount() > 0)
            {
                fOutputDL->insert(rgDataOut);
            }

            more = fInputDL->next(fInputIterator, &rgDataIn);
        }
    }
    catch (...)
    {
        handleException(std::current_exception(),
                        logging::ERR_IN_PROCESS,
                        logging::ERR_ALWAYS_CRITICAL,
                        "TupleAnnexStep::executeNoOrderBy()");
    }

    while (more)
        more = fInputDL->next(fInputIterator, &rgDataIn);

    // Bug 3136, let mini stats to be formatted if traceOn.
    fOutputDL->endOfInput();
}

void TupleAnnexStep::executeNoOrderByWithDistinct()
{
    utils::setThreadName("TASwoOrdDist");
    scoped_ptr<DistinctMap_t> distinctMap(new DistinctMap_t(10, TAHasher(this), TAEq(this)));
    vector<RGData> dataVec;
    vector<RGData> dataVecSkip;
    RGData rgDataIn;
    RGData rgDataOut;
    RGData rgDataSkip;
    RowGroup rowGroupSkip;
    Row rowSkip;
    bool more = false;

    rgDataOut.reinit(fRowGroupOut);
    fRowGroupOut.setData(&rgDataOut);
    fRowGroupOut.resetRowGroup(0);
    fRowGroupOut.getRow(0, &fRowOut);

    fRowGroupOut.initRow(&row1);
    fRowGroupOut.initRow(&row2);

    rowGroupSkip = fRowGroupOut;
    rgDataSkip.reinit(rowGroupSkip);
    rowGroupSkip.setData(&rgDataSkip);
    rowGroupSkip.resetRowGroup(0);
    rowGroupSkip.initRow(&rowSkip);
    rowGroupSkip.getRow(0, &rowSkip);

    try
    {
        more = fInputDL->next(fInputIterator, &rgDataIn);

        if (traceOn()) dlTimes.setFirstReadTime();

        StepTeleStats sts;
        sts.query_uuid = fQueryUuid;
        sts.step_uuid = fStepUuid;
        sts.msg_type = StepTeleStats::ST_START;
        sts.total_units_of_work = 1;
        postStepStartTele(sts);

        while (more && !cancelled() && !fLimitHit)
        {
            fRowGroupIn.setData(&rgDataIn);
            fRowGroupIn.getRow(0, &fRowIn);

            for (uint64_t i = 0; i < fRowGroupIn.getRowCount() && !cancelled() && !fLimitHit; ++i)
            {
                pair<DistinctMap_t::iterator, bool> inserted;
                Row* rowPtr;

                if (distinctMap->size() < fLimitStart)
                    rowPtr = &rowSkip;
                else
                    rowPtr = &fRowOut;

                if (fConstant)
                    fConstant->fillInConstants(fRowIn, *rowPtr);
                else
                    copyRow(fRowIn, rowPtr);

                fRowIn.nextRow();
                inserted = distinctMap->insert(rowPtr->getPointer());
                ++fRowsProcessed;

                if (inserted.second)
                {
                    if (UNLIKELY(fRowsReturned >= fLimitCount))
                    {
                        fLimitHit = true;
                        fJobList->abortOnLimit((JobStep*) this);
                        break;
                    }

                    // skip first limit-start rows
                    if (distinctMap->size() <= fLimitStart)
                    {
                        rowGroupSkip.incRowCount();
                        rowSkip.nextRow();
                        if (UNLIKELY(rowGroupSkip.getRowCount() >= rowgroup::rgCommonSize))
                        {
                            // allocate new RGData for skipped rows below the fLimitStart
                            // offset (do not take it into account in RM assuming there
                            // are few skipped rows
                            dataVecSkip.push_back(rgDataSkip);
                            rgDataSkip.reinit(rowGroupSkip);
                            rowGroupSkip.setData(&rgDataSkip);
                            rowGroupSkip.resetRowGroup(0);
                            rowGroupSkip.getRow(0, &rowSkip);
                        }
                        continue;
                    }

                    ++fRowsReturned;

                    fRowGroupOut.incRowCount();
                    fRowOut.nextRow();

                    if (UNLIKELY(fRowGroupOut.getRowCount() >= rowgroup::rgCommonSize))
                    {
                        dataVec.push_back(rgDataOut);
                        rgDataOut.reinit(fRowGroupOut);
                        fRowGroupOut.setData(&rgDataOut);
                        fRowGroupOut.resetRowGroup(0);
                        fRowGroupOut.getRow(0, &fRowOut);
                    }
                }
            }

            more = fInputDL->next(fInputIterator, &rgDataIn);
        }

        if (fRowGroupOut.getRowCount() > 0)
            dataVec.push_back(rgDataOut);

        for (vector<RGData>::iterator i = dataVec.begin(); i != dataVec.end(); i++)
        {
            rgDataOut = *i;
            fRowGroupOut.setData(&rgDataOut);
            fOutputDL->insert(rgDataOut);
        }
    }
    catch (...)
    {
        handleException(std::current_exception(),
                        logging::ERR_IN_PROCESS,
                        logging::ERR_ALWAYS_CRITICAL,
                        "TupleAnnexStep::executeNoOrderByWithDistinct()");
    }

    while (more)
        more = fInputDL->next(fInputIterator, &rgDataIn);

    // Bug 3136, let mini stats to be formatted if traceOn.
    fOutputDL->endOfInput();
}

void TupleAnnexStep::executeWithOrderBy()
{
    utils::setThreadName("TASwOrd");
    RGData rgDataIn;
    RGData rgDataOut;
    bool more = false;

    try
    {
        more = fInputDL->next(fInputIterator, &rgDataIn);

        if (traceOn()) dlTimes.setFirstReadTime();

        StepTeleStats sts;
        sts.query_uuid = fQueryUuid;
        sts.step_uuid = fStepUuid;
        sts.msg_type = StepTeleStats::ST_START;
        sts.total_units_of_work = 1;
        postStepStartTele(sts);

        while (more && !cancelled())
        {
            fRowGroupIn.setData(&rgDataIn);
            fRowGroupIn.getRow(0, &fRowIn);

            for (uint64_t i = 0; i < fRowGroupIn.getRowCount() && !cancelled(); ++i)
            {
                fOrderBy->processRow(fRowIn);
                fRowIn.nextRow();
            }

            more = fInputDL->next(fInputIterator, &rgDataIn);
        }

        fOrderBy->finalize();

        if (!cancelled())
        {
            while (fOrderBy->getData(rgDataIn))
            {
                if (fConstant == NULL &&
                        fRowGroupOut.getColumnCount() == fRowGroupIn.getColumnCount())
                {
                    rgDataOut = rgDataIn;
                    fRowGroupOut.setData(&rgDataOut);
                }
                else
                {
                    fRowGroupIn.setData(&rgDataIn);
                    fRowGroupIn.getRow(0, &fRowIn);

                    rgDataOut.reinit(fRowGroupOut, fRowGroupIn.getRowCount());
                    fRowGroupOut.setData(&rgDataOut);
                    fRowGroupOut.resetRowGroup(fRowGroupIn.getBaseRid());
                    fRowGroupOut.setDBRoot(fRowGroupIn.getDBRoot());
                    fRowGroupOut.getRow(0, &fRowOut);

                    for (uint64_t i = 0; i < fRowGroupIn.getRowCount(); ++i)
                    {
                        if (fConstant)
                            fConstant->fillInConstants(fRowIn, fRowOut);
                        else
                            copyRow(fRowIn, &fRowOut);

                        fRowGroupOut.incRowCount();
                        fRowOut.nextRow();
                        fRowIn.nextRow();
                    }
                }

                if (fRowGroupOut.getRowCount() > 0)
                {
                    fRowsReturned += fRowGroupOut.getRowCount();
                    fOutputDL->insert(rgDataOut);
                }
            }
        }
    }
    catch (...)
    {
        handleException(std::current_exception(),
                        logging::ERR_IN_PROCESS,
                        logging::ERR_ALWAYS_CRITICAL,
                        "TupleAnnexStep::executeWithOrderBy()");
    }

    while (more)
        more = fInputDL->next(fInputIterator, &rgDataIn);

    // Bug 3136, let mini stats to be formatted if traceOn.
    fOutputDL->endOfInput();
}

/*
    The m() iterates over thread's LimitedOrderBy instances,
    reverts the rules and then populates the final collection
    used for final sorting. The method uses OrderByRow that
    combination of Row::data and comparison rules.
    When m() finishes with thread's LOBs it iterates over
    final sorting collection, populates rgDataOut, then
    sends it into outputDL.
    Changing this method don't forget to make changes in
    finalizeParallelOrderBy() that is a clone.
    !!!The method doesn't set Row::baseRid
*/
void TupleAnnexStep::finalizeParallelOrderByDistinct()
{
    utils::setThreadName("TASwParOrdDistM");
    uint64_t count = 0;
    uint64_t offset = 0;
    uint32_t rowSize = 0;

    rowgroup::RGData rgDataOut;
    rgDataOut.reinit(fRowGroupOut, rowgroup::rgCommonSize);
    fRowGroupOut.setData(&rgDataOut);
    fRowGroupOut.resetRowGroup(0);
    // Calculate offset here
    fRowGroupOut.getRow(0, &fRowOut);
    ordering::SortingPQ finalPQ;
    scoped_ptr<DistinctMap_t> distinctMap(new DistinctMap_t(10, TAHasher(this), TAEq(this)));
    fRowGroupIn.initRow(&row1);
    fRowGroupIn.initRow(&row2);

    try
    {
        for(uint64_t id = 1; id <= fMaxThreads; id++)
        {
            if (cancelled())
            {
                break;
            }
            // Revert the ordering rules before we
            // add rows into the final PQ.
            fOrderByList[id]->getRule().revertRules();
            ordering::SortingPQ &currentPQ = fOrderByList[id]->getQueue();
            finalPQ.reserve(finalPQ.size()+currentPQ.size());
            pair<DistinctMap_t::iterator, bool> inserted;
            while (currentPQ.size())
            {
                ordering::OrderByRow &topOBRow =
                    const_cast<ordering::OrderByRow&>(currentPQ.top());
                inserted = distinctMap->insert(topOBRow.fData);
                if (inserted.second)
                {
                    finalPQ.push(topOBRow);
                }
                currentPQ.pop();
            }
        }
    }
    catch (...)
    {
        handleException(std::current_exception(),
                        logging::ERR_IN_PROCESS,
                        logging::ERR_ALWAYS_CRITICAL,
                        "TupleAnnexStep::finalizeParallelOrderByDistinct()");
    }

    // OFFSET processing
    while (finalPQ.size() && offset < fLimitStart)
    {
        offset++;
        finalPQ.pop();
    }

    // Calculate rowSize only once
    if (finalPQ.size())
    {
        ordering::OrderByRow& topOBRow =
            const_cast<ordering::OrderByRow&>(finalPQ.top());
        fRowIn.setData(topOBRow.fData);
        if (!fConstant)
        {
            copyRow(fRowIn, &fRowOut);
        }
        else
        {
            fConstant->fillInConstants(fRowIn, fRowOut);
        }
        rowSize = fRowOut.getSize();
        fRowGroupOut.incRowCount();
        fRowOut.nextRow(rowSize);
        finalPQ.pop();
        count++;
    }

    if (!fConstant)
    {
        while(finalPQ.size())
        {
            if (cancelled())
            {
                break;
            }

            while (count < fLimitCount && finalPQ.size()
                    && fRowGroupOut.getRowCount() < rowgroup::rgCommonSize)
            {
                ordering::OrderByRow &topOBRow =
                    const_cast<ordering::OrderByRow&>(finalPQ.top());

                fRowIn.setData(topOBRow.fData);
                copyRow(fRowIn, &fRowOut);
                fRowGroupOut.incRowCount();
                fRowOut.nextRow(rowSize);

                finalPQ.pop();
                count++;
                if (fRowGroupOut.getRowCount() == rowgroup::rgCommonSize)
                {
                    break;
                }
            }

            if (fRowGroupOut.getRowCount() > 0)
            {
                fRowsReturned += fRowGroupOut.getRowCount();
                fOutputDL->insert(rgDataOut);
                rgDataOut.reinit(fRowGroupIn, rowgroup::rgCommonSize);
                fRowGroupOut.setData(&rgDataOut);
                fRowGroupOut.resetRowGroup(0);
                fRowGroupOut.getRow(0, &fRowOut);
            }
            else
            {
                break;
            }
        } // end of limit bound while loop
    }
    else // Add ConstantColumns striped earlier
    {
        while(finalPQ.size())
        {
            if (cancelled())
            {
                break;
            }

            while (count < fLimitCount && finalPQ.size()
                    && fRowGroupOut.getRowCount() < rowgroup::rgCommonSize)
            {
                ordering::OrderByRow &topOBRow =
                    const_cast<ordering::OrderByRow&>(finalPQ.top());

                fRowIn.setData(topOBRow.fData);
                fConstant->fillInConstants(fRowIn, fRowOut);
                fRowGroupOut.incRowCount();
                fRowOut.nextRow(rowSize);

                finalPQ.pop();
                count++;
                if (fRowGroupOut.getRowCount() == rowgroup::rgCommonSize)
                {
                    break;
                }
            }

            if (fRowGroupOut.getRowCount() > 0)
            {
                fRowsReturned += fRowGroupOut.getRowCount();
                fOutputDL->insert(rgDataOut);
                rgDataOut.reinit(fRowGroupOut, rowgroup::rgCommonSize);
                fRowGroupOut.setData(&rgDataOut);
                fRowGroupOut.resetRowGroup(0);
                fRowGroupOut.getRow(0, &fRowOut);
            }
            else
            {
                    break;
            }
        } // end of limit bound while loop
    } // end of if-else

    if (fRowGroupOut.getRowCount() > 0)
    {
        fRowsReturned += fRowGroupOut.getRowCount();
        fOutputDL->insert(rgDataOut);
    }

    fOutputDL->endOfInput();

    StepTeleStats sts;
    sts.query_uuid = fQueryUuid;
    sts.step_uuid = fStepUuid;
    sts.msg_type = StepTeleStats::ST_SUMMARY;
    sts.total_units_of_work = sts.units_of_work_completed = 1;
    sts.rows = fRowsReturned;
    postStepSummaryTele(sts);

    if (traceOn())
    {
        if (dlTimes.FirstReadTime().tv_sec == 0)
            dlTimes.setFirstReadTime();

        dlTimes.setLastReadTime();
        dlTimes.setEndOfInputTime();
        printCalTrace();
    }
}

/*
    The m() iterates over thread's LimitedOrderBy instances,
    reverts the rules and then populates the final collection
    used for final sorting. The method uses OrderByRow that
    combination of Row::data and comparison rules.
    When m() finishes with thread's LOBs it iterates over
    final sorting collection, populates rgDataOut, then
    sends it into outputDL.
    Changing this method don't forget to make changes in
    finalizeParallelOrderByDistinct() that is a clone.
    !!!The method doesn't set Row::baseRid
*/
void TupleAnnexStep::finalizeParallelOrderBy()
{
    utils::setThreadName("TASwParOrdMerge");
    uint64_t count = 0;
    uint64_t offset = 0;
    uint32_t rowSize = 0;

    rowgroup::RGData rgDataOut;
    ordering::SortingPQ finalPQ;
    rgDataOut.reinit(fRowGroupOut, rowgroup::rgCommonSize);
    fRowGroupOut.setData(&rgDataOut);
    fRowGroupOut.resetRowGroup(0);
    // Calculate offset here
    fRowGroupOut.getRow(0, &fRowOut);

    try
    {
        for(uint64_t id = 1; id <= fMaxThreads; id++)
        {
            if (cancelled())
            {
                break;
            }
            // Revert the ordering rules before we
            // add rows into the final PQ.
            fOrderByList[id]->getRule().revertRules();
            ordering::SortingPQ &currentPQ = fOrderByList[id]->getQueue();
            finalPQ.reserve(currentPQ.size());
            while (currentPQ.size())
            {
                ordering::OrderByRow &topOBRow =
                    const_cast<ordering::OrderByRow&>(currentPQ.top());
                finalPQ.push(topOBRow);
                currentPQ.pop();
            }
        }
    }
    catch (...)
    {
        handleException(std::current_exception(),
                        logging::ERR_IN_PROCESS,
                        logging::ERR_ALWAYS_CRITICAL,
                        "TupleAnnexStep::finalizeParallelOrderBy()");
    }

    // OFFSET processing
    while (finalPQ.size() && offset < fLimitStart)
    {
        offset++;
        finalPQ.pop();
    }

    // Calculate rowSize only once
    if (finalPQ.size())
    {
        ordering::OrderByRow& topOBRow =
            const_cast<ordering::OrderByRow&>(finalPQ.top());
        fRowIn.setData(topOBRow.fData);
        if (!fConstant)
        {
            copyRow(fRowIn, &fRowOut);
        }
        else
        {
            fConstant->fillInConstants(fRowIn, fRowOut);
        }
        rowSize = fRowOut.getSize();
        fRowGroupOut.incRowCount();
        fRowOut.nextRow(rowSize);
        finalPQ.pop();
        count++;
    }

    if (!fConstant)
    {
        while(finalPQ.size())
        {
            if (cancelled())
            {
                break;
            }

            while (count < fLimitCount && finalPQ.size()
                    && fRowGroupOut.getRowCount() < rowgroup::rgCommonSize)
            {
                ordering::OrderByRow &topOBRow =
                    const_cast<ordering::OrderByRow&>(finalPQ.top());

                fRowIn.setData(topOBRow.fData);
                copyRow(fRowIn, &fRowOut);
                fRowGroupOut.incRowCount();
                fRowOut.nextRow(rowSize);

                finalPQ.pop();
                count++;
            }

            if (fRowGroupOut.getRowCount() > 0)
            {
                fRowsReturned += fRowGroupOut.getRowCount();
                fOutputDL->insert(rgDataOut);
                rgDataOut.reinit(fRowGroupIn, rowgroup::rgCommonSize);
                fRowGroupOut.setData(&rgDataOut);
                fRowGroupOut.resetRowGroup(0);
                fRowGroupOut.getRow(0, &fRowOut);
            }
            else
            {
                break;
            }
        } // end of limit bound while loop
    }
    else // Add ConstantColumns striped earlier
    {
        while(finalPQ.size())
        {
            if (cancelled())
            {
                break;
            }

            while (count < fLimitCount && finalPQ.size()
                    && fRowGroupOut.getRowCount() < rowgroup::rgCommonSize)
            {
                ordering::OrderByRow &topOBRow =
                    const_cast<ordering::OrderByRow&>(finalPQ.top());

                fRowIn.setData(topOBRow.fData);
                fConstant->fillInConstants(fRowIn, fRowOut);
                fRowGroupOut.incRowCount();
                fRowOut.nextRow(rowSize);

                finalPQ.pop();
                count++;
                if (fRowGroupOut.getRowCount() == rowgroup::rgCommonSize)
                {
                    break;
                }
            }

            if (fRowGroupOut.getRowCount() > 0)
            {
                fRowsReturned += fRowGroupOut.getRowCount();
                fOutputDL->insert(rgDataOut);
                rgDataOut.reinit(fRowGroupOut, rowgroup::rgCommonSize);
                fRowGroupOut.setData(&rgDataOut);
                fRowGroupOut.resetRowGroup(0);
                fRowGroupOut.getRow(0, &fRowOut);
            }
            else
            {
                break;
            }
        } // end of limit bound while loop
    } // end of if-else

    if (fRowGroupOut.getRowCount() > 0)
    {
        fRowsReturned += fRowGroupOut.getRowCount();
        fOutputDL->insert(rgDataOut);
    }

    fOutputDL->endOfInput();

    StepTeleStats sts;
    sts.query_uuid = fQueryUuid;
    sts.step_uuid = fStepUuid;
    sts.msg_type = StepTeleStats::ST_SUMMARY;
    sts.total_units_of_work = sts.units_of_work_completed = 1;
    sts.rows = fRowsReturned;
    postStepSummaryTele(sts);

    if (traceOn())
    {
        if (dlTimes.FirstReadTime().tv_sec == 0)
            dlTimes.setFirstReadTime();

        dlTimes.setLastReadTime();
        dlTimes.setEndOfInputTime();
        printCalTrace();
    }
}

void TupleAnnexStep::executeParallelOrderBy(uint64_t id)
{
    utils::setThreadName("TASwParOrd");
    RGData rgDataIn;
    RGData rgDataOut;
    bool more = false;
    uint64_t dlOffset = 0;
    uint32_t rowSize = 0;

    uint64_t rowCount = 0;
    uint64_t doubleRGSize = 2*rowgroup::rgCommonSize;
    rowgroup::Row r = fRowIn;
    rowgroup::RowGroup rg = fRowGroupIn;
    rg.initRow(&r);
    LimitedOrderBy *limOrderBy = fOrderByList[id];
    ordering::SortingPQ &currentPQ = limOrderBy->getQueue();
    if (limOrderBy->getLimitCount() < QUEUE_RESERVE_SIZE)
    {
        currentPQ.reserve(limOrderBy->getLimitCount());
    }
    else
    {
        currentPQ.reserve(QUEUE_RESERVE_SIZE);
    }

    try
    {
        more = fInputDL->next(fInputIteratorsList[id], &rgDataIn);
        if (more) dlOffset++;

        while (more && !cancelled())
        {
            if (dlOffset%fMaxThreads == id-1)
            {
                if (cancelled())
                    break;

                if (currentPQ.capacity()-currentPQ.size() < doubleRGSize)
                {
                    currentPQ.reserve(QUEUE_RESERVE_SIZE);
                }

                rg.setData(&rgDataIn);
                rg.getRow(0, &r);
                if (!rowSize)
                {
                    rowSize = r.getSize();
                }
                rowCount = rg.getRowCount();

                for (uint64_t i = 0; i < rowCount; ++i)
                {
                    limOrderBy->processRow(r);
                    r.nextRow(rowSize);
               }
            }

            // *DRRTUY Implement a method to skip elements in FIFO
            more = fInputDL->next(fInputIteratorsList[id], &rgDataIn);
            if(more) dlOffset++;
        }
    }
    catch (...)
    {
        handleException(std::current_exception(),
                        logging::ERR_IN_PROCESS,
                        logging::ERR_ALWAYS_CRITICAL,
                        "TupleAnnexStep::executeParallelOrderBy()");
    }

    // read out the input DL
    while (more)
        more = fInputDL->next(fInputIteratorsList[id], &rgDataIn);

    // Count finished sorting threads under mutex and run final
    // sort step when the last thread converges
    fParallelFinalizeMutex.lock();
    fFinishedThreads++;
    if (fFinishedThreads == fMaxThreads)
    {
        fParallelFinalizeMutex.unlock();
        if(fDistinct)
        {
            finalizeParallelOrderByDistinct();
        }
        else
        {
            finalizeParallelOrderBy();
        }
    }
    else
    {
        fParallelFinalizeMutex.unlock();
    }
}

const RowGroup& TupleAnnexStep::getOutputRowGroup() const
{
    return fRowGroupOut;
}

const RowGroup& TupleAnnexStep::getDeliveredRowGroup() const
{
    return fRowGroupDeliver;
}

void TupleAnnexStep::deliverStringTableRowGroup(bool b)
{
    fRowGroupOut.setUseStringTable(b);
    fRowGroupDeliver.setUseStringTable(b);
}

bool TupleAnnexStep::deliverStringTableRowGroup() const
{
    idbassert(fRowGroupOut.usesStringTable() == fRowGroupDeliver.usesStringTable());
    return fRowGroupDeliver.usesStringTable();
}

const string TupleAnnexStep::toString() const
{
    ostringstream oss;
    oss << "AnnexStep ";
    oss << "  ses:" << fSessionId << " txn:" << fTxnId << " st:" << fStepId;

    oss << " in:";

    for (unsigned i = 0; i < fInputJobStepAssociation.outSize(); i++)
        oss << fInputJobStepAssociation.outAt(i);

    oss << " out:";

    for (unsigned i = 0; i < fOutputJobStepAssociation.outSize(); i++)
        oss << fOutputJobStepAssociation.outAt(i);

    if (fOrderBy)
        oss << "    " << fOrderBy->toString();

    if (fConstant)
        oss << "    " << fConstant->toString();

    oss << endl;

    return oss.str();
}

void TupleAnnexStep::printCalTrace()
{
    time_t t = time (0);
    char timeString[50];
    ctime_r (&t, timeString);
    timeString[strlen (timeString ) - 1] = '\0';
    ostringstream logStr;
    logStr  << "ses:" << fSessionId << " st: " << fStepId << " finished at " << timeString
            << "; total rows returned-" << fRowsReturned << endl
            << "\t1st read " << dlTimes.FirstReadTimeString()
            << "; EOI " << dlTimes.EndOfInputTimeString() << "; runtime-"
            << JSTimeStamp::tsdiffstr(dlTimes.EndOfInputTime(), dlTimes.FirstReadTime())
            << "s;\n\tUUID " << uuids::to_string(fStepUuid) << endl
            << "\tJob completion status " << status() << endl;
    logEnd(logStr.str().c_str());
    fExtendedInfo += logStr.str();
    formatMiniStats();
}

void TupleAnnexStep::formatMiniStats()
{
    ostringstream oss;
    oss << "TNS ";
    oss << "UM "
        << "- "
        << "- "
        << "- "
        << "- "
        << "- "
        << "- "
        << JSTimeStamp::tsdiffstr(dlTimes.EndOfInputTime(), dlTimes.FirstReadTime()) << " "
        << fRowsReturned << " ";
    fMiniInfo += oss.str();
}

}   //namespace
// vim:ts=4 sw=4:

max_stars_repo_name: zettadb/zettalib

max_issues_repo_name: zettadb/zettalib

max_forks_repo_name: zettadb/zettalib


text:
#ifndef _MODEL_HPP_
#define _MODEL_HPP_

#include "Trajectory.h"
#include <vector>
#include "Policy.h"
#include <Eigen/Dense>

using namespace std;
using namespace Eigen;

/*
This class builds a model of an MDPs with tabular states and actions, and a finite  horizon, L.
*/

class Model {
public:
	// Adding a defauly constructor so that I can extend it
	Model();

	/*
	Take historical data in trajs and build a model. numState and numActions are the total numbers of possible states and actions
	*/
	Model(const vector<Trajectory*> & trajs, int numStates, int numActions, int L, bool JiangStyle);

	/*
	Use the model to directly estimate the value of a policy. This uses value iteration.
	*/
	//double getPolicyValue(const Policy & pi) const;

	// V and Q predictions - can be loaded for a specific evaluation policy
	void loadEvalPolicy(const Policy & pi, const int & L); // int L = Gridworld::getMaxTrajLen();
	vector<VectorXd> actionProbabilities; // [s][a]
	vector<VectorXd> V; // [t](s) - t in [0,L].
	vector<MatrixXd> Q; // [t](s,a)
	//vector<MatrixXd> Rsa; // [t](s,a) - Prediction of R_t given that S_0 =s and A_0=a
	//vector<vector<MatrixXd>> Rsas; // [t][s](a,sPrime) - Prediction of R_t given that S_0 =s and A_0=a, and S_1 = sPrime
	//MatrixXd Rs; // (s,t)
	double evalPolicyValue;

	// Generate trajectories from the provided policy
	vector<Trajectory> generateTrajectories(const Policy & pi, int N, mt19937_64 & generator) const;

	// Estimate value of policy under model using Monte Carlo returns
  double evalMonteCarlo(const Policy & pi, int N, mt19937_64 & generator) const;

	// We exposed these variables in order to do DRv2 (version 2 implementations) without having to compute Rsa and Rsas tables above (compute intensive)
	vector<vector<vector<double>>> R;				// R[s][a][s']. Size = [numStates][numActions][numStates+1]
	vector<vector<vector<double>>> P;				// P[s][a][s']. Size = [numStates][numActions][numStates+1]

//private:
	int N;
	int L;
	int numStates;
	int numActions;
	vector<double> d0;								// d0[s] = Pr(S_0=s). Size = [numStates]

	// How many times was each (s), (s,a), and (s,a,s') tuple seen?
	vector<vector<int>> stateActionCounts;
	vector<vector<int>> stateActionCounts_includingHorizon;				 // Includes transitions to terminal absorbing state due to time horizon
	vector<vector<vector<int>>> stateActionStateCounts;
	vector<vector<vector<int>>> stateActionStateCounts_includingHorizon; // Includes transitions to terminal absorbing state due to time horizon
};

#endif

ext: hpp

max_stars_repo_name: LARG/regression-importance-sampling

max_issues_repo_name: LARG/regression-importance-sampling

max_forks_repo_name: LARG/regression-importance-sampling


text:
#include <facter/facts/resolvers/uptime_resolver.hpp>
#include <facter/facts/collection.hpp>
#include <facter/facts/fact.hpp>
#include <facter/facts/scalar_value.hpp>
#include <facter/facts/map_value.hpp>
#include <boost/format.hpp>

using namespace std;

namespace facter { namespace facts { namespace resolvers {

    uptime_resolver::uptime_resolver() :
        resolver(
            "uptime",
            {
                fact::system_uptime,
                fact::uptime,
                fact::uptime_days,
                fact::uptime_hours,
                fact::uptime_seconds
            })
    {
    }

    void uptime_resolver::resolve(collection& facts)
    {
        auto seconds = get_uptime();
        if (seconds < 0) {
            return;
        }

        auto minutes = (seconds / 60) % 60;
        auto hours = seconds / (60 * 60);
        auto days = seconds / (60 * 60 * 24);
        string uptime;

        switch (days) {
            case 0:
                uptime = (boost::format("%d:%02d hours") % hours % minutes).str();
                break;
            case 1:
                uptime = "1 day";
                break;
            default:
                uptime = (boost::format("%d days") % days).str();
                break;
        }

        // Add hidden facts
        facts.add(fact::uptime_seconds, make_value<integer_value>(seconds, true));
        facts.add(fact::uptime_hours, make_value<integer_value>(hours, true));
        facts.add(fact::uptime_days, make_value<integer_value>(days, true));
        facts.add(fact::uptime, make_value<string_value>(uptime, true));

        auto value = make_value<map_value>();
        value->add("seconds", make_value<integer_value>(seconds));
        value->add("hours", make_value<integer_value>(hours));
        value->add("days", make_value<integer_value>(days));
        value->add("uptime", make_value<string_value>(move(uptime)));
        facts.add(fact::system_uptime, move(value));
    }

}}}  // namespace facter::facts::resolvers

ext: cc

max_stars_repo_name: whopper/cfacter

max_issues_repo_name: whopper/cfacter

max_forks_repo_name: whopper/cfacter


text:
#include "LtpEngine.h"
#include <iostream>
#include <inttypes.h>
#include <boost/bind/bind.hpp>
#include <boost/make_unique.hpp>

static const boost::posix_time::time_duration static_tokenMaxLimitDurationWindow(boost::posix_time::milliseconds(100));
static const boost::posix_time::time_duration static_tokenRefreshTimeDurationWindow(boost::posix_time::milliseconds(20));

LtpEngine::LtpEngine(const uint64_t thisEngineId, const uint8_t engineIndexForEncodingIntoRandomSessionNumber,
    const uint64_t mtuClientServiceData, uint64_t mtuReportSegment,
    const boost::posix_time::time_duration & oneWayLightTime, const boost::posix_time::time_duration & oneWayMarginTime,
    const uint64_t ESTIMATED_BYTES_TO_RECEIVE_PER_SESSION, const uint64_t maxRedRxBytesPerSession, bool startIoServiceThread,
    uint32_t checkpointEveryNthDataPacketSender, uint32_t maxRetriesPerSerialNumber, const bool force32BitRandomNumbers, const uint64_t maxSendRateBitsPerSecOrZeroToDisable) :
    M_ESTIMATED_BYTES_TO_RECEIVE_PER_SESSION(ESTIMATED_BYTES_TO_RECEIVE_PER_SESSION),
    M_MAX_RED_RX_BYTES_PER_SESSION(maxRedRxBytesPerSession),
    M_THIS_ENGINE_ID(thisEngineId),
    M_MTU_CLIENT_SERVICE_DATA(mtuClientServiceData),
    M_ONE_WAY_LIGHT_TIME(oneWayLightTime),
    M_ONE_WAY_MARGIN_TIME(oneWayMarginTime),
    M_TRANSMISSION_TO_ACK_RECEIVED_TIME((oneWayLightTime * 2) + (oneWayMarginTime * 2)),
    M_FORCE_32_BIT_RANDOM_NUMBERS(force32BitRandomNumbers),
    m_checkpointEveryNthDataPacketSender(checkpointEveryNthDataPacketSender),
    m_maxRetriesPerSerialNumber(maxRetriesPerSerialNumber),
    m_workLtpEnginePtr(boost::make_unique< boost::asio::io_service::work>(m_ioServiceLtpEngine)),
    m_timeManagerOfCancelSegments(m_ioServiceLtpEngine, oneWayLightTime, oneWayMarginTime, boost::bind(&LtpEngine::CancelSegmentTimerExpiredCallback, this, boost::placeholders::_1, boost::placeholders::_2)),
    m_tokenRefreshTimer(m_ioServiceLtpEngine),
    m_maxSendRateBitsPerSecOrZeroToDisable(maxSendRateBitsPerSecOrZeroToDisable),
    m_tokenRefreshTimerIsRunning(false),
    m_lastTimeTokensWereRefreshed(boost::posix_time::special_values::neg_infin)
{
    m_ltpRxStateMachine.SetCancelSegmentContentsReadCallback(boost::bind(&LtpEngine::CancelSegmentReceivedCallback, this,
        boost::placeholders::_1, boost::placeholders::_2, boost::placeholders::_3,
        boost::placeholders::_4, boost::placeholders::_5));
    m_ltpRxStateMachine.SetCancelAcknowledgementSegmentContentsReadCallback(boost::bind(&LtpEngine::CancelAcknowledgementSegmentReceivedCallback, this,
        boost::placeholders::_1, boost::placeholders::_2, boost::placeholders::_3,
        boost::placeholders::_4));
    m_ltpRxStateMachine.SetReportAcknowledgementSegmentContentsReadCallback(boost::bind(&LtpEngine::ReportAcknowledgementSegmentReceivedCallback, this,
        boost::placeholders::_1, boost::placeholders::_2, boost::placeholders::_3,
        boost::placeholders::_4));
    m_ltpRxStateMachine.SetReportSegmentContentsReadCallback(boost::bind(&LtpEngine::ReportSegmentReceivedCallback, this,
        boost::placeholders::_1, boost::placeholders::_2, boost::placeholders::_3,
        boost::placeholders::_4));
    m_ltpRxStateMachine.SetDataSegmentContentsReadCallback(boost::bind(&LtpEngine::DataSegmentReceivedCallback, this,
        boost::placeholders::_1, boost::placeholders::_2, boost::placeholders::_3,
        boost::placeholders::_4, boost::placeholders::_5, boost::placeholders::_6));

    m_rng.SetEngineIndex(engineIndexForEncodingIntoRandomSessionNumber);
    SetMtuReportSegment(mtuReportSegment);

    UpdateRate(m_maxSendRateBitsPerSecOrZeroToDisable);
    if (m_maxSendRateBitsPerSecOrZeroToDisable) {
        const uint64_t tokenLimit = m_tokenRateLimiter.GetRemainingTokens();
        std::cout << "LtpEngine: rate bitsPerSec = " << m_maxSendRateBitsPerSecOrZeroToDisable << "  token limit = " << tokenLimit << "\n";
    }

    Reset();

    if (startIoServiceThread) {
        m_ioServiceLtpEngineThreadPtr = boost::make_unique<boost::thread>(boost::bind(&boost::asio::io_service::run, &m_ioServiceLtpEngine));
    }
}

LtpEngine::~LtpEngine() {
    std::cout << "LtpEngine Stats:" << std::endl; //print before reset
    std::cout << "m_numCheckpointTimerExpiredCallbacks: " << m_numCheckpointTimerExpiredCallbacks << std::endl;
    std::cout << "m_numDiscretionaryCheckpointsNotResent: " << m_numDiscretionaryCheckpointsNotResent << std::endl;
    std::cout << "m_numReportSegmentTimerExpiredCallbacks: " << m_numReportSegmentTimerExpiredCallbacks << std::endl;
    std::cout << "m_numReportSegmentsUnableToBeIssued: " << m_numReportSegmentsUnableToBeIssued << std::endl;
    std::cout << "m_numReportSegmentsTooLargeAndNeedingSplit: " << m_numReportSegmentsTooLargeAndNeedingSplit << std::endl;
    std::cout << "m_numReportSegmentsCreatedViaSplit: " << m_numReportSegmentsCreatedViaSplit << std::endl;
    std::cout << "m_countAsyncSendsLimitedByRate " << m_countAsyncSendsLimitedByRate << std::endl << std::endl;

    if (m_ioServiceLtpEngineThreadPtr) {
        boost::asio::post(m_ioServiceLtpEngine, boost::bind(&LtpEngine::Reset, this));
        m_workLtpEnginePtr.reset(); //erase the work object (destructor is thread safe) so that io_service thread will exit when it runs out of work
        m_ioServiceLtpEngineThreadPtr->join();
        m_ioServiceLtpEngineThreadPtr.reset(); //delete it
    }
}

void LtpEngine::Reset() {
    m_mapSessionNumberToSessionSender.clear();
    m_mapSessionIdToSessionReceiver.clear();
    m_ltpRxStateMachine.InitRx();
    m_sendersIterator = m_mapSessionNumberToSessionSender.begin();
    m_receiversIterator = m_mapSessionIdToSessionReceiver.begin();
    m_closedSessionDataToSend.clear();
    m_timeManagerOfCancelSegments.Reset();
    m_listCancelSegmentTimerInfo.clear();
    m_listSendersNeedingDeleted.clear();
    m_listReceiversNeedingDeleted.clear();

    m_countAsyncSendsLimitedByRate = 0;
    m_numCheckpointTimerExpiredCallbacks = 0;
    m_numDiscretionaryCheckpointsNotResent = 0;
    m_numReportSegmentTimerExpiredCallbacks = 0;
    m_numReportSegmentsUnableToBeIssued = 0;
    m_numReportSegmentsTooLargeAndNeedingSplit = 0;
    m_numReportSegmentsCreatedViaSplit = 0;
}

void LtpEngine::SetCheckpointEveryNthDataPacketForSenders(uint64_t checkpointEveryNthDataPacketSender) {
    m_checkpointEveryNthDataPacketSender = checkpointEveryNthDataPacketSender;
}

void LtpEngine::SetMtuReportSegment(uint64_t mtuReportSegment) {
    //(5 * 10) + (receptionClaims.size() * (2 * 10)); //5 sdnvs * 10 bytes sdnv max + reception claims * 2sdnvs per claim
    //70 bytes worst case minimum for 1 claim
    if (mtuReportSegment < 70) {
        std::cerr << "error in LtpEngine::SetMtuReportSegment: mtuReportSegment must be at least 70 bytes!!!!.. setting to 70 bytes" << std::endl;
        m_maxReceptionClaims = 1;
    }
    m_maxReceptionClaims = (mtuReportSegment - 50) / 20;
    std::cout << "max reception claims = " << m_maxReceptionClaims << std::endl;
}

bool LtpEngine::PacketIn(const uint8_t * data, const std::size_t size, Ltp::SessionOriginatorEngineIdDecodedCallback_t * sessionOriginatorEngineIdDecodedCallbackPtr) {
    std::string errorMessage;
    const bool success = m_ltpRxStateMachine.HandleReceivedChars(data, size, errorMessage, sessionOriginatorEngineIdDecodedCallbackPtr);
    if (!success) {
        std::cerr << "error in LtpEngine::PacketIn: " << errorMessage << std::endl;
        m_ltpRxStateMachine.InitRx();
    }
    PacketInFullyProcessedCallback(success);
    return success;
}

bool LtpEngine::PacketIn(const std::vector<boost::asio::const_buffer> & constBufferVec) { //for testing
    for (std::size_t i = 0; i < constBufferVec.size(); ++i) {
        const boost::asio::const_buffer & cb = constBufferVec[i];
        if (!PacketIn((const uint8_t *) cb.data(), cb.size())) {
            return false;
        }
    }
    return true;
}

void LtpEngine::PacketIn_ThreadSafe(const uint8_t * data, const std::size_t size, Ltp::SessionOriginatorEngineIdDecodedCallback_t * sessionOriginatorEngineIdDecodedCallbackPtr) {
    boost::asio::post(m_ioServiceLtpEngine, boost::bind(&LtpEngine::PacketIn, this, data, size, sessionOriginatorEngineIdDecodedCallbackPtr));
}
void LtpEngine::PacketIn_ThreadSafe(const std::vector<boost::asio::const_buffer> & constBufferVec) { //for testing
    boost::asio::post(m_ioServiceLtpEngine, boost::bind(&LtpEngine::PacketIn, this, constBufferVec));
}

void LtpEngine::SignalReadyForSend_ThreadSafe() { //called by HandleUdpSend
    boost::asio::post(m_ioServiceLtpEngine, boost::bind(&LtpEngine::TrySendPacketIfAvailable, this));
}

void LtpEngine::TrySendPacketIfAvailable() {
    if (m_ioServiceLtpEngineThreadPtr) { //if not running inside a unit test
        //RATE STUFF (the TrySendPacketIfAvailable and OnTokenRefresh_TimerExpired run in the same thread)
        if (m_maxSendRateBitsPerSecOrZeroToDisable) { //if rate limiting enabled
            if (!m_tokenRateLimiter.CanTakeTokens()) { //no tokens available for next send, TrySendPacketIfAvailable() will be called at the next m_tokenRefreshTimer expiration
                TryRestartTokenRefreshTimer(); //make sure this is running so that tokens can be replenished
                return;
            }
        }
        std::vector<boost::asio::const_buffer> constBufferVec;
        boost::shared_ptr<std::vector<std::vector<uint8_t> > >  underlyingDataToDeleteOnSentCallback;
        uint64_t sessionOriginatorEngineId;
        if (NextPacketToSendRoundRobin(constBufferVec, underlyingDataToDeleteOnSentCallback, sessionOriginatorEngineId)) {
            if (m_maxSendRateBitsPerSecOrZeroToDisable) { //if rate limiting enabled
                std::size_t bytesToSend = 0;
                for (std::size_t i = 0; i < constBufferVec.size(); ++i) {
                    bytesToSend += constBufferVec[i].size();
                }
                m_tokenRateLimiter.TakeTokens(bytesToSend);
                TryRestartTokenRefreshTimer(); //tokens were taken, so make sure this is running so that tokens can be replenished
            }
            SendPacket(constBufferVec, underlyingDataToDeleteOnSentCallback, sessionOriginatorEngineId); //virtual call to child implementation
        }
    }
}

void LtpEngine::PacketInFullyProcessedCallback(bool success) {}

void LtpEngine::SendPacket(std::vector<boost::asio::const_buffer> & constBufferVec, boost::shared_ptr<std::vector<std::vector<uint8_t> > > & underlyingDataToDeleteOnSentCallback, const uint64_t sessionOriginatorEngineId) {}

bool LtpEngine::NextPacketToSendRoundRobin(std::vector<boost::asio::const_buffer> & constBufferVec, boost::shared_ptr<std::vector<std::vector<uint8_t> > > & underlyingDataToDeleteOnSentCallback, uint64_t & sessionOriginatorEngineId) {
    while (!m_listSendersNeedingDeleted.empty()) {
        std::map<uint64_t, std::unique_ptr<LtpSessionSender> >::iterator txSessionIt = m_mapSessionNumberToSessionSender.find(m_listSendersNeedingDeleted.front());
        if (txSessionIt != m_mapSessionNumberToSessionSender.end()) { //found
            if (txSessionIt->second->NextDataToSend(constBufferVec, underlyingDataToDeleteOnSentCallback)) { //if the session to be deleted still has data to send, send it before deletion
                sessionOriginatorEngineId = M_THIS_ENGINE_ID;
                return true;
            }
            else {
                m_numCheckpointTimerExpiredCallbacks += txSessionIt->second->m_numCheckpointTimerExpiredCallbacks;
                m_numDiscretionaryCheckpointsNotResent += txSessionIt->second->m_numDiscretionaryCheckpointsNotResent;
                if (txSessionIt == m_sendersIterator) {
                    m_sendersIterator = m_mapSessionNumberToSessionSender.erase(txSessionIt); //revalidate iterator
                }
                else {
                    m_mapSessionNumberToSessionSender.erase(txSessionIt);
                }
                ////std::cout << "deleted session sender " << m_listSendersNeedingDeleted.front() << std::endl;
            }
        }
        else {
            std::cout << "error in LtpEngine::NextPacketToSendRoundRobin: could not find session sender " << m_listSendersNeedingDeleted.front() << " to delete" << std::endl;
        }
        m_listSendersNeedingDeleted.pop_front();
    }

    while (!m_listReceiversNeedingDeleted.empty()) {
        std::map<Ltp::session_id_t, std::unique_ptr<LtpSessionReceiver> >::iterator rxSessionIt = m_mapSessionIdToSessionReceiver.find(m_listReceiversNeedingDeleted.front());
        if (rxSessionIt != m_mapSessionIdToSessionReceiver.end()) { //found rx Session
            if (rxSessionIt->second->NextDataToSend(constBufferVec, underlyingDataToDeleteOnSentCallback)) { //if the session to be deleted still has data to send, send it before deletion
                sessionOriginatorEngineId = rxSessionIt->first.sessionOriginatorEngineId;
                return true;
            }
            else {
                //erase session
                m_numReportSegmentTimerExpiredCallbacks += rxSessionIt->second->m_numReportSegmentTimerExpiredCallbacks;
                m_numReportSegmentsUnableToBeIssued += rxSessionIt->second->m_numReportSegmentsUnableToBeIssued;
                m_numReportSegmentsTooLargeAndNeedingSplit += rxSessionIt->second->m_numReportSegmentsTooLargeAndNeedingSplit;
                m_numReportSegmentsCreatedViaSplit += rxSessionIt->second->m_numReportSegmentsCreatedViaSplit;
                if (rxSessionIt == m_receiversIterator) {
                    m_receiversIterator = m_mapSessionIdToSessionReceiver.erase(rxSessionIt); //revalidate iterator
                }
                else {
                    m_mapSessionIdToSessionReceiver.erase(rxSessionIt);
                }
                ////std::cout << "deleted session receiver sessionNumber " << m_listReceiversNeedingDeleted.front().sessionNumber << std::endl;
            }
        }
        else {
            std::cout << "error in LtpEngine::NextPacketToSendRoundRobin: could not find session receiver " << m_listReceiversNeedingDeleted.front() << " to delete" << std::endl;
        }
        m_listReceiversNeedingDeleted.pop_front();
    }

    if (!m_listCancelSegmentTimerInfo.empty()) {
        //6.15.  Start Cancel Timer
        //This procedure is triggered by arrival of a link state cue indicating
        //the de - queuing(for transmission) of a Cx segment.
        //Response: the expected arrival time of the CAx segment that will be
        //produced on reception of this Cx segment is computed and a countdown
        //timer for this arrival time is started.
        cancel_segment_timer_info_t & info = m_listCancelSegmentTimerInfo.front();

        //send Cancel Segment
        underlyingDataToDeleteOnSentCallback = boost::make_shared<std::vector<std::vector<uint8_t> > >(1);
        Ltp::GenerateCancelSegmentLtpPacket((*underlyingDataToDeleteOnSentCallback)[0],
            info.sessionId, info.reasonCode, info.isFromSender, NULL, NULL);
        constBufferVec.resize(1);
        constBufferVec[0] = boost::asio::buffer((*underlyingDataToDeleteOnSentCallback)[0]);
        sessionOriginatorEngineId = info.sessionId.sessionOriginatorEngineId;

        const uint8_t * const infoPtr = (uint8_t*)&info;
        m_timeManagerOfCancelSegments.StartTimer(info.sessionId, std::vector<uint8_t>(infoPtr, infoPtr + sizeof(info)));
        //std::cout << "start cancel timer\n";
        m_listCancelSegmentTimerInfo.pop_front();
        return true;
    }

    if (!m_closedSessionDataToSend.empty()) { //includes report ack segments and cancel ack segments from closed sessions (which do not require timers)
        //highest priority
        underlyingDataToDeleteOnSentCallback = boost::make_shared<std::vector<std::vector<uint8_t> > >(1);
        (*underlyingDataToDeleteOnSentCallback)[0] = std::move(m_closedSessionDataToSend.front().second);
        sessionOriginatorEngineId = m_closedSessionDataToSend.front().first;
        m_closedSessionDataToSend.pop_front();
        constBufferVec.resize(1);
        constBufferVec[0] = boost::asio::buffer((*underlyingDataToDeleteOnSentCallback)[0]);
        return true;
    }

    bool foundDataToSend = false;
    for (unsigned int i = 0; i < 2; ++i) {
        while ((m_sendersIterator != m_mapSessionNumberToSessionSender.end()) && !foundDataToSend) {
            foundDataToSend = m_sendersIterator->second->NextDataToSend(constBufferVec, underlyingDataToDeleteOnSentCallback);
            sessionOriginatorEngineId = M_THIS_ENGINE_ID;
            ++m_sendersIterator;
        }
        while ((m_receiversIterator != m_mapSessionIdToSessionReceiver.end()) && !foundDataToSend) {
            foundDataToSend = m_receiversIterator->second->NextDataToSend(constBufferVec, underlyingDataToDeleteOnSentCallback);
            sessionOriginatorEngineId = m_receiversIterator->first.sessionOriginatorEngineId;
            ++m_receiversIterator;
        }
        if ((m_sendersIterator == m_mapSessionNumberToSessionSender.end()) && (m_receiversIterator == m_mapSessionIdToSessionReceiver.end())) {
            m_sendersIterator = m_mapSessionNumberToSessionSender.begin();
            m_receiversIterator = m_mapSessionIdToSessionReceiver.begin();
            if (!foundDataToSend) {
                continue; //continue once
            }
        }
    }
    return foundDataToSend;
}

/*
    Transmission Request

   In order to request transmission of a block of client service data,
   the client service MUST provide the following parameters to LTP:

      Destination client service ID.

      Destination LTP engine ID.

      Client service data to send, as an array of bytes.

      Length of the data to be sent.

      Length of the red-part of the data.  This value MUST be in the
      range from zero to the total length of data to be sent.

   On reception of a valid transmission request from a client service,
   LTP proceeds as follows.
   */
void LtpEngine::TransmissionRequest(uint64_t destinationClientServiceId, uint64_t destinationLtpEngineId,
    LtpClientServiceDataToSend && clientServiceDataToSend, std::shared_ptr<LtpTransmissionRequestUserData> && userDataPtrToTake, uint64_t lengthOfRedPart)
{ //only called directly by unit test (not thread safe)

    uint64_t randomSessionNumberGeneratedBySender;
    uint64_t randomInitialSenderCheckpointSerialNumber; //incremented by 1 for new
    if(M_FORCE_32_BIT_RANDOM_NUMBERS) {
        randomSessionNumberGeneratedBySender = m_rng.GetRandomSession32(m_randomDevice);
        randomInitialSenderCheckpointSerialNumber = m_rng.GetRandomSerialNumber32(m_randomDevice);
    }
    else {
        randomSessionNumberGeneratedBySender = m_rng.GetRandomSession64(m_randomDevice);
        randomInitialSenderCheckpointSerialNumber = m_rng.GetRandomSerialNumber64(m_randomDevice);
    }
    Ltp::session_id_t senderSessionId(M_THIS_ENGINE_ID, randomSessionNumberGeneratedBySender);
    m_mapSessionNumberToSessionSender[randomSessionNumberGeneratedBySender] = boost::make_unique<LtpSessionSender>(
        randomInitialSenderCheckpointSerialNumber, std::move(clientServiceDataToSend), std::move(userDataPtrToTake),
        lengthOfRedPart, M_MTU_CLIENT_SERVICE_DATA, senderSessionId, destinationClientServiceId,
        M_ONE_WAY_LIGHT_TIME, M_ONE_WAY_MARGIN_TIME, m_ioServiceLtpEngine,
        boost::bind(&LtpEngine::NotifyEngineThatThisSenderNeedsDeletedCallback, this, boost::placeholders::_1, boost::placeholders::_2, boost::placeholders::_3, boost::placeholders::_4),
        boost::bind(&LtpEngine::TrySendPacketIfAvailable, this),
        boost::bind(&LtpEngine::InitialTransmissionCompletedCallback, this, boost::placeholders::_1, boost::placeholders::_2), m_checkpointEveryNthDataPacketSender, m_maxRetriesPerSerialNumber);

    if (m_sessionStartCallback) {
        //At the sender, the session start notice informs the client service of the initiation of the transmission session.
        m_sessionStartCallback(senderSessionId);
    }
}
void LtpEngine::TransmissionRequest(uint64_t destinationClientServiceId, uint64_t destinationLtpEngineId,
    const uint8_t * clientServiceDataToCopyAndSend, uint64_t length, std::shared_ptr<LtpTransmissionRequestUserData> && userDataPtrToTake, uint64_t lengthOfRedPart)
{  //only called directly by unit test (not thread safe)
    TransmissionRequest(destinationClientServiceId, destinationLtpEngineId,
        std::move(std::vector<uint8_t>(clientServiceDataToCopyAndSend, clientServiceDataToCopyAndSend + length)),
        std::move(userDataPtrToTake), lengthOfRedPart
    );
}
void LtpEngine::TransmissionRequest(uint64_t destinationClientServiceId, uint64_t destinationLtpEngineId,
    const uint8_t * clientServiceDataToCopyAndSend, uint64_t length, uint64_t lengthOfRedPart)
{  //only called directly by unit test (not thread safe)
    TransmissionRequest(destinationClientServiceId, destinationLtpEngineId,
        std::move(std::vector<uint8_t>(clientServiceDataToCopyAndSend, clientServiceDataToCopyAndSend + length)),
        std::move(std::shared_ptr<LtpTransmissionRequestUserData>()), lengthOfRedPart
    );
}
void LtpEngine::TransmissionRequest(boost::shared_ptr<transmission_request_t> & transmissionRequest) {
    TransmissionRequest(transmissionRequest->destinationClientServiceId, transmissionRequest->destinationLtpEngineId,
        std::move(transmissionRequest->clientServiceDataToSend),
        std::move(transmissionRequest->userDataPtr),
        transmissionRequest->lengthOfRedPart
    );
    TrySendPacketIfAvailable();
}
void LtpEngine::TransmissionRequest_ThreadSafe(boost::shared_ptr<transmission_request_t> && transmissionRequest) {
    //The arguments to bind are copied or moved, and are never passed by reference unless wrapped in std::ref or std::cref.
    boost::asio::post(m_ioServiceLtpEngine, boost::bind(&LtpEngine::TransmissionRequest, this, std::move(transmissionRequest)));
}

void LtpEngine::CancellationRequest_ThreadSafe(const Ltp::session_id_t & sessionId) {
    //The arguments to bind are copied or moved, and are never passed by reference unless wrapped in std::ref or std::cref.
    boost::asio::post(m_ioServiceLtpEngine, boost::bind(&LtpEngine::CancellationRequest, this, sessionId));
}

//4.2.  Cancellation Request
//In order to request cancellation of a session, either as the sender
//or as the receiver of the associated data block, the client service
//must provide the session ID identifying the session to be canceled.
bool LtpEngine::CancellationRequest(const Ltp::session_id_t & sessionId) { //only called directly by unit test (not thread safe)

    //On reception of a valid cancellation request from a client service,
    //LTP proceeds as follows.

    //First, the internal "Cancel Session" procedure(Section 6.19) is
    //invoked.
    //6.19.  Cancel Session
    //Response: all segments of the affected session that are currently
    //queued for transmission can be deleted from the outbound traffic
    //queues.All countdown timers currently associated with the session
    //are deleted.Note : If the local LTP engine is the sender, then all
    //remaining data retransmission buffer space allocated to the session
    //can be released.

    if (M_THIS_ENGINE_ID == sessionId.sessionOriginatorEngineId) {
        //if the session is being canceled by the sender (i.e., the
        //session originator part of the session ID supplied in the
        //cancellation request is the local LTP engine ID):

        std::map<uint64_t, std::unique_ptr<LtpSessionSender> >::iterator txSessionIt = m_mapSessionNumberToSessionSender.find(sessionId.sessionNumber);
        if (txSessionIt != m_mapSessionNumberToSessionSender.end()) { //found

            //-If none of the data segments previously queued for transmission
            //as part of this session have yet been de - queued and transmitted
            //-- i.e., if the destination engine cannot possibly be aware of
            //this session -- then the session is simply closed; the "Close
            //Session" procedure (Section 6.20) is invoked.

            //- Otherwise, a CS(cancel by block sender) segment with the
            //reason - code USR_CNCLD MUST be queued for transmission to the
            //destination LTP engine specified in the transmission request
            //that started this session.
            //erase session
            m_numCheckpointTimerExpiredCallbacks += txSessionIt->second->m_numCheckpointTimerExpiredCallbacks;
            m_numDiscretionaryCheckpointsNotResent += txSessionIt->second->m_numDiscretionaryCheckpointsNotResent;
            if (txSessionIt == m_sendersIterator) {
                m_sendersIterator = m_mapSessionNumberToSessionSender.erase(txSessionIt); //revalidate iterator
            }
            else {
                m_mapSessionNumberToSessionSender.erase(txSessionIt);
            }
            std::cout << "LtpEngine::CancellationRequest deleted session sender session number " << sessionId.sessionNumber << std::endl;

            //send Cancel Segment to receiver (NextPacketToSendRoundRobin() will create the packet and start the timer)
            m_listCancelSegmentTimerInfo.emplace_back();
            cancel_segment_timer_info_t & info = m_listCancelSegmentTimerInfo.back();
            info.sessionId = sessionId;
            info.retryCount = 1;
            info.isFromSender = true;
            info.reasonCode = CANCEL_SEGMENT_REASON_CODES::USER_CANCELLED;

            TrySendPacketIfAvailable();
            return true;
        }
        return false;
    }
    else { //not sender, try receiver
        std::map<Ltp::session_id_t, std::unique_ptr<LtpSessionReceiver> >::iterator rxSessionIt = m_mapSessionIdToSessionReceiver.find(sessionId);
        if (rxSessionIt != m_mapSessionIdToSessionReceiver.end()) { //found rx Session

            //if the session is being canceled by the receiver:
            //-If there is no transmission queue - set bound for the sender
            //(possibly because the local LTP engine is running on a receive -
            //only device), then the session is simply closed; the "Close
            //Session" procedure (Section 6.20) is invoked.
            //- Otherwise, a CR(cancel by block receiver) segment with reason -
            //code USR_CNCLD MUST be queued for transmission to the block
            //sender.

            //erase session
            m_numReportSegmentTimerExpiredCallbacks += rxSessionIt->second->m_numReportSegmentTimerExpiredCallbacks;
            m_numReportSegmentsUnableToBeIssued += rxSessionIt->second->m_numReportSegmentsUnableToBeIssued;
            m_numReportSegmentsTooLargeAndNeedingSplit += rxSessionIt->second->m_numReportSegmentsTooLargeAndNeedingSplit;
            m_numReportSegmentsCreatedViaSplit += rxSessionIt->second->m_numReportSegmentsCreatedViaSplit;
            if (rxSessionIt == m_receiversIterator) {
                m_receiversIterator = m_mapSessionIdToSessionReceiver.erase(rxSessionIt); //revalidate iterator
            }
            else {
                m_mapSessionIdToSessionReceiver.erase(rxSessionIt);
            }
            std::cout << "LtpEngine::CancellationRequest deleted session receiver session number " << sessionId.sessionNumber << std::endl;

            //send Cancel Segment to sender (NextPacketToSendRoundRobin() will create the packet and start the timer)
            m_listCancelSegmentTimerInfo.emplace_back();
            cancel_segment_timer_info_t & info = m_listCancelSegmentTimerInfo.back();
            info.sessionId = sessionId;
            info.retryCount = 1;
            info.isFromSender = false;
            info.reasonCode = CANCEL_SEGMENT_REASON_CODES::USER_CANCELLED;

            TrySendPacketIfAvailable();
            return true;
        }
        return false;
    }

}

void LtpEngine::CancelSegmentReceivedCallback(const Ltp::session_id_t & sessionId, CANCEL_SEGMENT_REASON_CODES reasonCode, bool isFromSender,
    Ltp::ltp_extensions_t & headerExtensions, Ltp::ltp_extensions_t & trailerExtensions)
{
    if (isFromSender) { //to receiver
        std::map<Ltp::session_id_t, std::unique_ptr<LtpSessionReceiver> >::iterator rxSessionIt = m_mapSessionIdToSessionReceiver.find(sessionId);
        if (rxSessionIt != m_mapSessionIdToSessionReceiver.end()) { //found
            if (m_receptionSessionCancelledCallback) {
                m_receptionSessionCancelledCallback(sessionId, reasonCode); //No subsequent delivery notices will be issued for this session.
            }
            //erase session
            m_numReportSegmentTimerExpiredCallbacks += rxSessionIt->second->m_numReportSegmentTimerExpiredCallbacks;
            m_numReportSegmentsUnableToBeIssued += rxSessionIt->second->m_numReportSegmentsUnableToBeIssued;
            m_numReportSegmentsTooLargeAndNeedingSplit += rxSessionIt->second->m_numReportSegmentsTooLargeAndNeedingSplit;
            m_numReportSegmentsCreatedViaSplit += rxSessionIt->second->m_numReportSegmentsCreatedViaSplit;
            if (rxSessionIt == m_receiversIterator) {
                m_receiversIterator = m_mapSessionIdToSessionReceiver.erase(rxSessionIt); //revalidate iterator
            }
            else {
                m_mapSessionIdToSessionReceiver.erase(rxSessionIt);
            }
            std::cout << "LtpEngine::CancelSegmentReceivedCallback deleted session receiver session number " << sessionId.sessionNumber << std::endl;
            //Send CAx after outer if-else statement

        }
        else { //not found
            //The LTP receiver might also receive a retransmitted CS segment at the
            //CLOSED state(either if the CAS segment previously transmitted was
            //lost or if the CS timer expired prematurely at the LTP sender).In
            //such a case, the CAS is scheduled for retransmission.

            //Send CAx after outer if-else statement
        }
    }
    else { //to sender
        std::map<uint64_t, std::unique_ptr<LtpSessionSender> >::iterator txSessionIt = m_mapSessionNumberToSessionSender.find(sessionId.sessionNumber);
        if (txSessionIt != m_mapSessionNumberToSessionSender.end()) { //found
            if (m_transmissionSessionCancelledCallback) {
                m_transmissionSessionCancelledCallback(sessionId, reasonCode, txSessionIt->second->m_userDataPtr);
            }
            //erase session
            m_numCheckpointTimerExpiredCallbacks += txSessionIt->second->m_numCheckpointTimerExpiredCallbacks;
            m_numDiscretionaryCheckpointsNotResent += txSessionIt->second->m_numDiscretionaryCheckpointsNotResent;
            if (txSessionIt == m_sendersIterator) {
                m_sendersIterator = m_mapSessionNumberToSessionSender.erase(txSessionIt); //revalidate iterator
            }
            else {
                m_mapSessionNumberToSessionSender.erase(txSessionIt);
            }
            std::cout << "LtpEngine::CancelSegmentReceivedCallback deleted session sender session number " << sessionId.sessionNumber << std::endl;
            //Send CAx after outer if-else statement
        }
        else { //not found
            //Note that while at the CLOSED state:
            //If the session was canceled
            //by the receiver by issuing a CR segment, the receiver may retransmit
            //the CR segment(either prematurely or because the acknowledging CAR
            //segment got lost).In this case, the LTP sender retransmits the
            //acknowledging CAR segment and stays in the CLOSED state.
            //send CAx to receiver

            //Send CAx after outer if-else statement
        }
    }
    //send CAx to receiver or sender (whether or not session exists) (here because all data in session is erased)
    m_closedSessionDataToSend.emplace_back();
    Ltp::GenerateCancelAcknowledgementSegmentLtpPacket(m_closedSessionDataToSend.back().second,
        sessionId, isFromSender, NULL, NULL);
    m_closedSessionDataToSend.back().first = sessionId.sessionOriginatorEngineId;
    TrySendPacketIfAvailable();
}

void LtpEngine::CancelAcknowledgementSegmentReceivedCallback(const Ltp::session_id_t & sessionId, bool isToSender,
    Ltp::ltp_extensions_t & headerExtensions, Ltp::ltp_extensions_t & trailerExtensions)
{
    if (isToSender) {
        if (sessionId.sessionOriginatorEngineId != M_THIS_ENGINE_ID) {
            std::cerr << "error in CA received to sender: sessionId.sessionOriginatorEngineId (" << sessionId.sessionOriginatorEngineId << ") != M_THIS_ENGINE_ID (" << M_THIS_ENGINE_ID << ")\n";
            return;
        }
    }
    else {//to receiver
        if (sessionId.sessionOriginatorEngineId == M_THIS_ENGINE_ID) {
            std::cerr << "error in CA received to receiver: sessionId.sessionOriginatorEngineId (" << sessionId.sessionOriginatorEngineId << ") == M_THIS_ENGINE_ID (" << M_THIS_ENGINE_ID << ")\n";
            return;
        }
    }

    //6.18.  Stop Cancel Timer
    //This procedure is triggered by the reception of a CAx segment.
    //
    //Response: the timer associated with the Cx segment is deleted, and
    //the session of which the segment is one token is closed, i.e., the
    //"Close Session" procedure(Section 6.20) is invoked.
    //std::cout << "CancelAcknowledgementSegmentReceivedCallback1\n";
    if (!m_timeManagerOfCancelSegments.DeleteTimer(sessionId)) {
        std::cout << "LtpEngine::CancelAcknowledgementSegmentReceivedCallback didn't delete timer\n";
    }
    //std::cout << "CancelAcknowledgementSegmentReceivedCallback2\n";
}

void LtpEngine::CancelSegmentTimerExpiredCallback(Ltp::session_id_t cancelSegmentTimerSerialNumber, std::vector<uint8_t> & userData) {
    std::cout << "CancelSegmentTimerExpiredCallback!\n";
    cancel_segment_timer_info_t info;
    if (userData.size() != sizeof(info)) {
        std::cerr << "error in LtpEngine::CancelSegmentTimerExpiredCallback: userData.size() != sizeof(info)\n";
        return;
    }
    memcpy(&info, userData.data(), sizeof(info));

    if (info.retryCount <= m_maxRetriesPerSerialNumber) {
        //resend Cancel Segment to receiver (NextPacketToSendRoundRobin() will create the packet and start the timer)
        ++info.retryCount;
        m_listCancelSegmentTimerInfo.push_back(info);

        TrySendPacketIfAvailable();
    }

}

void LtpEngine::NotifyEngineThatThisSenderNeedsDeletedCallback(const Ltp::session_id_t & sessionId, bool wasCancelled, CANCEL_SEGMENT_REASON_CODES reasonCode, std::shared_ptr<LtpTransmissionRequestUserData> & userDataPtr) {
    if (wasCancelled) {
        //send Cancel Segment to receiver (NextPacketToSendRoundRobin() will create the packet and start the timer)
        m_listCancelSegmentTimerInfo.emplace_back();
        cancel_segment_timer_info_t & info = m_listCancelSegmentTimerInfo.back();
        info.sessionId = sessionId;
        info.retryCount = 1;
        info.isFromSender = true;
        info.reasonCode = reasonCode;

        if (m_transmissionSessionCancelledCallback) {
            m_transmissionSessionCancelledCallback(sessionId, reasonCode, userDataPtr);
        }
    }
    else {
        if (m_transmissionSessionCompletedCallback) {
            m_transmissionSessionCompletedCallback(sessionId, userDataPtr);
        }
    }

    m_listSendersNeedingDeleted.push_back(sessionId.sessionNumber);
    SignalReadyForSend_ThreadSafe(); //posts the TrySendPacketIfAvailable(); so this won't be deleteted during execution
}

void LtpEngine::NotifyEngineThatThisReceiverNeedsDeletedCallback(const Ltp::session_id_t & sessionId, bool wasCancelled, CANCEL_SEGMENT_REASON_CODES reasonCode) {
    if (wasCancelled) {
        //send Cancel Segment to sender (NextPacketToSendRoundRobin() will create the packet and start the timer)
        m_listCancelSegmentTimerInfo.emplace_back();
        cancel_segment_timer_info_t & info = m_listCancelSegmentTimerInfo.back();
        info.sessionId = sessionId;
        info.retryCount = 1;
        info.isFromSender = false;
        info.reasonCode = reasonCode;

        if (m_receptionSessionCancelledCallback) {
            m_receptionSessionCancelledCallback(sessionId, reasonCode);
        }
    }
    else { //not cancelled

    }

    m_listReceiversNeedingDeleted.push_back(sessionId);
    SignalReadyForSend_ThreadSafe(); //posts the TrySendPacketIfAvailable(); so this won't be deleteted during execution
}

void LtpEngine::InitialTransmissionCompletedCallback(const Ltp::session_id_t & sessionId, std::shared_ptr<LtpTransmissionRequestUserData> & userDataPtr) {
    if (m_initialTransmissionCompletedCallback) {
        m_initialTransmissionCompletedCallback(sessionId, userDataPtr);
    }
}

void LtpEngine::ReportAcknowledgementSegmentReceivedCallback(const Ltp::session_id_t & sessionId, uint64_t reportSerialNumberBeingAcknowledged,
    Ltp::ltp_extensions_t & headerExtensions, Ltp::ltp_extensions_t & trailerExtensions)
{
    if (sessionId.sessionOriginatorEngineId == M_THIS_ENGINE_ID) {
        std::cerr << "error in RA received: sessionId.sessionOriginatorEngineId == M_THIS_ENGINE_ID\n";
        return;
    }
    std::map<Ltp::session_id_t, std::unique_ptr<LtpSessionReceiver> >::iterator rxSessionIt = m_mapSessionIdToSessionReceiver.find(sessionId);
    if (rxSessionIt != m_mapSessionIdToSessionReceiver.end()) { //found
        rxSessionIt->second->ReportAcknowledgementSegmentReceivedCallback(reportSerialNumberBeingAcknowledged, headerExtensions, trailerExtensions);
    }
    TrySendPacketIfAvailable();
}

void LtpEngine::ReportSegmentReceivedCallback(const Ltp::session_id_t & sessionId, const Ltp::report_segment_t & reportSegment,
    Ltp::ltp_extensions_t & headerExtensions, Ltp::ltp_extensions_t & trailerExtensions)
{
    if (sessionId.sessionOriginatorEngineId != M_THIS_ENGINE_ID) {
        std::cerr << "error in RS received: sessionId.sessionOriginatorEngineId(" << sessionId.sessionOriginatorEngineId << ")  != M_THIS_ENGINE_ID(" << M_THIS_ENGINE_ID << ")" << std::endl;
        return;
    }
    std::map<uint64_t, std::unique_ptr<LtpSessionSender> >::iterator txSessionIt = m_mapSessionNumberToSessionSender.find(sessionId.sessionNumber);
    if (txSessionIt != m_mapSessionNumberToSessionSender.end()) { //found
        txSessionIt->second->ReportSegmentReceivedCallback(reportSegment, headerExtensions, trailerExtensions);
    }
    else { //not found
        //Note that while at the CLOSED state, the LTP sender might receive an
        //RS segment(if the last transmitted RA segment before session close
        //got lost or if the LTP receiver retransmitted the RS segment
        //prematurely), in which case it retransmits an acknowledging RA
        //segment and stays in the CLOSED state.
        m_closedSessionDataToSend.emplace_back();
        Ltp::GenerateReportAcknowledgementSegmentLtpPacket(m_closedSessionDataToSend.back().second,
            sessionId, reportSegment.reportSerialNumber, NULL, NULL);
        m_closedSessionDataToSend.back().first = sessionId.sessionOriginatorEngineId;
    }
    TrySendPacketIfAvailable();
}

//SESSION START:
//At the receiver, this notice indicates the beginning of a
//new reception session, and is delivered upon arrival of the first
//data segment carrying a new session ID.

void LtpEngine::DataSegmentReceivedCallback(uint8_t segmentTypeFlags, const Ltp::session_id_t & sessionId,
    std::vector<uint8_t> & clientServiceDataVec, const Ltp::data_segment_metadata_t & dataSegmentMetadata,
    Ltp::ltp_extensions_t & headerExtensions, Ltp::ltp_extensions_t & trailerExtensions)
{
    if (sessionId.sessionOriginatorEngineId == M_THIS_ENGINE_ID) {
        std::cerr << "error in DS received: sessionId.sessionOriginatorEngineId(" << sessionId.sessionOriginatorEngineId << ") == M_THIS_ENGINE_ID(" << M_THIS_ENGINE_ID << ")\n";
        return;
    }

    std::map<Ltp::session_id_t, std::unique_ptr<LtpSessionReceiver> >::iterator rxSessionIt = m_mapSessionIdToSessionReceiver.find(sessionId);
    if (rxSessionIt == m_mapSessionIdToSessionReceiver.end()) { //not found.. new session started
        //first check if the session has been closed prevously before recreating
        std::map<uint64_t, std::unique_ptr<LtpSessionRecreationPreventer> >::iterator it = m_mapSessionOriginatorEngineIdToLtpSessionRecreationPreventer.find(sessionId.sessionOriginatorEngineId);
        if (it == m_mapSessionOriginatorEngineIdToLtpSessionRecreationPreventer.end()) {
            std::cout << "create new LtpSessionRecreationPreventer for sessionOriginatorEngineId " << sessionId.sessionOriginatorEngineId << std::endl;
            it = m_mapSessionOriginatorEngineIdToLtpSessionRecreationPreventer.insert(
                std::pair<uint64_t, std::unique_ptr<LtpSessionRecreationPreventer> >(sessionId.sessionOriginatorEngineId, boost::make_unique<LtpSessionRecreationPreventer>(1000))).first;
        }
        if (!it->second->AddSession(sessionId.sessionNumber)) {
            std::cout << "preventing old session from being recreated for " << sessionId << std::endl;
            return;
        }
        //if(m_ltpSessionRecreationPreventer.AddSession(sessionId.sessionNumber))
        const uint64_t randomNextReportSegmentReportSerialNumber = (M_FORCE_32_BIT_RANDOM_NUMBERS) ? m_rng.GetRandomSerialNumber32(m_randomDevice) : m_rng.GetRandomSerialNumber64(m_randomDevice); //incremented by 1 for new
        std::unique_ptr<LtpSessionReceiver> session = boost::make_unique<LtpSessionReceiver>(randomNextReportSegmentReportSerialNumber, m_maxReceptionClaims,
            M_ESTIMATED_BYTES_TO_RECEIVE_PER_SESSION, M_MAX_RED_RX_BYTES_PER_SESSION,
            sessionId, dataSegmentMetadata.clientServiceId, M_ONE_WAY_LIGHT_TIME, M_ONE_WAY_MARGIN_TIME, m_ioServiceLtpEngine,
            boost::bind(&LtpEngine::NotifyEngineThatThisReceiverNeedsDeletedCallback, this, boost::placeholders::_1, boost::placeholders::_2, boost::placeholders::_3),
            boost::bind(&LtpEngine::TrySendPacketIfAvailable, this), m_maxRetriesPerSerialNumber);

        std::pair<std::map<Ltp::session_id_t, std::unique_ptr<LtpSessionReceiver> >::iterator, bool> res =
            m_mapSessionIdToSessionReceiver.insert(std::pair< Ltp::session_id_t, std::unique_ptr<LtpSessionReceiver> >(sessionId, std::move(session)));
        if (res.second == false) { //fragment key was not inserted
            std::cerr << "error new rx session cannot be inserted??\n";
            return;
        }
        rxSessionIt = res.first;

        if (m_sessionStartCallback) {
            //At the receiver, this notice indicates the beginning of a new reception session, and is delivered upon arrival of the first data segment carrying a new session ID.
            m_sessionStartCallback(sessionId);
        }
    }
    rxSessionIt->second->DataSegmentReceivedCallback(segmentTypeFlags, clientServiceDataVec, dataSegmentMetadata, headerExtensions, trailerExtensions, m_redPartReceptionCallback, m_greenPartSegmentArrivalCallback);
    TrySendPacketIfAvailable();
}

void LtpEngine::SetSessionStartCallback(const SessionStartCallback_t & callback) {
    m_sessionStartCallback = callback;
}
void LtpEngine::SetRedPartReceptionCallback(const RedPartReceptionCallback_t & callback) {
    m_redPartReceptionCallback = callback;
}
void LtpEngine::SetGreenPartSegmentArrivalCallback(const GreenPartSegmentArrivalCallback_t & callback) {
    m_greenPartSegmentArrivalCallback = callback;
}
void LtpEngine::SetReceptionSessionCancelledCallback(const ReceptionSessionCancelledCallback_t & callback) {
    m_receptionSessionCancelledCallback = callback;
}
void LtpEngine::SetTransmissionSessionCompletedCallback(const TransmissionSessionCompletedCallback_t & callback) {
    m_transmissionSessionCompletedCallback = callback;
}
void LtpEngine::SetInitialTransmissionCompletedCallback(const InitialTransmissionCompletedCallback_t & callback) {
    m_initialTransmissionCompletedCallback = callback;
}
void LtpEngine::SetTransmissionSessionCancelledCallback(const TransmissionSessionCancelledCallback_t & callback) {
    m_transmissionSessionCancelledCallback = callback;
}

std::size_t LtpEngine::NumActiveReceivers() const {
    return m_mapSessionIdToSessionReceiver.size();
}
std::size_t LtpEngine::NumActiveSenders() const {
    return m_mapSessionNumberToSessionSender.size();
}

void LtpEngine::UpdateRate(const uint64_t maxSendRateBitsPerSecOrZeroToDisable) {
    m_maxSendRateBitsPerSecOrZeroToDisable = maxSendRateBitsPerSecOrZeroToDisable;
    if (maxSendRateBitsPerSecOrZeroToDisable) {
        const uint64_t rateBytesPerSecond = m_maxSendRateBitsPerSecOrZeroToDisable >> 3;
        m_tokenRateLimiter.SetRate(
            rateBytesPerSecond,
            boost::posix_time::seconds(1),
            static_tokenMaxLimitDurationWindow //token limit of rateBytesPerSecond / (1000ms/100ms) = rateBytesPerSecond / 10
        );
    }
}

void LtpEngine::UpdateRate_ThreadSafe(const uint64_t maxSendRateBitsPerSecOrZeroToDisable) {
    boost::asio::post(m_ioServiceLtpEngine, boost::bind(&LtpEngine::UpdateRate, this, maxSendRateBitsPerSecOrZeroToDisable));
}

//restarts the token refresh timer if it is not running from now
void LtpEngine::TryRestartTokenRefreshTimer() {
    if (!m_tokenRefreshTimerIsRunning) {
        const boost::posix_time::ptime nowPtime = boost::posix_time::microsec_clock::universal_time();
        if (m_lastTimeTokensWereRefreshed.is_neg_infinity()) {
            m_lastTimeTokensWereRefreshed = nowPtime;
        }
        m_tokenRefreshTimer.expires_at(nowPtime + static_tokenRefreshTimeDurationWindow);
        m_tokenRefreshTimer.async_wait(boost::bind(&LtpEngine::OnTokenRefresh_TimerExpired, this, boost::asio::placeholders::error));
        m_tokenRefreshTimerIsRunning = true;
    }
}
//restarts the token refresh timer if it is not running from the given ptime
void LtpEngine::TryRestartTokenRefreshTimer(const boost::posix_time::ptime & nowPtime) {
    if (!m_tokenRefreshTimerIsRunning) {
        if (m_lastTimeTokensWereRefreshed.is_neg_infinity()) {
            m_lastTimeTokensWereRefreshed = nowPtime;
        }
        m_tokenRefreshTimer.expires_at(nowPtime + static_tokenRefreshTimeDurationWindow);
        m_tokenRefreshTimer.async_wait(boost::bind(&LtpEngine::OnTokenRefresh_TimerExpired, this, boost::asio::placeholders::error));
        m_tokenRefreshTimerIsRunning = true;
    }
}

void LtpEngine::OnTokenRefresh_TimerExpired(const boost::system::error_code& e) {
    const boost::posix_time::ptime nowPtime = boost::posix_time::microsec_clock::universal_time();
    const boost::posix_time::time_duration diff = nowPtime - m_lastTimeTokensWereRefreshed;
    m_tokenRateLimiter.AddTime(diff);
    m_lastTimeTokensWereRefreshed = nowPtime;
    m_tokenRefreshTimerIsRunning = false;
    if (e != boost::asio::error::operation_aborted) {
        // Timer was not cancelled, take necessary action.

        //If more tokens can be added, restart the timer so more tokens will be added at the next timer expiration.
        //Otherwise, if full, don't restart the timer and the next send packet operation will start it.
        if (!m_tokenRateLimiter.HasFullBucketOfTokens()) {
            TryRestartTokenRefreshTimer(nowPtime); //do this first before TrySendPacketIfAvailable so nowPtime can be used
        }

        TrySendPacketIfAvailable();
    }
    else {
        //std::cout << "timer cancelled\n";
    }
}

ext: cpp


text:
/*
 * Copyright (C) 2021 LEIDOS.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
#include <lanelet2_core/primitives/RegulatoryElement.h>
#include <boost/algorithm/string.hpp>
#include <lanelet2_extension/regulatory_elements/SignalizedIntersection.h>
#include "RegulatoryHelpers.h"
#include <lanelet2_extension/logging/logger.h>

namespace lanelet
{
    // C++ 14 vs 17 constant defintion
    #if __cplusplus < 201703L
        // Forward declare static constexpr
        constexpr char SignalizedIntersection::RuleName[];  // instantiate string in cpp file
        constexpr const char CarmaRoleNameString::IntersectionEntry[];
        constexpr const char CarmaRoleNameString::IntersectionExit[];
        constexpr const char CarmaRoleNameString::IntersectionInterior[];
    #endif

    ConstLanelets SignalizedIntersection::getEntryLanelets() const {return getParameters<ConstLanelet>(CarmaRoleNameString::IntersectionEntry);}

    ConstLanelets SignalizedIntersection::getExitLanelets() const {return getParameters<ConstLanelet>(CarmaRoleNameString::IntersectionExit);}

    ConstLanelets SignalizedIntersection::getInteriorLanelets() const {return getParameters<ConstLanelet>(CarmaRoleNameString::IntersectionInterior);}

    std::unique_ptr<lanelet::RegulatoryElementData> SignalizedIntersection::buildData(Id id, const Lanelets& entry, const Lanelets& exit,
                                                                const Lanelets& interior)
    {
        // Add parameters
        RuleParameterMap rules;
        rules[lanelet::CarmaRoleNameString::IntersectionEntry].insert(rules[lanelet::CarmaRoleNameString::IntersectionEntry].end(), entry.begin(),
                                                        entry.end());
        rules[lanelet::CarmaRoleNameString::IntersectionExit].insert(rules[lanelet::CarmaRoleNameString::IntersectionExit].end(), exit.begin(),
                                                        exit.end());
        rules[lanelet::CarmaRoleNameString::IntersectionInterior].insert(rules[lanelet::CarmaRoleNameString::IntersectionInterior].end(), interior.begin(),
                                                        interior.end());

        // Add attributes
        AttributeMap attribute_map({ { AttributeNamesString::Type, AttributeValueString::RegulatoryElement },
                                    { AttributeNamesString::Subtype, RuleName }});

        return std::make_unique<RegulatoryElementData>(id, rules, attribute_map);
    }

    std::vector<CarmaTrafficSignalConstPtr> SignalizedIntersection::getTrafficSignals(const ConstLanelet& llt) const
    {
        return llt.regulatoryElementsAs<CarmaTrafficSignal>();
    }

    void SignalizedIntersection::addLanelet(Lanelet lanelet, IntersectionSection section)
    {
        switch (section)
        {
            case IntersectionSection::ENTRY:
                parameters()[CarmaRoleNameString::IntersectionEntry].emplace_back(RuleParameter(lanelet));
                break;
            case IntersectionSection::EXIT:
                parameters()[CarmaRoleNameString::IntersectionExit].emplace_back(RuleParameter(lanelet));
                break;
            case IntersectionSection::INTERIOR:
                parameters()[CarmaRoleNameString::IntersectionInterior].emplace_back(RuleParameter(lanelet));
                break;
            default:
                throw std::invalid_argument("Invalid section is passed as IntersectionSection");
        }

        section_lookup[lanelet.id()] = section;
    }

    namespace {

        template <typename T>
        bool findAndErase(const T& primitive, RuleParameters* member)
        {
            if (member == nullptr)
            {
                LOG_ERROR_STREAM("member is null pointer!");
                return false;
            }

            auto it = std::find(member->begin(), member->end(), RuleParameter(primitive));

            if (it == member->end())
            {
                return false;
            }

            member->erase(it);

            return true;
        }

        class GetIdVisitor : public RuleParameterVisitor {
            public:
                static Id id(const ConstRuleParameter& param) {
                    GetIdVisitor visitor;
                    boost::apply_visitor(visitor, param);
                    return visitor.id_;
                }
                template <typename PrimT>
                void appendID(const PrimT& p) {
                    id_ = p.id();
                }

                void operator()(const ConstPoint3d& p) override { appendID(p); }
                void operator()(const ConstLineString3d& l) override { appendID(l); }
                void operator()(const ConstPolygon3d& p) override { appendID(p); }
                void operator()(const ConstWeakLanelet& ll) override {
                    if (!ll.expired()) {
                    appendID(ll.lock());
                    }
                }
                void operator()(const ConstWeakArea& ar) override {
                    if (!ar.expired()) {
                    appendID(ar.lock());
                    }
                }

            private:
                Id id_{};
        };

    } //namespace

    bool SignalizedIntersection::removeLanelet(const Lanelet& llt)
    {
        // if successfully found and erased, then return true, else false
        auto iter = section_lookup.find(llt.id());

        if (iter == section_lookup.end())
            return false; // regem is not here
        else if ( iter->second == IntersectionSection::ENTRY)
            findAndErase(RuleParameter(llt), &parameters().find(CarmaRoleNameString::IntersectionEntry)->second);
        else if ( iter->second == IntersectionSection::EXIT)
            findAndErase(RuleParameter(llt), &parameters().find(CarmaRoleNameString::IntersectionExit)->second);
        else if ( iter->second == IntersectionSection::INTERIOR)
            findAndErase(RuleParameter(llt), &parameters().find(CarmaRoleNameString::IntersectionInterior)->second);

        section_lookup.erase(llt.id());

        return true;
    }

    Optional<lanelet::ConstLineString3d> SignalizedIntersection::getStopLine(const ConstLanelet& llt) const
    {
        Optional<lanelet::ConstLineString3d> stop_line = boost::none;
        // stop line geometry should be same for any traffic signals
        auto traffic_signals = llt.regulatoryElementsAs<CarmaTrafficSignal>();
        if (!traffic_signals.empty())
            stop_line = traffic_signals.front()->stopLine().front();

        return stop_line;
    }

    SignalizedIntersection::SignalizedIntersection(const lanelet::RegulatoryElementDataPtr& data) : RegulatoryElement(data) {

        // save section_lookup for faster lookup
        for (const auto& param : data->parameters[lanelet::CarmaRoleNameString::IntersectionEntry])
        {
            section_lookup.insert({GetIdVisitor::id(param), IntersectionSection::ENTRY});
        }
        for (const auto& param : data->parameters[lanelet::CarmaRoleNameString::IntersectionExit])
        {
            section_lookup.insert({GetIdVisitor::id(param), IntersectionSection::EXIT});
        }
        for (const auto& param : data->parameters[lanelet::CarmaRoleNameString::IntersectionInterior])
        {
            section_lookup.insert({GetIdVisitor::id(param), IntersectionSection::INTERIOR});
        }
    }

    namespace
    {
    // this object actually does the registration work for us
    lanelet::RegisterRegulatoryElement<lanelet::SignalizedIntersection> reg;
    }  // namespace

} //namespace lanelet