AlpsKnowledgeBrokerMPI.h

Go to the documentation of this file.

/*===========================================================================*
 * This file is part of the Abstract Library for Parallel Search (ALPS).     *
 *                                                                           *
 * ALPS is distributed under the Eclipse Public License as part of the       *
 * COIN-OR repository (http://www.coin-or.org).                              *
 *                                                                           *
 * Authors:                                                                  *
 *                                                                           *
 *          Yan Xu, Lehigh University                                        *
 *          Aykut Bulut, Lehigh University                                   *
 *          Ted Ralphs, Lehigh University                                    *
 *                                                                           *
 * Conceptual Design:                                                        *
 *                                                                           *
 *          Yan Xu, Lehigh University                                        *
 *          Ted Ralphs, Lehigh University                                    *
 *          Laszlo Ladanyi, IBM T.J. Watson Research Center                  *
 *          Matthew Saltzman, Clemson University                             *
 *                                                                           *
 *                                                                           *
 * Copyright (C) 2001-2023, Lehigh University, Yan Xu, Aykut Bulut, and      *
 *                          Ted Ralphs.                                      *
 * All Rights Reserved.                                                      *
 *===========================================================================*/
 
 
#ifndef AlpsKnowledgeBrokerMPI_h_
#define AlpsKnowledgeBrokerMPI_h_
 
#include <cmath>
#include <iosfwd>
 
// #undef SEEK_SET
// #undef SEEK_END
// #undef SEEK_CUR
#include <mpi.h>
 
#include "AlpsEnumProcessT.h"
#include "AlpsKnowledge.h"
#include "AlpsKnowledgeBroker.h"
#include "AlpsParams.h"
 
//#############################################################################
 
class AlpsKnowledgeBrokerMPI : public AlpsKnowledgeBroker {
 
 private:
    AlpsKnowledgeBrokerMPI(const AlpsKnowledgeBrokerMPI&);
    AlpsKnowledgeBrokerMPI& operator=(const AlpsKnowledgeBrokerMPI&);
 
 protected:
 
    int processNum_;
 
    int hubNum_;
 
    int globalRank_;
 
    MPI_Comm clusterComm_;
 
    MPI_Comm hubComm_;
 
    MPI_Group hubGroup_;
 
    int clusterSize_;
 
    int userClusterSize_;
 
    int clusterRank_;
 
    int* hubRanks_;
 
    int myHubRank_;
 
    int masterRank_;
 
    AlpsProcessType processType_;
 
    AlpsProcessType* processTypeList_;
 
    bool hubWork_;
 
    MPI_Request subTreeRequest_;
 
    MPI_Request solRequestL_;
    MPI_Request solRequestR_;
 
    MPI_Request modelKnowRequestL_;
    MPI_Request modelKnowRequestR_;
 
    MPI_Request forwardRequestL_;
    MPI_Request forwardRequestR_;
 
    double incumbentValue_;
 
    int incumbentID_;
 
    bool updateIncumbent_;
 
    double workQuality_;
 
    double clusterWorkQuality_;
 
    double systemWorkQuality_;
 
    double* hubWorkQualities_;
 
    double* workerWorkQualities_;
 
    double workQuantity_;
 
    double clusterWorkQuantity_;
 
    double systemWorkQuantity_;
 
    double systemWorkQuantityForce_;
 
    double* hubWorkQuantities_;
 
    double* workerWorkQuantities_;
 
    bool* workerReported_;
 
    bool* hubReported_;
 
    bool allHubReported_;
 
    int masterDoBalance_;
 
    int hubDoBalance_;
 
    int* workerNodeProcesseds_;
 
    int clusterNodeProcessed_;
 
    int* hubNodeProcesseds_;
 
    int sendCount_;
 
    int recvCount_;
 
    int clusterSendCount_;
 
    int clusterRecvCount_;
 
    int systemSendCount_;
 
    int systemRecvCount_;
 
    int masterIndexBatch_;
 
    AlpsTimer masterTimer_;
 
    AlpsTimer hubTimer_;
 
    AlpsTimer workerTimer_;
 
    double rampUpTime_;
 
    double rampDownTime_;
 
    double idleTime_;
 
    double msgTime_;
 
    AlpsPsStats psStats_;
 
    bool forceTerminate_;
 
    bool blockTermCheck_;
 
    bool blockHubReport_;
 
    bool blockWorkerReport_;
 
    bool blockAskForWork_;
 
    char *attachBuffer_;
 
    char *largeBuffer_;
 
    char *largeBuffer2_;
 
    char *smallBuffer_;
 
    double masterBalancePeriod_;
 
    double hubReportPeriod_;
 
    int modelGenID_;
 
    int modelGenPos_;
 
    AlpsSubTree* rampUpSubTree_;
 
    int unitWorkNodes_;
 
    int haltSearch_;
 
 protected:
 
    void init();
 
    void masterMain(AlpsTreeNode* root);
 
    void hubMain();
 
    void workerMain();
 
    // The same subtree will be explored next time if it still have
    // unexplored nodes.
    AlpsReturnStatus doOneUnitWork(int unitWork,
                                   double unitTime,
                                   AlpsExitStatus & exitStatus,
                                   int & numNodesProcessed,
                                   int & numNodesBranched,
                                   int & numNodesDiscarded,
                                   int & numNodesPartial,
                                   int & depth,
                                   bool & betterSolution);
 
    void processMessages(char *&buffer,
                         MPI_Status &status,
                         MPI_Request &request);
 
    void rootInitMaster(AlpsTreeNode* root);
    void rootInitHub();
    void rootInitWorker();
 
    void spiralMaster(AlpsTreeNode* root);
    void spiralHub();
    void spiralWorker();
 
    //------------------------------------------------------
 
    void masterAskHubDonate(int donorID,
                            int receiverID,
                            double receiverWorkload);
 
    void hubAskWorkerDonate(int donorID,
                            int receiverID,
                            double receiverWorkload);
 
    void updateWorkloadInfo();
 
    virtual int getNumNodeLeftSystem()
    { return static_cast<int>(systemWorkQuantity_); }
 
    void donateWork(char*& buf,
                    int tag,
                    MPI_Status* status,
                    int recvID = -1,
                    double recvWL = 0.0);
 
    void hubAllocateDonation(char*& buf, MPI_Status* status);
 
    void hubBalanceWorkers();
 
    void hubSatisfyWorkerRequest(char*& buf, MPI_Status* status);
 
    // NOTE: comm is hubComm_ or MPI_COMM_WORLD.
    void hubReportStatus(int tag, MPI_Comm comm);
 
    // NOTE: comm is clusterComm_ or MPI_COMM_WORLD.
    void hubUpdateCluStatus(char*& buf, MPI_Status* status, MPI_Comm comm);
 
    void hubsShareWork(char*& buf, MPI_Status* status);
 
    void masterBalanceHubs();
 
    // NOTE: comm is hubComm or MPI_COMM_WORLD.
    void masterUpdateSysStatus(char*& buf, MPI_Status* status, MPI_Comm comm);
 
    void refreshSysStatus();
 
    void refreshClusterStatus();
 
    // NOTE: comm is clusterComm_ or MPI_COMM_WORLD.
    void workerReportStatus(int tag, MPI_Comm comm);
 
    //------------------------------------------------------
 
    void workerAskIndices();
 
    void workerRecvIndices(char *&bufLarge);
 
    void masterSendIndices(char *&bufLarge);
 
    //------------------------------------------------------
 
    void broadcastModel(const int id, const int source);
 
    void sendIncumbent();
 
    bool unpackSetIncumbent(char*& buf, MPI_Status* status);
 
    void collectBestSolution(int destination);
 
    void tellMasterRecv();
 
    // Not used
    void tellHubRecv();
 
    void packEncoded(AlpsEncoded* enc,
                     char*& buf,
                     int& size,
                     int& position,
                     MPI_Comm comm);
 
    AlpsEncoded* unpackEncoded(char*& buf,
                               int& position,
                               MPI_Comm comm,
                               int size = -1);
 
    // NOTE: comm is hubComm_ or clusterComm_
    void receiveSizeBuf(char*& buf,
                        int sender,
                        int tag,
                        MPI_Comm comm,
                        MPI_Status* status);
 
    // NOTE: comm is hubComm_ or clusterComm_
    void receiveRampUpNode(int sender,
                           MPI_Comm comm,
                           MPI_Status* status);
 
    void receiveSubTree(char*& buf, int sender, MPI_Status* status);
 
    // NOTE: comm is hubComm_ or clusterComm_.
    void sendSizeBuf(char*& buf,
                     int size,
                     int position,
                     const int target,
                     const int tag,
                     MPI_Comm comm);
 
    // NOTE: comm is hubComm_ or clusterComm_.
    void sendRampUpNode(const int target, MPI_Comm comm);
 
    void sendNodeModelGen(int receiver, int doUnitWork);
 
    bool sendSubTree(const int target, AlpsSubTree*& st, int tag);
 
    // NOTE: comm is hubComm_ or clusterComm_.
    void sendFinishInit(const int target, MPI_Comm comm);
 
    void deleteSubTrees();
 
 
    void forwardModelKnowledge();
 
    // NOTE: comm is hubComm_ or MPI_COMM_WORLD.
    void sendModelKnowledge(MPI_Comm comm, int receiver=-1);
 
    // NOTE: comm is hubComm_ or MPI_COMM_WORLD.
    void receiveModelKnowledge(MPI_Comm comm);
 
    void incSendCount(const char* how, int s = 1);
    void decSendCount(const char* how, int s = 1);
    void incRecvCount(const char* how, int s = 1);
    void decRecvCount(const char* how, int s = 1);
 
    void masterForceHubTerm();
 
    void hubForceWorkerTerm();
 
    void changeWorkingSubTree(double & changeWorkThreshold);
 
    void sendErrorCodeToMaster(int errorCode);
 
    void recvErrorCode(char *& bufLarge);
 
    void spiralRecvProcessNode();
 
    void spiralDonateNode();
 
 public:
 
    AlpsKnowledgeBrokerMPI()
        :
        AlpsKnowledgeBroker()
        {
            init();
        }
 
    AlpsKnowledgeBrokerMPI(int argc, 
                           char* argv[], 
                           AlpsModel& model,
                           bool showBanner = true)
        :
        AlpsKnowledgeBroker() 
        {    
            init();
            initializeSearch(argc, argv, model, showBanner);
        }
    
    ~AlpsKnowledgeBrokerMPI();
 
    virtual int getProcRank() const { return globalRank_; }
 
    virtual int getMasterRank() const { return masterRank_; }
 
    virtual AlpsProcessType getProcType() const { return processType_; }
 
    void initializeSearch(int argc,
                          char* argv[],
                          AlpsModel& model,
                          bool showBanner = true);
 
    void search(AlpsModel *model);
 
    void rootSearch(AlpsTreeNode* root);
 
    virtual double getIncumbentValue() const {
        double bestObj = ALPS_OBJ_MAX;
        if (AlpsKnowledgeBroker::hasKnowledge(AlpsKnowledgeTypeSolution)) {
            bestObj = getBestKnowledge(AlpsKnowledgeTypeSolution).second;
            if (incumbentValue_ > bestObj) {
                return bestObj;
            }
        }
        return incumbentValue_;
    }
 
  //todo(aykut) why not this returns to incumbentValue_
    virtual double getBestQuality() const {
        if (globalRank_ == masterRank_) {
            if (getNumKnowledges(AlpsKnowledgeTypeSolution) > 0) {
                return getBestKnowledge(AlpsKnowledgeTypeSolution).second;
            }
            else {
                return ALPS_OBJ_MAX;
            }
        }
        else {
            return ALPS_OBJ_MAX;
        }
    }
 
    virtual double getBestEstimateQuality() { return systemWorkQuality_; }
 
    virtual void printBestSolution(char* outputFile = 0) const;
 
    virtual void searchLog();
 
    //------------------------------------------------------
 
    void sendKnowledge(AlpsKnowledgeType type,
                       int sender,
                       int receiver,
                       char *& msgBuffer,
                       int msgSize,
                       int msgTag,
                       MPI_Comm comm,
                       bool blocking);
 
    void receiveKnowledge(AlpsKnowledgeType type,
                          int sender,
                          int receiver,
                          char *& msgBuffer,
                          int msgSize,
                          int msgTag,
                          MPI_Comm comm,
                          MPI_Status* status,
                          bool blocking);
 
    void requestKnowledge(AlpsKnowledgeType type,
                          int sender,
                          int receiver,
                          char *& msgBuffer,
                          int msgSize,
                          int msgTag,
                          MPI_Comm comm,
                          bool blocking);
 
};
#endif
 
//#############################################################################