text:
/* Copyright (c) 2013 by Wayne Wright, Round Rock, Texas.
** See license at http://github.com/w5xd/plm/blob/master/LICENSE.md */
#include <cstring>
#include <sstream>
#include <chrono>
#include <iomanip>
#include <algorithm>
#include "InsteonDevice.h"
#include "PlmMonitor.h"
#include <boost/date_time/posix_time/posix_time.hpp> // cuz c++11 chrono can't print times w/o warnings for statics

namespace w5xdInsteon {

const unsigned char InUseFlag = 0x80;
const unsigned char CMD_READ_WRITE_LINK_TABLE = 0x2f;
const unsigned char CMD_EXTENDED_GET_SET = 0x2e;

enum {EXT_MSG_LENGTH = 23,
    EXT_D1=9,
    EXT_D2=10,
    EXT_D3=11,
    EXT_D4=12,
    EXT_D5=13,
    EXT_D6=14,
    };

void InsteonDevice::PlaceCheckSum(unsigned char *extMsg)
{
    // This sum is undocumented, but required for, apparently,
    // extended commands that do a SET.
    unsigned char x = 0;
    for (int i = 6; i <= 20; i++) x += extMsg[i];
    extMsg[21] = ~x + 1;
}

bool InsteonDeviceAddr::operator <(const InsteonDeviceAddr &other) const
{
    if (m_addr[0] < other.m_addr[0])
        return true;
    if (m_addr[0] == other.m_addr[0])
    {
        if (m_addr[1] < other.m_addr[1])
            return true;
        if (m_addr[1] == other.m_addr[1])
        {
            if (m_addr[2] < other.m_addr[2])
                return true;
        }
    }
    return false;
}

const unsigned char InsteonLinkEntry::CONTROLLER_FLAG = 0x40;

void InsteonLinkEntry::print(std::ostream &oss)const
{
    oss << std::hex << std::setw(2) <<  std::setfill(' ') << (int)m_flag << " ";
    if (m_flag & InsteonLinkEntry::CONTROLLER_FLAG)  oss << "c"; else oss << "r";
    oss << " " <<  std::setw(2) <<  std::setfill(' ') <<
        (int)m_group << " ";
    bufferToStream(oss, m_addr, 3);
    oss << " " << std::setw(2)<< std::setfill(' ') <<(int)m_LinkSpecific1 <<
        " " << std::setw(2)<< std::setfill(' ') <<(int)m_LinkSpecific2 <<
        " " << std::setw(2)<< std::setfill(' ') <<(int)m_LinkSpecific3;
}

InsteonDevice::InsteonDevice(PlmMonitor *p, const unsigned char addr[3]) :
    m_plm(p)
   ,m_LinkTableComplete(false)
   ,m_finalAddr(0)
   ,m_lastAcqCommand1(0)
   ,m_incomingMessageCount(0)
   ,m_lastRequestedAddr(0)
{
    m_addr = addr;
}

bool InsteonDevice::operator < (const InsteonDevice &other) const
{    return m_addr < other.m_addr;}

static const char * const FlagLabels[] = {
    "P2P", "P2P ACK", "P2P Cleanup", "P2P Cleanup ACK",
    "Broadcast",   "P2P NAK", "Broadcast Group",
    "P2P Cleanup NAK"
};

void InsteonDevice::incomingMessage(const std::vector<unsigned char> &v, std::shared_ptr<InsteonCommand>)
{
    {
        std::unique_lock<std::mutex> l(m_mutex);
        m_incomingMessageCount += 1;
    }
    if (v.size() >= 25)
    {
        static const unsigned char StartExtended[] = { 0x2, 0x51};
        if (memcmp(&v[START_BYTE], StartExtended, sizeof(StartExtended)) == 0)
        {
            if (m_plm->m_verbosity >= PlmMonitor::MESSAGE_ON)
                dumpFlags(m_plm->cerr(), v);
            if ((v[FLAG_BYTE] & 0xF0) == EXTMSG_BIT)
            {
                if ((v[COMMAND1] == 0x2f) && (v[COMMAND2] == 0))
                {   // link database record
                    unsigned which = v[COMMAND2+3];
                    which <<= 8;
                    which |= v[COMMAND2+4];
                    unsigned char flag = v[COMMAND2+6];
                    std::unique_lock<std::mutex> l(m_mutex);
                    if (!m_LinkTableComplete)
                    {
#ifdef _DEBUGXXX   // force retry of a particular record to test the retry code
                        if ((which == 0xfe7) && (m_lastRequestedAddr == 0))
                            ;
                        else
#endif
                        if (flag & InUseFlag)
                            m_LinkTable[which] = InsteonLinkEntry(&v[COMMAND2+6]);
                        else if (flag != 0)
                            m_UnusedLinks.insert(which);
                        else
                            m_finalAddr = which;
                        // see if we got them all
                        if ((flag == 0) || (m_lastRequestedAddr != 0))
                        {
                            for (unsigned maddr = 0xfff; maddr != m_finalAddr; maddr -= 8)
                            {
                                if (m_LinkTable.find(maddr) == m_LinkTable.end() &&
                                    m_UnusedLinks.find(maddr) == m_UnusedLinks.end())
                                {
                                    // oops.. didn't get them all
                                    if (maddr != m_lastRequestedAddr)
                                        reqAllLinkData(maddr);
                                    return;
                                }
                            }
                            m_LinkTableComplete = true;
                            m_condition.notify_all();
                        }
                    }
                }
                else if ((v[COMMAND1] == CMD_EXTENDED_GET_SET) && (v[COMMAND2] == 0))
                {
                    std::unique_lock<std::mutex> l(m_mutex);
                    ExtendedGetResults_t::iterator itor = m_ExtendedGetResult.insert(
                        ExtendedGetResults_t::value_type(v[COMMAND2 + 1],
                        ExtendedGetResults_t::mapped_type())).first;
                    itor->second.assign(v.begin()+2, v.end());
                    m_condition.notify_all();
                }
                else if ((v[COMMAND1] == 0x3) && (v[COMMAND2] == 0))
                {
                    m_productData.assign(v.begin()+2, v.end());
                    m_condition.notify_all();
                }
            }
            else if ((v[FLAG_BYTE] & 0xF0) == (EXTMSG_BIT|ACK_BIT))
            {
            }
        }
    } else if (v.size() >= 11)
    {
        static const unsigned char StartStandard[] = { 0x2, 0x50};
        if ((memcmp(&v[START_BYTE], StartStandard, sizeof(StartStandard)) == 0))
        {
            if (m_plm->m_verbosity >= PlmMonitor::MESSAGE_ON)
                dumpFlags(m_plm->cerr(), v);

            unsigned char flagbyte = 0xF0 & v[FLAG_BYTE];
            if (flagbyte == GROUP_BIT)
            { // acq this group clean-up message
                static const unsigned char SetAcqMsg[3] = { 0x02, PlmMonitor::SET_ACQ_MSG_BYTE, 0 };
                m_plm->queueCommand(SetAcqMsg, sizeof(SetAcqMsg), 4, false);
            }
            else if (flagbyte == ACK_BIT)
            {
                std::unique_lock<std::mutex> l(m_mutex);
                m_lastAcqCommand1 = v[COMMAND1];
                m_condition.notify_all();
            }
        }
    }
}

void InsteonDevice::dumpFlags(std::ostream &os, const std::vector<unsigned char> &v)
{
    unsigned char flag = v[FLAG_BYTE];
    int offset = 0x7 & (flag >> 5);
    int rtLeft = 0x3 & (flag >> 2);
    int rtAllowed = 0x3 & flag;

    os << FlagLabels[offset] << ", left:" << rtLeft << " allw:" << rtAllowed <<
        " cmd: 0x" << std::hex << std::setw(2) << std::setfill('0') << (int)v[COMMAND1] <<
        ", 0x"     << std::hex << std::setw(2) << std::setfill('0') << (int)v[COMMAND2] << std::endl;
}

int InsteonDevice::numberOfLinks(int msecToWait) const
{
    auto start(std::chrono::steady_clock::now());
    std::unique_lock<std::mutex> l(m_mutex);
    auto toWait =std::chrono::milliseconds(msecToWait);
    while (!m_LinkTableComplete && (std::chrono::steady_clock::now() - start) < toWait)
    {
        int countBefore = m_incomingMessageCount;
        m_condition.wait_for(l, toWait);
        if (!m_LinkTableComplete)
        {
            if (countBefore != m_incomingMessageCount)   // If something changed, then reset the clock
                start = std::chrono::steady_clock::now();
        }
    }
    return m_LinkTableComplete ? static_cast<int>(m_LinkTable.size()) : -1;
}

const char * InsteonDevice::printLinkTable()
{
    std::unique_lock<std::mutex> l(m_mutex);
    if (!m_LinkTableComplete)
    {
       m_plm->cerr() << "No link table retrieved yet" << std::endl;
       return 0;
    }
    std::ostringstream linkTablePrinted;
    linkTablePrinted << "printLinkTable for device " << m_addr << std::endl << "addr flag group  ID       ls1 ls2 ls3" << std::endl;
    for (LinkTable_t::const_iterator itor = m_LinkTable.begin();
            itor != m_LinkTable.end();
            itor++)
    {
        linkTablePrinted << std::hex << std::setw(3) << std::setfill(' ') << (int)itor->first << " " ;
        itor->second.print(linkTablePrinted);
        linkTablePrinted << std::endl;
    }
    linkTablePrinted << "end table" << std::endl << std::flush;
    m_linkTablePrinted = linkTablePrinted.str();
    return m_linkTablePrinted.c_str();
}

void InsteonDevice::reqAllLinkData(unsigned addr)
{
    unsigned char reqAllLinkDatab[EXTMSG_COMMAND_LEN] =
    { 0x02, 0x62, 0x11, 0x11, 0x11, 0x1F, CMD_READ_WRITE_LINK_TABLE, 0x00, };
    memset(&reqAllLinkDatab[7], 0, sizeof(reqAllLinkDatab) - 7);
    memcpy(&reqAllLinkDatab[2], m_addr, 3);
    if (addr != 0)
    {
        reqAllLinkDatab[OFFSET_D5] = 8;  // request one record
        reqAllLinkDatab[OFFSET_D3] = (addr >> 8) & 0xFF;
        reqAllLinkDatab[OFFSET_D4] = addr & 0xFF;
    }
    m_lastRequestedAddr = addr;
    m_plm->queueCommand(reqAllLinkDatab, sizeof(reqAllLinkDatab), 23);
}

int InsteonDevice::startGatherLinkTable()
{
    {
        std::unique_lock<std::mutex> l(m_mutex);
        m_LinkTable.clear();
        m_UnusedLinks.clear();
        m_LinkTableComplete = false;
        m_finalAddr = 0;
        reqAllLinkData(0);
    }
    return 1;
}

int InsteonDevice::linkPlm(bool amController, unsigned char grp, unsigned char ls1, unsigned char ls2, unsigned char ls3)
{
    int res = m_plm->createLink(this, !amController, grp, 0x80, ls1, ls2, ls3);
    int res2 = createLinkWithModem(grp, amController, ls1, ls2, ls3);
    return std::min(res, res2);
}

int InsteonDevice::unLinkPlm(bool amController, unsigned char grp, unsigned char ls3)
{
     int res = m_plm->removeLink(this, !amController, grp);
     int res2 =  removeLinks(m_plm->insteonID(), grp,  amController, ls3);
     return std::min(res, res2);
}

int InsteonDevice::createLinkWithModem(unsigned char group, bool amController, InsteonDevice *other, unsigned char og)
{
    for (LinkTable_t::const_iterator itor = m_LinkTable.begin(); itor != m_LinkTable.end(); itor++)
    {
        if ((itor->second.m_addr == other->m_addr) && (itor->second.m_group == og))
        {
            if (!((itor->second.m_flag & InsteonLinkEntry::CONTROLLER_FLAG ? true : false) ^ amController))
            {
                return createLinkWithModem(group, amController, itor->second.m_LinkSpecific1,
                                                                itor->second.m_LinkSpecific2,
                                                                itor->second.m_LinkSpecific3);
            }
        }
    }
    return -1;
}

void InsteonDevice::InitExtMsg(unsigned char *extMsg, unsigned char cmd1)
{
    unsigned char init[EXTMSG_COMMAND_LEN] =
    { 0x02, 0x62,
        0x11, 0x11, 0x11,   // to addr
        0x1F, cmd1, 0x00,   // flag, cmd1, cmd2
        0x01, 0x02,     // d1, d2
        0x0f, 0xf7,     // d3, d4
        0x08,           // d5
        0xa0, // flag,
        0,    // group#
            0, 0, 0, // addr linked
            0, 0, 0, // ls1, ls2, ls3
             0x00};
    memcpy(extMsg, init, sizeof(init));
}

int InsteonDevice::extendedGet(unsigned char btn, unsigned char *pBuf, unsigned bufsize)
{
    unsigned char extMsg[EXTMSG_COMMAND_LEN];
    InitExtMsg(extMsg, CMD_EXTENDED_GET_SET);
    memcpy(&extMsg[OFFSET_TO_ADDR], m_addr, sizeof(m_addr));
    extMsg[OFFSET_CMD2] = 0;
    memset(&extMsg[OFFSET_D1], 0, 14);
    extMsg[OFFSET_D1] = btn;
    extMsg[OFFSET_D2] = 0;  // data request
    ExtendedGetResults_t::iterator itor;
    {
        std::unique_lock<std::mutex> l(m_mutex);
        itor = m_ExtendedGetResult.find(btn);
        if (itor != m_ExtendedGetResult.end())
            m_ExtendedGetResult.erase(itor);
    }
    static const auto secondsToWait = std::chrono::seconds(5);
    std::unique_lock<std::mutex> l(m_mutex);
    itor = m_ExtendedGetResult.end();
    int attempts = 2;
    while (itor == m_ExtendedGetResult.end() && (attempts-- >= 0))
    {
        auto start(std::chrono::steady_clock::now());
        std::shared_ptr<InsteonCommand> p = m_plm->queueCommand(extMsg, sizeof(extMsg), 23);
        while (itor == m_ExtendedGetResult.end() &&
            (std::chrono::steady_clock::now() - start) < secondsToWait)
        {
            int msgCount = m_incomingMessageCount;
            m_condition.wait_for(l, secondsToWait);
            itor = m_ExtendedGetResult.find(btn);
            if ((msgCount != m_incomingMessageCount) || (p->m_timesSent == 0))
                start = std::chrono::steady_clock::now();

            if (static_cast<int>(m_plm->m_verbosity) >= PlmMonitor::MESSAGE_EVERY_IO)
            {
                m_plm->cerr() << std::dec << "Get: at " << boost::posix_time::microsec_clock::universal_time() << " btn " << (int)btn <<
                    " msgCount " << msgCount << " attempts " << attempts <<
                    " m_incomingMessageCount " << m_incomingMessageCount <<
                    (itor == m_ExtendedGetResult.end() ? " not found" : " found") << std::endl;
            }
        }
    }
    if (itor == m_ExtendedGetResult.end())
        return -1;
    unsigned ret = static_cast<unsigned>(itor->second.size());
    static const int IGNORE_CHARS = 9;
    if (ret >= IGNORE_CHARS)
        ret -= IGNORE_CHARS;
    else
        return -2;
    if (!pBuf)
        return 0;   // successfully return nothing
    if (ret > bufsize)
        ret = bufsize;
    if (ret)
        memcpy(pBuf, &itor->second[IGNORE_CHARS], ret);
    return  ret;
}

const char * InsteonDevice::printExtendedGet(unsigned char btn)
{
    std::unique_lock<std::mutex> l(m_mutex);
    ExtendedGetResults_t::const_iterator itor = m_ExtendedGetResult.find(btn);
    if (itor == m_ExtendedGetResult.end() )
    {
       m_plm->cerr() << "No extended get/set response retrieved yet" << std::endl;
       return 0;
    }
    std::ostringstream oss;
    oss << "Extended state for device " << m_addr << std::endl <<
"                                             D1   D2   D3   D4   D5   D6   D7   D8" << std::endl <<
"                                             btn  resp unu  unu  X10H X10U RAMP ONLV" << std::endl;
    bufferToStream(oss, &itor->second[0], static_cast<int>(itor->second.size()));
    oss << std::endl;
    m_extendedGetPrint[btn] = oss.str();
    return m_extendedGetPrint[btn].c_str();
}

int InsteonDevice::createLinkWithModem(unsigned char group, bool amController,
                                       unsigned char ls1, unsigned char ls2, unsigned char ls3)
{
    if (!linktableComplete()) return -2;
    // Calculate d3 and d4 as the next link byte offset to use
    // Notice if the controllers already have a link to the same addr
    // and group and overwrite that one.
    InsteonDeviceAddr plmAddr(m_plm->insteonID());
    int addr = linkAddr(plmAddr, group, amController, ls3);
    unsigned char extMsg[EXTMSG_COMMAND_LEN];
    InitExtMsg(extMsg, CMD_READ_WRITE_LINK_TABLE);
    extMsg[OFFSET_LINK_GROUP] = group;
    extMsg[OFFSET_LINK_LS1] = ls1;
    extMsg[OFFSET_LINK_LS2] = ls2;
    extMsg[OFFSET_LINK_LS3] = ls3;

    extMsg[OFFSET_D3] = static_cast<unsigned char>(addr >> 8);
    extMsg[OFFSET_D4] = static_cast<unsigned char>(addr);

    memcpy(&extMsg[OFFSET_TO_ADDR], m_addr, sizeof(m_addr));
    memcpy(&extMsg[OFFSET_LINK_ADDR], plmAddr, 3);
    extMsg[OFFSET_LINK_FLAG] = amController ? 0xe2 : 0xa2;  // flag as Controller/responder
    if (m_plm->m_verbosity >= static_cast<int>(m_plm->MESSAGE_ON))
        m_plm->cerr() << "Attemping link on address " << std::hex << addr << " and group " << (int)group << std::endl;

    PlaceCheckSum(extMsg);
    std::shared_ptr<InsteonCommand> p = m_plm->sendCommandAndWait(extMsg, sizeof(extMsg), 23);
	invalidateLinkTable();
    return p->m_answerState;
}

int InsteonDevice::createLink(InsteonDevice *responder, unsigned char group,
                              unsigned char ls1, unsigned char ls2, unsigned char ls3)
{
    // creates 2 links. One in "this" as controller, and the other
    // in "responder" as responder. ls1, ls2, ls3 are added to the link table on both
    // controller and responder. (Because their values on the controller appear to be irrelevant)
    if (memcmp(&m_addr[0], &responder->m_addr[0], sizeof(m_addr)) == 0)
        return -1; // can't link to self
    if (!linktableComplete()) return -2;
    if (!responder->linktableComplete()) return -3;

    unsigned char extMsg[EXTMSG_COMMAND_LEN];
    InitExtMsg(extMsg, CMD_READ_WRITE_LINK_TABLE);
    extMsg[OFFSET_LINK_GROUP] = group;
    extMsg[OFFSET_LINK_LS1] = ls1;
    extMsg[OFFSET_LINK_LS2] = ls2;
    extMsg[OFFSET_LINK_LS3] = ls3;

    // Calculate d3 and d4 as the next link byte offset to use
    // Notice if the controllers already have a link to the same addr
    // and group and overwrite that one.
    int addr = linkAddr(responder->m_addr, group, true, ls3);
    extMsg[OFFSET_D3] = static_cast<unsigned char>(addr >> 8);
    extMsg[OFFSET_D4] = static_cast<unsigned char>(addr);

    memcpy(&extMsg[OFFSET_TO_ADDR], m_addr, sizeof(m_addr));
    memcpy(&extMsg[OFFSET_LINK_ADDR], responder->m_addr, sizeof(responder->m_addr));
    extMsg[OFFSET_LINK_FLAG] = 0xea;  // flag as Controller
    PlaceCheckSum(extMsg);
    std::shared_ptr<InsteonCommand> p = m_plm->sendCommandAndWait(extMsg, sizeof(extMsg), 23); // controller

    addr = responder->linkAddr(m_addr, group, false, ls3);
    extMsg[OFFSET_D3] = static_cast<unsigned char>(addr >> 8);
    extMsg[OFFSET_D4] = static_cast<unsigned char>(addr);
    extMsg[OFFSET_D5] = 8;
    memcpy(&extMsg[OFFSET_LINK_ADDR], m_addr, sizeof(m_addr));
    memcpy(&extMsg[OFFSET_TO_ADDR], responder->m_addr, sizeof(responder->m_addr));
    extMsg[OFFSET_LINK_FLAG] = 0xaa;  // flag as Responder
    PlaceCheckSum(extMsg);
    p = m_plm->sendCommandAndWait(extMsg, sizeof(extMsg), 23);
	invalidateLinkTable();
    return 1;
}

int InsteonDevice::createModemGroupToMatch(int grp)
{
    int ret = 0;
    static const int GROUP_1 = 1;
    if (!m_LinkTableComplete) return -1;
    std::vector<InsteonDevice *> othersToLink;
    {
        std::unique_lock<std::mutex> l(m_mutex);
        for (LinkTable_t::iterator itor = m_LinkTable.begin(); itor != m_LinkTable.end(); itor++)
        {
            InsteonLinkEntry &link = itor->second;
            if ((link.m_flag & InsteonLinkEntry::CONTROLLER_FLAG) && (link.m_group == GROUP_1))
            {   // controller on group 1
                othersToLink.push_back(m_plm->getDeviceAccess<InsteonDevice>(link.m_addr));
            }
        }
    }
    if (othersToLink.empty()) return 0;
    for (unsigned i = 0; i < othersToLink.size(); i++)
    {
        othersToLink[i]->startGatherLinkTable();
        if (othersToLink[i]->numberOfLinks() <= 0)
            return -3;
    }

    if (grp <= 0)
        grp = m_plm->nextUnusedControlGroup();
    if (grp <= 0)
        return -2;

    // try to create a link in our modem first
    // The ls3 == 1 means "this link can be commanded instead of the individual device
    int res = m_plm->createLink(this, true, grp, 0xA0, 0, 0, 1);
    if (res < 0)
        return -4;
    for (unsigned i = 0; i < othersToLink.size(); i++)
        if (m_plm->createLink(othersToLink[i], true, grp, 0xA0, 0, 0, 0) < 0)
            return -5;

    createLinkWithModem(grp, false, 0xff, 0, 0);
    for (unsigned i = 0; i < othersToLink.size(); i++)
        othersToLink[i]->createLinkWithModem(grp, false, this, GROUP_1);
    return ret;
}

int InsteonDevice::linkAddr(const InsteonDeviceAddr &dev, unsigned char grp, bool isControl, unsigned char ls3) const
{
    int addr = 0xfff;   // first address per documentation
    if (!m_LinkTable.empty())
        addr = m_LinkTable.begin()->first - 8;  // one previous to existing smallest number
    int nextAddr = 0xfff;
    for (LinkTable_t::const_reverse_iterator itor = m_LinkTable.rbegin(); itor != m_LinkTable.rend(); itor++)
    {
        if (itor->first != nextAddr)    // there is a gap here. use it
            addr = nextAddr;
        if ((itor->second.m_addr == dev) && (itor->second.m_group == grp))
        {
            if (!((itor->second.m_flag & InsteonLinkEntry::CONTROLLER_FLAG ? true : false) ^ isControl)) // control bits match?
            {
                if (isControl || (ls3 == 0) || (ls3 == itor->second.m_LinkSpecific3))
                {   // for responders, the ls3 info must match in order to override.
                    // ls3 is "part of" the responder that is to respond.
                    // Caller setting ls3 to zero matches all "parts"
                    addr = itor->first;
                    break;
                }
            }
        }
        nextAddr = itor->first - 8;
    }
    return addr;
}

int InsteonDevice::removeLink(InsteonDevice *responder, unsigned char group, unsigned char ls3)
{
    int r1 = responder->removeLinks(m_addr, group, false, ls3);
    int r2 = 0;
    if (!responder->amRespondingTo(m_addr, group))   // only if removed last responder link does controller  also get removed
        r2 = removeLinks(responder->m_addr, group, true, ls3);
    if (r1 <= 0)
        return r1;
    if (r2 <= 0)
        return r2;
    return 0;
}

bool InsteonDevice::amRespondingTo(const InsteonDeviceAddr &addr, unsigned char group)const
{
    for (LinkTable_t::const_reverse_iterator itor = m_LinkTable.rbegin(); itor != m_LinkTable.rend(); itor++)
    {
        const InsteonLinkEntry &link = itor->second;
        if ((link.m_flag & InUseFlag) &&
            (link.m_addr == addr) &&
            (link.m_group == group) &&
            !(link.m_flag & InsteonLinkEntry::CONTROLLER_FLAG ? true : false)
            )
            return true;
    }
    return false;
}

int InsteonDevice::removeLinks(const InsteonDeviceAddr &addr, unsigned char group, bool amController, unsigned char ls3)
{
    if (m_LinkTable.empty())
        return 0;
    for (LinkTable_t::reverse_iterator itor = m_LinkTable.rbegin(); itor != m_LinkTable.rend(); itor++)
    {
        InsteonLinkEntry &link = itor->second;
        if ((link.m_flag & InUseFlag) &&
            (link.m_addr == addr) &&
            (link.m_group == group) &&
            !((link.m_flag & InsteonLinkEntry::CONTROLLER_FLAG ? true : false) ^ amController) &&
            (amController || (ls3 == 0) || (ls3 == link.m_LinkSpecific3)) // responder only matches if ls3 matches as well
            )
        {
                unsigned char extMsg[EXTMSG_COMMAND_LEN];
                unsigned int maddr = itor->first;
                InitExtMsg(extMsg, CMD_READ_WRITE_LINK_TABLE);
                extMsg[OFFSET_LINK_GROUP] = link.m_group;
                extMsg[OFFSET_LINK_LS1] = link.m_LinkSpecific1;
                extMsg[OFFSET_LINK_LS2] = link.m_LinkSpecific2;
                extMsg[OFFSET_LINK_LS3] = link.m_LinkSpecific3;
                extMsg[OFFSET_D3] = static_cast<unsigned char>(maddr >> 8);
                extMsg[OFFSET_D4] = static_cast<unsigned char>(maddr);
                extMsg[OFFSET_D5] = 8;

                memcpy(&extMsg[OFFSET_TO_ADDR], m_addr, sizeof(m_addr));
                memcpy(&extMsg[OFFSET_LINK_ADDR], link.m_addr, 3);
                extMsg[OFFSET_LINK_FLAG] = link.m_flag & ~InUseFlag;  // clear the InUseFlag
                if (itor->first == m_LinkTable.begin()->first) // if removing the link with lowest address
                    memset(&extMsg[OFFSET_LINK_FLAG], 0, 8);   //...then tell device truncate table here.
                m_lastAcqCommand1 = 0;
                PlaceCheckSum(extMsg);
                std::shared_ptr<InsteonCommand> p = m_plm->sendCommandAndWait(extMsg, sizeof(extMsg), 23);
                if (p->m_answerState < 0)
                    return p->m_answerState;
                else
                {
                    link.m_flag &= ~InUseFlag;

                    // hang around for 5 seconds or until receive an acq, which is first
                    static const auto secondsToWait = std::chrono::seconds(5);
                    auto start(std::chrono::steady_clock::now());
                    std::unique_lock<std::mutex> l(m_mutex);
                    while ((m_lastAcqCommand1 != 0x2f) && (std::chrono::steady_clock::now() - start) < secondsToWait)
                    {
                        m_condition.wait_for(l, secondsToWait);
                    }
                }
        }
    }
    return 0;
}

// http://software.x10.com/pub/manuals/xtdcode.pdf
const char InsteonDevice::X10HouseCodeToLetter[] =
{    'M', 'E', 'C', 'K', 'O', 'G', 'A', 'I', 'N', 'F', 'D', 'L', 'P', 'H', 'B', 'J'};
const char InsteonDevice::X10HouseLetterToBits[] =
{    6, 0xe, 2, 0xa, 1, 9, 5, 0xd, 7, 0xf, 3, 0xb, 0, 8, 4, 0xc};
const char InsteonDevice::X10BitsToWheelCode[]=
{   13, 5, 3, 11, 15, 7, 1, 9, 14, 6, 4, 12, 16, 8, 2, 10};
const char InsteonDevice::X10WheelCodeToBits[] =
{   0, 0x0c, 0x1c, 0x4, 0x14, 0x2, 0x12, 0xa, 0x1a, 0xe, 0x1e, 0x6, 0x16, 0x0, 0x10, 0x8, 0x18};

int InsteonDevice::getX10Code(char &houseCode, unsigned char &unit, unsigned char btn) const
{
    ExtendedGetResults_t::const_iterator itor = m_ExtendedGetResult.find(btn);
    if (itor == m_ExtendedGetResult.end())
        return -1;
    if (itor->second.size() < EXT_MSG_LENGTH)
        return -2;
    unsigned char hc = itor->second[EXT_D5];
    unsigned char u = itor->second[EXT_D6];
    if ((hc == 0x20) || (u == 0x20))
    {
        houseCode = 0;
        unit = 0;
    }
    else
    {
        houseCode = X10HouseCodeToLetter[hc & 0xF];
        unit = X10BitsToWheelCode[0xf & (u >> 1)];
    }
    return 1;
}

int InsteonDevice::setX10Code(char houseCode, unsigned char unit, unsigned char btn)
{
    unsigned char hc = 0x20, u = 0x20;
    if ((houseCode != 0) || (unit != 0))
    {
        if ((houseCode < 'A') || (houseCode > 'P'))
            return -12;
        if ((unit < 1) || (unit > 16))
            return -13;
        hc = X10HouseLetterToBits[houseCode - 'A'];
        u = X10WheelCodeToBits[unit];
    }
    return sendExtendedCommand(btn, 0x4, hc, u);
}

int InsteonDevice::sendExtendedCommand(unsigned char btn, unsigned char d2, unsigned char d3, unsigned char d4)
{
    unsigned char extMsg[EXTMSG_COMMAND_LEN];
    InitExtMsg(extMsg, CMD_EXTENDED_GET_SET); // extended Set/Get
    memset(&extMsg[OFFSET_D1], 0, 14);
    memcpy(&extMsg[OFFSET_TO_ADDR], m_addr, sizeof(m_addr));
    extMsg[OFFSET_CMD2] = 0;
    extMsg[OFFSET_D1] = btn;
    extMsg[OFFSET_D2] = d2;      // SET command--X10
    extMsg[OFFSET_D3] =  d3;
    extMsg[OFFSET_D4] =  d4;
    extMsg[OFFSET_D5] = 0;
    extMsg[OFFSET_D6] = 0;
    PlaceCheckSum(extMsg);
    m_plm->queueCommand(extMsg, sizeof(extMsg), 23);
    return 0;
}

int InsteonDevice::getProductData()
{   // keypad responds. togglelinc does not
    unsigned char getPd[8] =  { 0x02, 0x62, 0, 0, 0, 0xF, 0x3, 0};
    memcpy(&getPd[2], m_addr, 3);
    std::unique_lock<std::mutex> l(m_mutex);
   auto start(std::chrono::steady_clock::now());
    m_productData.clear();
    std::shared_ptr<InsteonCommand> p = m_plm->queueCommand(getPd, sizeof(getPd), 9);
    static const auto secondsToWait = std::chrono::seconds(5);
    while (m_productData.empty() &&
        (std::chrono::steady_clock::now() - start) < secondsToWait)
    {
        m_condition.wait_for(l, secondsToWait);
    }
    return static_cast<int>(m_productData.size());
}

int InsteonDevice::truncateUnusedLinks()
{
    std::set<unsigned> UnusedLinks;
    LinkTable_t::key_type leastAddr;
    {
        std::unique_lock<std::mutex> l(m_mutex);
        if (!m_LinkTableComplete) return -2;
        if (m_LinkTable.empty()) return 0;
        leastAddr = m_LinkTable.begin()->first;
        UnusedLinks = m_UnusedLinks;
    }
    int ret = 0;
    for (
        std::set<unsigned>::iterator itor = UnusedLinks.begin();
        itor != UnusedLinks.end();
        itor ++)
        {
            unsigned linkToZero = *itor;
            if (linkToZero >= leastAddr)
                break;
            unsigned char extMsg[EXTMSG_COMMAND_LEN];
            InitExtMsg(extMsg, CMD_READ_WRITE_LINK_TABLE);
            memset(&extMsg[OFFSET_D1], 0, 14);
            extMsg[OFFSET_D2] = 2;
            memcpy(&extMsg[OFFSET_TO_ADDR], m_addr, sizeof(m_addr));
            extMsg[OFFSET_D3] = static_cast<unsigned char>(linkToZero >> 8);
            extMsg[OFFSET_D4] = static_cast<unsigned char>(linkToZero);
            extMsg[OFFSET_D5] = 8;
            PlaceCheckSum(extMsg);
            std::shared_ptr<InsteonCommand> p = m_plm->sendCommandAndWait(extMsg, sizeof(extMsg), 23);
            if (p->m_answerState < 0)
            {
                ret = p->m_answerState;
                break;
            }
            ret += 1;
        }
    return ret;
}

std::ostream & operator << (std::ostream &os, const InsteonDeviceAddr &dev)
{
    bufferToStream(os, dev.m_addr, sizeof(dev.m_addr));
    return os;
}

}


text:
#pragma once

#include <boost/container/small_vector.hpp>
#include <boost/lexical_cast.hpp>

#include "bytell_hash_map.hpp"
#include "operators/multi_predicate_join/multi_predicate_join_evaluator.hpp"
#include "resolve_type.hpp"
#include "scheduler/abstract_task.hpp"
#include "scheduler/current_scheduler.hpp"
#include "scheduler/job_task.hpp"
#include "storage/create_iterable_from_segment.hpp"
#include "storage/segment_iterate.hpp"
#include "type_comparison.hpp"
#include "uninitialized_vector.hpp"

/*
  This file includes the functions that cover the main steps of our hash join implementation
  (e.g., build() and probe()). These free functions are put into this header file to separate
  them from the process flow of the join hash and to make the better testable.
*/
namespace opossum {

using Hash = size_t;

/*
This is how elements of the input relations are saved after materialization.
The original value is used to detect hash collisions.
*/
template <typename T>
struct PartitionedElement {
  PartitionedElement() : row_id(NULL_ROW_ID), value(T()) {}
  PartitionedElement(RowID row, T val) : row_id(row), value(val) {}

  RowID row_id;
  T value;
};

// Initializing the partition vector takes some time. This is not necessary, because it will be overwritten anyway.
// The uninitialized_vector behaves like a regular std::vector, but the entries are initially invalid.
template <typename T>
using Partition = std::conditional_t<std::is_trivially_destructible_v<T>, uninitialized_vector<PartitionedElement<T>>,
                                     std::vector<PartitionedElement<T>>>;

// The small_vector holds the first n values in local storage and only resorts to heap storage after that. 1 is chosen
// as n because in many cases, we join on primary key attributes where by definition we have only one match on the
// smaller side.
using SmallPosList = boost::container::small_vector<RowID, 1>;

// In case we consider runtime to be more relevant, the flat hash map performs better (measured to be mostly on par
// with bytell hash map and in some cases up to 5% faster) but is significantly larger than the bytell hash map.
template <typename T>
using HashTable = ska::bytell_hash_map<T, SmallPosList>;

/*
This struct contains radix-partitioned data in a contiguous buffer, as well as a list of offsets for each partition.
The offsets denote the accumulated sizes (we cannot use the last element's position because we could not recognize
empty first containers).

This struct is used in two phases:
  - the result of the materialization phase, at this time partitioned by chunks
    as we parallelize the materialization phase via chunks
  - the result of the radix clustering phase

As the radix clustering might be skipped (when radix_bits == 0), both the materialization as well as the radix
clustering methods yield RadixContainers.
*/
template <typename T>
struct RadixContainer {
  std::shared_ptr<Partition<T>> elements;
  std::vector<size_t> partition_offsets;

  // bit vector to store NULL flags
  std::shared_ptr<std::vector<bool>> null_value_bitvector;
};

inline std::vector<size_t> determine_chunk_offsets(const std::shared_ptr<const Table>& table) {
  const auto chunk_count = table->chunk_count();
  auto chunk_offsets = std::vector<size_t>(chunk_count);

  size_t offset = 0;
  for (ChunkID chunk_id{0}; chunk_id < chunk_count; ++chunk_id) {
    chunk_offsets[chunk_id] = offset;
    offset += table->get_chunk(chunk_id)->size();
  }
  return chunk_offsets;
}

template <typename T, typename HashedType, bool retain_null_values>
RadixContainer<T> materialize_input(const std::shared_ptr<const Table>& in_table, ColumnID column_id,
                                    const std::vector<size_t>& chunk_offsets,
                                    std::vector<std::vector<size_t>>& histograms, const size_t radix_bits) {
  const std::hash<HashedType> hash_function;
  // list of all elements that will be partitioned
  auto elements = std::make_shared<Partition<T>>(in_table->row_count());

  [[maybe_unused]] auto null_value_bitvector = std::make_shared<std::vector<bool>>();
  if constexpr (retain_null_values) {
    null_value_bitvector->resize(in_table->row_count());
  }

  // fan-out
  const size_t num_partitions = 1ull << radix_bits;

  // currently, we just do one pass
  size_t pass = 0;
  size_t mask = static_cast<uint32_t>(pow(2, radix_bits * (pass + 1)) - 1);

  // create histograms per chunk
  histograms.resize(chunk_offsets.size());

  std::vector<std::shared_ptr<AbstractTask>> jobs;
  jobs.reserve(in_table->chunk_count());

  for (ChunkID chunk_id{0}; chunk_id < in_table->chunk_count(); ++chunk_id) {
    jobs.emplace_back(std::make_shared<JobTask>([&, chunk_id]() {
      // Get information from work queue
      auto output_offset = chunk_offsets[chunk_id];
      auto output_iterator = elements->begin() + output_offset;
      auto segment = in_table->get_chunk(chunk_id)->get_segment(column_id);

      [[maybe_unused]] auto null_value_bitvector_iterator = null_value_bitvector->begin();
      if constexpr (retain_null_values) {
        null_value_bitvector_iterator += output_offset;
      }

      // prepare histogram
      auto histogram = std::vector<size_t>(num_partitions);

      auto reference_chunk_offset = ChunkOffset{0};

      segment_with_iterators<T>(*segment, [&](auto it, const auto end) {
        using IterableType = typename decltype(it)::IterableType;

        while (it != end) {
          const auto& value = *it;
          ++it;

          if (!value.is_null() || retain_null_values) {
            // TODO(anyone): static_cast is almost always safe, since HashType is big enough. Only for double-vs-long
            // joins an information loss is possible when joining with longs that cannot be losslessly converted to
            // double
            const Hash hashed_value = hash_function(static_cast<HashedType>(value.value()));

            /*
            For ReferenceSegments we do not use the RowIDs from the referenced tables.
            Instead, we use the index in the ReferenceSegment itself. This way we can later correctly dereference
            values from different inputs (important for Multi Joins).
            */
            if constexpr (is_reference_segment_iterable_v<IterableType>) {
              *(output_iterator++) = PartitionedElement<T>{RowID{chunk_id, reference_chunk_offset}, value.value()};
            } else {
              *(output_iterator++) = PartitionedElement<T>{RowID{chunk_id, value.chunk_offset()}, value.value()};
            }

            // In case we care about NULL values, store the NULL flag
            if constexpr (retain_null_values) {
              if (value.is_null()) {
                *null_value_bitvector_iterator = true;
              }
            }

            const Hash radix = hashed_value & mask;
            ++histogram[radix];
            ++null_value_bitvector_iterator;
          }
          // reference_chunk_offset is only used for ReferenceSegments
          if constexpr (is_reference_segment_iterable_v<IterableType>) {
            ++reference_chunk_offset;
          }
        }
      });

      if constexpr (std::is_same_v<Partition<T>, uninitialized_vector<PartitionedElement<T>>>) {  // NOLINT
        // Because the vector is uninitialized, we need to manually fill up all slots that we did not use
        auto output_offset_end = chunk_id < chunk_offsets.size() - 1 ? chunk_offsets[chunk_id + 1] : elements->size();
        while (output_iterator != elements->begin() + output_offset_end) {
          *(output_iterator++) = PartitionedElement<T>{};
        }
      }

      histograms[chunk_id] = std::move(histogram);
    }));
    jobs.back()->schedule();
  }
  CurrentScheduler::wait_for_tasks(jobs);

  return RadixContainer<T>{elements, std::vector<size_t>{elements->size()}, null_value_bitvector};
}

/*
Build all the hash tables for the partitions of the build column. One job per partition
*/

// For semi and anti joins, we only care whether a value exists or not, so there is no point in tracking the position
// in the input table of more than one occurrence of a value. However, if we have secondary predicates, we do need to
// track all occurrences of a value as that first position might be disqualified later.
enum class JoinHashBuildMode { AllPositions, SinglePosition };

template <typename BuildColumnType, typename HashedType, JoinHashBuildMode mode>
std::vector<std::optional<HashTable<HashedType>>> build(const RadixContainer<BuildColumnType>& radix_container) {
  /*
  NUMA notes:
  The hashtables for each partition P should also reside on the same node as the two vectors buildP and probeP.
  */
  std::vector<std::optional<HashTable<HashedType>>> hashtables;
  hashtables.resize(radix_container.partition_offsets.size());

  std::vector<std::shared_ptr<AbstractTask>> jobs;
  jobs.reserve(radix_container.partition_offsets.size());

  for (size_t current_partition_id = 0; current_partition_id < radix_container.partition_offsets.size();
       ++current_partition_id) {
    const auto build_partition_begin =
        current_partition_id == 0 ? 0 : radix_container.partition_offsets[current_partition_id - 1];
    const auto build_partition_end = radix_container.partition_offsets[current_partition_id];  // make end non-inclusive
    const auto build_partition_size = build_partition_end - build_partition_begin;

    // Skip empty partitions, so that we don't have too many empty hash tables
    if (build_partition_size == 0) {
      continue;
    }

    jobs.emplace_back(std::make_shared<JobTask>(
        [&, build_partition_begin, build_partition_end, current_partition_id, build_partition_size]() {
          auto& build_partition = static_cast<Partition<BuildColumnType>&>(*radix_container.elements);

          // slightly oversize the hash table to avoid unnecessary rebuilds
          auto hashtable = HashTable<HashedType>(static_cast<size_t>(build_partition_size * 1.2));

          for (size_t partition_offset = build_partition_begin; partition_offset < build_partition_end;
               ++partition_offset) {
            auto& element = build_partition[partition_offset];

            if (element.row_id == NULL_ROW_ID) {
              // Skip initialized PartitionedElements that might remain after materialization phase.
              continue;
            }

            auto casted_value = static_cast<HashedType>(std::move(element.value));
            auto it = hashtable.find(casted_value);
            if (it != hashtable.end()) {
              if constexpr (mode == JoinHashBuildMode::AllPositions) {
                it->second.emplace_back(element.row_id);
              }
            } else {
              hashtable.emplace(casted_value, SmallPosList{element.row_id});
            }
          }

          hashtables[current_partition_id] = std::move(hashtable);
        }));
    jobs.back()->schedule();
  }

  CurrentScheduler::wait_for_tasks(jobs);

  return hashtables;
}

template <typename T, typename HashedType, bool retain_null_values>
RadixContainer<T> partition_radix_parallel(const RadixContainer<T>& radix_container,
                                           const std::vector<size_t>& chunk_offsets,
                                           std::vector<std::vector<size_t>>& histograms, const size_t radix_bits) {
  if constexpr (retain_null_values) {
    DebugAssert(radix_container.null_value_bitvector->size() == radix_container.elements->size(),
                "partition_radix_parallel() called with NULL consideration but radix container does not store any NULL "
                "value information");
  }

  const std::hash<HashedType> hash_function;

  // materialized items of radix container
  const auto& container_elements = *radix_container.elements;
  [[maybe_unused]] const auto& null_value_bitvector = *radix_container.null_value_bitvector;

  // fan-out
  const size_t num_partitions = 1ull << radix_bits;

  // currently, we just do one pass
  size_t pass = 0;
  size_t mask = static_cast<uint32_t>(pow(2, radix_bits * (pass + 1)) - 1);

  // allocate new (shared) output
  auto output = std::make_shared<Partition<T>>();
  output->resize(container_elements.size());

  [[maybe_unused]] auto output_nulls = std::make_shared<std::vector<bool>>();
  if constexpr (retain_null_values) {
    output_nulls->resize(null_value_bitvector.size());
  }

  RadixContainer<T> radix_output;
  radix_output.elements = output;
  radix_output.partition_offsets.resize(num_partitions);
  radix_output.null_value_bitvector = output_nulls;

  // use histograms to calculate partition offsets
  size_t offset = 0;
  std::vector<std::vector<size_t>> output_offsets_by_chunk(chunk_offsets.size(), std::vector<size_t>(num_partitions));
  for (size_t partition_id = 0; partition_id < num_partitions; ++partition_id) {
    for (ChunkID chunk_id{0}; chunk_id < chunk_offsets.size(); ++chunk_id) {
      output_offsets_by_chunk[chunk_id][partition_id] = offset;
      offset += histograms[chunk_id][partition_id];
    }
    radix_output.partition_offsets[partition_id] = offset;
  }

  std::vector<std::shared_ptr<AbstractTask>> jobs;
  jobs.reserve(chunk_offsets.size());

  for (ChunkID chunk_id{0}; chunk_id < chunk_offsets.size(); ++chunk_id) {
    jobs.emplace_back(std::make_shared<JobTask>([&, chunk_id]() {
      size_t input_offset = chunk_offsets[chunk_id];
      auto& output_offsets = output_offsets_by_chunk[chunk_id];

      size_t input_size = 0;
      if (chunk_id < chunk_offsets.size() - 1) {
        input_size = chunk_offsets[chunk_id + 1] - input_offset;
      } else {
        input_size = container_elements.size() - input_offset;
      }

      for (size_t chunk_offset = input_offset; chunk_offset < input_offset + input_size; ++chunk_offset) {
        auto& element = container_elements[chunk_offset];

        // In case of NULL-removing inner-joins, we ignore all NULL values.
        // Such values can be created in several ways: join input already has non-phyiscal NULL values (non-physical
        // means no RowID, e.g., created during an OUTER join), a physical value is NULL but is ignored for an inner
        // join (hence, we overwrite the RowID with NULL_ROW_ID), or it is simply a remainder of the pre-sized
        // RadixPartition which is initialized with default values (i.e., NULL_ROW_IDs).
        if (!retain_null_values && element.row_id == NULL_ROW_ID) {
          continue;
        }

        const size_t radix = hash_function(static_cast<HashedType>(element.value)) & mask;

        // In case NULL values have been materialized in materialize_input(),
        // we need to keep them during the radix clustering phase.
        if constexpr (retain_null_values) {
          (*output_nulls)[output_offsets[radix]] = null_value_bitvector[chunk_offset];
        }

        (*output)[output_offsets[radix]] = element;
        ++output_offsets[radix];
      }
    }));
    jobs.back()->schedule();
  }

  CurrentScheduler::wait_for_tasks(jobs);

  return radix_output;
}

/*
  In the probe phase we take all partitions from the probe partition, iterate over them and compare each join candidate
  with the values in the hash table. Since build and probe are hashed using the same hash function, we can reduce the
  number of hash tables that need to be looked into to just 1.
  */
template <typename ProbeColumnType, typename HashedType, bool keep_null_values>
void probe(const RadixContainer<ProbeColumnType>& probe_radix_container,
           const std::vector<std::optional<HashTable<HashedType>>>& hash_tables,
           std::vector<PosList>& pos_lists_build_side, std::vector<PosList>& pos_lists_probe_side, const JoinMode mode,
           const Table& build_table, const Table& probe_table,
           const std::vector<OperatorJoinPredicate>& secondary_join_predicates) {
  std::vector<std::shared_ptr<AbstractTask>> jobs;
  jobs.reserve(probe_radix_container.partition_offsets.size());

  /*
    NUMA notes:
    At this point both input relations are partitioned using radix partitioning.
    Probing will be done per partition for both sides.
    Therefore, inputs for one partition should be located on the same NUMA node,
    and the job that probes that partition should also be on that NUMA node.
  */
  for (size_t current_partition_id = 0; current_partition_id < probe_radix_container.partition_offsets.size();
       ++current_partition_id) {
    const auto partition_begin =
        current_partition_id == 0 ? 0 : probe_radix_container.partition_offsets[current_partition_id - 1];
    const auto partition_end = probe_radix_container.partition_offsets[current_partition_id];  // make end non-inclusive

    // Skip empty partitions to avoid empty output chunks
    if (partition_begin == partition_end) {
      continue;
    }

    jobs.emplace_back(std::make_shared<JobTask>([&, partition_begin, partition_end, current_partition_id]() {
      // Get information from work queue
      auto& partition = static_cast<Partition<ProbeColumnType>&>(*probe_radix_container.elements);
      PosList pos_list_build_side_local;
      PosList pos_list_probe_local;

      if constexpr (keep_null_values) {
        DebugAssert(
            probe_radix_container.null_value_bitvector->size() == probe_radix_container.elements->size(),
            "Hash join probe called with NULL consideration but inputs do not store any NULL value information");
      }

      if (hash_tables[current_partition_id]) {
        const auto& hash_table = hash_tables.at(current_partition_id).value();

        // Accessors are not thread-safe, so we create one evaluator per job
        std::optional<MultiPredicateJoinEvaluator> multi_predicate_join_evaluator;
        if (!secondary_join_predicates.empty()) {
          multi_predicate_join_evaluator.emplace(build_table, probe_table, mode, secondary_join_predicates);
        }

        // simple heuristic to estimate result size: half of the partition's rows will match
        // a more conservative pre-allocation would be the size of the build cluster
        const size_t expected_output_size =
            static_cast<size_t>(std::max(10.0, std::ceil((partition_end - partition_begin) / 2)));
        pos_list_build_side_local.reserve(static_cast<size_t>(expected_output_size));
        pos_list_probe_local.reserve(static_cast<size_t>(expected_output_size));

        for (size_t partition_offset = partition_begin; partition_offset < partition_end; ++partition_offset) {
          auto& probe_column_element = partition[partition_offset];

          if (mode == JoinMode::Inner && probe_column_element.row_id == NULL_ROW_ID) {
            // From previous joins, we could potentially have NULL values that do not refer to
            // an actual probe_column_element but to the NULL_ROW_ID. Hence, we can only skip for inner joins.
            continue;
          }

          const auto& iter = hash_table.find(static_cast<HashedType>(probe_column_element.value));

          if (iter != hash_table.end()) {
            // Key exists, thus we have at least one hit for the primary predicate
            const auto& primary_predicate_matching_rows = iter->second;

            // Since we cannot store NULL values directly in off-the-shelf containers,
            // we need to the check the NULL bit vector here because a NULL value (represented
            // as a zero) yields the same rows as an actual zero value.
            // For inner joins, we skip NULL values and output them for outer joins.
            // Note, if the materialization/radix partitioning phase did not explicitly consider
            // NULL values, they will not be handed to the probe function.
            if constexpr (keep_null_values) {
              if ((*probe_radix_container.null_value_bitvector)[partition_offset]) {
                pos_list_build_side_local.emplace_back(NULL_ROW_ID);
                pos_list_probe_local.emplace_back(probe_column_element.row_id);
                // ignore found matches and continue with next probe item
                continue;
              }
            }

            // If NULL values are discarded, the matching probe_column_element pairs will be written to the result pos
            // lists.
            if (!multi_predicate_join_evaluator) {
              for (const auto& row_id : primary_predicate_matching_rows) {
                pos_list_build_side_local.emplace_back(row_id);
                pos_list_probe_local.emplace_back(probe_column_element.row_id);
              }
            } else {
              auto match_found = false;
              for (const auto& row_id : primary_predicate_matching_rows) {
                if (multi_predicate_join_evaluator->satisfies_all_predicates(row_id, probe_column_element.row_id)) {
                  pos_list_build_side_local.emplace_back(row_id);
                  pos_list_probe_local.emplace_back(probe_column_element.row_id);
                  match_found = true;
                }
              }

              // We have not found matching items for all predicates.
              if constexpr (keep_null_values) {
                if (!match_found) {
                  pos_list_build_side_local.emplace_back(NULL_ROW_ID);
                  pos_list_probe_local.emplace_back(probe_column_element.row_id);
                }
              }
            }

          } else {
            // We have not found matching items for the first predicate. Only continue for non-equi join modes.
            // We use constexpr to prune this conditional for the equi-join implementation.
            // Note, the outer relation (i.e., left relation for LEFT OUTER JOINs) is the probing
            // relation since the relations are swapped upfront.
            if constexpr (keep_null_values) {
              pos_list_build_side_local.emplace_back(NULL_ROW_ID);
              pos_list_probe_local.emplace_back(probe_column_element.row_id);
            }
          }
        }
      } else {
        // When there is no hash table, we might still need to handle the values of the probe side for LEFT
        // and RIGHT joins. We use constexpr to prune this conditional for the equi-join implementation.
        if constexpr (keep_null_values) {
          // We assume that the relations have been swapped previously, so that the outer relation is the probing
          // relation.
          // Since we did not find a hash table, we know that there is no match in the build column for this partition.
          // Hence we are going to write NULL values for each row.

          pos_list_build_side_local.reserve(partition_end - partition_begin);
          pos_list_probe_local.reserve(partition_end - partition_begin);

          for (size_t partition_offset = partition_begin; partition_offset < partition_end; ++partition_offset) {
            auto& row = partition[partition_offset];
            pos_list_build_side_local.emplace_back(NULL_ROW_ID);
            pos_list_probe_local.emplace_back(row.row_id);
          }
        }
      }

      pos_lists_build_side[current_partition_id] = std::move(pos_list_build_side_local);
      pos_lists_probe_side[current_partition_id] = std::move(pos_list_probe_local);
    }));
    jobs.back()->schedule();
  }

  CurrentScheduler::wait_for_tasks(jobs);
}

template <typename ProbeColumnType, typename HashedType, JoinMode mode>
void probe_semi_anti(const RadixContainer<ProbeColumnType>& radix_probe_column,
                     const std::vector<std::optional<HashTable<HashedType>>>& hash_tables,
                     std::vector<PosList>& pos_lists, const Table& build_table, const Table& probe_table,
                     const std::vector<OperatorJoinPredicate>& secondary_join_predicates) {
  std::vector<std::shared_ptr<AbstractTask>> jobs;
  jobs.reserve(radix_probe_column.partition_offsets.size());

  [[maybe_unused]] const auto* probe_column_null_values =
      radix_probe_column.null_value_bitvector ? radix_probe_column.null_value_bitvector.get() : nullptr;

  for (size_t current_partition_id = 0; current_partition_id < radix_probe_column.partition_offsets.size();
       ++current_partition_id) {
    const auto partition_begin =
        current_partition_id == 0 ? 0 : radix_probe_column.partition_offsets[current_partition_id - 1];
    const auto partition_end = radix_probe_column.partition_offsets[current_partition_id];  // make end non-inclusive

    // Skip empty partitions to avoid empty output chunks
    if (partition_begin == partition_end) {
      continue;
    }

    jobs.emplace_back(std::make_shared<JobTask>([&, partition_begin, partition_end, current_partition_id]() {
      // Get information from work queue
      auto& partition = static_cast<Partition<ProbeColumnType>&>(*radix_probe_column.elements);

      PosList pos_list_local;

      if (hash_tables[current_partition_id]) {
        // Valid hashtable found, so there is at least one match in this partition

        // Accessors are not thread-safe, so we create one evaluator per job
        MultiPredicateJoinEvaluator multi_predicate_join_evaluator(build_table, probe_table, mode,
                                                                   secondary_join_predicates);

        for (size_t partition_offset = partition_begin; partition_offset < partition_end; ++partition_offset) {
          auto& probe_column_element = partition[partition_offset];

          if constexpr (mode == JoinMode::Semi) {
            // NULLs on the probe side are never emitted
            if (probe_column_element.row_id.chunk_offset == INVALID_CHUNK_OFFSET) {
              continue;
            }
          } else if constexpr (mode == JoinMode::AntiNullAsFalse) {  // NOLINT - doesn't like else if constexpr
            // NULL values on the probe side always lead to the tuple being emitted for AntiNullAsFalse, irrespective
            // of secondary predicates (`NULL("as false") AND <anything>` is always false)
            if ((*probe_column_null_values)[partition_offset]) {
              pos_list_local.emplace_back(probe_column_element.row_id);
              continue;
            }
          } else if constexpr (mode == JoinMode::AntiNullAsTrue) {  // NOLINT - doesn't like else if constexpr
            if ((*probe_column_null_values)[partition_offset]) {
              // Primary predicate is TRUE, as long as we do not support secondary predicates with AntiNullAsTrue.
              // This means that the probe value never gets emitted
              continue;
            }
          }

          auto any_build_column_value_matches = false;
          const auto& hashtable = hash_tables[current_partition_id].value();
          const auto it = hashtable.find(static_cast<HashedType>(probe_column_element.value));

          if (it != hashtable.end()) {
            const auto& matching_rows = it->second;

            for (const auto& row_id : matching_rows) {
              if (multi_predicate_join_evaluator.satisfies_all_predicates(row_id, probe_column_element.row_id)) {
                any_build_column_value_matches = true;
                break;
              }
            }
          }

          if ((mode == JoinMode::Semi && any_build_column_value_matches) ||
              ((mode == JoinMode::AntiNullAsTrue || mode == JoinMode::AntiNullAsFalse) &&
               !any_build_column_value_matches)) {
            pos_list_local.emplace_back(probe_column_element.row_id);
          }
        }
      } else if constexpr (mode == JoinMode::AntiNullAsFalse) {  // NOLINT - doesn't like else if constexpr
        // no hashtable on other side, but we are in AntiNullAsFalse mode which means all tuples from the probing side
        // get emitted.
        pos_list_local.reserve(partition_end - partition_begin);
        for (size_t partition_offset = partition_begin; partition_offset < partition_end; ++partition_offset) {
          auto& probe_column_element = partition[partition_offset];
          pos_list_local.emplace_back(probe_column_element.row_id);
        }
      } else if constexpr (mode == JoinMode::AntiNullAsTrue) {  // NOLINT - doesn't like else if constexpr
        // no hashtable on other side, but we are in AntiNullAsTrue mode which means all tuples from the probing side
        // get emitted. That is, except NULL values, which only get emitted if the build table is empty.
        pos_list_local.reserve(partition_end - partition_begin);
        for (size_t partition_offset = partition_begin; partition_offset < partition_end; ++partition_offset) {
          auto& probe_column_element = partition[partition_offset];
          // A NULL on the probe side never gets emitted, except when the build table is empty.
          // This is because `NULL NOT IN <empty list>` is actually true
          if ((*probe_column_null_values)[partition_offset] && build_table.row_count() > 0) {
            continue;
          }
          pos_list_local.emplace_back(probe_column_element.row_id);
        }
      }

      pos_lists[current_partition_id] = std::move(pos_list_local);
    }));
    jobs.back()->schedule();
  }

  CurrentScheduler::wait_for_tasks(jobs);
}

using PosLists = std::vector<std::shared_ptr<const PosList>>;
using PosListsByChunk = std::vector<std::shared_ptr<PosLists>>;

/**
 * Returns a vector where each entry with index i references a PosLists object. The PosLists object
 * contains the position list of every segment/chunk in column i.
 * @param input_table
 */
// See usage in _on_execute() for doc.
inline PosListsByChunk setup_pos_lists_by_chunk(const std::shared_ptr<const Table>& input_table) {
  DebugAssert(input_table->type() == TableType::References, "Function only works for reference tables");

  std::map<PosLists, std::shared_ptr<PosLists>> shared_pos_lists_by_pos_lists;

  PosListsByChunk pos_lists_by_segment(input_table->column_count());
  auto pos_lists_by_segment_it = pos_lists_by_segment.begin();

  const auto& input_chunks = input_table->chunks();

  // For every column, for every chunk
  for (ColumnID column_id{0}; column_id < input_table->column_count(); ++column_id) {
    // Get all the input pos lists so that we only have to pointer cast the segments once
    auto pos_list_ptrs = std::make_shared<PosLists>(input_table->chunk_count());
    auto pos_lists_iter = pos_list_ptrs->begin();

    // Iterate over every chunk and add the chunks segment with column_id to pos_list_ptrs
    for (ChunkID chunk_id{0}; chunk_id < input_table->chunk_count(); ++chunk_id) {
      const auto& ref_segment_uncasted = input_chunks[chunk_id]->segments()[column_id];
      const auto ref_segment = std::static_pointer_cast<const ReferenceSegment>(ref_segment_uncasted);
      *pos_lists_iter = ref_segment->pos_list();
      ++pos_lists_iter;
    }

    // pos_list_ptrs contains all position lists of the reference segments for the column_id.
    auto iter = shared_pos_lists_by_pos_lists.emplace(*pos_list_ptrs, pos_list_ptrs).first;

    *pos_lists_by_segment_it = iter->second;
    ++pos_lists_by_segment_it;
  }

  return pos_lists_by_segment;
}

/**
 *
 * @param output_segments [in/out] Vector to which the newly created reference segments will be written.
 * @param input_table Table which all the position lists reference
 * @param input_pos_list_ptrs_sptrs_by_segments Contains all position lists to all columns of input table
 * @param pos_list contains the positions of rows to use from the input table
 */
inline void write_output_segments(Segments& output_segments, const std::shared_ptr<const Table>& input_table,
                                  const PosListsByChunk& input_pos_list_ptrs_sptrs_by_segments,
                                  std::shared_ptr<PosList> pos_list) {
  std::map<std::shared_ptr<PosLists>, std::shared_ptr<PosList>> output_pos_list_cache;

  // We might use this later, but want to have it outside of the for loop
  std::shared_ptr<Table> dummy_table;

  // Add segments from input table to output chunk
  // for every column for every row in pos_list: get corresponding PosList of input_pos_list_ptrs_sptrs_by_segments
  // and add it to new_pos_list which is added to output_segments
  for (ColumnID column_id{0}; column_id < input_table->column_count(); ++column_id) {
    if (input_table->type() == TableType::References) {
      if (input_table->chunk_count() > 0) {
        const auto& input_table_pos_lists = input_pos_list_ptrs_sptrs_by_segments[column_id];

        auto iter = output_pos_list_cache.find(input_table_pos_lists);
        if (iter == output_pos_list_cache.end()) {
          // Get the row ids that are referenced
          auto new_pos_list = std::make_shared<PosList>(pos_list->size());
          auto new_pos_list_iter = new_pos_list->begin();
          for (const auto& row : *pos_list) {
            if (row.chunk_offset == INVALID_CHUNK_OFFSET) {
              *new_pos_list_iter = row;
            } else {
              const auto& referenced_pos_list = *(*input_table_pos_lists)[row.chunk_id];
              *new_pos_list_iter = referenced_pos_list[row.chunk_offset];
            }
            ++new_pos_list_iter;
          }

          iter = output_pos_list_cache.emplace(input_table_pos_lists, new_pos_list).first;
        }

        auto reference_segment = std::static_pointer_cast<const ReferenceSegment>(
            input_table->get_chunk(ChunkID{0})->get_segment(column_id));
        output_segments.push_back(std::make_shared<ReferenceSegment>(
            reference_segment->referenced_table(), reference_segment->referenced_column_id(), iter->second));
      } else {
        // If there are no Chunks in the input_table, we can't deduce the Table that input_table is referencing to.
        // pos_list will contain only NULL_ROW_IDs anyway, so it doesn't matter which Table the ReferenceSegment that
        // we output is referencing. HACK, but works fine: we create a dummy table and let the ReferenceSegment ref
        // it.
        if (!dummy_table) dummy_table = Table::create_dummy_table(input_table->column_definitions());
        output_segments.push_back(std::make_shared<ReferenceSegment>(dummy_table, column_id, pos_list));
      }
    } else {
      output_segments.push_back(std::make_shared<ReferenceSegment>(input_table, column_id, pos_list));
    }
  }
}

}  // namespace opossum

ext: hpp

max_stars_repo_name: hyrise-mp/hyrise

max_issues_repo_name: hyrise-mp/hyrise

max_forks_repo_name: hyrise-mp/hyrise


text:
#include <rose.h>

// DQ (2/9/2010): Testing use of ROE to compile ROSE.
#ifndef USE_ROSE

#include <string>
#include <boost/algorithm/string/predicate.hpp>

#include "sageGeneric.h"

#include "RtedTransformation.h"

#include "RtedSymbols.h"
#include "DataStructures.h"

namespace SI = SageInterface;
namespace SB = SageBuilder;

/*********************************************************
 * Check if a function call is interesting, i.e. contains a
 * call to a function that we need to check the parameters of
 ********************************************************/
bool isStringModifyingFunctionCall(const std::string& name) {
   return (  name == "memcpy"
          || name == "memmove"
          || name == "strcpy"
          || name == "strncpy"
          || name == "strcat"
          || name == "strchr"
          || name == "strlen"     // \pp actually not string modifying...
                                  //     and I suspect that this gives the wrong
                                  //     size of the argument lists for
                                  //     the (commented out) call in
                                  //     RtedTransformation::insertFuncCall.
          || name == "strncat"
          || name == "strpbrk"
          || name == "strspn"
          || name == "strstr"
          );
}

/***************************************************************
 * This dimension is used to calculate additional parameters
 * necessary for the function call, e.g.
 * if dim = 1 then the parameter is followed by one more
 * element wich is its size
 **************************************************************/
static
size_t getDimensionForFuncCall(const std::string& name)
{
   int dim = 0;

   if (  name == "memcpy"
      || name == "memmove"
      || name == "strcpy"
      || name == "strncpy"
      || name == "strcat"
      || name == "strchr"
      || name == "strncat"
      || name == "strpbrk"
      || name == "strspn"
      || name == "strstr"
      || name == "fopen"
      )
   {
     dim = 2;
   }
   else if (name == "strlen")
   {
     dim = 1;
   }

   return dim;
}

bool isFileIOFunctionCall(const std::string& name) {
   return (  name == "fopen"
          || name == "fgetc"
          || name == "fputc"
          || name == "fclose"
          || name == "::std::fstream"
          );
}

bool isGlobalFunctionOnIgnoreList(const std::string& name) {
   return (  name == "calloc"
          || name == "free"
          || name == "malloc"
          || name == "printf"
          || name == "realloc"
          || name == "upc_all_alloc"
          || name == "upc_local_alloc"
          || name == "upc_alloc"
          || name == "upc_free"
          || name == "upc_global_alloc"
          || isLibFunctionRequiringArgCheck(name)
          );
}

bool isLibFunctionRequiringArgCheck(const std::string& name)
{
  return (  name == "upc_threadof"
         || name == "upc_addrfield"
         );
}

namespace
{
  struct IsIntType
  {
    const SgType* res;

    IsIntType()
    : res(NULL)
    {}

    template <class SageNode>
    void skip(const SageNode& n) { res = n.get_base_type(); }

    void found(const SgType& n) { res = &n; }

    void handle(const SgNode&)                { ROSE_ASSERT(false); }

    void handle(const SgType&)                { /* not an int */ }

    void handle(const SgTypeChar& n)          { found(n); }
    void handle(const SgTypeSignedChar& n)    { found(n); }
    void handle(const SgTypeUnsignedChar& n)  { found(n); }
    void handle(const SgTypeShort& n)         { found(n); }
    void handle(const SgTypeSignedShort& n)   { found(n); }
    void handle(const SgTypeUnsignedShort& n) { found(n); }
    void handle(const SgTypeInt& n)           { found(n); }
    void handle(const SgTypeSignedInt& n)     { found(n); }
    void handle(const SgTypeUnsignedInt& n)   { found(n); }
    void handle(const SgTypeLong& n)          { found(n); }
    void handle(const SgTypeSignedLong& n)    { found(n); }
    void handle(const SgTypeUnsignedLong& n)  { found(n); }

    void handle(const SgModifierType& n)      { skip(n); }
    void handle(const SgTypedefType& n)       { skip(n); }

    operator const SgType*() { return res; }
  };
};

static
bool isIntType(const SgType* n)
{
  const SgType* curr = NULL;
  const SgType* next = n;

  do
  {
    curr = next;
    next = sg::dispatch(IsIntType(), curr);
  } while (next && next != curr);

  return (next != NULL);
}

static
bool isFunctionCallReturningInt(SgExpression* exp)
{
  SgFunctionCallExp* funcall = isSgFunctionCallExp(exp);

  return (funcall && isIntType(funcall->get_type()));
}

SgFunctionCallExp* RtedTransformation::convertIntToString(SgExpression* i)
{
  ROSE_ASSERT(symbols.roseConvertIntToString);

  SgFunctionRefExp*  memRef_r = SB::buildFunctionRefExp(symbols.roseConvertIntToString);
  SgExprListExp*     arg_list = SB::buildExprListExp();

  SI::appendExpression(arg_list, i);
  return SB::buildFunctionCallExp(memRef_r, arg_list);
}

static
SgName mkRtedName()
{
  static size_t counter = 0;

  std::stringstream ss;

  ss << "tmp" << counter << "_rted";
  ++counter;

  return ss.str();
}

void RtedTransformation::insertFuncCall(RtedArguments& args)
{
  // fixed arguments
  // 1 = name
  // 2 = unparsedStmt for Error message
  // 3 = Left Hand side variable, if assignment
  // 4 = SourceInfo
  // 5 = number of the following arguments (\pp ??)

  SgStatement*      stmt = args.stmt;
  ROSE_ASSERT(stmt);

  SgScopeStatement* scope = stmt->get_scope();
  ROSE_ASSERT(scope);

  SgExpression*     callNameExp = SB::buildStringVal(args.f_name);
  SgExprListExp*    arg_list = SB::buildExprListExp();

  SI::appendExpression(arg_list, callNameExp);
  SI::appendExpression(arg_list, SB::buildStringVal(removeSpecialChar(stmt->unparseToString())));

  // this one is new, it indicates the variable on the left hand side of the statment,
  // if available
  ROSE_ASSERT(args.leftHandSideAssignmentExprStr);
  SI::appendExpression(arg_list, args.leftHandSideAssignmentExprStr);
  std::cerr << " ... Left hand side variable : " << args.leftHandSideAssignmentExprStr << std::endl;

  appendFileInfo(arg_list, stmt);

  // how many additional arguments does this function need?
  const size_t         dimFuncCall = getDimensionForFuncCall(args.f_name);

  // nr of args + 2 (for sepcialFunctions) + 6 =  args+6;
  SgExpressionPtrList& rose_args = args.arguments;
  const size_t         size = dimFuncCall + rose_args.size();
  SgIntVal*            sizeExp = SB::buildIntVal(size);

  SI::appendExpression(arg_list, sizeExp);

  // \pp modifying the dimension count looks strange (in fact when comparing
  //     isStringModifyingFunctionCall to getDimensionForFuncCall one
  //     notices that the only affected call function is strlen)!
  //     Since I do not understand the code below, the refactored/reformmated
  //     code below preserves the original behaviour.
  //     dimIsTwo is newly introduced and is true when the original code had
  //       dimFuncCall = 2.
  //     Subsequently, dimIsTwo replaces all occurances of (dimFuncCall == 2).
  // \pp why do we need dimIsTwo anyways, the arguments seem to be not used in the backend
  const bool                          dimIsTwo = (dimFuncCall == 2) || isStringModifyingFunctionCall(args.f_name);
  SgExprListExp*                      vararg_list = SB::buildExprListExp();

  // iterate over all arguments of the function call, e.g. strcpy(arg1, arg2);
  SgExpressionPtrList::const_iterator it = rose_args.begin();

  for (; it != rose_args.end(); ++it)
  {
     SgExpression* exp = SI::deepCopy(*it);

     // ************ unary operation *******************************
     if (isSgUnaryOp(exp))
     {
        // \pp why are unary operations skipped?
        // \pp \memleak as exp is no longer referenced
        exp = isSgUnaryOp(exp)->get_operand();
     }

     if (isSgVarRefExp(exp))
     {
        SgVarRefExp* var = isSgVarRefExp(exp);
        SgType*      vartype = var->get_type();

        // \pp \todo what about skipping modifier types?
        SgType*      base_type = skip_ArrPtrType(vartype);

        std::cerr << " isSgVarRefExp :: type : " << vartype->class_name() << std::endl;
        std::cerr << "     base_type: " << base_type->class_name() << std::endl;

        // --------- varRefExp is TypeChar -----------------------
        if (isSgTypeChar(vartype) || isSgTypeChar(base_type)) {
           std::string        name = var->get_symbol()->get_declaration()->get_name().str();
           SgInitializedName* initName = var->get_symbol()->get_declaration();
           ROSE_ASSERT(initName);

           SgType*            inittype = initName->get_typeptr();
           ROSE_ASSERT(inittype);

           SgArrayType*       array = isSgArrayType(inittype);

           if (array)
           {
              SgExpression * expr = array->get_index();

              std::cerr << " Found 00 - btype : " << array->get_base_type()->class_name() << "  index type: " << expr << std::endl;

              // this is the variable that we pass
              SI::appendExpression(vararg_list, var);
              if (dimIsTwo)
              {
                expr = ( expr == 0 ? SB::buildStringVal("00")
                                   : SB::buildStringVal(expr->unparseToString() + "\0")
                       );

                SI::appendExpression(vararg_list, expr);
              }

              std::cerr << ">>> Found variable : " << name << "  with size : " << expr->unparseToString() << "  and val : " << var << std::endl;
           }
           else if (isSgPointerType(inittype))
           {
              std::cerr << "Cant determine the size of the following object : " << var->class_name() << "  with base type : "
                        << base_type->class_name() << std::endl;

              // this is a pointer to char* but not an array allocation yet
              // We have to pass var so we can check whether the memories overlap!
              SI::appendExpression(vararg_list, var);

              if (dimIsTwo)
              {
                // if this is a pointer to e.g. malloc we cant statically determine the
                // size of the allocated address. We need to pass the variable
                // and check in the runtime system if the size for the variable is known!
                SgExpression* manglName = SB::buildStringVal(var->get_symbol()->get_declaration()->get_mangled_name().str());

                SI::appendExpression(vararg_list, manglName);
              }
           }
           else
           {
              std::cerr << "Cant determine the size of the following object : " << var->class_name() << "  with type : "
                        << inittype->class_name() << std::endl;
              std::cerr << "   and base type : " << base_type->class_name() << std::endl;

              // this is most likely a single char, the size of it is 1
              SgExpression* andSign = SB::buildAddressOfOp(var);
              SgExpression* charCast = SB::buildCastExp(andSign, SB::buildPointerType(SB::buildCharType()));

              SI::appendExpression(vararg_list, charCast);

              if (dimIsTwo)
              {
                SgExpression* numberToString = SB::buildStringVal("001");

                SI::appendExpression(vararg_list, numberToString);
              }
           }
        }
        // --------- varRefExp is PointerType -----------------------
        else if (isSgPointerType(vartype))
        {
           // handle pointers as parameters
           std::cerr << "RtedTransformation - unknown type : " << vartype->class_name() << std::endl;
           std::cerr << "   and base type : " << base_type->class_name() << "  var:" << var->unparseToString() << "   stmt:"
                     << stmt->unparseToString() << std::endl;

           if (var) {
              SgInitializedName* initName = var->get_symbol()->get_declaration();
              ROSE_ASSERT(initName);
              // if it is a initName in a parameterlist than we do not want to send it
              // as a mangled name because it will be something like: L22R__ARG1
              // instead we send it as a normal name that can be found on the stack
              SgExpression* manglName = SB::buildStringVal(initName->get_mangled_name().str());
              if (isSgFunctionParameterList(initName->get_parent())) {
                 //       cerr << "\n!!!!!!!!!!parent of initName = " << initName->get_parent()->class_name() << endl;
                 manglName = SB::buildStringVal(initName->get_name().str());
              }
              ROSE_ASSERT(manglName);
              SI::appendExpression(vararg_list, manglName);
           }
           //ROSE_ASSERT(false);
        }
        // --------- varRefExp is IntegerType -----------------------
        else
        {
           // handle integers as parameters
           if (!isSgTypeInt(vartype)) {
              std::cerr << "RtedTransformation - unknown type : " << vartype->class_name() << std::endl;
              ROSE_ASSERT(false);
           }

           ROSE_ASSERT(dimIsTwo);

           SI::appendExpression(vararg_list, convertIntToString(var));
           std::cerr << " Created Function call  convertToString" << std::endl;
        }
     }
     // --------- this is not a varRefExp ----
     else if (isSgStringVal(exp))
     {
        std::string theString = isSgStringVal(exp)->get_value();

        SI::appendExpression(vararg_list, SB::buildStringVal(theString + "\0"));

        if (dimIsTwo)
        {
           size_t        sizeString = theString.size() + 1; // add the '\0' to it
           std::string   sizeStringStr = RoseBin_support::ToString(sizeString);
           SgExpression* numberToString = SB::buildStringVal(sizeStringStr);

           SI::appendExpression(vararg_list, numberToString);
        }
     }
     else if (isFunctionCallReturningInt(exp))
     {
       SI::appendExpression(vararg_list, convertIntToString(exp));

       if (dimIsTwo)
       {
         SI::appendExpression(vararg_list, SB::buildStringVal("-1"));
       }
     }
     else
     {
        // default create a string
        std::string   theString = exp->unparseToString();
        SgExpression* stringExp = SB::buildStringVal(theString + "\0");

        SI::appendExpression(vararg_list, stringExp);

        if (dimIsTwo)
        {
          // \pp why is this needed
          size_t        sizeString = theString.size();
          std::string   sizeStringStr = RoseBin_support::ToString(sizeString);
          SgExpression* numberToString = SB::buildStringVal(sizeStringStr);

          SI::appendExpression(vararg_list, numberToString);
        }
     }
  }

  // insert temporary array variable to work around not-an lvalue array error
  SgName                  tmpName = mkRtedName();
  SgType*                 constCharPtr = roseConstCharPtrType();
  SgType*                 constCharPtrArr = SB::buildArrayType(constCharPtr, NULL);
  SgAggregateInitializer* tmparrInit = genAggregateInitializer(vararg_list, constCharPtr);
  SgVariableDeclaration*  tmparr = SB::buildVariableDeclaration( tmpName, constCharPtrArr, tmparrInit, scope );

  SageInterface::insertStatement(stmt, tmparr, true /* before */);

  SI::appendExpression( arg_list, SB::buildVarRefExp(tmparr) );

  ROSE_ASSERT(symbols.roseFunctionCall);
  insertCheck( ilBefore,
               stmt,
               symbols.roseFunctionCall,
               arg_list,
               "RS : Calling Function, parameters: (#args, function name, filename, linenr, linenrTransformed, error message, left hand var, other parameters)"
             );
}

static
SgType* select_type(SgExpression* expr)
{
  ROSE_ASSERT(expr);

  return expr->get_type();
}

template <class RoseFunctionRefExp>
static
void insertAssertFunctionSignature(RtedTransformation& rt, SgFunctionCallExp* fce, RoseFunctionRefExp* fnref)
{
  // we aren't able to check the function signature at compile time, so we
  // insert a check to do so at runtime

  using SageBuilder::buildExprListExp;
  using SageBuilder::buildStringVal;
  using SageBuilder::buildIntVal;

  ROSE_ASSERT(fce && fnref);

  SgExprListExp*       arg_list = buildExprListExp();
  SgExpressionPtrList& fnArgs = fce->get_args()->get_expressions();
  SgType*              fnReturn = isSgFunctionType( fnref->get_type() ) -> get_return_type();
  SgTypePtrList        fnArgTypes;

  std::transform(fnArgs.begin(), fnArgs.end(), std::back_inserter(fnArgTypes), select_type);

  SI::appendExpression( arg_list, buildStringVal( fnref->get_symbol()->get_name()) );
  rt.appendSignature( arg_list, fnReturn, fnArgTypes );

  SgStatement*         stmt = getSurroundingStatement(*fce);

  rt.appendFileInfo( arg_list, stmt->get_scope(), fce->get_file_info() );

  insertCheckOnStmtLevel(ilBefore, fce, rt.symbols.roseAssertFunctionSignature, arg_list);
}

void RtedTransformation::insertAssertFunctionSignature(SgFunctionCallExp* fncall)
{
   ROSE_ASSERT(fncall);

   SgFunctionRefExp* fn_ref = isSgFunctionRefExp(fncall -> get_function());
   if (fn_ref)
   {
     ::insertAssertFunctionSignature(*this, fncall, fn_ref);
     return;
   }

#if 0
   // \pp not needed according to the code (and comment) below
   SgDotExp* dotExp = isSgDotExp(fncall -> get_function());
   if (dotExp) {
      SgMemberFunctionRefExp* mfn_ref = isSgMemberFunctionRefExp(dotExp->get_rhs_operand());
      // tps: actually, we might not need this for C++
      return;
      // ::insertAssertFunctionSignature(*this, fncall, mfn_ref);
   }
#endif
}

void RtedTransformation::insertFreeCall(SgExpression* freeExp, AllocKind ak)
{
   ROSE_ASSERT( freeExp );

   // stmt wraps a call to free -- most likely an expression statement
   // alert the RTS before the call to detect errors such as double-free
   SgStatement*  stmt = getSurroundingStatement(*freeExp);
   SgExpression* address_expression = NULL;

   requiresParentIsBasicBlock(*stmt);

   if ((ak & akCxxHeap) == akCxxHeap)
   {
     SgDeleteExp* delExp = isSgDeleteExp(freeExp);
     ROSE_ASSERT(delExp);

     address_expression = delExp -> get_variable();
   }
   else
   {
     SgFunctionCallExp*   callExp = isSgFunctionCallExp(freeExp);
     ROSE_ASSERT(callExp);

     SgExpressionPtrList& args = callExp->get_args()->get_expressions();
     ROSE_ASSERT( args.size() == 1 );

     address_expression = args.front();
   }

   ROSE_ASSERT( stmt && address_expression );

   const bool     upcShared = ((ak & akUpcShared) == akUpcShared);
   SgExprListExp* arg_list = SB::buildExprListExp();

   SI::appendExpression( arg_list, mkAddress(address_expression, upcShared) );
   appendAllocKind( arg_list, ak );
   appendFileInfo( arg_list, stmt );

   // have to check validity of call to free before the call itself
   insertCheck(ilBefore, stmt, symbols.roseFreeMemory, arg_list);

   // upc_free requires the thread to be the only operator on the heap
   if (upcShared)
   {
     ROSE_ASSERT(symbols.roseUpcBeginExclusive && symbols.roseUpcEndExclusive);

     insertCheck(ilBefore, stmt, symbols.roseUpcBeginExclusive, SB::buildExprListExp());
     insertCheck(ilAfter,  stmt, symbols.roseUpcEndExclusive,   SB::buildExprListExp());
   }
}

void RtedTransformation::insertReallocateCall(SgFunctionCallExp* realloc_call)
{
   ROSE_ASSERT(realloc_call);

   // stmt wraps a call to realloc -- most likely an expression statement
   // alert the RTS before the call
   SgStatement*         stmt = getSurroundingStatement(*realloc_call);
   SgExpressionPtrList& args = realloc_call->get_args()->get_expressions();
   ROSE_ASSERT( args.size() == 2 );

   SgExpression* address_expression = args.front();
   SgExpression* size_expression = args.back();

   ROSE_ASSERT( address_expression && size_expression );

   // roseReallocate( ptr, source info );
   SgExprListExp* arg_list = SB::buildExprListExp();

   SI::appendExpression( arg_list, address_expression );
   SI::appendExpression( arg_list, size_expression );
   appendFileInfo( arg_list, stmt );

   // have to check validity of call to realloc before the call itself
   insertCheck(ilBefore, stmt, symbols.roseReallocateMemory, arg_list);
}

/***************************************************************
 * Get the variable on the left hand side of an assignment
 * starting on the right hand side below the assignment in the tree
 **************************************************************/
static
SgExpression*
getExpressionLeftOfAssignmentFromChildOnRight(SgFunctionCallExp* n) {
   SgExpression* expr = NULL;
   SgNode* tempNode = n;
   while (!isSgAssignOp(tempNode) && !isSgProject(tempNode)) {
      tempNode = tempNode->get_parent();
      ROSE_ASSERT(tempNode);
   }
   if (isSgAssignOp(tempNode)) {
      expr = isSgAssignOp(tempNode)->get_lhs_operand();
   }
   return expr;
}

static
std::string
getMangledNameOfExpression(SgExpression* expr)
{
  std::string manglName;
  // look for varRef in the expr and return its mangled name
  const SgNodePtrList& var_refs = NodeQuery::querySubTree(expr,V_SgVarRefExp);

  if (var_refs.size()==0)
  {
    // there should be at least one var ref
    std::cerr << " getMangledNameOfExpression: varRef on left hand side not found : " << expr->unparseToString() << std::endl;
  }
  else if (var_refs.size()==1)
  {
    // correct, found on var ref
    SgVarRefExp* varRef = isSgVarRefExp(*(var_refs.begin()));
    ROSE_ASSERT(varRef && varRef->get_symbol()->get_declaration());
    manglName = varRef->get_symbol()->get_declaration()->get_mangled_name();
    std::cerr << " getMangledNameOfExpression: found varRef: " << manglName << std::endl;
  }
  else if (var_refs.size()>1)
  {
    // error
    std::cerr << " getMangledNameOfExpression: Too many varRefs on left hand side : " << var_refs.size() << std::endl;
  }

  return manglName;
}

void RtedTransformation::addFileIOFunctionCall(SgVarRefExp* n, bool read)
{
   // treat the IO variable as a function call
   SgInitializedName*         name = n -> get_symbol() -> get_declaration();
   std::string                fname = name->get_type()->unparseToString();
   const char*                readstr = (read ? "r" : "w");

   SgExpression*              ex = SB::buildStringVal(readstr);
   SgStatement*               stmt = getSurroundingStatement(*n);
   std::vector<SgExpression*> args;

   args.push_back(ex);

   std::cerr << "@@@ my interesting func: " << fname << std::endl;
   function_call.push_back(RtedArguments(stmt, fname, readstr, "", "", n, NULL, NULL, NULL, args));
}

static
bool upcHeapManagementNeeded(const std::string& fnname)
{
  // in order to avoid deadlocks we need to exit UPC reader locks for
  //   the heap when a thread can get blocked.
  //   The general deadlock scenario is:
  //   Thread 1: holds the heap lock (e.g., as reader) and waits at some
  //             synchronized function (e.g., collective call, lock acquisition).
  //   Thread 2: calls upc_free just before calling the collective
  //             function.
  static const std::string upc_prefix = "upc_";
  static const std::string upc_collective_prefix = "upc_all_";

  if (!boost::starts_with(fnname, upc_prefix)) return false;

  return (  fnname == "upc_lock"
         || boost::starts_with(fnname, upc_collective_prefix)
         );
}

struct FunctionCallInfo
{
  RtedTransformation&  trans;
  bool                 memberCall;
  std::string          name;
  std::string          mangled_name;
  SgExpression*        target;

  explicit
  FunctionCallInfo(RtedTransformation& master)
  : trans(master), memberCall(false), name(), mangled_name(), target(NULL)
  {}

  void memberFunctionCall(SgMemberFunctionRefExp& mrefExp);
  void memberFunctionCall(SgBinaryOp& n);
  void ptrToMemberFunctionCall(SgBinaryOp& n);

  void handle(SgNode&) { ROSE_ASSERT(false); }

  void handle(SgExpression& n)
  {
    // this is a computed function (e.g., through a function pointer)
  }

  void handle(SgFunctionRefExp& n)
  {
    SgFunctionDeclaration* decl = n.getAssociatedFunctionDeclaration();

    name = decl->get_name();
    mangled_name = decl->get_mangled_name().str();
    target = &n;
  }

  void handle(SgArrowExp& n) { memberFunctionCall(n); }
  void handle(SgDotExp& n) { memberFunctionCall(n); }

  void handle(SgMemberFunctionRefExp& n)
  {
    // \pp member function calls can occur in context of dot and arrow
    //     expressions, but they can also be "naked" within a member body
    memberFunctionCall(n);
  }

  void handle(SgDotStarOp& n) { ptrToMemberFunctionCall(n); }
  void handle(SgArrowStarOp& n) { ptrToMemberFunctionCall(n); }
};

void FunctionCallInfo::memberFunctionCall(SgMemberFunctionRefExp& mrefExp)
{
   SgMemberFunctionDeclaration* mdecl = mrefExp.getAssociatedMemberFunctionDeclaration();

   memberCall = true;
   name = mdecl->get_name();
   mangled_name = mdecl->get_mangled_name().str();
   target = &mrefExp;
}

void FunctionCallInfo::memberFunctionCall(SgBinaryOp& n)
{
   SgMemberFunctionRefExp*      mrefExp = isSgMemberFunctionRefExp(n.get_rhs_operand());
   ROSE_ASSERT(mrefExp);

   memberFunctionCall(*mrefExp);
}

void FunctionCallInfo::ptrToMemberFunctionCall(SgBinaryOp& n)
{
   // \pp \todo this needs to be handled appropriately
   // e.g. : (testclassA.*testclassAPtr)()
   SgExpression* right = n.get_rhs_operand();
   // we want to make sure that the right hand side is not NULL
   // abort();
   // tps (09/24/2009) : this part is new and needs to be tested
   // testcode breaks at different location right now.
   SgVarRefExp* varrefRight = isSgVarRefExp(right);
   if (varrefRight)
   {
      trans.variable_access_varref.push_back(varrefRight);
   }
}

static
void store_if_needed(RtedTransformation::UpcBlockingOpsContainer& cont, SgFunctionCallExp& call)
{
  SgStatement* stmt = getSurroundingStatement(call);
  SgStatement* last = cont.empty() ? NULL : cont.back();

  // if the stmt is already going to be protected, then we can skip it
  if (last != stmt) cont.push_back( stmt );
}

/***************************************************************
 * Check if the current node is a "interesting" function call
 **************************************************************/
ReturnInfo::Kind RtedTransformation::visit_FunctionCall(SgFunctionCallExp& fcexp)
{
  SgExpression* const                 fun = fcexp.get_function();
  const FunctionCallInfo              callee = sg::dispatch(FunctionCallInfo(*this), fun);

  // \pp \todo \note The current implementation cannot handle function targets
  //                 determined at runtime (e.g., function pointer).
  if (!callee.target) return ReturnInfo::rtNone;

  // \pp \todo the rest of this function's logic could/should be pushed
  //           into FunctionCallInfo.

  // find out if this is a IO-CALL like : myfile << "something" ;
  // with fstream myfile;
  SgExprListExp* const                exprlist = isSgExprListExp(fcexp.get_args());
  ROSE_ASSERT(exprlist);

  SgExpressionPtrList&                args = exprlist->get_expressions();
  SgExpressionPtrList::const_iterator it = args.begin();

  for (; it!=args.end(); ++it)
  {
    SgExpression* fre = *it;
    SgVarRefExp*  varRef = isSgVarRefExp(fre);

    if (varRef && isFileIOVariable(fre->get_type()))
    {
      // indicate whether this is a stream operator
      addFileIOFunctionCall(varRef, callee.name == "operator>>");
    }
  }

  std::cerr << "\n@@@@ Found a function call: " << callee.name;
  std::cerr << "   : fcexp.get_function() : " << fun->class_name()
            << "   parent : " << fcexp.get_parent()->class_name() << "  : " << fcexp.get_parent()->unparseToString()
            << "\n   : type: : " << fun->get_type()->class_name()
            << "   : parent type: : " << isSgExpression(fun->get_parent())->get_type()->class_name()
            << "   unparse: " << fcexp.unparseToString()
            << std::endl;

  // \pp \todo this should ask whether the function was defined in global scope
  //           or within namespace std (instead of asking for membership).
  //           This is to avoid false positives on functions with the same name
  //           but declared within a namespace.
  bool handled = !callee.memberCall;

  if (handled)
  {
    if (isStringModifyingFunctionCall(callee.name) || isFileIOFunctionCall(callee.name))
    {
       // if this is a function call that has a variable on the left hand size,
       // then we want to push that variable first,
       // this is used, e.g. with  File* fp = fopen("file","r");
       // Therefore we need to go up and see if there is an AssignmentOperator
       // and get the var on the left side
       SgStatement*   stmt = getSurroundingStatement(fcexp);
       SgExpression*  varOnLeft = getExpressionLeftOfAssignmentFromChildOnRight(&fcexp);
       SgExpression*  varOnLeftStr=NULL;
       if (varOnLeft) {
          // need to get the mangled_name of the varRefExp on left hand side
          varOnLeftStr = SB::buildStringVal(getMangledNameOfExpression(varOnLeft));
       } else {
          varOnLeftStr = SB::buildStringVal("NoAssignmentVar");
       }

       ROSE_ASSERT(varOnLeftStr && stmt );
       function_call.push_back( RtedArguments( stmt,
                                               callee.name,
                                               callee.mangled_name,
                                               "",
                                               "",
                                               callee.target,
                                               varOnLeftStr,
                                               varOnLeft,
                                               exprlist,
                                               exprlist->get_expressions()
                                             )
                              );
    }
    else if ("free" == callee.name)
    {
       frees.push_back( Deallocations::value_type(&fcexp, akCHeap) );
    }
    else if( "realloc" == callee.name )
    {
       reallocs.push_back( &fcexp );
    }
    else if( "upc_free" == callee.name )
    {
       frees.push_back( Deallocations::value_type(&fcexp, akUpcShared) );
    }
    else if (upcHeapManagementNeeded(callee.name))
    {
      store_if_needed( upcBlockingOps, fcexp );
    }
    else if (!isGlobalFunctionOnIgnoreList(callee.name))
    {
      handled = false;
    }
  }

  ReturnInfo::Kind res = ReturnInfo::rtNone;

  if (!handled)
  {
     // if we're able to, use the function definition's body as the end of
     // scope (for line number complaints).  If not, the callsite is good too.
     SgStatement*           fncallStmt = getSurroundingStatement( fcexp );

     if (!fncallStmt->get_file_info()->isCompilerGenerated())
     {
       res = functionReturnType(fcexp.get_type());

       if ( getDefiningDeclaration( fcexp ) == NULL )
       {
          // FIXME 2: We may be adding a lot of unnecessary signature checks
          // If we don't have the definition, we must be doing separate
          // compilation.  We will then have to check the signature at runtime
          // since there's no guarantee our prototype, if any, is accurate.

          // \pp Why do not we verify the used and actual function signatures at
          //     start-up time?
          function_call_missing_def.push_back( &fcexp );
       }
     }
  }

  return res;
}

void RtedTransformation::transformUnusedPointerReturn(CallSiteContainer::value_type callexp)
{
  // call rted_function checking that the memory the pointer points to is
  // is reachable, after this pointer disappears
  SgExprListExp*       args = SB::buildExprListExp();
  SgExprStatement*     parent = isSgExprStatement(callexp->get_parent());

  if (!parent)
  {
    std::cerr << "---> " << typeid(*callexp->get_parent()).name() << std::endl;
    std::cerr << "---> " << callexp->get_parent()->unparseToString() << std::endl;
    ROSE_ASSERT(false);
  }

  SI::appendExpression(args, mkAddress(callexp, false /* non-shared */));

  SgFunctionRefExp*    funref = SB::buildFunctionRefExp(symbols.roseCheckForMemoryLeak);
  SgFunctionCallExp*   check = SB::buildFunctionCallExp(funref, args);

  check->set_parent(parent);
  parent->set_expression(check);
}

#endif

ext: cpp

max_stars_repo_name: maurizioabba/rose

max_issues_repo_name: sujankh/rose-matlab

max_forks_repo_name: sujankh/rose-matlab