text:
// STD headers
#include <assert.h>
#include <fstream>
#include <functional>
#include <iomanip>
#include <iostream>
#include <list>
#include <string>
#include <thread>
#include <tuple>
#include <unordered_map>
#include <unordered_set>
#include <vector>

// Boost headers
#include <boost/bimap.hpp>
#include <boost/program_options.hpp>

// Cereal headers
#include <cereal/archives/binary.hpp>
#include <cereal/types/vector.hpp>

// Gurobi headers
#include <gurobi_c++.h>

// Custom headers
#include "flow_dict.hpp"
#include "simulator.hpp"
#include "utils.hpp"

// Typedefs
namespace bopt = boost::program_options;
typedef std::tuple<std::string, bool, std::string, std::string,
double> Edge;  // Format: (flow_id, in_cache, src, dst, cost)

// Constant hyperparameters
constexpr unsigned int kNumThreads = 24;

/**
 * Implements a Multi-Commodity Min-Cost Flow-based
 * solver for the Delayed Hits caching problem.
 */
class MCMCFSolver {
private:
    typedef std::vector<size_t> Indices;

    // Trace and cache parameters
    const uint z_;
    const uint c_;
    const bool is_forced_;
    const size_t trace_len_;
    const std::string trace_file_path_;
    const std::string model_file_prefix_;
    const std::string packets_file_path_;
    const std::vector<std::string> trace_;

    // Housekeeping
    std::vector<double> solution_;
    std::unordered_set<size_t> cache_node_idxs_;
    std::unordered_map<std::string, Indices> indices_;
    boost::bimap<size_t, std::string> idx_to_nodename_;
    std::unordered_map<size_t, double> idx_to_miss_cost_;
    std::unordered_map<std::string, std::string> sink_nodes_;
    std::unordered_map<std::string, Indices> cache_interval_start_idxs_;

    // "Split nodes" are used to enforce forced admission, if required.
    // We detect nodes where the forced admission constraint *might* be
    // violated (i.e. an object doesn't remain in the cache for atleast
    // one timestep), and add manually a constraint that enforces this.
    std::unordered_set<size_t> split_node_idxs_;

    // Internal helper methods
    double getMissCostStartingAt(const size_t idx) const;
    const std::string& cchNodeForMemAt(const size_t idx) const;
    size_t getNextMemNodeIdx(const size_t idx, const std::string& flow_id,
                             std::unordered_map<std::string,
                             size_t>& next_idxs_map) const;

    // Thread-safe, concurrent helper methods
    void populateOutOfCchEdgesConcurrent(
        const std::list<std::string>& flow_ids,
        std::list<Edge>& edges, std::unordered_set<size_t>& cache_node_idxs);

    void populateCchCapacityConstraintsConcurrent(
        const std::pair<size_t, size_t> range,
        const belatedly::FlowDict& flows, std::list<GRBLinExpr>& exprs) const;

    // Setup and checkpointing
    void setup();
    bool loadOptimizedModelSolution(belatedly::FlowDict& flows);
    void saveOptimizedModelSolution(const belatedly::FlowDict& flows) const;

    // Model creation
    void populateOutOfCchEdges(std::list<Edge>& all_edges);
    void createFlowVariables(GRBModel& model, belatedly::FlowDict&
                             flows, const std::list<Edge>& edges) const;

    void populateCchCapacityConstraints(
        GRBModel& model, const belatedly::FlowDict& flows) const;

    void populateSplitNodeFlowConstraints(
        GRBModel& model, const belatedly::FlowDict& flows) const;

    void populateFlowConservationConstraints(
        GRBModel& model, const belatedly::FlowDict& flows) const;

    void addMissCostStartingAt(const size_t start_idx,
                               const double coefficient,
                               std::vector<utils::Packet>& packets) const;

    // Cost computation and validation
    double getCostLowerBound(const belatedly::FlowDict& flows,
                             const std::list<Edge>& edges) const;

    double getCostUpperBoundAndValidateSolution(
        const belatedly::FlowDict& flows, const std::list<Edge>& edges) const;

public:
    MCMCFSolver(const std::string& trace_fp, const std::string& model_fp,
                const std::string& packets_fp, const bool forced, const
                uint z, const uint c, const std::vector<std::string>& trace) :
                z_(z), c_(c), is_forced_(forced), trace_len_(trace.size()),
                trace_file_path_(trace_fp), model_file_prefix_(model_fp),
                packets_file_path_(packets_fp), trace_(trace) {}

    void solve();
};

/**
 * Given an index, returns the corresponding cache node.
 *
 * Thread-safe.
 */
const std::string& MCMCFSolver::cchNodeForMemAt(const size_t idx) const {
    return idx_to_nodename_.left.at(idx);
}

/**
 * Returns the cost of a cache miss starting at idx.
 *
 * Thread-safe.
 */
double MCMCFSolver::getMissCostStartingAt(const size_t idx) const {
    const std::string& flow_id = trace_[idx];
    double miss_cost = z_;

    // Iterate over the next z timesteps, and add the
    // latency cost corresponding to each subsequent
    // packet of the same flow.
    for (size_t offset = 1; offset < z_; offset++) {
        if (idx + offset >= trace_len_) { break; }
        else if (trace_[idx + offset] == flow_id) {
            miss_cost += (z_ - offset);
        }
    }
    return miss_cost;
}

/**
 * Returns the index of the first mem node corresponding to the given
 * flow ID occuring after idx. If no reference to this flow is found
 * in (idx, trace_len_), returns maxval.
 *
 * The state for each flow is cached in next_idxs_map. This dict must
 * be cleared before starting a new operation (that is, when indices
 * are no longer expected to continue increasing monotonically).
 *
 * Thread-safe.
 */
size_t MCMCFSolver::getNextMemNodeIdx(
    const size_t idx, const std::string& flow_id,
    std::unordered_map<std::string, size_t>& next_idxs_map) const {
    const std::vector<size_t>& indices = indices_.at(flow_id);
    size_t indices_len = indices.size();

    while ((next_idxs_map[flow_id] < indices_len) &&
           (indices[next_idxs_map[flow_id]] <= idx)) {
        next_idxs_map[flow_id]++;
    }
    // If the next index is beyond the idxs list for
    // this flow, return the corresponding sink node.
    size_t next_idx = next_idxs_map[flow_id];
    if (next_idx == indices_len) {
        return std::numeric_limits<size_t>::max();
    }
    else {
        size_t next_mem_idx = indices[next_idx];
        assert(trace_[next_mem_idx] == flow_id);
        return next_mem_idx;
    }
}

/**
 * Attempts to load the optimal solution and the flow mappings
 * for the given model. Returns true if successful, else false.
 */
bool MCMCFSolver::loadOptimizedModelSolution(belatedly::FlowDict& flows) {
    if (model_file_prefix_.empty()) { return false; }
    std::ifstream solution_ifs(model_file_prefix_ + ".sol");
    const std::string flows_fp = model_file_prefix_ + ".flows";

    // Either of the required model files is missing, indicating failure
    if (!solution_ifs.good() || !std::ifstream(flows_fp).good()) {
        return false;
    }
    // Load the solutions vector
    {
        cereal::BinaryInputArchive ar(solution_ifs);
        ar(solution_);
    }
    // Load the flow mappings
    flows.loadFlowMappings(flows_fp);

    // Sanity check: The number of decision variables should
    // be equal to the number of entries in the flows map.
    assert(solution_.size() == flows.numVariables());
    return true;
}

/**
 * Saves the optimal solution and flow mappings to disk.
 */
void MCMCFSolver::saveOptimizedModelSolution(
    const belatedly::FlowDict& flows) const {

    if (model_file_prefix_.empty()) { return; }
    std::ofstream solution_ofs(model_file_prefix_ + ".sol");
    const std::string flows_fp = model_file_prefix_ + ".flows";

    // Save the solutions vector
    cereal::BinaryOutputArchive oarchive(solution_ofs);
    oarchive(solution_);

    // Save the flow mappings
    flows.saveFlowMappings(flows_fp);
}

/**
 * Populate the indices map and internal data structures.
 */
void MCMCFSolver::setup() {
    size_t num_nonempty_packets = 0;
    for (size_t idx = 0; idx < trace_len_; idx++) {
        const std::string& flow_id = trace_[idx];

        // If the packet is non-empty, append the idx to
        // the corresponding flow's idxs. Also, populate
        // the miss-costs map.
        if (!flow_id.empty()) {
            num_nonempty_packets++;
            indices_[flow_id].push_back(idx);
            idx_to_miss_cost_[idx] = getMissCostStartingAt(idx);
        }
        // Found a possible violation of forced admission; track this as a split node
        if (is_forced_ && !flow_id.empty() && (idx > 0) && (idx < trace_len_ - z_) &&
            (flow_id == trace_[idx - 1]) && (flow_id == trace_[idx + z_])) {
            split_node_idxs_.insert(idx);
        }
        // Populate the nodenames map
        idx_to_nodename_.insert(boost::bimap<size_t, std::string>::value_type(
            idx, "cch_t" + std::to_string(idx + z_ - 1) + "_" + flow_id));
    }
    // Populate the sink nodes map and
    // prime the cache intervals map.
    for (const auto& iter : indices_) {
        const std::string& flow_id = iter.first;
        sink_nodes_[flow_id] = ("x_" + flow_id);
        cache_interval_start_idxs_[flow_id];
    }
    // Debug
    std::cout << "Finished setup with " << trace_len_
              << " packets (" << num_nonempty_packets
              << " nonempty), " << indices_.size()
              << " flows." << std::endl;
}

/**
 * Helper method. Populates the out-of-cache edges (mem->cch/admittance,
 * and cch->mem/eviction) for the given flow IDs in a thread-safe manner.
 *
 * Important note: This method relies on the fact that separate threads
 * operate on disjoint sets of flow IDs (and only ever access different
 * STL containers concurrently), precluding the need for mutexes.
 *
 * Thread-safe.
 */
void MCMCFSolver::populateOutOfCchEdgesConcurrent(
    const std::list<std::string>& flow_ids, std::list<Edge>& edges,
    std::unordered_set<size_t>& local_cache_node_idxs) {

    // Iterate over the given flow IDs to populate the edges list
    std::unordered_map<std::string, size_t> next_idxs_map;
    for (const std::string& flow_id : flow_ids) {
        std::string evicted_reference = std::string();
        bool has_flow_started = false;

        // Iterate over each timestep and fetch the next mem reference
        for (size_t idx = 0; idx < trace_len_; idx++) {
            const std::string& trace_flow_id = trace_[idx];
            bool is_same_flow = (!trace_flow_id.empty() &&
                                 (trace_flow_id == flow_id));
            if (has_flow_started) {

                // Fetch the next mem reference corresponding to this flow
                const size_t next_reference_idx = getNextMemNodeIdx(
                    idx + z_ - 1, flow_id, next_idxs_map);

                const bool is_dst_sink_node = (next_reference_idx ==
                                               std::numeric_limits<size_t>::max());

                // Fetch the dst node for this out-of-cache edge
                const std::string& next_reference = is_dst_sink_node ?
                                                    sink_nodes_.at(flow_id) :
                                                    cchNodeForMemAt(next_reference_idx);
                if (is_same_flow) {
                    // In optional admission, unconditionally create an
                    // outedge from this node to the next mem reference.
                    if (!is_forced_) {
                        evicted_reference = next_reference;
                        edges.push_back(std::make_tuple(
                            flow_id, false, cchNodeForMemAt(idx), next_reference,
                            (is_dst_sink_node ? 0.0 : idx_to_miss_cost_.at(next_reference_idx)))
                        );
                    }
                    // The following node in the trace (mem_t{idx + z}) maps to
                    // this flow, but the previous node in the caching sequence
                    // (cch_t{idx + z - 1}) also maps to it. Thus, this is the
                    // final opportunity to reach mem_t{idx + z}, and we create
                    // an edge to it.
                    else if (split_node_idxs_.find(idx) != split_node_idxs_.end()) {
                        assert(!is_dst_sink_node);
                        edges.push_back(std::make_tuple(
                            flow_id, false, cchNodeForMemAt(idx),
                            next_reference, idx_to_miss_cost_.at(next_reference_idx))
                        );
                    }
                    // Discard the next mem reference for this flow
                    else {
                        evicted_reference.clear();
                    }

                    // Mark this idx as the start of a new interval for this flow
                    local_cache_node_idxs.insert(idx);
                    cache_interval_start_idxs_.at(flow_id).push_back(idx);
                }
                // If we did not already created an eviction edge to the
                // next mem reference for this flow, create one to it.
                else if (evicted_reference != next_reference) {
                    evicted_reference = next_reference;
                    edges.push_back(std::make_tuple(
                        flow_id, false, cchNodeForMemAt(idx), next_reference,
                        (is_dst_sink_node ? 0.0 : idx_to_miss_cost_.at(next_reference_idx)))
                    );

                    // Mark this idx as the start of a new interval for this flow
                    local_cache_node_idxs.insert(idx);
                    cache_interval_start_idxs_.at(flow_id).push_back(idx);
                }
            }
            // The flow has not yet started and the packet at this timestep
            // belongs to this flow, indicating that this is the first ever
            // request to it.
            else if (is_same_flow) {
                has_flow_started = true;

                // In optional admission, unconditionally create an
                // outedge from this node to the next mem reference.
                if (!is_forced_) {
                    // Fetch the next mem reference corresponding to this flow
                    const size_t next_reference_idx = getNextMemNodeIdx(
                        idx + z_ - 1, flow_id, next_idxs_map);

                    const bool is_dst_sink_node = (next_reference_idx ==
                                                   std::numeric_limits<size_t>::max());

                    // Fetch the dst node for this out-of-cache edge
                    const std::string& next_reference = is_dst_sink_node ?
                                                        sink_nodes_.at(flow_id) :
                                                        cchNodeForMemAt(next_reference_idx);
                    evicted_reference = next_reference;
                    edges.push_back(std::make_tuple(
                        flow_id, false, cchNodeForMemAt(idx), next_reference,
                        (is_dst_sink_node ? 0.0 : idx_to_miss_cost_.at(next_reference_idx)))
                    );
                }
                // Mark this idx as the start of a new interval for this flow
                local_cache_node_idxs.insert(idx);
                cache_interval_start_idxs_.at(flow_id).push_back(idx);
            }
        }
    }
}

/**
 * Populates the out-of-cache (eviction and admittance) edges.
 */
void MCMCFSolver::populateOutOfCchEdges(std::list<Edge>& all_edges) {
    size_t num_out_of_cch_edges = 0;
    size_t num_total_flows = indices_.size();
    size_t num_flows_per_thread = int(ceil(num_total_flows /
                                      static_cast<double>(kNumThreads)));

    // Temporary containers for each thread
    size_t num_flows_allotted = 0;
    std::array<std::list<Edge>, kNumThreads> edges;
    std::array<std::list<std::string>, kNumThreads> flow_ids;
    std::array<std::unordered_set<size_t>, kNumThreads> local_cache_node_idxs;

    // Partition the entire set of flow IDs into kNumThreads disjoint sets
    auto iter = indices_.begin();
    for (size_t t_idx = 0; t_idx < kNumThreads; t_idx++) {
        for (size_t i = 0; (i < num_flows_per_thread) &&
             (iter != indices_.end()); i++, iter++) {
            num_flows_allotted++;
            flow_ids[t_idx].push_back(iter->first);
        }
    }
    // Sanity check
    assert(num_flows_allotted == num_total_flows);

    // Launch kNumThreads on the corresponding sets of flow IDs
    std::thread workers[kNumThreads];
    for (size_t t_idx = 0; t_idx < kNumThreads; t_idx++) {
        workers[t_idx] = std::thread(&MCMCFSolver::populateOutOfCchEdgesConcurrent, this,
                                     std::ref(flow_ids[t_idx]), std::ref(edges[t_idx]),
                                     std::ref(local_cache_node_idxs[t_idx]));
    }
    // Wait for all the worker threads to finish execution
    for (size_t t_idx = 0; t_idx < kNumThreads; t_idx++) {
        workers[t_idx].join();
    }
    // Then, combine their results
    for (size_t t_idx = 0; t_idx < kNumThreads; t_idx++) {
        num_out_of_cch_edges += edges[t_idx].size();
        all_edges.splice(all_edges.end(), edges[t_idx]);
        cache_node_idxs_.insert(local_cache_node_idxs[t_idx].begin(),
                                local_cache_node_idxs[t_idx].end());
    }
    // Next, sort the cache intervals for each flow in-place
    for (auto& iter : cache_interval_start_idxs_) {
        std::vector<size_t>& start_idxs = iter.second;
        std::sort(start_idxs.begin(), start_idxs.end());
    }
    // Finally, for the last packet in the trace, unconditionally
    // create an outedge to the corresponding flow's sink node.
    size_t idx = (trace_len_ - 1);
    const std::string& flow_id = trace_[idx];
    if (is_forced_ && !flow_id.empty()) {
        num_out_of_cch_edges++;
        all_edges.push_back(std::make_tuple(flow_id, false,
                                            cchNodeForMemAt(idx),
                                            sink_nodes_.at(flow_id), 0.0));
    }
    // Debug
    std::cout << "Finished populating " << num_out_of_cch_edges
              << " out-of-cache edges concurrently, using "
              << kNumThreads << " threads." << std::endl;
}

/**
 * Given a list of edges, populates the flow tupledict.
 */
void MCMCFSolver::
createFlowVariables(GRBModel& model, belatedly::FlowDict& flows,
                    const std::list<Edge>& edges) const {
    size_t num_total_cache_intervals = 0;

    // First, create variables representing mem->cch and cch->mem edges
    for (const Edge& edge : edges) {
        flows.addVariable(std::get<0>(edge),
                          std::get<1>(edge),
                          std::get<2>(edge),
                          std::get<3>(edge),
                          std::get<4>(edge));
    }
    // Next, for every flow, create decision variables
    // corresponding to each possible caching interval.
    for (const auto& iter : cache_interval_start_idxs_) {
        const std::string& flow_id = iter.first;
        const std::vector<size_t>& start_idxs = iter.second;
        const size_t num_intervals = (start_idxs.size() - 1);

        // Iterate over each interval
        for (size_t i = 0; i < num_intervals; i++) {
            size_t idx = start_idxs[i];
            size_t next_idx = start_idxs[i + 1];

            // Sanity check: Both indices should be in the nodes set
            assert((cache_node_idxs_.find(idx) != cache_node_idxs_.end()) &&
                   (cache_node_idxs_.find(next_idx) != cache_node_idxs_.end()));

            // Create a decision variable corresponding to cch_{idx}->cch_{next_idx}
            num_total_cache_intervals++;
            flows.addVariable(flow_id, true,
                              cchNodeForMemAt(idx),
                              cchNodeForMemAt(next_idx), 0.0);
        }
    }
    // Finally, update the model
    model.update();

    // Debug
    std::cout << "Finished creating " << flows.numVariables() << " flow variables (for "
              << edges.size() << " edges, plus " << num_total_cache_intervals
              << " caching intervals)." << std::endl;
}

/**
 * Helper method. Populates the cache capacity constraints for
 * the given range of trace indices in a thread-safe manner.
 *
 * Thread-safe.
 */
void MCMCFSolver::populateCchCapacityConstraintsConcurrent(
    const std::pair<size_t, size_t> range, const belatedly::
    FlowDict& flows, std::list<GRBLinExpr>& exprs) const {
    std::unordered_map<std::string, GRBVar> decision_variables;
    std::unordered_map<std::string, size_t> current_idxs;

    // Starting index is out-of-bounds, do nothing
    if (range.first >= trace_len_) { return; }

    // First, for each flow ID, forward its current iterator idx
    // to point to the appropriate location in its indices list.
    for (const auto& iter : cache_interval_start_idxs_) {
        const std::vector<size_t>& intervals = iter.second;
        const std::string& flow_id = iter.first;

        size_t i = 0;
        while (i < (intervals.size() - 1) &&
               range.first > intervals[i + 1]) { i++; }

        // Update the current iterator idx
        current_idxs[flow_id] = i;
    }
    // Next, generate a linear-expression corresponding to the sum of
    // all flow decision variables at every timestep, and impose that
    // it is LEQ the cache size as a constraint.
    for (size_t idx = range.first; idx < range.second; idx++) {
        if (cache_node_idxs_.find(idx) ==
            cache_node_idxs_.end()) { continue; }

        bool is_split_node = (split_node_idxs_.find(idx) !=
                              split_node_idxs_.end());
        GRBLinExpr node_capacity_expr;
        bool is_expr_changed = false;

        // Update the current decision variable for each
        // flow and add it to the LHS of the constraints.
        for (const auto& iter : cache_interval_start_idxs_) {
            const std::string& flow_id = iter.first;
            const std::vector<size_t>& intervals = iter.second;
            bool is_split_node_for_flow = (is_split_node &&
                                           (trace_[idx] == flow_id));

            // If this trace index either precedes the first decision
            // variable, or is either at or beyond the last decision
            // variable for this flow, do nothing.
            size_t i = current_idxs[flow_id];
            if (idx < intervals[0] || idx >= intervals.back()) {
                continue;
            }
            // Reached the first interval or the end
            // of the current interval for this flow.
            else if ((decision_variables.find(flow_id) ==
                      decision_variables.end()) ||
                     (idx == intervals[i + 1])) {

                // If this is the end of the current
                // interval, update the idx iterator.
                if (idx == intervals[i + 1]) {
                    current_idxs[flow_id] = ++i;
                }
                // Fetch the current interval
                size_t start_idx = intervals[i];
                size_t end_idx = intervals[i + 1];

                // This is a regular node
                GRBVar decision_variable = (
                    flows.getVariable(flow_id, true,
                                      cchNodeForMemAt(start_idx),
                                      cchNodeForMemAt(end_idx)));

                // Update the decision variable for this flow
                decision_variables[flow_id] = decision_variable;
                node_capacity_expr += decision_variable;
                is_expr_changed = true;
            }
            // Else, add the current decision variable to the expression
            else {
                GRBVar decision_variable = decision_variables.at(flow_id);
                node_capacity_expr += decision_variable;
                assert(!is_split_node_for_flow);
            }
        }
        // Finally, add the expression as a cache constraints
        if (is_expr_changed) { exprs.push_back(node_capacity_expr); }
    }
}

/**
 * Populates the cache capacity constraints for the given flow variables.
 */
void MCMCFSolver::populateCchCapacityConstraints(
    GRBModel& model, const belatedly::FlowDict& flows) const {
    std::array<std::list<GRBLinExpr>, kNumThreads> exprs;
    std::thread workers[kNumThreads];
    size_t num_constraints = 0;

    // Partition the trace into kNumThreads disjoint sets
    size_t start_idx = 0;
    size_t idxs_per_thread = int(ceil(trace_len_ /
                                 static_cast<double>(kNumThreads)));

    // Launch kNumThreads on the corresponding range of indices
    for (size_t t_idx = 0; t_idx < kNumThreads; t_idx++) {
        size_t end_idx = std::min(trace_len_,
                                  start_idx + idxs_per_thread);

        workers[t_idx] = std::thread(
            &MCMCFSolver::populateCchCapacityConstraintsConcurrent, this,
            std::make_pair(start_idx, end_idx), std::ref(flows),
            std::ref(exprs[t_idx]));

        // Update the starting idx
        start_idx = end_idx;
    }
    // Wait for all the worker threads to finish execution
    for (size_t t_idx = 0; t_idx < kNumThreads; t_idx++) {
        workers[t_idx].join();
    }
    // Populate the model constraints
    for (size_t t_idx = 0; t_idx < kNumThreads; t_idx++) {
        for (auto& expr : exprs[t_idx]) {
            num_constraints++;
            model.addConstr(expr, GRB_LESS_EQUAL, c_, "");
        }
        exprs[t_idx].clear();
    }
    // Debug
    std::cout << "Finished adding " << num_constraints << " arc-capacity"
              << " constraints for cch->cch edges concurrently, using "
              << kNumThreads << " threads." << std::endl;
}

/**
 * Populates flow constraints for split nodes. These ensure that once
 * flow enters the cache, it remains there for at least one timestep.
 */
void MCMCFSolver::populateSplitNodeFlowConstraints(
    GRBModel& model, const belatedly::FlowDict& flows) const {
    size_t num_split_node_constraints = 0;

    for (const size_t idx : split_node_idxs_) {
        const std::string& flow_id = trace_[idx];
        num_split_node_constraints++;

        // Fetch the decision variable corresponding
        // to the cch_{idx}->next_mem_node edge.
        const GRBVar outflow = flows.getVariable(
            flow_id, false, cchNodeForMemAt(idx),
            cchNodeForMemAt(idx + z_));

        // Fetch the expression corresponding
        // to the mem_{idx}->cch_{idx} edge.
        const GRBLinExpr inflow = (
            flows.sumAcrossSrcNodes(flow_id, false,
                                    cchNodeForMemAt(idx)));

        // Ensure that the outflow is LEQ (1 - inflow)
        model.addConstr(outflow + inflow, GRB_LESS_EQUAL, 1, "");
    }
    // Debug
    std::cout << "Finished populating " << num_split_node_constraints
              << " flow constraints for split nodes." << std::endl;
}

/**
 * Populates the flow conservation constraints for all nodes.
 */
void MCMCFSolver::populateFlowConservationConstraints(
    GRBModel& model, const belatedly::FlowDict& flows) const {

    // Flow-conservation constraints for sink nodes
    size_t num_sink_node_constraints = 0;
    for (const auto& iter : indices_) {
        num_sink_node_constraints++;
        const std::string& flow_id = iter.first;
        const std::string& node = sink_nodes_.at(flow_id);
        model.addConstr(flows.sumAcrossSrcNodes(flow_id, node), GRB_EQUAL, 1, "");
    }
    // Debug
    std::cout << "Finished adding " << num_sink_node_constraints
              << " flow-conservation constraints for sink nodes."
              << std::endl;

    // Flow-conservation constraints for cache nodes
    size_t num_cache_node_constraints = 0;
    for (const auto& iter : cache_interval_start_idxs_) {
        const std::string& flow_id = iter.first;
        const std::vector<size_t>& start_idxs = iter.second;

        for (size_t idx : start_idxs) {
            num_cache_node_constraints++;
            const std::string& node = cchNodeForMemAt(idx);
            assert(cache_node_idxs_.find(idx) != cache_node_idxs_.end());

            // This is the source node for this flow
            if (idx == start_idxs.front()) {
                model.addConstr(flows.sumAcrossDstNodes(flow_id, node), GRB_EQUAL, 1, "");
            }
            // Else, this is a regular node
            else {
                model.addConstr(flows.sumAcrossSrcNodes(flow_id, node), GRB_EQUAL,
                                flows.sumAcrossDstNodes(flow_id, node), "");
            }
        }
    }
    // Debug
    std::cout << "Finished adding " << num_cache_node_constraints
              << " flow-conservation constraints for cache nodes."
              << std::endl;
}

/**
 * Helper method. Given a start idx and cost coefficient (flow fraction),
 * updates latencies of subsequent same-flow packets that occur within z
 * timesteps of start_idx.
 */
void MCMCFSolver::
addMissCostStartingAt(const size_t start_idx, const double coefficient,
                      std::vector<utils::Packet>& packets) const {
    if (utils::DoubleApproxEqual(coefficient, 0.)) { return; }

    // Iterate over the next z timesteps, and add the weighted
    // latency cost to each subsequent packet of the same flow.
    const std::string& flow_id = trace_[start_idx];
    for (size_t offset = 0; offset < z_; offset++) {
        const size_t idx = start_idx + offset;

        if (idx >= trace_len_) { break; }
        else if (trace_[idx] == flow_id) {
            packets[idx].addLatency((z_ - offset) * coefficient);
        }
    }
}

/**
 * Returns a (tight) lower-bound on the cost of the LP solution.
 */
double MCMCFSolver::getCostLowerBound(const belatedly::FlowDict& flows,
                                      const std::list<Edge>& edges) const {
    // Create packets corresponding to each timestep
    std::vector<utils::Packet> processed_packets;
    for (const std::string& flow_id : trace_) {
        processed_packets.push_back(utils::Packet(flow_id));
    }
    // Update the packet latency corresponding
    // to the first occurence of each flow.
    for (const auto& elem : indices_) {
        const size_t src_idx = elem.second.front();
        addMissCostStartingAt(src_idx, 1., processed_packets);
    }
    // Next, add the latency cost of flow
    // routed through out-of-cch edges.
    for (const auto& edge : edges) {
        const bool in_cache = std::get<1>(edge);
        const std::string& src = std::get<2>(edge);
        const std::string& dst = std::get<3>(edge);
        const std::string& flow_id = std::get<0>(edge);

        // This is an out-of-cch edge and dst is not a sink node
        if (!in_cache && dst != sink_nodes_.at(flow_id)) {
            double flow_fraction = solution_[
                flows.getVariableIdx(flow_id, in_cache, src, dst)];

            size_t dst_idx = idx_to_nodename_.right.at(dst);
            addMissCostStartingAt(dst_idx, flow_fraction,
                                  processed_packets);
        }
    }
    // Compute the total latency for this solution
    double total_latency = 0;
    for (const utils::Packet& packet : processed_packets) {
        const std::string& flow_id = packet.getFlowId();
        const double latency = packet.getTotalLatency();
        if (!flow_id.empty()) {

            // Sanity check: For non-empty packets, latency should be in [0, z]
            assert(utils::DoubleApproxGreaterThanOrEqual(latency, 0.) &&
                   utils::DoubleApproxGreaterThanOrEqual(z_, latency));

            total_latency += latency;
        }
        // Else, for empty packets, do nothing
        else { assert(latency == 0.); }
    }
    return total_latency;
}

/**
 * Validates the computed LP solution.
 */
double MCMCFSolver::getCostUpperBoundAndValidateSolution(
    const belatedly::FlowDict& flows, const std::list<Edge>& edges) const {
    std::vector<std::tuple<size_t, std::string, double>> evictions;
    std::vector<std::tuple<size_t, std::string, double>> inductions;

    // Parse the eviction schedule from the computed LP solution
    bool is_solution_integral = true;
    size_t num_fractional_vars = 0;
    for (const auto& edge : edges) {
        const bool in_cache = std::get<1>(edge);
        const std::string& flow_id = std::get<0>(edge);

        // If this is an out-of-cch edge, fetch the
        // decision variable corresponding to it.
        if (!in_cache) {
            const size_t src_idx = idx_to_nodename_.right.at(
                    std::get<2>(edge)) + z_ - 1;

            const size_t dst_idx =
                (std::get<3>(edge) == sink_nodes_.at(flow_id)) ?
                std::numeric_limits<size_t>::max() :
                idx_to_nodename_.right.at(std::get<3>(edge)) + z_ - 1;

            double value = solution_[flows.getVariableIdx(
                flow_id, in_cache, std::get<2>(edge),
                std::get<3>(edge))];

            // The solution has non-integral variables
            if (!utils::DoubleApproxEqual(value, 1.) &&
                !utils::DoubleApproxEqual(value, 0.)) {
                is_solution_integral = false;
                num_fractional_vars++;
            }
            // Perform an eviction at this node
            if (value > 0.) {
                evictions.push_back(std::make_tuple(src_idx, flow_id, value));
                if (dst_idx != std::numeric_limits<size_t>::max()) {
                    inductions.push_back(std::make_tuple(dst_idx, flow_id, value));
                }
            }
        }
    }
    // Populate the inductions list for source nodes
    for (const auto& item : indices_) {
        const std::string& flow_id = item.first;
        const size_t src_idx = item.second.front() + z_ - 1;
        inductions.push_back(std::make_tuple(src_idx, flow_id, 1.));
    }
    // Sort the eviction schedule
    std::sort(evictions.begin(), evictions.end(), [](
        const std::tuple<size_t, std::string, double>& a,
            const std::tuple<size_t, std::string, double>& b) {
                return (std::get<0>(a) < std::get<0>(b)); });

    // Sort the induction schedule
    std::sort(inductions.begin(), inductions.end(), [](
        const std::tuple<size_t, std::string, double>& a,
            const std::tuple<size_t, std::string, double>& b) {
                return (std::get<0>(a) < std::get<0>(b)); });
    // Debug
    std::cout << "Solution is "
              << (is_solution_integral ? "INTEGRAL." : "FRACTIONAL.") << std::endl;

    // For fractional solutions, output the fraction of non-integer decision variables
    if (!is_solution_integral) {
        double percentage_fractional = (num_fractional_vars * 100.) / flows.numVariables();
        std::cout << "Warning: Solution has " << num_fractional_vars << " fractional "
                  << "decision variables (" << flows.numVariables() << " in total => "
                  << std::fixed << std::setprecision(2) << percentage_fractional
                  << "%)." << std::endl;
    }
    // Finally, run the cache simulator and validate the computed LP solution
    belatedly::CacheSimulator simulator(packets_file_path_, is_solution_integral,
                                        is_forced_, z_, c_, trace_, evictions,
                                        inductions);
    return simulator.run();
}

/**
 * Solve the MCMCF instance for the given trace.
 */
void MCMCFSolver::solve() {

    // Populate the internal structs
    setup();
    std::list<Edge> edges;
    populateOutOfCchEdges(edges);

    // Create optimization model
    GRBEnv env = GRBEnv();
    GRBModel model = GRBModel(env);
    belatedly::FlowDict flows = belatedly::FlowDict(model);

    // If the solution does not already exist, re-run the optimizer
    bool is_loaded_from_file = loadOptimizedModelSolution(flows);
    double opt_cost = std::numeric_limits<double>::max();
    if (!is_loaded_from_file) {

        // Create variables, populate constraints
        createFlowVariables(model, flows, edges);
        populateCchCapacityConstraints(model, flows);
        populateSplitNodeFlowConstraints(model, flows);
        populateFlowConservationConstraints(model, flows);

        // Compute the optimal solution
        model.set(GRB_IntAttr_ModelSense, GRB_MINIMIZE);
        model.set(GRB_DoubleParam_NodefileStart, 5);
        model.set(GRB_IntParam_Threads, 1);
        model.set(GRB_IntParam_Method, 1);
        model.optimize();

        // Populate the solution vector
        int status = model.get(GRB_IntAttr_Status);
        assert(status == GRB_OPTIMAL); // Sanity check
        for (size_t idx = 0; idx < flows.numVariables(); idx++) {
            solution_.push_back(flows.getVariableAt(idx).get(GRB_DoubleAttr_X));
        }

        // Save the optimal solution
        saveOptimizedModelSolution(flows);

        // Fetch the optimal cost
        opt_cost = model.get(GRB_DoubleAttr_ObjVal);
        for (const auto& iter : indices_) {
            size_t first_idx = iter.second.front();
            opt_cost += idx_to_miss_cost_.at(first_idx);
        }

        // Print the cost corresponding to the optimal solution
        std::cout << "Optimal cost is: " << std::fixed
                  << std::setprecision(3) << opt_cost
                  << std::endl << std::endl;
    }

    // Note: This point onwards, we cannot assume that the Gurobi model parameters are
    // available (since the solution may have been loaded from file). Instead, we use
    // the flow mappings and the solutions vector in the remainder of the pipeline.
    double lower_bound = getCostLowerBound(flows, edges);
    double upper_bound = getCostUpperBoundAndValidateSolution(flows, edges);
    double delta_percent = ((upper_bound - lower_bound) / lower_bound) * 100;

    // Debug
    std::cout << "Lower-bound (LB) on the total latency cost is: "
              << std::fixed << std::setprecision(3) << lower_bound
              << "." << std::endl;

    std::cout << "Upper-bound (UB) on the total latency cost is: "
              << std::fixed << std::setprecision(3) << upper_bound
              << " (" << delta_percent << "% worse than LB)."
              << std::endl << std::endl;

    // If the model was solved in this instance, also validate the opt_cost
    if (!is_loaded_from_file) {
        assert(utils::DoubleApproxEqual(opt_cost, lower_bound, 1e-2));
    }
}

int main(int argc, char **argv) {
    // Parameters
    uint z;
    double c_scale;
    std::string trace_fp;
    std::string model_fp;
    std::string packets_fp;

    // Program options
    bopt::variables_map variables;
    bopt::options_description desc{"BELATEDLY's MCMCF-based solver for"
                                   " the Delayed Hits caching problem"};
    try {
        // Command-line arguments
        desc.add_options()
            ("help",        "Prints this message")
            ("trace",       bopt::value<std::string>(&trace_fp)->required(),            "Input trace file path")
            ("cscale",      bopt::value<double>(&c_scale)->required(),                  "Parameter: Cache size (%Concurrent Flows)")
            ("zfactor",     bopt::value<uint>(&z)->required(),                          "Parameter: Z")
            ("model",       bopt::value<std::string>(&model_fp)->default_value(""),     "[Optional] Output model file prefix (for checkpointing)")
            ("packets",     bopt::value<std::string>(&packets_fp)->default_value(""),   "[Optional] Output packets file path")
            ("forced,f",                                                                "[Optional] Use forced admission");

        // Parse model parameters
        bopt::store(bopt::parse_command_line(argc, argv, desc), variables);

        // Handle help flag
        if (variables.count("help")) {
            std::cout << desc << std::endl;
            return 0;
        }
        bopt::notify(variables);
    }
    // Flag argument errors
    catch(const bopt::required_option& e) {
        std::cerr << "Error: " << e.what() << std::endl;
        return 1;
    }
    catch(...) {
        std::cerr << "Unknown Error." << std::endl;
        return 1;
    }

    // Use forced admission?
    bool is_forced = (variables.count("forced") != 0);

    // Parse the trace file, and compute the absolute cache size
    std::vector<std::string> trace = utils::parseTrace(trace_fp);
    size_t num_cfs = utils::getFlowCounts(trace).num_concurrent_flows;
    uint c = std::max(1u, static_cast<uint>(round((num_cfs * c_scale) / 100.)));

    // Debug
    std::cout << "Optimizing trace: " << trace_fp << " (with " << num_cfs
              << " concurrent flows) using z = " << z << ", c = " << c
              << ", and " << (is_forced ? "forced" : "optional")
              << " admission." << std::endl;

    // Instantiate the solver and optimize
    MCMCFSolver solver(trace_fp, model_fp, packets_fp,
                       is_forced, z, c, trace);
    solver.solve();
}

max_stars_repo_name: Ahziu/Delayed-Hits

max_issues_repo_name: Ahziu/Delayed-Hits

max_forks_repo_name: Ahziu/Delayed-Hits


text:
#include "test.h"

#include "loki/worker.h"
#include "odin/worker.h"
#include "thor/worker.h"

#include <boost/property_tree/ptree.hpp>

#include <algorithm> // std::copy

#if !defined(VALHALLA_SOURCE_DIR)
#define VALHALLA_SOURCE_DIR
#endif

namespace bpt = boost::property_tree;

using namespace valhalla;
namespace vt = valhalla::thor;
namespace vk = valhalla::loki;
namespace vo = valhalla::odin;

#include "mjolnir/directededgebuilder.h"
#include "mjolnir/graphtilebuilder.h"

namespace {

void write_config(const std::string& filename) {
  std::ofstream file;
  try {
    file.open(filename, std::ios_base::trunc);
    file << "{ \
      \"mjolnir\": { \
      \"concurrency\": 1, \
       \"tile_dir\": \"test/data/paris_bss_tiles\", \
      } \
    }";
  } catch (...) {}
  file.close();
}

// NOTE: the default radius is set to 10 below. this is because of the stupid way we connect transit
// and bike share stations to the graph by essentially duplicating the edge instead of creating a
// false node. if you duplicate the edge but you use a radius of 0, you get a random edge candidate in
// loki which leads to random results when you make changes to the way the tiles are built. so to
// avoid that we set a radius here to get both sets of edges and let the algorithm take the cheaper
// one. this only worked before by luck
boost::property_tree::ptree get_conf(const char* tiles) {
  std::stringstream ss;
  ss << R"({
      "mjolnir":{"tile_dir":"test/data/)"
     << tiles << R"(", "concurrency": 1},
      "loki":{
        "actions":["route"],
        "logging":{"long_request": 100},
        "service_defaults":{"minimum_reachability": 2,"radius": 10,"search_cutoff": 35000, "node_snap_tolerance": 5, "street_side_tolerance": 5, "heading_tolerance": 60, "street_side_max_distance": 1000}
      },
      "thor":{"logging":{"long_request": 100}},
      "odin":{"logging":{"long_request": 100}},
      "skadi":{"actons":["height"],"logging":{"long_request": 5}},
      "meili":{"customizable": ["turn_penalty_factor","max_route_distance_factor","max_route_time_factor","search_radius"],
              "mode":"auto","grid":{"cache_size":100240,"size":500},
              "default":{"beta":3,"breakage_distance":2000,"geometry":false,"gps_accuracy":5.0,"interpolation_distance":10,
              "max_route_distance_factor":5,"max_route_time_factor":5,"max_search_radius":200,"route":true,
              "search_radius":15.0,"sigma_z":4.07,"turn_penalty_factor":200}},
      "service_limits": {
        "auto": {"max_distance": 5000000.0, "max_locations": 20,"max_matrix_distance": 400000.0,"max_matrix_locations": 50},
        "auto_shorter": {"max_distance": 5000000.0,"max_locations": 20,"max_matrix_distance": 400000.0,"max_matrix_locations": 50},
        "bicycle": {"max_distance": 500000.0,"max_locations": 50,"max_matrix_distance": 200000.0,"max_matrix_locations": 50},
        "bus": {"max_distance": 5000000.0,"max_locations": 50,"max_matrix_distance": 400000.0,"max_matrix_locations": 50},
        "hov": {"max_distance": 5000000.0,"max_locations": 20,"max_matrix_distance": 400000.0,"max_matrix_locations": 50},
        "isochrone": {"max_contours": 4,"max_distance": 25000.0,"max_locations": 1,"max_time": 120},
        "max_avoid_locations": 50,"max_radius": 200,"max_reachability": 100,"max_alternates":2,
        "multimodal": {"max_distance": 500000.0,"max_locations": 50,"max_matrix_distance": 0.0,"max_matrix_locations": 0},
        "pedestrian": {"max_distance": 250000.0,"max_locations": 50,"max_matrix_distance": 200000.0,"max_matrix_locations": 50,"max_transit_walking_distance": 10000,"min_transit_walking_distance": 1},
        "skadi": {"max_shape": 750000,"min_resample": 10.0},
        "trace": {"max_distance": 200000.0,"max_gps_accuracy": 100.0,"max_search_radius": 100,"max_shape": 16000,"max_best_paths":4,"max_best_paths_shape":100},
        "transit": {"max_distance": 500000.0,"max_locations": 50,"max_matrix_distance": 200000.0,"max_matrix_locations": 50},
        "truck": {"max_distance": 5000000.0,"max_locations": 20,"max_matrix_distance": 400000.0,"max_matrix_locations": 50},
        "bikeshare": {"max_distance": 5000000.0,"max_locations": 20,"max_matrix_distance": 400000.0,"max_matrix_locations": 50}
      }
    })";
  boost::property_tree::ptree conf;
  rapidjson::read_json(ss, conf);
  return conf;
}

struct route_tester {
  route_tester(const boost::property_tree::ptree& _conf)
      : conf(_conf), reader(std::make_shared<GraphReader>(conf.get_child("mjolnir"))),
        loki_worker(conf, reader), thor_worker(conf, reader), odin_worker(conf) {
  }
  Api test(const std::string& request_json) {
    Api request;
    ParseApi(request_json, valhalla::Options::route, request);
    loki_worker.route(request);
    thor_worker.route(request);
    odin_worker.narrate(request);
    return request;
  }
  boost::property_tree::ptree conf;
  std::shared_ptr<GraphReader> reader;
  vk::loki_worker_t loki_worker;
  vt::thor_worker_t thor_worker;
  vo::odin_worker_t odin_worker;
};

using TravelMode = valhalla::DirectionsLeg_TravelMode;
using BssManeuverType = valhalla::DirectionsLeg_Maneuver_BssManeuverType;

void test(const std::string& request,
          const std::vector<TravelMode>& expected_travel_modes,
          const std::vector<std::string>& expected_route,
          // We mark only the maneuvers that are RentBike and ReturnBike
          const std::map<size_t, BssManeuverType>& expected_bss_maneuver) {

  auto conf = get_conf("paris_bss_tiles");
  route_tester tester(conf);
  auto response = tester.test(request);
  const auto& legs = response.trip().routes(0).legs();
  const auto& directions = response.directions().routes(0).legs();

  std::cout << legs.begin()->shape() << std::endl;
  auto shape = midgard::decode<std::vector<midgard::PointLL>>(legs.begin()->shape());
  for (const auto& node : legs.begin()->node()) {
    if (node.has_edge()) {
      std::vector<midgard::PointLL> subshape(shape.begin() + node.edge().begin_shape_index(),
                                             shape.begin() + node.edge().end_shape_index() + 1);
      std::cout << midgard::encode(subshape) << std::endl;
    }
  }

  EXPECT_EQ(legs.size(), 1) << "Should have 1 leg";

  // We going to count how many times the travel_mode has been changed
  std::vector<valhalla::DirectionsLeg_TravelMode> travel_modes;

  std::vector<std::string> route;

  for (const auto& d : directions) {
    size_t idx = -1;
    for (const auto& m : d.maneuver()) {
      auto it = expected_bss_maneuver.find(++idx);
      if (it == expected_bss_maneuver.end()) {
        EXPECT_EQ(m.bss_maneuver_type(),
                  BssManeuverType::DirectionsLeg_Maneuver_BssManeuverType_kNoneAction)
            << "BSS maneuver type at " + std::to_string(idx) + " is incorrect";

      } else {
        EXPECT_EQ(m.bss_maneuver_type(), it->second)
            << "BSS maneuver type at " + std::to_string(idx) + " is incorrect";
      }
      travel_modes.push_back(m.travel_mode());
      std::string name;
      for (const auto& n : m.street_name()) {
        name += n.value() + " ";
      }
      if (!name.empty()) {
        name.pop_back();
      }
      if (!name.empty()) {
        route.push_back(name);
      }
    }
  }

  travel_modes.erase(std::unique(travel_modes.begin(), travel_modes.end()), travel_modes.end());
  std::copy(travel_modes.begin(), travel_modes.end(), std::ostream_iterator<int>(std::cout, ", "));
  std::cout << std::endl;
  EXPECT_TRUE(std::equal(travel_modes.begin(), travel_modes.end(), expected_travel_modes.begin(),
                         expected_travel_modes.end()))
      << "Should have " + std::to_string(expected_travel_modes.size()) + " travel_modes";

  std::copy(route.begin(), route.end(), std::ostream_iterator<std::string>(std::cout, ", "));
  std::cout << std::endl;
  EXPECT_TRUE(std::equal(route.begin(), route.end(), expected_route.begin(), expected_route.end()))
      << "The route is incorrect";
}

/*
 * In this test, we make it free to rent/return, so the algorithm will prefer bike share routes
 */
TEST(AstarBss, test_With_Mode_Changes) {
  std::string request =
      R"({"locations":[{"lat":48.86481,"lon":2.361015},{"lat":48.859782,"lon":2.36101}],"costing":"bikeshare",
	       "costing_options":{"pedestrian":{"bss_rent_cost":0,"bss_rent_penalty":0},
	                          "bicycle"   :{"bss_return_cost":0,"bss_return_penalty":0}}})";
  std::vector<valhalla::DirectionsLeg_TravelMode>
      expected_travel_modes{valhalla::DirectionsLeg_TravelMode::DirectionsLeg_TravelMode_kPedestrian,
                            valhalla::DirectionsLeg_TravelMode::DirectionsLeg_TravelMode_kBicycle,
                            valhalla::DirectionsLeg_TravelMode::DirectionsLeg_TravelMode_kPedestrian};

  std::vector<std::string> expected_route{"Rue Perrée",        "Rue Perrée",        "Rue Perrée",
                                          "Rue Caffarelli",    "Rue de Bretagne",   "Rue de Turenne",
                                          "Rue du Parc Royal", "Place de Thorigny", "Rue de la Perle",
                                          "Rue de la Perle"};

  const std::map<size_t, BssManeuverType>&
      expected_bss_maneuver{{2, DirectionsLeg_Maneuver_BssManeuverType_kRentBikeAtBikeShare},
                            {9, DirectionsLeg_Maneuver_BssManeuverType_kReturnBikeAtBikeShare}};

  test(request, expected_travel_modes, expected_route, expected_bss_maneuver);
}

/*
 * In this test, we increase the bss rent/return penalty considerably, so the algorithm will avoid
 * bike share stations
 */
TEST(AstarBss, test_BSS_mode_Without_Mode_Changes) {

  std::string request =
      R"({"locations":[{"lat":48.865020,"lon":2.369113},{"lat":48.859782,"lon":2.36101}],
	       "costing":"bikeshare",
	       "costing_options":{"pedestrian":{"bss_rent_cost":0,"bss_rent_penalty":1800},
	                          "bicycle"   :{"bss_return_cost":0,"bss_return_penalty":1800}}})";

  std::vector<valhalla::DirectionsLeg_TravelMode> expected_travel_modes{
      valhalla::DirectionsLeg_TravelMode::DirectionsLeg_TravelMode_kPedestrian};

  // yes... the departure is still projected on the bss connection..
  std::vector<std::string> expected_route{"Rue du Grand Prieuré", "Rue de Crussol",
                                          "Rue Amelot",           "Place Pasdeloup",
                                          "Boulevard du Temple",  "Rue des Filles du Calvaire",
                                          "Rue de Turenne",       "Rue Vieille du Temple",
                                          "Rue de la Perle"};

  const std::map<size_t, BssManeuverType>& expected_bss_maneuver{};

  test(request, expected_travel_modes, expected_route, expected_bss_maneuver);
}

/*
 * In this test, we increase the bss rent/return cost considerably, so the algorithm will avoid bike
 * share stations
 */
TEST(AstarBss, test_BSS_mode_Without_Mode_Changes_2) {
  std::string request =
      R"({"locations":[{"lat":48.865020,"lon":2.369113},{"lat":48.859782,"lon":2.36101}],
	       "costing":"bikeshare",
	       "costing_options":{"pedestrian":{"bss_rent_cost":1800,"bss_rent_penalty":0},
	                          "bicycle"   :{"bss_return_cost":1800,"bss_return_penalty":0}}})";

  std::vector<valhalla::DirectionsLeg_TravelMode> expected_travel_modes{
      valhalla::DirectionsLeg_TravelMode::DirectionsLeg_TravelMode_kPedestrian};

  // yes... the departure is still projected on the bss connection..
  std::vector<std::string> expected_route{"Rue du Grand Prieuré", "Rue de Crussol",
                                          "Rue Amelot",           "Place Pasdeloup",
                                          "Boulevard du Temple",  "Rue des Filles du Calvaire",
                                          "Rue de Turenne",       "Rue Vieille du Temple",
                                          "Rue de la Perle"};

  const std::map<size_t, BssManeuverType>& expected_bss_maneuver{};

  test(request, expected_travel_modes, expected_route, expected_bss_maneuver);
}

// We test if the bss connection edges respect the forward/reverse access
TEST(AstarBss, test_With_Mode_Changes_2) {
  std::string request =
      R"({"locations":[{"lat":48.8601411,"lon":2.3716413},{"lat":48.8594916,"lon":2.3602581}],"costing":"bikeshare",
	       "costing_options":{"pedestrian":{"bss_rent_cost":0,"bss_rent_penalty":0},
	                          "bicycle"   :{"bss_return_cost":0,"bss_return_penalty":0}}})";
  std::vector<valhalla::DirectionsLeg_TravelMode>
      expected_travel_modes{valhalla::DirectionsLeg_TravelMode::DirectionsLeg_TravelMode_kPedestrian,
                            valhalla::DirectionsLeg_TravelMode::DirectionsLeg_TravelMode_kBicycle,
                            valhalla::DirectionsLeg_TravelMode::DirectionsLeg_TravelMode_kPedestrian};
  std::vector<std::string> expected_route{"Rue Pelée",
                                          "Rue Pelée",
                                          "Rue Alphonse Baudin",
                                          "Rue Saint-Sébastien",
                                          "Boulevard Beaumarchais",
                                          "Rue du Pont aux Choux",
                                          "Rue de Turenne",
                                          "Rue du Parc Royal",
                                          "Place de Thorigny",
                                          "Rue de la Perle",
                                          "Rue de la Perle",
                                          "Rue Vieille du Temple"};
  const std::map<size_t, BssManeuverType>&
      expected_bss_maneuver{{1, DirectionsLeg_Maneuver_BssManeuverType_kRentBikeAtBikeShare},
                            {10, DirectionsLeg_Maneuver_BssManeuverType_kReturnBikeAtBikeShare}};

  test(request, expected_travel_modes, expected_route, expected_bss_maneuver);
}

// When pedestrian is chosen as travel_mode, the departure edge must NOT be a bss connections edge
TEST(AstarBss, test_Pedestrian) {
  std::string request =
      R"({"locations":[{"lat":48.859895,"lon":2.3610976338},{"lat":48.86271911,"lon":2.367111146}],"costing":"pedestrian"})";
  std::vector<valhalla::DirectionsLeg_TravelMode> expected_travel_modes{
      valhalla::DirectionsLeg_TravelMode::DirectionsLeg_TravelMode_kPedestrian};
  std::vector<std::string> expected_route{"Rue de la Perle", "Rue Vieille du Temple", "Rue Froissart",
                                          "Rue Commines", "Rue Amelot"};
  // There shouldn't be any bss maneuvers
  const std::map<size_t, BssManeuverType>& expected_bss_maneuver{};

  test(request, expected_travel_modes, expected_route, expected_bss_maneuver);
}

// When bicycle is chosen as travel_mode, the departure edge must NOT be a bss connections edge
TEST(AstarBss, test_Bicycle) {
  std::string request =
      R"({"locations":[{"lat":48.859895,"lon":2.3610976338},{"lat":48.86271911,"lon":2.367111146}],"costing":"bicycle"})";
  std::vector<valhalla::DirectionsLeg_TravelMode> expected_travel_modes{
      valhalla::DirectionsLeg_TravelMode::DirectionsLeg_TravelMode_kBicycle};
  std::vector<std::string> expected_route{"Rue de la Perle", "Rue des Archives", "Rue de Bretagne",
                                          "Rue Commines", "Rue Amelot"};
  // There shouldn't be any bss maneuvers
  const std::map<size_t, BssManeuverType>& expected_bss_maneuver{};

  test(request, expected_travel_modes, expected_route, expected_bss_maneuver);
}

// When auto is chosen as travel_mode, the departure edge must NOT be a bss connections edge
TEST(AstarBss, test_Auto) {
  std::string request =
      R"({"locations":[{"lat":48.859895,"lon":2.3610976338},{"lat":48.86271911,"lon":2.367111146}],"costing":"auto"})";
  std::vector<valhalla::DirectionsLeg_TravelMode> expected_travel_modes{
      valhalla::DirectionsLeg_TravelMode::DirectionsLeg_TravelMode_kDrive};
  std::vector<std::string> expected_route{"Rue de la Perle",   "Rue Vieille du Temple",
                                          "Rue Barbette",      "Rue Elzévir",
                                          "Place de Thorigny", "Rue de Thorigny",
                                          "Rue Debelleyme",    "Rue de Turenne",
                                          "Rue Commines",      "Rue Amelot"};
  // There shouldn't be any bss maneuvers
  const std::map<size_t, BssManeuverType>& expected_bss_maneuver{};

  test(request, expected_travel_modes, expected_route, expected_bss_maneuver);
}

// When auto is chosen as travel_mode, the departure edge must NOT be a bss connections edge
TEST(AstarBss, test_Truck) {
  std::string request =
      R"({"locations":[{"lat":48.859895,"lon":2.3610976338},{"lat":48.86271911,"lon":2.367111146}],"costing":"truck"})";
  std::vector<valhalla::DirectionsLeg_TravelMode> expected_travel_modes{
      valhalla::DirectionsLeg_TravelMode::DirectionsLeg_TravelMode_kDrive};
  std::vector<std::string> expected_route{"Rue de la Perle",        "Rue des Archives",
                                          "Rue Pastourelle",        "Rue du Temple",
                                          "Place de la République", "Boulevard du Temple",
                                          "Rue Oberkampf",          "Rue Amelot"};
  // There shouldn't be any bss maneuvers
  const std::map<size_t, BssManeuverType>& expected_bss_maneuver{};

  test(request, expected_travel_modes, expected_route, expected_bss_maneuver);
}

class AstarBSSTestEnv : public ::testing::Environment {
public:
  void SetUp() override {
    const std::string config_file = "test/test_astar_bss";
    write_config(config_file);
  }
};

} // anonymous namespace

int main(int argc, char* argv[]) {
  testing::AddGlobalTestEnvironment(new AstarBSSTestEnv);
  testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
}

ext: cc

max_stars_repo_name: flashspys/valhalla

max_issues_repo_name: flashspys/valhalla

max_forks_repo_name: flashspys/valhalla


text:
// Copyright Louis Dionne 2013-2016
// Distributed under the Boost Software License, Version 1.0.
// (See accompanying file LICENSE.md or copy at http://boost.org/LICENSE_1_0.txt)

#include <boost/hana/assert.hpp>
#include <boost/hana/at_key.hpp>
#include <boost/hana/fold_left.hpp>
#include <boost/hana/integral_constant.hpp>
#include <boost/hana/map.hpp>
#include <boost/hana/pair.hpp>
#include <boost/hana/range.hpp>
namespace hana = boost::hana;

//
// Make sure that http://stackoverflow.com/q/32702383/627587 works.
//

auto at = [](auto& map, auto key) -> auto& {
    return map[key];
};

template <typename Map, typename Keys>
auto& traverse(Map& map, Keys const& keys){
    return hana::fold_left(keys, map, at);
}

int main() {
    auto xs = hana::make_map(hana::make_pair(hana::int_c<0>,
                hana::make_map(hana::make_pair(hana::int_c<1>,
                    hana::make_map(hana::make_pair(hana::int_c<2>,
                        hana::make_map(hana::make_pair(hana::int_c<3>, 10))))))));

    int& i = traverse(xs, hana::range_c<int, 0, 4>);
    BOOST_HANA_RUNTIME_CHECK(i == 10);
    i = 99;
    BOOST_HANA_RUNTIME_CHECK(traverse(xs, hana::range_c<int, 0, 4>) == 99);
}

ext: cpp

max_issues_repo_name: lipper/arangodb

max_forks_repo_name: lipper/arangodb


text:
#include <boost/bind.hpp>
#include <cryptopp/dh.h>
#include <cryptopp/sha.h>
#include "CryptoConst.h"
#include "Log.h"
#include "Timestamp.h"
#include "RouterContext.h"
#include "Transports.h"
#include "SSU.h"
#include "SSUSession.h"

namespace i2p
{
namespace transport
{
	SSUSession::SSUSession (SSUServer& server, boost::asio::ip::udp::endpoint& remoteEndpoint,
		std::shared_ptr<const i2p::data::RouterInfo> router, bool peerTest ): TransportSession (router),
		m_Server (server), m_RemoteEndpoint (remoteEndpoint), m_Timer (GetService ()),
		m_PeerTest (peerTest),m_State (eSessionStateUnknown), m_IsSessionKey (false),
		m_RelayTag (0),m_Data (*this), m_IsDataReceived (false)
	{
		m_CreationTime = i2p::util::GetSecondsSinceEpoch ();
	}

	SSUSession::~SSUSession ()
	{
	}

	boost::asio::io_service& SSUSession::GetService ()
	{
		return IsV6 () ? m_Server.GetServiceV6 () : m_Server.GetService ();
	}

	void SSUSession::CreateAESandMacKey (const uint8_t * pubKey)
	{
		CryptoPP::DH dh (i2p::crypto::elgp, i2p::crypto::elgg);
		uint8_t sharedKey[256];
		if (!dh.Agree (sharedKey, m_DHKeysPair->privateKey, pubKey))
		{
		    LogPrint (eLogError, "Couldn't create shared key");
			return;
		};

		uint8_t * sessionKey = m_SessionKey, * macKey = m_MacKey;
		if (sharedKey[0] & 0x80)
		{
			sessionKey[0] = 0;
			memcpy (sessionKey + 1, sharedKey, 31);
			memcpy (macKey, sharedKey + 31, 32);
		}
		else if (sharedKey[0])
		{
			memcpy (sessionKey, sharedKey, 32);
			memcpy (macKey, sharedKey + 32, 32);
		}
		else
		{
			// find first non-zero byte
			uint8_t * nonZero = sharedKey + 1;
			while (!*nonZero)
			{
				nonZero++;
				if (nonZero - sharedKey > 32)
				{
					LogPrint ("First 32 bytes of shared key is all zeros. Ignored");
					return;
				}
			}

			memcpy (sessionKey, nonZero, 32);
			CryptoPP::SHA256().CalculateDigest(macKey, nonZero, 64 - (nonZero - sharedKey));
		}
		m_IsSessionKey = true;
		m_SessionKeyEncryption.SetKey (m_SessionKey);
		m_SessionKeyDecryption.SetKey (m_SessionKey);
	}

	void SSUSession::ProcessNextMessage (uint8_t * buf, size_t len, const boost::asio::ip::udp::endpoint& senderEndpoint)
	{
		m_NumReceivedBytes += len;
		i2p::transport::transports.UpdateReceivedBytes (len);
		if (m_State == eSessionStateIntroduced)
		{
			// HolePunch received
			LogPrint ("SSU HolePunch of ", len, " bytes received");
			m_State = eSessionStateUnknown;
			Connect ();
		}
		else
		{
			if (!len) return; // ignore zero-length packets
			if (m_State == eSessionStateEstablished)
				ScheduleTermination ();

			if (m_IsSessionKey && Validate (buf, len, m_MacKey)) // try session key first
				DecryptSessionKey (buf, len);
			else
			{
				// try intro key depending on side
				auto introKey = GetIntroKey ();
				if (introKey && Validate (buf, len, introKey))
					Decrypt (buf, len, introKey);
				else
				{
					// try own intro key
					auto address = i2p::context.GetRouterInfo ().GetSSUAddress ();
					if (!address)
					{
						LogPrint (eLogError, "SSU is not supported");
						return;
					}
					if (Validate (buf, len, address->key))
						Decrypt (buf, len, address->key);
					else
					{
						LogPrint (eLogError, "MAC verification failed ", len, " bytes from ", senderEndpoint);
						m_Server.DeleteSession (shared_from_this ());
						return;
					}
				}
			}
			// successfully decrypted
			ProcessMessage (buf, len, senderEndpoint);
		}
	}

	void SSUSession::ProcessMessage (uint8_t * buf, size_t len, const boost::asio::ip::udp::endpoint& senderEndpoint)
	{
		len -= (len & 0x0F); // %16, delete extra padding
		if (len <= sizeof (SSUHeader)) return; // drop empty message
		//TODO: since we are accessing a uint8_t this is unlikely to crash due to alignment but should be improved
		SSUHeader * header = (SSUHeader *)buf;
		switch (header->GetPayloadType ())
		{
			case PAYLOAD_TYPE_DATA:
				ProcessData (buf + sizeof (SSUHeader), len - sizeof (SSUHeader));
			break;
			case PAYLOAD_TYPE_SESSION_REQUEST:
				ProcessSessionRequest (buf, len, senderEndpoint);
			break;
			case PAYLOAD_TYPE_SESSION_CREATED:
				ProcessSessionCreated (buf, len);
			break;
			case PAYLOAD_TYPE_SESSION_CONFIRMED:
				ProcessSessionConfirmed (buf, len);
			break;
			case PAYLOAD_TYPE_PEER_TEST:
				LogPrint (eLogDebug, "SSU peer test received");
				ProcessPeerTest (buf + sizeof (SSUHeader), len - sizeof (SSUHeader), senderEndpoint);
			break;
			case PAYLOAD_TYPE_SESSION_DESTROYED:
			{
				LogPrint (eLogDebug, "SSU session destroy received");
				m_Server.DeleteSession (shared_from_this ());
				break;
			}
			case PAYLOAD_TYPE_RELAY_RESPONSE:
				ProcessRelayResponse (buf, len);
				if (m_State != eSessionStateEstablished)
					m_Server.DeleteSession (shared_from_this ());
			break;
			case PAYLOAD_TYPE_RELAY_REQUEST:
				LogPrint (eLogDebug, "SSU relay request received");
				ProcessRelayRequest (buf + sizeof (SSUHeader), len - sizeof (SSUHeader), senderEndpoint);
			break;
			case PAYLOAD_TYPE_RELAY_INTRO:
				LogPrint (eLogDebug, "SSU relay intro received");
				ProcessRelayIntro (buf + sizeof (SSUHeader), len - sizeof (SSUHeader));
			break;
			default:
				LogPrint (eLogWarning, "Unexpected SSU payload type ", (int)header->GetPayloadType ());
		}
	}

	void SSUSession::ProcessSessionRequest (uint8_t * buf, size_t len, const boost::asio::ip::udp::endpoint& senderEndpoint)
	{
		LogPrint (eLogDebug, "Session request received");
		m_RemoteEndpoint = senderEndpoint;
		if (!m_DHKeysPair)
			m_DHKeysPair = transports.GetNextDHKeysPair ();
		CreateAESandMacKey (buf + sizeof (SSUHeader));
		SendSessionCreated (buf + sizeof (SSUHeader));
	}

	void SSUSession::ProcessSessionCreated (uint8_t * buf, size_t len)
	{
		if (!m_RemoteRouter || !m_DHKeysPair)
		{
			LogPrint (eLogWarning, "Unsolicited session created message");
			return;
		}

		LogPrint (eLogDebug, "Session created received");
		m_Timer.cancel (); // connect timer
		SignedData s; // x,y, our IP, our port, remote IP, remote port, relayTag, signed on time
		uint8_t * payload = buf + sizeof (SSUHeader);
		uint8_t * y = payload;
		CreateAESandMacKey (y);
		s.Insert (m_DHKeysPair->publicKey, 256); // x
		s.Insert (y, 256); // y
		payload += 256;
		uint8_t addressSize = *payload;
		payload += 1; // size
		uint8_t * ourAddress = payload;
		boost::asio::ip::address ourIP;
		if (addressSize == 4) // v4
		{
			boost::asio::ip::address_v4::bytes_type bytes;
			memcpy (bytes.data (), ourAddress, 4);
			ourIP = boost::asio::ip::address_v4 (bytes);
		}
		else // v6
		{
			boost::asio::ip::address_v6::bytes_type bytes;
			memcpy (bytes.data (), ourAddress, 16);
			ourIP = boost::asio::ip::address_v6 (bytes);
		}
		s.Insert (ourAddress, addressSize); // our IP
		payload += addressSize; // address
		uint16_t ourPort = bufbe16toh (payload);
		s.Insert (payload, 2); // our port
		payload += 2; // port
		LogPrint ("Our external address is ", ourIP.to_string (), ":", ourPort);
		i2p::context.UpdateAddress (ourIP);
		if (m_RemoteEndpoint.address ().is_v4 ())
			s.Insert (m_RemoteEndpoint.address ().to_v4 ().to_bytes ().data (), 4); // remote IP v4
		else
			s.Insert (m_RemoteEndpoint.address ().to_v6 ().to_bytes ().data (), 16); // remote IP v6
		s.Insert<uint16_t> (htobe16 (m_RemoteEndpoint.port ())); // remote port
		s.Insert (payload, 8); // relayTag and signed on time
		m_RelayTag = bufbe32toh (payload);
		payload += 4; // relayTag
		payload += 4; // signed on time
		// decrypt signature
		size_t signatureLen = m_RemoteIdentity.GetSignatureLen ();
		size_t paddingSize = signatureLen & 0x0F; // %16
		if (paddingSize > 0) signatureLen += (16 - paddingSize);
		//TODO: since we are accessing a uint8_t this is unlikely to crash due to alignment but should be improved
		m_SessionKeyDecryption.SetIV (((SSUHeader *)buf)->iv);
		m_SessionKeyDecryption.Decrypt (payload, signatureLen, payload);
		// verify
		if (!s.Verify (m_RemoteIdentity, payload))
			LogPrint (eLogError, "SSU signature verification failed");
		m_RemoteIdentity.DropVerifier ();

		SendSessionConfirmed (y, ourAddress, addressSize + 2);
	}

	void SSUSession::ProcessSessionConfirmed (uint8_t * buf, size_t len)
	{
		LogPrint (eLogDebug, "Session confirmed received");
		uint8_t * payload = buf + sizeof (SSUHeader);
		payload++; // identity fragment info
		uint16_t identitySize = bufbe16toh (payload);
		payload += 2; // size of identity fragment
		m_RemoteIdentity.FromBuffer (payload, identitySize);
		m_Data.UpdatePacketSize (m_RemoteIdentity.GetIdentHash ());
		payload += identitySize; // identity
		payload += 4; // signed-on time
		size_t paddingSize = (payload - buf) + m_RemoteIdentity.GetSignatureLen ();
		paddingSize &= 0x0F;  // %16
		if (paddingSize > 0) paddingSize = 16 - paddingSize;
		payload += paddingSize;
		// TODO: verify signature (need data from session request), payload points to signature
		m_Data.Send (CreateDeliveryStatusMsg (0));
		Established ();
	}

	void SSUSession::SendSessionRequest ()
	{
		auto introKey = GetIntroKey ();
		if (!introKey)
		{
			LogPrint (eLogError, "SSU is not supported");
			return;
		}

		uint8_t buf[320 + 18]; // 304 bytes for ipv4, 320 for ipv6
		uint8_t * payload = buf + sizeof (SSUHeader);
		memcpy (payload, m_DHKeysPair->publicKey, 256); // x
		bool isV4 = m_RemoteEndpoint.address ().is_v4 ();
		if (isV4)
		{
			payload[256] = 4;
			memcpy (payload + 257, m_RemoteEndpoint.address ().to_v4 ().to_bytes ().data(), 4);
		}
		else
		{
			payload[256] = 16;
			memcpy (payload + 257, m_RemoteEndpoint.address ().to_v6 ().to_bytes ().data(), 16);
		}

		uint8_t iv[16];
		CryptoPP::RandomNumberGenerator& rnd = i2p::context.GetRandomNumberGenerator ();
		rnd.GenerateBlock (iv, 16); // random iv
		FillHeaderAndEncrypt (PAYLOAD_TYPE_SESSION_REQUEST, buf, isV4 ? 304 : 320, introKey, iv, introKey);
		m_Server.Send (buf, isV4 ? 304 : 320, m_RemoteEndpoint);
	}

	void SSUSession::SendRelayRequest (uint32_t iTag, const uint8_t * iKey)
	{
		auto address = i2p::context.GetRouterInfo ().GetSSUAddress ();
		if (!address)
		{
			LogPrint (eLogError, "SSU is not supported");
			return;
		}

		uint8_t buf[96 + 18];
		uint8_t * payload = buf + sizeof (SSUHeader);
		htobe32buf (payload, iTag);
		payload += 4;
		*payload = 0; // no address
		payload++;
		htobuf16(payload, 0); // port = 0
		payload += 2;
		*payload = 0; // challenge
		payload++;
		memcpy (payload, (const uint8_t *)address->key, 32);
		payload += 32;
		CryptoPP::RandomNumberGenerator& rnd = i2p::context.GetRandomNumberGenerator ();
		htobe32buf (payload, rnd.GenerateWord32 ()); // nonce

		uint8_t iv[16];
		rnd.GenerateBlock (iv, 16); // random iv
		if (m_State == eSessionStateEstablished)
			FillHeaderAndEncrypt (PAYLOAD_TYPE_RELAY_REQUEST, buf, 96, m_SessionKey, iv, m_MacKey);
		else
			FillHeaderAndEncrypt (PAYLOAD_TYPE_RELAY_REQUEST, buf, 96, iKey, iv, iKey);
		m_Server.Send (buf, 96, m_RemoteEndpoint);
	}

	void SSUSession::SendSessionCreated (const uint8_t * x)
	{
		auto introKey = GetIntroKey ();
		auto address = IsV6 () ? i2p::context.GetRouterInfo ().GetSSUV6Address () :
			i2p::context.GetRouterInfo ().GetSSUAddress (true); //v4 only
		if (!introKey || !address)
		{
			LogPrint (eLogError, "SSU is not supported");
			return;
		}
		CryptoPP::RandomNumberGenerator& rnd = i2p::context.GetRandomNumberGenerator ();
		SignedData s; // x,y, remote IP, remote port, our IP, our port, relayTag, signed on time
		s.Insert (x, 256); // x

		uint8_t buf[384 + 18];
		uint8_t * payload = buf + sizeof (SSUHeader);
		memcpy (payload, m_DHKeysPair->publicKey, 256);
		s.Insert (payload, 256); // y
		payload += 256;
		if (m_RemoteEndpoint.address ().is_v4 ())
		{
			// ipv4
			*payload = 4;
			payload++;
			memcpy (payload, m_RemoteEndpoint.address ().to_v4 ().to_bytes ().data(), 4);
			s.Insert (payload, 4); // remote endpoint IP V4
			payload += 4;
		}
		else
		{
			// ipv6
			*payload = 16;
			payload++;
			memcpy (payload, m_RemoteEndpoint.address ().to_v6 ().to_bytes ().data(), 16);
			s.Insert (payload, 16); // remote endpoint IP V6
			payload += 16;
		}
		htobe16buf (payload, m_RemoteEndpoint.port ());
		s.Insert (payload, 2); // remote port
		payload += 2;
		if (address->host.is_v4 ())
			s.Insert (address->host.to_v4 ().to_bytes ().data (), 4); // our IP V4
		else
			s.Insert (address->host.to_v6 ().to_bytes ().data (), 16); // our IP V6
		s.Insert<uint16_t> (htobe16 (address->port)); // our port
		uint32_t relayTag = 0;
		if (i2p::context.GetRouterInfo ().IsIntroducer ())
		{
			relayTag = rnd.GenerateWord32 ();
			if (!relayTag) relayTag = 1;
			m_Server.AddRelay (relayTag, m_RemoteEndpoint);
		}
		htobe32buf (payload, relayTag);
		payload += 4; // relay tag
		htobe32buf (payload, i2p::util::GetSecondsSinceEpoch ()); // signed on time
		payload += 4;
		s.Insert (payload - 8, 8); // relayTag and signed on time
		s.Sign (i2p::context.GetPrivateKeys (), payload); // DSA signature
		// TODO: fill padding with random data

		uint8_t iv[16];
		rnd.GenerateBlock (iv, 16); // random iv
		// encrypt signature and padding with newly created session key
		size_t signatureLen = i2p::context.GetIdentity ().GetSignatureLen ();
		size_t paddingSize = signatureLen & 0x0F; // %16
		if (paddingSize > 0) signatureLen += (16 - paddingSize);
		m_SessionKeyEncryption.SetIV (iv);
		m_SessionKeyEncryption.Encrypt (payload, signatureLen, payload);
		payload += signatureLen;
		size_t msgLen = payload - buf;

		// encrypt message with intro key
		FillHeaderAndEncrypt (PAYLOAD_TYPE_SESSION_CREATED, buf, msgLen, introKey, iv, introKey);
		Send (buf, msgLen);
	}

	void SSUSession::SendSessionConfirmed (const uint8_t * y, const uint8_t * ourAddress, size_t ourAddressLen)
	{
		uint8_t buf[512 + 18];
		uint8_t * payload = buf + sizeof (SSUHeader);
		*payload = 1; // 1 fragment
		payload++; // info
		size_t identLen = i2p::context.GetIdentity ().GetFullLen (); // 387+ bytes
		htobe16buf (payload, identLen);
		payload += 2; // cursize
		i2p::context.GetIdentity ().ToBuffer (payload, identLen);
		payload += identLen;
		uint32_t signedOnTime = i2p::util::GetSecondsSinceEpoch ();
		htobe32buf (payload, signedOnTime); // signed on time
		payload += 4;
		auto signatureLen = i2p::context.GetIdentity ().GetSignatureLen ();
		size_t paddingSize = ((payload - buf) + signatureLen)%16;
		if (paddingSize > 0) paddingSize = 16 - paddingSize;
		// TODO: fill padding
		payload += paddingSize; // padding size

		// signature
		SignedData s; // x,y, our IP, our port, remote IP, remote port, relayTag, our signed on time
		s.Insert (m_DHKeysPair->publicKey, 256); // x
		s.Insert (y, 256); // y
		s.Insert (ourAddress, ourAddressLen); // our address/port as seem by party
		if (m_RemoteEndpoint.address ().is_v4 ())
			s.Insert (m_RemoteEndpoint.address ().to_v4 ().to_bytes ().data (), 4); // remote IP V4
		else
			s.Insert (m_RemoteEndpoint.address ().to_v6 ().to_bytes ().data (), 16); // remote IP V6
		s.Insert<uint16_t> (htobe16 (m_RemoteEndpoint.port ())); // remote port
		s.Insert (htobe32 (m_RelayTag)); // relay tag
		s.Insert (htobe32 (signedOnTime)); // signed on time
		s.Sign (i2p::context.GetPrivateKeys (), payload); // DSA signature
		payload += signatureLen;

		size_t msgLen = payload - buf;
		uint8_t iv[16];
		CryptoPP::RandomNumberGenerator& rnd = i2p::context.GetRandomNumberGenerator ();
		rnd.GenerateBlock (iv, 16); // random iv
		// encrypt message with session key
		FillHeaderAndEncrypt (PAYLOAD_TYPE_SESSION_CONFIRMED, buf, msgLen, m_SessionKey, iv, m_MacKey);
		Send (buf, msgLen);
	}

	void SSUSession::ProcessRelayRequest (uint8_t * buf, size_t len, const boost::asio::ip::udp::endpoint& from)
	{
		uint32_t relayTag = bufbe32toh (buf);
		auto session = m_Server.FindRelaySession (relayTag);
		if (session)
		{
			buf += 4; // relay tag
			uint8_t size = *buf;
			buf++; // size
			buf += size; // address
			buf += 2; // port
			uint8_t challengeSize = *buf;
			buf++; // challenge size
			buf += challengeSize;
			uint8_t * introKey = buf;
			buf += 32; // introkey
			uint32_t nonce = bufbe32toh (buf);
			SendRelayResponse (nonce, from, introKey, session->m_RemoteEndpoint);
			SendRelayIntro (session.get (), from);
		}
	}

	void SSUSession::SendRelayResponse (uint32_t nonce, const boost::asio::ip::udp::endpoint& from,
		const uint8_t * introKey, const boost::asio::ip::udp::endpoint& to)
	{
		uint8_t buf[80 + 18]; // 64 Alice's ipv4 and 80 Alice's ipv6
		uint8_t * payload = buf + sizeof (SSUHeader);
		// Charlie's address always v4
		if (!to.address ().is_v4 ())
		{
			LogPrint (eLogError, "Charlie's IP must be v4");
			return;
		}
		*payload = 4;
		payload++; // size
		htobe32buf (payload, to.address ().to_v4 ().to_ulong ()); // Charlie's IP
		payload += 4; // address
		htobe16buf (payload, to.port ()); // Charlie's port
		payload += 2; // port
		// Alice
		bool isV4 = from.address ().is_v4 (); // Alice's
		if (isV4)
		{
			*payload = 4;
			payload++; // size
			memcpy (payload, from.address ().to_v4 ().to_bytes ().data (), 4); // Alice's IP V4
			payload += 4; // address
		}
		else
		{
			*payload = 16;
			payload++; // size
			memcpy (payload, from.address ().to_v6 ().to_bytes ().data (), 16); // Alice's IP V6
			payload += 16; // address
		}
		htobe16buf (payload, from.port ()); // Alice's port
		payload += 2; // port
		htobe32buf (payload, nonce);

		if (m_State == eSessionStateEstablished)
		{
			// encrypt with session key
			FillHeaderAndEncrypt (PAYLOAD_TYPE_RELAY_RESPONSE, buf, isV4 ? 64 : 80);
			Send (buf, isV4 ? 64 : 80);
		}
		else
		{
			// ecrypt with Alice's intro key
			uint8_t iv[16];
			CryptoPP::RandomNumberGenerator& rnd = i2p::context.GetRandomNumberGenerator ();
			rnd.GenerateBlock (iv, 16); // random iv
			FillHeaderAndEncrypt (PAYLOAD_TYPE_RELAY_RESPONSE, buf, isV4 ? 64 : 80, introKey, iv, introKey);
			m_Server.Send (buf, isV4 ? 64 : 80, from);
		}
		LogPrint (eLogDebug, "SSU relay response sent");
	}

	void SSUSession::SendRelayIntro (SSUSession * session, const boost::asio::ip::udp::endpoint& from)
	{
		if (!session) return;
		// Alice's address always v4
		if (!from.address ().is_v4 ())
		{
			LogPrint (eLogError, "Alice's IP must be v4");
			return;
		}
		uint8_t buf[48 + 18];
		uint8_t * payload = buf + sizeof (SSUHeader);
		*payload = 4;
		payload++; // size
		htobe32buf (payload, from.address ().to_v4 ().to_ulong ()); // Alice's IP
		payload += 4; // address
		htobe16buf (payload, from.port ()); // Alice's port
		payload += 2; // port
		*payload = 0; // challenge size
		uint8_t iv[16];
		CryptoPP::RandomNumberGenerator& rnd = i2p::context.GetRandomNumberGenerator ();
		rnd.GenerateBlock (iv, 16); // random iv
		FillHeaderAndEncrypt (PAYLOAD_TYPE_RELAY_INTRO, buf, 48, session->m_SessionKey, iv, session->m_MacKey);
		m_Server.Send (buf, 48, session->m_RemoteEndpoint);
		LogPrint (eLogDebug, "SSU relay intro sent");
	}

	void SSUSession::ProcessRelayResponse (uint8_t * buf, size_t len)
	{
		LogPrint (eLogDebug, "Relay response received");
		uint8_t * payload = buf + sizeof (SSUHeader);
		uint8_t remoteSize = *payload;
		payload++; // remote size
		//boost::asio::ip::address_v4 remoteIP (bufbe32toh (payload));
		payload += remoteSize; // remote address
		//uint16_t remotePort = bufbe16toh (payload);
		payload += 2; // remote port
		uint8_t ourSize = *payload;
		payload++; // our size
		boost::asio::ip::address ourIP;
		if (ourSize == 4)
		{
			boost::asio::ip::address_v4::bytes_type bytes;
			memcpy (bytes.data (), payload, 4);
			ourIP = boost::asio::ip::address_v4 (bytes);
		}
		else
		{
			boost::asio::ip::address_v6::bytes_type bytes;
			memcpy (bytes.data (), payload, 16);
			ourIP = boost::asio::ip::address_v6 (bytes);
		}
		payload += ourSize; // our address
		uint16_t ourPort = bufbe16toh (payload);
		payload += 2; // our port
		LogPrint ("Our external address is ", ourIP.to_string (), ":", ourPort);
		i2p::context.UpdateAddress (ourIP);
	}

	void SSUSession::ProcessRelayIntro (uint8_t * buf, size_t len)
	{
		uint8_t size = *buf;
		if (size == 4)
		{
			buf++; // size
			boost::asio::ip::address_v4 address (bufbe32toh (buf));
			buf += 4; // address
			uint16_t port = bufbe16toh (buf);
			// send hole punch of 1 byte
			m_Server.Send (buf, 0, boost::asio::ip::udp::endpoint (address, port));
		}
		else
			LogPrint (eLogWarning, "Address size ", size, " is not supported");
	}

	void SSUSession::FillHeaderAndEncrypt (uint8_t payloadType, uint8_t * buf, size_t len,
		const uint8_t * aesKey, const uint8_t * iv, const uint8_t * macKey)
	{
		if (len < sizeof (SSUHeader))
		{
			LogPrint (eLogError, "Unexpected SSU packet length ", len);
			return;
		}
		//TODO: we are using a dirty solution here but should work for now
		SSUHeader * header = (SSUHeader *)buf;
		memcpy (header->iv, iv, 16);
		header->flag = payloadType << 4; // MSB is 0
		htobe32buf (&(header->time), i2p::util::GetSecondsSinceEpoch ());
		uint8_t * encrypted = &header->flag;
		uint16_t encryptedLen = len - (encrypted - buf);
		i2p::crypto::CBCEncryption encryption;
		encryption.SetKey (aesKey);
		encryption.SetIV (iv);
		encryption.Encrypt (encrypted, encryptedLen, encrypted);
		// assume actual buffer size is 18 (16 + 2) bytes more
		memcpy (buf + len, iv, 16);
		htobe16buf (buf + len + 16, encryptedLen);
		i2p::crypto::HMACMD5Digest (encrypted, encryptedLen + 18, macKey, header->mac);
	}

	void SSUSession::FillHeaderAndEncrypt (uint8_t payloadType, uint8_t * buf, size_t len)
	{
		if (len < sizeof (SSUHeader))
		{
			LogPrint (eLogError, "Unexpected SSU packet length ", len);
			return;
		}
		//TODO: we are using a dirty solution here but should work for now
		SSUHeader * header = (SSUHeader *)buf;
		i2p::context.GetRandomNumberGenerator ().GenerateBlock (header->iv, 16); // random iv
		m_SessionKeyEncryption.SetIV (header->iv);
		header->flag = payloadType << 4; // MSB is 0
		htobe32buf (&(header->time), i2p::util::GetSecondsSinceEpoch ());
		uint8_t * encrypted = &header->flag;
		uint16_t encryptedLen = len - (encrypted - buf);
		m_SessionKeyEncryption.Encrypt (encrypted, encryptedLen, encrypted);
		// assume actual buffer size is 18 (16 + 2) bytes more
		memcpy (buf + len, header->iv, 16);
		htobe16buf (buf + len + 16, encryptedLen);
		i2p::crypto::HMACMD5Digest (encrypted, encryptedLen + 18, m_MacKey, header->mac);
	}

	void SSUSession::Decrypt (uint8_t * buf, size_t len, const uint8_t * aesKey)
	{
		if (len < sizeof (SSUHeader))
		{
			LogPrint (eLogError, "Unexpected SSU packet length ", len);
			return;
		}
		//TODO: since we are accessing a uint8_t this is unlikely to crash due to alignment but should be improved
		SSUHeader * header = (SSUHeader *)buf;
		uint8_t * encrypted = &header->flag;
		uint16_t encryptedLen = len - (encrypted - buf);
		i2p::crypto::CBCDecryption decryption;
		decryption.SetKey (aesKey);
		decryption.SetIV (header->iv);
		decryption.Decrypt (encrypted, encryptedLen, encrypted);
	}

	void SSUSession::DecryptSessionKey (uint8_t * buf, size_t len)
	{
		if (len < sizeof (SSUHeader))
		{
			LogPrint (eLogError, "Unexpected SSU packet length ", len);
			return;
		}
		//TODO: since we are accessing a uint8_t this is unlikely to crash due to alignment but should be improved
		SSUHeader * header = (SSUHeader *)buf;
		uint8_t * encrypted = &header->flag;
		uint16_t encryptedLen = len - (encrypted - buf);
		if (encryptedLen > 0)
		{
			m_SessionKeyDecryption.SetIV (header->iv);
			m_SessionKeyDecryption.Decrypt (encrypted, encryptedLen, encrypted);
		}
	}

	bool SSUSession::Validate (uint8_t * buf, size_t len, const uint8_t * macKey)
	{
		if (len < sizeof (SSUHeader))
		{
			LogPrint (eLogError, "Unexpected SSU packet length ", len);
			return false;
		}
		//TODO: since we are accessing a uint8_t this is unlikely to crash due to alignment but should be improved
		SSUHeader * header = (SSUHeader *)buf;
		uint8_t * encrypted = &header->flag;
		uint16_t encryptedLen = len - (encrypted - buf);
		// assume actual buffer size is 18 (16 + 2) bytes more
		memcpy (buf + len, header->iv, 16);
		htobe16buf (buf + len + 16, encryptedLen);
		uint8_t digest[16];
		i2p::crypto::HMACMD5Digest (encrypted, encryptedLen + 18, macKey, digest);
		return !memcmp (header->mac, digest, 16);
	}

	void SSUSession::Connect ()
	{
		if (m_State == eSessionStateUnknown)
		{
			// set connect timer
			ScheduleConnectTimer ();
			m_DHKeysPair = transports.GetNextDHKeysPair ();
			SendSessionRequest ();
		}
	}

	void SSUSession::WaitForConnect ()
	{
		if (!m_RemoteRouter) // incoming session
			ScheduleConnectTimer ();
		else
			LogPrint (eLogError, "SSU wait for connect for outgoing session");
	}

	void SSUSession::ScheduleConnectTimer ()
	{
		m_Timer.cancel ();
		m_Timer.expires_from_now (boost::posix_time::seconds(SSU_CONNECT_TIMEOUT));
		m_Timer.async_wait (std::bind (&SSUSession::HandleConnectTimer,
			shared_from_this (), std::placeholders::_1));
}

	void SSUSession::HandleConnectTimer (const boost::system::error_code& ecode)
	{
		if (!ecode)
		{
			// timeout expired
			LogPrint ("SSU session was not established after ", SSU_CONNECT_TIMEOUT, " second");
			Failed ();
		}
	}

	void SSUSession::Introduce (uint32_t iTag, const uint8_t * iKey)
	{
		if (m_State == eSessionStateUnknown)
		{
			// set connect timer
			m_Timer.expires_from_now (boost::posix_time::seconds(SSU_CONNECT_TIMEOUT));
			m_Timer.async_wait (std::bind (&SSUSession::HandleConnectTimer,
				shared_from_this (), std::placeholders::_1));
		}
		SendRelayRequest (iTag, iKey);
	}

	void SSUSession::WaitForIntroduction ()
	{
		m_State = eSessionStateIntroduced;
		// set connect timer
		m_Timer.expires_from_now (boost::posix_time::seconds(SSU_CONNECT_TIMEOUT));
		m_Timer.async_wait (std::bind (&SSUSession::HandleConnectTimer,
			shared_from_this (), std::placeholders::_1));
	}

	void SSUSession::Close ()
	{
		m_State = eSessionStateClosed;
		SendSesionDestroyed ();
		transports.PeerDisconnected (shared_from_this ());
		m_Data.Stop ();
		m_Timer.cancel ();
	}

	void SSUSession::Done ()
	{
		GetService ().post (std::bind (&SSUSession::Failed, shared_from_this ()));
	}

	void SSUSession::Established ()
	{
		m_State = eSessionStateEstablished;
		if (m_DHKeysPair)
		{
			delete m_DHKeysPair;
			m_DHKeysPair = nullptr;
		}
		m_Data.Start ();
		m_Data.Send (ToSharedI2NPMessage(CreateDatabaseStoreMsg ()));
		transports.PeerConnected (shared_from_this ());
		if (m_PeerTest && (m_RemoteRouter && m_RemoteRouter->IsPeerTesting ()))
			SendPeerTest ();
		ScheduleTermination ();
	}

	void SSUSession::Failed ()
	{
		if (m_State != eSessionStateFailed)
		{
			m_State = eSessionStateFailed;
			m_Server.DeleteSession (shared_from_this ());
		}
	}

	void SSUSession::ScheduleTermination ()
	{
		m_Timer.cancel ();
		m_Timer.expires_from_now (boost::posix_time::seconds(SSU_TERMINATION_TIMEOUT));
		m_Timer.async_wait (std::bind (&SSUSession::HandleTerminationTimer,
			shared_from_this (), std::placeholders::_1));
	}

	void SSUSession::HandleTerminationTimer (const boost::system::error_code& ecode)
	{
		if (ecode != boost::asio::error::operation_aborted)
		{
			LogPrint ("SSU no activity fo ", SSU_TERMINATION_TIMEOUT, " seconds");
			Failed ();
		}
	}

	const uint8_t * SSUSession::GetIntroKey () const
	{
		if (m_RemoteRouter)
		{
			// we are client
			auto address = m_RemoteRouter->GetSSUAddress ();
			return address ? (const uint8_t *)address->key : nullptr;
		}
		else
		{
			// we are server
			auto address = i2p::context.GetRouterInfo ().GetSSUAddress ();
			return address ? (const uint8_t *)address->key : nullptr;
		}
	}

	void SSUSession::SendI2NPMessages (const std::vector<std::shared_ptr<I2NPMessage> >& msgs)
	{
		GetService ().post (std::bind (&SSUSession::PostI2NPMessages, shared_from_this (), msgs));
	}

	void SSUSession::PostI2NPMessages (std::vector<std::shared_ptr<I2NPMessage> > msgs)
	{
		if (m_State == eSessionStateEstablished)
		{
			for (auto it: msgs)
				if (it) m_Data.Send (it);
		}
	}

	void SSUSession::ProcessData (uint8_t * buf, size_t len)
	{
		m_Data.ProcessMessage (buf, len);
		m_IsDataReceived = true;
	}

	void SSUSession::FlushData ()
	{
		if (m_IsDataReceived)
		{
			m_Data.FlushReceivedMessage ();
			m_IsDataReceived = false;
		}
	}

	void SSUSession::ProcessPeerTest (const uint8_t * buf, size_t len, const boost::asio::ip::udp::endpoint& senderEndpoint)
	{
		uint32_t nonce = bufbe32toh (buf); // 4 bytes
		uint8_t size = buf[4];	// 1 byte
		uint32_t address = (size == 4) ? buf32toh(buf + 5) : 0; // big endian, size bytes
		uint16_t port = buf16toh(buf + size + 5); // big endian, 2 bytes
		const uint8_t * introKey = buf + size + 7;
		if (port && !address)
		{
			LogPrint (eLogWarning, "Address of ", size, " bytes not supported");
			return;
		}
		switch (m_Server.GetPeerTestParticipant (nonce))
		{
			// existing test
			case ePeerTestParticipantAlice1:
			{
				if (m_State == eSessionStateEstablished)
				{
					LogPrint (eLogDebug, "SSU peer test from Bob. We are Alice");
					if (i2p::context.GetStatus () == eRouterStatusTesting) // still not OK
						i2p::context.SetStatus (eRouterStatusFirewalled);
				}
				else
				{
					LogPrint (eLogDebug, "SSU first peer test from Charlie. We are Alice");
					i2p::context.SetStatus (eRouterStatusOK);
					m_Server.UpdatePeerTest (nonce, ePeerTestParticipantAlice2);
					SendPeerTest (nonce, senderEndpoint.address ().to_v4 ().to_ulong (),
						senderEndpoint.port (), introKey, true, false); // to Charlie
				}
				break;
			}
			case ePeerTestParticipantAlice2:
			{
				if (m_State == eSessionStateEstablished)
					LogPrint (eLogDebug, "SSU peer test from Bob. We are Alice");
				else
				{
					// peer test successive
					LogPrint (eLogDebug, "SSU second peer test from Charlie. We are Alice");
					i2p::context.SetStatus (eRouterStatusOK);
					m_Server.RemovePeerTest (nonce);
				}
				break;
			}
			case ePeerTestParticipantBob:
			{
				LogPrint (eLogDebug, "SSU peer test from Charlie. We are Bob");
				auto session = m_Server.GetPeerTestSession (nonce); // session with Alice from PeerTest
				if (session && session->m_State == eSessionStateEstablished)
					session->Send (PAYLOAD_TYPE_PEER_TEST, buf, len); // back to Alice
				m_Server.RemovePeerTest (nonce); // nonce has been used
				break;
			}
			case ePeerTestParticipantCharlie:
			{
				LogPrint (eLogDebug, "SSU peer test from Alice. We are Charlie");
				SendPeerTest (nonce, senderEndpoint.address ().to_v4 ().to_ulong (),
					senderEndpoint.port (), introKey); // to Alice with her actual address
				m_Server.RemovePeerTest (nonce); // nonce has been used
				break;
			}
			// test not found
			case ePeerTestParticipantUnknown:
			{
				if (m_State == eSessionStateEstablished)
				{
					// new test
					if (port)
					{
						LogPrint (eLogDebug, "SSU peer test from Bob. We are Charlie");
						m_Server.NewPeerTest (nonce, ePeerTestParticipantCharlie);
						Send (PAYLOAD_TYPE_PEER_TEST, buf, len); // back to Bob
						SendPeerTest (nonce, be32toh (address), be16toh (port), introKey); // to Alice with her address received from Bob
					}
					else
					{
						LogPrint (eLogDebug, "SSU peer test from Alice. We are Bob");
						auto session = m_Server.GetRandomEstablishedSession (shared_from_this ()); // Charlie
						if (session)
						{
							m_Server.NewPeerTest (nonce, ePeerTestParticipantBob, shared_from_this ());
							session->SendPeerTest (nonce, senderEndpoint.address ().to_v4 ().to_ulong (),
								senderEndpoint.port (), introKey, false); // to Charlie with Alice's actual address
						}
					}
				}
				else
					LogPrint (eLogError, "SSU unexpected peer test");
			}
		}
	}

	void SSUSession::SendPeerTest (uint32_t nonce, uint32_t address, uint16_t port,
		const uint8_t * introKey, bool toAddress, bool sendAddress)
	// toAddress is true for Alice<->Chalie communications only
	// sendAddress is false if message comes from Alice
	{
		uint8_t buf[80 + 18];
		uint8_t iv[16];
		uint8_t * payload = buf + sizeof (SSUHeader);
		htobe32buf (payload, nonce);
		payload += 4; // nonce
		// address and port
		if (sendAddress && address)
		{
			*payload = 4;
			payload++; // size
			htobe32buf (payload, address);
			payload += 4; // address
		}
		else
		{
			*payload = 0;
			payload++; //size
		}
		htobe16buf (payload, port);
		payload += 2; // port
		// intro key
		if (toAddress)
		{
			// send our intro key to address instead it's own
			auto addr = i2p::context.GetRouterInfo ().GetSSUAddress ();
			if (addr)
				memcpy (payload, addr->key, 32); // intro key
			else
				LogPrint (eLogError, "SSU is not supported. Can't send peer test");
		}
		else
			memcpy (payload, introKey, 32); // intro key

		// send
		CryptoPP::RandomNumberGenerator& rnd = i2p::context.GetRandomNumberGenerator ();
		rnd.GenerateBlock (iv, 16); // random iv
		if (toAddress)
		{
			// encrypt message with specified intro key
			FillHeaderAndEncrypt (PAYLOAD_TYPE_PEER_TEST, buf, 80, introKey, iv, introKey);
			boost::asio::ip::udp::endpoint e (boost::asio::ip::address_v4 (address), port);
			m_Server.Send (buf, 80, e);
		}
		else
		{
			// encrypt message with session key
			FillHeaderAndEncrypt (PAYLOAD_TYPE_PEER_TEST, buf, 80);
			Send (buf, 80);
		}
	}

	void SSUSession::SendPeerTest ()
	{
		// we are Alice
		LogPrint (eLogDebug, "SSU sending peer test");
		auto address = i2p::context.GetRouterInfo ().GetSSUAddress ();
		if (!address)
		{
			LogPrint (eLogError, "SSU is not supported. Can't send peer test");
			return;
		}
		uint32_t nonce = i2p::context.GetRandomNumberGenerator ().GenerateWord32 ();
		if (!nonce) nonce = 1;
		m_PeerTest = false;
		m_Server.NewPeerTest (nonce, ePeerTestParticipantAlice1);
		SendPeerTest (nonce, 0, 0, address->key, false, false); // address and port always zero for Alice
	}

	void SSUSession::SendKeepAlive ()
	{
		if (m_State == eSessionStateEstablished)
		{
			uint8_t buf[48 + 18];
			uint8_t	* payload = buf + sizeof (SSUHeader);
			*payload = 0; // flags
			payload++;
			*payload = 0; // num fragments
			// encrypt message with session key
			FillHeaderAndEncrypt (PAYLOAD_TYPE_DATA, buf, 48);
			Send (buf, 48);
			LogPrint (eLogDebug, "SSU keep-alive sent");
			ScheduleTermination ();
		}
	}

	void SSUSession::SendSesionDestroyed ()
	{
		if (m_IsSessionKey)
		{
			uint8_t buf[48 + 18];
			// encrypt message with session key
			FillHeaderAndEncrypt (PAYLOAD_TYPE_SESSION_DESTROYED, buf, 48);
			try
			{
				Send (buf, 48);
			}
			catch (std::exception& ex)
			{
				LogPrint (eLogError, "SSU send session destoriyed exception ", ex.what ());
			}
			LogPrint (eLogDebug, "SSU session destroyed sent");
		}
	}

	void SSUSession::Send (uint8_t type, const uint8_t * payload, size_t len)
	{
		uint8_t buf[SSU_MTU_V4 + 18];
		size_t msgSize = len + sizeof (SSUHeader);
		size_t paddingSize = msgSize & 0x0F; // %16
		if (paddingSize > 0) msgSize += (16 - paddingSize);
		if (msgSize > SSU_MTU_V4)
		{
			LogPrint (eLogWarning, "SSU payload size ", msgSize, " exceeds MTU");
			return;
		}
		memcpy (buf + sizeof (SSUHeader), payload, len);
		// encrypt message with session key
		FillHeaderAndEncrypt (type, buf, msgSize);
		Send (buf, msgSize);
	}

	void SSUSession::Send (const uint8_t * buf, size_t size)
	{
		m_NumSentBytes += size;
		i2p::transport::transports.UpdateSentBytes (size);
		m_Server.Send (buf, size, m_RemoteEndpoint);
	}
}
}