DispatcherServer.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 OBJECTQUERYMODEL_DISPATCHER_H
#define OBJECTQUERYMODEL_DISPATCHER_H

#include "ServerFunctionality.h"
#include "PDBLogger.h"
#include "PDBWork.h"
#include "PartitionPolicy.h"
#include "UseTemporaryAllocationBlock.h"
#include "PDBVector.h"

#include "NodeDispatcherData.h"
#include "StorageClient.h"
#include "StatisticsDB.h"

#include <string>
#include <queue>
#include <unordered_map>
#include <vector>



namespace pdb {

// The DispatcherServer partitions and then forwards a Vector of pdb::Objects received from a
// DispatcherClient to the proper storage servers
// So far, there are two dispatching policies are supported:
// -- Random Policy: the received Vector will be sent to any storage node determined randomly
// -- Round-Robin Policy: the first received Vector will be sent to the first storage node, 
//    and so on.


class DispatcherServer : public ServerFunctionality {

public:
    DispatcherServer(PDBLoggerPtr logger, std::shared_ptr<StatisticsDB> statisticsDB);

    ~DispatcherServer();

    void initialize();

    void registerHandlers(PDBServer& forMe) override;

    void registerStorageNodes(Handle<Vector<Handle<NodeDispatcherData>>> storageNodes);

    void registerSet(std::pair<std::string, std::string> setAndDatabase,
                     PartitionPolicyPtr partitionPolicy);

    bool dispatchData(std::pair<std::string, std::string> setAndDatabase,
                      std::string type,
                      Handle<Vector<Handle<Object>>> toDispatch);
    bool dispatchBytes(std::pair<std::string, std::string> setAndDatabase,
                       std::string type,
                       char* bytes,
                       size_t numBytes);


    void waitAllRequestsProcessed() {
        pthread_mutex_lock(&mutex);
        while (numRequestsInProcessing > 0) {
            pthread_mutex_unlock(&mutex);
            sleep(1);
            pthread_mutex_lock(&mutex);
        }
        pthread_mutex_unlock(&mutex);
    }

private:
    PDBLoggerPtr logger;
    std::shared_ptr<StatisticsDB> statisticsDB;
    Handle<Vector<Handle<NodeDispatcherData>>> storageNodes;
    std::map<std::pair<std::string, std::string>, PartitionPolicyPtr> partitionPolicies;

    bool validateTypes(const std::string& databaseName,
                       const std::string& setName,
                       const std::string& typeName,
                       std::string& errMsg);

    bool sendData(std::pair<std::string, std::string> setAndDatabase,
                  std::string type,
                  Handle<NodeDispatcherData> destination,
                  Handle<Vector<Handle<Object>>> toSend);

    bool sendBytes(std::pair<std::string, std::string> setAndDatabase,
                   std::string type,
                   Handle<NodeDispatcherData> destination,
                   char* bytes,
                   size_t numBytes);

    Handle<NodeDispatcherData> findNode(NodeID nodeId);
    int numRequestsInProcessing = 0;
    pthread_mutex_t mutex;
};
}


#endif  // OBJECTQUERYMODEL_DISPATCHER_H