use a worker pool to ingest entries faster #1964

ozkatz · 2021-05-18T16:10:57Z

No description provided.

arielshaqed

Neat!

I'm not sure how much speedup you can get from something like this -- each client goroutine appears to do less work than the server, and the server can load multiple DB threads but they all contend for the same resource. Still, at least now the client does the most it can do.

arielshaqed · 2021-05-19T05:35:14Z

cmd/lakectl/cmd/ingest.go

+
+func stageWorker(wg *sync.WaitGroup, requests <-chan *stageRequest, responses chan<- *stageResponse) {
+	defer wg.Done()
+	client := getClient()


Can we share clients? I think HTTP connection pooling/reuse anyway happens at a level below, so it might manage to reduce connection re-use which could increase performance (or decrease performance...).

From a different angle - getClient should be called once, it can fail and we do not want to dump multiple errors.
They all share the same transport - but we don't need to perform the same operation multiple times to construct it.

arielshaqed · 2021-05-19T05:38:39Z

cmd/lakectl/cmd/ingest.go

+
+			// update every 10k records
+			if summary.Objects%10000 == 0 {
+				Write("Staged {{ .Objects | green }} objects so far...\n", summary)


Suggested change

Write("Staged {{ .Objects | green }} objects so far...\n", summary)

Write("Staged {{ .Objects | green }} objects so far...\r", summary)

might give a cute single-line "animation". (But you'd have to write that \n later, when done)

Nice idea. Done.

itaiad200

Tested it, works like a charm!

itaiad200 · 2021-05-19T07:16:28Z

cmd/lakectl/cmd/ingest.go

@@ -81,5 +135,6 @@ func init() {
 	_ = ingestCmd.MarkFlagRequired("to")
 	ingestCmd.Flags().Bool("dry-run", false, "only print the paths to be ingested")
 	ingestCmd.Flags().BoolP("verbose", "v", false, "print stats for each individual object staged")
+	ingestCmd.Flags().IntP("concurrency", "C", 64, "max concurrent API calls to make to the lakeFS server")


Why capital 'C'?

Suggested change

ingestCmd.Flags().IntP("concurrency", "C", 64, "max concurrent API calls to make to the lakeFS server")

ingestCmd.Flags().IntP("concurrency", "c", 64, "max concurrent API calls to make to the lakeFS server")

because we use -c on the global lakectl command as short for --config

nopcoder · 2021-05-19T07:15:02Z

cmd/lakectl/cmd/ingest.go

@@ -12,6 +14,32 @@ const ingestSummaryTemplate = `
 Staged {{ .Objects | yellow }} external objects (total of {{ .Bytes | human_bytes | yellow }})
 `

+type stageRequest struct {
+	ctx        context.Context


ctx can be passed to the workers once on creation.

nopcoder · 2021-05-19T07:19:35Z

cmd/lakectl/cmd/ingest.go

+				if lakefsURI.Path != nil && *lakefsURI.Path != "" {
+					path := *lakefsURI.Path
+					if strings.HasSuffix(*lakefsURI.Path, "/") {


This can be done once outside the loop - just set prefix with the right value

nopcoder · 2021-05-19T07:21:49Z

cmd/lakectl/cmd/ingest.go

+		}()
+
+		for response := range responses {
+			DieOnResponseError(response.resp, response.error)


we just die - so no point in passing errors from the worker - can reduce all the specific response channel and just process responses

nopcoder · 2021-05-19T07:30:22Z

cmd/lakectl/cmd/ingest.go

+			summary.Bytes += api.Int64Value(response.resp.JSON201.SizeBytes)
+
+			// update every 10k records
+			if summary.Objects%10000 == 0 {


Consider using time since instead of round numbers or optional progress bar

nopcoder · 2021-05-19T07:37:07Z

cmd/lakectl/cmd/ingest.go

+
+func stageWorker(wg *sync.WaitGroup, requests <-chan *stageRequest, responses chan<- *stageResponse) {
+	defer wg.Done()
+	client := getClient()


From a different angle - getClient should be called once, it can fail and we do not want to dump multiple errors.
They all share the same transport - but we don't need to perform the same operation multiple times to construct it.

ozkatz · 2021-05-19T11:26:46Z

Neat!

I'm not sure how much speedup you can get from something like this -- each client goroutine appears to do less work than the server, and the server can load multiple DB threads but they all contend for the same resource. Still, at least now the client does the most it can do.

@arielshaqed the main speedup comes from the fact that this is a lakectl command - there's good chance a user will run it locally from their machine. In that case, the latency between user -> lakeFS server could be big, so doing it serially one object at a time really adds up.

arielshaqed

Neat. Love the progress-by-time idea & implementation.

nopcoder

👍

use a worker pool to ingest entries

8818048

ozkatz requested review from nopcoder and itaiad200 May 18, 2021 16:10

arielshaqed approved these changes May 19, 2021

View reviewed changes

itaiad200 approved these changes May 19, 2021

View reviewed changes

nopcoder suggested changes May 19, 2021

View reviewed changes

implemenet review comments

5009fdb

ozkatz requested review from nopcoder, arielshaqed and itaiad200 May 19, 2021 11:21

arielshaqed approved these changes May 19, 2021

View reviewed changes

nopcoder approved these changes May 19, 2021

View reviewed changes

ozkatz merged commit 4bfead3 into master May 19, 2021

ozkatz deleted the feature/concurrent-ingest branch May 19, 2021 11:44

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

use a worker pool to ingest entries faster #1964

use a worker pool to ingest entries faster #1964

ozkatz commented May 18, 2021

arielshaqed left a comment

arielshaqed May 19, 2021

nopcoder May 19, 2021

ozkatz May 19, 2021

arielshaqed May 19, 2021

ozkatz May 19, 2021

itaiad200 left a comment

itaiad200 May 19, 2021

ozkatz May 19, 2021

nopcoder May 19, 2021

ozkatz May 19, 2021

nopcoder May 19, 2021

ozkatz May 19, 2021

nopcoder May 19, 2021

ozkatz May 19, 2021

nopcoder May 19, 2021

ozkatz May 19, 2021

nopcoder May 19, 2021

ozkatz commented May 19, 2021

arielshaqed left a comment

nopcoder left a comment

	Write("Staged {{ .Objects \| green }} objects so far...\n", summary)
	Write("Staged {{ .Objects \| green }} objects so far...\r", summary)

	ingestCmd.Flags().IntP("concurrency", "C", 64, "max concurrent API calls to make to the lakeFS server")
	ingestCmd.Flags().IntP("concurrency", "c", 64, "max concurrent API calls to make to the lakeFS server")

use a worker pool to ingest entries faster #1964

use a worker pool to ingest entries faster #1964

Conversation

ozkatz commented May 18, 2021

arielshaqed left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

itaiad200 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ozkatz commented May 19, 2021

arielshaqed left a comment

Choose a reason for hiding this comment

nopcoder left a comment

Choose a reason for hiding this comment