scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-10-08 03:47:51 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Create a pool of goroutines bounded to hardware limits. This should resolve "connection refused" error from spawning too many goroutines at once. (#2138)

Browse source
This commit is contained in:
souhailanoor 2022-06-29 16:22:37 -05:00 • committed by GitHub
parent aaad12ed5b
commit 91e59e4376
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 15 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
api.go
View file

@ -15,6 +15,7 @@ import (
"math"
"net/url"
"os"
"runtime"
"sort"
"strconv"
"strings"
@ -1556,11 +1557,15 @@ func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest,
options.IgnoreKeyCheck = true
var eg errgroup.Group
guard := make(chan struct{}, runtime.NumCPU()) // only run as many goroutines as CPUs available
for _, subReq := range reqs {
// TODO: if local node owns this shard we don't need to go through the client
guard <- struct{}{} // would block if guard channel is already filled
subReq := subReq
eg.Go(func() error {
return api.server.defaultClient.Import(ctx, qcx, subReq, options)
err := api.server.defaultClient.Import(ctx, qcx, subReq, options)
<-guard
return err
})
}
return eg.Wait()
@ -1851,7 +1856,8 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu
options.IgnoreKeyCheck = true
start := 0
shard := req.ColumnIDs[0] / ShardWidth
var eg errgroup.Group // TODO make this a pooled errgroup
var eg errgroup.Group
guard := make(chan struct{}, runtime.NumCPU()) // only run as many goroutines as CPUs available
for i, colID := range req.ColumnIDs {
if colID/ShardWidth != shard {
subreq := &ImportValueRequest{
@ -1865,9 +1871,11 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu
} else if req.FloatValues != nil {
subreq.FloatValues = req.FloatValues[start:i]
}
guard <- struct{}{} // would block if guard channel is already filled
eg.Go(func() error {
return api.server.defaultClient.ImportValue(ctx, qcx, subreq, options)
err := api.server.defaultClient.ImportValue(ctx, qcx, subreq, options)
<-guard
return err
})
start = i
shard = colID / ShardWidth

6
ctl/import_test.go
View file

@ -249,8 +249,8 @@ func TestImportCommand_KeyReplication(t *testing.T) {
// create a large import file in order to test the
// translateStoreBufferSize growth logic.
keyBytes := []byte{}
for row := 0; row < 100; row++ {
for col := 0; col < 100; col++ {
for row := 0; row < 1000; row++ {
for col := 0; col < 1000; col++ {
x := fmt.Sprintf("foo%d,bar%d\n", row, col)
keyBytes = append(keyBytes, x...)
}
@ -303,7 +303,7 @@ func TestImportCommand_KeyReplication(t *testing.T) {
}
// Read body and unmarshal response.
exp := `{"results":[100]}` + "\n"
exp := `{"results":[1000]}` + "\n"
if body, err := ioutil.ReadAll(resp.Body); err != nil {
return fmt.Errorf("reading: %s", err)
} else if !reflect.DeepEqual(body, []byte(exp)) {