feat(transfer): Optimized approach for OCM transfer by implementing a concurrent worker pool. #1420

dynamic-Archu · 2025-04-22T10:24:22Z

Optimized approach for OCM transfer by implementing a concurrent worker pool, replacing the previous sequential transfer logic.

What this PR does / why we need it

This PR introduces an optimized approach for OCM transfer by implementing a concurrent worker pool, replacing the previous sequential transfer logic.

Improvements:

Reduced transfer time from ~45 minutes to ~23 minutes 40 seconds.
Achieved nearly 2x speedup by leveraging concurrent processing.
Introduced a configurable concurrent worker pool to parallelize transfers.

Technical Approach:

Replaced copyVersion with a new function copyVersionWithWorkerPool, using a bounded goroutine worker pool
Introduced a transferTask abstraction to encapsulate resource and source transfer logic
Enabled parallel transfer of artifacts using a concurrent worker pool (default size: 5 workers)
Tuned worker pool size for optimal performance and resource utilization.

Why:

Introducing concurrency significantly improves transfer speed and overall efficiency.

Which issue(s) this PR fixes

Optimized approach for OCM transfer by implementing a concurrent worker pool, replacing the previous sequential transfer logic.

jakobmoellerdev

Thanks for the contribution.

Please adjust the following:

Ensure Context Cancellation is still possible
Ensure tests are green
Please keep existing comments and keep your change set minimal for the change needed.
Please consider what happens if children have joint dependencies

Component A
=> Component B
=> Component C

Component B
=> Component D
Component C
=> Component D

Currently you are triggering multiple transfers for Component D which is incorrect.

Cheers!

dynamic-Archu · 2025-04-24T06:08:01Z

@jakobmoellerdev

Thanks for the feedback! I have a few questions regarding the points you've raised:

Could you clarify the specific logic you expect for context cancellation ?
For the scenario with joint dependencies (i.e., Component B and Component C both depending on Component D), would the correct approach be to ignore the second transfer of Component D? Or is the ideal solution to ensure that only one transfer occurs for Component D, possibly by adding a check to see if it has already been triggered?
Should we be considering any optimization patterns to avoid triggering multiple transfers?

jakobmoellerdev · 2025-04-24T07:01:14Z

@jakobmoellerdev

Thanks for the feedback! I have a few questions regarding the points you've raised:

Could you clarify the specific logic you expect for context cancellation ?

I expect that the worker pool can be cancelled if a parent context is cancelled. you removed the parent context right now. I think you can just add this back in and create a new subcontext: https://github.com/open-component-model/ocm/pull/1420/files#diff-d341af7fc25e4e0ea76361047d0e578a1e197d1e598fcba3ffc5ce866c42358dL38

For the scenario with joint dependencies (i.e., Component B and Component C both depending on Component D), would the correct approach be to ignore the second transfer of Component D? Or is the ideal solution to ensure that only one transfer occurs for Component D, possibly by adding a check to see if it has already been triggered?

It is imperative that the transfer for a single component version in the handler happens at most once per transfer attempt. Retries are handled within there, so we should avoid triggering it multiple times. There are a few ways you can do this, for example with mutexes or synced maps.

Should we be considering any optimization patterns to avoid triggering multiple transfers?

See above :)

One other thing: If the transfer of one component version fails, we should cancel all other active transfers. For that you should possibly use an error group from the sync package

Updated existing implementation with new changes

jakobmoellerdev · 2025-05-13T13:40:38Z

api/ocm/tools/transfer/transfer.go

 	}

+	// subp := common.AddPrinterGap(ctx, "  ")
+	var wg sync.WaitGroup


I just noticed this while taking a look here again, but this is not implemented as described in the PR:

Enabled parallel transfer of artifacts using a concurrent worker pool (default size: 5 workers, configurable)
Tuned worker pool size for optimal performance and resource utilization.

You should make this sync group limited and configurable

This comment is still not resolved

Is the expectation here that once one of the go routine fails, all others should abort? if so, then you should use an errgroup.WithContext. WDYT?

jakobmoellerdev · 2025-05-13T13:43:35Z

api/ocm/tools/transfer/transfer.go

-func CopyVersionWithContext(cctx context.Context, log logging.Logger, hist common.History, src ocm.ComponentVersionAccess, t ocm.ComponentVersionAccess, handler TransferHandler) (rerr error) {
-	return copyVersion(cctx, common.GetPrinter(cctx), log, hist, src, t, src.GetDescriptor().Copy(), handler)
+func CopyVersionWithContext(cctx context.Context, printer common.Printer, log logging.Logger, hist common.History, src ocm.ComponentVersionAccess, t ocm.ComponentVersionAccess, handler TransferHandler) (rerr error) {
+	return copyVersionWithWorkerPool(cctx, printer, log, hist, src, t, src.GetDescriptor().Copy(), handler, 5)


The 5 workers seem very specific to your trial run and should not be used for OCM in general.

I have 2 suggestions for possible value setting:

Default to a runtime-dynamic value, for example runtime.NumCPU() that defaults to active discovered CPUs by the GO runtime

Default to a sequential value, i.e. 1

Choose whichever you prefer.

That being said, to properly make it configurable like you describe in your PR, I am missing exactly that: a configurable flag. As such you will need to introduce both a setting/option as well as a CLI flag for concurrency. Ideally this should also be configurable in .ocmconfig. I don't mind for now if you want to have this in separate PRs but imo the merge should not happen without ANY configuration ability.

Thanks!

jakobmoellerdev · 2025-05-13T13:44:06Z

api/ocm/tools/transfer/transfer.go

+			i, r := i, r
+			tasks <- transferTask{
+				id: fmt.Sprintf("resource-%d", i),
+				task: func() error {


This functional transfer task is very hard to read and debug, so I suggest you split this into a separate function.

dynamic-Archu · 2025-05-13T15:38:18Z

Hi @jakobmoellerdev,

Thank you for the review and feedback ! I'll look into the suggestions you provided.

In the meantime, I wanted to raise a concern I noticed while testing the OCM transfer process locally using the latest version of the OCM codebase. When I run it after cloning, I consistently encounter the following error:

Error: transfer resources and sources: commit and ref cannot be specified at the same time

This error appears even when using the latest OCM transfer.go code (without my changes), and although the transfer process completes, the blobs are not getting transferred to the JFrog repository.
When I run the same with my modified code, the components are getting transferred correctly, but the blobs are still missing in the JFrog repo.

It would be great if someone could help verify this behavior and share any insights.

jakobmoellerdev · 2025-05-13T18:43:20Z

There is a related change I did quite some time back that influenced the github access method: #1406 i introduced this limitation because ref and commit are mutually exclusive to the access method. Im assuming you have incorrect specs in your component version

Updated existing code with new changes based on the previous feedbacks

dynamic-Archu · 2025-05-16T12:08:30Z

Hi @jakobmoellerdev ,

Thank you for the previous review and inputs! I've made the necessary changes based on your comments. Could you please review the latest changes and let me know if everything looks good?

Additionally, I wanted to follow up regarding the issue I'm facing during local testing - the blobs are still not getting transferred and I'm consistently seeing the following error: transfer resources and sources: commit and ref cannot be specified at the same time

You also suggested that the issue could be due to incorrect specs in the component version.
Could this error also be the reason why the blobs are not being transferred?

Looking forward to your input. Thanks again!

jakobmoellerdev · 2025-05-16T12:20:13Z

The issue you were facing is a regression we reverted now in the latest RCs and on main, you shouldnt have this error anymore. :)

for the lint errors

lint errors

dynamic-Archu · 2025-05-26T04:45:48Z

Hi @jakobmoellerdev,

There are a couple of checks that are failing, and I’d appreciate your guidance on how to proceed:

Test Failure – There seems to be a mismatch in the expected vs. actual output in components/transfer/cmd_test.go under transfers ctf.
gci Formatting Error – I'm also encountering an import sorting issue flagged by gci in components/transfer/ctf/ctf.go.

Could you please share some insights or point me in the right direction to resolve these?

Thanks in advance!

jakobmoellerdev

Regarding the lint error. you shoud fixup your import ordering on the project such that

  gci:
    sections:
      - standard
      - blank
      - dot
      - default
      - prefix(ocm.software/ocm)
    custom-order: true

is set. your file edit breaks these rules. please see the GCI lint documentation on how to fix that and your IDE documentation on how to configure import ordering. You can also run golangci-lint in --fix mode to correct this automatically.

Regarding the test run:
you are changing the way that notifyArtifactInfo is called in your transfer logic.

In your code https://github.com/open-component-model/ocm/pull/1420/files#diff-d341af7fc25e4e0ea76361047d0e578a1e197d1e598fcba3ffc5ce866c42358dR316 I am assuming that you changed how changed and valueNeeded are evaluated and that causes a different handling. Please double check that the behavior on the breaking test is the same as before your change.

Before:

var old compdesc.Resource

				hint := ocmcpi.ArtifactNameHint(a, src)
				old, err = cur.GetResourceByIdentity(r.Meta().GetIdentity(srccd.Resources))

				changed := err != nil || old.Digest == nil || !old.Digest.Equal(r.Meta().Digest)
				valueNeeded := err == nil && needsTransport(src.GetContext(), r, &old)
				if changed || valueNeeded {
					var msgs []interface{}
					if !errors.IsErrNotFound(err) {
						if err != nil {
							return err
						}
						if !changed && valueNeeded {
							msgs = []interface{}{"copy"}
						} else {
							msgs = []interface{}{"overwrite"}
						}
					}
					notifyArtifactInfo(printer, log, "resource", i, r.Meta(), hint, msgs...)
					err = handler.HandleTransferResource(r, m, hint, t)
				} else {
					if err == nil { // old resource found -> keep current access method
						t.SetResource(r.Meta(), old.Access, ocm.ModifyElement(), ocm.SkipVerify(), ocm.DisableExtraIdentityDefaulting())
					}
					notifyArtifactInfo(printer, log, "resource", i, r.Meta(), hint, "already present")
				}

after

				hint := ocmcpi.ArtifactNameHint(a, src)
				old, err := cur.GetResourceByIdentity(r.Meta().GetIdentity(srccd.Resources))
				changed := err != nil || old.Digest == nil || !old.Digest.Equal(r.Meta().Digest)
				valueNeeded := err == nil && needsTransport(src.GetContext(), r, &old)
				if changed || valueNeeded {
					notifyArtifactInfo(printer, log, "resource", i, r.Meta(), hint, "copy")
					return handler.HandleTransferResource(r, m, hint, t)
				} else if err == nil {
					t.SetResource(r.Meta(), old.Access, ocm.ModifyElement(), ocm.SkipVerify(), ocm.DisableExtraIdentityDefaulting())
					notifyArtifactInfo(printer, log, "resource", i, r.Meta(), hint, "already present")
				}

api/ocm/tools/transfer/transfer.go

jakobmoellerdev · 2025-05-27T07:20:24Z

api/ocm/tools/transfer/transfer.go

 	}

+	// subp := common.AddPrinterGap(ctx, "  ")
+	var wg sync.WaitGroup


This comment is still not resolved

api/ocm/tools/transfer/transfer.go

jakobmoellerdev · 2025-05-27T07:24:08Z

api/ocm/tools/transfer/transfer.go

-
-		nested := finalize.Nested()
+	log.Info("  transferring resources and sources using worker pool", "workers", maxWorkers)
+	tasks := make(chan transferTask)


can you not make the tasks a bounded channel too?

Yes, absolutely! I've now updated the tasks channel to be a bounded (buffered) channel within the copyVersionWithWorkerPool function.

This helps to:

Decouple the task producer from the workers slightly.

Smooth out throughput by allowing a small backlog of tasks to accumulate.

Control memory usage by limiting the number of in-flight tasks.

The buffer size is set to maxWorkers * 2 (with a minimum of 1), which provides a good balance.

Here's the updated line in copyVersionWithWorkerPool:

// ... log.Info(" transferring resources and sources using worker pool", "workers", maxWorkers) // Make tasks a bounded (buffered) channel to smooth out task distribution. taskBufferSize := maxWorkers * 2 if taskBufferSize < 1 { // Ensure a minimum buffer size taskBufferSize = 1 } tasks := make(chan transferTask, taskBufferSize) // <-- Now a buffered channel errChan := make(chan error, len(src.GetResources())+len(src.GetSources())) // ...

Thanks for the suggestion!

I do not think it is necessary for you to introduce an unknown buffer size. instead you can introduce a goroutine pool that is buffered to the number of resources+sources no? (same as the errChan)

jakobmoellerdev · 2025-05-27T07:24:36Z

api/ocm/tools/transfer/transfer.go

+	}
+
+	go func() {
+		for i, r := range src.GetResources() {


why do you need to run a go routine to submit transfer tasks to the pool?

The task submission loop (go func() { ... tasks <- ... close(tasks) }) runs in its own goroutine primarily to:

Avoid Deadlock: If it ran in the main goroutine, sending tasks could block (if the channel is full), preventing wg.Wait() and close(tasks) from being reached.

Proper Channel Closure: Ensures close(tasks) is called only after all tasks have been submitted, allowing worker range loops to terminate correctly.

This pattern allows the task producer to run concurrently with the workers and manage channel lifecycle gracefully.

I understand why in general a submission is done in a goroutine. However, im wondering why you need it. You know the length of elements in the channel in advance so it is always buffered correctly and can never lock. as such you do not have a deadlock. that also simplifies channel closure.

Attribute update

code

Update init.go

jakobmoellerdev

This looks great already, thanks for all the efforts 🥳 There are some minor corrections and one major question to answer which we already briefly discussed before:

Do you want to change the default OCM behavior to run concurrent transfers. If yes, then how do we retain the old serialized ordered behavior? It is such a big switch that we have to make sure users can opt in to the parallel transfer IMO.

jakobmoellerdev · 2025-07-16T11:04:50Z

api/ocm/tools/transfer/transfer.go

 type TransportClosure = common.NameVersionInfo[*struct{}]

+// TransferWorkersEnvVar is the environment variable to configure the number of transfer workers.
+const TransferWorkersEnvVar = "OCM_TRANSFER_WORKERS" // This constant is now technically unused, but kept for context.


please make the env variable read out inside the attribute coding and change to OCM_MAX_WORKERS

jakobmoellerdev · 2025-07-16T11:06:00Z

api/ocm/tools/transfer/transfer.go

 	}

+	// subp := common.AddPrinterGap(ctx, "  ")
+	var wg sync.WaitGroup


Is the expectation here that once one of the go routine fails, all others should abort? if so, then you should use an errgroup.WithContext. WDYT?

jakobmoellerdev · 2025-07-16T11:06:19Z

api/ocm/tools/transfer/transfer.go

-		// If no copy is done, merge must keep the access methods in target!!!
 		if !doMerge || doCopy {
-			err = copyVersion(ctx, printer, log, state.History, src, t, n, handler)
+			// *** FIX: Pass 'ctx' argument here ***


Missing TODO

jakobmoellerdev · 2025-07-16T11:06:48Z

api/ocm/tools/transfer/transfer.go

+	// If attributeWorkers is 0, it means either user explicitly set 0, or attribute was not set (default to 0).
+	// In both cases, this signals to use the CPU-based auto-detection.
+	if attributeWorkers == 0 {
+		return determineWorkersFromCPU()


This default should become part of the attribute get logic imo

jakobmoellerdev · 2025-07-16T11:08:57Z

Before I forget, we introduced a DCO requirement so you need to sign off your commits now!

comment changes Signed-off-by: Archanaa N <[email protected]>

Signed-off-by: Archanaa [email protected]

jakobmoellerdev

We have quite a few people that look forward to this. if you need help in resolving open comments, please feel free to join our community call!

https://ocm.software/community/engagement/#community-calls

jakobmoellerdev · 2025-09-03T07:13:08Z

api/ocm/tools/transfer/transfer.go

-			if err != nil {
-				unstr = nil
+			// Capture the error specifically for this operation
+			specErr := error(nil) // Declare a local error variable


Suggested change

specErr := error(nil) // Declare a local error variable

var specErr error

jakobmoellerdev · 2025-09-03T07:13:28Z

api/ocm/tools/transfer/transfer.go

+			// Capture the error specifically for this operation
+			specErr := error(nil) // Declare a local error variable
+			unstr, specErr = runtimeutil.ToUnstructuredTypedObject(tgt.GetSpecification())
+			if specErr != nil {


inline this assignment and if statement

jakobmoellerdev · 2025-09-03T07:14:02Z

api/ocm/tools/transfer/transfer.go

+func calculateEffectiveTransferWorkers(ctx context.Context) int {
+	// First, obtain the OCM data context from the provided Go context.
+	// `ocm.DefaultContext()` is a common way to get it if it's not directly `datacontext.Context`.
+	ocmCtx := ocm.DefaultContext()


shouldnt the ocm context be retrieved from the caller context?

Patch 1

Revert "Patch 1"

dynamic-Archu · 2025-10-06T08:02:50Z

Hi @jakobmoellerdev , could you please trigger all the remaining test cases ?

jakobmoellerdev · 2025-10-06T11:34:24Z

Hey there: tests/lint still failing, PTAL. Also please sign and make sure DCO is passing. also run go generate as there are diffs. Thanks!

Signed-off-by: Archanaa.N <[email protected]>

Updated transfer.go

2c89f38

Optimized approach for OCM transfer by implementing a concurrent worker pool, replacing the previous sequential transfer logic.

dynamic-Archu requested a review from a team as a code owner April 22, 2025 10:24

github-actions bot added the size/m Medium label Apr 22, 2025

dynamic-Archu changed the title ~~Updated transfer.go~~ feat(transfer): Optimized approach for OCM transfer by implementing a concurrent worker pool. Apr 22, 2025

github-actions bot added the kind/feature new feature, enhancement, improvement, extension label Apr 22, 2025

Merge branch 'main' into dynamic-Archu-patch-1

2bc54fb

frewilhelm added the needs/validation Validate the issue and assign a priority label Apr 22, 2025

jakobmoellerdev requested changes Apr 23, 2025

View reviewed changes

dynamic-Archu added 2 commits April 24, 2025 14:18

Merge branch 'main' into dynamic-Archu-patch-1

f7352f2

Updated transfer.go

bf1f9e5

Updated existing implementation with new changes

jakobmoellerdev reviewed May 13, 2025

View reviewed changes

Update transfer.go

8c30e54

Updated existing code with new changes based on the previous feedbacks

dynamic-Archu added 5 commits May 17, 2025 11:47

Merge branch 'main' into dynamic-Archu-patch-1

01c7a17

Updated transfer.go

45c44cb

for the lint errors

Merge branch 'main' into dynamic-Archu-patch-1

182aeab

Update transfer.go

30d6f56

lint errors

Merge branch 'main' into dynamic-Archu-patch-1

1c40c7f

Merge branch 'main' into dynamic-Archu-patch-1

3f8feb0

jakobmoellerdev reviewed May 27, 2025

View reviewed changes

dynamic-Archu added 4 commits June 6, 2025 10:03

Update transfer.go

8679921

Update transfer.go

652e25c

Update transfer.go

4e19e51

Merge branch 'main' into dynamic-Archu-patch-1

eb15c0b

dynamic-Archu added 9 commits July 2, 2025 11:22

Merge branch 'main' into dynamic-Archu-patch-1

8ef7222

Merge branch 'main' into dynamic-Archu-patch-1

167ca50

Merge branch 'main' into dynamic-Archu-patch-1

d776796

Attribute update transfer.go

134034e

Attribute update

Create maxworkersattr.go

71780af

Update maxworkersattr.go

908c503

code

Update init.go

040ffef

Update init.go

Merge branch 'main' into dynamic-Archu-patch-1

4f36d21

Merge branch 'main' into dynamic-Archu-patch-1

fad83e5

jakobmoellerdev requested changes Jul 16, 2025

View reviewed changes

dynamic-Archu and others added 3 commits July 16, 2025 21:09

Update transfer.go - comment changes

f963953

comment changes Signed-off-by: Archanaa N <[email protected]>

Update transfer.go

8350f2c

Signed-off-by: Archanaa [email protected]

Merge branch 'main' into dynamic-Archu-patch-1

a2424ab

jakobmoellerdev reviewed Sep 3, 2025

View reviewed changes

jakobmoellerdev removed the needs/validation Validate the issue and assign a priority label Sep 3, 2025

Umang2608 and others added 3 commits September 26, 2025 18:00

Update transfer.go

29e9ce7

Update maxworkersattr.go

a8e1082

Merge pull request #2 from Umang2608/patch-1

114f00b

Patch 1

github-actions bot added the size/l Large label Oct 3, 2025

dynamic-Archu added 5 commits October 3, 2025 16:26

Revert "Patch 1"

862b3aa

Merge pull request #3 from dynamic-Archu/revert-2-patch-1

e2ec35b

Revert "Patch 1"

Merge branch 'main' into dynamic-Archu-patch-1

6979f2a

Update transfer_u.go

e345377

Update maxworkersattr_u.go

9db9057

dynamic-Archu added 3 commits October 7, 2025 10:39

Update transfer.go

54830a1

Merge branch 'main' into dynamic-Archu-patch-1

3ef9a1d

Update transfer.go

5573842

Signed-off-by: Archanaa.N <[email protected]>

	specErr := error(nil) // Declare a local error variable
	var specErr error

feat(transfer): Optimized approach for OCM transfer by implementing a concurrent worker pool. #1420

Are you sure you want to change the base?

feat(transfer): Optimized approach for OCM transfer by implementing a concurrent worker pool. #1420

Conversation

dynamic-Archu commented Apr 22, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What this PR does / why we need it

Which issue(s) this PR fixes

Uh oh!

jakobmoellerdev left a comment

Choose a reason for hiding this comment

Uh oh!

dynamic-Archu commented Apr 24, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jakobmoellerdev commented Apr 24, 2025 • 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!

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!

dynamic-Archu commented May 13, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jakobmoellerdev commented May 13, 2025

Uh oh!

dynamic-Archu commented May 16, 2025

Uh oh!

jakobmoellerdev commented May 16, 2025

Uh oh!

dynamic-Archu commented May 26, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jakobmoellerdev left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dynamic-Archu Jun 23, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

jakobmoellerdev left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

jakobmoellerdev commented Jul 16, 2025

Uh oh!

jakobmoellerdev left a comment

Choose a reason for hiding this comment

Uh oh!

dynamic-Archu commented Apr 22, 2025 •

edited

Loading

dynamic-Archu commented Apr 24, 2025 •

edited

Loading

jakobmoellerdev commented Apr 24, 2025 •

edited

Loading

dynamic-Archu commented May 13, 2025 •

edited

Loading

dynamic-Archu commented May 26, 2025 •

edited

Loading

dynamic-Archu Jun 23, 2025 •

edited

Loading

jakobmoellerdev left a comment •

edited

Loading

dynamic-Archu Sep 8, 2025 •

edited

Loading