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                                        *
 *          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-2013, Lehigh University, Yan Xu, and Ted Ralphs.       *
 *===========================================================================*/

#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)
        :
        AlpsKnowledgeBroker() 
        {    
            init();
            initializeSearch(argc, argv, model);
        }
    
    ~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);

    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_;
    }
 
    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

//#############################################################################