Re-add rowDataChan buffering to avoid Get call stalling. Closes #382

turbot · Aug 22, 2022 · a712d9d · a712d9d
1 parent dccf629
commit a712d9d
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 7 deletions.
diff --git a/plugin/plugin.go b/plugin/plugin.go
@@ -575,9 +575,9 @@ func (p *Plugin) executeForConnection(ctx context.Context, req *proto.ExecuteReq
 	}
 
 	// asyncronously fetch items
-	log.Printf("[TRACE] calling fetchItemsAsync, table: %s, matrixItem: %v, limit: %d,  connectionCallId: %s\"", table.Name, queryData.Matrix, limit, connectionCallId)
-	if err := table.fetchItemsAsync(ctx, queryData); err != nil {
-		log.Printf("[WARN] fetchItemsAsync returned an error, table: %s, error: %v", table.Name, err)
+	log.Printf("[TRACE] calling fetchItems, table: %s, matrixItem: %v, limit: %d,  connectionCallId: %s\"", table.Name, queryData.Matrix, limit, connectionCallId)
+	if err := table.fetchItems(ctx, queryData); err != nil {
+		log.Printf("[WARN] fetchItems returned an error, table: %s, error: %v", table.Name, err)
 		return err
 
 	}

diff --git a/plugin/query_data.go b/plugin/query_data.go
@@ -20,6 +20,9 @@ import (
 	"github.com/turbot/steampipe-plugin-sdk/v4/telemetry"
 )
 
+// how may rows do we cache in the rowdata channel
+const rowDataBufferSize = 100
+
 // NOTE - any field added here must also be added to ShallowCopy
 
 type QueryData struct {
@@ -117,8 +120,8 @@ func newQueryData(connectionCallId string, plugin *Plugin, queryContext *QueryCo
 
 		// asyncronously read items using the 'get' or 'list' API
 		// items are streamed on rowDataChan, errors returned on errorChan
-		rowDataChan: make(chan *RowData),
-		errorChan:   make(chan error),
+		rowDataChan: make(chan *RowData, rowDataBufferSize),
+		errorChan:   make(chan error, 1),
 		outputChan:  outputChan,
 		listWg:      &wg,
 

diff --git a/plugin/table_fetch.go b/plugin/table_fetch.go
@@ -49,8 +49,8 @@ When executing for each matrix item, the matrix item is put into the context, av
 */
 
 // call either 'get' or 'list'.
-func (t *Table) fetchItemsAsync(ctx context.Context, queryData *QueryData) error {
-	ctx, span := telemetry.StartSpan(ctx, t.Plugin.Name, "Table.fetchItemsAsync (%s)", t.Name)
+func (t *Table) fetchItems(ctx context.Context, queryData *QueryData) error {
+	ctx, span := telemetry.StartSpan(ctx, t.Plugin.Name, "Table.fetchItems (%s)", t.Name)
 
 	defer span.End()
 
@@ -162,6 +162,12 @@ func (t *Table) doGetForQualValues(ctx context.Context, queryData *QueryData, ke
 	var getWg sync.WaitGroup
 	var errorChan = make(chan (error), len(qualValueList.Values))
 
+	// NOTE: ensure QueryData.rowDataChan can buffer sufficient items
+	// (we normally expect it would be sufficient)
+	if len(qualValueList.Values) > rowDataBufferSize {
+		queryData.rowDataChan = make(chan *RowData, len(qualValueList.Values))
+	}
+
 	// we will make a copy of  queryData and update KeyColumnQuals to replace the list value with a single qual value
 	for _, qv := range qualValueList.Values {
 		// make a shallow copy of the query data and modify the quals