#include "tracing/node_trace_writer.h"

#include "util-inl.h"

#include <fcntl.h>

#include <cstring>

namespace node {

namespace tracing {

NodeTraceWriter::NodeTraceWriter(const std::string& log_file_pattern)

    : log_file_pattern_(log_file_pattern) {}

void NodeTraceWriter::InitializeOnThread(uv_loop_t* loop) {

  CHECK_NULL(tracing_loop_);

  tracing_loop_ = loop;

  flush_signal_.data = this;

  int err = uv_async_init(tracing_loop_, &flush_signal_,

                          [](uv_async_t* signal) {

    NodeTraceWriter* trace_writer =

        ContainerOf(&NodeTraceWriter::flush_signal_, signal);

    trace_writer->FlushPrivate();

  });

  CHECK_EQ(err, 0);

  exit_signal_.data = this;

  err = uv_async_init(tracing_loop_, &exit_signal_, ExitSignalCb);

  CHECK_EQ(err, 0);

}

void NodeTraceWriter::WriteSuffix() {

  // If our final log file has traces, then end the file appropriately.

  // This means that if no trace events are recorded, then no trace file is

  // produced.

  bool should_flush = false;

  {

    Mutex::ScopedLock scoped_lock(stream_mutex_);

    if (total_traces_ > 0) {

      total_traces_ = kTracesPerFile;  // Act as if we reached the file limit.

      should_flush = true;

    }

  }

  if (should_flush) {

    Flush(true);

  }

}

NodeTraceWriter::~NodeTraceWriter() {

  WriteSuffix();

  uv_fs_t req;

  if (fd_ != -1) {

    CHECK_EQ(0, uv_fs_close(nullptr, &req, fd_, nullptr));

    uv_fs_req_cleanup(&req);

  }

  uv_async_send(&exit_signal_);

  Mutex::ScopedLock scoped_lock(request_mutex_);

  while (!exited_) {

    exit_cond_.Wait(scoped_lock);

  }

}

void replace_substring(std::string* target,

                       const std::string& search,

                       const std::string& insert) {

  size_t pos = target->find(search);

  for (; pos != std::string::npos; pos = target->find(search, pos)) {

    target->replace(pos, search.size(), insert);

    pos += insert.size();

  }

}

void NodeTraceWriter::OpenNewFileForStreaming() {

  ++file_num_;

  uv_fs_t req;

  // Evaluate a JS-style template string, it accepts the values ${pid} and

  // ${rotation}

  std::string filepath(log_file_pattern_);

  replace_substring(&filepath, "${pid}", std::to_string(uv_os_getpid()));

  replace_substring(&filepath, "${rotation}", std::to_string(file_num_));

  if (fd_ != -1) {

    CHECK_EQ(uv_fs_close(nullptr, &req, fd_, nullptr), 0);

    uv_fs_req_cleanup(&req);

  }

  fd_ = uv_fs_open(nullptr, &req, filepath.c_str(),

      O_CREAT | O_WRONLY | O_TRUNC, 0644, nullptr);

  uv_fs_req_cleanup(&req);

  if (fd_ < 0) {

    fprintf(stderr, "Could not open trace file %s: %s\n",

                    filepath.c_str(),

                    uv_strerror(fd_));

    fd_ = -1;

  }

}

void NodeTraceWriter::AppendTraceEvent(TraceObject* trace_event) {

  Mutex::ScopedLock scoped_lock(stream_mutex_);

  // If this is the first trace event, open a new file for streaming.

  if (total_traces_ == 0) {

    OpenNewFileForStreaming();

    // Constructing a new JSONTraceWriter object appends "{\"traceEvents\":["

    // to stream_.

    // In other words, the constructor initializes the serialization stream

    // to a state where we can start writing trace events to it.

    // Repeatedly constructing and destroying json_trace_writer_ allows

    // us to use V8's JSON writer instead of implementing our own.

    json_trace_writer_.reset(TraceWriter::CreateJSONTraceWriter(stream_));

  }

  ++total_traces_;

  json_trace_writer_->AppendTraceEvent(trace_event);

}

void NodeTraceWriter::FlushPrivate() {

  std::string str;

  int highest_request_id;

  {

    Mutex::ScopedLock stream_scoped_lock(stream_mutex_);

    if (total_traces_ >= kTracesPerFile) {

      total_traces_ = 0;

      // Destroying the member JSONTraceWriter object appends "]}" to

      // stream_ - in other words, ending a JSON file.

      json_trace_writer_.reset();

    }

    // str() makes a copy of the contents of the stream.

    str = stream_.str();

    stream_.str("");

    stream_.clear();

  }

  {

    Mutex::ScopedLock request_scoped_lock(request_mutex_);

    highest_request_id = num_write_requests_;

  }

  WriteToFile(std::move(str), highest_request_id);

}

void NodeTraceWriter::Flush(bool blocking) {

  Mutex::ScopedLock scoped_lock(request_mutex_);

  {

    // We need to lock the mutexes here in a nested fashion; stream_mutex_

    // protects json_trace_writer_, and without request_mutex_ there might be

    // a time window in which the stream state changes?

    Mutex::ScopedLock stream_mutex_lock(stream_mutex_);

    if (!json_trace_writer_)

      return;

  }

  int request_id = ++num_write_requests_;

  int err = uv_async_send(&flush_signal_);

  CHECK_EQ(err, 0);

  if (blocking) {

    // Wait until data associated with this request id has been written to disk.

    // This guarantees that data from all earlier requests have also been

    // written.

    while (request_id > highest_request_id_completed_) {

      request_cond_.Wait(scoped_lock);

    }

  }

}

void NodeTraceWriter::WriteToFile(std::string&& str, int highest_request_id) {

  if (fd_ == -1) return;

  uv_buf_t buf = uv_buf_init(nullptr, 0);

  {

    Mutex::ScopedLock lock(request_mutex_);

    write_req_queue_.emplace(WriteRequest {

      std::move(str), highest_request_id

    });

    if (write_req_queue_.size() == 1) {

      buf = uv_buf_init(

          const_cast<char*>(write_req_queue_.front().str.c_str()),

          write_req_queue_.front().str.length());

    }

  }

  // Only one write request for the same file descriptor should be active at

  // a time.

  if (buf.base != nullptr && fd_ != -1) {

    StartWrite(buf);

  }

}

void NodeTraceWriter::StartWrite(uv_buf_t buf) {

  int err = uv_fs_write(

      tracing_loop_, &write_req_, fd_, &buf, 1, -1,

      [](uv_fs_t* req) {

        NodeTraceWriter* writer =

            ContainerOf(&NodeTraceWriter::write_req_, req);

        writer->AfterWrite();

      });

  CHECK_EQ(err, 0);

}

void NodeTraceWriter::AfterWrite() {

  CHECK_GE(write_req_.result, 0);

  uv_fs_req_cleanup(&write_req_);

  uv_buf_t buf = uv_buf_init(nullptr, 0);

  {

    Mutex::ScopedLock scoped_lock(request_mutex_);

    int highest_request_id = write_req_queue_.front().highest_request_id;

    write_req_queue_.pop();

    highest_request_id_completed_ = highest_request_id;

    request_cond_.Broadcast(scoped_lock);

    if (!write_req_queue_.empty()) {

      buf = uv_buf_init(

          const_cast<char*>(write_req_queue_.front().str.c_str()),

          write_req_queue_.front().str.length());

    }

  }

  if (buf.base != nullptr && fd_ != -1) {

    StartWrite(buf);

  }

}

// static

void NodeTraceWriter::ExitSignalCb(uv_async_t* signal) {

  NodeTraceWriter* trace_writer =

      ContainerOf(&NodeTraceWriter::exit_signal_, signal);

  // Close both flush_signal_ and exit_signal_.

  uv_close(reinterpret_cast<uv_handle_t*>(&trace_writer->flush_signal_),

           [](uv_handle_t* signal) {

             NodeTraceWriter* trace_writer =

                 ContainerOf(&NodeTraceWriter::flush_signal_,

                             reinterpret_cast<uv_async_t*>(signal));

             uv_close(

                 reinterpret_cast<uv_handle_t*>(&trace_writer->exit_signal_),

                 [](uv_handle_t* signal) {

                   NodeTraceWriter* trace_writer =

                       ContainerOf(&NodeTraceWriter::exit_signal_,

                                   reinterpret_cast<uv_async_t*>(signal));

                   Mutex::ScopedLock scoped_lock(trace_writer->request_mutex_);

                   trace_writer->exited_ = true;

                   trace_writer->exit_cond_.Signal(scoped_lock);

                 });

           });

}

}  // namespace tracing

}  // namespace node