-
Notifications
You must be signed in to change notification settings - Fork 973
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge branch 'main' into shwap-bitswap-get-size
- Loading branch information
Showing
46 changed files
with
781 additions
and
678 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -12,16 +12,6 @@ on: | |
| pull_request: | ||
| workflow_dispatch: | ||
| inputs: | ||
| version-bump: | ||
| # Friendly description to be shown in the UI instead of 'name' | ||
| description: "Semver type of new version (major / minor / patch)" | ||
| # Input has to be provided for the workflow to run | ||
| required: true | ||
| type: choice | ||
| options: | ||
| - patch | ||
| - minor | ||
| - major | ||
| tag-as: | ||
| description: 'Tag for snapshot release (optional)' | ||
| required: false | ||
|
|
@@ -88,31 +78,9 @@ jobs: | |
| with: | ||
| go-version: ${{ needs.setup.outputs.go-version }} | ||
|
|
||
| # If this was a workflow dispatch event, we need to generate and push a tag | ||
| # for goreleaser to grab | ||
| version_bump: | ||
| needs: [hadolint, yamllint, markdown-lint, go-ci, setup] | ||
| runs-on: ubuntu-latest | ||
| permissions: "write-all" | ||
| steps: | ||
| - uses: actions/checkout@v4 | ||
|
|
||
| - name: Bump version and push tag | ||
| # Placing the if condition here is a workaround for needing to block | ||
| # on this step during workflow dispatch events but the step not | ||
| # needing to run on tags. If we had the if condition on the full | ||
| # version_bump section, it would skip and not run, which would result | ||
| # in goreleaser not running either. | ||
| if: ${{ github.event_name == 'workflow_dispatch' }} | ||
| uses: mathieudutour/[email protected] | ||
| with: | ||
| github_token: ${{ secrets.GITHUB_TOKEN }} | ||
| default_bump: ${{ inputs.version-bump }} | ||
| release_branches: ${{ needs.setup.outputs.branch }} | ||
|
|
||
| # Generate the release with goreleaser to include pre-built binaries | ||
| goreleaser: | ||
| needs: [version_bump, setup] | ||
| needs: [setup] | ||
| runs-on: ubuntu-latest | ||
| if: | | ||
| github.event_name == 'workflow_dispatch' || | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,45 @@ | ||
| package utils | ||
|
|
||
| import ( | ||
| "context" | ||
| "sync" | ||
| ) | ||
|
|
||
| // Sessions manages concurrent sessions for the specified key. | ||
| // It ensures only one session can proceed for each key, avoiding duplicate efforts. | ||
| // If a session is already active for the given key, it waits until the session completes or | ||
| // context error occurs. | ||
| type Sessions struct { | ||
| active sync.Map | ||
| } | ||
|
|
||
| func NewSessions() *Sessions { | ||
| return &Sessions{} | ||
| } | ||
|
|
||
| // StartSession attempts to start a new session for the given key. It provides a release function | ||
| // to clean up the session lock for this key, once the session is complete. | ||
| func (s *Sessions) StartSession( | ||
| ctx context.Context, | ||
| key any, | ||
| ) (endSession func(), err error) { | ||
| // Attempt to load or initialize a channel to track the sampling session for this height | ||
| lockChan, alreadyActive := s.active.LoadOrStore(key, make(chan struct{})) | ||
| if alreadyActive { | ||
| // If a session is already active, wait for it to complete | ||
| select { | ||
| case <-lockChan.(chan struct{}): | ||
| case <-ctx.Done(): | ||
| return func() {}, ctx.Err() | ||
| } | ||
| // previous session has completed, try to obtain the lock for this session | ||
| return s.StartSession(ctx, key) | ||
| } | ||
|
|
||
| // Provide a function to release the lock once session is complete | ||
| releaseLock := func() { | ||
| close(lockChan.(chan struct{})) | ||
| s.active.Delete(key) | ||
| } | ||
| return releaseLock, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,127 @@ | ||
| package utils | ||
|
|
||
| import ( | ||
| "context" | ||
| "sync" | ||
| "sync/atomic" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| // TestSessionsSerialExecution verifies that multiple sessions for the same key are executed | ||
| // sequentially. | ||
| func TestSessionsSerialExecution(t *testing.T) { | ||
| ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second) | ||
| t.Cleanup(cancel) | ||
|
|
||
| sessions := NewSessions() | ||
| key := "testKey" | ||
| activeCount := atomic.Int32{} | ||
| var wg sync.WaitGroup | ||
|
|
||
| numSessions := 20 | ||
|
|
||
| for i := 0; i < numSessions; i++ { | ||
| wg.Add(1) | ||
| go func(id int) { | ||
| defer wg.Done() | ||
| endSession, err := sessions.StartSession(ctx, key) | ||
| require.NoError(t, err) | ||
| old := activeCount.Add(1) | ||
| require.Equal(t, int32(1), old) | ||
| // Simulate some work | ||
| time.Sleep(50 * time.Millisecond) | ||
| old = activeCount.Add(-1) | ||
| require.Equal(t, int32(0), old) | ||
| // Release the session | ||
| endSession() | ||
| }(i) | ||
| } | ||
|
|
||
| wg.Wait() | ||
| } | ||
|
|
||
| func TestSessionsContextCancellation(t *testing.T) { | ||
| sessions := NewSessions() | ||
| key := "testCancelKey" | ||
|
|
||
| // Start the first session which will hold the lock for a while | ||
| wg := sync.WaitGroup{} | ||
| wg.Add(1) | ||
| go func() { | ||
| defer wg.Done() | ||
| ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) | ||
| defer cancel() | ||
|
|
||
| release, err := sessions.StartSession(ctx, key) | ||
| if err != nil { | ||
| t.Errorf("First session: failed to start: %v", err) | ||
| return | ||
| } | ||
|
|
||
| // Hold the session for 1 second | ||
| time.Sleep(1 * time.Second) | ||
| release() | ||
| }() | ||
|
|
||
| // Give the first goroutine a moment to acquire the session | ||
| time.Sleep(100 * time.Millisecond) | ||
|
|
||
| // Attempt to start a second session with a context that times out before the first session releases | ||
| ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) | ||
| t.Cleanup(cancel) | ||
|
|
||
| _, err := sessions.StartSession(ctx, key) | ||
| require.ErrorIs(t, err, context.DeadlineExceeded) | ||
|
|
||
| // Attempt to start a second session with a context that is canceled before the first session | ||
| // releases | ||
| ctx, cancel = context.WithCancel(context.Background()) | ||
| cancel() | ||
|
|
||
| _, err = sessions.StartSession(ctx, key) | ||
| require.ErrorIs(t, err, context.Canceled) | ||
|
|
||
| wg.Wait() | ||
| } | ||
|
|
||
| // TestSessions_ConcurrentDifferentKeys ensures that sessions with different keys run concurrently. | ||
| func TestSessions_ConcurrentDifferentKeys(t *testing.T) { | ||
| sessions := NewSessions() | ||
| numKeys := 20 | ||
| var wg sync.WaitGroup | ||
| startCh := make(chan struct{}) | ||
| activeSessions := atomic.Int32{} | ||
| maxActive := int32(0) | ||
|
|
||
| for i := 0; i < numKeys; i++ { | ||
| wg.Add(1) | ||
| go func(key int) { | ||
| defer wg.Done() | ||
| ctx := context.Background() | ||
| endSession, err := sessions.StartSession(ctx, key) | ||
| require.NoError(t, err) | ||
|
|
||
| active := activeSessions.Add(1) | ||
| if active > maxActive { | ||
| maxActive = active | ||
| } | ||
|
|
||
| // Wait to simulate work | ||
| time.Sleep(100 * time.Millisecond) | ||
|
|
||
| activeSessions.Add(-1) | ||
| endSession() | ||
| }(i) | ||
| } | ||
|
|
||
| // Start all goroutines | ||
| close(startCh) | ||
| wg.Wait() | ||
|
|
||
| if maxActive > int32(numKeys) { | ||
| t.Errorf("Expected %d concurrent active sessions, but got %d", numKeys, maxActive) | ||
| } | ||
| } |
Oops, something went wrong.