text:
/* This file is part of VoltDB.
 * Copyright (C) 2008-2018 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.
 */

#ifndef VOLTDB_PROCEDURECALLBACK_HPP_
#define VOLTDB_PROCEDURECALLBACK_HPP_
#include "InvocationResponse.hpp"
#include <boost/shared_ptr.hpp>
namespace voltdb {

/*
 * Abstract base class for callbacks to provide to the
 * API with procedure invocations
 */
class ProcedureCallback {
public:

    enum AbandonReason { NOT_ABANDONED, TOO_BUSY };

    /*
     * Invoked when a response to an invocation is available or
     * the connection to the node the invocation was sent to was lost.
     * Callbacks should not throw user exceptions.
     *
     * The API surrounds the callback in a try catch for std::exception
     * and passes any caught exceptions to the status listener.
     * @return true if the event loop should break after invoking this callback, false otherwise
     */
    virtual bool callback(InvocationResponse response)  = 0;
    virtual void abandon(AbandonReason reason) {}
    // Mechanism for procedure to over-ride abandon property set in client in event of backpressure.
    // @return true: allow abandoning of requests in case of back pressure
    //         false: don't abandon the requests in back pressure scenario.
    virtual bool allowAbandon() const {return true;}
    virtual ~ProcedureCallback() {}
};
}

#endif /* VOLTDB_PROCEDURECALLBACK_HPP_ */

ext: hpp

max_stars_repo_name: AVAInformationSystems/voltdb-client-cpp

max_issues_repo_name: AVAInformationSystems/voltdb-client-cpp

max_forks_repo_name: AVAInformationSystems/voltdb-client-cpp


text:
//
// Created by soliareofastora on 27.05.2021.
//

#include <boost/python.hpp>

#include "atoms_file_io.h"
#include "load_contact_maps.h"
#include "python_utils.h"

namespace py = boost::python;
namespace np = py::numpy;

BOOST_PYTHON_MODULE (libAtomDistanceIO) {
  py::def("initialize", Initialize);
  py::def("save_atoms", SaveAtomsFile);
  py::def("load_contact_map", LoadContactMap);
  py::def("load_aligned_contact_map", LoadAlignedContactMap);
}

ext: cpp

max_stars_repo_name: bioinf-mcb/metagenomic-deepFRI

max_issues_repo_name: bioinf-mcb/metagenomic-deepFRI

max_forks_repo_name: bioinf-mcb/Metagenomic-DeepFRI


text:
#include "MultivariateNormalDistribution.hpp"

#include <rw/kinematics/Kinematics.hpp>
#include <rw/loaders/rwxml/XMLRWLoader.hpp>
#include <rw/models/SerialDevice.hpp>
#include <rwlibs/calibration/CalibrationMeasurement.hpp>
#include <rwlibs/calibration/WorkCellCalibration.hpp>
#include <rwlibs/calibration/xml/XMLCalibrationMeasurementFile.hpp>

#include <boost/program_options/parsers.hpp>
#include <boost/program_options/variables_map.hpp>

using namespace rw::kinematics;
using namespace rw::models;
using namespace rwlibs::calibration;

boost::program_options::options_description optionsDescription("Options");
boost::program_options::positional_options_description positionalOptionsDescription;
boost::program_options::variables_map variablesMap;

std::string workCellFilePath;
std::string deviceName1;
std::string deviceName2;
std::string sensorFrameName1;
std::string sensorFrameName2;
std::string markerFrameName1;
std::string markerFrameName2;
std::string measurementFilePath;
int measurementCount;
bool addNoise;

int parseArguments(int argumentCount, char** arguments);
void printHelp();
std::vector<CalibrationMeasurement::Ptr> generateMeasurements(rw::models::SerialDevice::Ptr serialDevice,
	rw::kinematics::Frame::Ptr referenceFrame, rw::kinematics::Frame::Ptr measurementFrame, rw::kinematics::State state, unsigned int measurementCount,
	bool addNoise);

void printMeasurements(const std::vector<CalibrationMeasurement::Ptr>& measurements, rw::models::WorkCell::Ptr workCell, WorkCellCalibration::Ptr workCellCalibration);

int main(int argumentCount, char** arguments) {
	if (int parseResult = parseArguments(argumentCount, arguments) < 1)
		return parseResult;

	sensorFrameName2 = "";
	deviceName2 = "";
	markerFrameName2 = "";

	std::cout << "Initializing calibration measurement tool:" << std::endl;

	// Load workcell.
	std::cout << "\tLoading work cell [ " << workCellFilePath << " ].. ";
	std::cout.flush();
	rw::models::WorkCell::Ptr workCell = rw::loaders::XMLRWLoader::load(workCellFilePath);
	if (workCell.isNull()) {
		std::cout << "FAILED." << std::endl;
		return -1;
	}
	std::cout << "Loaded [ " << workCell->getName() << " ]." << std::endl;
	rw::kinematics::State state = workCell->getDefaultState();

	// Find device and frames.
	std::cout << "\tFinding device [ " << deviceName1 << " ].. ";
	std::cout.flush();

	std::vector<SerialDevice::Ptr> devices;
	SerialDevice::Ptr device = deviceName1.empty() ? workCell->getDevices().front().cast<SerialDevice>() : workCell->findDevice<SerialDevice>(deviceName1);
	if (device.isNull()) {
		std::cout << "Failed to find device: "<<deviceName1 << std::endl;
		return -1;
	}
	devices.push_back(device);
	if (deviceName2.empty() == false) {
		std::cout << "\tFinding device [ " << deviceName2 << " ].. ";
		std::cout.flush();
		SerialDevice::Ptr device = workCell->findDevice<SerialDevice>(deviceName2);
		if (device.isNull()) {
			std::cout << "Failed to find device: "<<deviceName2 << std::endl;
			return -1;
		}
		devices.push_back(device);
	}
	std::cout << "Found "<<devices.size()<<" devices." << std::endl;

	rw::models::SerialDevice::Ptr serialDevice = device.cast<rw::models::SerialDevice>();

	std::cout << "\tFinding reference frame [ " << sensorFrameName1 << " ].. ";
	std::cout.flush();
	rw::kinematics::Frame::Ptr sensorFrame = sensorFrameName1.empty() ? workCell->findFrame("WORLD") : workCell->findFrame(sensorFrameName1);
	if (sensorFrame.isNull()) {
		std::cout << "Failed to find sensor frame: "<<sensorFrameName1<< std::endl;
		return -1;
	}

	std::vector<Frame*> sensorFrames;
	sensorFrames.push_back(sensorFrame.get());
	if (sensorFrameName2.empty() == false) {
		std::cout << "\tFinding reference frame [ " << sensorFrameName2 << " ].. ";
		std::cout.flush();
		rw::kinematics::Frame::Ptr sensorFrame = workCell->findFrame(sensorFrameName2);
		if (sensorFrame.isNull()) {
			std::cout << "Failed to find sensor frame: "<<sensorFrameName2<< std::endl;
			return -1;
		}
		sensorFrames.push_back(sensorFrame.get());
	}

	std::cout << "Found "<<sensorFrames.size()<<" sensor frames." << std::endl;

	std::cout << "\tFinding marker frame [ " << markerFrameName1 << " ].. ";
	std::cout.flush();
	rw::kinematics::Frame* markerFrame = markerFrameName1.empty() ? device->getEnd() : workCell->findFrame(markerFrameName1);
	if (markerFrame == NULL) {
		std::cout << "Failed to find marker frame: "<<markerFrameName1<< std::endl;
		return -1;
	}
	std::vector<Frame*> markerFrames;
	markerFrames.push_back(markerFrame);
	if (markerFrameName2.empty() == false) {
		std::cout << "\tFinding marker frame [ " << markerFrameName2 << " ].. ";
		std::cout.flush();
		rw::kinematics::Frame* markerFrame = workCell->findFrame(markerFrameName2);
		if (markerFrame == NULL) {
			std::cout << "Failed to find marker frame: "<<markerFrameName2<< std::endl;
			return -1;
		}
		markerFrames.push_back(markerFrame);
	}
	std::cout << "Found "<<markerFrames.size()<<" marker frames." << std::endl;
	if (devices.size() != markerFrames.size()) {
		std::cout<<"Failure: The number of devices and marker frames must be equal"<<std::endl;
		return -1;
	}

	std::cout << "Initializing artificial calibration..";
	std::cout.flush();

	typedef std::pair<SerialDevice::Ptr, Frame*> DevAndFrame;
	std::vector<DevAndFrame> deviceAndFramePairs;
	for (size_t i = 0; i<devices.size(); i++) {
		deviceAndFramePairs.push_back(std::make_pair(devices[i], markerFrames[i]));
	}

	WorkCellCalibration::Ptr artificialCalibration(rw::common::ownedPtr(new WorkCellCalibration(deviceAndFramePairs, sensorFrames)));
	artificialCalibration->getFixedFrameCalibrations()->getCalibration(0)->setCorrectionTransform(rw::math::Transform3D<>(rw::math::Vector3D<>(10.0 / 1000.0, -8.0 / 1000.0, 7 / 1000.0),
																														  rw::math::RPY<>(1.7 * rw::math::Deg2Rad, 0.7 * rw::math::Deg2Rad, -2.0 * rw::math::Deg2Rad)));
	artificialCalibration->getFixedFrameCalibrations()->getCalibration(1)->setCorrectionTransform(rw::math::Transform3D<>(rw::math::Vector3D<>(-9.0 / 1000.0, 11.0 / 1000.0, 17.0 / 1000.0),
																														  rw::math::RPY<>(1 * rw::math::Deg2Rad, 1.3 * rw::math::Deg2Rad, -1.4 * rw::math::Deg2Rad)));
	//artificialCalibration->getFixedFrameCalibrations()->getCalibration(1)->setCorrectionTransform(rw::math::Transform3D<>(rw::math::Vector3D<>(3.0 / 1000.0, -11.0 / 1000.0, -3.0 / 1000.0), rw::math::RPY<>(1.3 * rw::math::Deg2Rad, -1.2 * rw::math::Deg2Rad, 0.7 * rw::math::Deg2Rad)));
	//artificialCalibration->getFixedFrameCalibrations()->getCalibration(2)->setCorrectionTransform(rw::math::Transform3D<>(rw::math::Vector3D<>(1.0 / 1000.0, 2.0 / 1000.0, -3.0 / 1000.0), rw::math::RPY<>(-0.3 * rw::math::Deg2Rad, 0.2 * rw::math::Deg2Rad, 0.1 * rw::math::Deg2Rad)));

	CompositeCalibration<ParallelAxisDHCalibration>::Ptr artificialCompositeLinkCalibration = artificialCalibration->getCompositeLinkCalibration();
	for (unsigned int calibrationIndex = 0; calibrationIndex < (unsigned int)artificialCompositeLinkCalibration->getCalibrationCount(); calibrationIndex++) {
		ParallelAxisDHCalibration::Ptr artificialLinkCalibration = artificialCompositeLinkCalibration->getCalibration(calibrationIndex);
		CalibrationParameterSet parameterSet = artificialLinkCalibration->getParameterSet();
		if (parameterSet(ParallelAxisDHCalibration::PARAMETER_A).isEnabled())
			parameterSet(ParallelAxisDHCalibration::PARAMETER_A) = 0.3 / 100.0;
		if (parameterSet(ParallelAxisDHCalibration::PARAMETER_B).isEnabled())
			parameterSet(ParallelAxisDHCalibration::PARAMETER_B) = -0.2 / 100.0;
		if (parameterSet(ParallelAxisDHCalibration::PARAMETER_D).isEnabled())
			parameterSet(ParallelAxisDHCalibration::PARAMETER_D) = -0.1 / 100.0;
		if (parameterSet(ParallelAxisDHCalibration::PARAMETER_ALPHA).isEnabled())
			parameterSet(ParallelAxisDHCalibration::PARAMETER_ALPHA) = -0.6 * rw::math::Deg2Rad;
		if (parameterSet(ParallelAxisDHCalibration::PARAMETER_BETA).isEnabled())
			parameterSet(ParallelAxisDHCalibration::PARAMETER_BETA) = 0.5 * rw::math::Deg2Rad;
		if (parameterSet(ParallelAxisDHCalibration::PARAMETER_THETA).isEnabled())
			parameterSet(ParallelAxisDHCalibration::PARAMETER_THETA) = 0.4 * rw::math::Deg2Rad;
		artificialLinkCalibration->setParameterSet(parameterSet);
	}
	CompositeCalibration<JointEncoderCalibration>::Ptr artificialCompositeJointCalibration = artificialCalibration->getCompositeJointEncoderCalibration();
	for (unsigned int calibrationIndex = 0; calibrationIndex < (unsigned int)artificialCompositeJointCalibration->getCalibrationCount(); calibrationIndex++) {
		JointEncoderCalibration::Ptr artificialJointCalibration = artificialCompositeJointCalibration->getCalibration(calibrationIndex);
		CalibrationParameterSet parameterSet = artificialJointCalibration->getParameterSet();
		if (parameterSet(JointEncoderCalibration::PARAMETER_TAU).isEnabled())
			parameterSet(JointEncoderCalibration::PARAMETER_TAU) = 0.003;
		if (parameterSet(JointEncoderCalibration::PARAMETER_SIGMA).isEnabled())
			parameterSet(JointEncoderCalibration::PARAMETER_SIGMA) = -0.002;
		artificialJointCalibration->setParameterSet(parameterSet);
	}
	std::cout << " Initialized." << std::endl;

	std::cout << "Applying artificial calibration..";
	std::cout.flush();
	//artificialCalibration->apply();
	std::cout << " Applied." << std::endl;

	std::vector<CalibrationMeasurement::Ptr> measurements;
	std::cout << "Generating measurements [ Count: " << measurementCount << " - Noise: " << (addNoise ? "Enabled" : "Disabled") << " ]..";
	for (size_t i = 0; i<deviceAndFramePairs.size(); i++) {
		for (size_t j = 0; j<sensorFrames.size(); j++) {
			std::vector<CalibrationMeasurement::Ptr> m = generateMeasurements(deviceAndFramePairs[i].first, sensorFrames[j], deviceAndFramePairs[i].second, state, measurementCount, addNoise);
			measurements.insert(measurements.end(), m.begin(), m.end());
		}
	}
	std::cout << " Generated." << std::endl;

	std::cout << "Reverting artificial calibration..";
	std::cout.flush();
	std::cout.flush();
	artificialCalibration->revert();
	std::cout << " Reverted." << std::endl;

	std::cout << "Saving measurement file [ " << measurementFilePath << " ]..";
	std::cout.flush();
	XMLCalibrationMeasurementFile<>::save(measurements, measurementFilePath);
	std::cout << " Saved." << std::endl;

	std::cout << "Residual summary:" << std::endl;
	printMeasurements(measurements, workCell, artificialCalibration);

	return 0;
}

int parseArguments(int argumentCount, char** arguments) {
	optionsDescription.add_options()("help", "Print help message")("workCellFile", boost::program_options::value<std::string>(&workCellFilePath)->required(),
		"Set the work cell file path")
		("measurementFile", boost::program_options::value<std::string>(&measurementFilePath)->required(), "Set the measurement file path")
		("device1",	boost::program_options::value<std::string>(&deviceName1), "Set the device 1 name")
		("device2",	boost::program_options::value<std::string>(&deviceName2)->default_value(""), "Set the device 2 name")
		("sensorFrame1", boost::program_options::value<std::string>(&sensorFrameName1), "Set sensor frame 1 name")
		("sensorFrame2", boost::program_options::value<std::string>(&sensorFrameName2)->default_value(""), "Set sensor frame 2 name")
		("markerFrame1", boost::program_options::value<std::string>(&markerFrameName1), "Set the marker frame 1 name")
		("markerFrame2", boost::program_options::value<std::string>(&markerFrameName2)->default_value(""), "Set the marker frame 2 name")
		("measurementCount", boost::program_options::value<int>(&measurementCount)->default_value(100), "Set the number of measurements")
		("addNoise", boost::program_options::value<bool>(&addNoise)->default_value(false), "Enable/disable noise on measurements");

	positionalOptionsDescription.add("workCellFile", 1).add("measurementFile", 1);

	try {
		boost::program_options::store(
			boost::program_options::command_line_parser(argumentCount, arguments).options(optionsDescription).positional(positionalOptionsDescription).run(),
			variablesMap);

		if (variablesMap.count("help")) {
			printHelp();
			return 0;
		}

		boost::program_options::notify(variablesMap);
	} catch (boost::program_options::error& error) {
		std::cerr << "Error: " << error.what() << std::endl;
		std::cerr << std::endl;
		printHelp();
		return -1;
	}

	return 1;
}

void printHelp() {
	std::cerr << "Usage:" << std::endl;
	std::cerr << "   sdurw_calibration-mtool [work cell file] [measurement file]" << std::endl;
	std::cerr << std::endl;
	std::cerr << optionsDescription << std::endl;
}

std::vector<CalibrationMeasurement::Ptr> generateMeasurements(rw::models::SerialDevice::Ptr serialDevice,
															  rw::kinematics::Frame::Ptr sensorFrame,
															  rw::kinematics::Frame::Ptr markerFrame,
															  rw::kinematics::State state,
															  unsigned int measurementCount,
															  bool addNoise)
{
		MultivariateNormalDistribution<double, 6> mvnd((unsigned int)time(0));

		std::vector<CalibrationMeasurement::Ptr> measurements;
		for (unsigned int measurementIndex = 0; measurementIndex < measurementCount; measurementIndex++) {
			rw::math::Q q = rw::math::Math::ranQ(serialDevice->getBounds());
			serialDevice->setQ(q, state);

			rw::math::Transform3D<> transform = rw::kinematics::Kinematics::frameTframe(sensorFrame.get(), markerFrame.get(), state);
			if (transform.R()(2,2) > -0.5)
				continue;

			Eigen::Matrix<double, 6, 6> covariance = Eigen::Matrix<double, 6, 6>::Identity();
			if (addNoise) {
				Eigen::Matrix<double, 6, 6> random = Eigen::Matrix<double, 6, 6>::Random();
				covariance = random.transpose() * random;
				covariance.block<3, 3>(0, 0) /= 50.0;
				covariance.block<3, 3>(3, 3) /= 5.0;
				covariance.block<3, 3>(3, 0) /= 1000.0;
				covariance.block<3, 3>(0, 3) /= 1000.0;
				covariance /= 10e4;

				Eigen::Matrix<double, 6, 1> mvndVector = mvnd.draw(covariance);
				transform.P() = rw::math::Vector3D<>(mvndVector(0), mvndVector(1), mvndVector(2)) + transform.P();
				transform.R() = rw::math::RPY<>(mvndVector(3), mvndVector(4), mvndVector(5)).toRotation3D() * transform.R();
			}
			CalibrationMeasurement::Ptr measurement = rw::common::ownedPtr(new CalibrationMeasurement(q, transform, covariance));
			measurement->setDeviceName(serialDevice->getName());
			measurement->setMarkerFrameName(markerFrame->getName());
			measurement->setSensorFrameName(sensorFrame->getName());
			measurements.push_back(measurement);
		}

		return measurements;
}

void printMeasurements(const std::vector<CalibrationMeasurement::Ptr>& measurements, rw::models::WorkCell::Ptr workCell, WorkCellCalibration::Ptr workCellCalibration) {
	const unsigned int measurementCount = (unsigned int)measurements.size();

	Eigen::VectorXd distances(measurementCount), angles(measurementCount);
	Eigen::VectorXd calibratedDistances(measurementCount), calibratedAngles(measurementCount);
	rw::kinematics::State state = workCell->getDefaultState();
	for (unsigned int measurementIndex = 0; measurementIndex < measurementCount; measurementIndex++) {
		SerialDevice::Ptr serialDevice = workCell->findDevice<SerialDevice>(measurements[measurementIndex]->getDeviceName());
		Frame::Ptr sensorFrame = workCell->findFrame(measurements[measurementIndex]->getSensorFrameName());
		Frame::Ptr markerFrame = workCell->findFrame(measurements[measurementIndex]->getMarkerFrameName());
		serialDevice->setQ(measurements[measurementIndex]->getQ(), state);

		const rw::math::Transform3D<> tfmMeasurement = measurements[measurementIndex]->getTransform();
		const rw::math::Transform3D<> tfmModel = rw::kinematics::Kinematics::frameTframe(sensorFrame.get(), markerFrame.get(), state);
		workCellCalibration->apply();
		const rw::math::Transform3D<> tfmCalibratedModel =
			rw::kinematics::Kinematics::frameTframe(sensorFrame.get(), markerFrame.get(), state);
		workCellCalibration->revert();
		const rw::math::Transform3D<> tfmError = rw::math::Transform3D<>(tfmModel.P() - tfmMeasurement.P(), tfmModel.R() * rw::math::inverse(tfmMeasurement.R()));
		const rw::math::Transform3D<> tfmCalibratedError = rw::math::Transform3D<>(tfmCalibratedModel.P() - tfmMeasurement.P(), tfmCalibratedModel.R() * rw::math::inverse(tfmMeasurement.R()));

		distances(measurementIndex) = tfmError.P().norm2(), calibratedDistances(measurementIndex) = tfmCalibratedError.P().norm2();
		angles(measurementIndex) = rw::math::EAA<>(tfmError.R()).angle(), calibratedAngles(measurementIndex) = rw::math::EAA<>(
			tfmCalibratedError.R()).angle();

		std::cout << "\tMeasurement " << measurementIndex + 1 << ": [ Uncalibrated: " << distances(measurementIndex) * 1000.0 << " mm / "
			<< angles(measurementIndex) * rw::math::Rad2Deg << " \u00B0 - Calibrated: " << calibratedDistances(measurementIndex) * 1000.0 << " mm / "
			<< calibratedAngles(measurementIndex) * rw::math::Rad2Deg << " \u00B0 ]" << std::endl;
	}
	std::cout << "\tSummary - Uncalibrated: [ Avg: " << distances.mean() * 1000.0 << " mm / " << angles.mean() * rw::math::Rad2Deg << " \u00B0 - Min: "
		<< distances.minCoeff() * 1000.0 << " mm / " << angles.minCoeff() * rw::math::Rad2Deg << " \u00B0 - Max: " << distances.maxCoeff() * 1000.0 << " mm / "
		<< angles.maxCoeff() * rw::math::Rad2Deg << " \u00B0 ]" << std::endl;
	std::cout << "\tSummary - Calibrated: [ Avg: " << calibratedDistances.mean() * 1000.0 << " mm / " << calibratedAngles.mean() * rw::math::Rad2Deg << " \u00B0 - Min: "
		<< calibratedDistances.minCoeff() * 1000.0 << " mm / " << calibratedAngles.minCoeff() * rw::math::Rad2Deg << " \u00B0 - Max: "
		<< calibratedDistances.maxCoeff() * 1000.0 << " mm / " << calibratedAngles.maxCoeff() * rw::math::Rad2Deg << " \u00B0 ]" << std::endl;
}


text:
// @COPYRIGHT@
// Licensed under MIT license.
// See LICENSE.TXT file in the project root for more information.
// ==============================================================

#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <atomic>
#include <malloc.h>
#include <iostream>
#include <unistd.h>
#include <assert.h>
#include <string.h>
#include <sys/mman.h>
#include <sstream>
#include <functional>
#include <unordered_set>
#include <vector>
#include <unordered_map>
#include <algorithm>
#include <list>
#include <bits/stdc++.h>
#include <boost/property_tree/json_parser.hpp>
#include <boost/optional.hpp>
#include <nlohmann/json.hpp>
#include "pin.H"
#include "cctlib.H"
#include "shadow_memory.H"
#include "rbtree.h"
#include <xmmintrin.h>
#include <immintrin.h>
#include <sys/time.h>
#include <sys/resource.h>

extern "C" {
#include "xed-interface.h"
#include "xed-common-hdrs.h"
}

#include <google/sparse_hash_map>
#include <google/dense_hash_map>
using google::sparse_hash_map;  // namespace where class lives by default
using google::dense_hash_map;
using json = nlohmann::json;

// Short alias for this namespace
namespace pt = boost::property_tree;
using namespace std;
using namespace PinCCTLib;

#define MAX_REUSE_DISTANCE_BINS (32) // 4 GB
#define MAX_REUSE_DISTANCE (1ULL<< (MAX_REUSE_DISTANCE_BINS)) // 4 GB

#define CACHELINE_BIT (6)
#define CACHELINE_SIZE (1L<<CACHELINE_BIT)

#define NUM_BLOCKS (3)

#define FIRST_USE (ULLONG_MAX)

#define MULTI_THREADED

struct{
    char dummy1[128];
    xed_state_t  xedState;
    char dummy2[128];
} InsReuseGlobals;

using RBTree_t = RBTree<uint64_t, uint32_t, uint64_t>;

constexpr struct {
  size_t blkSize;
  char * blkDescription;
} BLK_INFO[NUM_BLOCKS] = {
  {CACHELINE_SIZE, (char *) "64B CacheLineReuse"},
  {4096, (char *) "4K OS PageSizeReuse"},
  {1L<<21, (char *) "2M Huge PageSizeReuse"},
};

struct InsReuseThreadData{
    char padding1[128];
    uint64_t bytesLoad;
    long long numIns;
    RBTree_t insRBTree;
    uint64_t numInsExecuted;
    uint64_t footprint;
    uint64_t insReuseHisto[MAX_REUSE_DISTANCE_BINS];
    ShadowMemory<uint64_t> smIns;
    struct{
        RBTree_t rbTree;
        uint64_t numBlocksCounter;
        uint64_t footprint;
        uint64_t reuseHisto[MAX_REUSE_DISTANCE_BINS];
        void * prevBlock;
        ShadowMemory<uint64_t> sm;
    } blockData[NUM_BLOCKS];

    bool sampleFlag;
    char padding2[128];
};

static struct {
    PIN_LOCK lock;
    uint64_t numInsExecuted;
    uint64_t footprint;
    uint64_t insReuseHisto[MAX_REUSE_DISTANCE_BINS];
    struct{
        uint64_t numBlocksCounter;
        uint64_t footprint;
        uint64_t reuseHisto[MAX_REUSE_DISTANCE_BINS];
    } blockData[NUM_BLOCKS];
} GLOBAL_STATS;

// key for accessing TLS storage in the threads. initialized once in main()
static  TLS_KEY client_tls_key;
static InsReuseThreadData* gSingleThreadedTData;

// function to access thread-specific data
inline InsReuseThreadData* ClientGetTLS(const THREADID threadId) {
#ifdef MULTI_THREADED
    InsReuseThreadData* tdata =
    static_cast<InsReuseThreadData*>(PIN_GetThreadData(client_tls_key, threadId));
    return tdata;
#else
    return gSingleThreadedTData;
#endif
}

static inline size_t GetBlock(size_t addr, size_t blockMask) {
    return addr & (blockMask);
}

static INT32 Usage() {
    PIN_ERROR("Pin tool to profile instruction reuse distance.\n" + KNOB_BASE::StringKnobSummary() + "\n");
    return -1;
}

// Main for RedSpy, initialize the tool, register instrumentation functions and call the target program.
static FILE* gTraceFile;
static ofstream gJSONFile;

// Initialized the needed data structures before launching the target program
static void ClientInit(int argc, char* argv[]) {
    // Create output file
    char name[MAX_FILE_PATH] = "insReuse.out.";
    char* envPath = getenv("INS_REUSE_CLIENT_OUTPUT_FILE");

    if(envPath) {
        // assumes max of MAX_FILE_PATH
        strcpy(name, envPath);
    }

    gethostname(name + strlen(name), MAX_FILE_PATH - strlen(name));
    pid_t pid = getpid();
    sprintf(name + strlen(name), "%d", pid);
    cerr << "\n Creating log file at:" << name << "\n";
    gTraceFile = fopen(name, "w");
    // print the arguments passed
    fprintf(gTraceFile, "\n");

    for(int i = 0 ; i < argc; i++) {
        fprintf(gTraceFile, "%s ", argv[i]);
    }

    fprintf(gTraceFile, "\n");

    int i = 0;
    for(i = 0 ; i < argc; i++) {
        if (0 == strcmp(argv[i], "--")) {
            i++;
            break;
        }
    }
    json j;
    if (i < argc) {
        j["exe"] =  argv[i];
        i++;
    }
    stringstream ss;
    for(; i < argc; i++) {
        ss << argv[i] <<" ";
    }
    j["args"] =  ss.str();
    gJSONFile.open(string(name) + ".json");
    gJSONFile << j.dump(4) << "\n";

    // Init Xed
    // Init XED for decoding instructions
    xed_state_init(&InsReuseGlobals.xedState, XED_MACHINE_MODE_LONG_64, (xed_address_width_enum_t) 0, XED_ADDRESS_WIDTH_64b);

    // init some globals
    PIN_InitLock (&GLOBAL_STATS.lock);
    GLOBAL_STATS.numInsExecuted = 0;
    GLOBAL_STATS.footprint = 0;
    for (int i=0; i < MAX_REUSE_DISTANCE_BINS; i++) {
    	GLOBAL_STATS.insReuseHisto[i] = 0;
    }
    for (int j = 0; j < NUM_BLOCKS; j++) {
        GLOBAL_STATS.blockData[j].numBlocksCounter = 0;
        GLOBAL_STATS.blockData[j].footprint = 0;
        for (int i=0; i < MAX_REUSE_DISTANCE_BINS; i++) {
    	    GLOBAL_STATS.blockData[j].reuseHisto[i] = 0;
        }
    }
    fflush(gTraceFile);
    gJSONFile.flush();
}

static inline void UpdateInsReuseStats(uint64_t distance, uint64_t count, InsReuseThreadData* tData){
    uint64_t bin;

    // Bin 0: [0, 1)
    // Bin 1  [1, 2)
    // Bin 2  [2, 4)
    // Bin 3  [4, 8)

    if (distance >= MAX_REUSE_DISTANCE) {
        bin = MAX_REUSE_DISTANCE_BINS-1;
    } else if (distance == 0) {
        bin = 0;
    } else {
        bin = (sizeof(uint64_t)*8 - 1 - __builtin_clzl(distance)) + 1 /* 1 for dedicated bin 0*/;
    }
    tData->insReuseHisto[bin] += count;
}

static inline void UpdateBlockReuseStats(uint64_t distance, uint64_t count, uint64_t* reuseHisto){
    // Bin 0: [0, 1)
    // Bin 1  [1, 2)
    // Bin 2  [2, 4)
    // Bin 3  [4, 8)

    uint64_t bin;
    if (distance >= MAX_REUSE_DISTANCE) {
        bin = MAX_REUSE_DISTANCE_BINS-1;
    } else if (distance == 0) {
        bin = 0;
    } else {
        bin = (sizeof(uint64_t)*8 - 1 - __builtin_clzl(distance)) + 1 /* 1 for dedicated bin 0*/;
    }
    // One instruction is at this distance
    reuseHisto[bin] ++;
    // The rest of the instructions on the same cacheline have a zero distance
    if (count > 1) {
        reuseHisto[0] += count-1;
    }
}

static inline uint64_t ComputeInsReuseDistance(uint64_t prevTick, uint64_t newTick, uint32_t v, InsReuseThreadData * tData, RBTree_t * rbt){
    // Find how many RB-Tree nodes are to the right of this node.
    uint64_t reuseDist;
    auto node = rbt->FindSumGreaterEqual(prevTick, &reuseDist);
    if (node) {
        //  Delete the node from RB-tree
        auto retNode = rbt->Delete(node);
        // reinsert the node with new tick
        retNode->key =  newTick;
        retNode->value =  v;
        rbt->Insert(retNode);
        return reuseDist;
    } else {
        return FIRST_USE;
    }
}

static inline uint64_t ComputeBlockReuseDistance(uint64_t prevTick, uint64_t newTick, uint32_t v, InsReuseThreadData * tData, RBTree_t * rbt){
    // Find how many RB-Tree nodes are to the right of this node.
    uint64_t reuseDist;
    auto node = rbt->FindSumGreaterThan(prevTick, &reuseDist);
    if (node) {
        //  Delete the node from RB-tree
        auto retNode = rbt->Delete(node);
        // reinsert the node with new tick
        retNode->key =  newTick;
        retNode->value =  v;
        rbt->Insert(retNode);
        return reuseDist;
    } else {
        return FIRST_USE;
    }
}

static inline void AnalyzeInsLevelReuse(void * insAddr, uint32_t numInsInBBL,  THREADID threadId){
	assert(0 != numInsInBBL);
	InsReuseThreadData* tData = ClientGetTLS(threadId);
	tuple<uint64_t[SHADOW_PAGE_SIZE]> &t = tData->smIns.GetOrCreateShadowBaseAddress((uint64_t)insAddr);
	uint64_t * shadowMemAddr = &(get<0>(t)[PAGE_OFFSET((uint64_t)insAddr)]);

	uint64_t prevTick = *shadowMemAddr;
	uint64_t newTick = tData->numInsExecuted;

	// Update the number of instructions executed
	if (prevTick == 0 /* first use */) {
		tData->numInsExecuted += numInsInBBL;
		tData->footprint += numInsInBBL;
		// However, needs a new insertion
		auto newNode = new TreeNode<uint64_t, uint32_t, uint64_t>(tData->numInsExecuted, numInsInBBL);
		tData->insRBTree.Insert(newNode);
		*shadowMemAddr = tData->numInsExecuted;
	} else {
		// Update the tick if prevTick != newTick
		if (prevTick != newTick) {
			tData->numInsExecuted += numInsInBBL;
		}
		uint64_t reuseDistance = ComputeInsReuseDistance(prevTick,  tData->numInsExecuted, numInsInBBL, tData, & tData->insRBTree);
		assert(FIRST_USE != reuseDistance);
		UpdateInsReuseStats(reuseDistance, numInsInBBL, tData);
		*shadowMemAddr = tData->numInsExecuted;
	}
}

template <int blockSize, int blkIdx>
static inline void AnalyzeBlockLevelReuse(void * block, uint32_t numInsInBlock,  THREADID threadId){
	assert(0 != numInsInBlock);
	// blockSize must be a power of 2.
	assert(1 == __builtin_popcountll(blockSize));

	InsReuseThreadData* tData = ClientGetTLS(threadId);
	auto blockData = & (tData->blockData[blkIdx]);

	// Fast path: if block == prevBlock, simply increment the histo with 0 reuse distance and return
	if (blockData->prevBlock == block) {
		UpdateBlockReuseStats(0 /*reuseDistance*/, numInsInBlock, blockData->reuseHisto);
		return;
	}
	// update blockData->prevBlock to block.
	blockData->prevBlock = block;

	tuple<uint64_t[SHADOW_PAGE_SIZE]> &t = blockData->sm.GetOrCreateShadowBaseAddress((size_t)block);
	uint64_t * shadowMemAddr = &(get<0>(t)[PAGE_OFFSET((size_t)block)]);
	uint64_t prevTick = *shadowMemAddr;
	uint64_t newTick = ++blockData->numBlocksCounter;

	if (prevTick == 0 /* first use */) {
		// needs a new insertion
		auto newNode = new TreeNode<uint64_t, uint32_t, uint64_t>(newTick, 1);
		blockData->rbTree.Insert(newNode);
		blockData->footprint ++;

		if (numInsInBlock > 1) {
			// The rest of the instructions on the same cacheline have a zero distance
			blockData->reuseHisto[0] += numInsInBlock-1;
		}
	} else {
		// TODO: Optimization: if the last block access is same as this one, we can bypass this RB-Tree deletion, insertion
		// and instead directly increment the key. Need to be careful not to violate any tree properties.
		// May have to check to make sure that the node value is also unchanged.
		//uint64_t reuseDistance = ComputeBlockReuseDistance(prevTick, blockData->numBlocksCounter, numInsInBlock, tData, & blockData->rbTree);
		uint64_t reuseDistance = ComputeBlockReuseDistance(prevTick, newTick, 1, tData, &blockData->rbTree);
		assert(FIRST_USE != reuseDistance);
		UpdateBlockReuseStats(reuseDistance, numInsInBlock, blockData->reuseHisto);
	}
	*shadowMemAddr = newTick;
}

static void InstrumentInsLevelReuse(TRACE trace, void* f) {
    // Instruction-level reuse distance
    for (BBL bbl = TRACE_BblHead(trace); BBL_Valid(bbl); bbl = BBL_Next(bbl)){
        uint32_t totInsInBbl = BBL_NumIns(bbl);
        ADDRINT insAddr = BBL_Address(bbl);
        BBL_InsertCall(bbl,
                       IPOINT_BEFORE,(AFUNPTR)AnalyzeInsLevelReuse,
                       IARG_ADDRINT, insAddr,
                       IARG_UINT32, totInsInBbl,
                       IARG_THREAD_ID, IARG_END);
    }
}

template <int blockSize, int blkIdx>
static void InstrumentMemBlockLevelReuse(TRACE trace, void* f) {
    // blockSize must be a power of 2.
    assert(1 == __builtin_popcountll(blockSize));
    const size_t blockBits = __builtin_ctzll(blockSize);
    const size_t blockMask = ~(blockSize-1);

    // Cacheline-level reuse distance
    for (BBL bbl = TRACE_BblHead(trace); BBL_Valid(bbl); bbl = BBL_Next(bbl)){
        uint32_t numInsInMemBlock = 0;
        size_t prevMemBlock = GetBlock((uint64_t) INS_Address(BBL_InsHead(bbl)), blockMask);
        for(INS ins = BBL_InsHead(bbl); INS_Valid(ins); ins=INS_Next(ins)) {
            // Does instruction straddle two mem blocks?
            size_t insStartMemBlock = GetBlock((size_t) INS_Address(ins), blockMask);
            size_t insEndMemBlock = GetBlock((size_t) (INS_Address(ins) + INS_Size(ins) - 1), blockMask);

            // +1 for the ongoing mem block
            if(insStartMemBlock == prevMemBlock) {
                numInsInMemBlock ++;
            }

            // Assumption: ins can never span more than one memory block.
            // Continue adding more to this mem block
            if (insEndMemBlock == prevMemBlock) {
                continue;
            }
            // either the ins ends in a new block or begins on a new block.

            INS_InsertCall(ins,
                           IPOINT_BEFORE,(AFUNPTR)AnalyzeBlockLevelReuse<blockSize, blkIdx>,
                           IARG_ADDRINT, prevMemBlock,
                           IARG_UINT32, numInsInMemBlock,
                           IARG_THREAD_ID, IARG_END);
            numInsInMemBlock = 1;
            prevMemBlock = insEndMemBlock;
        }

        INS_InsertCall(BBL_InsTail(bbl),
                       IPOINT_BEFORE, (AFUNPTR)AnalyzeBlockLevelReuse<blockSize, blkIdx>,
                       IARG_ADDRINT, prevMemBlock,
                       IARG_UINT32, numInsInMemBlock,
                       IARG_THREAD_ID, IARG_END);

    }
}

//instrument the trace, count the number of ins in the trace and instrument each BBL
static void InstrumentTrace(TRACE trace, void* f) {
    InstrumentInsLevelReuse(trace, f);

    // Not using a loop yet, although it can be done. I want to write it such that porting to non c++11 is not too hard.
    InstrumentMemBlockLevelReuse<BLK_INFO[0].blkSize, 0>(trace, f);
    InstrumentMemBlockLevelReuse<BLK_INFO[1].blkSize, 1>(trace, f);
    InstrumentMemBlockLevelReuse<BLK_INFO[2].blkSize, 2>(trace, f);
}

// On each Unload of a loaded image, the accummulated redundancy information is dumped
static VOID ImageUnload(IMG img, VOID* v) {
    fprintf(gTraceFile, "\n TODO .. Multi-threading is not well supported.");
    THREADID  threadid =  PIN_ThreadId();
    fprintf(gTraceFile, "\nUnloading %s", IMG_Name(img).c_str());
    PIN_LockClient();
    PIN_UnlockClient();
}

static void DumpHisto(uint64_t * histo, uint64_t footprint, string key1, string key2){
    double total = 0;
    for(int i = 0; i < MAX_REUSE_DISTANCE_BINS; i++) {
        total += histo[i];
    }
    for(int i = 0; i < MAX_REUSE_DISTANCE_BINS; i++) {
        fprintf(gTraceFile, "\n %2d %e (%.2lf%%)", i, (double) histo[i], histo[i]/total*100);
    }
    fflush(gTraceFile);

    json j;
    j["Source"] = key1;
    j["Metric"] = key2;
    for(int i = 0; i < MAX_REUSE_DISTANCE_BINS; i++) {
	j["raw"].push_back(histo[i]);
	j["relative"].push_back(histo[i]*1.0/total);
    }
    j["Footprint"] = footprint;
    gJSONFile << j.dump(4) << "\n";
}

static VOID ThreadFiniFunc(THREADID threadid, const CONTEXT *ctxt, INT32 code, VOID *v) {
    InsReuseThreadData* tData = ClientGetTLS(threadid);

    // Take the good old lock (don't expect much contention unless threads are created and destroyed as if there is no tomorrow)
    PIN_GetLock(&GLOBAL_STATS.lock, threadid);

    GLOBAL_STATS.numInsExecuted += tData->numInsExecuted;
    // can't add the footprint because it is not additive (non unique)
    for(int i = 0; i < MAX_REUSE_DISTANCE_BINS; i++) {
        GLOBAL_STATS.insReuseHisto[i] += tData->insReuseHisto[i];
    }
    fprintf(gTraceFile, "\nTID %d instruction-reuse histo (ins footprint = %e)", threadid, (double)tData->footprint);

    DumpHisto(tData->insReuseHisto, tData->footprint,"TID " + to_string(threadid), "InsReuse");

    for(int j = 0; j < NUM_BLOCKS; j++) {
        GLOBAL_STATS.blockData[j].numBlocksCounter += tData->blockData[j].numBlocksCounter;
	// can't add the footprint because it is not additive (non unique)
        for(int i = 0; i < MAX_REUSE_DISTANCE_BINS; i++) {
            GLOBAL_STATS.blockData[j].reuseHisto[i] += tData->blockData[j].reuseHisto[i];
        }
        fprintf(gTraceFile, "\nTID %d %s histo (%zu byte blks footprint = %e)", threadid, BLK_INFO[j].blkDescription, BLK_INFO[j].blkSize, (double) tData->blockData[j].footprint);

        DumpHisto(tData->blockData[j].reuseHisto, tData->blockData[j].footprint, "TID " + to_string(threadid), BLK_INFO[j].blkDescription);
    }

    // release the lock
    PIN_ReleaseLock(&GLOBAL_STATS.lock);
}

/*
 static void PrintStats() {
 struct rusage rusage;
 getrusage(RUSAGE_SELF, &rusage);
 size_t peakRSS =  (size_t)(rusage.ru_maxrss);
 }

 */

static VOID FiniFunc(INT32 code, VOID *v) {

    // Peak memory
    struct rusage rusage;
    getrusage(RUSAGE_SELF, &rusage);
    size_t peakRSS =  (size_t)(rusage.ru_maxrss);
    struct timeval ut = rusage.ru_utime;
    struct timeval st = rusage.ru_stime;

    THREADID  threadId =  PIN_ThreadId();
    InsReuseThreadData* tData = ClientGetTLS(threadId);
    fprintf(gTraceFile, "\nWhole program instruction-reuse histo ((ins footprint = %e)", (double)GLOBAL_STATS.footprint);
    DumpHisto(GLOBAL_STATS.insReuseHisto, GLOBAL_STATS.footprint /* basically 0 */, "Whole program", "InsReuse");
    for(int i = 0; i < NUM_BLOCKS; i++) {
        fprintf(gTraceFile, "\nWhole program %s histo (%zu byte blks footprint = %e)", BLK_INFO[i].blkDescription, BLK_INFO[i].blkSize, (double)GLOBAL_STATS.blockData[i].footprint);
	DumpHisto(GLOBAL_STATS.blockData[i].reuseHisto, GLOBAL_STATS.blockData[i].footprint /* basically 0*/, "Whole program", BLK_INFO[i].blkDescription);

    }

    fprintf(gTraceFile, "\n ------- \n");
    fprintf(gTraceFile, "\nutime: %lu", ut.tv_sec * 1000000 + ut.tv_usec);
    fprintf(gTraceFile, "\nstime: %lu", st.tv_sec * 1000000 + st.tv_usec);
    fprintf(gTraceFile, "\nRSS: %zu", peakRSS);
    fprintf(gTraceFile, "\n EOF");

    json j;
    j["utime"] = ut.tv_sec * 1000000 + ut.tv_usec;
    j["stime"] = st.tv_sec * 1000000 + st.tv_usec;
    j["RSS"] = peakRSS;
    gJSONFile << j.dump(4) << "\n";
    gJSONFile.close();
    fclose(gTraceFile);
}

static void InitThreadData(InsReuseThreadData* tdata){
    tdata->bytesLoad = 0;
    tdata->sampleFlag = true;
    tdata->numIns = 0;

    for (int i = 0; i < NUM_BLOCKS; i++) {
        tdata->blockData[i].numBlocksCounter = 0x42; // the start of clock
        tdata->blockData[i].footprint = 0; // the start of clock
        tdata->blockData[i].prevBlock = NULL; // last accessed block
        memset(tdata->blockData[i].reuseHisto, 0, sizeof(uint64_t) * MAX_REUSE_DISTANCE_BINS);
    }

    tdata->numInsExecuted = 0x42; // the start of clock
    tdata->footprint = 0x0; // the start of clock
    memset(tdata->insReuseHisto, 0, sizeof(uint64_t) * MAX_REUSE_DISTANCE_BINS);
}

static VOID ThreadStart(THREADID threadid, CONTEXT* ctxt, INT32 flags, VOID* v) {
    InsReuseThreadData* tdata =  new InsReuseThreadData();
    InitThreadData(tdata);
    //    __sync_fetch_and_add(&gClientNumThreads, 1);
#ifdef MULTI_THREADED
    PIN_SetThreadData(client_tls_key, tdata, threadid);
#else
    gSingleThreadedTData = tdata;
#endif
}

int main(int argc, char* argv[]) {
    // Initialize PIN
    if(PIN_Init(argc, argv))
        return Usage();

    // Initialize Symbols, we need them to report functions and lines
    PIN_InitSymbols();

    // Init Client
    ClientInit(argc, argv);
    // Intialize CCTLib
    //PinCCTLibInit(INTERESTING_INS_ALL, gTraceFile, InstrumentInsCallback, 0);

    // Obtain  a key for TLS storage.
    client_tls_key = PIN_CreateThreadDataKey(0 /*TODO have a destructir*/);
    // Register ThreadStart to be called when a thread starts.
    PIN_AddThreadStartFunction(ThreadStart, 0);
    // fini function for post-mortem analysis
    PIN_AddThreadFiniFunction(ThreadFiniFunc, 0);
    PIN_AddFiniFunction(FiniFunc, 0);

    TRACE_AddInstrumentFunction(InstrumentTrace, 0);

    // Register ImageUnload to be called when an image is unloaded
    IMG_AddUnloadFunction(ImageUnload, 0);

    // Launch program now
    PIN_StartProgram();
    return 0;
}

max_stars_repo_name: CCTLib/CCTLib

max_issues_repo_name: CCTLib/CCTLib

max_forks_repo_name: CCTLib/CCTLib


text:
#ifndef G4_VERSION_HH
#define G4_VERSION_HH

#include <boost/version.hpp>
#include <G4Version.hh>

#ifdef G4UI_USE_QT
    #include <QtGlobal>
#endif

inline void printVersionInfo()
{
    std::cout << "g4 app, version 0.0.0" << std::endl;
    std::cout << "Built: " << __DATE__ << " " << __TIME__ << std::endl;
    std::cout << "Compiled with: " << std::endl;
    std::cout << "- Geant4: " << G4Version << " " << G4Date << std::endl;
    #ifdef __GNUC__
        std::cout << "- gcc: " << __GNUC__ << "." << __GNUC_MINOR__ << "." << __GNUC_PATCHLEVEL__ << std::endl;
    #endif
    #ifdef G4UI_USE_QT
        std::cout << "- Qt: " << qVersion() << std::endl;
    #endif
    std::cout << "- Boost: "
      << BOOST_VERSION / 100000     << "."  // major version
      << BOOST_VERSION / 100 % 1000 << "."  // minor version
      << BOOST_VERSION % 100                // patch level
      << std::endl;
}

#endif

ext: hh


text:
//  Unit test for boost::any.
//
//  See http://www.boost.org for most recent version, including documentation.
//
//  Copyright Antony Polukhin, 2013-2014.
//
//  Distributed under the Boost
//  Software License, Version 1.0. (See accompanying file
//  LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt).

#include <cstdlib>
#include <string>
#include <utility>

#include <boost/any.hpp>
#include "test.hpp"
#include <boost/move/move.hpp>

#ifdef BOOST_NO_CXX11_RVALUE_REFERENCES

int main()
{
    return EXIT_SUCCESS;
}

#else

namespace any_tests
{
    typedef test<const char *, void (*)()> test_case;
    typedef const test_case * test_case_iterator;

    extern const test_case_iterator begin, end;
}

int main()
{
    using namespace any_tests;
    tester<test_case_iterator> test_suite(begin, end);
    return test_suite() ? EXIT_SUCCESS : EXIT_FAILURE;
}

namespace any_tests // test suite
{
    void test_move_construction();
    void test_move_assignment();
    void test_copy_construction();
    void test_copy_assignment();

    void test_move_construction_from_value();
    void test_move_assignment_from_value();
    void test_copy_construction_from_value();
    void test_copy_assignment_from_value();
    void test_construction_from_const_any_rv();
    void test_cast_to_rv();

    const test_case test_cases[] =
    {
        { "move construction of any",             test_move_construction      },
        { "move assignment of any",               test_move_assignment        },
        { "copy construction of any",             test_copy_construction      },
        { "copy assignment of any",               test_copy_assignment        },

        { "move construction from value",         test_move_construction_from_value },
        { "move assignment from value",           test_move_assignment_from_value  },
        { "copy construction from value",         test_copy_construction_from_value },
        { "copy assignment from value",           test_copy_assignment_from_value },
        { "constructing from const any&&",        test_construction_from_const_any_rv },
        { "casting to rvalue reference",          test_cast_to_rv }
    };

    const test_case_iterator begin = test_cases;
    const test_case_iterator end =
        test_cases + (sizeof test_cases / sizeof *test_cases);

    class move_copy_conting_class {
    public:
        static unsigned int moves_count;
        static unsigned int copy_count;

        move_copy_conting_class(){}
        move_copy_conting_class(move_copy_conting_class&& /*param*/) {
            ++ moves_count;
        }

        move_copy_conting_class& operator=(move_copy_conting_class&& /*param*/) {
            ++ moves_count;
            return *this;
        }

        move_copy_conting_class(const move_copy_conting_class&) {
            ++ copy_count;
        }
        move_copy_conting_class& operator=(const move_copy_conting_class& /*param*/) {
            ++ copy_count;
            return *this;
        }
    };

    unsigned int move_copy_conting_class::moves_count = 0;
    unsigned int move_copy_conting_class::copy_count = 0;
}

namespace any_tests // test definitions
{
    using namespace boost;

    void test_move_construction()
    {
        any value0 = move_copy_conting_class();
        move_copy_conting_class::copy_count = 0;
        move_copy_conting_class::moves_count = 0;
        any value(boost::move(value0));

        check(value0.empty(), "moved away value is empty");
        check_false(value.empty(), "empty");
        check_equal(value.type(), typeindex::type_id<move_copy_conting_class>(), "type");
        check_non_null(any_cast<move_copy_conting_class>(&value), "any_cast<move_copy_conting_class>");
        check_equal(
            move_copy_conting_class::copy_count, 0u,
            "checking copy counts");
        check_equal(
            move_copy_conting_class::moves_count, 0u,
            "checking move counts");
    }

    void test_move_assignment()
    {
        any value0 = move_copy_conting_class();
        any value = move_copy_conting_class();
        move_copy_conting_class::copy_count = 0;
        move_copy_conting_class::moves_count = 0;
        value = boost::move(value0);

        check(value0.empty(), "moved away is empty");
        check_false(value.empty(), "empty");
        check_equal(value.type(), typeindex::type_id<move_copy_conting_class>(), "type");
        check_non_null(any_cast<move_copy_conting_class>(&value), "any_cast<move_copy_conting_class>");
        check_equal(
            move_copy_conting_class::copy_count, 0u,
            "checking copy counts");
        check_equal(
            move_copy_conting_class::moves_count, 0u,
            "checking move counts");
    }

    void test_copy_construction()
    {
        any value0 = move_copy_conting_class();
        move_copy_conting_class::copy_count = 0;
        move_copy_conting_class::moves_count = 0;
        any value(value0);

        check_false(value0.empty(), "copyed value is not empty");
        check_false(value.empty(), "empty");
        check_equal(value.type(), typeindex::type_id<move_copy_conting_class>(), "type");
        check_non_null(any_cast<move_copy_conting_class>(&value), "any_cast<move_copy_conting_class>");
        check_equal(
            move_copy_conting_class::copy_count, 1u,
            "checking copy counts");
        check_equal(
            move_copy_conting_class::moves_count, 0u,
            "checking move counts");
    }

    void test_copy_assignment()
    {
        any value0 = move_copy_conting_class();
        any value = move_copy_conting_class();
        move_copy_conting_class::copy_count = 0;
        move_copy_conting_class::moves_count = 0;
        value = value0;

        check_false(value0.empty(), "copyied value is not empty");
        check_false(value.empty(), "empty");
        check_equal(value.type(), typeindex::type_id<move_copy_conting_class>(), "type");
        check_non_null(any_cast<move_copy_conting_class>(&value), "any_cast<move_copy_conting_class>");
        check_equal(
            move_copy_conting_class::copy_count, 1u,
            "checking copy counts");
        check_equal(
            move_copy_conting_class::moves_count, 0u,
            "checking move counts");
    }

     void test_move_construction_from_value()
    {
        move_copy_conting_class value0;
        move_copy_conting_class::copy_count = 0;
        move_copy_conting_class::moves_count = 0;
#ifndef BOOST_NO_TEMPLATE_PARTIAL_SPECIALIZATION
        any value(boost::move(value0));
#else
        any value(value0);
#endif

        check_false(value.empty(), "empty");
        check_equal(value.type(), typeindex::type_id<move_copy_conting_class>(), "type");
        check_non_null(any_cast<move_copy_conting_class>(&value), "any_cast<move_copy_conting_class>");

#ifndef BOOST_NO_TEMPLATE_PARTIAL_SPECIALIZATION
        check_equal(
            move_copy_conting_class::copy_count, 0u,
            "checking copy counts");
        check_equal(
            move_copy_conting_class::moves_count, 1u,
            "checking move counts");
#endif

     }

    void test_move_assignment_from_value()
    {
        move_copy_conting_class value0;
        any value;
        move_copy_conting_class::copy_count = 0;
        move_copy_conting_class::moves_count = 0;
#ifndef BOOST_NO_TEMPLATE_PARTIAL_SPECIALIZATION
        value = boost::move(value0);
#else
        value = value0;
#endif

        check_false(value.empty(), "empty");
        check_equal(value.type(), typeindex::type_id<move_copy_conting_class>(), "type");
        check_non_null(any_cast<move_copy_conting_class>(&value), "any_cast<move_copy_conting_class>");

#ifndef BOOST_NO_TEMPLATE_PARTIAL_SPECIALIZATION
        check_equal(
            move_copy_conting_class::copy_count, 0u,
            "checking copy counts");
        check_equal(
            move_copy_conting_class::moves_count, 1u,
            "checking move counts");
#endif

    }

    void test_copy_construction_from_value()
    {
        move_copy_conting_class value0;
        move_copy_conting_class::copy_count = 0;
        move_copy_conting_class::moves_count = 0;
        any value(value0);

        check_false(value.empty(), "empty");
        check_equal(value.type(), typeindex::type_id<move_copy_conting_class>(), "type");
        check_non_null(any_cast<move_copy_conting_class>(&value), "any_cast<move_copy_conting_class>");

        check_equal(
            move_copy_conting_class::copy_count, 1u,
            "checking copy counts");
        check_equal(
            move_copy_conting_class::moves_count, 0u,
            "checking move counts");
     }

    void test_copy_assignment_from_value()
    {
        move_copy_conting_class value0;
        any value;
        move_copy_conting_class::copy_count = 0;
        move_copy_conting_class::moves_count = 0;
        value = value0;

        check_false(value.empty(), "empty");
        check_equal(value.type(), typeindex::type_id<move_copy_conting_class>(), "type");
        check_non_null(any_cast<move_copy_conting_class>(&value), "any_cast<move_copy_conting_class>");

        check_equal(
            move_copy_conting_class::copy_count, 1u,
            "checking copy counts");
        check_equal(
            move_copy_conting_class::moves_count, 0u,
            "checking move counts");
    }

    const any helper_method() {
        return true;
    }

    const bool helper_method1() {
        return true;
    }

    void test_construction_from_const_any_rv()
    {
        any values[] = {helper_method(), helper_method1() };
        (void)values;
    }

    void test_cast_to_rv()
    {
        move_copy_conting_class value0;
        any value;
        value = value0;
        move_copy_conting_class::copy_count = 0;
        move_copy_conting_class::moves_count = 0;

        move_copy_conting_class value1 = any_cast<move_copy_conting_class&&>(value);

        check_equal(
            move_copy_conting_class::copy_count, 0u,
            "checking copy counts");
        check_equal(
            move_copy_conting_class::moves_count, 1u,
            "checking move counts");
        (void)value1;
/* Following code shall fail to compile
        const any cvalue = value0;
        move_copy_conting_class::copy_count = 0;
        move_copy_conting_class::moves_count = 0;

        move_copy_conting_class value2 = any_cast<move_copy_conting_class&&>(cvalue);

        check_equal(
            move_copy_conting_class::copy_count, 1u,
            "checking copy counts");
        check_equal(
            move_copy_conting_class::moves_count, 0u,
            "checking move counts");
        (void)value2;
*/
    }

}

#endif


text:
#include <cstdlib>
#include <iostream>
#include <string>
#include <boost/timer/timer.hpp>
#include <boost/thread/tss.hpp>
#include <boost/python.hpp>

#include "common/god.h"
#include "common/logging.h"
#include "common/threadpool.h"
#include "common/search.h"
#include "common/printer.h"
#include "common/sentence.h"
#include "common/sentences.h"
#include "common/exception.h"
#include "common/translation_task.h"

using namespace amunmt;
using namespace std;

God god_;

void init(const std::string& options) {
  god_.Init(options);
}

boost::python::list translate(boost::python::list& in)
{
  size_t miniSize = god_.Get<size_t>("mini-batch");
  size_t maxiSize = god_.Get<size_t>("maxi-batch");
  int miniWords = god_.Get<int>("mini-batch-words");

  std::vector<std::future< std::shared_ptr<Histories> >> results;
  SentencesPtr maxiBatch(new Sentences());

  for(int lineNum = 0; lineNum < boost::python::len(in); ++lineNum) {
    std::string line = boost::python::extract<std::string>(boost::python::object(in[lineNum]));
    //cerr << "line=" << line << endl;

    maxiBatch->push_back(SentencePtr(new Sentence(god_, lineNum, line)));

    if (maxiBatch->size() >= maxiSize) {

      maxiBatch->SortByLength();
      while (maxiBatch->size()) {
        SentencesPtr miniBatch = maxiBatch->NextMiniBatch(miniSize, miniWords);

        results.emplace_back(
          god_.GetThreadPool().enqueue(
              [miniBatch]{ return TranslationTask(::god_, miniBatch); }
              )
        );
      }

      maxiBatch.reset(new Sentences());
    }
  }

  // last batch
  if (maxiBatch->size()) {
    maxiBatch->SortByLength();
    while (maxiBatch->size()) {
      SentencesPtr miniBatch = maxiBatch->NextMiniBatch(miniSize, miniWords);
      results.emplace_back(
        god_.GetThreadPool().enqueue(
            [miniBatch]{ return TranslationTask(::god_, miniBatch); }
            )
      );
    }
  }

  // resort batch into line number order
  Histories allHistories;
  for (auto&& result : results) {
    std::shared_ptr<Histories> histories = result.get();
    allHistories.Append(*histories);
  }
  allHistories.SortByLineNum();

  // output
  std::stringstream ss;
  Printer(god_, allHistories, ss);
  string str = ss.str();
  boost::python::list output;
  output.append(str);

  return output;
}

boost::python::list translate_single(boost::python::list& in)
{
  boost::timer::cpu_timer timer2;

  size_t miniSize = god_.Get<size_t>("mini-batch");
  size_t maxiSize = god_.Get<size_t>("maxi-batch");
  int miniWords = god_.Get<int>("mini-batch-words");
  //LOG(info)->info("DEBUG.....");
  std::vector<std::future< std::shared_ptr<Histories> >> results;
  //SentencesPtr maxiBatch(new Sentences());

  std::string line = boost::python::extract<std::string>(boost::python::object(in[0]));

  SentencesPtr miniBatch(new Sentences());
  miniBatch->push_back(SentencePtr(new Sentence(god_, 0, line)));

  /*results.emplace_back(
    god_.GetThreadPool().enqueue(
        [miniBatch]{ return TranslationTask(::god_, miniBatch); }
        )
  );*/
  //God *GodPtr = &god_;
  //std::shared_ptr<Histories> tmpHistories = TranslationTask(god_, miniBatch);
  Search& search = god_.GetSearch();
  std::shared_ptr<Histories> tmpHistories = search.Translate(*miniBatch);

  LOG(info)->info("Translated took {}", timer2.format(3, "%ws"));

  // last batch
  /*if (maxiBatch->size()) {
    maxiBatch->SortByLength();
    while (maxiBatch->size()) {
      SentencesPtr miniBatch = maxiBatch->NextMiniBatch(miniSize, miniWords);
      results.emplace_back(
        god_.GetThreadPool().enqueue(
            [miniBatch]{ return TranslationTask(::god_, miniBatch); }
            )
      );
    }
  }*/

  // resort batch into line number order
  Histories allHistories;
  /*for (auto&& result : results) {
    std::shared_ptr<Histories> histories = result.get();
    allHistories.Append(*histories);
  }*/
  allHistories.Append(*tmpHistories);
  //allHistories.SortByLineNum();

  // output
  std::stringstream ss;
  Printer(god_, allHistories, ss);
  string str = ss.str();
  boost::python::list output;
  //boost::python::list abc;
  output.append(str);
  //debug
  //string str_time = timer2.format(3, "%ws");
  //output.append(str_time);

  //output.append("FUCK YOU");

  return output;
}

BOOST_PYTHON_MODULE(libamunmt)
{
  boost::python::def("init", init);
  boost::python::def("translate", translate);
  boost::python::def("translate_single", translate_single);
}


text:
/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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.
 */

#include "MQClientAPIImpl.h"
#include <assert.h>
#include <boost/filesystem.hpp>
#include <boost/filesystem/fstream.hpp>
#include <fstream>
#include "CommunicationMode.h"
#include "Logging.h"
#include "MQDecoder.h"
#include "PullResultExt.h"

namespace rocketmq {
//<!************************************************************************
MQClientAPIImpl::MQClientAPIImpl(
    const string& mqClientId, ClientRemotingProcessor* clientRemotingProcessor,
    int pullThreadNum, uint64_t tcpConnectTimeout,
    uint64_t tcpTransportTryLockTimeout, string unitName)
    : m_firstFetchNameSrv(true), m_mqClientId(mqClientId) {
  m_pRemotingClient.reset(new TcpRemotingClient(
      pullThreadNum, tcpConnectTimeout, tcpTransportTryLockTimeout));
  m_pRemotingClient->registerProcessor(CHECK_TRANSACTION_STATE,
                                       clientRemotingProcessor);
  m_pRemotingClient->registerProcessor(RESET_CONSUMER_CLIENT_OFFSET,
                                       clientRemotingProcessor);
  m_pRemotingClient->registerProcessor(GET_CONSUMER_STATUS_FROM_CLIENT,
                                       clientRemotingProcessor);
  m_pRemotingClient->registerProcessor(GET_CONSUMER_RUNNING_INFO,
                                       clientRemotingProcessor);
  m_pRemotingClient->registerProcessor(NOTIFY_CONSUMER_IDS_CHANGED,
                                       clientRemotingProcessor);
  m_pRemotingClient->registerProcessor(CONSUME_MESSAGE_DIRECTLY,
                                       clientRemotingProcessor);

  m_topAddressing.reset(new TopAddressing(unitName));
}

MQClientAPIImpl::~MQClientAPIImpl() {
  m_pRemotingClient = NULL;
  m_topAddressing = NULL;
}

void MQClientAPIImpl::stopAllTcpTransportThread() {
  m_pRemotingClient->stopAllTcpTransportThread();
}

bool MQClientAPIImpl::writeDataToFile(string filename, string data,
                                      bool isSync) {
  if (data.size() == 0) return false;

  FILE* pFd = fopen(filename.c_str(), "w+");
  if (NULL == pFd) {
    LOG_ERROR("fopen failed, filename:%s", filename.c_str());
    return false;
  }

  int byte_write = 0;
  int byte_left = data.size();
  const char* pData = data.c_str();
  while (byte_left > 0) {
    byte_write = fwrite(pData, sizeof(char), byte_left, pFd);
    if (byte_write == byte_left) {
      if (ferror(pFd)) {
        LOG_ERROR("write data fail, data len:" SIZET_FMT ", file:%s, msg:%s",
                  data.size(), filename.c_str(), strerror(errno));
        fclose(pFd);
        return false;
      }
    }
    byte_left -= byte_write;
    pData += byte_write;
  }
  pData = NULL;

  if (isSync) {
    LOG_INFO("fsync with filename:%s", filename.c_str());
    fflush(pFd);
  }
  fclose(pFd);

  return true;
}

string MQClientAPIImpl::fetchNameServerAddr(const string& NSDomain) {
  try {
    string homeDir(UtilAll::getHomeDirectory());
    string storePath = homeDir + "/logs/rocketmq-cpp/snapshot";

    boost::filesystem::path dir(storePath);
    boost::system::error_code ec;
    if (!boost::filesystem::exists(dir, ec)) {
      if (!boost::filesystem::create_directory(dir, ec)) {
        LOG_ERROR("create data dir:%s error", storePath.c_str());
        return "";
      }
    }
    string file(storePath);
    string fileBak(storePath);
    vector<string> ret_;
    int retSize = UtilAll::Split(ret_, m_mqClientId, "@");
    if (retSize == 2) {
      file.append("/nameserver_addr-").append(ret_[retSize - 1]);
    } else {
      LOG_ERROR("split mqClientId:%s fail", m_mqClientId.c_str());
      file.append("/nameserver_addr-DEFAULT");
    }
    boost::filesystem::path snapshot_file(file);
    fileBak.append("/nameserver_addr.bak");
    const string addrs = m_topAddressing->fetchNSAddr(NSDomain);
    if (addrs.empty()) {
      if (m_nameSrvAddr.empty()) {
        LOG_INFO("Load the name server snapshot local file:%s", file.c_str());
        if (boost::filesystem::exists(snapshot_file)) {
          ifstream snapshot_file(file, ios::binary);
          istreambuf_iterator<char> beg(snapshot_file), end;
          string filecontent(beg, end);
          updateNameServerAddr(filecontent);
          m_nameSrvAddr = filecontent;
        } else {
          LOG_WARN("The name server snapshot local file not exists");
        }
      }
    } else {
      if (m_firstFetchNameSrv == true) {
        // it is the first time, so need to create the name server snapshot
        // local file
        m_firstFetchNameSrv = false;
      }
      if (addrs.compare(m_nameSrvAddr) != 0) {
        LOG_INFO("name server address changed, old: %s, new: %s",
                 m_nameSrvAddr.c_str(), addrs.c_str());
        updateNameServerAddr(addrs);
        m_nameSrvAddr = addrs;
      } else {
        if (!m_firstFetchNameSrv) return m_nameSrvAddr;
      }
      // update the snapshot local file if nameSrv changes or
      // m_firstFetchNameSrv==true
      if (writeDataToFile(fileBak, addrs, true)) {
        if (!UtilAll::ReplaceFile(fileBak, file))
          LOG_ERROR("could not rename bak file:%s", strerror(errno));
      }
    }

    if (!boost::filesystem::exists(snapshot_file)) {
      // the name server snapshot local file maybe deleted by force, create it
      if (writeDataToFile(fileBak, m_nameSrvAddr, true)) {
        if (!UtilAll::ReplaceFile(fileBak, file))
          LOG_ERROR("could not rename bak file:%s", strerror(errno));
      }
    }
  } catch (...) {
  }
  return m_nameSrvAddr;
}

void MQClientAPIImpl::updateNameServerAddr(const string& addrs) {
  if (m_pRemotingClient != NULL)
    m_pRemotingClient->updateNameServerAddressList(addrs);
}

void MQClientAPIImpl::callSignatureBeforeRequest(
    const string& addr, RemotingCommand& request,
    const SessionCredentials& session_credentials) {
  ClientRPCHook rpcHook(session_credentials);
  rpcHook.doBeforeRequest(addr, request);
}

// Note: all request rules: throw exception if got broker error response,
// exclude getTopicRouteInfoFromNameServer and unregisterClient
void MQClientAPIImpl::createTopic(
    const string& addr, const string& defaultTopic, TopicConfig topicConfig,
    const SessionCredentials& sessionCredentials) {
  string topicWithProjectGroup = topicConfig.getTopicName();
  CreateTopicRequestHeader* requestHeader = new CreateTopicRequestHeader();
  requestHeader->topic = (topicWithProjectGroup);
  requestHeader->defaultTopic = (defaultTopic);
  requestHeader->readQueueNums = (topicConfig.getReadQueueNums());
  requestHeader->writeQueueNums = (topicConfig.getWriteQueueNums());
  requestHeader->perm = (topicConfig.getPerm());
  requestHeader->topicFilterType = (topicConfig.getTopicFilterType());

  RemotingCommand request(UPDATE_AND_CREATE_TOPIC, requestHeader);
  callSignatureBeforeRequest(addr, request, sessionCredentials);
  request.Encode();

  unique_ptr<RemotingCommand> response(
      m_pRemotingClient->invokeSync(addr, request));

  if (response) {
    switch (response->getCode()) {
      case SUCCESS_VALUE:
        return;
      default:
        break;
    }
    THROW_MQEXCEPTION(MQBrokerException, response->getRemark(),
                      response->getCode());
  }
  THROW_MQEXCEPTION(MQBrokerException, "response is null", -1);
}

SendResult MQClientAPIImpl::sendMessage(
    const string& addr, const string& brokerName, const MQMessage& msg,
    SendMessageRequestHeader* pRequestHeader, int timeoutMillis, int maxRetrySendTimes,
    int communicationMode, SendCallback* pSendCallback,
    const SessionCredentials& sessionCredentials) {
  RemotingCommand request(SEND_MESSAGE, pRequestHeader);
  string body = msg.getBody();
  request.SetBody(body.c_str(), body.length());
  request.setMsgBody(body);
  callSignatureBeforeRequest(addr, request, sessionCredentials);
  request.Encode();

  switch (communicationMode) {
    case ComMode_ONEWAY:
      m_pRemotingClient->invokeOneway(addr, request);
      break;
    case ComMode_ASYNC:
      sendMessageAsync(addr, brokerName, msg, request, pSendCallback, timeoutMillis, maxRetrySendTimes, 1);
      break;
    case ComMode_SYNC:
      return sendMessageSync(addr, brokerName, msg, request, timeoutMillis);
    default:
      break;
  }
  return SendResult();
}

void MQClientAPIImpl::sendHearbeat(
    const string& addr, HeartbeatData* pHeartbeatData,
    const SessionCredentials& sessionCredentials) {
  RemotingCommand request(HEART_BEAT, NULL);

  string body;
  pHeartbeatData->Encode(body);
  request.SetBody(body.data(), body.length());
  request.setMsgBody(body);
  callSignatureBeforeRequest(addr, request, sessionCredentials);
  request.Encode();

  if (m_pRemotingClient->invokeHeartBeat(addr, request)) {
    LOG_INFO("sendheartbeat to broker:%s success", addr.c_str());
  }
}

void MQClientAPIImpl::unregisterClient(
    const string& addr, const string& clientID, const string& producerGroup,
    const string& consumerGroup, const SessionCredentials& sessionCredentials) {
  LOG_INFO("unregisterClient to broker:%s", addr.c_str());
  RemotingCommand request(UNREGISTER_CLIENT,
                          new UnregisterClientRequestHeader(
                              clientID, producerGroup, consumerGroup));
  callSignatureBeforeRequest(addr, request, sessionCredentials);
  request.Encode();

  unique_ptr<RemotingCommand> response(
      m_pRemotingClient->invokeSync(addr, request));

  if (response) {
    switch (response->getCode()) {
      case SUCCESS_VALUE:
        LOG_INFO("unregisterClient to:%s success", addr.c_str());
        return;
      default:
        break;
    }
    LOG_WARN("unregisterClient fail:%s,%d", response->getRemark().c_str(),
             response->getCode());
  }
}

// return NULL if got no response or error response
TopicRouteData* MQClientAPIImpl::getTopicRouteInfoFromNameServer(
    const string& topic, int timeoutMillis,
    const SessionCredentials& sessionCredentials) {
  RemotingCommand request(GET_ROUTEINTO_BY_TOPIC,
                          new GetRouteInfoRequestHeader(topic));
  callSignatureBeforeRequest("", request, sessionCredentials);
  request.Encode();

  unique_ptr<RemotingCommand> pResponse(
      m_pRemotingClient->invokeSync("", request, timeoutMillis));

  if (pResponse != NULL) {
    if (((*(pResponse->GetBody())).getSize() == 0) ||
        ((*(pResponse->GetBody())).getData() != NULL)) {
      switch (pResponse->getCode()) {
        case SUCCESS_VALUE: {
          const MemoryBlock* pbody = pResponse->GetBody();
          if (pbody->getSize()) {
            TopicRouteData* topicRoute = TopicRouteData::Decode(pbody);
            return topicRoute;
          }
        }
        case TOPIC_NOT_EXIST: {
          return NULL;
        }
        default:
          break;
      }
      LOG_WARN("%s,%d", pResponse->getRemark().c_str(), pResponse->getCode());
      return NULL;
    }
  }
  return NULL;
}

TopicList* MQClientAPIImpl::getTopicListFromNameServer(
    const SessionCredentials& sessionCredentials) {
  RemotingCommand request(GET_ALL_TOPIC_LIST_FROM_NAMESERVER, NULL);
  callSignatureBeforeRequest("", request, sessionCredentials);
  request.Encode();

  unique_ptr<RemotingCommand> pResponse(
      m_pRemotingClient->invokeSync("", request));
  if (pResponse != NULL) {
    if (((*(pResponse->GetBody())).getSize() == 0) ||
        ((*(pResponse->GetBody())).getData() != NULL)) {
      switch (pResponse->getCode()) {
        case SUCCESS_VALUE: {
          const MemoryBlock* pbody = pResponse->GetBody();
          if (pbody->getSize()) {
            TopicList* topicList = TopicList::Decode(pbody);
            return topicList;
          }
        }
        default:
          break;
      }

      THROW_MQEXCEPTION(MQClientException, pResponse->getRemark(),
                        pResponse->getCode());
    }
  }
  return NULL;
}

int MQClientAPIImpl::wipeWritePermOfBroker(const string& namesrvAddr,
                                           const string& brokerName,
                                           int timeoutMillis) {
  return 0;
}

void MQClientAPIImpl::deleteTopicInBroker(const string& addr,
                                          const string& topic,
                                          int timeoutMillis) {}

void MQClientAPIImpl::deleteTopicInNameServer(const string& addr,
                                              const string& topic,
                                              int timeoutMillis) {}

void MQClientAPIImpl::deleteSubscriptionGroup(const string& addr,
                                              const string& groupName,
                                              int timeoutMillis) {}

string MQClientAPIImpl::getKVConfigByValue(const string& projectNamespace,
                                           const string& projectGroup,
                                           int timeoutMillis) {
  return "";
}

KVTable MQClientAPIImpl::getKVListByNamespace(const string& projectNamespace,
                                              int timeoutMillis) {
  return KVTable();
}

void MQClientAPIImpl::deleteKVConfigByValue(const string& projectNamespace,
                                            const string& projectGroup,
                                            int timeoutMillis) {}

SendResult MQClientAPIImpl::sendMessageSync(const string& addr,
                                            const string& brokerName,
                                            const MQMessage& msg,
                                            RemotingCommand& request,
                                            int timeoutMillis) {
  //<!block util response;
  unique_ptr<RemotingCommand> pResponse(
      m_pRemotingClient->invokeSync(addr, request, timeoutMillis));
  if (pResponse != NULL) {
    try {
      LOG_DEBUG("sendMessageSync success:%s to addr:%s,brokername:%s",
                msg.toString().c_str(), addr.c_str(), brokerName.c_str());
      SendResult result = processSendResponse(brokerName, msg, pResponse.get());
      return result;
    } catch (...) {
      LOG_ERROR("send error");
    }
  }
  THROW_MQEXCEPTION(MQClientException, "response is null", -1);
}

void MQClientAPIImpl::sendMessageAsync(const string& addr,
                                       const string& brokerName,
                                       const MQMessage& msg,
                                       RemotingCommand& request,
                                       SendCallback* pSendCallback,
                                       int64 timeoutMilliseconds,
                                       int maxRetryTimes,
                                       int retrySendTimes) {
  int64 begin_time = UtilAll::currentTimeMillis();
  //<!delete in future;
  AsyncCallbackWrap* cbw = new SendCallbackWrap(brokerName, msg, pSendCallback, this);

  LOG_DEBUG("sendMessageAsync request:%s, timeout:%lld, maxRetryTimes:%d retrySendTimes:%d", request.ToString().data(), timeoutMilliseconds, maxRetryTimes, retrySendTimes);

  if (m_pRemotingClient->invokeAsync(addr, request, cbw, timeoutMilliseconds, maxRetryTimes, retrySendTimes) ==
    false) {
    LOG_WARN("invokeAsync failed to addr:%s,topic:%s, timeout:%lld, maxRetryTimes:%d, retrySendTimes:%d",
	  addr.c_str(), msg.getTopic().data(), timeoutMilliseconds, maxRetryTimes, retrySendTimes);
	  //when getTcp return false, need consider retrySendTimes
	  int retry_time = retrySendTimes + 1;
	  int64 time_out = timeoutMilliseconds - (UtilAll::currentTimeMillis() - begin_time);
	  while (retry_time < maxRetryTimes && time_out > 0) {
		  begin_time = UtilAll::currentTimeMillis();
		  if (m_pRemotingClient->invokeAsync(addr, request, cbw, time_out, maxRetryTimes, retry_time) == false) {
		    retry_time += 1;
		    time_out = time_out - (UtilAll::currentTimeMillis() - begin_time);
			  LOG_WARN("invokeAsync retry failed to addr:%s,topic:%s, timeout:%lld, maxRetryTimes:%d, retrySendTimes:%d",
				addr.c_str(), msg.getTopic().data(), time_out, maxRetryTimes, retry_time);
			  continue;
		  } else {
			  return; //invokeAsync success
		  }
	  }

    LOG_ERROR("sendMessageAsync failed to addr:%s,topic:%s, timeout:%lld, maxRetryTimes:%d, retrySendTimes:%d",
	  addr.c_str(), msg.getTopic().data(), time_out, maxRetryTimes, retrySendTimes);

    if (cbw) {
      cbw->onException();
      deleteAndZero(cbw);
    } else {
      THROW_MQEXCEPTION(MQClientException, "sendMessageAsync failed", -1);
    }
  }
}

void MQClientAPIImpl::deleteOpaqueForDropPullRequest(const MQMessageQueue& mq, int opaque) {
    m_pRemotingClient->deleteOpaqueForDropPullRequest(mq, opaque);
}

PullResult* MQClientAPIImpl::pullMessage(
    const string& addr, PullMessageRequestHeader* pRequestHeader,
    int timeoutMillis, int communicationMode, PullCallback* pullCallback,
    void* pArg, const SessionCredentials& sessionCredentials) {
  RemotingCommand request(PULL_MESSAGE, pRequestHeader);
  callSignatureBeforeRequest(addr, request, sessionCredentials);
  request.Encode();

  switch (communicationMode) {
    case ComMode_ONEWAY:
      break;
    case ComMode_ASYNC:
      pullMessageAsync(addr, request, timeoutMillis, pullCallback, pArg);
      break;
    case ComMode_SYNC:
      return pullMessageSync(addr, request, timeoutMillis);
    default:
      break;
  }

  return NULL;
}

void MQClientAPIImpl::pullMessageAsync(const string& addr,
                                       RemotingCommand& request,
                                       int timeoutMillis,
                                       PullCallback* pullCallback, void* pArg) {
  //<!delete in future;
  AsyncCallbackWrap* cbw = new PullCallbackWarp(pullCallback, this, pArg);
  MQMessageQueue mq;
  AsyncArg* pAsyncArg = static_cast<AsyncArg*>(pArg);
  if (pAsyncArg && pAsyncArg->pPullRequest) {
    mq = pAsyncArg->mq;
    pAsyncArg->pPullRequest->setLatestPullRequestOpaque(request.getOpaque());
    LOG_DEBUG("pullMessageAsync set opaque:%d, mq:%s",
        pAsyncArg->pPullRequest->getLatestPullRequestOpaque(),mq.toString().c_str());
  }

  if (m_pRemotingClient->invokeAsync(addr, request, cbw, timeoutMillis) ==
      false) {
    LOG_ERROR("pullMessageAsync failed of addr:%s, opaque:%d, mq:%s", addr.c_str(), request.getOpaque(), mq.toString().data());
    if (pAsyncArg && pAsyncArg->pPullRequest) {
        pAsyncArg->pPullRequest->setLatestPullRequestOpaque(0);
    }
    deleteAndZero(cbw);
    THROW_MQEXCEPTION(MQClientException, "pullMessageAsync failed", -1);
  }
}

PullResult* MQClientAPIImpl::pullMessageSync(const string& addr,
                                             RemotingCommand& request,
                                             int timeoutMillis) {
  unique_ptr<RemotingCommand> pResponse(
      m_pRemotingClient->invokeSync(addr, request, timeoutMillis));
  if (pResponse != NULL) {
    if (((*(pResponse->GetBody())).getSize() == 0) ||
        ((*(pResponse->GetBody())).getData() != NULL)) {
      try {
        PullResult* pullResult =
            processPullResponse(pResponse.get());  // pullMessage will handle
                                                   // exception from
                                                   // processPullResponse
        return pullResult;
      } catch (MQException& e) {
        LOG_ERROR(e.what());
        return NULL;
      }
    }
  }
  return NULL;
}

SendResult MQClientAPIImpl::processSendResponse(const string& brokerName,
                                                const MQMessage& msg,
                                                RemotingCommand* pResponse) {
  SendStatus sendStatus = SEND_OK;
  int res = 0;
  switch (pResponse->getCode()) {
    case FLUSH_DISK_TIMEOUT:
      sendStatus = SEND_FLUSH_DISK_TIMEOUT;
      break;
    case FLUSH_SLAVE_TIMEOUT:
      sendStatus = SEND_FLUSH_SLAVE_TIMEOUT;
      break;
    case SLAVE_NOT_AVAILABLE:
      sendStatus = SEND_SLAVE_NOT_AVAILABLE;
      break;
    case SUCCESS_VALUE:
      sendStatus = SEND_OK;
      break;
    default:
      res = -1;
      break;
  }
  if (res == 0) {
    SendMessageResponseHeader* responseHeader =
        (SendMessageResponseHeader*)pResponse->getCommandHeader();
    MQMessageQueue messageQueue(msg.getTopic(), brokerName,
                                responseHeader->queueId);
	string unique_msgId = msg.getProperty(MQMessage::PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
    return SendResult(sendStatus, unique_msgId, responseHeader->msgId, messageQueue,
                      responseHeader->queueOffset);
  }
  LOG_ERROR("processSendResponse error remark:%s, error code:%d",
            (pResponse->getRemark()).c_str(), pResponse->getCode());
  THROW_MQEXCEPTION(MQClientException, pResponse->getRemark(),
                    pResponse->getCode());
}

PullResult* MQClientAPIImpl::processPullResponse(RemotingCommand* pResponse) {
  PullStatus pullStatus = NO_NEW_MSG;
  switch (pResponse->getCode()) {
    case SUCCESS_VALUE:
      pullStatus = FOUND;
      break;
    case PULL_NOT_FOUND:
      pullStatus = NO_NEW_MSG;
      break;
    case PULL_RETRY_IMMEDIATELY:
      pullStatus = NO_MATCHED_MSG;
      break;
    case PULL_OFFSET_MOVED:
      pullStatus = OFFSET_ILLEGAL;
      break;
    default:
      THROW_MQEXCEPTION(MQBrokerException, pResponse->getRemark(),
                        pResponse->getCode());
      break;
  }

  PullMessageResponseHeader* responseHeader =
      static_cast<PullMessageResponseHeader*>(pResponse->getCommandHeader());

  if (!responseHeader) {
    LOG_ERROR("processPullResponse:responseHeader is NULL");
    THROW_MQEXCEPTION(MQClientException,
                      "processPullResponse:responseHeader is NULL", -1);
  }
  //<!get body,delete outsite;
  MemoryBlock bodyFromResponse =
      *(pResponse->GetBody());  // response data judgement had been done outside
                                // of processPullResponse
  if (bodyFromResponse.getSize() == 0) {
    if (pullStatus != FOUND) {
      return new PullResultExt(pullStatus, responseHeader->nextBeginOffset,
                               responseHeader->minOffset,
                               responseHeader->maxOffset,
                               (int)responseHeader->suggestWhichBrokerId);
    } else {
      THROW_MQEXCEPTION(MQClientException,
                        "memoryBody size is 0, but pullStatus equals found",
                        -1);
    }
  } else {
    return new PullResultExt(
        pullStatus, responseHeader->nextBeginOffset, responseHeader->minOffset,
        responseHeader->maxOffset, (int)responseHeader->suggestWhichBrokerId,
        bodyFromResponse);
  }
}

//<!***************************************************************************
int64 MQClientAPIImpl::getMinOffset(
    const string& addr, const string& topic, int queueId, int timeoutMillis,
    const SessionCredentials& sessionCredentials) {
  GetMinOffsetRequestHeader* pRequestHeader = new GetMinOffsetRequestHeader();
  pRequestHeader->topic = topic;
  pRequestHeader->queueId = queueId;

  RemotingCommand request(GET_MIN_OFFSET, pRequestHeader);
  callSignatureBeforeRequest(addr, request, sessionCredentials);
  request.Encode();

  unique_ptr<RemotingCommand> response(
      m_pRemotingClient->invokeSync(addr, request, timeoutMillis));

  if (response) {
    switch (response->getCode()) {
      case SUCCESS_VALUE: {
        GetMinOffsetResponseHeader* responseHeader =
            (GetMinOffsetResponseHeader*)response->getCommandHeader();

        int64 offset = responseHeader->offset;
        return offset;
      }
      default:
        break;
    }
    THROW_MQEXCEPTION(MQBrokerException, response->getRemark(),
                      response->getCode());
  }
  THROW_MQEXCEPTION(MQBrokerException, "response is null", -1);
}

int64 MQClientAPIImpl::getMaxOffset(
    const string& addr, const string& topic, int queueId, int timeoutMillis,
    const SessionCredentials& sessionCredentials) {
  GetMaxOffsetRequestHeader* pRequestHeader = new GetMaxOffsetRequestHeader();
  pRequestHeader->topic = topic;
  pRequestHeader->queueId = queueId;

  RemotingCommand request(GET_MAX_OFFSET, pRequestHeader);
  callSignatureBeforeRequest(addr, request, sessionCredentials);
  request.Encode();

  unique_ptr<RemotingCommand> response(
      m_pRemotingClient->invokeSync(addr, request, timeoutMillis));

  if (response) {
    switch (response->getCode()) {
      case SUCCESS_VALUE: {
        GetMaxOffsetResponseHeader* responseHeader =
            (GetMaxOffsetResponseHeader*)response->getCommandHeader();

        int64 offset = responseHeader->offset;
        return offset;
      }
      default:
        break;
    }
    THROW_MQEXCEPTION(MQBrokerException, response->getRemark(),
                      response->getCode());
  }
  THROW_MQEXCEPTION(MQBrokerException, "response is null", -1);
}

int64 MQClientAPIImpl::searchOffset(
    const string& addr, const string& topic, int queueId, uint64_t timestamp,
    int timeoutMillis, const SessionCredentials& sessionCredentials) {
  SearchOffsetRequestHeader* pRequestHeader = new SearchOffsetRequestHeader();
  pRequestHeader->topic = topic;
  pRequestHeader->queueId = queueId;
  pRequestHeader->timestamp = timestamp;

  RemotingCommand request(SEARCH_OFFSET_BY_TIMESTAMP, pRequestHeader);
  callSignatureBeforeRequest(addr, request, sessionCredentials);
  request.Encode();

  unique_ptr<RemotingCommand> response(
      m_pRemotingClient->invokeSync(addr, request, timeoutMillis));

  if (response) {
    switch (response->getCode()) {
      case SUCCESS_VALUE: {
        SearchOffsetResponseHeader* responseHeader =
            (SearchOffsetResponseHeader*)response->getCommandHeader();

        int64 offset = responseHeader->offset;
        return offset;
      }
      default:
        break;
    }
    THROW_MQEXCEPTION(MQBrokerException, response->getRemark(),
                      response->getCode());
  }
  THROW_MQEXCEPTION(MQBrokerException, "response is null", -1);
}

MQMessageExt* MQClientAPIImpl::viewMessage(
    const string& addr, int64 phyoffset, int timeoutMillis,
    const SessionCredentials& sessionCredentials) {
  ViewMessageRequestHeader* pRequestHeader = new ViewMessageRequestHeader();
  pRequestHeader->offset = phyoffset;

  RemotingCommand request(VIEW_MESSAGE_BY_ID, pRequestHeader);
  callSignatureBeforeRequest(addr, request, sessionCredentials);
  request.Encode();

  unique_ptr<RemotingCommand> response(
      m_pRemotingClient->invokeSync(addr, request, timeoutMillis));

  if (response) {
    switch (response->getCode()) {
      case SUCCESS_VALUE: {
      }
      default:
        break;
    }
    THROW_MQEXCEPTION(MQBrokerException, response->getRemark(),
                      response->getCode());
  }
  THROW_MQEXCEPTION(MQBrokerException, "response is null", -1);
}

int64 MQClientAPIImpl::getEarliestMsgStoretime(
    const string& addr, const string& topic, int queueId, int timeoutMillis,
    const SessionCredentials& sessionCredentials) {
  GetEarliestMsgStoretimeRequestHeader* pRequestHeader =
      new GetEarliestMsgStoretimeRequestHeader();
  pRequestHeader->topic = topic;
  pRequestHeader->queueId = queueId;

  RemotingCommand request(GET_EARLIEST_MSG_STORETIME, pRequestHeader);
  callSignatureBeforeRequest(addr, request, sessionCredentials);
  request.Encode();

  unique_ptr<RemotingCommand> response(
      m_pRemotingClient->invokeSync(addr, request, timeoutMillis));

  if (response) {
    switch (response->getCode()) {
      case SUCCESS_VALUE: {
        GetEarliestMsgStoretimeResponseHeader* responseHeader =
            (GetEarliestMsgStoretimeResponseHeader*)
                response->getCommandHeader();

        int64 timestamp = responseHeader->timestamp;
        return timestamp;
      }
      default:
        break;
    }
    THROW_MQEXCEPTION(MQBrokerException, response->getRemark(),
                      response->getCode());
  }
  THROW_MQEXCEPTION(MQBrokerException, "response is null", -1);
}

void MQClientAPIImpl::getConsumerIdListByGroup(
    const string& addr, const string& consumerGroup, vector<string>& cids,
    int timeoutMillis, const SessionCredentials& sessionCredentials) {
  GetConsumerListByGroupRequestHeader* pRequestHeader =
      new GetConsumerListByGroupRequestHeader();
  pRequestHeader->consumerGroup = consumerGroup;

  RemotingCommand request(GET_CONSUMER_LIST_BY_GROUP, pRequestHeader);
  callSignatureBeforeRequest(addr, request, sessionCredentials);
  request.Encode();

  unique_ptr<RemotingCommand> pResponse(
      m_pRemotingClient->invokeSync(addr, request, timeoutMillis));

  if (pResponse != NULL) {
    if ((pResponse->GetBody()->getSize() == 0) ||
        (pResponse->GetBody()->getData() != NULL)) {
      switch (pResponse->getCode()) {
        case SUCCESS_VALUE: {
          const MemoryBlock* pbody = pResponse->GetBody();
          if (pbody->getSize()) {
            GetConsumerListByGroupResponseBody::Decode(pbody, cids);
            return;
          }
        }
        default:
          break;
      }
      THROW_MQEXCEPTION(MQBrokerException, pResponse->getRemark(),
                        pResponse->getCode());
    }
  }
  THROW_MQEXCEPTION(MQBrokerException, "response is null", -1);
}

int64 MQClientAPIImpl::queryConsumerOffset(
    const string& addr, QueryConsumerOffsetRequestHeader* pRequestHeader,
    int timeoutMillis, const SessionCredentials& sessionCredentials) {
  RemotingCommand request(QUERY_CONSUMER_OFFSET, pRequestHeader);
  callSignatureBeforeRequest(addr, request, sessionCredentials);
  request.Encode();

  unique_ptr<RemotingCommand> response(
      m_pRemotingClient->invokeSync(addr, request, timeoutMillis));

  if (response) {
    switch (response->getCode()) {
      case SUCCESS_VALUE: {
        QueryConsumerOffsetResponseHeader* responseHeader =
            (QueryConsumerOffsetResponseHeader*)response->getCommandHeader();
        int64 consumerOffset = responseHeader->offset;
        return consumerOffset;
      }
      default:
        break;
    }
    THROW_MQEXCEPTION(MQBrokerException, response->getRemark(),
                      response->getCode());
  }
  THROW_MQEXCEPTION(MQBrokerException, "response is null", -1);
  return -1;
}

void MQClientAPIImpl::updateConsumerOffset(
    const string& addr, UpdateConsumerOffsetRequestHeader* pRequestHeader,
    int timeoutMillis, const SessionCredentials& sessionCredentials) {
  RemotingCommand request(UPDATE_CONSUMER_OFFSET, pRequestHeader);
  callSignatureBeforeRequest(addr, request, sessionCredentials);
  request.Encode();

  unique_ptr<RemotingCommand> response(
      m_pRemotingClient->invokeSync(addr, request, timeoutMillis));

  if (response) {
    switch (response->getCode()) {
      case SUCCESS_VALUE: {
        return;
      }
      default:
        break;
    }
    THROW_MQEXCEPTION(MQBrokerException, response->getRemark(),
                      response->getCode());
  }
  THROW_MQEXCEPTION(MQBrokerException, "response is null", -1);
}

void MQClientAPIImpl::updateConsumerOffsetOneway(
    const string& addr, UpdateConsumerOffsetRequestHeader* pRequestHeader,
    int timeoutMillis, const SessionCredentials& sessionCredentials) {
  RemotingCommand request(UPDATE_CONSUMER_OFFSET, pRequestHeader);
  callSignatureBeforeRequest(addr, request, sessionCredentials);
  request.Encode();

  m_pRemotingClient->invokeOneway(addr, request);
}

void MQClientAPIImpl::consumerSendMessageBack(
    MQMessageExt& msg, const string& consumerGroup, int delayLevel,
    int timeoutMillis, const SessionCredentials& sessionCredentials) {
  ConsumerSendMsgBackRequestHeader* pRequestHeader =
      new ConsumerSendMsgBackRequestHeader();
  pRequestHeader->group = consumerGroup;
  pRequestHeader->offset = msg.getCommitLogOffset();
  pRequestHeader->delayLevel = delayLevel;

  string addr = socketAddress2IPPort(msg.getStoreHost());
  RemotingCommand request(CONSUMER_SEND_MSG_BACK, pRequestHeader);
  callSignatureBeforeRequest(addr, request, sessionCredentials);
  request.Encode();

  unique_ptr<RemotingCommand> response(
      m_pRemotingClient->invokeSync(addr, request, timeoutMillis));

  if (response) {
    switch (response->getCode()) {
      case SUCCESS_VALUE: {
        return;
      }
      default:
        break;
    }
    THROW_MQEXCEPTION(MQBrokerException, response->getRemark(),
                      response->getCode());
  }
  THROW_MQEXCEPTION(MQBrokerException, "response is null", -1);
}

void MQClientAPIImpl::lockBatchMQ(
    const string& addr, LockBatchRequestBody* requestBody,
    vector<MQMessageQueue>& mqs, int timeoutMillis,
    const SessionCredentials& sessionCredentials) {
  RemotingCommand request(LOCK_BATCH_MQ, NULL);
  string body;
  requestBody->Encode(body);
  request.SetBody(body.data(), body.length());
  request.setMsgBody(body);
  callSignatureBeforeRequest(addr, request, sessionCredentials);
  request.Encode();

  unique_ptr<RemotingCommand> pResponse(
      m_pRemotingClient->invokeSync(addr, request, timeoutMillis));

  if (pResponse != NULL) {
    if (((*(pResponse->GetBody())).getSize() == 0) ||
        ((*(pResponse->GetBody())).getData() != NULL)) {
      switch (pResponse->getCode()) {
        case SUCCESS_VALUE: {
          const MemoryBlock* pbody = pResponse->GetBody();
          if (pbody->getSize()) {
            LockBatchResponseBody::Decode(pbody, mqs);
          }
          return;
        } break;
        default:
          break;
      }
      THROW_MQEXCEPTION(MQBrokerException, pResponse->getRemark(),
                        pResponse->getCode());
    }
  }
  THROW_MQEXCEPTION(MQBrokerException, "response is null", -1);
}

void MQClientAPIImpl::unlockBatchMQ(
    const string& addr, UnlockBatchRequestBody* requestBody, int timeoutMillis,
    const SessionCredentials& sessionCredentials) {
  RemotingCommand request(UNLOCK_BATCH_MQ, NULL);
  string body;
  requestBody->Encode(body);
  request.SetBody(body.data(), body.length());
  request.setMsgBody(body);
  callSignatureBeforeRequest(addr, request, sessionCredentials);
  request.Encode();

  unique_ptr<RemotingCommand> pResponse(
      m_pRemotingClient->invokeSync(addr, request, timeoutMillis));

  if (pResponse != NULL) {
    switch (pResponse->getCode()) {
      case SUCCESS_VALUE: {
        return;
      } break;
      default:
        break;
    }
    THROW_MQEXCEPTION(MQBrokerException, pResponse->getRemark(),
                      pResponse->getCode());
  }
  THROW_MQEXCEPTION(MQBrokerException, "response is null", -1);
}

//<!************************************************************************
}  //<!end namespace;

max_stars_repo_name: wlliqipeng/rocketmq-client-cpp

max_issues_repo_name: yewenchen/rocketmq-client-cpp

max_forks_repo_name: yewenchen/rocketmq-client-cpp