text:
// Copyright (c) YugaByte, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
// in compliance with the License.  You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software distributed under the License
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
// or implied.  See the License for the specific language governing permissions and limitations
// under the License.
//

#include "yb/rpc/compressed_stream.h"

#include <lz4.h>
#include <snappy-sinksource.h>
#include <snappy.h>
#include <zlib.h>

#include <boost/preprocessor/cat.hpp>
#include <boost/range/iterator_range.hpp>

#include "yb/gutil/casts.h"

#include "yb/rpc/circular_read_buffer.h"
#include "yb/rpc/outbound_data.h"
#include "yb/rpc/refined_stream.h"

#include "yb/util/logging.h"
#include "yb/util/result.h"
#include "yb/util/size_literals.h"
#include "yb/util/status_format.h"

using namespace std::literals;

DEFINE_int32(stream_compression_algo, 0, "Algorithm used for stream compression. "
                                         "0 - no compression, 1 - gzip, 2 - snappy, 3 - lz4.");

namespace yb {
namespace rpc {

using SmallRefCntBuffers = boost::container::small_vector_base<RefCntBuffer>;

namespace {

class Compressor {
 public:
  virtual std::string ToString() const = 0;

  // Initialize compressor, required since we don't use exceptions to return error from ctor.
  virtual Status Init() = 0;

  // Compress specified vector of input buffers into single output buffer.
  virtual Status Compress(
      const SmallRefCntBuffers& input, RefinedStream* stream, OutboundDataPtr data) = 0;

  // Decompress specified input slice to specified output buffer.
  virtual Result<ReadBufferFull> Decompress(StreamReadBuffer* inp, StreamReadBuffer* out) = 0;

  // Connection header associated with this compressor.
  virtual OutboundDataPtr ConnectionHeader() = 0;

  virtual ~Compressor() = default;
};

size_t EntrySize(const RefCntBuffer& buffer) {
  return buffer.size();
}

size_t EntrySize(const iovec& iov) {
  return iov.iov_len;
}

char* EntryData(const RefCntBuffer& buffer) {
  return buffer.data();
}

char* EntryData(const iovec& iov) {
  return static_cast<char*>(iov.iov_base);
}

template <class Collection>
size_t TotalLen(const Collection& input) {
  size_t result = 0;
  for (const auto& buf : input) {
    result += EntrySize(buf);
  }
  return result;
}

template <class Compressor>
OutboundDataPtr GetConnectionHeader() {
  // Compressed stream header has signature YBx, where x - compressor identifier.
  static auto result = std::make_shared<StringOutboundData>(
      "YB"s + Compressor::kId, Compressor::kId + "ConnectionHeader"s);
  return result;
}

template <class SliceDecompressor>
Result<ReadBufferFull> DecompressBySlices(
    StreamReadBuffer* inp, StreamReadBuffer* out, const SliceDecompressor& slice_decompressor) {
  size_t consumed = 0;
  auto out_vecs = VERIFY_RESULT(out->PrepareAppend());
  auto out_it = out_vecs.begin();
  size_t appended = 0;

  for (const auto& iov : inp->AppendedVecs()) {
    Slice slice(static_cast<char*>(iov.iov_base), iov.iov_len);
    for (;;) {
      if (out_it->iov_len == 0) {
        if (++out_it == out_vecs.end()) {
          break;
        }
      }
      size_t len = VERIFY_RESULT(slice_decompressor(&slice, out_it->iov_base, out_it->iov_len));
      appended += len;
      IoVecRemovePrefix(len, &*out_it);
      if (slice.empty()) {
        break;
      }
    }
    consumed += iov.iov_len - slice.size();
    if (!slice.empty()) {
      break;
    }
  }
  out->DataAppended(appended);
  inp->Consume(consumed, Slice());
  return ReadBufferFull(out->Full());
}

class ZlibCompressor : public Compressor {
 public:
  static const char kId = 'G';
  static const int kIndex = 1;

  explicit ZlibCompressor(MemTrackerPtr mem_tracker) {
  }

  ~ZlibCompressor() {
    if (deflate_inited_) {
      int res = deflateEnd(&deflate_stream_);
      LOG_IF(WARNING, res != Z_OK && res != Z_DATA_ERROR)
          << "Failed to destroy deflate stream: " << res;
    }
    if (inflate_inited_) {
      int res = inflateEnd(&inflate_stream_);
      LOG_IF(WARNING, res != Z_OK) << "Failed to destroy inflate stream: " << res;
    }
  }

  OutboundDataPtr ConnectionHeader() override {
    return GetConnectionHeader<ZlibCompressor>();
  }

  Status Init() override {
    memset(&deflate_stream_, 0, sizeof(deflate_stream_));
    int res = deflateInit(&deflate_stream_, /* level= */ Z_DEFAULT_COMPRESSION);
    if (res != Z_OK) {
      return STATUS_FORMAT(RuntimeError, "Cannot init deflate stream: $0", res);
    }
    deflate_inited_ = true;

    memset(&inflate_stream_, 0, sizeof(inflate_stream_));
    res = inflateInit(&inflate_stream_);
    if (res != Z_OK) {
      return STATUS_FORMAT(RuntimeError, "Cannot init inflate stream: $0", res);
    }
    inflate_inited_ = true;

    return Status::OK();
  }

  std::string ToString() const override {
    return "Zlib";
  }

  Status Compress(
      const SmallRefCntBuffers& input, RefinedStream* stream, OutboundDataPtr data) override {
    RefCntBuffer output(deflateBound(&deflate_stream_, TotalLen(input)));
    deflate_stream_.avail_out = static_cast<unsigned int>(output.size());
    deflate_stream_.next_out = output.udata();

    for (auto it = input.begin(); it != input.end();) {
      const auto& buf = *it++;
      deflate_stream_.next_in = const_cast<Bytef*>(buf.udata());
      deflate_stream_.avail_in = static_cast<unsigned int>(buf.size());

      for (;;) {
        auto res = deflate(&deflate_stream_, it == input.end() ? Z_PARTIAL_FLUSH : Z_NO_FLUSH);
        if (res == Z_STREAM_END) {
          if (deflate_stream_.avail_in != 0) {
            return STATUS_FORMAT(
                RuntimeError, "Stream end when input data still available: $0",
                deflate_stream_.avail_in);
          }
          break;
        }
        if (res != Z_OK) {
          return STATUS_FORMAT(RuntimeError, "Compression failed: $0", res);
        }
        if (deflate_stream_.avail_in == 0) {
          break;
        }
      }
    }

    output.Shrink(deflate_stream_.next_out - output.udata());

    // Send compressed data to underlying stream.
    return stream->SendToLower(std::make_shared<SingleBufferOutboundData>(
        std::move(output), std::move(data)));
  }

  Result<ReadBufferFull> Decompress(StreamReadBuffer* inp, StreamReadBuffer* out) override {
    return DecompressBySlices(
        inp, out, [this](Slice* input, void* out, size_t outlen) -> Result<size_t> {
      inflate_stream_.next_in = const_cast<Bytef*>(pointer_cast<const Bytef*>(input->data()));
      inflate_stream_.avail_in = narrow_cast<uInt>(input->size());
      inflate_stream_.next_out = static_cast<Bytef*>(out);
      inflate_stream_.avail_out = narrow_cast<uInt>(outlen);

      int res = inflate(&inflate_stream_, Z_NO_FLUSH);
      if (res != Z_OK && res != Z_BUF_ERROR) {
        return STATUS_FORMAT(RuntimeError, "Decompression failed: $0", res);
      }

      input->remove_prefix(input->size() - inflate_stream_.avail_in);
      return outlen - inflate_stream_.avail_out;
    });
  }

 private:
  z_stream deflate_stream_;
  z_stream inflate_stream_;
  bool deflate_inited_ = false;
  bool inflate_inited_ = false;
};

// Source implementation that provides input from range of buffers.
template <class It>
class RangeSource : public snappy::Source {
 public:
  explicit RangeSource(const It& begin, const It& end)
      : current_(begin), end_(end),
        available_(TotalLen(boost::make_iterator_range(begin, end))) {}

  void Limit(size_t value) {
    available_ = value;
  }

  size_t Available() const override {
    return available_;
  }

  const char* Peek(size_t* len) override {
    if (available_ == 0) {
      *len = 0;
      return nullptr;
    }
    *len = std::min(EntrySize(*current_) - current_pos_, available_);
    return EntryData(*current_) + current_pos_;
  }

  void Rewind(size_t n) {
    available_ += n;
    for (;;) {
      if (current_pos_ >= n) {
        current_pos_ -= n;
        break;
      }
      n -= current_pos_;
      --current_;
      current_pos_ = EntrySize(*current_);
    }
  }

  void Skip(size_t n) override {
    if (!n) {
      return;
    }
    available_ -= n;
    if ((current_pos_ += n) >= EntrySize(*current_)) {
      current_pos_ = 0;
      ++current_;
      DCHECK(available_ == 0 ? current_ == end_ : current_ < end_)
          << "Available: " << available_ << ", left buffers: " << std::distance(current_, end_)
          << ", n: " << n;
    }
  }

 private:
  It current_; // Current buffer that we are reading.
  It end_;
  size_t current_pos_ = 0; // Position in current buffer.
  size_t available_; // How many bytes left in all buffers.
};

// Sink implementation that provides output to io vecs.
class IoVecsSink : public snappy::Sink {
 public:
  explicit IoVecsSink(IoVecs* out) : out_(out->begin()) {}

  size_t total_appended() const {
    return total_appended_;
  }

  void Append(const char* bytes, size_t n) override {
    total_appended_ += n;

    if (bytes == out_->iov_base) {
      if (out_->iov_len == n) {
        ++out_;
      } else {
        IoVecRemovePrefix(n, &*out_);
      }
      return;
    }

    while (n > 0) {
      size_t current_len = out_->iov_len;
      if (current_len >= n) {
        memcpy(out_->iov_base, bytes, n);
        IoVecRemovePrefix(n, &*out_);
        n = 0;
      } else {
        memcpy(out_->iov_base, bytes, current_len);
        bytes += current_len;
        n -= current_len;
        ++out_;
      }
    }
  }

  char* GetAppendBufferVariable(
      size_t min_size, size_t desired_size_hint, char* scratch,
      size_t scratch_size, size_t* allocated_size) override {
    if (min_size <= out_->iov_len) {
      *allocated_size = out_->iov_len;
      return static_cast<char*>(out_->iov_base);
    }
    return Sink::GetAppendBufferVariable(
        min_size, desired_size_hint, scratch, scratch_size, allocated_size);
  }

 private:
  IoVecs::iterator out_;
  size_t total_appended_ = 0;
};

// Binary search to find max value so that max_compressed_len(value) fits into header_len bytes.
template <class F>
static size_t FindMaxChunkSize(size_t header_len, const F& max_compressed_len) {
  size_t max_value = (1ULL << (8 * header_len)) - 1;
  size_t l = 1;
  size_t r = max_value;
  while (r > l) {
    size_t m = (l + r + 1) / 2;
    if (implicit_cast<size_t>(max_compressed_len(narrow_cast<int>(m))) > max_value) {
      r = m - 1;
    } else {
      l = m;
    }
  }
  return l;
}

// Snappy does not support stream compression.
// So we have to add block len, to start decompression only from block start.
// We use 2 bytes for block len, so have to limit uncompressed block so that in worst case
// compressed block size would fit into 2 bytes.
constexpr size_t kSnappyHeaderLen = 2;
const size_t kSnappyMaxChunkSize = FindMaxChunkSize(kSnappyHeaderLen, &snappy::MaxCompressedLength);

class SnappyCompressor : public Compressor {
 public:
  static constexpr char kId = 'S';
  static constexpr int kIndex = 2;
  static constexpr size_t kHeaderLen = kSnappyHeaderLen;

  explicit SnappyCompressor(MemTrackerPtr mem_tracker) {
  }

  ~SnappyCompressor() {
  }

  OutboundDataPtr ConnectionHeader() override {
    return GetConnectionHeader<SnappyCompressor>();
  }

  Status Init() override {
    return Status::OK();
  }

  std::string ToString() const override {
    return "Snappy";
  }

  Status Compress(
      const SmallRefCntBuffers& input, RefinedStream* stream, OutboundDataPtr data) override {
    RangeSource<SmallRefCntBuffers::const_iterator> source(input.begin(), input.end());
    auto input_size = source.Available();
    bool stop = false;
    while (!stop) {
      // Split input into chunks of size kSnappyMaxChunkSize or less.
      if (input_size > kSnappyMaxChunkSize) {
        source.Limit(kSnappyMaxChunkSize);
        input_size -= kSnappyMaxChunkSize;
      } else {
        source.Limit(input_size);
        stop = true;
      }
      RefCntBuffer output(kHeaderLen + snappy::MaxCompressedLength(source.Available()));
      snappy::UncheckedByteArraySink sink(output.data() + kHeaderLen);
      auto compressed_len = snappy::Compress(&source, &sink);
      BigEndian::Store16(output.data(), compressed_len);
      output.Shrink(kHeaderLen + compressed_len);
      RETURN_NOT_OK(stream->SendToLower(std::make_shared<SingleBufferOutboundData>(
          std::move(output),
          // We processed last buffer, attach data to it, so it will be notified when this buffer
          // is transferred.
          stop ? std::move(data) : nullptr)));
    }
    return Status::OK();
  }

  Result<ReadBufferFull> Decompress(StreamReadBuffer* inp, StreamReadBuffer* out) override {
    auto inp_vecs = inp->AppendedVecs();
    RangeSource<IoVecs::const_iterator> source(inp_vecs.begin(), inp_vecs.end());

    auto total_consumed = 0;
    auto out_vecs = VERIFY_RESULT(out->PrepareAppend());
    auto total_output = IoVecsFullSize(out_vecs);
    IoVecsSink sink(&out_vecs);

    size_t input_size = source.Available();
    auto result = ReadBufferFull::kFalse;
    while (total_consumed + kHeaderLen <= input_size) {
      source.Limit(input_size - total_consumed);
      // Fetch chunk len.
      size_t len = 0;
      auto data = source.Peek(&len);
      size_t size;
      if (len >= kHeaderLen) {
        // Size fully contained in the first buffer.
        size = BigEndian::Load16(data);
        source.Skip(kHeaderLen);
      } else {
        // Size distributed between 2 blocks.
        // Since blocks are not empty and size exactly 2, we should expect that first block
        // contains 1 byte of size.
        if (len != 1) {
          return STATUS(RuntimeError, "Expected to peek at least one byte");
        }
        char buf[kHeaderLen];
        buf[0] = *data;
        source.Skip(1);
        data = source.Peek(&len);
        buf[1] = *data;
        source.Skip(1);
        size = BigEndian::Load16(buf);
      }
      // Check whether we already received full chunk.
      VLOG_WITH_FUNC(4)
          << "Total consumed: " << total_consumed << ", size: " << size << ", input: "
          << input_size;
      if (total_consumed + kHeaderLen + size > input_size) {
        break;
      }
      source.Limit(size);
      uint32_t length = 0;
      auto old_available = source.Available();
      if (!snappy::GetUncompressedLength(&source, &length)) {
        return STATUS(RuntimeError, "GetUncompressedLength failed");
      }

      // Check if we have space for decompressed block.
      VLOG_WITH_FUNC(4)
          << "Total appended: " << sink.total_appended() << ", length: " << length
          << ", total_output: " << total_output;
      if (sink.total_appended() + length > total_output) {
        result = ReadBufferFull::kTrue;
        break;
      }
      // Rollback to state before GetUncompressedLength.
      source.Rewind(old_available - source.Available());
      if (!snappy::Uncompress(&source, &sink)) {
        return STATUS(RuntimeError, "Decompression failed");
      }
      total_consumed += kHeaderLen + size;
    }

    out->DataAppended(sink.total_appended());
    inp->Consume(total_consumed, Slice());

    return result;
  }
};

// LZ4 could compress/decompress only continuous block of memory into similar block.
// So we do that same as for Snappy, but decompression is even more complex.
constexpr size_t kLZ4HeaderLen = 2;
const size_t kLZ4MaxChunkSize = FindMaxChunkSize(kLZ4HeaderLen, &LZ4_compressBound);
const size_t kLZ4BufferSize = 64_KB;

class LZ4DecompressState {
 public:
  LZ4DecompressState(char* input_buffer, char* output_buffer, Slice* prev_decompress_data_left)
      : input_buffer_(input_buffer), output_buffer_(output_buffer),
        prev_decompress_data_left_(prev_decompress_data_left) {}

  Result<ReadBufferFull> Execute(StreamReadBuffer* inp, StreamReadBuffer* out) {
    outvecs_ = VERIFY_RESULT(out->PrepareAppend());
    out_it_ = outvecs_.begin();

    // Check if we previously decompressed some data that did not fit into output buffer.
    // So copy it now. See DecompressChunk for details.
    while (!prev_decompress_data_left_->empty()) {
      auto len = std::min(prev_decompress_data_left_->size(), out_it_->iov_len);
      memcpy(out_it_->iov_base, prev_decompress_data_left_->data(), len);
      total_add_ += len;
      IoVecRemovePrefix(len, &*out_it_);
      prev_decompress_data_left_->remove_prefix(len);
      if (out_it_->iov_len == 0) {
        if (++out_it_ == outvecs_.end()) {
          out->DataAppended(total_add_);
          return ReadBufferFull(out->Full());
        }
      }
    }

    // Remaining piece of data in the previous vec.
    Slice prev_input_slice;
    for (const auto& input_vec : inp->AppendedVecs()) {
      Slice input_slice(static_cast<char*>(input_vec.iov_base), input_vec.iov_len);
      if (!prev_input_slice.empty()) {
        size_t chunk_size;
        if (prev_input_slice.size() >= kLZ4HeaderLen) {
          // Size fully contained in the previous block.
          chunk_size = BigEndian::Load16(prev_input_slice.data());
        } else if (prev_input_slice.size() + input_slice.size() < kLZ4HeaderLen) {
          // Did not receive header yet. So exit and wait for more data received.
          break;
        } else {
          // Size distributed between 2 blocks.
          char buf[kLZ4HeaderLen];
          memcpy(buf, prev_input_slice.data(), prev_input_slice.size());
          memcpy(buf + prev_input_slice.size(), input_slice.data(),
                 kLZ4HeaderLen - prev_input_slice.size());
          chunk_size = BigEndian::Load16(buf);
        }
        if (kLZ4HeaderLen + chunk_size > prev_input_slice.size() + input_slice.size()) {
          // Did not receive full block yet. So exit and wait for more data received.
          // TODO Here we rely on the fact that we use circular buffer and could have at most 2
          //      blocks. In general case chunk could be distributed across several buffers.
          break;
        }
        if (prev_input_slice.size() > kLZ4HeaderLen) {
          // Data is distributed between 2 buffers.
          // Have to copy it into separate input buffer so it will be a single continuous block.
          size_t size_in_prev_slice = prev_input_slice.size() - kLZ4HeaderLen;
          size_t size_in_current_slice = chunk_size - size_in_prev_slice;
          memcpy(input_buffer_, prev_input_slice.data() + kLZ4HeaderLen, size_in_prev_slice);
          memcpy(input_buffer_ + size_in_prev_slice, input_slice.data(), size_in_current_slice);
          RETURN_NOT_OK(DecompressChunk(Slice(input_buffer_, chunk_size)));
          input_slice.remove_prefix(size_in_current_slice);
        } else {
          // Only header (or part of header) was in previous buffer.
          // Could decompress from current buffer.
          input_slice.remove_prefix(kLZ4HeaderLen - prev_input_slice.size());
          RETURN_NOT_OK(DecompressChunk(input_slice.Prefix(chunk_size)));
          input_slice.remove_prefix(chunk_size);
        }
      }
      // Decompress all chunks contained in current buffer.
      while (input_slice.size() >= kLZ4HeaderLen && out_it_ != outvecs_.end()) {
        size_t chunk_size = BigEndian::Load16(input_slice.data());
        if (input_slice.size() < kLZ4HeaderLen + chunk_size) {
          break;
        }
        input_slice.remove_prefix(kLZ4HeaderLen);
        RETURN_NOT_OK(DecompressChunk(input_slice.Prefix(chunk_size)));
        input_slice.remove_prefix(chunk_size);
      }
      prev_input_slice = input_slice;
      // DecompressChunk could increase out_it_, so check whether we still have output space.
      if (out_it_ == outvecs_.end()) {
        break;
      }
    }
    inp->Consume(total_consumed_, Slice());
    out->DataAppended(total_add_);
    return ReadBufferFull(out->Full());
  }

 private:
  Status DecompressChunk(const Slice& input) {
    int res = LZ4_decompress_safe(
        input.cdata(), static_cast<char*>(out_it_->iov_base), narrow_cast<int>(input.size()),
        narrow_cast<int>(out_it_->iov_len));
    if (res <= 0) {
      // Unfortunately LZ4 does not provide information whether decryption failed because
      // of wrong data or it just does not fit into output buffer.
      // Try to decode to buffer that is big enough for max possible decompressed chunk.
      res = LZ4_decompress_safe(
          input.cdata(), output_buffer_, narrow_cast<int>(input.size()), kLZ4BufferSize);
      if (res <= 0) {
        return STATUS_FORMAT(RuntimeError, "Decompress failed: $0", res);
      }

      // Copy data from output buffer to provided read buffer.
      size_t size = res;
      char* buf = output_buffer_;
      while (out_it_ != outvecs_.end()) {
        size_t len = std::min<size_t>(size, out_it_->iov_len);
        memcpy(out_it_->iov_base, buf, len);
        IoVecRemovePrefix(len, &*out_it_);
        size -= len;
        total_add_ += len;
        buf += len;
        if (out_it_->iov_len == 0) {
          // Need to fill next output io vec.
          ++out_it_;
        }
        if (size == 0) {
          // Fully copied all decompressed data.
          break;
        }
      }
      if (size != 0) {
        // Have more decompressed data than provided read buffer could accept.
        *prev_decompress_data_left_ = Slice(buf, size);
      }
    } else {
      IoVecRemovePrefix(res, &*out_it_);
      total_add_ += res;
      if (out_it_->iov_len == 0) {
        ++out_it_;
      }
    }

    // Header was consumed by the caller, so we also add it here.
    total_consumed_ += kLZ4HeaderLen + input.size();
    return Status::OK();
  }

  // LZ4 operates on continuous chunks of memory, so we use input and output buffers to combine
  // several iovecs into one continuous chunk when necessary.
  char* input_buffer_;
  char* output_buffer_;
  Slice* prev_decompress_data_left_;

  IoVecs outvecs_;
  IoVecs::iterator out_it_;
  size_t total_add_ = 0;
  size_t total_consumed_ = 0;
};

class LZ4Compressor : public Compressor {
 public:
  static constexpr char kId = 'L';
  static constexpr int kIndex = 3;
  static constexpr size_t kHeaderLen = kLZ4HeaderLen;

  explicit LZ4Compressor(MemTrackerPtr mem_tracker) {
    if (mem_tracker) {
      consumption_ = ScopedTrackedConsumption(std::move(mem_tracker), 2 * kLZ4BufferSize);
    }
  }

  OutboundDataPtr ConnectionHeader() override {
    return GetConnectionHeader<LZ4Compressor>();
  }

  Status Init() override {
    return Status::OK();
  }

  std::string ToString() const override {
    return "LZ4";
  }

  Status Compress(
      const SmallRefCntBuffers& input, RefinedStream* stream, OutboundDataPtr data) override {
    // Increment iterator in loop body to be able to check whether it is last iteration or not.
    for (auto input_it = input.begin(); input_it != input.end();) {
      Slice input_slice = input_it->AsSlice();
      ++input_it;
      while (!input_slice.empty()) {
        Slice chunk;
        // Split input into chunks of size kLZ4MaxChunkSize or less.
        if (input_slice.size() > kLZ4MaxChunkSize) {
          chunk = input_slice.Prefix(kLZ4MaxChunkSize);
        } else {
          chunk = input_slice;
        }
        input_slice.remove_prefix(chunk.size());
        RefCntBuffer output(kHeaderLen + LZ4_compressBound(narrow_cast<int>(chunk.size())));
        int res = LZ4_compress(
            chunk.cdata(), output.data() + kHeaderLen, narrow_cast<int>(chunk.size()));
        if (res <= 0) {
          return STATUS_FORMAT(RuntimeError, "LZ4 compression failed: $0", res);
        }
        BigEndian::Store16(output.data(), res);
        output.Shrink(kHeaderLen + res);
        RETURN_NOT_OK(stream->SendToLower(std::make_shared<SingleBufferOutboundData>(
            std::move(output),
            // We processed last buffer, attach data to it, so it will be notified when this buffer
            // is transferred.
            input_slice.empty() && input_it == input.end() ? std::move(data) : nullptr)));
      }
    }

    return Status::OK();
  }

  Result<ReadBufferFull> Decompress(StreamReadBuffer* inp, StreamReadBuffer* out) override {
    LZ4DecompressState state(
        decompress_input_buf_, decompress_output_buf_, &prev_decompress_data_left_);
    return state.Execute(inp, out);
  }

 private:
  char decompress_input_buf_[kLZ4BufferSize];
  char decompress_output_buf_[kLZ4BufferSize];
  Slice prev_decompress_data_left_;
  ScopedTrackedConsumption consumption_;
};

#undef LZ4
#define YB_COMPRESSION_ALGORITHMS (Zlib)(Snappy)(LZ4)

#define YB_CREATE_COMPRESSOR_CASE(r, data, name) \
  case BOOST_PP_CAT(name, Compressor)::data: \
      return std::make_unique<BOOST_PP_CAT(name, Compressor)>(std::move(mem_tracker));

std::unique_ptr<Compressor> CreateCompressor(char sign, MemTrackerPtr mem_tracker) {
  switch (sign) {
BOOST_PP_SEQ_FOR_EACH(YB_CREATE_COMPRESSOR_CASE, kId, YB_COMPRESSION_ALGORITHMS)
    default:
      return nullptr;
  }
}

std::unique_ptr<Compressor> CreateOutboundCompressor(MemTrackerPtr mem_tracker) {
  auto algo = FLAGS_stream_compression_algo;
  if (!algo) {
    return nullptr;
  }
  switch (algo) {
BOOST_PP_SEQ_FOR_EACH(YB_CREATE_COMPRESSOR_CASE, kIndex, YB_COMPRESSION_ALGORITHMS)
    default:
      YB_LOG_EVERY_N_SECS(DFATAL, 5) << "Unknown compression algorithm: " << algo;
      return nullptr;
  }
}

class CompressedRefiner : public StreamRefiner {
 public:
  CompressedRefiner() = default;

 private:
  void Start(RefinedStream* stream) override {
    stream_ = stream;
  }

  Status ProcessHeader() override {
    constexpr int kHeaderLen = 3;

    auto data = stream_->ReadBuffer().AppendedVecs();
    if (data.empty() || data[0].iov_len < kHeaderLen) {
      // Did not receive enough bytes to make a decision.
      // So just wait more bytes.
      return Status::OK();
    }

    const auto* bytes = static_cast<const uint8_t*>(data[0].iov_base);
    if (bytes[0] == 'Y' && bytes[1] == 'B') {
      compressor_ = CreateCompressor(bytes[2], stream_->buffer_tracker());
      if (compressor_) {
        RETURN_NOT_OK(compressor_->Init());
        RETURN_NOT_OK(stream_->StartHandshake());
        stream_->ReadBuffer().Consume(kHeaderLen, Slice());
        return Status::OK();
      }
    }

    // Don't use compression on this stream.
    return stream_->Established(RefinedStreamState::kDisabled);
  }

  Status Send(OutboundDataPtr data) override {
    boost::container::small_vector<RefCntBuffer, 10> input;
    data->Serialize(&input);
    return compressor_->Compress(input, stream_, std::move(data));
  }

  Status Handshake() override {
    if (stream_->local_side() == LocalSide::kClient) {
      compressor_ = CreateOutboundCompressor(stream_->buffer_tracker());
      if (!compressor_) {
        return stream_->Established(RefinedStreamState::kDisabled);
      }
      RETURN_NOT_OK(compressor_->Init());
      RETURN_NOT_OK(stream_->SendToLower(compressor_->ConnectionHeader()));
    }

    return stream_->Established(RefinedStreamState::kEnabled);
  }

  Result<ReadBufferFull> Read(StreamReadBuffer* out) override {
    VLOG_WITH_PREFIX(4) << __func__;

    return compressor_->Decompress(&stream_->ReadBuffer(), out);
  }

  const Protocol* GetProtocol() override {
    return CompressedStreamProtocol();
  }

  std::string ToString() const override {
    return compressor_ ? compressor_->ToString() : "PLAIN";
  }

  const std::string& LogPrefix() const {
    return stream_->LogPrefix();
  }

  RefinedStream* stream_ = nullptr;
  std::unique_ptr<Compressor> compressor_ = nullptr;
};

} // namespace

const Protocol* CompressedStreamProtocol() {
  static Protocol result("tcpc");
  return &result;
}

StreamFactoryPtr CompressedStreamFactory(
    StreamFactoryPtr lower_layer_factory, const MemTrackerPtr& buffer_tracker) {
  return std::make_shared<RefinedStreamFactory>(
      std::move(lower_layer_factory), buffer_tracker, [](const StreamCreateData& data) {
    return std::make_unique<CompressedRefiner>();
  });
}

}  // namespace rpc
}  // namespace yb

max_stars_repo_name: asad-awadia/yugabyte-db

max_issues_repo_name: asad-awadia/yugabyte-db

max_forks_repo_name: asad-awadia/yugabyte-db


text:
///////////////////////////////////////////////////////////////////////////////////////////////////
/// \file cqmc/timing/timing.cpp
///
/// \brief   implementation of functions and classes used in timing the code
///
///////////////////////////////////////////////////////////////////////////////////////////////////

#include <iostream>
#include <sstream>
#include <string>
#include <map>
#include <list>

#include <boost/format.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/scoped_ptr.hpp>

#include "formic/utils/mpi_interface.h"

#include "engine_timing.h"
#include "formic/utils/exception.h"

boost::scoped_ptr<std::pair<std::string,cqmc::Stopwatch> > cqmc::timers::propose_1e( new std::pair<std::string,cqmc::Stopwatch>(std::string("propose_1e"), cqmc::Stopwatch()) );
boost::scoped_ptr<std::pair<std::string,cqmc::Stopwatch> > cqmc::timers::propose_2e( new std::pair<std::string,cqmc::Stopwatch>(std::string("propose_2e"), cqmc::Stopwatch()) );

///////////////////////////////////////////////////////////////////////////////////////////////////
/// \brief   constructs a new stopwatch
///
///////////////////////////////////////////////////////////////////////////////////////////////////
cqmc::Stopwatch::Stopwatch() : _running(false), _start_time(0), _elapsed_time(0) {}

///////////////////////////////////////////////////////////////////////////////////////////////////
/// \brief   resets the stopwatch
///
///////////////////////////////////////////////////////////////////////////////////////////////////
void cqmc::Stopwatch::reset(const std::string & name) {
  if (_running)
    throw formic::Exception("cannot reset Stopwatch \"%s\" because it is running") % name;
  _elapsed_time = 0;
}

///////////////////////////////////////////////////////////////////////////////////////////////////
/// \brief   starts the stopwatch
///
///////////////////////////////////////////////////////////////////////////////////////////////////
void cqmc::Stopwatch::start(const std::string & name) {
  if (_running)
    throw formic::Exception("cannot start Stopwatch \"%s\" because it is already running") % name;
  _start_time = MPI_Wtime();
  _running = true;
}

///////////////////////////////////////////////////////////////////////////////////////////////////
/// \brief   stops the stopwatch
///
///////////////////////////////////////////////////////////////////////////////////////////////////
void cqmc::Stopwatch::stop(const std::string & name) {
  if (!_running)
    throw formic::Exception("cannot stop Stopwatch \"%s\" because it is not running") % name;
  _elapsed_time = _elapsed_time + ( MPI_Wtime() - _start_time );
  _running = false;
}

///////////////////////////////////////////////////////////////////////////////////////////////////
/// \brief   returns the elapsed time in seconds
///
///////////////////////////////////////////////////////////////////////////////////////////////////
double cqmc::Stopwatch::elapsed_seconds() const {
  double total = _elapsed_time;
  if (_running)
    total = total + ( MPI_Wtime() - _start_time );
  return total;
}

///////////////////////////////////////////////////////////////////////////////////////////////////
/// \brief   container to hold timers in use by the cqmc code
///
///////////////////////////////////////////////////////////////////////////////////////////////////
static std::map<std::string, cqmc::Stopwatch> cqmc_timer_container;

///////////////////////////////////////////////////////////////////////////////////////////////////
/// \brief   prints all timers in cqmc_timer_container
///
///////////////////////////////////////////////////////////////////////////////////////////////////
std::string cqmc::print_timers() {

  struct local_funcs {

    static void print_timer(std::stringstream & ss,
                            const cqmc::Stopwatch & sw,
                            const std::string & name) {

      // do not print the timer if it recorded no time
      if ( sw.elapsed_seconds() == 0.0 )
        return;

      // print the timer
      ss << boost::format("  %-30s      %15.6f seconds") % name % sw.elapsed_seconds() << std::endl;

    }

  };

  std::stringstream ss;

  ss << std::endl;
  ss << boost::format("Printing all timers used during this execution:") << std::endl;

  local_funcs::print_timer(ss, cqmc::timers::propose_1e->second, cqmc::timers::propose_1e->first);
  local_funcs::print_timer(ss, cqmc::timers::propose_2e->second, cqmc::timers::propose_2e->first);

  for (std::map<std::string, cqmc::Stopwatch>::const_iterator it = cqmc_timer_container.begin();
       it != cqmc_timer_container.end(); it++)
    local_funcs::print_timer(ss, it->second, it->first);
    //ss << boost::format("  %-30s      %15.6f seconds") % it->first % it->second.elapsed_seconds() << std::endl;

  ss << std::endl;

  return ss.str();

}

///////////////////////////////////////////////////////////////////////////////////////////////////
/// \brief   starts a timer
///
/// \param[in]     name     a string identifying the timer
///
///////////////////////////////////////////////////////////////////////////////////////////////////
void cqmc::start_timer(const std::string & name) {
  cqmc_timer_container[name].start(name);
}

///////////////////////////////////////////////////////////////////////////////////////////////////
/// \brief   stops a timer
///
/// \param[in]     name     a string identifying the timer
///
///////////////////////////////////////////////////////////////////////////////////////////////////
void cqmc::stop_timer(const std::string & name) {
  cqmc_timer_container[name].stop(name);
}

///////////////////////////////////////////////////////////////////////////////////////////////////
/// \brief   resets a timer
///
/// \param[in]     name     a string identifying the timer
///
///////////////////////////////////////////////////////////////////////////////////////////////////
void cqmc::reset_timer(const std::string & name) {
  cqmc_timer_container[name].reset(name);
}

ext: cpp

max_stars_repo_name: eugeneswalker/qmcpack

max_issues_repo_name: eugeneswalker/qmcpack

max_forks_repo_name: williamfgc/qmcpack


text:
//
// Copyright 2018-2020 Mateusz Loskot <mateusz at loskot dot net>
//
// Distributed under the Boost Software License, Version 1.0
// See accompanying file LICENSE_1_0.txt or copy at
// http://www.boost.org/LICENSE_1_0.txt
//
#include <boost/gil/algorithm.hpp>
#include <boost/gil/image.hpp>
#include <boost/gil/image_view_factory.hpp>

#include <boost/core/lightweight_test.hpp>

namespace gil = boost::gil;

void test_lambda_expression()
{
    gil::gray8_pixel_t const gray128(128);
    gil::gray8_image_t image(2, 2, gray128);

    int sum{0};
    gil::for_each_pixel(gil::view(image), [&sum](gil::gray8_pixel_t& p) {
        sum += gil::at_c<0>(p);
    });
    BOOST_TEST_EQ(sum, 2 * 2 * 128);
}

struct accumulator
{
    void operator()(gil::gray8_pixel_t const& p) {
        sum += gil::at_c<0>(p);
    }

    int sum = 0;
};

void test_function_object_1d_traversable()
{
    gil::gray8_pixel_t const gray128(128);
    gil::gray8_image_t image(2, 2, gray128);

    accumulator acc;
    acc = gil::for_each_pixel(
        gil::const_view(image),
        acc
    );
    BOOST_TEST_EQ(acc.sum, 2 * 2 * 128);
}

void test_function_object_not_1d_traversable()
{
    gil::gray8_pixel_t const gray128(128);
    gil::gray8_image_t image(4, 4, gray128);

    accumulator acc;
    acc = gil::for_each_pixel(
        gil::subimage_view(gil::const_view(image), gil::point_t{1, 1}, gil::point_t{2, 2}),
        acc
    );
    BOOST_TEST_EQ(acc.sum, 2 * 2 * 128);
}

int main()
{
    test_lambda_expression();
    test_function_object_1d_traversable();
    test_function_object_not_1d_traversable();

    return ::boost::report_errors();
}


text:
/*

Copyright (c) 2005-2019, University of Oxford.
All rights reserved.

University of Oxford means the Chancellor, Masters and Scholars of the
University of Oxford, having an administrative office at Wellington
Square, Oxford OX1 2JD, UK.

This file is part of Chaste.

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

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

*/

#ifndef ABSTRACTOUTPUTMODIFIER_HPP_
#define ABSTRACTOUTPUTMODIFIER_HPP_

#include "ChasteSerialization.hpp"
#include <boost/serialization/string.hpp>
#include "ClassIsAbstract.hpp"

#include <string>
#include "AbstractTetrahedralMesh.hpp"

/**
 * A plug-in class for on-the-fly output.  This is designed so that a user can insert something in
 * order to monitor the progress of a simulation or to produce "post processed" output during the simulation.
 */
class AbstractOutputModifier
{
private:
    /** For testing */
    friend class TestMonodomainProblem;

    /** Needed for serialization. */
    friend class boost::serialization::access;
    /**
     * Archive this modifier.  Just calls the base class version.
     *
     * @param archive the archive
     * @param version the current version of this class
     */
    template<class Archive>
    void serialize(Archive & archive, const unsigned int version)
    {
        archive & mFilename;
        archive & mFlushTime;
    }

protected:
    /** Constructor that does nothing, for archiving */
    AbstractOutputModifier()
        : mFilename(), mFlushTime(0.0)
    {}

    std::string mFilename; /**<The file which is eventually produced by this modifier*/

    /** Simulation time period between flushes to disk */
    double mFlushTime;

public:
    /**
     * Standard construction method contains only the name of the file which this simulation modifier should produce.
     *
     * Note that construction occurs before the Solve() loop and probably before initialisation.  This means that it
     * will not be possible to view certain data (e.g the mesh) at the time of construction
     *
     * Note the problem passes parameters in a non-templated fashion in order to keep the interface as lightweight as
     * possible.
     * @param rFilename  The file which is eventually produced by this modifier
     * @param flushTime  The (simulation) time between flushing the file to disk. Default (0) means don't flush.
     *
     */
    AbstractOutputModifier(const std::string& rFilename, double flushTime=0.0)
        : mFilename(rFilename),
          mFlushTime(flushTime)
    {}
    /**
     * Destructor should be overridden when necessary
     */
    virtual ~AbstractOutputModifier()
    {

    }

    /**
     * Initialise the modifier (open a file or make some memory) when the solve loop is starting
     *
     * Note the problem passes parameters in a non-templated fashion in order to keep the interface as lightweight as
     * possible.  That is, it might have been slicker to pass in the mesh but that would require multiple templates.
     * @param pVectorFactory  The vector factory which is associated with the calling problem's mesh
     * @param rNodePermutation The permutation associated with the calling problem's mesh (when running with parallel partitioning)
     */
    virtual void InitialiseAtStart(DistributedVectorFactory* pVectorFactory, const std::vector<unsigned>& rNodePermutation)=0;

    /**
     * Finalise the modifier (close a file or dump the calculation to disk)
     */
    virtual void FinaliseAtEnd()=0;

    /**
     * Process a solution time-step (dump a small line to file or compute the latest activation times)
     * @param time  The current simulation time
     * @param solution  A working copy of the solution at the current time-step.  This is the PETSc vector which is distributed across the processes.
     * @param problemDim  The calling problem dimension. Used here to avoid probing the size of the solution vector
     */
    virtual void ProcessSolutionAtTimeStep(double time, Vec solution, unsigned problemDim)=0;
};

CLASS_IS_ABSTRACT(AbstractOutputModifier)

#endif /* ABSTRACTOUTPUTMODIFIER_HPP_ */

ext: hpp


text:
#define BOOST_TEST_DYN_LINK
#define BOOST_TEST_MODULE JPetParamsTest
#include <boost/test/unit_test.hpp>
#include <boost/any.hpp>

#include "./JPetParams/JPetParams.h"

BOOST_AUTO_TEST_SUITE(FirstSuite)

BOOST_AUTO_TEST_CASE(defaultConstr)
{
  JPetParams params;

  BOOST_REQUIRE(params.getParamManager() == nullptr);
  BOOST_REQUIRE(params.getOptions().empty());
}

BOOST_AUTO_TEST_CASE(nonDefaultConstr)
{
  jpet_options_tools::OptsStrAny opts;
  opts["blaOption"] = std::string("value");
  auto mgr = std::make_shared<JPetParamManager>(new JPetParamManager);
  JPetParams params(opts, mgr);
  BOOST_REQUIRE_EQUAL(params.getParamManager(), mgr.get());
  BOOST_REQUIRE_EQUAL(params.getOptions().size(), 1u);
}

BOOST_AUTO_TEST_CASE(copyCtr)
{
  jpet_options_tools::OptsStrAny opts;
  opts["blaOption"] = std::string("value");
  auto mgr = std::make_shared<JPetParamManager>(new JPetParamManager);
  JPetParams params(opts, mgr);
  auto params2 = params;
  BOOST_REQUIRE_EQUAL(params2.getParamManager(), mgr.get());
  BOOST_REQUIRE_EQUAL(params2.getOptions().size(), params.getOptions().size());
}

BOOST_AUTO_TEST_CASE(memoryLeaks)
{
  jpet_options_tools::OptsStrAny opts;
  opts["blaOption"] = std::string("value");
  JPetParams* params = new JPetParams(opts, std::make_shared<JPetParamManager>(new JPetParamManager));
  auto mgr = params->getParamManager();
  delete params;
  BOOST_REQUIRE(mgr != nullptr);
}

BOOST_AUTO_TEST_SUITE_END()

max_stars_repo_name: Alvarness/j-pet-framework

max_issues_repo_name: Alvarness/j-pet-framework

max_forks_repo_name: Alvarness/j-pet-framework


text:
//////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////
//
//  License: MIT, see included COPYING file
//
//
//
//////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////

#include <string.h>
#include <sstream>
#include <iostream>
#include <vector>
#include <algorithm>
#include <boost/algorithm/string.hpp>
#include <boost/lexical_cast.hpp>

#include "aligner.h"

const int MaxLibrarySize = 1000000;

////////////////////////////////////////////////////////////////////////////
Aligner::Aligner(){
	m_useSSW = 0;
	m_byAA = 0;
}

////////////////////////////////////////////////////////////////////////////
Aligner::Aligner(const char *idxFile, const char* name){

	m_useSSW = 0;
	m_byAA = 0;
	m_bwaAligner.Initialize(idxFile, name);
}

////////////////////////////////////////////////////////////////////////////
int Aligner::Initialize(const char *idxFile, const char * name,
	BWAalignConfig & config){

	return Initialize(idxFile, name,
		config.A, config.B,
		config.OINS, config.EINS,
		config.ODEL,config.EDEL,
		config.CLIP3, config.CLIP5,
		config.MIN_SEED_LEN, config.MIN_SCORE);

}

////////////////////////////////////////////////////////////////////////////
int Aligner::Initialize(const char *idxFile, const char * name,
	int a, int b,
	int o_ins, int e_ins,
	int o_del, int e_del,
	int pen_clip3, int pen_clip5,
	int min_seed_len,
	int T
	){

	if(idxFile == NULL) {return -1;}
	std::string idxFileStr = idxFile;
	if(idxFileStr.compare("") == 0){ return -1;}

	m_bwaAligner.Initialize(idxFile, name,
		a, b,
		o_ins, e_ins,
		o_del, e_del,
		pen_clip3, pen_clip5,
		min_seed_len,
		T
	);

	// read in all the seqs
	m_library.OpenSeqFiles(idxFile,"");
	int libsize = m_library.ReadNSeqs(MaxLibrarySize);
	for(int i=0;i<libsize; i++){
		std::string libname = m_library.m_bseqs[i].name;
		m_libraryEntries[libname] = i;
	}
	return 0;
}

////////////////////////////////////////////////////////////////////////////
Aligner::~Aligner(){
}
////////////////////////////////////////////////////////////////////////////
int Aligner::SetMultiHit(int multihit){
	m_bwaAligner.SetMultiHit(multihit);
	m_sswAligner.SetMultiHit(multihit);
	return 0;
}
////////////////////////////////////////////////////////////////////////////
void PrettyPrintAlignment(std::string outref, std::string outq){
	std::string matches = std::string(outref.size(),' ');
	int match = 0;
	int mismatch = 0;
	for(unsigned int i=0; i<outref.size(); i++){
		if(outref[i] == outq[i]){
			matches[i] = '|';
			match++;
		}else{
			mismatch++;
		}
	}

	printf("\tR/Q\n\t%s\n\t%s\n\t%s\n\tMatch:%d\n\tMisMa:%d\n",
		outref.c_str(), matches.c_str(), outq.c_str(), match, mismatch);
}
////////////////////////////////////////////////////////////////////////////
// generate an alignment for ref and query from the cigar string
void GetAlignment(std::string ref, std::string cigar,
	int pos, std::string q,  std::string & outref, std::string & outq){

	pos = pos - 1; //convert to zero-based
	if(pos<0){fprintf(stderr,"ERROR:Bad pos:%d\n",pos); exit(-1);}

	std::vector<char> cigM;
	std::vector<int> cigI;

	char numbuf[32];
	int numbufi = 0;
	for(unsigned int c=0;c<cigar.size();c++){
		if(isdigit(cigar[c])){
			numbuf[numbufi++] = cigar[c];
		}else{
			numbuf[numbufi] = '\0';
			numbufi = 0;
			cigM.push_back(cigar[c]);
			cigI.push_back(atoi(numbuf));
		}
	}

	outref = "";
	outq = "";
	unsigned int refcursor = 0;
	unsigned int qcursor = 0;
	std::string refadd;
	std::string qadd;
	for(unsigned int i = 0; i< cigM.size(); i++){
		switch(cigM[i]){
			case 'S':
				if(i==0){ // 5' softclip
					if(pos>cigI[i]){ // ref offset bigger
						refadd = ref.substr(0,pos);
						boost::algorithm::to_lower(refadd);
						outref += refadd;

						qadd = std::string(pos-cigI[i],'-') +
							q.substr(0,cigI[i]);
						boost::algorithm::to_lower(qadd);
						outq += qadd;
						refcursor += pos;
						qcursor += cigI[i];
					}else{
						refadd = std::string(cigI[i]-pos,'-') +
							ref.substr(0,pos);
						boost::algorithm::to_lower(refadd);
						outref += refadd;
						qadd = q.substr(0,cigI[i]);
						boost::algorithm::to_lower(qadd);
						outq += qadd;
						refcursor += pos;
						qcursor += cigI[i];
					}
				}else{ // 3' softclip
					if(ref.substr(refcursor).size() > //ref is longer
						q.substr(qcursor).size()){

						refadd = ref.substr(refcursor);
						boost::algorithm::to_lower(refadd);
						outref += refadd;

						qadd = q.substr(qcursor) +
							std::string(ref.substr(refcursor).size() -
								q.substr(qcursor).size(),'-');

						boost::algorithm::to_lower(qadd);
						outq += qadd;
					}else{
						refadd = ref.substr(refcursor) +
							std::string(q.substr(qcursor).size() -
								ref.substr(refcursor).size(),'-' );
						boost::algorithm::to_lower(refadd);
						outref += refadd;

						qadd = q.substr(qcursor);
						boost::algorithm::to_lower(qadd);
						outq += qadd;
					}

					refcursor = ref.size();
					qcursor = q.size();
				}
				assert(outref.size() == outq.size());
			break;
			case 'M':
			case 'X':
			case '=':
				if(i==0){
					outref += ref.substr(refcursor, pos+cigI[i]);
					outq += std::string(pos,'-') + q.substr(qcursor, cigI[i]);
					refcursor += pos + cigI[i];
				}else{
					outref += ref.substr(refcursor, cigI[i]);
					outq += q.substr(qcursor, cigI[i]);
					refcursor += cigI[i];
				}
				qcursor += cigI[i];
				assert(outref.size() == outq.size());
			break;
			case 'I':
				if(i==0){
					outref += ref.substr(0,pos) + std::string(cigI[i],'-');
					refcursor = pos;
					outq += std::string(pos,'-') + q.substr(qcursor, cigI[i]);
					qcursor += cigI[i];
				}else{
					outref += std::string(cigI[i],'-');
					outq += q.substr(qcursor, cigI[i]);
					qcursor += cigI[i];
				}
				assert(outref.size() == outq.size());
			break;
			case 'D':
				if(i==0){
					outq += std::string(pos,'-') + std::string(cigI[i],'-');
					qcursor += 0;
					outref += ref.substr(0,pos) + ref.substr(pos+1, cigI[i]);
					refcursor += pos + cigI[i];

				}else{
					outref += ref.substr(refcursor, cigI[i]);
					outq += std::string(cigI[i],'-');
					refcursor += cigI[i];
				}

				assert(outref.size() == outq.size());
			break;
			default:
				fprintf(stderr, "ERROR: unhandled cigar type:%c\n",cigM[i]);
				exit(-1);
			break;
		}
	}

	//fill in the last gaps
	if(refcursor < ref.size() || qcursor < q.size()){
		// C- -- CC CC
		// GG GG G- --
		outq += outq.substr(qcursor);
		outref += ref.substr(refcursor);
		if(outq.size() > outref.size()){
			outref += std::string(outq.size()-outref.size(), '-');
		}else if(outref.size() > outq.size()){
			outq += std::string(outref.size()-outq.size(), '-');
		}
	}

	if(outref.size() != outq.size()){
		fprintf(stderr, "ERROR:alignment string sizes do not match\n");
		exit(-1);
	}

}

////////////////////////////////////////////////////////////////////////////
// compute the SHM and their positions for query against a call
int Aligner::ComputeSHM(std::string call,
	const char *seq, int seqlen,
	std::string & cigar,
	std::string & shm,
	std::vector<int> & mutpos){

	if(!Ready()){ return -1; }

	std::vector<std::string> allcalls;
	std::istringstream callstream(call);
	std::string onecall;
	int callindex = 0;
	shm = "";
	while(getline(callstream, onecall, ',')){

		if(m_libraryEntries.find(onecall) == m_libraryEntries.end()){
			return -1;
		}

		int index = m_libraryEntries[onecall];

		char *rseq = m_library.m_bseqs[index].seq;

		std::string query = seq;
		std::string ref = rseq;

		std::string sswcigar;
		int rpos, mismatches;
		double score;

		m_sswAligner.Align(
			query.c_str(), query.size(), ref.c_str(), ref.size(),
			sswcigar, rpos, score, mismatches, m_byAA);

		std::string qalign;
		std::string ralign;

		GetAlignment(rseq, sswcigar, rpos, seq, ralign, qalign);

		//Find the first non "-" 5p char in ref
		//       |
		//   ----aagtTTTTTTT  # ref
		//   gjksacccTTTTTTT  # seq
		//   |
		//   aagtgggTTTTTTT  # ref
		//   ---acccTTTTTTT  # seq
		unsigned int startref = 0;
		int inFlank = 1;
		for(unsigned int i=0;i<ralign.size();i++){
			if(ralign[i] != '-') inFlank = 0;
			if(ralign[i] == '-' && inFlank) startref = i+1;
			if(!inFlank) break;
		}

		//Find the the first non '-' 3p char in ref
		unsigned int endref = ralign.size()-1;
		inFlank = 1;
		for(int i=endref;i>=0; i--){
			if(ralign[i] != '-') inFlank = 0;
			if(ralign[i] == '-' && inFlank) endref = i-1;
			if(!inFlank) break;
		}

		//map each align pos to its position in the query
		//and then mask it
		std::vector<int> querypos;
		int curQpos = 0;
		for(unsigned int i=0;i<qalign.size();i++){
			querypos.push_back(curQpos);
			if(qalign[i] != '-'){
				curQpos++;
				if(curQpos >= seqlen){ curQpos = seqlen-1;}
			}
		}

		int shmscore = 0;
		for(unsigned int i=startref;i<=endref;i++){
			if(ralign[i] != qalign[i]){
				if(callindex == 0){ //only tally mutpos for first call
					mutpos.push_back(querypos[i]);
				}
				shmscore++;
			}
		}

		if(callindex>0){
			shm = shm + ",";
		}
		shm = shm + boost::lexical_cast<std::string>( shmscore  );
		callindex++;

	}
	return 0;
}
////////////////////////////////////////////////////////////////////////////
int Aligner::Ready(){
	return(m_library.m_Nbseqs > 0);
}

////////////////////////////////////////////////////////////////////////////
//
int Aligner::RunClassifier(const char *seq, int seqlen,
		std::string & call, std::string & score,
		std::string & pos, std::string & cigar,
		int firstpos, // = 0,
		int lastpos // = -1,
	){
	call = "";
	pos = "";
	cigar = "";

	if(Ready()){
		if(m_useSSW || m_byAA){
			m_sswAligner.FindBestAlignment(seq, seqlen,
				call, score, pos, cigar, m_byAA, m_library, firstpos, lastpos);
		}else{
			m_bwaAligner.FindBestAlignment(seq, seqlen,
				call, score, pos, cigar, firstpos, lastpos);
		}
	}
	return 0;
}

////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////

max_stars_repo_name: Genentech/Absolve

max_issues_repo_name: Genentech/Absolve

max_forks_repo_name: Genentech/Absolve


text:
/*!
@file
Defines `boost::hana::SearchableSet`.

@copyright Louis Dionne 2015
Distributed under the Boost Software License, Version 1.0.
(See accompanying file LICENSE.md or copy at http://boost.org/LICENSE_1_0.txt)
 */

#ifndef BOOST_HANA_SANDBOX_SEARCHABLE_SET_HPP
#define BOOST_HANA_SANDBOX_SEARCHABLE_SET_HPP

#include <boost/hana/applicative.hpp>
#include <boost/hana/comparable.hpp>
#include <boost/hana/config.hpp>
#include <boost/hana/core/operators.hpp>
#include <boost/hana/functor.hpp>
#include <boost/hana/logical.hpp>
#include <boost/hana/maybe.hpp>
#include <boost/hana/monad.hpp>
#include <boost/hana/searchable.hpp>

namespace boost { namespace hana {
    //! @ingroup group-datatypes
    //! A `Monad` for searching infinite sets in finite time.
    //!
    //! Taken from http://math.andrej.com/2008/11/21/a-haskell-monad-for-infinite-search-in-finite-time/.
    struct SearchableSet {
        struct hana {
            struct operators
                : boost::hana::operators::of<Comparable, Monad>
            { };
        };
    };

    template <typename Find, typename = operators::adl>
    struct _sset {
        Find find;
        struct hana { using datatype = SearchableSet; };
    };

    BOOST_HANA_CONSTEXPR_LAMBDA auto searchable_set = [](auto pred) {
        return _sset<decltype(pred)>{pred};
    };

    BOOST_HANA_CONSTEXPR_LAMBDA auto singleton = [](auto x) {
        return searchable_set([=](auto p) { return x; });
    };

    BOOST_HANA_CONSTEXPR_LAMBDA auto doubleton = [](auto x, auto y) {
        return searchable_set([=](auto p) {
            return if_(p(x), x, y);
        });
    };

    BOOST_HANA_CONSTEXPR_LAMBDA auto union_ = [](auto xs, auto ys) {
        return flatten(doubleton(xs, ys));
    };

    //////////////////////////////////////////////////////////////////////////
    // Comparable
    //////////////////////////////////////////////////////////////////////////
    template <>
    struct equal_impl<SearchableSet, SearchableSet> {
        template <typename Xs, typename Ys>
        static constexpr auto apply(Xs xs, Ys ys)
        { return and_(subset(xs, ys), subset(ys, xs)); }
    };

    //////////////////////////////////////////////////////////////////////////
    // Functor
    //////////////////////////////////////////////////////////////////////////
    template <>
    struct transform_impl<SearchableSet> {
        template <typename Set, typename F>
        static constexpr auto apply(Set set, F f) {
            return searchable_set([=](auto q) {
                return f(set.find([=](auto x) { return q(f(x)); }));
            });
        }
    };

    //////////////////////////////////////////////////////////////////////////
    // Applicative
    //////////////////////////////////////////////////////////////////////////
    template <>
    struct lift_impl<SearchableSet> {
        template <typename X>
        static constexpr auto apply(X x)
        { return singleton(x); }
    };

    template <>
    struct ap_impl<SearchableSet> {
        template <typename F, typename Set>
        static constexpr auto apply(F fset, Set set) {
            return flatten(transform(fset, [=](auto f) {
                return transform(set, f);
            }));
        }
    };

    //////////////////////////////////////////////////////////////////////////
    // Monad
    //////////////////////////////////////////////////////////////////////////
    template <>
    struct flatten_impl<SearchableSet> {
        template <typename Set>
        static constexpr auto apply(Set set) {
            return searchable_set([=](auto p) {
                return set.find([=](auto set) {
                    return any_of(set, p);
                }).find(p);
            });
        }
    };

    //////////////////////////////////////////////////////////////////////////
    // Searchable
    //////////////////////////////////////////////////////////////////////////
    template <>
    struct find_if_impl<SearchableSet> {
        template <typename Set, typename Pred>
        static constexpr auto apply(Set set, Pred p) {
            auto x = set.find(p);
            return if_(p(x), just(x), nothing);
        }
    };

    template <>
    struct any_of_impl<SearchableSet> {
        template <typename Set, typename Pred>
        static constexpr auto apply(Set set, Pred p) {
            return p(set.find(p));
        }
    };
}} // end namespace boost::hana

#endif // !BOOST_HANA_SANDBOX_SEARCHABLE_SET_HPP

max_stars_repo_name: josephwinston/hana

max_issues_repo_name: josephwinston/hana

max_forks_repo_name: josephwinston/hana


text:
#include <DataManager.h>
#include <BookTools.h>
#include <cl-util/Configurable.h>

#include <boost/algorithm/string.hpp>

#include <iostream>
#include <vector>
#include <string>

#include <cl-util/factory.h>
#include <cl-util/debug_stream.h>

using namespace std;
using namespace clite::util;

class SimDebug :
    public WakeupHandler::listener,
    public MarketHandler::listener,
    public OrderHandler::listener

{

    factory<DataManager>::pointer dm_;
    DataManager &dm;
    char buf[160];

    bool showOrder, showData, showMaster;
    vector<bool> showBook;

    bool nextData, nextOrder, nextMove, nextWake, advancing;
    TimeVal nexttv;

    public:

    SimDebug ( )
        : dm_(factory<DataManager>::get(only::one)),
        dm(*dm_),
        showOrder(false), showData(false),
        showMaster(false), showBook(ECN::ECN_size, false),
        nextData(true), nextOrder(true),
        nextMove(true), nextWake(true)

    { dm.add_listener(this); }

    void interact () {
        using namespace boost::algorithm;

        while (true) {
            cout << (trc::compat::util::DateTime(dm.curtv()).gettimestring()) << "> ";
            cout.flush();
            string input;
            vector<string> sres;
            getline(cin, input);

            split(sres, input, is_space(), token_compress_on);

            if (sres.size() == 0) { continue; }

            // <symbol> <b/s> <size> <price> <ecn> <timeout>

            if (sres[0] == "buy" || sres[0] == "sell" || sres[0] == "short") {
                if (sres.size() != 6) {
                    cerr << "trade needs 6 args: <buy/sell/short> <size> <symbol> <price> <ecn> <timeout>." << endl;
                    continue;
                }
                int cid = dm.cid(sres[2].c_str());
                if (cid == -1) {
                    cerr << "symbol " << sres[1] << " not known." << endl;
                    continue;
                }

                Mkt::Side dir;
                if (sres[0] == "buy") dir = Mkt::BID;
                else if (sres[0] == "sell") dir = Mkt::ASK;
                else continue;
                int size = atoi(sres[1].c_str());
                double px = atof(sres[3].c_str());
                ECN::ECN ecn = ECN::ECN_parse(sres[4].c_str());
                int time = atoi(sres[5].c_str());
                dm.placeOrder(cid, ecn, size, px, dir, time);
                continue;
            }

            if (sres[0] == "cxl" && sres.size() > 1) {
                int cid = dm.cid(sres[1].c_str());
                if (cid == -1) { cerr << "symbol not known." << endl; continue; }
                if (sres.size() > 2) {
                    double px = atof(sres[2].c_str());
                    dm.cancelMarket(cid, Mkt::BID, px);
                    dm.cancelMarket(cid, Mkt::ASK, px);
                } else {
                    dm.cancelMarket(cid);
                }
                continue;
            }
            if (sres[0] == "show" && sres.size() > 2) {
                bool on = sres[2] == "on";
                if (sres[1] == "data") showData = on;
                if (sres[1] == "order") showOrder = on;
                if (sres[1] == "master") showMaster = on;
                ECN::ECN ecn = ECN::ECN_parse(sres[1].c_str());
                if (ecn != ECN::UNKN) {
                    showBook[ecn] = on;
                }

                continue;
            }

            if ((sres[0] == "n" || sres[0] == "next") && sres.size() == 1) {
                nextData = nextOrder = nextMove = true;
                return;
            }
            if ((sres[0] == "n" || sres[0] == "next") && sres.size() > 1) {
                if (sres[1] == "data")  nextData = true;
                else if (sres[1] == "order") nextOrder = true;
                else if (sres[1] == "move")  nextMove = true;
                else if (sres[1] == "wake")  nextWake = true;
                else continue;

                return;
            }

            if (sres[0] == "pos") {
                for (int i = 0; i < dm.cidsize(); ++i)
                    cout << dm.symbol(i) << ":\t" << dm.position(i) << endl;
                continue;
            }
            if (sres[0] == "print" && sres.size() > 2) {
                int cid = dm.cid(sres[1].c_str());
                ECN::ECN ecn = ECN::ECN_parse(sres[2].c_str());

                if (cid != -1) {
                    if (sres[2] == "book") {
                        printBook(cid);
                        continue;
                    } else if (sres[2] == "orders") {
                        printOrders(cid);
                        for (int i = 0; i < ECN::ECN_size; ++i)
                            printOrders(cid, ECN::ECN(i));
                        continue;
                    } else if (ecn != ECN::UNKN) {
                        printBook(cid, ecn);
                        continue;
                    }
                }
            }

            if (sres[0] == "adv" && sres.size() > 1) {
                vector<string> times;
                split(times, sres[1], is_any_of(":."));
                int hh = 0, mm = 0, ss = 0, us = 0;
                trc::compat::util::DateTime dt;
                switch(times.size()) {
                    case 4: us = atoi(times[3].c_str());
                    case 3: ss = atoi(times[2].c_str());
                    case 2: mm = atoi(times[1].c_str());
                    case 1: hh = atoi(times[0].c_str());
                            dt = dm.curtv();
                            dt.setmidnight();
                            dt.settime(hh,mm,ss);
                            nexttv = dt.getTimeVal();
                            nexttv.usec() = us;
                            advancing = true;
                            break;
                    default:
                            cerr << "time must be specified with 1-4 elements" << endl;
                            continue;
                }
                return;
            }

            if (sres[0] == "quit") {
                dm.stop("Debug: quit");
            }

            cerr << "I didn't understand that." << endl;
        }
    }

    void update ( const WakeUpdate &w ) {
        if (nextWake) {
            nextWake = false;
            interact();
        } else if (advancing && dm.curtv() >= nexttv) {
            advancing = false;
            interact();
        }
    }

    void update(const DataUpdate &du) {
        if (showData) {
            du.snprint(buf, sizeof(buf));
            cout << buf << endl;
        }
        if (showBook[du.ecn]) printBook(du.cid, du.ecn);
        if (showMaster) printBook(du.cid);
        if (nextData) {
            nextData = false;
            interact();
        }
    }

    void update (const OrderUpdate &ou) {
        if (showOrder) {
            ou.snprint(buf, sizeof(buf));
            cout << buf << endl;
        }
        if (nextOrder) {
            nextOrder = false;
            interact();
        }
    }

    void printBook(int cid) {
        int n = 10;
        size_t asz, bsz;
        double apx, bpx;
        bool bid, ask;
        printf("----- Book for %s -----\n",
                dm.symbol(cid));
        int i = 0;
        do {
            bid = getMarket(dm.masterBook(), cid, Mkt::BID, i, &bpx, &bsz);
            ask = getMarket(dm.masterBook(), cid, Mkt::ASK, i, &apx, &asz);
            if (bid && ask) printf("%7lu @ %7.2f |%1d| %7.2f @ %7lu \n", bsz, bpx, i, apx, asz);
            else if (bid)   printf("%7lu @ %7.2f |%1d| ------- @ ----- \n", bsz, bpx, i);
            else if (ask)   printf("----- @ ------- |%1d| %7.2f @ %7lu \n", i, apx, asz);
            ++i;
        } while ((bid || ask) && i < n);
    }

    void printBook(int cid, ECN::ECN ecn) {
        int n = 10;
        size_t asz, bsz;
        double apx, bpx;
        bool bid, ask;
        printf("----- %4s Book for %s -----\n",
                ECN::desc(ecn), dm.symbol(cid));
        int i = 0;
        do {
            bid = getMarket(dm.subBook(ecn), cid, Mkt::BID, i, &bpx, &bsz);
            ask = getMarket(dm.subBook(ecn), cid, Mkt::ASK, i, &apx, &asz);
            if (bid && ask) printf("%7lu @ %7.2f |%1d| %7.2f @ %7lu \n", bsz, bpx, i, apx, asz);
            else if (bid)   printf("%7lu @ %7.2f |%1d| ------- @ ----- \n", bsz, bpx, i);
            else if (ask)   printf("----- @ ------- |%1d| %7.2f @ %7lu \n", i, apx, asz);
            ++i;
        } while ((bid || ask) && i < n);
    }

    void printOrders(int cid) {
        size_t asz, bsz;
        double apx, bpx;
        bool bid, ask;
        printf("----- My Order Book for %s -----\n",
                dm.symbol(cid));
        int i = 0;
        do {
            bid = getMarket(dm.orderBook(), cid, Mkt::BID, i, &bpx, &bsz);
            ask = getMarket(dm.orderBook(), cid, Mkt::ASK, i, &apx, &asz);
            if (bid && ask) printf("%7lu @ %7.2f |%1d| %7.2f @ %7lu \n", bsz, bpx, i, apx, asz);
            else if (bid)   printf("%7lu @ %7.2f |%1d| ------- @ ----- \n", bsz, bpx, i);
            else if (ask)   printf("----- @ ------- |%1d| %7.2f @ %7lu \n", i, apx, asz);
            ++i;
        } while (bid || ask);
    }
    void printOrders(int cid, ECN::ECN ecn) {
        size_t asz, bsz;
        double apx, bpx;
        bool bid, ask;
        printf("----- My %s Order Book for %s -----\n",
                ECN::desc(ecn), dm.symbol(cid));
        int i = 0;
        do {
            bid = getMarket(dm.subOrderBook(ecn), cid, Mkt::BID, i, &bpx, &bsz);
            ask = getMarket(dm.subOrderBook(ecn), cid, Mkt::ASK, i, &apx, &asz);
            if (bid && ask) printf("%7lu @ %7.2f |%1d| %7.2f @ %7lu \n", bsz, bpx, i, apx, asz);
            else if (bid)   printf("%7lu @ %7.2f |%1d| ------- @ ----- \n", bsz, bpx, i);
            else if (ask)   printf("----- @ ------- |%1d| %7.2f @ %7lu \n", i, apx, asz);
            ++i;
        } while (bid || ask);
    }
};

int main ( int argc, char **argv ) {

    bool help = false;

    factory<DataManager>::pointer dm(factory<DataManager>::get(only::one));
    CmdLineFileConfig cfg(argc, argv, "config,C");

    cfg.defSwitch("help,h", &help, "print this help.");

    cfg.add(*dm);
    cfg.add(*debug_stream_config::get_config());
    cfg.add(*file_table_config::get_config());
    cfg.configure();
    if (help) { cerr << cfg << endl; return 1; }

    if(!dm->initialize()) return 2;

    SimDebug dbg;

    dm->run();
}

max_stars_repo_name: CodeApprenticeRai/newt

max_issues_repo_name: CodeApprenticeRai/newt

max_forks_repo_name: CodeApprenticeRai/newt


text:
/*
 * Copyright (C) 2013-2014, The OpenFlint Open Source Project
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS-IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/algorithm/string.hpp>
#include <boost/algorithm/string_regex.hpp>
#include <boost/format.hpp>
#include <boost/tokenizer.hpp>
#include <boost/regex.hpp>

#include "UDPServer.h"
#include "utils/Logging.h"
#include "net/NetworkManager.h"

namespace flint {

UDPServer::UDPServer(boost::asio::io_service &ioService,
		const std::string &listenIp, int listenPort) :
		timer_(ioService), socket_(ioService), listenIp_(listenIp), listenPort_(
				listenPort) {
	// TODO Auto-generated constructor stub
}

UDPServer::~UDPServer() {
	// TODO Auto-generated destructor stub
}

void UDPServer::init() {
//	LOG_INFO << "UDPServer init!!! -> " << listenIp_ << ":" << listenPort_;
	boost::asio::ip::address address = boost::asio::ip::address::from_string(
			"0.0.0.0");
	boost::asio::ip::udp::endpoint endpoint(address, listenPort_);

	// open
	socket_.open(endpoint.protocol());

	// reuse address, must set before binding!!!
	boost::asio::ip::udp::socket::reuse_address reuse_option(true);
	socket_.set_option(reuse_option);

	// set ttl
	boost::asio::ip::multicast::hops ttl(1);
	socket_.set_option(ttl);

	// enable loopback
	boost::asio::ip::multicast::enable_loopback loopback(true);
	socket_.set_option(loopback);

	// close quickly
	boost::asio::socket_base::linger linger_option(true, 0);
	socket_.set_option(linger_option);

	try {
		// bind
		socket_.bind(endpoint);
	} catch (boost::system::system_error &e) {
		LOG_ERROR << "UDPServer bind: " << e.what();
	}

	// join the multicast group.
	boost::asio::ip::address multicastAddress =
			boost::asio::ip::address::from_string(listenIp_);
	try {
		std::string ip = NetworkManager::getInstance()->getIpAddr();
		if (boost::iequals(ip, "")) {
//			LOG_INFO << "join group: " << multicastAddress.to_string();
			socket_.set_option(
					boost::asio::ip::multicast::join_group(multicastAddress));
		} else {
//			LOG_INFO << "join group: " << multicastAddress.to_string() << " | "
//					<< ip;
			socket_.set_option(
					boost::asio::ip::multicast::join_group(
							multicastAddress.to_v4(),
							boost::asio::ip::address::from_string(ip).to_v4()));
		}
	} catch (boost::system::system_error &e) {
		LOG_ERROR << "UDPServer join group: " << e.what();
	}
//	LOG_DEBUG << "UDPServer init!!!- end";
}

void UDPServer::onStart() {
//	LOG_INFO << "UDPServer start!!! -> " << listenIp_ << ":" << listenPort_;
	init();
	wait_packet();
}

void UDPServer::onStop() {
	socket_.close();
	timer_.cancel();
}

void UDPServer::wait_packet() {
//	LOG_DEBUG << "UDPServer wait packet!!!";
	recv_buffer_.fill(0);
	socket_.async_receive_from(boost::asio::buffer(recv_buffer_),
			client_endpoint_,
			boost::bind(&UDPServer::handle_packet, this,
					boost::asio::placeholders::error,
					boost::asio::placeholders::bytes_transferred));
//	LOG_DEBUG << "UDPServer wait packet!!! end";
}

void UDPServer::handle_packet(const boost::system::error_code& error,
		std::size_t len) {
//	LOG_DEBUG << "UDPServer handle packet!!!";
	if (!error || error == boost::asio::error::message_size) {
		if (len > 0) {
//			std::string message;
//			message.assign(recv_buffer_.data(), len);
			parseMessage(client_endpoint_.address().to_string(),
					client_endpoint_.port(), recv_buffer_.data(), len);
		}
		wait_packet();
	}
//	LOG_DEBUG << "UDPServer handle packet!!! end";
}

} /* namespace flint */

max_stars_repo_name: jsli/flingd-cc


text:
#pragma once
#ifndef CLIENT_H_
#define CLIENT_H_

#include <string.h>
#include <errno.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <pthread.h>
#include <memory>
#include <csignal>
#include <vector>
#include <random>
#include <unordered_map>
#include <dirent.h>
#include <boost/functional/hash.hpp>

#include <iostream>
#include <fstream>

#include <psp/libos/persephone.hh>
#include <psp/hist.hh>
#include "msg.hh"
#include "schedule.hh"
#include "common.hh"

#include <rte_mbuf.h>

template <typename T>
void sample_into(std::vector<T> &from, std::vector<T>&to,
                 auto lat_get,   // Must match form: ComparableType lat_get(const T&) {}
                 auto time_get,  // Must match form: ComparableType time_get(const T&) {}
                 size_t n) {
    if (from.empty()) {
        return;
    }
    auto time_cmp = [&](T&a, T&b) { return time_get(*a) < time_get(*b); };
    auto lat_cmp = [&](T&a, T&b) { return lat_get(*a) < lat_get(*b); };

    if (from.size() <= n) {
        to.insert(to.end(),
                  std::make_move_iterator(from.begin()),
                  std::make_move_iterator(from.end()));
        if (time_get) {
            std::sort(to.begin(), to.end(), time_cmp);
        }
        return;
    }

    std::sort(from.begin(), from.end(), lat_cmp);
    int spacing = from.size() / n;
    if (spacing == 0) {
        spacing = 1;
    }
    to.push_back(std::move(from[0]));
    for (unsigned int i = spacing; i < from.size() - 1; i+=spacing) {
        to.push_back(std::move(from[i]));
    }
    to.push_back(std::move(from[from.size() - 1]));
    if (time_get) {
        std::sort(to.begin(), to.end(), time_cmp);
    }
}

#define SCHED_SIZE 50000000
class Client : public Worker {
    // Log stuff
    private: Histogram_t hist;

    // Workload stuff
    public: std::vector<ClientRequest *> requests;
    private: uint32_t send_offset = 0;

    // Schedules
    public: std::vector<std::unique_ptr<Schedule>> schedules;
    private: std::vector<tp>::iterator send_time_it;
    private: tp next_send_time;
    private: boost::chrono::nanoseconds start_offset;
    private: ClientRequest *next_cr;

    // Init params
    private: int max_concurrency = 1;

    public: Client(
        int m, std::vector<std::unique_ptr<Schedule>> &scheds,
        std::vector<ClientRequest *> &&reqs) :
        Worker(WorkerType::CLIENT), requests(reqs), max_concurrency(m) {
            for (auto &s: scheds) {
                schedules.push_back(std::move(s));
            }
        }

    public: ~Client() {
        for (size_t i = 0; i < schedules.size(); ++i) {
            auto &s = schedules[i];
            log_info(
                "Duration: %.02lfs -> Sent: %u, Received: %u,"
                " %u sent but not answered, %u behind schedule, "
                " %u skipped, %u events processed, %u send attempts",
                ns_diff(s->start_time, s->end_time) / 1e9,
                s->send_index + 1, s->recv_requests,
                (s->send_index + 1) - s->recv_requests,
                s->n_requests - (s->send_index + 1),
                s->n_skipped, s->ev_count, s->attempts
            );
        }
        for (ClientRequest *req: requests) {
            delete req;
        }
    }

    public: int setup() override {
        PSP_NOTNULL(EPERM, udp_ctx);
        return 0;
    }

    private: int send_request(ClientRequest *cr) {
        if (udp_ctx->push_head - udp_ctx->push_tail == OUTBOUND_Q_LEN) {
            return EAGAIN;
        }
        //PSP_DEBUG("Sending request " << cr->id);
        unsigned long payload = (unsigned long) cr->mbuf;
        udp_ctx->outbound_queue[udp_ctx->push_head++ & (OUTBOUND_Q_LEN - 1)] = payload;
        PSP_OK(udp_ctx->send_packets());
        cr->sending = rdtscp(NULL);
        return 0;
    }

    // To fill vtable entry
    private: int process_request(unsigned long payload) override {
        return ENOTSUP;
    }

    private: int dequeue(unsigned long *payload) {
        return 0;
    }

    private: void gen_next_send_time() {
        next_send_time = (*send_time_it + start_offset);
        send_time_it++;
    }

    private: int gen_next_req(Schedule *sched) {
        if (likely(next_cr == nullptr)) {
            next_cr = requests[sched->send_index + send_offset];
            next_cr->id ^= (worker_id << 27); // XOR should be equivalent to OR here?
            char *payload_addr = nullptr;
            PSP_NULL(EINVAL, next_cr->mbuf);
            if (unlikely(udp_ctx->get_mbuf(&next_cr->mbuf, &payload_addr) == ENOMEM))
                return EAGAIN;
            switch (sched->ptype) {
                case pkt_type::IX: {
                    IXMessage req = {
                        .type = static_cast<uint16_t>(next_cr->type),
                        .seq_num = 42,
                        .queue_length = {0, 0, 0},
                        .client_id = static_cast<uint16_t>(worker_id),
                        .req_id = next_cr->id,
                        .pkts_length = sizeof(IXMessage),
                        .runNs = next_cr->run_ns,
                        .genNs = 0
                    };
                    memcpy(payload_addr + NET_HDR_SIZE, static_cast<void *>(&req), sizeof(req));
                    reinterpret_cast<rte_mbuf *>(next_cr->mbuf)->l4_len = sizeof(req);
                    break;
                }
                case pkt_type::RAW: {
                    char *id_addr = payload_addr + NET_HDR_SIZE;
                    *reinterpret_cast<uint32_t *>(id_addr) = next_cr->id;
                    memset(id_addr + sizeof(uint32_t), 1, sched->pkt_size - sizeof(uint32_t));
                    reinterpret_cast<rte_mbuf *>(next_cr->mbuf)->l4_len = sched->pkt_size;
                    break;
                }
                case pkt_type::PSP_MB: {
                    char *id_addr = payload_addr + NET_HDR_SIZE;
                    *reinterpret_cast<uint32_t *>(id_addr) = next_cr->id;
                    char *type_addr = id_addr + sizeof(uint32_t);
                    *reinterpret_cast<uint32_t *>(type_addr) = static_cast<uint32_t>(next_cr->type);
                    char *req_addr = type_addr +  sizeof(uint32_t);
                    *reinterpret_cast<uint32_t *>(req_addr) = sizeof(uint32_t);
                    *reinterpret_cast<uint32_t *>(req_addr + sizeof(uint32_t)) = next_cr->run_ns;
                    reinterpret_cast<rte_mbuf *>(next_cr->mbuf)->l4_len = sizeof(uint32_t) * 4;
                    break;
                }
                default:
                    PSP_UNREACHABLE();
            }
        }
        return 0;
    }

    private: int work(int status, unsigned long payload) override {
        for (auto &s: schedules) {
            PSP_OK(play_schedule(s.get()));
        }
        terminate = true;
        return 0;
    }

    private: int play_schedule(Schedule *sched) {
        // Adjust start time
        sched->start_time = take_time();
        start_offset = boost::chrono::duration_cast<boost::chrono::nanoseconds>(
            sched->start_time - sched->send_times[0]
        );
        send_time_it = sched->send_times.begin();
        sched->end_time = sched->send_times.back() + start_offset;
        PSP_DEBUG("Start time: " << since_epoch(sched->start_time)/1e9 << " s");
        PSP_DEBUG("start offset: " << start_offset.count()/1e9 << " s");
        PSP_DEBUG("Schedule start time : " << (since_epoch(sched->send_times[0]) + start_offset.count())/1e9 << " s");
        PSP_DEBUG("Schedule end time : " << since_epoch(sched->end_time)/1e9 << " s");
        next_cr = nullptr;
        gen_next_send_time();
        bool sending_complete = false;
        tp iter_time = take_time();
        int status = EAGAIN;
        terminate = false;
        PSP_INFO("Running schedule for " << ns_diff(sched->send_times[0], sched->send_times.back()) / 1e9);
        while (sched->recv_requests < sched->n_requests && !terminate) {
            /* While we are not due a request, check for answers */
            status = EAGAIN;
            do {
                if ((iter_time - sched->start_time) >= sched->max_duration) {
                    PSP_INFO(
                        "Terminating the schedule because " \
                        "(iter_time - sched->start_time) >= sched->max_duration"
                    );
                    terminate = true;
                    break;
                }
                status = udp_ctx->recv_packets();
                iter_time = take_time();
                sched->ev_count++;
            } while (status == EAGAIN and iter_time < next_send_time);
            int inflight = sched->send_index - sched->recv_requests;
            /* While we are due a request */
            while (iter_time >= next_send_time and iter_time <= sched->end_time and sched->send_index < sched->n_requests) {
                sched->attempts++;
                if (max_concurrency < 0 or (max_concurrency > inflight)) {
                    if (gen_next_req(sched) == 0) {
                        PSP_NOTNULL(EINVAL, next_cr);
                        int rtn = send_request(next_cr);
                        if (rtn == 0) {
                            sched->send_index++;
                            // reset next_cr so we can generate a new one
                            next_cr = nullptr;
                        } else if (rtn == EAGAIN) {
                            // Push queue is full. Need to drain it.
                            sched->n_skipped++;
                            break;
                        } else {
                            PSP_UNREACHABLE();
                        }
                    } else {
                        // Likely no more mbufs available. Need to drain push queue.
                        sched->n_skipped++;
                        break;
                    }
                } else {
                    // Max concurrency reached
                    sched->n_skipped++;
                }
                gen_next_send_time();
                if (next_send_time >= sched->end_time) {
                    PSP_INFO("Sending complete");
                    sending_complete = true;
                }
                inflight = sched->send_index - sched->recv_requests;
            }
            if (inflight == 0 and (sending_complete or iter_time > sched->end_time)) {
                PSP_INFO(
                    "Terminating the schedule because " \
                    " sending complete or iter_time > sched->end_time"
                );
                terminate = true;
                continue;
            }
            /* We dequeued some answers from the NIC */
            if (status == 0) {
                PSP_DEBUG("Attempting to dequeue up to " << udp_ctx->pop_head - udp_ctx->pop_tail);
                uint64_t pop_time = rdtscp(NULL);
                size_t batch_dequeued = 0;
                while (udp_ctx->pop_head > udp_ctx->pop_tail and batch_dequeued < MAX_RX_BURST) {
                    unsigned long req = udp_ctx->inbound_queue[udp_ctx->pop_tail++ & (INBOUND_Q_LEN - 1)];
                    uint32_t rid;
                    switch (sched->ptype) {
                        case pkt_type::IX: {
                            IXMessage *msg = rte_pktmbuf_mtod_offset(
                                static_cast<rte_mbuf *>((void*)req), IXMessage *, NET_HDR_SIZE
                             );
                            rid = msg->req_id;
                            break;
                        }
                        default: {
                            rid = *rte_pktmbuf_mtod_offset(
                                static_cast<rte_mbuf *>((void*)req), uint32_t *, NET_HDR_SIZE
                             );
                        }
                    }
                    /*
                    std::cout << *(rte_ipv4_hdr*)(((rte_mbuf *)req)->buf_addr+128+14) << std::endl;
                    std::cout << *(rte_udp_hdr*)(((rte_mbuf *)req)->buf_addr+128+34) << std::endl;
                    */
                    auto &cr = requests[rid ^ (worker_id << 27)];
                    PSP_DEBUG(
                        "[client " << worker_id << "] received response to "
                        << rid << "(" << (rid ^ (worker_id << 27)) << ")"
                    );
                    cr->completed = pop_time;
                    PSP_OK(udp_ctx->free_mbuf(req));
                    batch_dequeued++;
                }
                sched->recv_requests += batch_dequeued;
            }
        }
        if (sending_complete) {
            sched->last_send_time = sched->end_time;
        } else {
            sched->last_send_time = *send_time_it + start_offset;
        }
        sched->end_time = take_time();
        send_offset += sched->n_requests;
        PSP_INFO("Ran schedule for " << ns_diff(sched->start_time, sched->end_time) / 1e9);
        return 0;
    }

    public: int log_latency(std::ostream &output, std::ostream &hist_output, int downsample) {
        if (requests.size() == 0) {
            return ENOENT;
        }
        if (downsample > 0) {
            std::unordered_map<enum ReqType, std::vector<ClientRequest *>> requests_by_type{};
            for (ClientRequest *req: requests) {
                if (requests_by_type.find(req->type) == requests_by_type.end()) {
                    requests_by_type[req->type] = std::vector<ClientRequest *>{};
                }
                requests_by_type[req->type].push_back(req);
            }
            std::vector<ClientRequest *> downsampled_requests;
            for (auto &reqtype : requests_by_type) {
                sample_into(reqtype.second, downsampled_requests,
                            req_latency, req_time, downsample);
            }
            for (auto &req : downsampled_requests) {
                output << worker_id << "\t" << *req << std::endl;
            }
        } else if (downsample == 0) {
            for (ClientRequest *req: requests) {
                output << worker_id << "\t" << *req << std::endl;
            }
        } else if (downsample == -1) {
            // Discard first 10%
            std::vector<ClientRequest *> pruned_requests(requests.begin() + requests.size() * 0.1, requests.end());
            //std::vector<ClientRequest *> pruned_requests(requests.begin()+1e5, requests.end());

            // Short all requests by response time
            auto lat_cmp = [](ClientRequest *a, ClientRequest *b) { return req_latency(*a) < req_latency(*b); };
            std::sort(pruned_requests.begin(), pruned_requests.end(), lat_cmp);

            // Histograms
            PSP_INFO("Processing " << pruned_requests.size() << " samples");
            hist.buckets.clear(); // Not sure why I have to clean it there?
            hist.min = 0; hist.max = 0; hist.count = 0; hist.total = 0;
            for (uint64_t i = 0; i < pruned_requests.size(); ++i) {
                if (pruned_requests[i]->completed == 0) {
                    continue;
                }
                // Store values in nanoseconds
                insert_value(&hist, (pruned_requests[i]->completed - pruned_requests[i]->sending) / cycles_per_ns);
            }
            if (hist.count == 0) {
                PSP_WARN("No values inserted in histogram");
                return -1;
            }
            // First line is histo for all samples
            hist_output << "TYPE\tMIN\tMAX\tCOUNT\tTOTAL";
            for (uint64_t i = 0; i < hist.buckets.size(); ++i) {
                if (hist.buckets[i] > 0) {
                    hist_output << "\t" << i * 1000;
                }
            }
            hist_output << std::endl;
            hist_output << "UNKNOWN\t"
                        << hist.min << "\t"
                        << hist.max << "\t"
                        << hist.count << "\t"
                        << hist.total;
            for (unsigned i = 0; i < hist.buckets.size(); ++i) {
                if (hist.buckets[i] > 0) {
                    hist_output << "\t" << hist.buckets[i];
                }
            }
            hist_output << std::endl;
            check_hist(pruned_requests, hist.count - 1);
            // Then one line per request type
            std::unordered_map<enum ReqType, std::vector<ClientRequest *>> requests_by_type{};
            for (uint64_t i = 0; i < pruned_requests.size(); ++i) {
                if (pruned_requests[i]->completed == 0) {
                    continue;
                }
                auto &req_ptr = pruned_requests[i];
                if (requests_by_type.find(req_ptr->type) == requests_by_type.end()) {
                    requests_by_type[req_ptr->type] = std::vector<ClientRequest *>{};
                }
                requests_by_type[req_ptr->type].push_back(req_ptr);
            }
            for (auto &rtype: requests_by_type) {
                hist.buckets.clear();
                hist.min = 0; hist.max = 0; hist.count = 0; hist.total = 0;
                auto reqs = rtype.second;
                for (uint64_t i = 0; i < reqs.size(); ++i) {
                    // Store values in nanoseconds
                    insert_value(&hist, (reqs[i]->completed - reqs[i]->sending) / cycles_per_ns);
                }
		if (hist.count == 0) {
		    PSP_WARN(
			"No values inserted in histogram for type "
                        << req_type_str[static_cast<int>(rtype.first)]
		    );
                    continue;
		}
                hist_output << "TYPE\tMIN\tMAX\tCOUNT\tTOTAL";
                for (uint64_t i = 0; i < hist.buckets.size(); ++i) {
                    if (hist.buckets[i] > 0) {
                        hist_output << "\t" << i * 1000;
                    }
                }
                hist_output << std::endl;
                hist_output << req_type_str[static_cast<int>(rtype.first)] << "\t"
                            << hist.min << "\t"
                            << hist.max << "\t"
                            << hist.count << "\t"
                            << hist.total;
                for (unsigned i = 0; i < hist.buckets.size(); ++i) {
                    if (hist.buckets[i] > 0) {
                        hist_output << "\t" << hist.buckets[i];
                    }
                }
                hist_output << std::endl;
                PSP_TRUE(EINVAL, hist.count == reqs.size());

                check_hist(reqs, reqs.size() - 1);
            }
        }
        return 0;
    }

    public: int log_throughput(std::ostream &output, long unsigned int epoch_size_ns, uint64_t first_cplted) {
        std::unordered_map<std::pair<enum ReqType, int>,
                           long unsigned int,
                           boost::hash<std::pair<enum ReqType, int>>> type_epoch_counts;
        for (ClientRequest *req: requests) {
            if (req->completed == 0)
                continue;
            auto type_epoch = std::pair(req->type, ((req->completed / cycles_per_ns) - first_cplted) / epoch_size_ns);
            if (type_epoch_counts.find(type_epoch) == type_epoch_counts.end()) {
                type_epoch_counts[type_epoch] = 0;
            }
            type_epoch_counts[type_epoch]++;
        }
        std::cout << " identified " << type_epoch_counts.size() << " epochs" << std::endl;
        for (auto &it : type_epoch_counts) {
            output << worker_id << "\t"
                   << (it.first.second * epoch_size_ns) << "\t"
                   << req_type_str[static_cast<int>(it.first.first)] << "\t"
                   << it.second << "\n";
        }
        return 0;
    }

    public: static uint64_t req_latency(const ClientRequest &req) {
        if (req.completed == 0) {
            return LLONG_MAX;
        } else {
            return (req.completed - req.sending) / cycles_per_ns;
        }
    }

    public: static uint64_t req_time(const ClientRequest &req) {
        return req.sending / cycles_per_ns;
    }

    private: int check_hist(std::vector<ClientRequest *> &reqs, uint64_t nr) {
        std::cout << "======================================" << std::endl;
        std::cout << "Hist min: " << hist.min
                  << ". Min: " << (reqs[0]->completed - reqs[0]->sending) / cycles_per_ns << std::endl;
        uint32_t c = 0;
        for (uint32_t i = 0; i < hist.buckets.size(); ++i) {
            if (hist.buckets[i] == 0)
                continue;
            if (c + hist.buckets[i] >= hist.count * .25 and c < hist.count * .25) {
                std::cout << std::fixed << "Hist 25th: " <<  (i*1000 + (i+1)*1000) / 2.0
                          << ". 25th: "
                          << (reqs[nr*25/100]->completed - reqs[nr*25/100]->sending) / cycles_per_ns
                          << std::endl;
            }
            if (c + hist.buckets[i] >= hist.count * .5 and c < hist.count * .5) {
                std::cout << std::fixed << "Hist 50th: " <<  (i*1000 + (i+1)*1000) / 2.0
                          << ". 50th: "
                          << (reqs[nr/2]->completed - reqs[nr/2]->sending) / cycles_per_ns
                          << std::endl;
            }
            if (c + hist.buckets[i] >= hist.count * .75 and c < hist.count * .75) {
                std::cout << std::fixed << "Hist 75th: " <<  (i*1000 + (i+1)*1000) / 2.0
                //std::cout << "Hist 75th: " << ((1UL << i) + (1UL << (i-1))) / 2.0
                          << ". 75th: "
                          << (reqs[nr*75/100]->completed - reqs[nr*75/100]->sending) / cycles_per_ns
                          << std::endl;
            }
            if (c + hist.buckets[i] >= hist.count * .9 and c < hist.count * .9) {
                std::cout << std::fixed << "Hist 90th: " <<  (i*1000 + (i+1)*1000) / 2.0
                          << ". 90th: "
                          << (reqs[nr*90/100]->completed - reqs[nr*90/100]->sending) / cycles_per_ns
                          << std::endl;
            }
            if (c + hist.buckets[i] >= hist.count * .99 and c < hist.count * .99) {
                std::cout << std::fixed << "Hist 99th: " <<  (i*1000 + (i+1)*1000) / 2.0
                          << ". 99th: "
                          << (reqs[nr*99/100]->completed - reqs[nr*99/100]->sending) / cycles_per_ns
                          << std::endl;
            }
            if (c + hist.buckets[i] >= hist.count * .999 and c < hist.count * .999) {
                std::cout << std::fixed << "Hist 99.9th: " <<  (i*1000 + (i+1)*1000) / 2.0
                          << ". 99.9th: "
                          << (reqs[nr*999/1000]->completed - reqs[nr*999/1000]->sending) / cycles_per_ns
                          << std::endl;
            }
            if (c + hist.buckets[i] >= hist.count * .9999 and c < hist.count * .9999) {
                std::cout << std::fixed << "Hist 99.99th: " <<  (i*1000 + (i+1)*1000) / 2.0
                          << ". 99.99th: "
                          << (reqs[nr*9999/10000]->completed - reqs[nr*9999/10000]->sending) / cycles_per_ns
                          << std::endl;
            }
            c += hist.buckets[i];
        }
        PSP_TRUE(EINVAL, c == hist.count);

        std::cout << "Hist max: " << hist.max
                  << ". Max: " << (reqs[nr]->completed - reqs[nr]->sending) / cycles_per_ns << std::endl;
        std::cout << "Hist count: " << c << ". Num requests: " << nr+1 << std::endl;
        std::cout << "======================================" << std::endl;
        return 0;
    }
};
#endif //CLIENT_H_

ext: hh

max_stars_repo_name: maxdml/Pers-phone

max_issues_repo_name: maxdml/psp

max_forks_repo_name: maxdml/psp