WIP feat: Implement StreamWriter and StructuredIrStreamWriter for IR data streaming (based on #74).

[zzxthehappiest]

Description

Implemented the desiredSize/ready and abort on top of #74

Done:

• Add and bind three new functions in StreamWriter class:
  • getLastWritePromise: return a Promise for the last write operation. Since the clp::WriterInterface::write returns void, to make the WritableStream can correctly backpressure (as the example here), we need await writer.ready (which is an async operation) in WebStreamWriter::write, so it also needs to return a Promise so in JS code the Promise produced by await writer.ready can be resolved correctly, avoid memory leak after main function exits (if not doing so, there will be a memory access exception).
  • desiredSize: return the desiredSize property of the WritableStream, it is a sync function.
  • abort: call the abort(reason) defined in the underlying sink in theWritableStream.
• Add a helper function in WebStreamWriter:
  • await_ready: get the readyPromise property of the WritableStream, so that we can add callback to make the WebStreamWriter::write correctly write the content after backpressure is finished.
• Rewrite WebStreamWriter::write:
  • Since the previous version is a pure sync so it is not able to use desiredSize/ready backpressure mechnism of WritableStream, so I make it a "fake"-async function.
• Refactor the unit test:
  • Modify the CMakeLists.txt now it copies the generated .js files to the unit test directly, so we don't need to hard code it as cmake-build-release which is only used by Clion.
  • Add the unit test for backpressure and abort.
  • Add the unit test for ClpFfiJs-worker, but haven't successfully run it for unknown reason.

Questions:

• For the following FIXME I am not sure what do they mean exactly:
  • In StreamWriter::write (link)
  • In StreamWriter::flush (link). For the flush I checked the Compressor.cpp, it seems we never use m_compressed_stream_writer.flush() so I am not sure do we really need to implement this in the WebStreamWriter as well as modify the currentStructuredIrStreamWriter::flush.
  • In StructuredIrStreamWriter::write (link)
  • In StructuredIrStreamWriter::close (link)
• For the following TODO I am not sure what do they mean exactly:
  • In StreamWriter::write (link)
  • In the unit test (link)
• To support the backpressure we need to support async; to support async we need to remove the -fwasm-exceptions compile flag and add -sASYNCIFY=1 flag, otherwise there is compile error says the async is not supported. However, there is an apparent performance cost of it, the 1000000 write operations takes around 0.7s slower than before (2.1s -> 2.8s). It seems these two compile flags have not been supported at the same time yet in the emsk framework. But from the stream standard, WritableStreamDefaultWriter's abort, close and write are all async (they all return Promise, link), as well as the same functions in the underlying sink (link). So I think it is worth to trade off the cost to support the async feature and we can potentially gain the performance back in other places (e.g., better caching policy when write).

Checklist

• The PR satisfies the contribution guidelines.
• This is a breaking change and that has been indicated in the PR title, OR this isn't a
  breaking change.
• Necessary docs have been updated, OR no docs need to be updated.

Validation performed

Run the unit test test-node.mjs.

[coderabbitai]

Important

Review skipped

Draft detected.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

🪧 Tips

Chat

There are 3 ways to chat with CodeRabbit:

• Review comments: Directly reply to a review comment made by CodeRabbit. Example:
  • I pushed a fix in commit <commit_id>, please review it.
  • Generate unit testing code for this file.
  • Open a follow-up GitHub issue for this discussion.
• Files and specific lines of code (under the "Files changed" tab): Tag @coderabbitai in a new review comment at the desired location with your query. Examples:
  • @coderabbitai generate unit testing code for this file.
  • @coderabbitai modularize this function.
• PR comments: Tag @coderabbitai in a new PR comment to ask questions about the PR branch. For the best results, please provide a very specific query, as very limited context is provided in this mode. Examples:
  • @coderabbitai gather interesting stats about this repository and render them as a table. Additionally, render a pie chart showing the language distribution in the codebase.
  • @coderabbitai read src/utils.ts and generate unit testing code.
  • @coderabbitai read the files in the src/scheduler package and generate a class diagram using mermaid and a README in the markdown format.
  • @coderabbitai help me debug CodeRabbit configuration file.

Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments.

CodeRabbit Commands (Invoked using PR comments)

• @coderabbitai pause to pause the reviews on a PR.
• @coderabbitai resume to resume the paused reviews.
• @coderabbitai review to trigger an incremental review. This is useful when automatic reviews are disabled for the repository.
• @coderabbitai full review to do a full review from scratch and review all the files again.
• @coderabbitai summary to regenerate the summary of the PR.
• @coderabbitai generate docstrings to generate docstrings for this PR.
• @coderabbitai generate sequence diagram to generate a sequence diagram of the changes in this PR.
• @coderabbitai resolve resolve all the CodeRabbit review comments.
• @coderabbitai configuration to show the current CodeRabbit configuration for the repository.
• @coderabbitai help to get help.

Other keywords and placeholders

• Add @coderabbitai ignore anywhere in the PR description to prevent this PR from being reviewed.
• Add @coderabbitai summary to generate the high-level summary at a specific location in the PR description.
• Add @coderabbitai anywhere in the PR title to generate the title automatically.

CodeRabbit Configuration File (.coderabbit.yaml)

• You can programmatically configure CodeRabbit by adding a .coderabbit.yaml file to the root of your repository.
• Please see the configuration documentation for more information.
• If your editor has YAML language server enabled, you can add the path at the top of this file to enable auto-completion and validation: # yaml-language-server: $schema=https://coderabbit.ai/integrations/schema.v2.json

Documentation and Community

• Visit our Documentation for detailed information on how to use CodeRabbit.
• Join our Discord Community to get help, request features, and share feedback.
• Follow us on X/Twitter for updates and announcements.

[junhaoliao]

@zzxthehappiest
Sorry for this late response. Thanks for helping and please see below for my answers:

• For the following FIXME I am not sure what do they mean exactly:
  • In StreamWriter::write (link)

Sorry for the confusion. I meant to say instead of implementing a write() method which accepts some chunk as bytes, we may want a separate interface (e.g., appendLogEvent(event: object) ) to accept some argument of type object which contains kv-pairs. In appendLogEvent, after serializing the object into msgpack bytes, appendLogEvent can call write(chunk) where the chunk will be the msgpack bytes.

However, during prototyping, as we may find via the commit history, it was found that serializing any JS-object in C++ is quite expensive (~4x slower) because there will be a lot of cross-VM communications. Instead, I decided to serialize the object -> msgpack in JS then pass the msgpack bytes into C++ code. Luckily, it can be easily achieved via the msgpackr library as we can see in the latest test.mjs.

Nevertheless, please ignore this action item for now. I am thinking of a different interface due to some limitations in the current implementation. Please see below for the explanation.

---

In the prototype, I was hoping to implement the CLP IR writer with some common / standard interface in JavaScript, so it may work better with other (JS-standard / 3rd-party) reader / writer / transform streams. However, after conducting some throughout research this week, I found WritableStreamDefaultWriter's extensibility is quite limited - it seems no web code developer is supposed to directly call the constructor of a WritableStreamDefaultWriter because such instances are only supposed to be created via getWriter() on a WritableStream instance. For flush control, the standard does provide a WritableStreamDefaultController class, but any control seems to only happen in the underlyingSink's write() method, which isn't directly achievable by simply extending WritableStreamDefaultWriter. Even worse, I also couldn't find any example about extending the WritableStreamDefaultWriter class in any official docs / online.

I believe some changes are needed in the interface. Here are some approaches I am picturing:

1. In addition to extending WritableStreamDefaultWriter as StructuredIrStreamWriter, we can also extend WritableStream as StructuredIrWritableStream. In the extended WritableStream class, getWriter() returns a StructuredIrStreamWriter instance for clp-ffi-js library users to write msgpack bytes as chunks. For flush control on write, StructuredIrWritableStream will execute its own logic before calling the underlying sink's write method. In addition, for explicit flush control by library users, StructuredIrWritableStream will also provide a flush() method.
   Below is some pseudo code for illustration:

   class StructuredIrWritableStream extends WritableStream {
       constructor(underlyingSink, strategy) {
           this.#writeToSink = underlyingSink.write
   
           super({
               ...underlyingSink,
               write: (chunk) => {
                   // our own flush-on-write logic
                   if (shouldFlush) {
                       underlyingSink.write(chunk)
                   }
               }
           }, strategy)
       }
   
       flush() {
           this.#writeToSink(chunk)
       }
   
       getWriter() {
           return StructuredIrStreamWriter(stream, DEFAULT_WRITER_OPTIONS)
       }
   
       getWriterWithOptions(options) {
           return StructuredIrStreamWriter(stream, options)
       }
   }

2. OR: Implement a custom StructuredIrStreamWriter instead of extending WritableStreamDefaultWriter, so we're not constrained by the lifecycle requirements of the WritableStream utilities.
3. OR: In addition to (2.) where we implement our own StructuredIrStreamWriter, provide a method createUnderlyingSink, which returns a UnderlyingSink dictionary to be used in WritableStream's constructor. Essentially, in this approach we are providing an "underlying sink" instead of a writer to our library users.

cc @LinZhihao-723 @davidlion @kirkrodrigues @hoophalab for comments, who are experts with writers in Python / Golang.

I'll respond to the rest of the questions in a separate thread to avoid spamming the experts.

[hoophalab]

I'm leaning toward the second approach (and optionally including the third). This way, we can have a writer with extended methods like flush() and getNumBytesSerialized().

We can expose a standard WritableStream through a asStream() -> WritableStream method (or toStream()), which users can then use with the TransformStream class. The corresponding WritableStreamDefaultWriter would only support a subset of the methods provided in the extended writer.

Junhao and I would love to hear your thoughts and discuss further: @LinZhihao-723 @davidlion @kirkrodrigues @davemarco

[junhaoliao]

@zzxthehappiest @hoophalab

@davidlion reviewed this and we discussed offline -
The primary reason for implementing existing JavaScript interfaces is to enable integration with other stream types, such as CompressionStream. (What @hoophalab proposed will work well for this purpose - CompressionStream extends TranformStream.) However, we should note CompressionStream doesn't yet support Zstandard (zstd) compression, the de-facto standard algorithm used in CLP IR formats.
Besides that, extending the WritableStream class specifically for logging purposes offers limited benefits, given its lack of adoption in this context. Instead, as @davidlion proposed, we want to follow the interface patterns already established in our clp-ffi-py and clp-ffi-go libraries, focusing on ensuring compatibility with popular JavaScript logging frameworks like Pino and Winston rather than adhering to the WritableStream lifecycle.

[zzxthehappiest]

@zzxthehappiest Sorry for this late response. Thanks for helping and please see below for my answers:

• For the following FIXME I am not sure what do they mean exactly:

  • In StreamWriter::write (link)

Sorry for the confusion. I meant to say instead of implementing a write() method which accepts some chunk as bytes, we may want a separate interface (e.g., appendLogEvent(event: object) ) to accept some argument of type object which contains kv-pairs. In appendLogEvent, after serializing the object into msgpack bytes, appendLogEvent can call write(chunk) where the chunk will be the msgpack bytes.

However, during prototyping, as we may find via the commit history, it was found that serializing any JS-object in C++ is quite expensive (~4x slower) because there will be a lot of cross-VM communications. Instead, I decided to serialize the object -> msgpack in JS then pass the msgpack bytes into C++ code. Luckily, it can be easily achieved via the msgpackr library as we can see in the latest test.mjs.

Nevertheless, please ignore this action item for now. I am thinking of a different interface due to some limitations in the current implementation. Please see below for the explanation.

In the prototype, I was hoping to implement the CLP IR writer with some common / standard interface in JavaScript, so it may work better with other (JS-standard / 3rd-party) reader / writer / transform streams. However, after conducting some throughout research this week, I found WritableStreamDefaultWriter's extensibility is quite limited - it seems no web code developer is supposed to directly call the constructor of a WritableStreamDefaultWriter because such instances are only supposed to be created via getWriter() on a WritableStream instance. For flush control, the standard does provide a WritableStreamDefaultController class, but any control seems to only happen in the underlyingSink's write() method, which isn't directly achievable by simply extending WritableStreamDefaultWriter. Even worse, I also couldn't find any example about extending the WritableStreamDefaultWriter class in any official docs / online.

I believe some changes are needed in the interface. Here are some approaches I am picturing:

1. In addition to extending WritableStreamDefaultWriter as StructuredIrStreamWriter, we can also extend WritableStream as StructuredIrWritableStream. In the extended WritableStream class, getWriter() returns a StructuredIrStreamWriter instance for clp-ffi-js library users to write msgpack bytes as chunks. For flush control on write, StructuredIrWritableStream will execute its own logic before calling the underlying sink's write method. In addition, for explicit flush control by library users, StructuredIrWritableStream will also provide a flush() method.
   Below is some pseudo code for illustration:

   class StructuredIrWritableStream extends WritableStream {
       constructor(underlyingSink, strategy) {
           this.#writeToSink = underlyingSink.write
   
           super({
               ...underlyingSink,
               write: (chunk) => {
                   // our own flush-on-write logic
                   if (shouldFlush) {
                       underlyingSink.write(chunk)
                   }
               }
           }, strategy)
       }
   
       flush() {
           this.#writeToSink(chunk)
       }
   
       getWriter() {
           return StructuredIrStreamWriter(stream, DEFAULT_WRITER_OPTIONS)
       }
   
       getWriterWithOptions(options) {
           return StructuredIrStreamWriter(stream, options)
       }
   }

2. OR: Implement a custom StructuredIrStreamWriter instead of extending WritableStreamDefaultWriter, so we're not constrained by the lifecycle requirements of the WritableStream utilities.
3. OR: In addition to (2.) where we implement our own StructuredIrStreamWriter, provide a method createUnderlyingSink, which returns a UnderlyingSink dictionary to be used in WritableStream's constructor. Essentially, in this approach we are providing an "underlying sink" instead of a writer to our library users.

cc @LinZhihao-723 @davidlion @kirkrodrigues @hoophalab for comments, who are experts with writers in Python / Golang.

I'll respond to the rest of the questions in a separate thread to avoid spamming the experts.

If I understand correctly, we are going to try the second option with what @hoophalab proposed?