text:
// Copyright (c) 2019-2021 Duality Blockchain Solutions Developers
// Distributed under the MIT software license, see the accompanying
// file COPYING or http://www.opensource.org/licenses/mit-license.php.

#include "bdap/auditdb.h"

#include "amount.h"
#include "base58.h"
#include "bdap/domainentrydb.h"
#include "bdap/fees.h"
#include "bdap/utils.h"
#include "coins.h"
#include "utilmoneystr.h"
#include "utiltime.h"
#include "validation.h"
#include "validationinterface.h"

#include <univalue.h>

#include <boost/thread.hpp>

CAuditDB *pAuditDB = NULL;

bool GetAuditTxId(const std::string& strTxId, CAudit& audit)
{
    if (!pAuditDB || !pAuditDB->ReadAuditTxId(vchFromString(strTxId), audit))
        return false;

    return !audit.IsNull();
}

bool AuditExists(const std::vector<unsigned char>& vchAudit)
{
    if (!pAuditDB)
        return false;

    return pAuditDB->AuditExists(vchAudit);
}

bool UndoAddAudit(const CAudit& audit)
{
    if (!pAuditDB)
        return false;

    return pAuditDB->EraseAuditTxId(vchFromString(audit.txHash.ToString()));
}

bool CAuditDB::AddAudit(const CAudit& audit)
{
    bool writeState = false;
    bool writeStateDN = false;
    bool auditHashWriteState = true;
    {
        LOCK(cs_bdap_audit);
        CAuditData auditData = audit.GetAuditData();
        for(const std::vector<unsigned char>& vchAuditHash : auditData.vAuditData) {
            std::vector<std::vector<unsigned char>> vvTxHash;
            CDBWrapper::Read(make_pair(std::string("audit"), vchAuditHash), vvTxHash);
            vvTxHash.push_back(vchFromString(audit.txHash.ToString()));
            // each hash points to a txid. The txid record stores the audit record.
            if (!Write(make_pair(std::string("audit"), vchAuditHash), vvTxHash)) {
                auditHashWriteState = false;
            }
        }
        if (audit.vchOwnerFullObjectPath.size() > 0) {
            std::vector<std::vector<unsigned char>> vvTxId;
            CDBWrapper::Read(make_pair(std::string("dn"), audit.vchOwnerFullObjectPath), vvTxId);
            vvTxId.push_back(vchFromString(audit.txHash.ToString()));

            writeStateDN = Write(make_pair(std::string("dn"), audit.vchOwnerFullObjectPath), vvTxId);
        }
        else {
            writeStateDN = true;
        }
        writeState = Write(make_pair(std::string("txid"), vchFromString(audit.txHash.ToString())), audit);
    }

    return writeState && auditHashWriteState && writeStateDN;
}

bool CAuditDB::ReadAuditTxId(const std::vector<unsigned char>& vchTxId, CAudit& audit)
{
    LOCK(cs_bdap_audit);
    return CDBWrapper::Read(make_pair(std::string("txid"), vchTxId), audit);
}

bool CAuditDB::ReadAuditDN(const std::vector<unsigned char>& vchOwnerFullObjectPath, std::vector<CAudit>& vAudits)
{
    LOCK(cs_bdap_audit);
    std::vector<std::vector<unsigned char>> vvTxId;
    bool readState = false;
    readState = CDBWrapper::Read(make_pair(std::string("dn"), vchOwnerFullObjectPath), vvTxId);

    if (readState) {
        for (const std::vector<unsigned char>& vchTxId : vvTxId) {
            CAudit audit;
            if (ReadAuditTxId(vchTxId, audit)) {
                vAudits.push_back(audit);
            }
        }
    }

    return (vAudits.size() > 0);
}

bool CAuditDB::ReadAuditHash(const std::vector<unsigned char>& vchAudit, std::vector<CAudit>& vAudits)
{
    LOCK(cs_bdap_audit);
    std::vector<std::vector<unsigned char>> vvTxId;
    bool readState = false;
    readState = CDBWrapper::Read(make_pair(std::string("audit"), vchAudit), vvTxId);

    if (readState) {
        for (const std::vector<unsigned char>& vchTxId : vvTxId) {
            CAudit audit;
            if (ReadAuditTxId(vchTxId, audit)) {
                vAudits.push_back(audit);
            }
        }
    }

    return (vAudits.size() > 0);
}

bool CAuditDB::AuditExists(const std::vector<unsigned char>& vchAudit)
{
    LOCK(cs_bdap_audit);
    return CDBWrapper::Exists(make_pair(std::string("audit"), vchAudit));
}

bool CAuditDB::EraseAuditTxId(const std::vector<unsigned char>& vchTxId)
{
    LOCK(cs_bdap_audit);
    CAudit audit;
    if (ReadAuditTxId(vchTxId, audit)) {
        for(const std::vector<unsigned char>& vchAudit : audit.GetAudits())
            CDBWrapper::Erase(make_pair(std::string("audit"), vchAudit));
    }
    if (audit.vchOwnerFullObjectPath.size() > 0) {
        std::vector<std::vector<unsigned char>> vvTxId;
        CDBWrapper::Read(make_pair(std::string("dn"), audit.vchOwnerFullObjectPath), vvTxId);
        if (vvTxId.size() == 1 && vvTxId[0] == vchTxId) {
            CDBWrapper::Erase(make_pair(std::string("dn"), audit.vchOwnerFullObjectPath));
        }
        else {
            std::vector<std::vector<unsigned char>> vvTxIdNew;
            for (const std::vector<unsigned char>& txid : vvTxId) {
                if (txid != vchTxId) {
                    vvTxIdNew.push_back(txid);
                }
            }
            Write(make_pair(std::string("dn"), audit.vchOwnerFullObjectPath), vvTxIdNew);
        }
    }
    return CDBWrapper::Erase(make_pair(std::string("txid"), vchTxId));
}

bool CAuditDB::EraseAudit(const std::vector<unsigned char>& vchAudit)
{
    LOCK(cs_bdap_audit);
    return CDBWrapper::Erase(make_pair(std::string("audit"), vchAudit));
}

bool CheckAuditDB()
{
    if (!pAuditDB)
        return false;

    return true;
}

bool FlushAuditLevelDB()
{
    {
        LOCK(cs_bdap_audit);
        if (pAuditDB != NULL)
        {
            if (!pAuditDB->Flush()) {
                LogPrintf("Failed to flush Audit BDAP database!");
                return false;
            }
        }
    }
    return true;
}

static bool CommonDataCheck(const CAudit& audit, const vchCharString& vvchOpParameters, std::string& errorMessage)
{
    if (audit.IsNull()) {
        errorMessage = "CommonDataCheck failed! Audit is null.";
        return false;
    }

    if (!audit.ValidateValues(errorMessage)) {
        errorMessage = "CommonDataCheck failed! Invalid audit value. " + errorMessage;
        return false;
    }

    if (vvchOpParameters.size() > 3) {
        errorMessage = "CommonDataCheck failed! Too many parameters.";
        return false;
    }

    if (vvchOpParameters.size() < 1) {
        errorMessage = "CommonDataCheck failed! Not enough parameters.";
        return false;
    }
    // check count parameter is not longer than
    if (vvchOpParameters[0].size() > 10) {
        errorMessage = "CommonDataCheck failed! Not enough parameters.";
        return false;
    }
    // check if count equals audit count.
    uint32_t nCount;
    ParseUInt32(stringFromVch(vvchOpParameters[0]), &nCount);
    if (nCount != audit.GetAudits().size()) {
        errorMessage = "CommonDataCheck failed! Parameter count does not match audits in data.";
        return false;
    }

    // check if signed.
    if (vvchOpParameters.size() == 2) {
        errorMessage = "CommonDataCheck failed! Only two parameters. Signed audits require 3 parameters.";
        return false;
    }

    if (vvchOpParameters.size() > 1 && audit.vchOwnerFullObjectPath != vvchOpParameters[1]) {
        errorMessage = "CommonDataCheck failed! Script operation account parameter does not match account in audit object.";
        return false;
    }
    // check FQDN size
    if (vvchOpParameters.size() > 1 && vvchOpParameters[1].size() > MAX_OBJECT_FULL_PATH_LENGTH) {
        errorMessage = "CommonDataCheck failed! Count parameter value too large.";
        return false;
    }
    // check pubkey size
    if (vvchOpParameters.size() > 1 && vvchOpParameters[2].size() > MAX_KEY_LENGTH) {
        errorMessage = "CommonDataCheck failed! Count parameter value too large.";
        return false;
    }
    return true;
}

static bool CheckNewAuditTxInputs(const CAudit& audit, const CScript& scriptOp, const vchCharString& vvchOpParameters, const uint256& txHash,
                               std::string& errorMessage, bool fJustCheck)
{
    if (!CommonDataCheck(audit, vvchOpParameters, errorMessage))
        return error(errorMessage.c_str());

    if (fJustCheck)
        return true;

    // check audit signature if required.
    if (audit.SignatureRequired()) {
        if (!audit.IsSigned()) {
            errorMessage = "CheckNewAuditTxInputs: - The audit requires a signature.  Add new audit failed!";
            return error(errorMessage.c_str());
        }
        CDomainEntry entry;
        if (!GetDomainEntry(audit.vchOwnerFullObjectPath, entry)) {
            errorMessage = "CheckNewAuditTxInputs: - Could not find specified audit account owner! " + stringFromVch(audit.vchOwnerFullObjectPath);
            return error(errorMessage.c_str());
        }
        CDynamicAddress address = entry.GetWalletAddress();
        CKeyID keyID;
        if (!address.GetKeyID(keyID)) {
            errorMessage = "CheckNewAuditTxInputs: - Could not get key id. " + address.ToString();
            return error(errorMessage.c_str());
        }
        // check signature and pubkey belongs to bdap account.
        CPubKey pubkey(vvchOpParameters[2]);

        CDynamicAddress addressCompare(pubkey.GetID());

        if (!(address == addressCompare)) {
            errorMessage = "CheckNewAuditTxInputs: - Wallet address does not match. ";
            return error(errorMessage.c_str());
        }

        if (!audit.CheckSignature(pubkey.Raw())) {
            errorMessage = "CheckNewAuditTxInputs: - Could not validate signature. ";
            return error(errorMessage.c_str());
        }

    }

    CAudit getAudit;
    if (GetAuditTxId(audit.txHash.ToString(), getAudit)) {
        if (audit.txHash != txHash) {
            errorMessage = "CheckNewAuditTxInputs: - The audit " + audit.txHash.ToString() + " already exists.  Add new audit failed!";
            return error(errorMessage.c_str());
        } else {
            LogPrintf("%s -- Already have audit %s in local database. Skipping add audit step.\n", __func__, audit.txHash.ToString());
            return true;
        }
    }

    if (!pAuditDB) {
        errorMessage = "CheckNewAuditTxInputs failed! Can not open LevelDB BDAP audit database.";
        return error(errorMessage.c_str());
    }
    if (!pAuditDB->AddAudit(audit)) {
        errorMessage = "CheckNewAuditTxInputs failed! Error adding new audit record to LevelDB.";
        pAuditDB->EraseAuditTxId(vchFromString(audit.txHash.ToString()));
        return error(errorMessage.c_str());
    }

    return FlushAuditLevelDB();
}

bool CheckAuditTx(const CTransactionRef& tx, const CScript& scriptOp, const int& op1, const int& op2, const std::vector<std::vector<unsigned char> >& vvchArgs,
                                const bool fJustCheck, const int& nHeight, const uint32_t& nBlockTime, const bool bSanityCheck, std::string& errorMessage)
{
    if (tx->IsCoinBase() && !fJustCheck && !bSanityCheck) {
        LogPrintf("*Trying to add BDAP audit in coinbase transaction, skipping...");
        return true;
    }

    LogPrint("bdap", "%s -- BDAP nHeight=%d, chainActive.Tip()=%d, op1=%s, op2=%s, hash=%s justcheck=%s\n", __func__, nHeight, chainActive.Tip()->nHeight, BDAPFromOp(op1).c_str(), BDAPFromOp(op2).c_str(), tx->GetHash().ToString().c_str(), fJustCheck ? "JUSTCHECK" : "BLOCK");

    // unserialize BDAP from txn, check if the audit is valid and does not conflict with a previous audit
    CAudit audit;
    std::vector<unsigned char> vchData;
    std::vector<unsigned char> vchHash;
    int nDataOut;
    bool bData = GetBDAPData(tx, vchData, vchHash, nDataOut);
    if(bData && !audit.UnserializeFromTx(tx, nHeight))
    {
        errorMessage = ("UnserializeFromData data in tx failed!");
        LogPrintf("%s -- %s \n", __func__, errorMessage);
        return error(errorMessage.c_str());
    }
    const std::string strOperationType = GetBDAPOpTypeString(op1, op2);
    CAmount monthlyFee, oneTimeFee, depositFee;
    if (strOperationType == "bdap_new_audit") {
        if (!audit.ValidateValues(errorMessage))
            return false;

        if (vvchArgs.size() > 3) {
            errorMessage = "Failed to get fees to add a new BDAP account";
            return false;
        }
        std::string strCount = stringFromVch(vvchArgs[0]);
        uint32_t nCount;
        ParseUInt32(stringFromVch(vvchArgs[0]), &nCount);
        if (!GetBDAPFees(OP_BDAP_NEW, OP_BDAP_AUDIT, BDAP::ObjectType::BDAP_AUDIT, (uint16_t)nCount, monthlyFee, oneTimeFee, depositFee)) {
            errorMessage = "Failed to get fees to add a new BDAP account";
            return false;
        }
        LogPrint("bdap", "%s -- nCount %d, oneTimeFee %d\n", __func__, nCount, FormatMoney(oneTimeFee));
        // extract amounts from tx.
        CAmount dataAmount, opAmount;
        if (!ExtractAmountsFromTx(tx, dataAmount, opAmount)) {
            errorMessage = "Unable to extract BDAP amounts from transaction";
            return false;
        }
        LogPrint("bdap", "%s -- dataAmount %d, opAmount %d\n", __func__, FormatMoney(dataAmount), FormatMoney(opAmount));
        if (monthlyFee > dataAmount) {
            LogPrintf("%s -- Invalid BDAP monthly registration fee amount for new BDAP account. Monthly paid %d but should be %d\n", __func__,
                                    FormatMoney(dataAmount), FormatMoney(monthlyFee));
            errorMessage = "Invalid BDAP monthly registration fee amount for new BDAP account";
            return false;
        }
        else {
            LogPrint("bdap", "%s -- Valid BDAP monthly registration fee amount for new BDAP account. Monthly paid %d, should be %d.\n", __func__,
                                    FormatMoney(dataAmount), FormatMoney(monthlyFee));
        }
        if (depositFee > opAmount) {
            LogPrintf("%s -- Invalid BDAP deposit fee amount for new BDAP account. Deposit paid %d but should be %d\n", __func__,
                                    FormatMoney(opAmount), FormatMoney(depositFee));
            errorMessage = "Invalid BDAP deposit fee amount for new BDAP account";
            return false;
        }
        else {
            LogPrint("bdap", "%s -- Valid BDAP deposit fee amount for new BDAP account. Deposit paid %d, should be %d\n", __func__,
                                    FormatMoney(opAmount), FormatMoney(depositFee));
        }

        return CheckNewAuditTxInputs(audit, scriptOp, vvchArgs, tx->GetHash(), errorMessage, fJustCheck);
    }

    return false;
}

max_stars_repo_name: AmirAbrams/dynamic-swap

max_issues_repo_name: AmirAbrams/dynamic-swap


text:
/*
    Copyright 2013 Adobe
    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 <adobe/config.hpp>

#define BOOST_TEST_MAIN

#include <boost/test/unit_test.hpp>

#include <adobe/enum_ops.hpp>

enum Number {
    num_neg_one = -1,
    num_0 = 0,
    num_1 = 1,
    num_2 = 2,
    num_3 = 3,
    num_4 = 4,
    num_5 = 5,
    num_6 = 6,
    num_7 = 7
};

enum class Views : int { None = 0, Text = 1 << 0, Icon = 1 << 1, Preview = 1 << 2 };

enum class Num : int {
    num_0 = 0,
    num_1 = 1,
    num_2 = 2,
    num_3 = 3,
    num_4 = 4,
    num_5 = 5,
    num_6 = 6,
    num_7 = 7
};

auto stlab_enable_bitmask_enum(Views) -> std::true_type;

ADOBE_DEFINE_BITSET_OPS(Number)

BOOST_AUTO_TEST_CASE(enum_bitset_ops) {
    Number x;

    // or
    x = num_1 | num_2;
    BOOST_CHECK(x == 3);

    // and
    x = num_3 & num_1;
    BOOST_CHECK(x == 1);

    // xor
    x = num_2 ^ num_3;
    BOOST_CHECK(x == 1);

    // not
    x = num_7 & (~num_5);
    BOOST_CHECK(x == 2);

    // or=
    x = num_1;
    x |= num_2;
    BOOST_CHECK(x == 3);

    // and=
    x = num_3;
    x &= num_1;
    BOOST_CHECK(x == 1);

    // xor
    x = num_2;
    x ^= num_3;
    BOOST_CHECK(x == 1);
}

BOOST_AUTO_TEST_CASE(enumclass_bitset_ops) {
    // the above proves our general cases
    // this just sanity checks enum classes
    Views x;
    x = Views::Text | Views::Icon;
    BOOST_CHECK((x & Views::Text) == Views::Text);
    BOOST_CHECK((x & Views::Preview) != Views::Preview);
}

ADOBE_DEFINE_ARITHMETIC_OPS(Number)
auto stlab_enable_arithmetic_enum(Num) -> std::true_type;

BOOST_AUTO_TEST_CASE(enum_arith_ops) {
    Number x;

    // +
    x = num_1 + num_2;
    BOOST_CHECK(x == 3);

    // -
    x = num_3 - num_1;
    BOOST_CHECK(x == 2);

    // - (unary)
    x = -num_neg_one;
    BOOST_CHECK(x == 1);

    // *
    x = num_2 * num_3;
    BOOST_CHECK(x == 6);

    // /
    x = num_5 / num_2;
    BOOST_CHECK(x == 2);

    // %
    x = num_5 % num_2;
    BOOST_CHECK(x == 1);

    // +=
    x = num_1;
    x += num_2;
    BOOST_CHECK(x == 3);

    // -=
    x = num_3;
    x -= num_1;
    BOOST_CHECK(x == 2);

    // *=
    x = num_2;
    x *= num_3;
    BOOST_CHECK(x == 6);

    // /=
    x = num_5;
    x /= num_2;
    BOOST_CHECK(x == 2);

    // %=
    x = num_5;
    x %= num_2;
    BOOST_CHECK(x == 1);
}

BOOST_AUTO_TEST_CASE(enumclass_arith_ops) {
    // the above proves our general cases
    // this just sanity checks enum classes
    Num x;
    x = Num::num_1;
    BOOST_CHECK((x + Num::num_3) == Num::num_4);
}

ext: cpp


text:
/**
 * Performance test: logistic.
 *
 * This test runs a simple logistic model 100 times with a fixed
 * seed. The test records to a csv file information about the current
 * build, whether the values match the expected known values, whether
 * values are identical between runs, and the runtime in seconds for
 * each of the 100 individual runs.
 *
 * The output file is a csv file located at
 * test/performance/performance.csv.  If the file is empty, a header
 * line will be written.  If the file is not empty and the header line
 * of the file matches, a data row will be appended.  If the file is
 * not empty and the header line is different, a header row and a data
 * row will be appended.
 *
 * Below are the test routines in this test.
 * 1. run: Runs the model 100 times. This test should always
 *    succeed. Test failure indicates that the model could not be run.
 * 2. values_from_tagged_version: Compares the first run's final
 *    iteration against known values. Test will succeed if the values
 *    match the known values; the test will fail otherwise.
 * 3. values_same_run_to_run: Compares the first run's final iteration
 *    against all runs' final iterations. Test will succeed if all the
 *    values match; the test will fail otherwise.
 * 4. write_results_to_disk: Writes the results to disk. Test will
 *    succeed if writing to an empty file or to one with a matching
 *    header; the test will fail if the header in the existing file
 *    does not match the current header.
 *
 * Note: the values_from_tagged_version will need to be updated
 * if Stan's RNG changes, the seeding mechanism changes, or the
 * algorithm changes. The other routines should be fine.
 *
 * Each run of this test produces a single, csv row with 106 columns
 * in the output file:
 *   1. current date. Formatted using std::ctime: "Www Mmm dd hh:mm:ss yyyy"
 *   2. git hash. The current 40-character SHA-1 git hash or "NA".
 *   3. git date. The date of the last commit or "NA".
 *   4. model name. "logistic"
 *   5. matches tagged version. Either "yes" or "no".
 *   6. all values the same. Either "yes" or "no".
 *   7--106. run time (in seconds).
 */

#include <gtest/gtest.h>
#include <test/test-models/performance/logistic.hpp>
#include <stan/version.hpp>
#include <ctime>
#include <test/performance/utility.hpp>
#include <boost/algorithm/string/trim.hpp>

class performance : public ::testing::Test {
 public:
  static void SetUpTestCase() {
    N = 100;
    seconds_per_run.resize(N);
    last_draws_per_run.resize(N);
    matches_tagged_version = false;
    all_values_same = false;
  }

  static int N;
  static std::vector<double> seconds_per_run;
  static std::vector<std::vector<double> > last_draws_per_run;
  static bool matches_tagged_version;
  static bool all_values_same;
};

int performance::N;
std::vector<double> performance::seconds_per_run;
std::vector<std::vector<double> > performance::last_draws_per_run;
bool performance::matches_tagged_version;
bool performance::all_values_same;

using stan::test::performance::get_date;
using stan::test::performance::get_git_date;
using stan::test::performance::get_git_hash;
using stan::test::performance::get_last_iteration_from_file;
using stan::test::performance::quote;
using stan::test::performance::run_command;
using stan::test::performance::run_command_output;

TEST_F(performance, run) {
  clock_t t;
  for (int n = 0; n < N; ++n) {
    std::cout << "iteration: " << n << " / " << N << std::endl;
    t = clock();  // start timer
    stan::test::performance::command<stan_model>(
        1000, 10000, "src/test/test-models/performance/logistic.data.R",
        "test/performance/logistic_output.csv", 0U);
    t = clock() - t;  // end timer
    seconds_per_run[n] = static_cast<double>(t) / CLOCKS_PER_SEC;
    last_draws_per_run[n]
        = get_last_iteration_from_file("test/performance/logistic_output.csv");
  }
  SUCCEED();
}

// evaluate
TEST_F(performance, values_from_tagged_version) {
  int N_values = 9;
  ASSERT_EQ(N_values, last_draws_per_run[0].size())
      << "last tagged version, 2.17.0, had " << N_values << " elements";

  std::vector<double> first_run = last_draws_per_run[0];
  EXPECT_FLOAT_EQ(-65.216301, first_run[0]) << "lp__: index 0";

  EXPECT_FLOAT_EQ(0.91851199, first_run[1]) << "accept_stat__: index 1";

  EXPECT_FLOAT_EQ(0.76885802, first_run[2]) << "stepsize__: index 2";

  EXPECT_FLOAT_EQ(2, first_run[3]) << "treedepth__: index 3";

  EXPECT_FLOAT_EQ(3, first_run[4]) << "n_leapfrog__: index 4";

  EXPECT_FLOAT_EQ(0, first_run[5]) << "divergent__: index 5";

  EXPECT_FLOAT_EQ(66.696503, first_run[6]) << "energy__: index 6";

  EXPECT_FLOAT_EQ(1.3577, first_run[7]) << "beta.1: index 7";

  EXPECT_FLOAT_EQ(-0.51189202, first_run[8]) << "beta.2: index 8";

  matches_tagged_version = !HasNonfatalFailure();
}

TEST_F(performance, values_same_run_to_run) {
  int N_values = last_draws_per_run[0].size();

  for (int i = 0; i < N_values; i++) {
    double expected_value = last_draws_per_run[0][i];
    for (int n = 1; n < N; n++) {
      EXPECT_FLOAT_EQ(expected_value, last_draws_per_run[n][i])
          << "expecting run to run values to be the same. Found run " << n
          << " to have different values than the 0th run for "
          << "index: " << i;
    }
  }
  all_values_same = !HasNonfatalFailure();
}

TEST_F(performance, check_output_is_same) {
  std::ifstream file_stream;
  file_stream.open("test/performance/logistic_output.csv", std::ios_base::in);
  ASSERT_TRUE(file_stream.good());

  std::string line, expected;

  getline(file_stream, line);
  ASSERT_EQ(
      "lp__,accept_stat__,stepsize__,treedepth__,n_leapfrog__,divergent__,"
      "energy__,beta.1,beta.2",
      line);
  ASSERT_TRUE(file_stream.good());

  getline(file_stream, line);
  ASSERT_EQ("# Adaptation terminated", line);
  ASSERT_TRUE(file_stream.good());

  file_stream.close();
}

TEST_F(performance, write_results_to_disk) {
  std::stringstream header;
  std::stringstream line;

  // current date / time
  header << quote("date");
  line << quote(get_date());

  // git hash
  header << "," << quote("git hash") << "," << quote("git date");
  line << "," << quote(get_git_hash()) << "," << quote(get_git_date());

  // model name: "logistic"
  header << "," << quote("model name");
  line << "," << quote("logistic");

  // matches tagged values
  header << "," << quote("matches tagged version");
  line << "," << quote(matches_tagged_version ? "yes" : "no");

  // all values same
  header << "," << quote("all values same");
  line << "," << quote(all_values_same ? "yes" : "no");

  // N times
  for (int n = 0; n < N; n++) {
    std::stringstream ss;
    ss << "run " << n + 1;
    header << "," << quote(ss.str());
    line << "," << seconds_per_run[n];
  }

  // append output to: test/performance/performance.csv
  bool write_header = false;
  std::fstream file_stream;

  file_stream.open("test/performance/performance.csv", std::ios_base::in);
  if (file_stream.peek() == std::fstream::traits_type::eof()) {
    write_header = true;
  } else {
    std::string file_header;
    std::getline(file_stream, file_header);

    EXPECT_EQ(file_header, header.str()) << "header of file is different";
    if (file_header != header.str())
      write_header = true;
  }
  file_stream.close();

  file_stream.open("test/performance/performance.csv", std::ios_base::app);
  if (write_header)
    file_stream << header.str() << std::endl;
  file_stream << line.str() << std::endl;
  file_stream.close();
}

max_stars_repo_name: yizhang-yiz/stan

max_issues_repo_name: yizhang-yiz/stan

max_forks_repo_name: yizhang-yiz/stan


text:
#include <n2o/converter/builtin_converters.hpp>

#include <boost/cast.hpp>

#include <n2o/converter/rvalue_from_js_data.hpp>
#include <n2o/converter/registry.hpp>
#include <n2o/type_id.hpp>

namespace n2o { namespace converter {

typedef v8::Handle<v8::Value> (*unaryfunc)(v8::Handle<v8::Value>);

namespace {

template <typename T, typename SlotPolicy>
struct slot_rvalue_from_js {
    slot_rvalue_from_js() {
        registry::insert(
                  & slot_rvalue_from_js<T, SlotPolicy>::convertible
                , & slot_rvalue_from_js<T, SlotPolicy>::construct
                , n2o::type_id<T>() // XXX why the namespace?
                , & SlotPolicy::get_jstype
                );
    }
private:
    static void* convertible(v8::Handle<v8::Value> v) {
        unaryfunc * slot = SlotPolicy::get_slot(v);
        return slot && *slot ? slot : 0;
    }
    static void construct(v8::Handle<v8::Value> v, rvalue_from_js_stage1_data* data) {
        unaryfunc creator = * static_cast<unaryfunc*>(data->convertible);
        v8::Local<v8::Value> intermediate = creator(v);

        void * storage = ((rvalue_from_js_storage<T>*)data)->storage.bytes;

        new (storage) T(SlotPolicy::extract(intermediate));

        // record successful construction
        data->convertible = storage;
    }
};

v8::Handle<v8::Value> identity_unaryfunc(v8::Handle<v8::Value> x) {
    return x;
}

unaryfunc js_value_identity = identity_unaryfunc;

struct int_rvalue_from_js_base {
    static unaryfunc* get_slot(v8::Handle<v8::Value> v) {
        return v->IsInt32() || v->IsUint32() ? &js_value_identity : 0; // XXX
    }
    static js_type_info const* get_jstype() { return 0; }
};

template <typename T>
struct signed_int_rvalue_from_js : int_rvalue_from_js_base {
    static T extract(v8::Handle<v8::Value> intermediate) {
        int64_t x = intermediate->IntegerValue();
        return boost::numeric_cast<T>(x);
    }
};

template <typename T>
struct unsigned_int_rvalue_from_js : int_rvalue_from_js_base {
    static T extract(v8::Handle<v8::Value> intermediate) {
        int64_t x = intermediate->IntegerValue();
        return boost::numeric_cast<T>(x);
    }
};

struct bool_rvalue_from_js {
    static unaryfunc* get_slot(v8::Handle<v8::Value> v) {
        return v->IsBoolean() ? &js_value_identity : 0;
    }
    static bool extract(v8::Handle<v8::Value> v) {
        return v->BooleanValue();
    }
    static js_type_info const* get_jstype() { return 0; }
};

struct float_rvalue_from_js {
    static unaryfunc* get_slot(v8::Handle<v8::Value> v) {
        return v->IsNumber() ? &js_value_identity : 0;
    }
    static double extract(v8::Handle<v8::Value> intermediate) {
        return intermediate->NumberValue();
    }
    static js_type_info const* get_jstype() { return 0; }
};

struct string_rvalue_from_js {
    static unaryfunc* get_slot(v8::Handle<v8::Value> v) {
        return v->IsString() ? &js_value_identity : 0;
    }
    static std::string extract(v8::Handle<v8::Value> intermediate) {
        v8::String::Utf8Value utf8(intermediate);
        return std::string(*utf8, utf8.length());
    }
    static js_type_info const* get_jstype() { return 0; }
};

} // end of anonymous namespace

v8::Local<v8::Value>
do_return_to_js(char x) {
    return v8::String::NewFromUtf8(v8::Isolate::GetCurrent(), &x, v8::String::kNormalString, 1);
}

v8::Local<v8::Value>
do_return_to_js(char const* x) {
    if (x) {
        return v8::String::NewFromUtf8(v8::Isolate::GetCurrent(), x, v8::String::kNormalString);
    }
    return v8::Undefined(v8::Isolate::GetCurrent());
}

v8::Local<v8::Value>
do_return_to_js(v8::Handle<v8::Value> x) {
    if ( ! x.IsEmpty()) {
        return x;
    }
    return v8::Undefined(v8::Isolate::GetCurrent());
}

#define REGISTER_INT_CONVERTERS_S(signedness, U)               \
    slot_rvalue_from_js<                                       \
              signedness U                                     \
            , signedness ## _int_rvalue_from_js<signedness U>  \
    >()

#define REGISTER_INT_CONVERTERS(U)        \
    REGISTER_INT_CONVERTERS_S(signed, U); \
    REGISTER_INT_CONVERTERS_S(unsigned, U)

void
initialize_builtin_converters() {
    // booleans
    slot_rvalue_from_js<bool, bool_rvalue_from_js>();

    // integers
    REGISTER_INT_CONVERTERS(char);
    REGISTER_INT_CONVERTERS(short);
    REGISTER_INT_CONVERTERS(int);
    REGISTER_INT_CONVERTERS(long);

    // floating point types
    slot_rvalue_from_js<float,  float_rvalue_from_js>();
    slot_rvalue_from_js<double, float_rvalue_from_js>();

    // strings
    slot_rvalue_from_js<std::string, string_rvalue_from_js>();
}

}} // end of namespace n2o::converter


text:
/*****************************************************************
 Author: J.-G. Young, <jean.gabriel.young@gmail.com>
******************************************************************
 This program creates a network using the complete structural
 preferential attachment model described in

  A) Hébert-Dufresne et al., Phys.  Rev. Lett. 107, 1, (2011).
  B) Hébert-Dufresne et al., Phys. Rev. E 85, 026108, (2012).
  C) Young et al., Phys. Rev. E (2016).

******************************************************************
 Usage:
******************************************************************
  Compilation:
    g++ -o3 -W -Wall -Wextra -pedantic -std=c++0x spa_main.cpp spa_network.cpp -lboost_program_options -o SPA

  Call:
    Run ./SPA -h to see the list of parameters.

*****************************************************************/

// Configuration file
#include "config.h"
// C libraries
#include <stdlib.h> // C error codes
// STL
#include <string>   // const char* to std::string()
#include <chrono>   // c++11, benchmarking
// boost
#if HAVE_LIBBOOST_PROGRAM_OPTIONS == 1
  #include <boost/program_options.hpp>
#endif
// Project files
#include "spa_network.hpp" // spa_network_t
#include "interface.hpp"  // boost_interface(3) and barbone_interface(3)
#include "output_functions.hpp"  // print_outputs(2)
#include "spa_algorithm.hpp" // spa_star(5) and spa_basic(4)
// Typdef and helpers
#include "types.hpp" // spa_parameters_t
#include "errors.hpp" // detailed error code
#include "includes/logger.hpp" // LOGGER macros

int main(int argc, char const *argv[]) {

  LOGGER_START("")

  // Interface (parse argv, handle parameter errors, help message, etc.)

  spa_parameters_t prog_params;

  #if HAVE_LIBBOOST_PROGRAM_OPTIONS == 1
    int interface_return = boost_interface(argc, argv, prog_params);
  #else
    int interface_return = barebone_interface(argc, argv, prog_params);
  #endif

  switch(interface_return) {
    case EXIT_FAILURE :
      LOGGER_WRITE(Logger::DEBUG, "Exiting with interface error: invalid invocation of "+std::string(argv[0])+",")
      LOGGER_STOP()
      return EXIT_FAILURE;

    case FILE_ERROR :
      LOGGER_WRITE(Logger::ERROR, "Exiting with interface error: could not open the configuration file.")
      LOGGER_STOP()
      return EXIT_FAILURE;

    case ELEM_MISSING_ERROR :
      LOGGER_WRITE(Logger::ERROR, "Exiting with interface error: missing parameters.")
      LOGGER_STOP()
      return EXIT_FAILURE;

    case LOGIC_ERROR :
      LOGGER_WRITE(Logger::ERROR, "Exiting with interface error: parameter error.")
      LOGGER_STOP()
      return EXIT_FAILURE;

    case STOP_WITH_SUCCESS :
      LOGGER_WRITE(Logger::DEBUG, "Exiting with interface success: successfully parsed input (help message or template produced).")
      LOGGER_STOP()
      return EXIT_SUCCESS;

    default: break;
  }

  // Object declarations
  #if HAVE_STEADY_CLOCK == 1
    const auto simulation_begins = std::chrono::steady_clock::now();
  #else
    const auto simulation_begins = std::chrono::monotonic_clock::now();
  #endif

  LOGGER_WRITE(Logger::DEBUG, "Instantiation of the spa_network_t class.")
  spa_network_t network(prog_params.final_size.value, prog_params.basic_size.value);

  LOGGER_WRITE(Logger::DEBUG, "Instantiation of the random number generator engine "+VAR_PRINT(prog_params.seed.value)+".")
  std::mt19937 engine(prog_params.seed.value);

  // Setup and simulation
  if (prog_params.edge_list.initialized || prog_params.degrees.initialized || prog_params.internal_degrees.initialized || prog_params.projected_internal_degrees.initialized) {
    LOGGER_WRITE(Logger::INFO, "Connectivity information requested: complete simulation mode.")
    network.initialize(prog_params.initial_size.value);
    spa_star(prog_params.p.value, prog_params.q.value, prog_params.r.value, network, engine);
  }
  else if (prog_params.connected_edge_list.initialized || prog_params.connected_degrees.initialized) {
    LOGGER_WRITE(Logger::INFO, "Legacy connectivity information requested: legacy simulation mode.")
    network.set_handle_links_status(false);
    network.initialize(prog_params.initial_size.value);
    spa_basic(prog_params.p.value, prog_params.q.value, network, engine);
  }
  else {
    LOGGER_WRITE(Logger::INFO, "No connectivity information requested: legacy simulation mode.")
    network.set_handle_links_status(false);
    network.initialize(prog_params.initial_size.value);
    spa_basic(prog_params.p.value, prog_params.q.value, network, engine);
  }
  LOGGER_WRITE(Logger::INFO, "Simulation completed.")

  // Timer
  #if HAVE_STEADY_CLOCK == 1
    const auto simulation_ends = std::chrono::steady_clock::now();
  #else
    const auto simulation_ends = std::chrono::monotonic_clock::now();
  #endif
  double simulation_duration = std::chrono::duration_cast<std::chrono::milliseconds>(simulation_ends-simulation_begins).count()/1000.0;

  // Outputs
  LOGGER_WRITE(Logger::INFO, "Processing outputs.")
  process_and_output(network, prog_params, simulation_duration); // this alters the network.

  LOGGER_WRITE(Logger::INFO, "Exiting without error.")
  LOGGER_STOP()

  return EXIT_SUCCESS;
}

max_stars_repo_name: spa-networks/spa

max_issues_repo_name: spa-networks/spa

max_forks_repo_name: spa-networks/spa


text:
#include <boost/enable_shared_from_this.hpp>
#include <boost/make_shared.hpp>
//#include <iostream>
//#include <string>

using namespace std;

class Test : public boost::enable_shared_from_this<Test> {
public :
   Test(int b) : a(b) { }
   int get() { return a; }
   ~Test() { cout << "destroyed" << endl; }
private :
   int a;
};

int main(int argc, char *argv[])
{
  boost::shared_ptr<Test> sp = boost::make_shared<Test>(100);
  // Note : you can't get the this pointer from a normal object, you must use it from a shared pointer.
  boost::shared_ptr<Test> sp1 = sp->shared_from_this();
  cout << sp1->get() << endl;
  cout << sp.use_count() << endl;

  return 0;
}

max_stars_repo_name: chenmingqiang/codesnippet

max_issues_repo_name: chenmingqiang/codesnippet

max_forks_repo_name: chenmingqiang/codesnippet


text:
/*

Copyright (c) 2006-2014, Arvid Norberg, Magnus Jonsson
All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions
are met:

    * Redistributions of source code must retain the above copyright
      notice, this list of conditions and the following disclaimer.
    * Redistributions in binary form must reproduce the above copyright
      notice, this list of conditions and the following disclaimer in
      the documentation and/or other materials provided with the distribution.
    * Neither the name of the author nor the names of its
      contributors may be used to endorse or promote products derived
      from this software without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
POSSIBILITY OF SUCH DAMAGE.

*/

#include <ctime>
#include <algorithm>
#include <set>
#include <deque>
#include <cctype>
#include <algorithm>

#ifdef _MSC_VER
#pragma warning(push, 1)
#endif

#include <boost/limits.hpp>
#include <boost/bind.hpp>

#ifdef _MSC_VER
#pragma warning(pop)
#endif

#include "libtorrent/extensions/ut_pex.hpp"
#include "libtorrent/extensions/ut_metadata.hpp"
#include "libtorrent/extensions/smart_ban.hpp"
#include "libtorrent/peer_id.hpp"
#include "libtorrent/torrent_info.hpp"
#include "libtorrent/tracker_manager.hpp"
#include "libtorrent/bencode.hpp"
#include "libtorrent/hasher.hpp"
#include "libtorrent/entry.hpp"
#include "libtorrent/session.hpp"
#include "libtorrent/fingerprint.hpp"
#include "libtorrent/entry.hpp"
#include "libtorrent/alert_types.hpp"
#include "libtorrent/invariant_check.hpp"
#include "libtorrent/file.hpp"
#include "libtorrent/bt_peer_connection.hpp"
#include "libtorrent/ip_filter.hpp"
#include "libtorrent/socket.hpp"
#include "libtorrent/aux_/session_impl.hpp"
#include "libtorrent/kademlia/dht_tracker.hpp"
#include "libtorrent/natpmp.hpp"
#include "libtorrent/upnp.hpp"
#include "libtorrent/magnet_uri.hpp"

using boost::shared_ptr;
using boost::weak_ptr;
using libtorrent::aux::session_impl;

namespace libtorrent
{

	TORRENT_EXPORT void TORRENT_LINK_TEST_NAME() {}

	// this function returns a session_settings object
	// which will optimize libtorrent for minimum memory
	// usage, with no consideration of performance.
	TORRENT_EXPORT session_settings min_memory_usage()
	{
		session_settings set;

		set.alert_queue_size = 100;

		set.max_allowed_in_request_queue = 100;

		// setting this to a low limit, means more
		// peers are more likely to request from the
		// same piece. Which means fewer partial
		// pieces and fewer entries in the partial
		// piece list
		set.whole_pieces_threshold = 2;
		set.use_parole_mode = false;
		set.prioritize_partial_pieces = true;

		// connect to 5 peers per second
		set.connection_speed = 5;

		// be extra nice on the hard drive when running
		// on embedded devices. This might slow down
		// torrent checking
		set.file_checks_delay_per_block = 5;

		// only have 4 files open at a time
		set.file_pool_size = 4;

		// we want to keep the peer list as small as possible
		set.allow_multiple_connections_per_ip = false;
		set.max_failcount = 2;
		set.inactivity_timeout = 120;

		// whenever a peer has downloaded one block, write
		// it to disk, and don't read anything from the
		// socket until the disk write is complete
		set.max_queued_disk_bytes = 1;

		// don't keep track of all upnp devices, keep
		// the device list small
		set.upnp_ignore_nonrouters = true;

		// never keep more than one 16kB block in
		// the send buffer
		set.send_buffer_watermark = 9;

		// don't use any disk cache
		set.cache_size = 0;
		set.cache_buffer_chunk_size = 1;
		set.use_read_cache = false;
		set.use_disk_read_ahead = false;

		set.close_redundant_connections = true;

		set.max_peerlist_size = 500;
		set.max_paused_peerlist_size = 50;

		// udp trackers are cheaper to talk to
		set.prefer_udp_trackers = true;

		set.max_rejects = 10;

		set.recv_socket_buffer_size = 16 * 1024;
		set.send_socket_buffer_size = 16 * 1024;

		// use less memory when checking pieces
		set.optimize_hashing_for_speed = false;

		// use less memory when reading and writing
		// whole pieces
		set.coalesce_reads = false;
		set.coalesce_writes = false;

		// disallow the buffer size to grow for the uTP socket
		set.utp_dynamic_sock_buf = false;

		// max 'bottled' http receive buffer/url torrent size
		set.max_http_recv_buffer_size = 1024 * 1024;

		return set;
	}

	TORRENT_EXPORT session_settings high_performance_seed()
	{
		session_settings set;

		// allow peers to request a lot of blocks at a time,
		// to be more likely to saturate the bandwidth-delay-
		// product.
		set.max_out_request_queue = 1500;
		set.max_allowed_in_request_queue = 2000;

		// don't throttle TCP, assume there is
		// plenty of bandwidth
		set.mixed_mode_algorithm = session_settings::prefer_tcp;

		// we will probably see a high rate of alerts, make it less
		// likely to loose alerts
		set.alert_queue_size = 50000;

		// allow 500 files open at a time
		set.file_pool_size = 500;

		// don't update access time for each read/write
		set.no_atime_storage = true;

		// as a seed box, we must accept multiple peers behind
		// the same NAT
		set.allow_multiple_connections_per_ip = true;

		// connect to 50 peers per second
		set.connection_speed = 50;

		// allow 8000 peer connections
		set.connections_limit = 8000;

		// allow lots of peers to try to connect simultaneously
		set.listen_queue_size = 200;

		// unchoke many peers
		set.unchoke_slots_limit = 500;

		// we need more DHT capacity to ping more peers
		// candidates before trying to connect
		set.dht_upload_rate_limit = 20000;

		// we're more interested in downloading than seeding
		// only service a read job every 1000 write job (when
		// disk is congested). Presumably on a big box, writes
		// are extremely cheap and reads are relatively expensive
		// so that's the main reason this ratio should be adjusted
		set.read_job_every = 100;

		// use 1 GB of cache
		set.cache_size = 32768 * 2;
		set.use_read_cache = true;
		set.cache_buffer_chunk_size = 128;
		set.read_cache_line_size = 32;
		set.write_cache_line_size = 32;
		set.low_prio_disk = false;
		// one hour expiration
		set.cache_expiry = 60 * 60;
		// this is expensive and could add significant
		// delays when freeing a large number of buffers
		set.lock_disk_cache = false;

		// the max number of bytes pending write before we throttle
		// download rate
		set.max_queued_disk_bytes = 10 * 1024 * 1024;
		// flush write cache in a way to minimize the amount we need to
		// read back once we want to hash-check the piece. i.e. try to
		// flush all blocks in-order
		set.disk_cache_algorithm = session_settings::avoid_readback;

		set.explicit_read_cache = false;
		// prevent fast pieces to interfere with suggested pieces
		// since we unchoke everyone, we don't need fast pieces anyway
		set.allowed_fast_set_size = 0;
		// suggest pieces in the read cache for higher cache hit rate
		set.suggest_mode = session_settings::suggest_read_cache;

		set.close_redundant_connections = true;

		set.max_rejects = 10;

		set.recv_socket_buffer_size = 1024 * 1024;
		set.send_socket_buffer_size = 1024 * 1024;

		set.optimize_hashing_for_speed = true;

		// don't let connections linger for too long
		set.request_timeout = 10;
		set.peer_timeout = 20;
		set.inactivity_timeout = 20;

		set.active_limit = 2000;
		set.active_tracker_limit = 2000;
		set.active_dht_limit = 600;
		set.active_seeds = 2000;

		set.choking_algorithm = session_settings::fixed_slots_choker;

		// in order to be able to deliver very high
		// upload rates, this should be able to cover
		// the bandwidth delay product. Assuming an RTT
		// of 500 ms, and a send rate of 20 MB/s, the upper
		// limit should be 10 MB
		set.send_buffer_watermark = 3 * 1024 * 1024;

		// put 1.5 seconds worth of data in the send buffer
		// this gives the disk I/O more heads-up on disk
		// reads, and can maximize throughput
		set.send_buffer_watermark_factor = 150;

		// always stuff at least 1 MiB down each peer
		// pipe, to quickly ramp up send rates
 		set.send_buffer_low_watermark = 1 * 1024 * 1024;

		// don't retry peers if they fail once. Let them
		// connect to us if they want to
		set.max_failcount = 1;

		// allow the buffer size to grow for the uTP socket
		set.utp_dynamic_sock_buf = true;

		// max 'bottled' http receive buffer/url torrent size
		set.max_http_recv_buffer_size = 6 * 1024 * 1024;

		// the disk cache performs better with the pool allocator
		set.use_disk_cache_pool = true;

		return set;
	}

	// wrapper around a function that's executed in the network thread
	// ans synchronized in the client thread
	template <class R>
	void fun_ret(R* ret, bool* done, condition_variable* e, mutex* m, boost::function<R(void)> f)
	{
		*ret = f();
		mutex::scoped_lock l(*m);
		*done = true;
		e->notify_all();
	}

	void fun_wrap(bool* done, condition_variable* e, mutex* m, boost::function<void(void)> f)
	{
		f();
		mutex::scoped_lock l(*m);
		*done = true;
		e->notify_all();
	}

#define TORRENT_ASYNC_CALL(x) \
	m_impl->m_io_service.dispatch(boost::bind(&session_impl:: x, m_impl.get()))

#define TORRENT_ASYNC_CALL1(x, a1) \
	m_impl->m_io_service.dispatch(boost::bind(&session_impl:: x, m_impl.get(), a1))

#define TORRENT_ASYNC_CALL2(x, a1, a2) \
	m_impl->m_io_service.dispatch(boost::bind(&session_impl:: x, m_impl.get(), a1, a2))

#define TORRENT_ASYNC_CALL3(x, a1, a2, a3) \
	m_impl->m_io_service.dispatch(boost::bind(&session_impl:: x, m_impl.get(), a1, a2, a3))

#define TORRENT_WAIT \
	mutex::scoped_lock l(m_impl->mut); \
	while (!done) { m_impl->cond.wait(l); };

#define TORRENT_SYNC_CALL(x) \
	bool done = false; \
	m_impl->m_io_service.dispatch(boost::bind(&fun_wrap, &done, &m_impl->cond, &m_impl->mut, boost::function<void(void)>(boost::bind(&session_impl:: x, m_impl.get())))); \
	TORRENT_WAIT

#define TORRENT_SYNC_CALL1(x, a1) \
	bool done = false; \
	m_impl->m_io_service.dispatch(boost::bind(&fun_wrap, &done, &m_impl->cond, &m_impl->mut, boost::function<void(void)>(boost::bind(&session_impl:: x, m_impl.get(), a1)))); \
	TORRENT_WAIT

#define TORRENT_SYNC_CALL2(x, a1, a2) \
	bool done = false; \
	m_impl->m_io_service.dispatch(boost::bind(&fun_wrap, &done, &m_impl->cond, &m_impl->mut, boost::function<void(void)>(boost::bind(&session_impl:: x, m_impl.get(), a1, a2)))); \
	TORRENT_WAIT

#define TORRENT_SYNC_CALL3(x, a1, a2, a3) \
	bool done = false; \
	m_impl->m_io_service.dispatch(boost::bind(&fun_wrap, &done, &m_impl->cond, &m_impl->mut, boost::function<void(void)>(boost::bind(&session_impl:: x, m_impl.get(), a1, a2, a3)))); \
	TORRENT_WAIT

#define TORRENT_SYNC_CALL4(x, a1, a2, a3, a4) \
	bool done = false; \
	m_impl->m_io_service.dispatch(boost::bind(&fun_wrap, &done, &m_impl->cond, &m_impl->mut, boost::function<void(void)>(boost::bind(&session_impl:: x, m_impl.get(), a1, a2, a3, a4)))); \
	TORRENT_WAIT

#define TORRENT_SYNC_CALL_RET(type, x) \
	bool done = false; \
	type r; \
	m_impl->m_io_service.dispatch(boost::bind(&fun_ret<type >, &r, &done, &m_impl->cond, &m_impl->mut, boost::function<type(void)>(boost::bind(&session_impl:: x, m_impl.get())))); \
	TORRENT_WAIT

#define TORRENT_SYNC_CALL_RET1(type, x, a1) \
	bool done = false; \
	type r; \
	m_impl->m_io_service.dispatch(boost::bind(&fun_ret<type >, &r, &done, &m_impl->cond, &m_impl->mut, boost::function<type(void)>(boost::bind(&session_impl:: x, m_impl.get(), a1)))); \
	TORRENT_WAIT

#define TORRENT_SYNC_CALL_RET2(type, x, a1, a2) \
	bool done = false; \
	type r; \
	m_impl->m_io_service.dispatch(boost::bind(&fun_ret<type >, &r, &done, &m_impl->cond, &m_impl->mut, boost::function<type(void)>(boost::bind(&session_impl:: x, m_impl.get(), a1, a2)))); \
	TORRENT_WAIT

#define TORRENT_SYNC_CALL_RET3(type, x, a1, a2, a3) \
	bool done = false; \
	type r; \
	m_impl->m_io_service.dispatch(boost::bind(&fun_ret<type >, &r, &done, &m_impl->cond, &m_impl->mut, boost::function<type(void)>(boost::bind(&session_impl:: x, m_impl.get(), a1, a2, a3)))); \
	TORRENT_WAIT

#ifndef TORRENT_CFG
#error TORRENT_CFG is not defined!
#endif

	// this is a dummy function that's exported and named based
	// on the configuration. The session.hpp file will reference
	// it and if the library and the client are built with different
	// configurations this will give a link error
	void TORRENT_EXPORT TORRENT_CFG() {}

#if defined _MSC_VER && defined TORRENT_DEBUG
	static void straight_to_debugger(unsigned int, _EXCEPTION_POINTERS*)
	{ throw; }
#endif

	void session::init(std::pair<int, int> listen_range, char const* listen_interface
		, fingerprint const& id, boost::uint32_t alert_mask)
	{
#if defined _MSC_VER && defined TORRENT_DEBUG
		// workaround for microsofts
		// hardware exceptions that makes
		// it hard to debug stuff
		::_set_se_translator(straight_to_debugger);
#endif

		m_impl.reset(new session_impl(listen_range, id, listen_interface, alert_mask));
	}

	void session::set_log_path(std::string const& p)
	{
#if defined TORRENT_VERBOSE_LOGGING || defined TORRENT_LOGGING \
	|| defined TORRENT_ERROR_LOGGING
		m_impl->set_log_path(p);
#endif
	}

	void session::start(int flags)
	{
#ifndef TORRENT_DISABLE_EXTENSIONS
		if (flags & add_default_plugins)
		{
			add_extension(create_ut_pex_plugin);
			add_extension(create_ut_metadata_plugin);
			add_extension(create_smart_ban_plugin);
		}
#endif

		m_impl->start_session();

		if (flags & start_default_features)
		{
			start_upnp();
			start_natpmp();
#ifndef TORRENT_DISABLE_DHT
			start_dht();
#endif
			start_lsd();
		}
	}

	session::~session()
	{
		TORRENT_ASSERT(m_impl);
		// if there is at least one destruction-proxy
		// abort the session and let the destructor
		// of the proxy to syncronize
		if (!m_impl.unique())
		{
			TORRENT_ASYNC_CALL(abort);
		}
	}

	void session::save_state(entry& e, boost::uint32_t flags) const
	{
		TORRENT_SYNC_CALL2(save_state, &e, flags);
	}

	void session::load_state(lazy_entry const& e)
	{
		// this needs to be synchronized since the lifespan
		// of e is tied to the caller
		TORRENT_SYNC_CALL1(load_state, &e);
	}

	feed_handle session::add_feed(feed_settings const& feed)
	{
		// if you have auto-download enabled, you must specify a download directory!
		TORRENT_ASSERT_PRECOND(!feed.auto_download || !feed.add_args.save_path.empty());
		TORRENT_SYNC_CALL_RET1(feed_handle, add_feed, feed);
		return r;
	}

	void session::remove_feed(feed_handle h)
	{
		TORRENT_ASYNC_CALL1(remove_feed, h);
	}

	void session::get_feeds(std::vector<feed_handle>& f) const
	{
		f.clear();
		TORRENT_SYNC_CALL1(get_feeds, &f);
	}

	void session::add_extension(boost::function<boost::shared_ptr<torrent_plugin>(torrent*, void*)> ext)
	{
#ifndef TORRENT_DISABLE_EXTENSIONS
		TORRENT_ASYNC_CALL1(add_extension, ext);
#endif
	}

	void session::add_extension(boost::shared_ptr<plugin> ext)
	{
#ifndef TORRENT_DISABLE_EXTENSIONS
		TORRENT_ASYNC_CALL1(add_ses_extension, ext);
#endif
	}

	void session::load_asnum_db(char const* file)
	{
#ifndef TORRENT_DISABLE_GEO_IP
		TORRENT_ASYNC_CALL1(load_asnum_db, std::string(file));
#endif
	}

	void session::load_country_db(char const* file)
	{
#ifndef TORRENT_DISABLE_GEO_IP
		TORRENT_ASYNC_CALL1(load_country_db, std::string(file));
#endif
	}

	int session::as_for_ip(address const& addr)
	{
#ifndef TORRENT_DISABLE_GEO_IP
		return m_impl->as_for_ip(addr);
#else
		return 0;
#endif
	}

#if TORRENT_USE_WSTRING
#ifndef TORRENT_NO_DEPRECATE
	void session::load_asnum_db(wchar_t const* file)
	{
#ifndef TORRENT_DISABLE_GEO_IP
		TORRENT_ASYNC_CALL1(load_asnum_dbw, std::wstring(file));
#endif
	}

	void session::load_country_db(wchar_t const* file)
	{
#ifndef TORRENT_DISABLE_GEO_IP
		TORRENT_ASYNC_CALL1(load_country_dbw, std::wstring(file));
#endif
	}
#endif // TORRENT_NO_DEPRECATE
#endif // TORRENT_USE_WSTRING

#ifndef TORRENT_NO_DEPRECATE
	void session::load_state(entry const& ses_state)
	{
		if (ses_state.type() == entry::undefined_t) return;
		std::vector<char> buf;
		bencode(std::back_inserter(buf), ses_state);
		lazy_entry e;
		error_code ec;
#if defined TORRENT_DEBUG || TORRENT_RELEASE_ASSERTS || !defined BOOST_NO_EXCEPTIONS
		int ret =
#endif
		lazy_bdecode(&buf[0], &buf[0] + buf.size(), e, ec);

		TORRENT_ASSERT(ret == 0);
#ifndef BOOST_NO_EXCEPTIONS
		if (ret != 0) throw libtorrent_exception(ec);
#endif
		TORRENT_SYNC_CALL1(load_state, &e);
	}

	entry session::state() const
	{
		entry ret;
		TORRENT_SYNC_CALL2(save_state, &ret, 0xffffffff);
		return ret;
	}
#endif

	void session::set_ip_filter(ip_filter const& f)
	{
		TORRENT_ASYNC_CALL1(set_ip_filter, f);
	}

	ip_filter session::get_ip_filter() const
	{
		TORRENT_SYNC_CALL_RET(ip_filter, get_ip_filter);
		return r;
	}

	void session::set_port_filter(port_filter const& f)
	{
		TORRENT_ASYNC_CALL1(set_port_filter, f);
	}

	void session::set_peer_id(peer_id const& id)
	{
		TORRENT_ASYNC_CALL1(set_peer_id, id);
	}

	peer_id session::id() const
	{
		TORRENT_SYNC_CALL_RET(peer_id, get_peer_id);
		return r;
	}

	io_service& session::get_io_service()
	{
		return m_impl->m_io_service;
	}

	void session::set_key(int key)
	{
		TORRENT_ASYNC_CALL1(set_key, key);
	}

	void session::get_torrent_status(std::vector<torrent_status>* ret
		, boost::function<bool(torrent_status const&)> const& pred
		, boost::uint32_t flags) const
	{
		TORRENT_SYNC_CALL3(get_torrent_status, ret, boost::ref(pred), flags);
	}

	void session::refresh_torrent_status(std::vector<torrent_status>* ret
		, boost::uint32_t flags) const
	{
		TORRENT_SYNC_CALL2(refresh_torrent_status, ret, flags);
	}

	void session::post_torrent_updates()
	{
		TORRENT_ASYNC_CALL(post_torrent_updates);
	}

	std::vector<torrent_handle> session::get_torrents() const
	{
		TORRENT_SYNC_CALL_RET(std::vector<torrent_handle>, get_torrents);
		return r;
	}

	torrent_handle session::find_torrent(sha1_hash const& info_hash) const
	{
		TORRENT_SYNC_CALL_RET1(torrent_handle, find_torrent_handle, info_hash);
		return r;
	}

#ifndef BOOST_NO_EXCEPTIONS
	torrent_handle session::add_torrent(add_torrent_params const& params)
	{
		error_code ec;
		TORRENT_SYNC_CALL_RET2(torrent_handle, add_torrent, params, boost::ref(ec));
		if (ec) throw libtorrent_exception(ec);
		return r;
	}
#endif

	torrent_handle session::add_torrent(add_torrent_params const& params, error_code& ec)
	{
		ec.clear();
		TORRENT_SYNC_CALL_RET2(torrent_handle, add_torrent, params, boost::ref(ec));
		return r;
	}

	void session::async_add_torrent(add_torrent_params const& params)
	{
		add_torrent_params* p = new add_torrent_params(params);
#ifndef TORRENT_NO_DEPRECATE
		if (params.tracker_url)
		{
			p->trackers.push_back(params.tracker_url);
			p->tracker_url = NULL;
		}
#endif
		TORRENT_ASYNC_CALL1(async_add_torrent, p);
	}

#ifndef BOOST_NO_EXCEPTIONS
#ifndef TORRENT_NO_DEPRECATE
	// if the torrent already exists, this will throw duplicate_torrent
	torrent_handle session::add_torrent(
		torrent_info const& ti
		, std::string const& save_path
		, entry const& resume_data
		, storage_mode_t storage_mode
		, bool paused
		, storage_constructor_type sc)
	{
		boost::intrusive_ptr<torrent_info> tip(new torrent_info(ti));
		add_torrent_params p(sc);
		p.ti = tip;
		p.save_path = save_path;
		if (resume_data.type() != entry::undefined_t)
		{
			bencode(std::back_inserter(p.resume_data), resume_data);
		}
		p.storage_mode = storage_mode;
		p.paused = paused;
		return add_torrent(p);
	}

	torrent_handle session::add_torrent(
		boost::intrusive_ptr<torrent_info> ti
		, std::string const& save_path
		, entry const& resume_data
		, storage_mode_t storage_mode
		, bool paused
		, storage_constructor_type sc
		, void* userdata)
	{
		add_torrent_params p(sc);
		p.ti = ti;
		p.save_path = save_path;
		if (resume_data.type() != entry::undefined_t)
		{
			bencode(std::back_inserter(p.resume_data), resume_data);
		}
		p.storage_mode = storage_mode;
		p.paused = paused;
		p.userdata = userdata;
		return add_torrent(p);
	}

	torrent_handle session::add_torrent(
		char const* tracker_url
		, sha1_hash const& info_hash
		, char const* name
		, std::string const& save_path
		, entry const& e
		, storage_mode_t storage_mode
		, bool paused
		, storage_constructor_type sc
		, void* userdata)
	{
		add_torrent_params p(sc);
		p.tracker_url = tracker_url;
		p.info_hash = info_hash;
		p.save_path = save_path;
		p.paused = paused;
		p.userdata = userdata;
		return add_torrent(p);
	}
#endif // TORRENT_NO_DEPRECATE
#endif // BOOST_NO_EXCEPTIONS

	void session::remove_torrent(const torrent_handle& h, int options)
	{
		if (!h.is_valid())
#ifdef BOOST_NO_EXCEPTIONS
			return;
#else
			throw_invalid_handle();
#endif
		TORRENT_ASYNC_CALL2(remove_torrent, h, options);
	}

#ifndef TORRENT_NO_DEPRECATE
	bool session::listen_on(
		std::pair<int, int> const& port_range
		, const char* net_interface, int flags)
	{
		error_code ec;
		TORRENT_SYNC_CALL4(listen_on, port_range, boost::ref(ec), net_interface, flags);
		return !!ec;
	}
#endif

	void session::listen_on(
		std::pair<int, int> const& port_range
		, error_code& ec
		, const char* net_interface, int flags)
	{
		TORRENT_SYNC_CALL4(listen_on, port_range, boost::ref(ec), net_interface, flags);
	}

	unsigned short session::listen_port() const
	{
		TORRENT_SYNC_CALL_RET(unsigned short, listen_port);
		return r;
	}

	unsigned short session::ssl_listen_port() const
	{
		TORRENT_SYNC_CALL_RET(unsigned short, ssl_listen_port);
		return r;
	}

	session_status session::status() const
	{
		TORRENT_SYNC_CALL_RET(session_status, status);
		return r;
	}

	void session::pause()
	{
		TORRENT_ASYNC_CALL(pause);
	}

	void session::resume()
	{
		TORRENT_ASYNC_CALL(resume);
	}

	bool session::is_paused() const
	{
		TORRENT_SYNC_CALL_RET(bool, is_paused);
		return r;
	}

	void session::get_cache_info(sha1_hash const& ih
		, std::vector<cached_piece_info>& ret) const
	{
		m_impl->m_disk_thread.get_cache_info(ih, ret);
	}

	cache_status session::get_cache_status() const
	{
		return m_impl->m_disk_thread.status();
	}

	void session::start_dht()
	{
#ifndef TORRENT_DISABLE_DHT
		// the state is loaded in load_state()
		TORRENT_ASYNC_CALL(start_dht);
#endif
	}

	void session::stop_dht()
	{
#ifndef TORRENT_DISABLE_DHT
		TORRENT_ASYNC_CALL(stop_dht);
#endif
	}

	void session::set_dht_settings(dht_settings const& settings)
	{
#ifndef TORRENT_DISABLE_DHT
		TORRENT_ASYNC_CALL1(set_dht_settings, settings);
#endif
	}

	dht_settings session::get_dht_settings() const
	{
#ifndef TORRENT_DISABLE_DHT
		TORRENT_SYNC_CALL_RET(dht_settings, get_dht_settings);
#else
		dht_settings r;
#endif
		return r;
	}

#ifndef TORRENT_NO_DEPRECATE
	void session::start_dht(entry const& startup_state)
	{
#ifndef TORRENT_DISABLE_DHT
		TORRENT_ASYNC_CALL1(start_dht, startup_state);
#endif
	}

	entry session::dht_state() const
	{
#ifndef TORRENT_DISABLE_DHT
		TORRENT_SYNC_CALL_RET(entry, dht_state);
		return r;
#else
		return entry();
#endif
	}
#endif // TORRENT_NO_DEPRECATE

	void session::add_dht_node(std::pair<std::string, int> const& node)
	{
#ifndef TORRENT_DISABLE_DHT
		TORRENT_ASYNC_CALL1(add_dht_node_name, node);
#endif
	}

	void session::add_dht_router(std::pair<std::string, int> const& node)
	{
#ifndef TORRENT_DISABLE_DHT
		TORRENT_ASYNC_CALL1(add_dht_router, node);
#endif
	}

	bool session::is_dht_running() const
	{
#ifndef TORRENT_DISABLE_DHT
		TORRENT_SYNC_CALL_RET(bool, is_dht_running);
		return r;
#else
		return false;
#endif
	}

	void session::dht_get_item(sha1_hash const& target)
	{
#ifndef TORRENT_DISABLE_DHT
		TORRENT_ASYNC_CALL1(dht_get_immutable_item, target);
#endif
	}

	void session::dht_get_item(boost::array<char, 32> key
		, std::string salt)
	{
#ifndef TORRENT_DISABLE_DHT
		TORRENT_ASYNC_CALL2(dht_get_mutable_item, key, salt);
#endif
	}

	sha1_hash session::dht_put_item(entry data)
	{
		std::vector<char> buf;
		bencode(std::back_inserter(buf), data);
		sha1_hash ret = hasher(&buf[0], buf.size()).final();

#ifndef TORRENT_DISABLE_DHT
		TORRENT_ASYNC_CALL2(dht_put_item, data, ret);
#endif
		return ret;
	}

	void session::dht_put_item(boost::array<char, 32> key
		, boost::function<void(entry&, boost::array<char,64>&
			, boost::uint64_t&, std::string const&)> cb
		, std::string salt)
	{
#ifndef TORRENT_DISABLE_DHT
		TORRENT_ASYNC_CALL3(dht_put_mutable_item, key, cb, salt);
#endif
	}

	void session::set_pe_settings(pe_settings const& settings)
	{
#ifndef TORRENT_DISABLE_ENCRYPTION
		TORRENT_ASYNC_CALL1(set_pe_settings, settings);
#endif
	}

	pe_settings session::get_pe_settings() const
	{
#ifndef TORRENT_DISABLE_ENCRYPTION
		TORRENT_SYNC_CALL_RET(pe_settings, get_pe_settings);
#else
		pe_settings r;
#endif
		return r;
	}

	bool session::is_listening() const
	{
		TORRENT_SYNC_CALL_RET(bool, is_listening);
		return r;
	}

	void session::set_settings(session_settings const& s)
	{
		TORRENT_ASYNC_CALL1(set_settings, s);
	}

	session_settings session::settings() const
	{
		TORRENT_SYNC_CALL_RET(session_settings, settings);
		return r;
	}

	void session::set_proxy(proxy_settings const& s)
	{
		TORRENT_ASYNC_CALL1(set_proxy, s);
	}

	proxy_settings session::proxy() const
	{
		TORRENT_SYNC_CALL_RET(proxy_settings, proxy);
		return r;
	}

#ifndef TORRENT_NO_DEPRECATE
	void session::set_peer_proxy(proxy_settings const& s)
	{
		TORRENT_ASYNC_CALL1(set_peer_proxy, s);
	}

	void session::set_web_seed_proxy(proxy_settings const& s)
	{
		TORRENT_ASYNC_CALL1(set_web_seed_proxy, s);
	}

	void session::set_tracker_proxy(proxy_settings const& s)
	{
		TORRENT_ASYNC_CALL1(set_tracker_proxy, s);
	}

	proxy_settings session::peer_proxy() const
	{
		TORRENT_SYNC_CALL_RET(proxy_settings, peer_proxy);
		return r;
	}

	proxy_settings session::web_seed_proxy() const
	{
		TORRENT_SYNC_CALL_RET(proxy_settings, web_seed_proxy);
		return r;
	}

	proxy_settings session::tracker_proxy() const
	{
		TORRENT_SYNC_CALL_RET(proxy_settings, tracker_proxy);
		return r;
	}

	void session::set_dht_proxy(proxy_settings const& s)
	{
#ifndef TORRENT_DISABLE_DHT
		TORRENT_ASYNC_CALL1(set_dht_proxy, s);
#endif
	}

	proxy_settings session::dht_proxy() const
	{
#ifndef TORRENT_DISABLE_DHT
		TORRENT_SYNC_CALL_RET(proxy_settings, dht_proxy);
		return r;
#else
		return proxy_settings();
#endif
	}
#endif // TORRENT_NO_DEPRECATE

	void session::set_i2p_proxy(proxy_settings const& s)
	{
#if TORRENT_USE_I2P
		TORRENT_ASYNC_CALL1(set_i2p_proxy, s);
#endif
	}

	proxy_settings session::i2p_proxy() const
	{
#if TORRENT_USE_I2P
		TORRENT_SYNC_CALL_RET(proxy_settings, i2p_proxy);
#else
		proxy_settings r;
#endif
		return r;
	}

#ifdef TORRENT_STATS
	void session::enable_stats_logging(bool s)
	{
		TORRENT_ASYNC_CALL1(enable_stats_logging, s);
	}
#endif

#ifndef TORRENT_NO_DEPRECATE
	int session::max_uploads() const
	{
		TORRENT_SYNC_CALL_RET(int, max_uploads);
		return r;
	}

	void session::set_max_uploads(int limit)
	{
		TORRENT_ASYNC_CALL1(set_max_uploads, limit);
	}

	int session::max_connections() const
	{
		TORRENT_SYNC_CALL_RET(int, max_connections);
		return r;
	}

	void session::set_max_connections(int limit)
	{
		TORRENT_ASYNC_CALL1(set_max_connections, limit);
	}

	int session::max_half_open_connections() const
	{
		TORRENT_SYNC_CALL_RET(int, max_half_open_connections);
		return r;
	}

	void session::set_max_half_open_connections(int limit)
	{
		TORRENT_ASYNC_CALL1(set_max_half_open_connections, limit);
	}

	int session::local_upload_rate_limit() const
	{
		TORRENT_SYNC_CALL_RET(int, local_upload_rate_limit);
		return r;
	}

	int session::local_download_rate_limit() const
	{
		TORRENT_SYNC_CALL_RET(int, local_download_rate_limit);
		return r;
	}

	int session::upload_rate_limit() const
	{
		TORRENT_SYNC_CALL_RET(int, upload_rate_limit);
		return r;
	}

	int session::download_rate_limit() const
	{
		TORRENT_SYNC_CALL_RET(int, download_rate_limit);
		return r;
	}

	void session::set_local_upload_rate_limit(int bytes_per_second)
	{
		TORRENT_ASYNC_CALL1(set_local_upload_rate_limit, bytes_per_second);
	}

	void session::set_local_download_rate_limit(int bytes_per_second)
	{
		TORRENT_ASYNC_CALL1(set_local_download_rate_limit, bytes_per_second);
	}

	void session::set_upload_rate_limit(int bytes_per_second)
	{
		TORRENT_ASYNC_CALL1(set_upload_rate_limit, bytes_per_second);
	}

	void session::set_download_rate_limit(int bytes_per_second)
	{
		TORRENT_ASYNC_CALL1(set_download_rate_limit, bytes_per_second);
	}

	int session::num_uploads() const
	{
		TORRENT_SYNC_CALL_RET(int, num_uploads);
		return r;
	}

	int session::num_connections() const
	{
		TORRENT_SYNC_CALL_RET(int, num_connections);
		return r;
	}
#endif // TORRENT_NO_DEPRECATE

	void session::set_alert_dispatch(boost::function<void(std::auto_ptr<alert>)> const& fun)
	{
		TORRENT_ASYNC_CALL1(set_alert_dispatch, fun);
	}

	std::auto_ptr<alert> session::pop_alert()
	{
		return m_impl->pop_alert();
	}

	void session::pop_alerts(std::deque<alert*>* alerts)
	{
		for (std::deque<alert*>::iterator i = alerts->begin()
			, end(alerts->end()); i != end; ++i)
			delete *i;
		alerts->clear();
		m_impl->pop_alerts(alerts);
	}

	alert const* session::wait_for_alert(time_duration max_wait)
	{
		return m_impl->wait_for_alert(max_wait);
	}

	void session::set_alert_mask(boost::uint32_t m)
	{
		TORRENT_ASYNC_CALL1(set_alert_mask, m);
	}

#ifndef TORRENT_NO_DEPRECATE
	size_t session::set_alert_queue_size_limit(size_t queue_size_limit_)
	{
		TORRENT_SYNC_CALL_RET1(size_t, set_alert_queue_size_limit, queue_size_limit_);
		return r;
	}

	void session::set_severity_level(alert::severity_t s)
	{
		int m = 0;
		switch (s)
		{
			case alert::debug: m = alert::all_categories; break;
			case alert::info: m = alert::all_categories & ~(alert::debug_notification
				| alert::progress_notification | alert::dht_notification); break;
			case alert::warning: m = alert::all_categories & ~(alert::debug_notification
				| alert::status_notification | alert::progress_notification
				| alert::dht_notification); break;
			case alert::critical: m = alert::error_notification | alert::storage_notification; break;
			case alert::fatal: m = alert::error_notification; break;
			default: break;
		}

		TORRENT_ASYNC_CALL1(set_alert_mask, m);
	}
#endif

	void session::start_lsd()
	{
		TORRENT_ASYNC_CALL(start_lsd);
	}

	void session::start_natpmp()
	{
		TORRENT_ASYNC_CALL(start_natpmp);
	}

	void session::start_upnp()
	{
		TORRENT_ASYNC_CALL(start_upnp);
	}

	int session::add_port_mapping(protocol_type t, int external_port, int local_port)
	{
		TORRENT_SYNC_CALL_RET3(int, add_port_mapping, int(t), external_port, local_port);
		return r;
	}

	void session::delete_port_mapping(int handle)
	{
		TORRENT_ASYNC_CALL1(delete_port_mapping, handle);
	}

	void session::stop_lsd()
	{
		TORRENT_ASYNC_CALL(stop_lsd);
	}

	void session::stop_natpmp()
	{
		TORRENT_ASYNC_CALL(stop_natpmp);
	}

	void session::stop_upnp()
	{
		TORRENT_ASYNC_CALL(stop_upnp);
	}

	connection_queue& session::get_connection_queue()
	{
		return m_impl->m_half_open;
	}

	session_settings::session_settings(std::string const& user_agent_)
		: version(LIBTORRENT_VERSION_NUM)
		, user_agent(user_agent_)
		, tracker_completion_timeout(30)
		, tracker_receive_timeout(10)
		, stop_tracker_timeout(5)
		, tracker_maximum_response_length(1024*1024)
		, piece_timeout(20)
		, request_timeout(50)
		, request_queue_time(3)
		, max_allowed_in_request_queue(500)
		, max_out_request_queue(500)
		, whole_pieces_threshold(20)
		, peer_timeout(120)
		, urlseed_timeout(20)
		, urlseed_pipeline_size(5)
		, urlseed_wait_retry(30)
		, file_pool_size(40)
		, allow_multiple_connections_per_ip(false)
		, max_failcount(3)
		, min_reconnect_time(60)
		, peer_connect_timeout(15)
		, ignore_limits_on_local_network(true)
		, connection_speed(6)
		, send_redundant_have(true)
		, lazy_bitfields(false)
		, inactivity_timeout(600)
		, unchoke_interval(15)
		, optimistic_unchoke_interval(30)
		, num_want(200)
		, initial_picker_threshold(4)
		, allowed_fast_set_size(10)
		, suggest_mode(no_piece_suggestions)
		, max_queued_disk_bytes(1024 * 1024)
		, max_queued_disk_bytes_low_watermark(0)
		, handshake_timeout(10)
		, use_dht_as_fallback(false)
		, free_torrent_hashes(true)
		, upnp_ignore_nonrouters(false)
		, send_buffer_low_watermark(512)
		, send_buffer_watermark(500 * 1024)
		, send_buffer_watermark_factor(50)
#ifndef TORRENT_NO_DEPRECATE
		// deprecated in 0.16
		, auto_upload_slots(true)
		, auto_upload_slots_rate_based(true)
#endif
		, choking_algorithm(fixed_slots_choker)
		, seed_choking_algorithm(round_robin)
		, use_parole_mode(true)
		, cache_size(1024)
		, cache_buffer_chunk_size(16)
		, cache_expiry(300)
		, use_read_cache(true)
		, explicit_read_cache(0)
		, explicit_cache_interval(30)
		, disk_io_write_mode(0)
		, disk_io_read_mode(0)
		, coalesce_reads(false)
		, coalesce_writes(false)
		, outgoing_ports(0,0)
		, peer_tos(0)
		, active_downloads(3)
		, active_seeds(5)
		, active_dht_limit(88) // don't announce more than once every 40 seconds
		, active_tracker_limit(1600) // don't announce to trackers more than once every 1.125 seconds
		, active_lsd_limit(60) // don't announce to local network more than once every 5 seconds
		, active_limit(15)
		, auto_manage_prefer_seeds(false)
		, dont_count_slow_torrents(true)
		, auto_manage_interval(30)
		, share_ratio_limit(2.f)
		, seed_time_ratio_limit(7.f)
		, seed_time_limit(24 * 60 * 60) // 24 hours
		, peer_turnover_interval(300)
		, peer_turnover(2 / 50.f)
		, peer_turnover_cutoff(.9f)
		, close_redundant_connections(true)
		, auto_scrape_interval(1800)
		, auto_scrape_min_interval(300)
		, max_peerlist_size(4000)
		, max_paused_peerlist_size(4000)
		, min_announce_interval(5 * 60)
		, prioritize_partial_pieces(false)
		, auto_manage_startup(60)
		, rate_limit_ip_overhead(true)
		, announce_to_all_trackers(false)
		, announce_to_all_tiers(false)
		, prefer_udp_trackers(true)
		, strict_super_seeding(false)
		, seeding_piece_quota(20)
#ifdef TORRENT_WINDOWS
		, max_sparse_regions(30000)
#else
		, max_sparse_regions(0)
#endif
		, lock_disk_cache(false)
		, max_rejects(50)
		, recv_socket_buffer_size(0)
		, send_socket_buffer_size(0)
		, optimize_hashing_for_speed(true)
		, file_checks_delay_per_block(0)
		, disk_cache_algorithm(avoid_readback)
		, read_cache_line_size(32)
		, write_cache_line_size(32)
		, optimistic_disk_retry(10 * 60)
		, disable_hash_checks(false)
		, allow_reordered_disk_operations(true)
		, allow_i2p_mixed(false)
		, max_suggest_pieces(10)
		, drop_skipped_requests(false)
		, low_prio_disk(true)
		, local_service_announce_interval(5 * 60)
		, dht_announce_interval(15 * 60)
		, udp_tracker_token_expiry(60)
		, volatile_read_cache(false)
		, guided_read_cache(false)
		, default_cache_min_age(1)
		, num_optimistic_unchoke_slots(0)
		, no_atime_storage(true)
		, default_est_reciprocation_rate(16000)
		, increase_est_reciprocation_rate(20)
		, decrease_est_reciprocation_rate(3)
		, incoming_starts_queued_torrents(false)
		, report_true_downloaded(false)
		, strict_end_game_mode(true)
		, broadcast_lsd(true)
		, enable_outgoing_utp(true)
		, enable_incoming_utp(true)
		, enable_outgoing_tcp(true)
		, enable_incoming_tcp(true)
		, max_pex_peers(50)
		, ignore_resume_timestamps(false)
		, no_recheck_incomplete_resume(false)
		, anonymous_mode(false)
		, force_proxy(false)
		, tick_interval(500)
		, report_web_seed_downloads(true)
		, share_mode_target(3)
		, upload_rate_limit(0)
		, download_rate_limit(0)
		, local_upload_rate_limit(0)
		, local_download_rate_limit(0)
		, dht_upload_rate_limit(4000)
		, unchoke_slots_limit(8)
		, half_open_limit(0)
		, connections_limit(200)
		, connections_slack(10)
		, utp_target_delay(100) // milliseconds
		, utp_gain_factor(3000) // bytes per rtt
		, utp_min_timeout(500) // milliseconds
		, utp_syn_resends(2)
		, utp_fin_resends(2)
		, utp_num_resends(3)
		, utp_connect_timeout(3000) // milliseconds
#ifndef TORRENT_NO_DEPRECATE
		, utp_delayed_ack(0) // milliseconds
#endif
		, utp_dynamic_sock_buf(false) // this doesn't seem quite reliable yet
		, utp_loss_multiplier(50) // specified in percent
		, mixed_mode_algorithm(peer_proportional)
		, rate_limit_utp(true)
		, listen_queue_size(5)
		, announce_double_nat(false)
		, torrent_connect_boost(10)
		, seeding_outgoing_connections(true)
		, no_connect_privileged_ports(true)
		, alert_queue_size(6000)
		, max_metadata_size(3*1024*1024)
		, smooth_connects(true)
		, always_send_user_agent(false)
		, apply_ip_filter_to_trackers(true)
		, read_job_every(10)
		, use_disk_read_ahead(true)
		, lock_files(false)
		, ssl_listen(4433)
		, tracker_backoff(250)
		, ban_web_seeds(true)
		, max_http_recv_buffer_size(4*1024*1024)
		, support_share_mode(true)
		, support_merkle_torrents(false)
		, report_redundant_bytes(true)
		, use_disk_cache_pool(false)
		, inactive_down_rate(2048)
		, inactive_up_rate(2048)
	{}

	session_settings::~session_settings() {}
}

max_stars_repo_name: eliransapir/PopcornTorrent


text:
//  Copyright (c) 2001-2008 Hartmut Kaiser
//
//  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)

#if !defined(BOOST_SPIRIT_KARMA_VERBATIM_MAR_02_2007_0303PM)
#define BOOST_SPIRIT_KARMA_VERBATIM_MAR_02_2007_0303PM

#if defined(_MSC_VER) && (_MSC_VER >= 1020)
#pragma once      // MS compatible compilers support #pragma once
#endif

#include <boost/spirit/home/support/unused.hpp>
#include <boost/spirit/home/support/attribute_of.hpp>
#include <boost/spirit/home/karma/domain.hpp>
#include <boost/fusion/include/at.hpp>
#include <boost/fusion/include/value_at.hpp>

namespace boost { namespace spirit { namespace karma
{
    ///////////////////////////////////////////////////////////////////////////
    //  The verbatim generator is used for verbatim[...] directives.
    ///////////////////////////////////////////////////////////////////////////
    struct verbatim
    {
        template <typename Component, typename Context, typename Unused>
        struct attribute
          : traits::attribute_of<
                karma::domain,
                typename result_of::right<Component>::type,
                Context
            >
        {
        };

        template <typename Component, typename OutputIterator,
            typename Context, typename Delimiter, typename Parameter>
        static bool
        generate(Component const& component, OutputIterator& sink,
            Context& ctx, Delimiter const& d, Parameter const& param)
        {
            //  the verbatim generator simply dispatches to the embedded
            //  generator while supplying unused as the new delimiter
            //  to avoid delimiting down the generator stream
            typedef typename
                spirit::result_of::right<Component>::type::director
            director;

            if (director::generate(spirit::right(component), sink, ctx,
                    unused, param))
            {
                karma::delimit(sink, d);           // always do post-delimiting
                return true;
            }
            return false;
        }

        template <typename Component, typename Context>
        static std::string what(Component const& component, Context const& ctx)
        {
            std::string result = "verbatim[";

            typedef typename
                spirit::result_of::right<Component>::type::director
            director;

            result += director::what(spirit::right(component), ctx);
            result += "]";
            return result;
        }
    };

}}}

#endif

ext: hpp

max_issues_repo_name: cpmech/vismatrix

max_forks_repo_name: cpmech/vismatrix


text:
#include <iostream>
#include <unistd.h>
#include <gtest/gtest.h>
#include <poll.h>
#include <chrono>
#include <boost/thread.hpp>
#include "coroutine.h"
using namespace std;
using namespace std::chrono;
using namespace co;

///poll test points:
// 1.timeout == 0
// 2.timeout == -1
// 3.timeout == 1
// 4.all of fds are valid.
// 5.all of fds are invalid.
// 6.some -1 in fds.
//X7.some file_fd in fds.
//X8.occurred ERR events
// 9.timeout return.
// 10.multi threads.

static int fill_send_buffer(int fd)
{
    static char* buf = new char[10240];
    int c = 0;
    for (;;) {
        pollfd pfd;
        pfd.fd = fd;
        pfd.events = POLLOUT;
        if (poll(&pfd, 1, 0) <= 0)
            break;
        c += write(fd, buf, 10240);
    }
    return c;
}

TEST(Poll, TimeoutIs0)
{
    go [] {
        int fds[2];
        int res = socketpair(AF_LOCAL, SOCK_STREAM, 0, fds);
        EXPECT_EQ(res, 0);

        {
            pollfd pfds[1] = {{fds[0], POLLIN, 0}};
            int n = poll(pfds, 1, 0);
            EXPECT_EQ(n, 0);
            EXPECT_EQ(pfds[0].revents, 0);
        }

        {
            pollfd pfds[1] = {{fds[0], POLLOUT, 0}};
            int n = poll(pfds, 1, 0);
            EXPECT_EQ(n, 1);
            EXPECT_EQ(pfds[0].revents, POLLOUT);
        }

        {
            pollfd pfds[2] = {{fds[0], POLLIN, 0}, {fds[1], POLLIN, 0}};
            int n = poll(pfds, 2, 0);
            EXPECT_EQ(n, 0);
            EXPECT_EQ(pfds[0].revents, 0);
            EXPECT_EQ(pfds[1].revents, 0);
        }

        {
            pollfd pfds[2] = {{fds[0], POLLOUT, 0}, {fds[1], POLLOUT, 0}};
            int n = poll(pfds, 2, 0);
            EXPECT_EQ(n, 2);
            EXPECT_EQ(pfds[0].revents, POLLOUT);
            EXPECT_EQ(pfds[1].revents, POLLOUT);
        }

        {
            pollfd pfds[2] = {{fds[0], POLLIN, 0}, {fds[1], POLLOUT, 0}};
            int n = poll(pfds, 2, 0);
            EXPECT_EQ(n, 1);
            EXPECT_EQ(pfds[0].revents, 0);
            EXPECT_EQ(pfds[1].revents, POLLOUT);
        }

        {
            pollfd pfds[2] = {{fds[0], POLLOUT, 0}, {fds[1], POLLIN, 0}};
            int n = poll(pfds, 2, 0);
            EXPECT_EQ(n, 1);
            EXPECT_EQ(pfds[0].revents, POLLOUT);
            EXPECT_EQ(pfds[1].revents, 0);
        }

        shutdown(fds[0], SHUT_RDWR);

        {
            pollfd pfds[2] = {{fds[0], POLLOUT, 0}, {fds[1], POLLOUT, 0}};
            int n = poll(pfds, 2, 0);
            EXPECT_EQ(n, 2);
            EXPECT_EQ(pfds[0].revents, POLLOUT|POLLHUP);
            EXPECT_EQ(pfds[1].revents, POLLOUT|POLLHUP);
        }

        EXPECT_EQ(close(fds[0]), 0);

        {
            pollfd pfds[2] = {{fds[0], POLLOUT, 0}, {fds[1], POLLIN|POLLOUT, 0}};
            int n = poll(pfds, 2, 0);
            EXPECT_EQ(n, 2);
            EXPECT_EQ(pfds[0].revents, POLLNVAL);
            EXPECT_EQ(pfds[1].revents, POLLIN|POLLOUT|POLLHUP);
        }

        EXPECT_EQ(g_Scheduler.GetCurrentTaskYieldCount(), 0u);

        fds[0] = -1;

        {
            pollfd pfds[1] = {{fds[0], POLLOUT, 0}};
            int n = poll(pfds, 1, 0);
            EXPECT_EQ(n, 0);
            EXPECT_EQ(pfds[0].revents, 0);
        }

        {
            pollfd pfds[2] = {{fds[0], POLLOUT, 0}, {fds[1], POLLIN|POLLOUT, 0}};
            int n = poll(pfds, 2, 0);
            EXPECT_EQ(n, 1);
            EXPECT_EQ(pfds[0].revents, 0);
            EXPECT_EQ(pfds[1].revents, POLLIN|POLLOUT|POLLHUP);
        }

        EXPECT_EQ(g_Scheduler.GetCurrentTaskYieldCount(), 0u);
        EXPECT_EQ(close(fds[1]), 0);
    };
    g_Scheduler.RunUntilNoTask();
    EXPECT_EQ(Task::GetTaskCount(), 0u);
}

TEST(Poll, TimeoutIsNegative1)
{
//    g_Scheduler.GetOptions().debug = dbg_all;
    go [] {
        int fds[2];
        int res = socketpair(AF_LOCAL, SOCK_STREAM, 0, fds);
        EXPECT_EQ(res, 0);

        // poll无限时等待数据到达
        {
            go [=]{
                sleep(1);
                ssize_t n = write(fds[1], "a", 1);
                (void)n;
            };
            auto start = system_clock::now();
            pollfd pfds[2] = {{fds[0], POLLIN, 0}, {fds[1], POLLIN, 0}};
            int n = poll(pfds, 2, -1);
            EXPECT_EQ(n, 1);
            EXPECT_EQ(pfds[0].revents, POLLIN);
            EXPECT_EQ(pfds[1].revents, 0);
            auto elapse = duration_cast<milliseconds>(system_clock::now() - start).count();
            EXPECT_GT(elapse, 999);
            EXPECT_LT(elapse, 1050);
            char buf;
            EXPECT_EQ(read(fds[0], &buf, 1), 1);
            EXPECT_EQ(buf, 'a');
        }

        {
            pollfd pfds[1] = {{fds[0], POLLOUT, 0}};
            int n = poll(pfds, 1, -1);
            EXPECT_EQ(n, 1);
            EXPECT_EQ(pfds[0].revents, POLLOUT);
        }

        {
            pollfd pfds[2] = {{fds[0], POLLOUT, 0}, {fds[1], POLLOUT, 0}};
            int n = poll(pfds, 2, -1);
            EXPECT_EQ(n, 2);
            EXPECT_EQ(pfds[0].revents, POLLOUT);
            EXPECT_EQ(pfds[1].revents, POLLOUT);
        }

        {
            pollfd pfds[2] = {{fds[0], POLLIN, 0}, {fds[1], POLLOUT, 0}};
            int n = poll(pfds, 2, -1);
            EXPECT_EQ(n, 1);
            EXPECT_EQ(pfds[0].revents, 0);
            EXPECT_EQ(pfds[1].revents, POLLOUT);
        }

        {
            pollfd pfds[2] = {{fds[0], POLLOUT, 0}, {fds[1], POLLIN, 0}};
            int n = poll(pfds, 2, -1);
            EXPECT_EQ(n, 1);
            EXPECT_EQ(pfds[0].revents, POLLOUT);
            EXPECT_EQ(pfds[1].revents, 0);
        }

        shutdown(fds[0], SHUT_RDWR);

        {
            pollfd pfds[2] = {{fds[0], POLLOUT, 0}, {fds[1], POLLOUT, 0}};
            int n = poll(pfds, 2, -1);
            EXPECT_EQ(n, 2);
            EXPECT_EQ(pfds[0].revents, POLLOUT|POLLHUP);
            EXPECT_EQ(pfds[1].revents, POLLOUT|POLLHUP);
        }

        close(fds[0]);

        {
            pollfd pfds[2] = {{fds[0], POLLOUT, 0}, {fds[1], POLLIN|POLLOUT, 0}};
            int n = poll(pfds, 2, -1);
            EXPECT_EQ(n, 2);
            EXPECT_EQ(pfds[0].revents, POLLNVAL);
            EXPECT_EQ(pfds[1].revents, POLLIN|POLLOUT|POLLHUP);
        }

        fds[0] = -1;

        {
            // poll invalid fds with negative timeout, as co_yield.
            pollfd pfds[1] = {{fds[0], POLLOUT, 0}};
            auto yc = g_Scheduler.GetCurrentTaskYieldCount();
            int n = poll(pfds, 1, -1);
            EXPECT_EQ(g_Scheduler.GetCurrentTaskYieldCount(), yc + 1);
            EXPECT_EQ(n, 0);
            EXPECT_EQ(pfds[0].revents, 0);
        }

        {
            pollfd pfds[2] = {{fds[0], POLLOUT, 0}, {fds[1], POLLIN|POLLOUT, 0}};
            int n = poll(pfds, 2, -1);
            EXPECT_EQ(n, 1);
            EXPECT_EQ(pfds[0].revents, 0);
            EXPECT_EQ(pfds[1].revents, POLLIN|POLLOUT|POLLHUP);
        }
    };
    g_Scheduler.RunUntilNoTask();
    EXPECT_EQ(Task::GetTaskCount(), 0u);
}

TEST(Poll, TimeoutIs1)
{
//    g_Scheduler.GetOptions().debug = dbg_all;
    go [] {
        int fds[2];
        int res = socketpair(AF_LOCAL, SOCK_STREAM, 0, fds);
        EXPECT_EQ(res, 0);

        // poll无限时等待数据到达
        {
            go [=]{
                co_sleep(200);
                ssize_t n = write(fds[1], "a", 1);
                (void)n;
            };
            auto start = system_clock::now();
            pollfd pfds[2] = {{fds[0], POLLIN, 0}, {fds[1], POLLIN, 0}};
            int n = poll(pfds, 2, 2000);
            EXPECT_EQ(n, 1);
            EXPECT_EQ(pfds[0].revents, POLLIN);
            EXPECT_EQ(pfds[1].revents, 0);
            auto elapse = duration_cast<milliseconds>(system_clock::now() - start).count();
            EXPECT_GT(elapse, 199);
            EXPECT_LT(elapse, 250);

            char buf;
            EXPECT_EQ(read(fds[0], &buf, 1), 1);
            EXPECT_EQ(buf, 'a');
        }

        {
            go [=]{
                co_sleep(500);
                ssize_t n = write(fds[1], "a", 1);
                (void)n;
            };
            auto start = system_clock::now();
            pollfd pfds[2] = {{fds[0], POLLIN, 0}, {fds[1], POLLIN, 0}};
            int n = poll(pfds, 2, 200);
            EXPECT_EQ(n, 0);
            EXPECT_EQ(pfds[0].revents, 0);
            EXPECT_EQ(pfds[1].revents, 0);
            auto elapse = duration_cast<milliseconds>(system_clock::now() - start).count();
            EXPECT_GT(elapse, 199);
            EXPECT_LT(elapse, 250);

            start = system_clock::now();
            n = poll(pfds, 2, 1000);
            EXPECT_EQ(n, 1);
            EXPECT_EQ(pfds[0].revents, POLLIN);
            EXPECT_EQ(pfds[1].revents, 0);
            elapse = duration_cast<milliseconds>(system_clock::now() - start).count();
            EXPECT_GT(elapse, 250);
            EXPECT_LT(elapse, 350);

            char buf;
            EXPECT_EQ(read(fds[0], &buf, 1), 1);
            EXPECT_EQ(buf, 'a');
        }

        {
            pollfd pfds[1] = {{fds[0], POLLOUT, 0}};
            int n = poll(pfds, 1, 1000);
            EXPECT_EQ(n, 1);
            EXPECT_EQ(pfds[0].revents, POLLOUT);
        }

        {
            pollfd pfds[2] = {{fds[0], POLLOUT, 0}, {fds[1], POLLOUT, 0}};
            int n = poll(pfds, 2, 1000);
            EXPECT_EQ(n, 2);
            EXPECT_EQ(pfds[0].revents, POLLOUT);
            EXPECT_EQ(pfds[1].revents, POLLOUT);
        }

        {
            pollfd pfds[2] = {{fds[0], POLLIN, 0}, {fds[1], POLLOUT, 0}};
            int n = poll(pfds, 2, 1000);
            EXPECT_EQ(n, 1);
            EXPECT_EQ(pfds[0].revents, 0);
            EXPECT_EQ(pfds[1].revents, POLLOUT);
        }

        {
            pollfd pfds[2] = {{fds[0], POLLOUT, 0}, {fds[1], POLLIN, 0}};
            int n = poll(pfds, 2, 1000);
            EXPECT_EQ(n, 1);
            EXPECT_EQ(pfds[0].revents, POLLOUT);
            EXPECT_EQ(pfds[1].revents, 0);
        }

        shutdown(fds[0], SHUT_RDWR);

        {
            pollfd pfds[2] = {{fds[0], POLLOUT, 0}, {fds[1], POLLOUT, 0}};
            int n = poll(pfds, 2, 1000);
            EXPECT_EQ(n, 2);
            EXPECT_EQ(pfds[0].revents, POLLOUT|POLLHUP);
            EXPECT_EQ(pfds[1].revents, POLLOUT|POLLHUP);
        }

        close(fds[0]);

        {
            pollfd pfds[2] = {{fds[0], POLLOUT, 0}, {fds[1], POLLIN|POLLOUT, 0}};
            int n = poll(pfds, 2, 1000);
            EXPECT_EQ(n, 2);
            EXPECT_EQ(pfds[0].revents, POLLNVAL);
            EXPECT_EQ(pfds[1].revents, POLLIN|POLLOUT|POLLHUP);
        }

        fds[0] = -1;

        {
            pollfd pfds[1] = {{fds[0], POLLOUT, 0}};
            auto start = system_clock::now();
            auto yc = g_Scheduler.GetCurrentTaskYieldCount();
            int n = poll(pfds, 1, 300);
            EXPECT_EQ(g_Scheduler.GetCurrentTaskYieldCount(), yc + 1);
            auto elapse = duration_cast<milliseconds>(system_clock::now() - start).count();
            EXPECT_GT(elapse, 299);
            EXPECT_LT(elapse, 350);
            EXPECT_EQ(n, 0);
            EXPECT_EQ(pfds[0].revents, 0);
        }

        {
            pollfd pfds[2] = {{fds[0], POLLOUT, 0}, {fds[1], POLLIN|POLLOUT, 0}};
            int n = poll(pfds, 2, 1000);
            EXPECT_EQ(n, 1);
            EXPECT_EQ(pfds[0].revents, 0);
            EXPECT_EQ(pfds[1].revents, POLLIN|POLLOUT|POLLHUP);
        }
    };
    g_Scheduler.RunUntilNoTask();
    EXPECT_EQ(Task::GetTaskCount(), 0u);
}

TEST(PollTrigger, MultiPollTimeout1)
{
//    g_Scheduler.GetOptions().debug = dbg_ioblock | dbg_fd_ctx;

    int fds[2];
    int res = socketpair(AF_LOCAL, SOCK_STREAM, 0, fds);
    EXPECT_EQ(res, 0);

    res = fill_send_buffer(fds[0]);
    cout << "fill " << res << " bytes." << endl;

    for (int i = 0; i < 10; ++i) {
        go [=] {
            pollfd pfds[1] = {{fds[0], POLLIN, 0}};
            auto start = high_resolution_clock::now();
            int n = poll(pfds, 1, 1000);
            auto end = system_clock::now();
            auto c = duration_cast<milliseconds>(end - start).count();
            EXPECT_LT(c, 1050);
            EXPECT_GT(c, 999);
            EXPECT_EQ(n, 0);
            EXPECT_EQ(pfds[0].revents, 0);
        };
    }

    for (int i = 0; i < 10; ++i) {
        go [=] {
            pollfd pfds[1] = {{fds[0], POLLOUT, 0}};
            auto start = high_resolution_clock::now();
            int n = poll(pfds, 1, 500);
            auto end = system_clock::now();
            auto c = duration_cast<milliseconds>(end - start).count();
            EXPECT_LT(c, 550);
            EXPECT_GT(c, 499);
            EXPECT_EQ(n, 0);
            EXPECT_EQ(pfds[0].revents, 0);
        };
    }
    g_Scheduler.RunUntilNoTask();
    EXPECT_EQ(Task::GetTaskCount(), 0u);

    g_Scheduler.GetOptions().debug = 0;
    close(fds[0]);
    close(fds[1]);
}

TEST(PollTrigger, MultiPollTimeout2)
{
//    g_Scheduler.GetOptions().debug = dbg_ioblock | dbg_fd_ctx;

    int fds[2];
    int res = socketpair(AF_LOCAL, SOCK_STREAM, 0, fds);
    EXPECT_EQ(res, 0);

    for (int i = 0; i < 10; ++i) {
        go [=] {
            pollfd pfds[2] = {{fds[0], POLLIN, 0}, {fds[1], POLLIN, 0}};
            auto start = high_resolution_clock::now();
            int n = poll(pfds, 2, 1000);
            auto end = system_clock::now();
            auto c = duration_cast<milliseconds>(end - start).count();
            EXPECT_LT(c, 1050);
            EXPECT_GT(c, 999);
            EXPECT_EQ(n, 0);
            EXPECT_EQ(pfds[0].revents, 0);
            EXPECT_EQ(pfds[1].revents, 0);
        };
    }

    g_Scheduler.RunUntilNoTask();
    EXPECT_EQ(Task::GetTaskCount(), 0u);

    res = fill_send_buffer(fds[0]);
    cout << "fill " << res << " bytes." << endl;
    res = fill_send_buffer(fds[1]);
    cout << "fill " << res << " bytes." << endl;

    for (int i = 0; i < 10; ++i) {
        go [=] {
            pollfd pfds[2] = {{fds[0], POLLOUT, 0}, {fds[1], POLLOUT, 0}};
            auto start = high_resolution_clock::now();
            int n = poll(pfds, 2, 500);
            auto end = system_clock::now();
            auto c = duration_cast<milliseconds>(end - start).count();
            EXPECT_LT(c, 550);
            EXPECT_GT(c, 499);
            EXPECT_EQ(n, 0);
            EXPECT_EQ(pfds[0].revents, 0);
            EXPECT_EQ(pfds[1].revents, 0);
        };
    }
    g_Scheduler.RunUntilNoTask();
    EXPECT_EQ(Task::GetTaskCount(), 0u);

    g_Scheduler.GetOptions().debug = 0;
    close(fds[0]);
    close(fds[1]);
}

TEST(PollTrigger, MultiPollTrigger)
{
//    g_Scheduler.GetOptions().debug = dbg_ioblock | dbg_fd_ctx;

    int fds[2];
    int res = socketpair(AF_LOCAL, SOCK_STREAM, 0, fds);
    EXPECT_EQ(res, 0);

    for (int i = 0; i < 1; ++i) {
        go [=] {
            pollfd pfds[1] = {{fds[0], POLLIN, 0}};
            auto start = high_resolution_clock::now();
            int n = poll(pfds, 1, 1000);
            auto end = system_clock::now();
            auto c = duration_cast<milliseconds>(end - start).count();
            EXPECT_LT(c, 550);
            EXPECT_GT(c, 499);
            EXPECT_EQ(n, 1);
            EXPECT_EQ(pfds[0].revents, POLLIN);
        };
    }

    for (int i = 0; i < 1; ++i) {
        go [=] {
            pollfd pfds[2] = {{fds[0], POLLOUT, 0}, {fds[1], POLLOUT, 0}};
            int n = poll(pfds, 2, 1000);
            EXPECT_EQ(n, 2);
            EXPECT_EQ(pfds[0].revents, POLLOUT);
            EXPECT_EQ(pfds[1].revents, POLLOUT);
        };
    }

    go [=] {
        co_sleep(500);
        ssize_t n = write(fds[1], "a", 1);
        (void)n;
    };
    g_Scheduler.RunUntilNoTask();
    EXPECT_EQ(Task::GetTaskCount(), 0u);

    g_Scheduler.GetOptions().debug = 0;
    close(fds[0]);
    close(fds[1]);
}

TEST(PollTrigger, MultiPollClose)
{
//    g_Scheduler.GetOptions().debug = dbg_ioblock | dbg_fd_ctx;

    int fds[2];
    int res = socketpair(AF_LOCAL, SOCK_STREAM, 0, fds);
    EXPECT_EQ(res, 0);

    for (int i = 0; i < 1; ++i) {
        go [=] {
            pollfd pfds[1] = {{fds[0], POLLIN, 0}};
            auto start = high_resolution_clock::now();
            int n = poll(pfds, 1, 1000);
            auto end = system_clock::now();
            auto c = duration_cast<milliseconds>(end - start).count();
            EXPECT_LT(c, 550);
            EXPECT_GT(c, 499);
            EXPECT_EQ(n, 1);
            EXPECT_EQ(pfds[0].revents, POLLNVAL);
            cout << "read wait exit" << endl;
        };
    }

    fill_send_buffer(fds[0]);

    for (int i = 0; i < 1; ++i) {
        go [=] {
            pollfd pfds[1] = {{fds[0], POLLOUT, 0}};
            int n = poll(pfds, 1, 1000);
            EXPECT_EQ(n, 1);
            EXPECT_EQ(pfds[0].revents, POLLNVAL);
            cout << "write wait exit" << endl;
        };
    }

    go [=] {
        co_sleep(500);
        cout << "close " << fds[0] << endl;
        close(fds[0]);
    };
    g_Scheduler.RunUntilNoTask();
    EXPECT_EQ(Task::GetTaskCount(), 0u);

//    g_Scheduler.GetOptions().debug = 0;
    close(fds[0]);
    close(fds[1]);
}

TEST(Poll, MultiThreads)
{
    for (int i = 0; i < 20; ++i)
        go [] {
            int fds[2];
            int res = socketpair(AF_LOCAL, SOCK_STREAM, 0, fds);
            EXPECT_EQ(res, 0);
            uint64_t yc = g_Scheduler.GetCurrentTaskYieldCount();
            pollfd pfds[1] = {{fds[0], POLLIN, 0}};
            auto start = high_resolution_clock::now();
            int n = poll(pfds, 1, 1000);
            auto end = system_clock::now();
            auto c = duration_cast<milliseconds>(end - start).count();
            EXPECT_LT(c, 1200);
            EXPECT_GT(c, 999);
            EXPECT_EQ(n, 0);
            EXPECT_EQ(g_Scheduler.GetCurrentTaskYieldCount(), yc + 1);
            EXPECT_EQ(close(fds[0]), 0);
            EXPECT_EQ(close(fds[1]), 0);
            close(fds[0]);
            close(fds[1]);
        };
    printf("coroutines create done.\n");
    boost::thread_group tg;
    for (int i = 0; i < 8; ++i)
        tg.create_thread([] {
                g_Scheduler.RunUntilNoTask();
                });
    tg.join_all();
    printf("coroutines run done.\n");
    if (Task::GetTaskCount()) // 可能会有一些Task还未删除，执行删除逻辑。
        g_Scheduler.RunUntilNoTask();
    EXPECT_EQ(Task::GetTaskCount(), 0u);
}

max_stars_repo_name: fengliangwang/libgo

max_issues_repo_name: fengliangwang/libgo

max_forks_repo_name: fengliangwang/libgo


text:
#include <iostream>
#include <boost/algorithm/string.hpp>

#include "glog/logging.h"
#include "folly/String.h"

#include "kafka_producer.h"

namespace predictor {

Status KafkaProducer::initialize(const std::string& configfile) {
  std::ifstream fin(configfile);
  rapidjson::IStreamWrapper isw(fin);
  rapidjson::Document config;
  config.ParseStream(isw);
  if (!config.IsObject()) {
    LOG(ERROR) << "parse kafka config file error:" << configfile;
    return Status::KAFKA_PRODUCER_CONFIG_FILE_ERROR;
  }
  if (!config.HasMember("kafka_producer_config")) {
    LOG(ERROR) << "kafka producer config member:kafka_producer_config is not found:" << configfile;
    return Status::KAFKA_PRODUCER_CONFIG_ERROR;
  }
  const rapidjson::Value& kafka_producer_config = config["kafka_producer_config"];
  if (!kafka_producer_config.IsObject()) {
    LOG(ERROR) << "parse kafka producer config member:kafka_config error:" << configfile;
    return Status::KAFKA_PRODUCER_CONFIG_ERROR;
  }
  if (!kafka_producer_config.HasMember("topic")) {
    LOG(ERROR) << "kafka producer config member:topic is not found:" << configfile;
    return Status::KAFKA_PRODUCER_CONFIG_TOPIC_ERROR;
  }
  topic_ = kafka_producer_config["topic"].GetString();
  if (kafka_producer_config.HasMember("partition_num")) {
    partition_num_ = kafka_producer_config["partition_num"].GetInt();
  }
  LOG(INFO) << "topic:" << topic_ << " partition:" << partition_num_;

  if (!kafka_producer_config.HasMember("kafka_config")) {
    LOG(ERROR) << "kafka producer config member:kafka_config is not found:" << configfile;
    return Status::KAFKA_PRODUCER_CONFIG_KAFKA_ERROR;
  }
  const rapidjson::Value& kafka_config = kafka_producer_config["kafka_config"];
  if (!kafka_config.IsArray()) {
    LOG(ERROR) << "get kafka producer kafka_config error:"  << configfile;
    return Status::KAFKA_PRODUCER_CONFIG_KAFKA_ERROR;
  }
  std::unordered_map<std::string, std::string> config_map;
  for (rapidjson::SizeType i = 0; i < kafka_config.Size() - 1; i = i + 2) {
    config_map[kafka_config[i].GetString()] = kafka_config[i + 1].GetString();
  }
  int ret = init(topic_, config_map);
  if (ret) {
    return Status::KAFKA_PRODUCER_KAFKA_INIT_ERROR;
  }
  return Status::OK;
}

bool KafkaProducer::sendKafka(const std::string& message, int64_t hashid) {
  std::unique_ptr<folly::IOBuf> iobuf = folly::IOBuf::wrapBuffer(message.c_str(), message.length());
  int partition = -1;
  if (hashid != -1 && partition_num_ != -1) {
    partition = hashid % partition_num_;
  }
  int ret = common::kafka::Producer::send(iobuf, partition);
  if (ret) {
    LOG(ERROR) << "send kafka error, message:" << message << "; size:" << message.length();
    return false;
  }
  return true;
}
}  // namespace predictor

ext: cc

max_stars_repo_name: algo-data-platform/PredictorService

max_issues_repo_name: algo-data-platform/PredictorService

max_forks_repo_name: algo-data-platform/PredictorService