QuerySchedulerServer.h

Go to the documentation of this file.

/*****************************************************************************
 *                                                                           *
 *  Copyright 2018 Rice University                                           *
 *                                                                           *
 *  Licensed under the Apache License, Version 2.0 (the "License");          *
 *  you may not use this file except in compliance with the License.         *
 *  You may obtain a copy of the License at                                  *
 *                                                                           *
 *      http://www.apache.org/licenses/LICENSE-2.0                           *
 *                                                                           *
 *  Unless required by applicable law or agreed to in writing, software      *
 *  distributed under the License is distributed on an "AS IS" BASIS,        *
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. *
 *  See the License for the specific language governing permissions and      *
 *  limitations under the License.                                           *
 *                                                                           *
 *****************************************************************************/
#ifndef QUERY_SCHEDULER_SERVER_H
#define QUERY_SCHEDULER_SERVER_H


#include "ServerFunctionality.h"
#include "ResourceInfo.h"
#include "StandardResourceInfo.h"
#include "Handle.h"
#include "PDBVector.h"
#include "QueryBase.h"
#include "ResourceInfo.h"
#include "SimpleSingleTableQueryProcessor.h"
#include "PDBLogger.h"
#include "TupleSetJobStage.h"
#include "AggregationJobStage.h"
#include "BroadcastJoinBuildHTJobStage.h"
#include "HashPartitionedJoinBuildHTJobStage.h"
#include "SequenceID.h"
#include "PhysicalOptimizer.h"
#include "ShuffleInfo.h"
#include "DistributedStorageManagerClient.h"
#include "StatisticsDB.h"
#include "RegisterReplica.h"
#include <vector>
#include <ExecuteComputation.h>

namespace pdb {

class QuerySchedulerServer : public ServerFunctionality {

public:

    QuerySchedulerServer(PDBLoggerPtr logger,
                         ConfigurationPtr conf,
                         std::shared_ptr<StatisticsDB> statisticsDB,
                         bool pseudoClusterMode = false,
                         double partitionToCoreRatio = 0.75);

    QuerySchedulerServer(int port,
                         PDBLoggerPtr logger,
                         ConfigurationPtr conf,
                         std::shared_ptr<StatisticsDB> statisticsDB,
                         bool pseudoClusterMode = false,
                         double partitionToCoreRatio = 0.75);

    ~QuerySchedulerServer();

    void registerHandlers(PDBServer& forMe) override;

    void collectStats();

    void cleanup() override;

    StatisticsPtr getStats();

    std::string getNextJobId() {
        time_t currentTime = time(nullptr);
        struct tm* local = localtime(&currentTime);
        this->jobId = "Job-" + std::to_string(local->tm_year + 1900) + "_" +
            std::to_string(local->tm_mon + 1) + "_" + std::to_string(local->tm_mday) + "_" +
            std::to_string(local->tm_hour) + "_" + std::to_string(local->tm_min) + "_" +
            std::to_string(local->tm_sec) + "_" + std::to_string(seqId.getNextSequenceID());
        return this->jobId;
    }

protected:

    void initialize();

    void initializeForServerMode();

    void initializeForPseudoClusterMode();

    void scheduleStages(std::vector<Handle<AbstractJobStage>>& stagesToSchedule,
                        std::shared_ptr<ShuffleInfo> shuffleInfo);


    void prepareAndScheduleStage(Handle<AbstractJobStage> &stage,
                                 unsigned long node,
                                 int &counter,
                                 PDBBuzzerPtr &callerBuzzer);

    template<typename T>
    bool scheduleStage(unsigned long node,
                       Handle<T>& stage,
                       PDBCommunicatorPtr communicator);

    PDBCommunicatorPtr getCommunicatorToNode(int port,
                                             std::string &ip);

    Handle <TupleSetJobStage> getStageToSend(unsigned long index,
                                             Handle <TupleSetJobStage> &stage);

    Handle <AggregationJobStage> getStageToSend(unsigned long index,
                                                Handle <AggregationJobStage> &stage);

    Handle <BroadcastJoinBuildHTJobStage> getStageToSend(unsigned long index,
                                                         Handle <BroadcastJoinBuildHTJobStage> &stage);

    Handle <HashPartitionedJoinBuildHTJobStage> getStageToSend(unsigned long index,
                                                               Handle <HashPartitionedJoinBuildHTJobStage> &stage);

    void collectStatsForNode(int node,
                             int &counter,
                             PDBBuzzerPtr &callerBuzzer);

    void updateStats(Handle<SetIdentifier> setToUpdateStats);

    pair<bool, basic_string<char>> executeComputation(Handle<ExecuteComputation> &request,
                                                      PDBCommunicatorPtr &sendUsingMe);

    pair<bool, basic_string<char>> registerReplica(Handle<RegisterReplica> &request,
                                                                     PDBCommunicatorPtr &sendUsingMe);


    void extractPipelineStages(int &jobStageId,
                               vector<Handle<AbstractJobStage>> &jobStages,
                               vector<Handle<SetIdentifier>> &intermediateSets);

    void createIntermediateSets(DistributedStorageManagerClient &dsmClient,
                                vector<Handle<SetIdentifier>> &intermediateSets);

    void removeIntermediateSets(DistributedStorageManagerClient &dsmClient);

    void removeUnusedIntermediateSets(DistributedStorageManagerClient &dsmClient,
                                      vector<Handle<SetIdentifier>> &intermediateSets);

    void requestStatistics(PDBCommunicatorPtr &communicator, bool &success, string &errMsg) const;


    std::vector<StandardResourceInfoPtr>* standardResources;

    int port;

    std::vector<Handle<SetIdentifier>> interGlobalSets;

    PDBLoggerPtr logger;

    ConfigurationPtr conf;

    std::shared_ptr<StatisticsDB> statisticsDB;

    bool pseudoClusterMode;

    pthread_mutex_t connection_mutex;

    SequenceID seqId;

    std::string jobId;

    double partitionToCoreRatio;

    std::shared_ptr<PhysicalOptimizer> physicalOptimizerPtr;

    StatisticsPtr statsForOptimization;

    std::shared_ptr<ShuffleInfo> shuffleInfo;
};
}


#endif