Skip to content

Commit

Permalink
Limit to single inflight package syncing operation (#289)
Browse files Browse the repository at this point in the history 
This PR revives work done in previous PRs (#118, #120) to make sure that only a single package syncing operation is ever in flight and also adds a test.

The previous PRs did not account for needing to also protect `initStatuses` and `SetPackageStatuses`, so that's why the Lock and Unlock statements are not just paired in doSync. If you think the intent would be clearer using a sync.WaitGroup, let me know.

The new test makes sure that the mutex correctly protects the local storage; if we comment out the calls to Lock/Unlock and run the test with the `-race` flag we can see the race condition taking place

<details>

```
# Without using the mutex we can see the race condition of messages sent in parallel
$ go test -run=TestPackageUpdatesInParallel -v -race -count=1
=== RUN   TestPackageUpdatesInParallel
==================
WARNING: DATA RACE
Write at 0x00c00003cdb0 by goroutine 10:
  github.com/open-telemetry/opamp-go/client/internal.(*InMemPackagesStore).SetLastReportedStatuses()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/inmempackagestore.go:95 +0x34
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).reportStatuses()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:335 +0x78
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).syncPackage()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:199 +0x4dc
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).doSync()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:132 +0x3b0
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).Sync.gowrap1()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:69 +0x4c

Previous read at 0x00c00003cdb0 by goroutine 8:
  github.com/open-telemetry/opamp-go/client/internal.(*InMemPackagesStore).LastReportedStatuses()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/inmempackagestore.go:91 +0x30
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).initStatuses()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:88 +0x64
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).Sync()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:59 +0x78
  github.com/open-telemetry/opamp-go/client/internal.TestPackageUpdatesInParallel.func1()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/httpsender_test.go:185 +0x68
  github.com/open-telemetry/opamp-go/client/types.CallbacksStruct.OnMessage()
      /Users/tpaschalis/GitRepos/opamp-go/client/types/callbacks.go:161 +0x84
  github.com/open-telemetry/opamp-go/client/types.(*CallbacksStruct).OnMessage()
      <autogenerated>:1 +0x20
  github.com/open-telemetry/opamp-go/client/internal.(*receivedProcessor).ProcessReceivedMessage()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/receivedprocessor.go:160 +0xe94
  github.com/open-telemetry/opamp-go/client/internal.TestPackageUpdatesInParallel.func2()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/httpsender_test.go:197 +0x4d0

Goroutine 10 (running) created at:
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).Sync()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:69 +0x15c
  github.com/open-telemetry/opamp-go/client/internal.TestPackageUpdatesInParallel.func1()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/httpsender_test.go:185 +0x68
  github.com/open-telemetry/opamp-go/client/types.CallbacksStruct.OnMessage()
      /Users/tpaschalis/GitRepos/opamp-go/client/types/callbacks.go:161 +0x84
  github.com/open-telemetry/opamp-go/client/types.(*CallbacksStruct).OnMessage()
      <autogenerated>:1 +0x20
  github.com/open-telemetry/opamp-go/client/internal.(*receivedProcessor).ProcessReceivedMessage()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/receivedprocessor.go:160 +0xe94
  github.com/open-telemetry/opamp-go/client/internal.TestPackageUpdatesInParallel.func3()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/httpsender_test.go:216 +0x4d0

Goroutine 8 (finished) created at:
  github.com/open-telemetry/opamp-go/client/internal.TestPackageUpdatesInParallel()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/httpsender_test.go:196 +0x4b0
  testing.tRunner()
      /opt/homebrew/Cellar/go/1.22.2/libexec/src/testing/testing.go:1689 +0x180
  testing.(*T).Run.gowrap1()
      /opt/homebrew/Cellar/go/1.22.2/libexec/src/testing/testing.go:1742 +0x40
==================
==================
WARNING: DATA RACE
Write at 0x00c0000999e0 by goroutine 11:
  runtime.mapaccess2_faststr()
      /opt/homebrew/Cellar/go/1.22.2/libexec/src/runtime/map_faststr.go:108 +0x42c
  github.com/open-telemetry/opamp-go/client/internal.(*InMemPackagesStore).CreatePackage()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/inmempackagestore.go:50 +0x74
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).syncPackage()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:203 +0x528
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).doSync()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:132 +0x3b0
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).Sync.gowrap1()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:69 +0x4c

Previous read at 0x00c0000999e0 by goroutine 10:
  runtime.mapdelete()
      /opt/homebrew/Cellar/go/1.22.2/libexec/src/runtime/map.go:696 +0x43c
  github.com/open-telemetry/opamp-go/client/internal.(*InMemPackagesStore).Packages()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/inmempackagestore.go:36 +0x64
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).deleteUnneededLocalPackages()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:303 +0x58
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).doSync()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:125 +0x1b4
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).Sync.gowrap1()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:69 +0x4c

Goroutine 11 (running) created at:
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).Sync()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:69 +0x15c
  github.com/open-telemetry/opamp-go/client/internal.TestPackageUpdatesInParallel.func1()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/httpsender_test.go:185 +0x68
  github.com/open-telemetry/opamp-go/client/types.CallbacksStruct.OnMessage()
      /Users/tpaschalis/GitRepos/opamp-go/client/types/callbacks.go:161 +0x84
  github.com/open-telemetry/opamp-go/client/types.(*CallbacksStruct).OnMessage()
      <autogenerated>:1 +0x20
  github.com/open-telemetry/opamp-go/client/internal.(*receivedProcessor).ProcessReceivedMessage()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/receivedprocessor.go:160 +0xe94
  github.com/open-telemetry/opamp-go/client/internal.TestPackageUpdatesInParallel.func2()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/httpsender_test.go:197 +0x4d0

Goroutine 10 (running) created at:
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).Sync()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:69 +0x15c
  github.com/open-telemetry/opamp-go/client/internal.TestPackageUpdatesInParallel.func1()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/httpsender_test.go:185 +0x68
  github.com/open-telemetry/opamp-go/client/types.CallbacksStruct.OnMessage()
      /Users/tpaschalis/GitRepos/opamp-go/client/types/callbacks.go:161 +0x84
  github.com/open-telemetry/opamp-go/client/types.(*CallbacksStruct).OnMessage()
      <autogenerated>:1 +0x20
  github.com/open-telemetry/opamp-go/client/internal.(*receivedProcessor).ProcessReceivedMessage()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/receivedprocessor.go:160 +0xe94
  github.com/open-telemetry/opamp-go/client/internal.TestPackageUpdatesInParallel.func3()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/httpsender_test.go:216 +0x4d0
==================
==================
WARNING: DATA RACE
Write at 0x00c00003cdb0 by goroutine 11:
  github.com/open-telemetry/opamp-go/client/internal.(*InMemPackagesStore).SetLastReportedStatuses()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/inmempackagestore.go:95 +0x34
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).reportStatuses()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:335 +0x78
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).syncPackage()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:229 +0x780
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).doSync()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:132 +0x3b0
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).Sync.gowrap1()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:69 +0x4c

Previous write at 0x00c00003cdb0 by goroutine 10:
  github.com/open-telemetry/opamp-go/client/internal.(*InMemPackagesStore).SetLastReportedStatuses()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/inmempackagestore.go:95 +0x34
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).reportStatuses()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:335 +0x78
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).syncPackage()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:229 +0x780
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).doSync()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:132 +0x3b0
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).Sync.gowrap1()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:69 +0x4c

Goroutine 11 (running) created at:
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).Sync()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:69 +0x15c
  github.com/open-telemetry/opamp-go/client/internal.TestPackageUpdatesInParallel.func1()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/httpsender_test.go:185 +0x68
  github.com/open-telemetry/opamp-go/client/types.CallbacksStruct.OnMessage()
      /Users/tpaschalis/GitRepos/opamp-go/client/types/callbacks.go:161 +0x84
  github.com/open-telemetry/opamp-go/client/types.(*CallbacksStruct).OnMessage()
      <autogenerated>:1 +0x20
  github.com/open-telemetry/opamp-go/client/internal.(*receivedProcessor).ProcessReceivedMessage()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/receivedprocessor.go:160 +0xe94
  github.com/open-telemetry/opamp-go/client/internal.TestPackageUpdatesInParallel.func2()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/httpsender_test.go:197 +0x4d0

Goroutine 10 (running) created at:
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).Sync()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:69 +0x15c
  github.com/open-telemetry/opamp-go/client/internal.TestPackageUpdatesInParallel.func1()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/httpsender_test.go:185 +0x68
  github.com/open-telemetry/opamp-go/client/types.CallbacksStruct.OnMessage()
      /Users/tpaschalis/GitRepos/opamp-go/client/types/callbacks.go:161 +0x84
  github.com/open-telemetry/opamp-go/client/types.(*CallbacksStruct).OnMessage()
      <autogenerated>:1 +0x20
  github.com/open-telemetry/opamp-go/client/internal.(*receivedProcessor).ProcessReceivedMessage()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/receivedprocessor.go:160 +0xe94
  github.com/open-telemetry/opamp-go/client/internal.TestPackageUpdatesInParallel.func3()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/httpsender_test.go:216 +0x4d0
==================
==================
WARNING: DATA RACE
Write at 0x00c00003cdb0 by goroutine 10:
  github.com/open-telemetry/opamp-go/client/internal.(*InMemPackagesStore).SetLastReportedStatuses()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/inmempackagestore.go:95 +0x34
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).reportStatuses()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:335 +0x78
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).doSync()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:151 +0x69c
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).Sync.gowrap1()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:69 +0x4c

Previous write at 0x00c00003cdb0 by goroutine 11:
  github.com/open-telemetry/opamp-go/client/internal.(*InMemPackagesStore).SetLastReportedStatuses()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/inmempackagestore.go:95 +0x34
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).reportStatuses()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:335 +0x78
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).doSync()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:151 +0x69c
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).Sync.gowrap1()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:69 +0x4c

Goroutine 10 (running) created at:
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).Sync()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:69 +0x15c
  github.com/open-telemetry/opamp-go/client/internal.TestPackageUpdatesInParallel.func1()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/httpsender_test.go:185 +0x68
  github.com/open-telemetry/opamp-go/client/types.CallbacksStruct.OnMessage()
      /Users/tpaschalis/GitRepos/opamp-go/client/types/callbacks.go:161 +0x84
  github.com/open-telemetry/opamp-go/client/types.(*CallbacksStruct).OnMessage()
      <autogenerated>:1 +0x20
  github.com/open-telemetry/opamp-go/client/internal.(*receivedProcessor).ProcessReceivedMessage()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/receivedprocessor.go:160 +0xe94
  github.com/open-telemetry/opamp-go/client/internal.TestPackageUpdatesInParallel.func3()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/httpsender_test.go:216 +0x4d0

Goroutine 11 (running) created at:
  github.com/open-telemetry/opamp-go/client/internal.(*packagesSyncer).Sync()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/packagessyncer.go:69 +0x15c
  github.com/open-telemetry/opamp-go/client/internal.TestPackageUpdatesInParallel.func1()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/httpsender_test.go:185 +0x68
  github.com/open-telemetry/opamp-go/client/types.CallbacksStruct.OnMessage()
      /Users/tpaschalis/GitRepos/opamp-go/client/types/callbacks.go:161 +0x84
  github.com/open-telemetry/opamp-go/client/types.(*CallbacksStruct).OnMessage()
      <autogenerated>:1 +0x20
  github.com/open-telemetry/opamp-go/client/internal.(*receivedProcessor).ProcessReceivedMessage()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/receivedprocessor.go:160 +0xe94
  github.com/open-telemetry/opamp-go/client/internal.TestPackageUpdatesInParallel.func2()
      /Users/tpaschalis/GitRepos/opamp-go/client/internal/httpsender_test.go:197 +0x4d0
==================
    testing.go:1398: race detected during execution of test
--- FAIL: TestPackageUpdatesInParallel (0.10s)
FAIL
exit status 1
FAIL	github.com/open-telemetry/opamp-go/client/internal	0.286s
```

</details>

Fixes #84
  • Loading branch information
@tpaschalis
tpaschalis authored Aug 29, 2024
1 parent 182ce37 commit 536037b
Show file tree
Hide file tree
Showing 10 changed files with 251 additions and 7 deletions.
1 change: 1 addition & 0 deletions client/httpclient.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -131,6 +131,7 @@ func (c *httpClient) runUntilStopped(ctx context.Context) {
&c.common.ClientSyncedState,
c.common.PackagesStateProvider,
c.common.Capabilities,
&c.common.PackageSyncMutex,
)
}

Expand Down
3 changes: 3 additions & 0 deletions client/internal/clientcommon.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,9 @@ type ClientCommon struct {
// PackagesStateProvider provides access to the local state of packages.
PackagesStateProvider types.PackagesStateProvider

// PackageSyncMutex makes sure only one package syncing operation happens at a time.
PackageSyncMutex sync.Mutex

// The transport-specific sender.
sender Sender

Expand Down
4 changes: 3 additions & 1 deletion client/internal/httpsender.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (
"fmt"
"io"
"net/http"
"sync"
"sync/atomic"
"time"

Expand Down Expand Up @@ -91,10 +92,11 @@ func (h *HTTPSender) Run(
clientSyncedState *ClientSyncedState,
packagesStateProvider types.PackagesStateProvider,
capabilities protobufs.AgentCapabilities,
packageSyncMutex *sync.Mutex,
) {
h.url = url
h.callbacks = callbacks
h.receiveProcessor = newReceivedProcessor(h.logger, callbacks, h, clientSyncedState, packagesStateProvider, capabilities)
h.receiveProcessor = newReceivedProcessor(h.logger, callbacks, h, clientSyncedState, packagesStateProvider, capabilities, packageSyncMutex)

for {
pollingTimer := time.NewTimer(time.Millisecond * time.Duration(atomic.LoadInt64(&h.pollingIntervalMs)))
Expand Down
126 changes: 125 additions & 1 deletion client/internal/httpsender_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -183,7 +183,7 @@ func TestRequestInstanceUidFlagReset(t *testing.T) {
clientSyncedState := &ClientSyncedState{}
clientSyncedState.SetFlags(protobufs.AgentToServerFlags_AgentToServerFlags_RequestInstanceUid)
capabilities := protobufs.AgentCapabilities_AgentCapabilities_Unspecified
sender.receiveProcessor = newReceivedProcessor(&sharedinternal.NopLogger{}, sender.callbacks, sender, clientSyncedState, nil, capabilities)
sender.receiveProcessor = newReceivedProcessor(&sharedinternal.NopLogger{}, sender.callbacks, sender, clientSyncedState, nil, capabilities, new(sync.Mutex))

// If we process a message with a nil AgentIdentification, or an incorrect NewInstanceUid.
sender.receiveProcessor.ProcessReceivedMessage(ctx,
Expand All @@ -208,3 +208,127 @@ func TestRequestInstanceUidFlagReset(t *testing.T) {
assert.Equal(t, sender.receiveProcessor.clientSyncedState.flags, protobufs.AgentToServerFlags_AgentToServerFlags_Unspecified)
cancel()
}

func TestPackageUpdatesInParallel(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
localPackageState := NewInMemPackagesStore()
sender := NewHTTPSender(&sharedinternal.NopLogger{})
blockSyncCh := make(chan struct{})
doneCh := make([]<-chan struct{}, 0)

// Use `ch` to simulate blocking behavior on the second call to Sync().
// This will allow both Sync() calls to be called in parallel; we will
// first make sure that both are inflight before manually releasing the
// channel so that both go through in sequence.
localPackageState.onAllPackagesHash = func() {
if localPackageState.lastReportedStatuses != nil {
<-blockSyncCh
}
}

var messages atomic.Int32
var mux sync.Mutex
sender.callbacks = types.CallbacksStruct{
OnMessageFunc: func(ctx context.Context, msg *types.MessageData) {
err := msg.PackageSyncer.Sync(ctx)
assert.NoError(t, err)
messages.Add(1)
doneCh = append(doneCh, msg.PackageSyncer.Done())
},
}

clientSyncedState := &ClientSyncedState{}
capabilities := protobufs.AgentCapabilities_AgentCapabilities_AcceptsPackages
sender.receiveProcessor = newReceivedProcessor(&sharedinternal.NopLogger{}, sender.callbacks, sender, clientSyncedState, localPackageState, capabilities, &mux)

sender.receiveProcessor.ProcessReceivedMessage(ctx,
&protobufs.ServerToAgent{
PackagesAvailable: &protobufs.PackagesAvailable{
Packages: map[string]*protobufs.PackageAvailable{
"package1": {
Type: protobufs.PackageType_PackageType_TopLevel,
Version: "1.0.0",
File: &protobufs.DownloadableFile{
DownloadUrl: "foo",
ContentHash: []byte{4, 5},
},
Hash: []byte{1, 2, 3},
},
},
AllPackagesHash: []byte{1, 2, 3, 4, 5},
},
})
sender.receiveProcessor.ProcessReceivedMessage(ctx,
&protobufs.ServerToAgent{
PackagesAvailable: &protobufs.PackagesAvailable{
Packages: map[string]*protobufs.PackageAvailable{
"package22": {
Type: protobufs.PackageType_PackageType_TopLevel,
Version: "1.0.0",
File: &protobufs.DownloadableFile{
DownloadUrl: "bar",
ContentHash: []byte{4, 5},
},
Hash: []byte{1, 2, 3},
},
},
AllPackagesHash: []byte{1, 2, 3, 4, 5},
},
})

// Make sure that both Sync calls have gone through _before_ releasing the first one.
// This means that they're both called in parallel, and that the race
// detector would always report a race condition, but proper locking makes
// sure that's not the case.
assert.Eventually(t, func() bool {
return messages.Load() == 2
}, 2*time.Second, 100*time.Millisecond, "both messages must have been processed successfully")

// Release the second Sync call so it can continue and wait for both of them to complete.
blockSyncCh <- struct{}{}
<-doneCh[0]
<-doneCh[1]

cancel()
}

func TestPackageUpdatesWithError(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
sender := NewHTTPSender(&sharedinternal.NopLogger{})

// We'll pass in a nil PackageStateProvider to force the Sync call to return with an error.
localPackageState := types.PackagesStateProvider(nil)
var messages atomic.Int32
var mux sync.Mutex
sender.callbacks = types.CallbacksStruct{
OnMessageFunc: func(ctx context.Context, msg *types.MessageData) {
// Make sure the call to Sync will return an error due to a nil PackageStateProvider
err := msg.PackageSyncer.Sync(ctx)
assert.Error(t, err)
messages.Add(1)
},
}

clientSyncedState := &ClientSyncedState{}

capabilities := protobufs.AgentCapabilities_AgentCapabilities_AcceptsPackages
sender.receiveProcessor = newReceivedProcessor(&sharedinternal.NopLogger{}, sender.callbacks, sender, clientSyncedState, localPackageState, capabilities, &mux)

// Send two messages in parallel.
sender.receiveProcessor.ProcessReceivedMessage(ctx,
&protobufs.ServerToAgent{
PackagesAvailable: &protobufs.PackagesAvailable{},
})
sender.receiveProcessor.ProcessReceivedMessage(ctx,
&protobufs.ServerToAgent{
PackagesAvailable: &protobufs.PackagesAvailable{},
})

// Make sure that even though the call to Sync errored out early, the lock
// was still released properly for both messages to be processed.
assert.Eventually(t, func() bool {
return messages.Load() == 2
}, 5*time.Second, 100*time.Millisecond, "both messages must have been processed successfully")

cancel()
}
5 changes: 5 additions & 0 deletions client/internal/inmempackagestore.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@ type InMemPackagesStore struct {
fileContents map[string][]byte
fileHashes map[string][]byte
lastReportedStatuses *protobufs.PackageStatuses

onAllPackagesHash func()
}

var _ types.PackagesStateProvider = (*InMemPackagesStore)(nil)
Expand All @@ -28,6 +30,9 @@ func NewInMemPackagesStore() *InMemPackagesStore {
}

func (l *InMemPackagesStore) AllPackagesHash() ([]byte, error) {
if l.onAllPackagesHash != nil {
l.onAllPackagesHash()
}
return l.allPackagesHash, nil
}

Expand Down
19 changes: 18 additions & 1 deletion client/internal/packagessyncer.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import (
"errors"
"fmt"
"net/http"
"sync"

"github.com/open-telemetry/opamp-go/client/types"
"github.com/open-telemetry/opamp-go/protobufs"
Expand All @@ -20,6 +21,7 @@ type packagesSyncer struct {
sender Sender

statuses *protobufs.PackageStatuses
mux *sync.Mutex
doneCh chan struct{}
}

Expand All @@ -30,6 +32,7 @@ func NewPackagesSyncer(
sender Sender,
clientSyncedState *ClientSyncedState,
packagesStateProvider types.PackagesStateProvider,
mux *sync.Mutex,
) *packagesSyncer {
return &packagesSyncer{
logger: logger,
Expand All @@ -38,6 +41,7 @@ func NewPackagesSyncer(
clientSyncedState: clientSyncedState,
localState: packagesStateProvider,
doneCh: make(chan struct{}),
mux: mux,
}
}

Expand All @@ -49,15 +53,24 @@ func (s *packagesSyncer) Sync(ctx context.Context) error {
}()

// Prepare package statuses.
// Grab a lock to make sure that package statuses are not overriden by
// another call to Sync running in parallel.
// In case Sync returns early with an error, take care of unlocking the
// mutex in this goroutine; otherwise it will be unlocked at the end
// of the sync operation.
s.mux.Lock()
if err := s.initStatuses(); err != nil {
s.mux.Unlock()
return err
}

if err := s.clientSyncedState.SetPackageStatuses(s.statuses); err != nil {
s.mux.Unlock()
return err
}

// Now do the actual syncing in the background.
// Now do the actual syncing in the background and release the lock from
// inside of the goroutine.
go s.doSync(ctx)

return nil
Expand Down Expand Up @@ -99,6 +112,10 @@ func (s *packagesSyncer) initStatuses() error {

// doSync performs the actual syncing process.
func (s *packagesSyncer) doSync(ctx context.Context) {
// Once doSync returns in a separate goroutine, make sure to release the
// mutex so that a new syncing process can take place.
defer s.mux.Unlock()

hash, err := s.localState.AllPackagesHash()
if err != nil {
s.logger.Errorf(ctx, "Package syncing failed: %V", err)
Expand Down
7 changes: 7 additions & 0 deletions client/internal/receivedprocessor.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package internal
import (
"context"
"fmt"
"sync"

"github.com/open-telemetry/opamp-go/client/types"
"github.com/open-telemetry/opamp-go/protobufs"
Expand All @@ -24,6 +25,9 @@ type receivedProcessor struct {

packagesStateProvider types.PackagesStateProvider

// packageSyncMutex protects against multiple package syncing operations at the same time.
packageSyncMutex *sync.Mutex

// Agent's capabilities defined at Start() time.
capabilities protobufs.AgentCapabilities
}
Expand All @@ -35,6 +39,7 @@ func newReceivedProcessor(
clientSyncedState *ClientSyncedState,
packagesStateProvider types.PackagesStateProvider,
capabilities protobufs.AgentCapabilities,
packageSyncMutex *sync.Mutex,
) receivedProcessor {
return receivedProcessor{
logger: logger,
Expand All @@ -43,6 +48,7 @@ func newReceivedProcessor(
clientSyncedState: clientSyncedState,
packagesStateProvider: packagesStateProvider,
capabilities: capabilities,
packageSyncMutex: packageSyncMutex,
}
}

Expand Down Expand Up @@ -122,6 +128,7 @@ func (r *receivedProcessor) ProcessReceivedMessage(ctx context.Context, msg *pro
r.sender,
r.clientSyncedState,
r.packagesStateProvider,
r.packageSyncMutex,
)
} else {
r.logger.Debugf(ctx, "Ignoring PackagesAvailable, agent does not have AcceptsPackages capability")
Expand Down
4 changes: 3 additions & 1 deletion client/internal/wsreceiver.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package internal
import (
"context"
"fmt"
"sync"

"github.com/gorilla/websocket"
"github.com/open-telemetry/opamp-go/client/types"
Expand Down Expand Up @@ -32,13 +33,14 @@ func NewWSReceiver(
clientSyncedState *ClientSyncedState,
packagesStateProvider types.PackagesStateProvider,
capabilities protobufs.AgentCapabilities,
packageSyncMutex *sync.Mutex,
) *wsReceiver {
w := &wsReceiver{
conn: conn,
logger: logger,
sender: sender,
callbacks: callbacks,
processor: newReceivedProcessor(logger, callbacks, sender, clientSyncedState, packagesStateProvider, capabilities),
processor: newReceivedProcessor(logger, callbacks, sender, clientSyncedState, packagesStateProvider, capabilities, packageSyncMutex),
stopped: make(chan struct{}),
}

Expand Down
Loading

0 comments on commit 536037b

Please sign in to comment.