feat: Implement batch sender. (#1995)

## Problem

The DFS scheduler implements concurrent resolution of resources, but only if resources are sent in batches to the top-level `res` channel. Thus, the same plugin syncs a lot quicker simply by batching resources before sending them through the channel.

## Solution this PR implements

```
// BatchSender is a helper struct that batches items and sends them in batches of batchSize or after batchTimeout.
//
// - If item is already a slice, it will be sent directly
// - Otherwise, it will be added to the current batch
// - If the current batch has reached the batch size, it will be sent immediately
// - Otherwise, a timer will be started to send the current batch after the batch timeout
type BatchSender struct {
```

Currently with a batch timeout of `100ms` and batch size of `100`, it manages a 5x improvement on a sync that reproduces the issue brought forward by the community (https://github.com/jeromewir/cq-source-concurrency-childtable-example):

<img width="982" alt="Screenshot 2024-12-03 at 17 17 23" src="https://github.com/user-attachments/assets/7a1f7dc5-4aca-4263-8aa6-0ec10d73cca3">

The PR tries to touch the least amount of scheduler code, and stays away from tricky language constructs as much as possible.

Author: marianogappa

scheduler/batchsender/batch_sender.go

@@ -0,0 +1,85 @@
+package batchsender
+
+import (
+	"sync"
+	"time"
+
+	"github.com/cloudquery/plugin-sdk/v4/helpers"
+)
+
+const (
+	batchSize    = 100
+	batchTimeout = 100 * time.Millisecond
+)
+
+// BatchSender is a helper struct that batches items and sends them in batches of batchSize or after batchTimeout.
+//
+// - If item is already a slice, it will be sent directly
+// - Otherwise, it will be added to the current batch
+// - If the current batch has reached the batch size, it will be sent immediately
+// - Otherwise, a timer will be started to send the current batch after the batch timeout
+type BatchSender struct {
+	sendFn    func(any)
+	items     []any
+	timer     *time.Timer
+	itemsLock sync.Mutex
+}
+
+func NewBatchSender(sendFn func(any)) *BatchSender {
+	return &BatchSender{sendFn: sendFn}
+}
+
+func (bs *BatchSender) Send(item any) {
+	if bs.timer != nil {
+		bs.timer.Stop()
+	}
+
+	items := helpers.InterfaceSlice(item)
+
+	// If item is already a slice, send it directly
+	// together with the current batch
+	if len(items) > 1 {
+		bs.flush(items...)
+		return
+	}
+
+	// Otherwise, add item to the current batch
+	bs.appendToBatch(items...)
+
+	// If the current batch has reached the batch size, send it
+	if len(bs.items) >= batchSize {
+		bs.flush()
+		return
+	}
+
+	// Otherwise, start a timer to send the current batch after the batch timeout
+	bs.timer = time.AfterFunc(batchTimeout, func() { bs.flush() })
+}
+
+func (bs *BatchSender) appendToBatch(items ...any) {
+	bs.itemsLock.Lock()
+	defer bs.itemsLock.Unlock()
+
+	bs.items = append(bs.items, items...)
+}
+
+func (bs *BatchSender) flush(items ...any) {
+	bs.itemsLock.Lock()
+	defer bs.itemsLock.Unlock()
+
+	bs.items = append(bs.items, items...)
+
+	if len(bs.items) == 0 {
+		return
+	}
+
+	bs.sendFn(bs.items)
+	bs.items = nil
+}
+
+func (bs *BatchSender) Close() {
+	if bs.timer != nil {
+		bs.timer.Stop()
+	}
+	bs.flush()
+}

scheduler/scheduler_dfs.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/cloudquery/plugin-sdk/v4/helpers"
+	"github.com/cloudquery/plugin-sdk/v4/scheduler/batchsender"
 	"github.com/cloudquery/plugin-sdk/v4/scheduler/metrics"
 	"github.com/cloudquery/plugin-sdk/v4/scheduler/resolvers"
 	"github.com/cloudquery/plugin-sdk/v4/schema"
@@ -121,9 +122,13 @@ func (s *syncClient) resolveTableDfs(ctx context.Context, table *schema.Table, c
 		}
 	}()
 
+	batchSender := batchsender.NewBatchSender(func(item any) {
+		s.resolveResourcesDfs(ctx, table, client, parent, item, resolvedResources, depth)
+	})
 	for r := range res {
-		s.resolveResourcesDfs(ctx, table, client, parent, r, resolvedResources, depth)
+		batchSender.Send(r)
 	}
+	batchSender.Close()
 
 	// we don't need any waitgroups here because we are waiting for the channel to close
 	endTime := time.Now()