RequestExecutor.cpp

Go to the documentation of this file.

#include "RequestExecutor.h"
 
namespace   cmlabs{

// Request Executor
 
RequestExecutor::RequestExecutor(uint32 id, const char* name) {
    executorID = id;
    if (name && id)
        executorName = utils::StringFormat("%s:%u", name, id);
    else if (name)
        executorName = name;
    else if (id)
        executorName = utils::StringFormat("Executor:%u", id);
    else
        executorName = "";
 
    lastRefID = 0;
    sentCount = 0;
    repliedCount = 0;
    shortReceivedCount = 0;
    longReceivedCount = 0;
    heartbeatIntervalMS = 3000;
    lastHeartbeat = 0;
    shouldContinue = true;
    isRunning = false;
    channel = NULL;
    manager = new NetworkManager();
    longReqLimit = 0;
 
    if (!ThreadManager::CreateThread(RequestExecutorRun, this, threadID, 0)) {
        shouldContinue = false;
    }
}
 
RequestExecutor::~RequestExecutor() {
    stop();
    shutdownNetwork();
 
    std::map<uint64, RequestReply*>::iterator i, e;
    mutex.enter(200, __FUNCTION__);
    for (i = requestMap.begin(), e = requestMap.end(); i != e; i++)
        delete(i->second);
    requestMap.clear();
 
    longExecQ.clear();
    shortExecQ.clear();
 
    delete(manager);
    mutex.leave();
}
 
bool RequestExecutor::shutdownNetwork() {
    if (channel)
        channel->shutdown();
    return true;
}
 
bool RequestExecutor::addLongRequestName(const char* name) {
    if (channel)
        return false;
    longReqNames.push_back(name);
    return true;
}
 
bool RequestExecutor::addGateway(uint32 id, std::string addr, uint16 port, uint8 encryption) {
    RequestGatewayConnection con;
    con.clear();
    con.id = id;
    con.addr = addr;
    con.port = port;
    con.encryption = encryption;
    
    if (!mutex.enter(200, __FUNCTION__))
        return false;
 
    DataMessage* msgConnect = new DataMessage();
    msgConnect->setString("URI", "ExecutorConnect");
    msgConnect->setString("LongRequests", utils::TextVectorConcat(longReqNames, ",", false).c_str());
 
    if (!channel)
        channel = manager->addTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, 0, this, con.conID, con.location, 1000, (const char*)msgConnect->data, msgConnect->getSize());
    else
        con.conID = channel->addTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, con.location, 1000, (const char*)msgConnect->data, msgConnect->getSize());
 
    delete msgConnect;
 
    connections.push_back(con);
    mutex.leave();
    return true;
 
    //if (!con.conID) {
    //  mutex.leave();
    //  return false;
    //}
 
    
 
    //RequestGatewayConnection con;
    //con.clear();
    //con.id = id;
    //con.addr = addr;
    //con.port = port;
    //con.encryption = encryption;
 
    //if (!mutex.enter(200, __FUNCTION__))
    //  return false;
 
    //if (!channel)
    //  channel = manager->createTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, true, con.id, this, con.conID, con.location);
    //else
    //  con.conID = channel->createTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, true, con.location);
 
    //if (!con.conID) {
    //  mutex.leave();
    //  return false;
    //}

    //DataMessage* msg = new DataMessage();
    //msg->setString("URI", "ExecutorConnect");
    //msg->setString("LongRequests", utils::TextVectorConcat(longReqNames, ",", false).c_str());
    //if (!channel->sendMessage(msg, con.conID)) {
    //  delete(msg);
    //  mutex.leave();
    //  return false;
    //}
    //
    //delete(msg);
    //gateways[con.conID] = con;
    //mutex.leave();
    //return true;
}
 
 
bool RequestExecutor::receiveNetworkEvent(NetworkEvent* evt, NetworkChannel* channel, uint64 conid) {
    uint64 now = GetTimeNow();
    std::list<RequestGatewayConnection>::iterator i, e = connections.end();
    if (!mutex.enter(200, __FUNCTION__))
        return false;
    i = connections.begin();
    while (i != e) {
        if ((*i).conID == conid) {
            switch (evt->type) {
            case NETWORKEVENT_DISCONNECT:
            case NETWORKEVENT_DISCONNECT_RETRYING:
                LogPrint(0, LOG_SYSTEM, 0, "[%s] Disconnected from gateway %u on %s:%u", executorName.c_str(), conid, (*i).addr.c_str(), (*i).port);
                (*i).lastFailTime = now;
                //(*i).conID = 0;
                break;
            case NETWORKEVENT_CONNECT:
            case NETWORKEVENT_RECONNECT:
                LogPrint(0, LOG_SYSTEM, 0, "[%s] Connected to gateway %u on %s:%u", executorName.c_str(), conid, (*i).addr.c_str(), (*i).port);
                (*i).lastFailTime = 0;
                (*i).lastConTime = now;
                break;
            default:
                break;
            }
        }
        i++;
    }
    delete evt;
    mutex.leave();
    return true;
 
    //std::map<uint64,RequestGatewayConnection>::iterator i;
    //if (!mutex.enter(200, __FUNCTION__))
    //  return false;
    //if ( (i = gateways.find(conid)) != gateways.end()) {
    //  switch(evt->type) {
    //      case NETWORKEVENT_DISCONNECT:
    //          i->second.lastFailTime = now;
    //          break;
    //      case NETWORKEVENT_RECONNECT:
    //          i->second.lastFailTime = 0;
    //          break;
    //      default:
    //          break;
    //  }
    //}
    //mutex.leave();
    //return true;
}
 
bool RequestExecutor::setLongRequestLimit(uint32 limit) {
    longReqLimit = limit;
    return true;
}
 
bool RequestExecutor::receiveMessage(DataMessage* msg, NetworkChannel* channel, uint64 conid) {
    // Network-thread ingress: wrap each request in a RequestReply that records
    // the gateway's reference id (gatewayRef) and connection (origin), restamp
    // the message with a local execRef used as the requestMap key, then
    // classify short vs long — by size when longReqLimit is set, otherwise by
    // substring match against the registered long-request names — and queue it
    // for the worker threads. Note the queued pointer is the same object owned
    // by the RequestReply; it is freed when the reply is sent (replyToGateway).
    if (!msg) return false;
    RequestReply* reply;
    RequestGatewayConnection con;
    bool isLongReq;
    //std::map<uint64,RequestGatewayConnection>::iterator i;
    uint64 now = GetTimeNow();
    const char* req = msg->getString("URI");
    //LogPrint(0, LOG_SYSTEM, 0, "      * Executor received message from Gateway %llu (%llu)", conid, msg->getReference());
 
    if (!mutex.enter(1000, __FUNCTION__)) {
        LogPrint(0, LOG_SYSTEM, 0, "Error locking Executor mutex");
        delete(msg);
        return true;
    }
 
    std::list<RequestGatewayConnection>::iterator i, e = connections.end();
    i = connections.begin();
    while (i != e) {
        if ((*i).conID == conid) {
            reply = new RequestReply();
            reply->origin = conid;
            reply->execRef = ++lastRefID;
            reply->gatewayRef = msg->getReference();
            msg->setReference(reply->execRef);
            reply->giveRequestMessage(msg);
            reply->setStatus(QUEUED);
            requestMap[reply->execRef] = reply;
 
            if (longReqLimit)
                isLongReq = (msg->getSize() >= longReqLimit);
            else
                isLongReq = false;
            if (!isLongReq) {
                std::vector<std::string>::iterator il = longReqNames.begin(), el = longReqNames.end();
                while (il != el) {
                    if (utils::stristr(req, il->c_str())) {
                        isLongReq = true;
                        break;
                    }
                    il++;
                }
            }
            //isLongReq = ( std::find(longReqNames.begin(), longReqNames.end(), req) != longReqNames.end());
            if (isLongReq) {
                longExecQ.add(msg);
                //longExecQSemaphore.signal();
                longReceivedCount++;
            }
            else {
                shortExecQ.add(msg);
                //shortExecQSemaphore.signal();
                shortReceivedCount++;
            }
            sendStatusNow();
            mutex.leave();
            return true;
        }
        i++;
    }
 
    delete(msg);
    mutex.leave();
    return true;
}
 
DataMessage* RequestExecutor::waitForLongRequest(uint32 timeoutMS) {
    DataMessage* msg;
    if (msg = longExecQ.waitForAndTakeFirst(timeoutMS)) {
        msg->setSendTime(GetTimeNow());
        return msg;
    }
    else
        return NULL;
}
 
DataMessage* RequestExecutor::waitForShortRequest(uint32 timeoutMS) {
    DataMessage* msg;
    if (msg = shortExecQ.waitForAndTakeFirst(timeoutMS)) {
        msg->setSendTime(GetTimeNow());
        return msg;
    }
    else
        return NULL;
}
 
bool RequestExecutor::replyToQuery(DataMessage* msg) {
    if (!msg) return false;
    uint64 execRef = msg->getReference();
 
    if (!mutex.enter(200, __FUNCTION__))
        return false;
 
    std::map<uint64,RequestReply*>::iterator i = requestMap.find(execRef);
    //std::map<uint64,RequestGatewayConnection>::iterator con;
 
    if (i == requestMap.end()) {
        mutex.leave();
        return false;
    }
 
    std::list<RequestGatewayConnection>::iterator iCon, eCon = connections.end();
    iCon = connections.begin();
    while (iCon != eCon) {
        if ((*iCon).conID == i->second->origin) {
            replyQ.add(msg);
            repliedCount++;
            mutex.leave();
            return true;
        }
        iCon++;
    }
 
    mutex.leave();
    return false;
}
 
 
bool RequestExecutor::replyToGateway(DataMessage* msg) {
    if (!msg)
        return false;
    uint64 execRef = msg->getReference();
 
    if (!mutex.enter(200, __FUNCTION__))
        return false;
 
    std::map<uint64,RequestReply*>::iterator i = requestMap.find(execRef);
    std::map<uint64,RequestGatewayConnection>::iterator con;
 
    if (i == requestMap.end()) {
        LogPrint(0, LOG_SYSTEM, 0, "Executor couldn't find request to reply to (%llu)", execRef);
        mutex.leave();
        //delete(msg);
        return false;
    }
    RequestReply* reply = i->second;
 
    std::list<RequestGatewayConnection>::iterator iCon, eCon = connections.end();
    iCon = connections.begin();
    while (iCon != eCon) {
        if ((*iCon).conID == reply->origin) {
            msg->setReference(reply->gatewayRef);
            if (sendMessageToGateway(msg, (*iCon))) {
                // success
                //delete(msg);
                requestMap.erase(i);
                delete(reply);
                sentCount++;
                mutex.leave();
                return true;
            }
            else {
                LogPrint(0, LOG_SYSTEM, 0, "   $ Executor couldn't reply to Gateway %llu (%llu)", iCon->conID, reply->gatewayRef);
                requestMap.erase(i);
                delete(reply);
                mutex.leave();
                return false;
            }
        }
        iCon++;
    }
 
    LogPrint(0, LOG_SYSTEM, 0, "   $ Executor couldn't find any Gateways to reply to (%llu)", reply->gatewayRef);
    requestMap.erase(i);
    delete(reply);
    mutex.leave();
    return false;
}
 
 
bool RequestExecutor::run() {
    // Worker loop with two duties: (1) drain replyQ — replies queued by
    // replyToQuery() from application worker threads — and push each back to
    // its originating gateway via replyToGateway(), which also erases the
    // request bookkeeping (deleting the RequestReply and the original request
    // message); (2) emit an ExecutorStatus heartbeat every heartbeatIntervalMS
    // so gateways can track liveness and queue depth for load balancing.
    // Incoming requests do NOT pass through here — they arrive on the network
    // thread in receiveMessage() and go straight into shortExecQ/longExecQ.
    DataMessage* msg;
    isRunning = true;
 
    sendStatusNow();
 
    while (shouldContinue) {
        while (msg = replyQ.waitForAndTakeFirst(50)) {
            if (!mutex.enter(200, __FUNCTION__)) {
                delete(msg);
                break;
            }
            if (!replyToGateway(msg)) {
                // #######
            }
            delete(msg);
            mutex.leave();
        }
 
        // Send status heartbeat
        if (!lastHeartbeat || (GetTimeAgeMS(lastHeartbeat) > heartbeatIntervalMS)) {
            sendStatusNow();
        }
    }
    isRunning = false;
    return true;
}
 
 
bool RequestExecutor::sendMessageToGateway(DataMessage* msg, RequestGatewayConnection& con) {
 
    uint64 now = GetTimeNow();
 
    // if (!channel || !channel->isConnected(con.conID))
    if (!channel)
        return false;
 
    // LogPrint(0, LOG_SYSTEM, 0, "Executor sending message to gateway...");
 
    //if (!con.conID || !channel) {
    //  if (!channel)
    //      channel = manager->createTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, true, con.id, this, con.conID, con.location);
    //  else
    //      con.conID = channel->createTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, true, con.location);
    //  if (!channel || !con.conID) {
    //      if (!con.lastFailTime)
    //          LogPrint(0, LOG_SYSTEM, 0, "[%s] Unable to connect to gateway %u on %s:%u, will retry later", executorName.c_str(), con.id, con.addr.c_str(), con.port);
    //      con.lastFailTime = now;
    //      return false;
    //  }
    //  DataMessage* msgConnect = new DataMessage();
    //  msgConnect->setString("URI", "ExecutorConnect");
    //  msgConnect->setString("LongRequests", utils::TextVectorConcat(longReqNames, ",", false).c_str());
    //  if (!channel->sendMessage(msgConnect, con.conID)) {
    //      LogPrint(0, LOG_SYSTEM, 0, "[%s] Unable to communicate with gateway %u on %s:%u, will retry later", executorName.c_str(), con.id, con.addr.c_str(), con.port);
    //      con.lastFailTime = now;
    //      // clearly not working, disconnect and try again later
    //      channel->endConnection(con.conID);
    //      con.conID = 0;
    //      delete msgConnect;
    //      return false;
    //  }
    //  delete msgConnect;
    //  if (!con.lastConTime)
    //      LogPrint(0, LOG_SYSTEM, 0, "[%s] Connected to gateway %u on %s:%u", executorName.c_str(), con.id, con.addr.c_str(), con.port);
    //  else
    //      LogPrint(0, LOG_SYSTEM, 0, "[%s] Reconnected to gateway %u on %s:%u", executorName.c_str(), con.id, con.addr.c_str(), con.port);
    //  con.lastConTime = now;
    //}
    if (!channel->sendMessage(msg, con.conID)) {
        if (!con.lastFailTime)
            LogPrint(0, LOG_SYSTEM, 0, "[%s] Unable to send data to gateway %u on %s:%u, disconnecting...", executorName.c_str(), con.id, con.addr.c_str(), con.port);
        con.lastFailTime = GetTimeNow();
        // clearly not working, disconnect and try again later
        //channel->endConnection(con.conID);
        //con.conID = 0;
        return false;
    }
    //if (msg->getReference())
    //  LogPrint(0, LOG_SYSTEM, 0, "      + Executor replying to Gateway %u (%llu)", con.id, msg->getReference());
    //else
    //  LogPrint(0, LOG_SYSTEM, 0, "      + Executor sending Heartbeat to Gateway %u", con.id);
    return true;
}
 
uint32 RequestExecutor::sendStatusNow() {
    // Heartbeat: broadcasts an ExecutorStatus message carrying the current
    // short/long queue depths to every configured gateway. The gateway feeds
    // these into its RequestConnection records for load balancing, and treats
    // a missing heartbeat (executorHeartbeatTimeout) as executor death.
 
    if (!conMutex.enter(1000, __FUNCTION__)) {
        LogPrint(0, LOG_SYSTEM, 0, "Mutex error sending heartbeat from Executor %u...", executorID);
        return 0;
    }
    if (!connections.size()) {
        //LogPrint(0, LOG_SYSTEM, 0, "No connections sending heartbeat from Executor %u...", executorID);
        conMutex.leave();
        return 0;
    }
 
    DataMessage* msg = new DataMessage();
    msg->setString("URI", "ExecutorStatus");
    msg->setInt("ShortQSize", shortExecQ.getCount());
    msg->setInt("LongQSize", longExecQ.getCount());
 
    uint32 success = 0, failed = 0, notConnected = 0;
 
    std::list<RequestGatewayConnection>::iterator iCon, eCon = connections.end();
    iCon = connections.begin();
    while (iCon != eCon) {
        if (sendMessageToGateway(msg, (*iCon)))
            success++;
        else
            failed++;
        iCon++;
    }
    conMutex.leave();
 
    delete(msg);
    lastHeartbeat = GetTimeNow();
    if (failed)
        LogPrint(0, LOG_SYSTEM, 2, "Failed sending heartbeat from Executor %u to %u gateways (%u succeeded)...", executorID, failed, success);
//  else
//      LogPrint(0, LOG_SYSTEM, 0, "Sent heartbeat from Executor %u to %u gateways...", executorID, success);
    return success;
}
 
THREAD_RET THREAD_FUNCTION_CALL RequestExecutorRun(THREAD_ARG arg) {
    if (arg == NULL) thread_ret_val(1);
    thread_ret_val((int)(((RequestExecutor*)arg)->run() ? 0 : 1));
}
 
 
 
 
 
TestRequestExecutor::TestRequestExecutor(uint32 id, uint32 processTimeMS) : RequestExecutor(id) {
    this->id = id;
    this->processTimeMS = processTimeMS;
    shouldContinue = true;
    uint32 myThreadID;
    ThreadManager::CreateThread(ShortExecutorRun, this, myThreadID, 0);
    ThreadManager::CreateThread(LongExecutorRun, this, myThreadID, 0);
}
 
 
bool TestRequestExecutor::shortRequestRun() {
    uint32 count = 0;
    DataMessage* replyMsg, *msg;
    while (shouldContinue) {
        if (msg = waitForShortRequest(20)) {
            replyMsg = new DataMessage();
            replyMsg->setReference(msg->getReference());
            replyToQuery(replyMsg);
        }
    }
    return true;
}
 
bool TestRequestExecutor::longRequestRun() {
    uint32 count = 0;
    DataMessage* replyMsg, *msg;
    std::list<DataMessage*> storedMessages;
    std::list<DataMessage*>::iterator i, e = storedMessages.end();
 
    while (shouldContinue) {
        if (msg = waitForLongRequest(20)) {
            if (!processTimeMS) {
                replyMsg = new DataMessage();
                replyMsg->setReference(msg->getReference());
                replyToQuery(replyMsg);
                //printf("--- Replying immediately ---\n");
            }
            else {
                msg->setRecvTime(GetTimeNow());
                storedMessages.push_back(msg);
            }
        }
 
        i = storedMessages.begin();
        while (i != e) {
            if ((msg = (*i)) && (GetTimeAgeMS(msg->getRecvTime()) >= (int32)processTimeMS)) {
                i = storedMessages.erase(i);
                replyMsg = new DataMessage();
                replyMsg->setReference(msg->getReference());
                replyToQuery(replyMsg);
                //printf("--- Replying delayed ---\n");
            }
            else
                i++;
        }
    }
    return true;
}
 
THREAD_RET THREAD_FUNCTION_CALL ShortExecutorRun(THREAD_ARG arg) {
    if (arg == NULL) thread_ret_val(1);
    thread_ret_val((int)(((TestRequestExecutor*)arg)->shortRequestRun() ? 0 : 1));
}
 
THREAD_RET THREAD_FUNCTION_CALL LongExecutorRun(THREAD_ARG arg) {
    if (arg == NULL) thread_ret_val(1);
    thread_ret_val((int)(((TestRequestExecutor*)arg)->longRequestRun() ? 0 : 1));
}
 
}