<commit_before>/*
 * Copyright (c) Meta Platforms, Inc. and affiliates.
 *
 * This source code is licensed under the MIT license found in the
 * LICENSE file in the root directory of this source tree.
 */

#include "ConnectionFifo.h"

#include <chrono>

#include <folly/Range.h>

namespace facebook {
namespace memcache {

namespace {

uint64_t timeSinceEpoch() {
  using namespace std::chrono;
  return duration_cast<microseconds>(steady_clock::now().time_since_epoch())
      .count();
}

class PipeIov {
 public:
  const iovec* iov() const {
    return iov_;
  }
  size_t size() const {
    return index_;
  }
  bool full() const {
    return index_ >= kMaxLen || remSize_ == 0;
  }

  void reset() {
    index_ = 0;
    remSize_ = PIPE_BUF;
  }
  size_t append(void* buf, size_t len) {
    assert(!full());

    size_t numBytes = std::min(len, remSize_);
    iov_[index_].iov_base = buf;
    iov_[index_].iov_len = numBytes;
    remSize_ -= numBytes;
    ++index_;
    return numBytes;
  }

 private:
  static constexpr size_t kMaxLen = 16;
  iovec iov_[kMaxLen];
  size_t index_ = 0;
  size_t remSize_ = PIPE_BUF;

  static_assert(
      IOV_MAX >= kMaxLen,
      "IOV_MAX must be larger than PipeIov::kMaxLen.");
};

/**
 * Helper to keep state of iteration over const iovec.
 */
class IovecIterator {
 public:
  IovecIterator(const struct iovec* iov, size_t iovcnt)
      : iov_(iov), iovcnt_(iovcnt) {
    assert(iovcnt > 0);

    curBuf_ = reinterpret_cast<char*>(iov_[0].iov_base);
    curBufLen_ = iov_[0].iov_len;
  }

  char* currentBuffer() const {
    return curBuf_;
  }
  size_t currentBufferLength() const {
    return curBufLen_;
  }
  bool hasData() const {
    return index_ < iovcnt_;
  }

  void advance(size_t bytes) {
    assert(hasData());
    assert(bytes <= curBufLen_);

    if (bytes < curBufLen_) {
      curBuf_ += bytes;
      curBufLen_ -= bytes;
    } else { // (bytes == bufLen)
      ++index_;
      if (hasData()) {
        curBuf_ = reinterpret_cast<char*>(iov_[index_].iov_base);
        curBufLen_ = iov_[index_].iov_len;
      }
    }
  }

 private:
  const struct iovec* iov_;
  const size_t iovcnt_;

  size_t index_{0};
  char* curBuf_{nullptr};
  size_t curBufLen_{0};
};

MessageHeader buildMsgHeader(
    const folly::AsyncTransportWrapper* transport,
    folly::StringPiece routerName) {
  MessageHeader header;
  header.setConnectionId(reinterpret_cast<uintptr_t>(transport));

  if (!transport) {
    return header;
  }

  try {
    folly::SocketAddress address;

    transport->getPeerAddress(&address);
    if (address.getFamily() == AF_INET || address.getFamily() == AF_INET6) {
      address.getAddressStr(
          header.peerAddressModifiable(), MessageHeader::kAddressMaxSize);
      header.setPeerPort(address.getPort());
    } else if (address.getFamily() == AF_UNIX) {
      // For unix sockets, just localAddress has the path.
      transport->getLocalAddress(&address);
      std::snprintf(
          header.peerAddressModifiable(),
          MessageHeader::kAddressMaxSize,
          "%s%s",
          kUnixSocketPrefix.data(),
          address.getPath().c_str());
    }

    transport->getLocalAddress(&address);
    header.setLocalPort(address.getPort());
  } catch (const std::exception& e) {
    VLOG(2) << "Error getting host/port to write to debug fifo: " << e.what();
  }

  // set router name
  header.routerNameModifiable()[0] = '\0';
  if (!routerName.empty()) {
    int res = std::snprintf(
        header.routerNameModifiable(),
        MessageHeader::kRouterNameMaxSize,
        "%s",
        routerName.str().c_str());
    if (res < 0) {
      LOG(ERROR) << "Error writing router name '" << routerName
                 << "' to debug fifo";
    }<commit_msg>Fix error setting transport local address in debug fifos when using Unix socket listeners (#388)<commit_after>/*
 * Copyright (c) Meta Platforms, Inc. and affiliates.
 *
 * This source code is licensed under the MIT license found in the
 * LICENSE file in the root directory of this source tree.
 */

#include "ConnectionFifo.h"

#include <chrono>

#include <folly/Range.h>

namespace facebook {
namespace memcache {

namespace {

uint64_t timeSinceEpoch() {
  using namespace std::chrono;
  return duration_cast<microseconds>(steady_clock::now().time_since_epoch())
      .count();
}

class PipeIov {
 public:
  const iovec* iov() const {
    return iov_;
  }
  size_t size() const {
    return index_;
  }
  bool full() const {
    return index_ >= kMaxLen || remSize_ == 0;
  }

  void reset() {
    index_ = 0;
    remSize_ = PIPE_BUF;
  }
  size_t append(void* buf, size_t len) {
    assert(!full());

    size_t numBytes = std::min(len, remSize_);
    iov_[index_].iov_base = buf;
    iov_[index_].iov_len = numBytes;
    remSize_ -= numBytes;
    ++index_;
    return numBytes;
  }

 private:
  static constexpr size_t kMaxLen = 16;
  iovec iov_[kMaxLen];
  size_t index_ = 0;
  size_t remSize_ = PIPE_BUF;

  static_assert(
      IOV_MAX >= kMaxLen,
      "IOV_MAX must be larger than PipeIov::kMaxLen.");
};

/**
 * Helper to keep state of iteration over const iovec.
 */
class IovecIterator {
 public:
  IovecIterator(const struct iovec* iov, size_t iovcnt)
      : iov_(iov), iovcnt_(iovcnt) {
    assert(iovcnt > 0);

    curBuf_ = reinterpret_cast<char*>(iov_[0].iov_base);
    curBufLen_ = iov_[0].iov_len;
  }

  char* currentBuffer() const {
    return curBuf_;
  }
  size_t currentBufferLength() const {
    return curBufLen_;
  }
  bool hasData() const {
    return index_ < iovcnt_;
  }

  void advance(size_t bytes) {
    assert(hasData());
    assert(bytes <= curBufLen_);

    if (bytes < curBufLen_) {
      curBuf_ += bytes;
      curBufLen_ -= bytes;
    } else { // (bytes == bufLen)
      ++index_;
      if (hasData()) {
        curBuf_ = reinterpret_cast<char*>(iov_[index_].iov_base);
        curBufLen_ = iov_[index_].iov_len;
      }
    }
  }

 private:
  const struct iovec* iov_;
  const size_t iovcnt_;

  size_t index_{0};
  char* curBuf_{nullptr};
  size_t curBufLen_{0};
};

MessageHeader buildMsgHeader(
    const folly::AsyncTransportWrapper* transport,
    folly::StringPiece routerName) {
  MessageHeader header;
  header.setConnectionId(reinterpret_cast<uintptr_t>(transport));

  if (!transport) {
    return header;
  }

  try {
    folly::SocketAddress address;

    transport->getPeerAddress(&address);
    if (address.getFamily() == AF_INET || address.getFamily() == AF_INET6) {
      address.getAddressStr(
          header.peerAddressModifiable(), MessageHeader::kAddressMaxSize);
      header.setPeerPort(address.getPort());
      header.setLocalPort(address.getPort());
    } else if (address.getFamily() == AF_UNIX) {
      // For unix sockets, just localAddress has the path.
      transport->getLocalAddress(&address);
      std::snprintf(
          header.peerAddressModifiable(),
          MessageHeader::kAddressMaxSize,
          "%s%s",
          kUnixSocketPrefix.data(),
          address.getPath().c_str());
    }

    transport->getLocalAddress(&address);
  } catch (const std::exception& e) {
    VLOG(2) << "Error getting host/port to write to debug fifo: " << e.what();
  }

  // set router name
  header.routerNameModifiable()[0] = '\0';
  if (!routerName.empty()) {
    int res = std::snprintf(
        header.routerNameModifiable(),
        MessageHeader::kRouterNameMaxSize,
        "%s",
        routerName.str().c_str());
    if (res < 0) {
      LOG(ERROR) << "Error writing router name '" << routerName
                 << "' to debug fifo";
    }
{'fraction_non_alphanumeric': 0.12868668165586564, 'fraction_numerical': 0.004893532601507737, 'mean_word_length': 3.067778375470683, 'pattern_counts': {'":': 0, '<': 43, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '113944', 'n_tokens_mistral': 2558, 'n_tokens_neox': 2525, 'n_words': 586}