[devtool] create stream_data_sink

Pull Request resolved: #8604

this diff create `StreamDataSink`, a datasink that adopts array as buffer and flush to disk when buffer is full.
ghstack-source-id: 268188577
@exported-using-ghexport

Differential Revision: [D69936705](https://our.internmc.facebook.com/intern/diff/D69936705/)

Author: Gasoonjia

devtools/etdump/data_sinks/stream_data_sink.cpp

@@ -0,0 +1,125 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ * All rights reserved.
+ *
+ * This source code is licensed under the BSD-style license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+#include <executorch/devtools/etdump/data_sinks/stream_data_sink.h>
+#include <executorch/devtools/etdump/utils.h>
+#include <fcntl.h> // open()
+#include <unistd.h> // For write and close
+#include <cstring>
+
+using ::executorch::runtime::Error;
+using ::executorch::runtime::Result;
+
+namespace executorch {
+namespace etdump {
+
+StreamDataSink::StreamDataSink(StreamDataSink&& other) noexcept
+    : buffer_(std::move(other.buffer_)),
+      buffer_offset_(other.buffer_offset_),
+      file_descriptor_(other.file_descriptor_),
+      total_written_bytes_(other.total_written_bytes_),
+      alignment_(other.alignment_) {
+  other.file_descriptor_ = -1;
+}
+
+Result<StreamDataSink> StreamDataSink::create(
+    void* buffer_data_ptr,
+    size_t buffer_size,
+    const char* file_path,
+    size_t alignment) {
+  // Check if alignment is a power of two
+  if (alignment == 0 || (alignment & (alignment - 1)) != 0) {
+    return Error::InvalidArgument;
+  }
+
+  // Open the file and get the file descriptor
+  // It will fail if the file already exists
+  int file_descriptor = open(file_path, O_WRONLY | O_CREAT | O_EXCL, 0644);
+  if (file_descriptor < 0) {
+    // Return an error if the file cannot be accessed or created
+    ET_LOG(
+        Error, "Failed to open %s: %s (%d)", file_path, strerror(errno), errno);
+    return Error::AccessFailed;
+  }
+
+  // Return the successfully created StreamDataSink
+  return StreamDataSink(
+      buffer_data_ptr, buffer_size, file_descriptor, alignment);
+}
+
+StreamDataSink::~StreamDataSink() {
+  // Flush to and close the file descriptor
+  if (file_descriptor_ >= 0) {
+    flush();
+    close(file_descriptor_);
+  }
+}
+
+Result<size_t> StreamDataSink::write(const void* ptr, size_t size) {
+  if (size == 0) {
+    // No data to write, return current offset
+    return buffer_offset_ + total_written_bytes_;
+  }
+
+  const uint8_t* data_ptr = static_cast<const uint8_t*>(ptr);
+
+  // Align the buffer offset
+  uint8_t* aligned_ptr =
+      internal::align_pointer(buffer_.data() + buffer_offset_, alignment_);
+
+  // Zero out the padding between data blobs
+  size_t n_zero_pad = aligned_ptr - (buffer_.data() + buffer_offset_);
+  memset(buffer_.data() + buffer_offset_, 0, n_zero_pad);
+
+  // Calculate the new offset
+  size_t new_offset = (aligned_ptr - buffer_.data()) + size;
+
+  if (new_offset > buffer_.size()) {
+    // If the new offset is out of range, flush the buffer and try again
+    Result<bool> ret = flush();
+    if (!ret.ok()) {
+      return ret.error();
+    }
+
+    aligned_ptr = internal::align_pointer(buffer_.data(), alignment_);
+    n_zero_pad = aligned_ptr - buffer_.data();
+    memset(buffer_.data(), 0, n_zero_pad);
+    new_offset = (aligned_ptr - buffer_.data()) + size;
+
+    if (new_offset > buffer_.size()) {
+      return Error::OutOfResources;
+    }
+  }
+
+  // Copy data to the aligned position
+  std::memcpy(aligned_ptr, data_ptr, size);
+  buffer_offset_ = new_offset;
+
+  return aligned_ptr - buffer_.data() + total_written_bytes_;
+}
+
+size_t StreamDataSink::get_used_bytes() const {
+  return total_written_bytes_ + buffer_offset_;
+}
+
+Result<bool> StreamDataSink::flush() {
+  if (buffer_offset_ > 0) {
+    ssize_t written = ::write(file_descriptor_, buffer_.data(), buffer_offset_);
+    if (written != buffer_offset_) {
+      return Error::Internal;
+    }
+
+    total_written_bytes_ += written;
+    buffer_offset_ = 0;
+  }
+
+  return true;
+}
+
+} // namespace etdump
+} // namespace executorch

devtools/etdump/data_sinks/stream_data_sink.h

@@ -0,0 +1,121 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ * All rights reserved.
+ *
+ * This source code is licensed under the BSD-style license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+// (c) Meta Platforms, Inc. and affiliates. Confidential and proprietary.
+
+#pragma once
+
+#include <executorch/devtools/etdump/data_sinks/data_sink_base.h>
+#include <executorch/runtime/core/exec_aten/exec_aten.h>
+#include <executorch/runtime/core/span.h>
+#include <unistd.h> // For file operations
+
+namespace executorch {
+namespace etdump {
+/**
+ * StreamDataSink is a concrete implementation of DataSinkBase that manages
+ * the storage of data blobs using a buffer and a file descriptor. It writes
+ * data to a buffer first and flushes it to a file when the buffer is full,
+ * or when the flush() method is called explicitly. It is useful for storing
+ * large amounts of data in a file without consuming excessive memory.
+ *
+ * Noted that
+ * - This class is demonstrated purpose only, and not intended to be used or
+ * existed in production.
+ * - The buffer is provided and owned by user at construction time. The user is
+ * responsible for ensuring the validity and ownership of these resources.
+ */
+class StreamDataSink : public DataSinkBase {
+ public:
+  /**
+   * Creates a StreamDataSink with a given buffer and file path.
+   *
+   * @param[in] buffer_data_ptr A pointer to the buffer used for temporary
+   * storage.
+   * @param[in] buffer_size The size of the buffer.
+   * @param[in] file_path The path to the file for writing data.
+   * @param[in] alignment The alignment requirement for the buffer. Default
+   * is 64.
+   * @returns A new StreamDataSink on success.
+   * @retval Error::InvalidArgument `alignment` is not a power of two.
+   * @retval Error::AccessFailed Cannot access/create the file using
+   * `file_path`.
+   */
+  static ::executorch::runtime::Result<StreamDataSink> create(
+      void* buffer_data_ptr,
+      size_t buffer_size,
+      const char* file_path,
+      size_t alignment = 64);
+
+  /**
+   * Destructor that ensures all buffered data is flushed to the file.
+   */
+  ~StreamDataSink() override;
+
+  // Delete copy constructor and copy assignment operator
+  StreamDataSink(const StreamDataSink&) = delete;
+  StreamDataSink& operator=(const StreamDataSink&) = delete;
+
+  StreamDataSink(StreamDataSink&& other) noexcept;
+  StreamDataSink& operator=(StreamDataSink&& other) = default;
+
+  /**
+   * Writes data into the debug storage aligned to the given alignment.
+   *
+   * @param[in] ptr A pointer to the data to be written into the storage.
+   * @param[in] size The size of the data in bytes.
+   * @return A Result object containing either:
+   *         - The offset of the starting location of the data within the
+   *           debug storage, or
+   *         - An error code indicating the failure reason.
+   */
+  ::executorch::runtime::Result<size_t> write(const void* ptr, size_t size)
+      override;
+
+  /**
+   * Gets the number of bytes currently used in the debug storage.
+   *
+   * @return The amount of data currently stored in bytes.
+   */
+  size_t get_used_bytes() const override;
+
+  /**
+   * Flushes all buffered data to the file. No alignment is applied.
+   *
+   * @return A Result object containing either:
+   *         - true, indicating the flush operation was successful, or
+   *         - An error code indicating the failure reason.
+   */
+  ::executorch::runtime::Result<bool> flush();
+
+ private:
+  /**
+   * Constructs a StreamDataSink with a given buffer and file descriptor.
+   *
+   * @param[in] buffer A span representing the buffer used for temporary
+   * storage.
+   * @param[in] file_descriptor A valid file descriptor for writing data.
+   */
+  StreamDataSink(
+      void* buffer_data_ptr,
+      size_t buffer_size,
+      int file_descriptor,
+      size_t alignment)
+      : buffer_({static_cast<uint8_t*>(buffer_data_ptr), buffer_size}),
+        buffer_offset_(0),
+        file_descriptor_(file_descriptor),
+        total_written_bytes_(0),
+        alignment_(alignment) {}
+  ::executorch::runtime::Span<uint8_t> buffer_;
+  size_t buffer_offset_;
+  int file_descriptor_;
+  size_t total_written_bytes_;
+  size_t alignment_;
+};
+} // namespace etdump
+} // namespace executorch

devtools/etdump/data_sinks/targets.bzl

@@ -47,3 +47,4 @@ def define_common_targets():
         )
 
         define_data_sink_target("buffer_data_sink", aten_suffix)
+        define_data_sink_target("stream_data_sink", aten_suffix)