<commit_before>/* Copyright 2018 The TensorFlow Authors. All Rights Reserved.

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 "tensorflow/python/client/session_ref.h"

#include <stdlib.h>
#include <memory>
#include <utility>

#include "tensorflow/core/lib/io/path.h"
#include "tensorflow/core/lib/io/record_writer.h"
#include "tensorflow/core/lib/strings/stringprintf.h"
#include "tensorflow/core/protobuf/master.pb.h"
#include "tensorflow/core/protobuf/named_tensor.pb.h"
#include "tensorflow/core/protobuf/replay_log.pb.h"

namespace tensorflow {

namespace {

// Scope helper to track active calls and manage session lifetime.
// SessionRef blocks closing until all active calls complete or are cancelled.
struct RunCounter {
  std::shared_ptr<Session> session;
  uint64* value;
  mutex* m;
  condition_variable* cv;

  explicit RunCounter(std::shared_ptr<Session> s, uint64* v, mutex* m,
                      condition_variable* cv)
      : session(std::move(s)), value(v), m(m), cv(cv) {
    mutex_lock l(*m);
    ++*value;
  }

  ~RunCounter() {
    mutex_lock l(*m);
    if (--*value == 0) {
      cv->notify_all();
    }
  }
};

std::string SessionToHandle(Session* session) {
  return strings::Printf("%llu", reinterpret_cast<uint64>(session));
}

// The Session interface has many methods of the form:
//
// X(a, b);
// X(RunOptions, a, b);
//
// Not all sessions support the second case (with an empty RunOptions()).
// We use this variable as a sentinel to dispatch to the correct call.
RunOptions* kEmptyRunOptions() {
  static RunOptions* options = new RunOptions();
  return options;
}

}  // namespace

// Run the given session operation, recording start and end timestamps.
// If the operation returns a bad status, return after flushing the current
// log request.  This should be run _after_ all request information has been
// added to the current op.
#define RUN_WITH_TIMESTAMP(OpName, ...)              \
  op.set_start_time_us(Env::Default()->NowMicros()); \
  Status status = session->OpName(__VA_ARGS__);      \
  op.set_end_time_us(Env::Default()->NowMicros());   \
  if (!status.ok()) {                                \
    Flush(op).IgnoreError();                         \
    return status;                                   \
  }

// Records requests (and optionally responses) performed against a session.
// The resulting replay log can be used with the `tf_replay` tool to replicate
// the operations against a simulated environment, without requiring the
// original code or cluster setup.
//
// Session logging by setting the TF_REPLAY_LOG_FILE environment variable.
class SessionLogger {
 public:
  SessionLogger() {
    std::string log_name = getenv("TF_REPLAY_LOG_FILE");
    LOG(INFO) << "Constructing new session logger for " << log_name;
    TF_CHECK_OK(
        Env::Default()->RecursivelyCreateDir(string(io::Dirname(log_name))));
    Env::Default()->DeleteFile(log_name).IgnoreError();

    TF_CHECK_OK(Env::Default()->NewWritableFile(log_name, &log_file_));
    log_writer_ = absl::make_unique<io::RecordWriter>(log_file_.get());
  }

  ~SessionLogger() {
    log_writer_->Close().IgnoreError();
    log_writer_.release();
    log_file_->Close().IgnoreError();
  }

  Status RecordNewSession(Session* session) {
    LOG(INFO) << "New session discovered.  Capturing devices...";
    ReplayOp op;
    NewReplaySession* req = op.mutable_new_replay_session();

    std::vector<DeviceAttributes> devices;
    Status status = session->ListDevices(&devices);
    if (status.ok()) {
      LOG(INFO) << "Found: " << devices.size() << " devices.";
      for (const DeviceAttributes& dev : devices) {
        *req->mutable_devices()->add_local_device() = dev;
      }
    } else {
      LOG(WARNING) << "Failed to list devices on session. Continuing.";
    }

    req->set_session_handle(SessionToHandle(session));
    return Flush(op);
  }

  Status RecordRun(Session* session,
                   const std::vector<std::pair<string, Tensor> >& inputs,
                   const std::vector<string>& output_tensor_names,
                   const std::vector<string>& target_node_names,
                   std::vector<Tensor>* outputs) {
    return RecordRun(session, *kEmptyRunOptions(), inputs, output_tensor_names,
                     target_node_names, outputs, nullptr);<commit_msg>Remove device capture from replay logs.<commit_after>/* Copyright 2018 The TensorFlow Authors. All Rights Reserved.

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 "tensorflow/python/client/session_ref.h"

#include <stdlib.h>
#include <memory>
#include <utility>

#include "tensorflow/core/lib/io/path.h"
#include "tensorflow/core/lib/io/record_writer.h"
#include "tensorflow/core/lib/strings/stringprintf.h"
#include "tensorflow/core/protobuf/master.pb.h"
#include "tensorflow/core/protobuf/named_tensor.pb.h"
#include "tensorflow/core/protobuf/replay_log.pb.h"

namespace tensorflow {

namespace {

// Scope helper to track active calls and manage session lifetime.
// SessionRef blocks closing until all active calls complete or are cancelled.
struct RunCounter {
  std::shared_ptr<Session> session;
  uint64* value;
  mutex* m;
  condition_variable* cv;

  explicit RunCounter(std::shared_ptr<Session> s, uint64* v, mutex* m,
                      condition_variable* cv)
      : session(std::move(s)), value(v), m(m), cv(cv) {
    mutex_lock l(*m);
    ++*value;
  }

  ~RunCounter() {
    mutex_lock l(*m);
    if (--*value == 0) {
      cv->notify_all();
    }
  }
};

std::string SessionToHandle(Session* session) {
  return strings::Printf("%llu", reinterpret_cast<uint64>(session));
}

// The Session interface has many methods of the form:
//
// X(a, b);
// X(RunOptions, a, b);
//
// Not all sessions support the second case (with an empty RunOptions()).
// We use this variable as a sentinel to dispatch to the correct call.
RunOptions* kEmptyRunOptions() {
  static RunOptions* options = new RunOptions();
  return options;
}

}  // namespace

// Run the given session operation, recording start and end timestamps.
// If the operation returns a bad status, return after flushing the current
// log request.  This should be run _after_ all request information has been
// added to the current op.
#define RUN_WITH_TIMESTAMP(OpName, ...)              \
  op.set_start_time_us(Env::Default()->NowMicros()); \
  Status status = session->OpName(__VA_ARGS__);      \
  op.set_end_time_us(Env::Default()->NowMicros());   \
  if (!status.ok()) {                                \
    Flush(op).IgnoreError();                         \
    return status;                                   \
  }

// Records requests (and optionally responses) performed against a session.
// The resulting replay log can be used with the `tf_replay` tool to replicate
// the operations against a simulated environment, without requiring the
// original code or cluster setup.
//
// Session logging by setting the TF_REPLAY_LOG_FILE environment variable.
class SessionLogger {
 public:
  SessionLogger() {
    std::string log_name = getenv("TF_REPLAY_LOG_FILE");
    LOG(INFO) << "Constructing new session logger for " << log_name;
    TF_CHECK_OK(
        Env::Default()->RecursivelyCreateDir(string(io::Dirname(log_name))));
    Env::Default()->DeleteFile(log_name).IgnoreError();

    TF_CHECK_OK(Env::Default()->NewWritableFile(log_name, &log_file_));
    log_writer_ = absl::make_unique<io::RecordWriter>(log_file_.get());
  }

  ~SessionLogger() {
    log_writer_->Close().IgnoreError();
    log_writer_.release();
    log_file_->Close().IgnoreError();
  }

  Status RecordNewSession(Session* session) {
    ReplayOp op;
    NewReplaySession* req = op.mutable_new_replay_session();
    req->set_session_handle(SessionToHandle(session));
    return Flush(op);
  }

  Status RecordRun(Session* session,
                   const std::vector<std::pair<string, Tensor> >& inputs,
                   const std::vector<string>& output_tensor_names,
                   const std::vector<string>& target_node_names,
                   std::vector<Tensor>* outputs) {
    return RecordRun(session, *kEmptyRunOptions(), inputs, output_tensor_names,
                     target_node_names, outputs, nullptr);
{'fraction_non_alphanumeric': 0.13344034242910646, 'fraction_numerical': 0.0032102728731942215, 'mean_word_length': 3.5084418716835506, 'pattern_counts': {'":': 0, '<': 46, '<?xml version=': 0, '>': 51, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3855970', 'n_tokens_mistral': 2720, 'n_tokens_neox': 2637, 'n_words': 781}