feat: add buffering layer to BulkWriter by thebrianchen · Pull Request #1470 · googleapis/nodejs-firestore

thebrianchen · 2021-04-12T16:03:20Z

This PR adds a buffer layer to BulkWriter to limit the number of outstanding RPCs that BulkWriter has. This helps prevent BulkWriter from OOMing when multiple RPCs start failing due to hotspotting or contention.

schmidt-sebastian · 2021-04-12T18:33:39Z

+    } else {
+      // Increment the _lastOp pointer to ensure that the promise returned by
+      // flush() includes this operation.
+      this._lastOp = this._lastOp.then(() => {


I think we should do this in both cases and remove the duplication between here and _sendFn.

schmidt-sebastian · 2021-04-12T18:34:37Z

+          }
+      );
+      this._bufferedOperations.push(() =>
+        this._sendFn(enqueueOnBatchCallback, bulkWriterOp)


Do we need to increment pendingOpsCount? (did not check)

codecov · 2021-04-12T22:32:19Z

Codecov Report

Merging #1470 (740882f) into master (cf1949f) will increase coverage by 0.00%.
The diff coverage is 100.00%.

@@           Coverage Diff           @@
##           master    #1470   +/-   ##
=======================================
  Coverage   98.21%   98.21%           
=======================================
  Files          32       32           
  Lines       19567    19651   +84     
  Branches     1373     1379    +6     
=======================================
+ Hits        19217    19301   +84     
  Misses        346      346           
  Partials        4        4

Impacted Files	Coverage Δ
dev/src/bulk-writer.ts	`100.00% <100.00%> (ø)`

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update cf1949f...740882f. Read the comment docs.

schmidt-sebastian

This looks great! Some optional suggestions.

schmidt-sebastian · 2021-04-13T15:31:25Z

+ * BulkWriter instance. BulkWriter buffers additional writes after this many
+ * pending operations in order to avoiding going OOM.
+ */
+const DEFAULT_MAXIMUM_PENDING_OPERATIONS_COUNT = 10000;


I don't think there is any actual benefit to having 500 RPCs outstanding. Do you see any differences in throughput? Ideally, the number of outstanding RPCs should be well below 100, so that we don't have to initialize new clients.

Changed to 500, after running some throughput tests. I don't see any drastic changes, but the throughput usually has pretty high variance unless I run tests for > 1 hour.

schmidt-sebastian · 2021-04-13T15:33:23Z

+   * The number of pending operations enqueued on this BulkWriter instance.
+   * @private
+   */
+  private _pendingOpsCount = 0;


'pending' might not be the right term since it does not necessarily indicate that only includes operations that have been sent. I wonder if there is a better term.

Couldn't think of a better term. It seemed like pending was the closest thing, but I added a separate comment to describe what it means.

schmidt-sebastian · 2021-04-13T15:34:36Z

+    this._lastOp = this._lastOp.then(() => {
+      return silencePromise(bulkWriterOp.promise);
+    });


Suggested change

this._lastOp = this._lastOp.then(() => {

return silencePromise(bulkWriterOp.promise);

});

this._lastOp = this._lastOp.then(() => silencePromise(bulkWriterOp.promise));

Optional

But is it? :)

schmidt-sebastian · 2021-04-13T15:41:32Z

+    // buffer.
+    if (this._pendingOpsCount < this._maxPendingOpCount) {
+      this._pendingOpsCount++;
+      this._sendFn(enqueueOnBatchCallback, bulkWriterOp);


You might be able to simplify this further if you move the incrementing and decrementing into _sendFn. You only have to increment once and you can move the call to _processBufferedOps() to:

this._lastOp = this._lastOp.then(() => silencePromise(op.promise));

Optional.

I moved the incrementing call into sendFn. However, _processedBufferedOps has to be part of the returned promise in order for it to get run. Any promise off of lastOp doesn't get run by Node, which then creates the race condition with bulkCommit() in _sendBatch.

schmidt-sebastian · 2021-04-13T15:44:31Z

+      .catch(err => {
+        this._processBufferedOps();
+        throw err;
+      });


You can probably return this._lastOp if you move this code to either line 862 above or _sendFn.

I can't return lastOp since it's Promise<void>, but the return type is Promise<WriteResult>.

schmidt-sebastian · 2021-04-13T15:45:01Z

+      },
+    ]);
+    // eslint-disable-next-line no-console
+    console.log('BCHEN start buffer test');


thebrianchen

Thanks!

thebrianchen · 2021-04-13T17:04:25Z

+   * The number of pending operations enqueued on this BulkWriter instance.
+   * @private
+   */
+  private _pendingOpsCount = 0;


Couldn't think of a better term. It seemed like pending was the closest thing, but I added a separate comment to describe what it means.

thebrianchen · 2021-04-13T17:04:54Z

+    this._lastOp = this._lastOp.then(() => {
+      return silencePromise(bulkWriterOp.promise);
+    });


thebrianchen · 2021-04-13T17:08:01Z

+      },
+    ]);
+    // eslint-disable-next-line no-console
+    console.log('BCHEN start buffer test');


thebrianchen · 2021-04-13T17:11:15Z

+      .catch(err => {
+        this._processBufferedOps();
+        throw err;
+      });


I can't return lastOp since it's Promise<void>, but the return type is Promise<WriteResult>.

thebrianchen · 2021-04-13T17:18:43Z

+    // buffer.
+    if (this._pendingOpsCount < this._maxPendingOpCount) {
+      this._pendingOpsCount++;
+      this._sendFn(enqueueOnBatchCallback, bulkWriterOp);


I moved the incrementing call into sendFn. However, _processedBufferedOps has to be part of the returned promise in order for it to get run. Any promise off of lastOp doesn't get run by Node, which then creates the race condition with bulkCommit() in _sendBatch.

thebrianchen · 2021-04-13T17:20:15Z

+ * BulkWriter instance. BulkWriter buffers additional writes after this many
+ * pending operations in order to avoiding going OOM.
+ */
+const DEFAULT_MAXIMUM_PENDING_OPERATIONS_COUNT = 10000;


Changed to 500, after running some throughput tests. I don't see any drastic changes, but the throughput usually has pretty high variance unless I run tests for > 1 hour.

thebrianchen self-assigned this Apr 12, 2021

thebrianchen requested review from a team April 12, 2021 16:03

product-auto-label bot added the api: firestore Issues related to the googleapis/nodejs-firestore API. label Apr 12, 2021

google-cla bot added the cla: yes This human has signed the Contributor License Agreement. label Apr 12, 2021

schmidt-sebastian reviewed Apr 12, 2021

View reviewed changes

feat: add buffering layer to BulkWriter

57a90b6

thebrianchen force-pushed the bc/bulk-buffer branch from 1b236d8 to 57a90b6 Compare April 12, 2021 23:42

Merge branch 'master' into bc/bulk-buffer

9651a54

thebrianchen requested review from schmidt-sebastian April 13, 2021 00:00

thebrianchen assigned schmidt-sebastian and unassigned thebrianchen Apr 13, 2021

schmidt-sebastian approved these changes Apr 13, 2021

View reviewed changes

schmidt-sebastian assigned thebrianchen and unassigned schmidt-sebastian Apr 13, 2021

thebrianchen commented Apr 13, 2021

View reviewed changes

resolve comments

fc262b5

thebrianchen force-pushed the bc/bulk-buffer branch from 1a163fc to fc262b5 Compare April 13, 2021 17:21

handle pendingOpsCount in one place

e3c4168

schmidt-sebastian approved these changes Apr 13, 2021

View reviewed changes

Brian Chen added 2 commits April 13, 2021 13:09

it is, at last

ec2262a

lint

740882f

thebrianchen merged commit 9cc9548 into master Apr 13, 2021

thebrianchen deleted the bc/bulk-buffer branch April 13, 2021 18:26

This was referenced Apr 27, 2021

feat: add buffering layer to BulkWriter googleapis/java-firestore#610

Closed

feat: add buffering layer to BulkWriter googleapis/java-firestore#611

Merged

gdieguez mentioned this pull request Jun 9, 2021

BulkWriter.close doesn't resolve when writing more than 500 documents #1531

Closed

Conversation

thebrianchen commented Apr 12, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codecov bot commented Apr 12, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

schmidt-sebastian left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

thebrianchen left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

thebrianchen commented Apr 12, 2021 •

edited

Loading

codecov bot commented Apr 12, 2021 •

edited

Loading