/* This file is part of VoltDB.
* Copyright (C) 2008-2015 VoltDB Inc.
*
* Permission is hereby granted, free of charge, to any person obtaining
* a copy of this software and associated documentation files (the
* "Software"), to deal in the Software without restriction, including
* without limitation the rights to use, copy, modify, merge, publish,
* distribute, sublicense, and/or sell copies of the Software, and to
* permit persons to whom the Software is furnished to do so, subject to
* the following conditions:
*
* The above copyright notice and this permission notice shall be
* included in all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
* MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
* IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
* OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
* ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
* OTHER DEALINGS IN THE SOFTWARE.
*/
#include
#include "Distributer.h"
#include "InvocationResponse.hpp"
#include "ProcedureCallback.hpp"
#include "ElasticHashinator.h"
#include "Table.h"
#include "TableIterator.h"
#include "Row.hpp"
#include
#include
#include
#include
#include
namespace voltdb {
boost::shared_mutex Distributer::m_procInfoLock;
const int Distributer::MP_INIT_PID = 16383;
ProcedureInfo::ProcedureInfo(const std::string & jsonText):PARAMETER_NONE(-1){
boost::property_tree::ptree pt;
//todo verify json
std::stringstream ss(jsonText);
boost::property_tree::read_json(ss, pt);
//"{"readOnly":true,"singlePartition":false}
//{"partitionParameter":0,"readOnly":true,"partitionParameterType":6,"singlePartition":true}
m_partitionParameter = pt.get("partitionParameter", PARAMETER_NONE);
m_partitionParameterType = pt.get("partitionParameterType", PARAMETER_NONE);
//Mandatory
m_readOnly = pt.get("readOnly");
m_multiPart = !pt.get("singlePartition");
debug_msg("m_partitionParameter " <hashinate((strVal.data()), strVal.size());
}
default:
//not supported
return -1;
}
if (val != INT64_MIN)
return m_hashinator->hashinate(val);
return -1;
}
int Distributer::getHashedPartitionForParameter(ByteBuffer ¶mBuffer, int parameterId){
int index = 5;//offset
bool wasNull;
std::string name = paramBuffer.getString(index, wasNull);
//Skip procedure name and size, client data
index += sizeof(int32_t) + name.size() + sizeof(int64_t);
//get number of parameters
paramBuffer.getInt16(index);
index += 2;
//partition key must be the first
if (parameterId > 0)
return -1;//throw
return parseParameter(paramBuffer, index);
}
ProcedureInfo* Distributer::getProcedure(const std::string& procName) throw (UnknownProcedureException)
{
std::map<:string procedureinfo>::iterator it = m_procedureInfo.find(procName);
if (it == m_procedureInfo.end())
return NULL;
return &it->second;
}
int Distributer::getHostIdByPartitionId(int partitionId)
{
std::map::iterator it = m_PartitionToHostId.find(partitionId);
if (it == m_PartitionToHostId.end())
return -1;
return it->second;
}
void Distributer::handleTopologyNotification(const std::vector<:table>& t){
// If The savedTopoTable is not the same as our notified one, we have to update the hashinator
if (m_savedTopoTable == t[0]) {
return;
}
updateAffinityTopology(t);
debug_msg("updateAffinityTopology after notification");
}
void Distributer::updateAffinityTopology(const std::vector<:table>& topoTable){
// Partition, Sites, Leader
// 0, 0:0, 0:0
// 1, 1:0, 1:0
// 2, 2:0, 2:0
// 3, 0:1, 0:1
// 4, 1:1, 1:1
// 5, 2:1, 2:1
// 16383, 2:2, 2:2
debug_msg("updateAffinityTopology ");
m_PartitionToHostId.clear();
voltdb::TableIterator tableIter = topoTable[0].iterator();
while (tableIter.hasNext())
{
voltdb::Row row = tableIter.next();
debug_msg(row.toString());
int partitionId = row.getInt32(0);
std::istringstream ss(row.getString(2));
int hostId;
std::string token;
//parse host Id
std::getline(ss, token, ':');
hostId = atoi(token.c_str());
debug_msg("updateAffinityTopology: partitionId=" < tokens(new char[realsize]);
hashRow.getVarbinary(1, realsize, (uint8_t*)(tokens.get()), &realsize);
m_hashinator.reset(new ElasticHashinator(tokens.get()));
m_isElastic = true;
}else{
m_isElastic = false;
}
//mark update status as finished
m_isUpdating = false;
m_savedTopoTable = topoTable[0];
}
void Distributer::updateProcedurePartitioning(const std::vector<:table>& procInfoTable){
debug_msg("updateProcedurePartitioning ");
boost::unique_lock<:shared_mutex> lock(m_procInfoLock);
m_procedureInfo.clear();
voltdb::TableIterator tableIter = procInfoTable[0].iterator();
while (tableIter.hasNext())
{
voltdb::Row row = tableIter.next();
std::string procedureName = row.getString(2);
std::string jsonString = row.getString(6);
m_procedureInfo.insert(std::pair<:string procedureinfo>(procedureName, ProcedureInfo(jsonString)));
}
}
}