Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Package Syncing #118

Closed
wants to merge 3 commits into from
Closed

Package Syncing #118

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0ccb1c1
Adding Mutex
kpratyus Jul 29, 2022
25b247c
Adding Mutex
kpratyus Jul 29, 2022
6a15fe0
Merge pull request #1 from kpratyus/sync_package
kpratyus Jul 29, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions client/internal/clientstate.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ var (
// It is safe to call methods of this struct concurrently.
type ClientSyncedState struct {
mutex sync.Mutex
packageSyncMutex sync.Mutex
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think this is the right place for this mutex. The ClientSyncedState struct has a specific purpose that is described above and adding a mutex for packager syncer does not fit that purpose.

receivedProcessor may be a better place for this mutex.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The receivedProcessor is constructed in the constructor of wsReceiver. As the mutex has to be unique for each client and the pointer has to be passed to `receivedProcessor'. So mutex has to be included in the constructor of client.common, wsReceiver and receivedProcessor. To avoid these many changes I added the mutex in ClientSyncedState. What's your opinion on this?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is one-to-one relationship between Client, ClientSyncedState and receivedProcessor, so I don't understand where do you see a problem. A mutex declared in receivedProcessor struct will be unique for each client, just like it is unique when it is in ClientSyncedState. There is no need to pass the mutex to receivedProcessor, receivedProcessor can contain the mutex.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Correct me if I am wrong here. For every client The function runOneCycle runs in a loop. Everytime the function runs it creates a new receiver object.
r := internal.NewWSReceiver( c.common.Logger, c.common.Callbacks, c.conn, c.sender, &c.common.ClientSyncedState, c.common.PackagesStateProvider, c.common.Capabilities, )
This in turn creates a new receivedProcessor' object everytime. processor: newReceivedProcessor(logger, callbacks, sender, clientSyncedState, packagesStateProvider, capabilities)`
So how is there a one to one relationship between client and receivedProcessor?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You are absolutely right and I was wrong! Due to reconnections receivedProcessor may be re-created while a previous sync is still happening. So, receivedProcessor is the wrong owner for the mutex, you are right.

I still don't like putting it in ClientSyncedState, it doesn't fit the purpose.

Here are a couple options:

  • Extend the purpose of ClientSyncedState to contain all of the client state. In that case the mutex can be a good fit there. It probably needs to be renamed to ClientState. I am not entirely sure this is a good idea though, since we are not really putting the entire client state in it, so it may become misleading.
  • Another good owner for the mutex is ClientCommon. However, this will require passing the mutex where receivedProcessor is created and store the pointer to the mutex in receivedProcessor. Probably a bit awkward but doable.

Maybe there is another option that I haven't thought of, please feel free to suggest.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

While raising the PR I also thought of these 2 options. Code clarity wise adding all the mutexes in a single structure(ClientSyncedState here) makes sense but I totally get the point that we want to keep ClientSyncedState as lightweight as possible.

Although awkward I think keeping it in ClientCommon is a better option. In future if there arises need of any other locks then it would be totally extendable as well.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Although awkward I think keeping it in ClientCommon is a better option. In future if there arises need of any other locks then it would be totally extendable as well.

Sounds good, let's go with this approach.


agentDescription *protobufs.AgentDescription
health *protobufs.AgentHealth
Expand Down
8 changes: 6 additions & 2 deletions client/internal/packagessyncer.go
View file
Open in desktop
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 Down Expand Up @@ -58,7 +59,7 @@ func (s *packagesSyncer) Sync(ctx context.Context) error {
}

// Now do the actual syncing in the background.
go s.doSync(ctx)
go s.doSync(ctx, &s.clientSyncedState.packageSyncMutex)

return nil
}
Expand Down Expand Up @@ -98,7 +99,7 @@ func (s *packagesSyncer) initStatuses() error {
}

// doSync performs the actual syncing process.
func (s *packagesSyncer) doSync(ctx context.Context) {
func (s *packagesSyncer) doSync(ctx context.Context, mutex *sync.Mutex) {
hash, err := s.localState.AllPackagesHash()
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why isn't the mutex locked right from the top of doSync? AllPackagesHash is also susceptible to the race. It reads data that is written by SetAllPackagesHash below.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, Missed that. I actually thought that it only handles logging. Will change the mutex. Thanks for pointing it out

if err != nil {
s.logger.Errorf("Package syncing failed: %V", err)
Expand All @@ -109,6 +110,9 @@ func (s *packagesSyncer) doSync(ctx context.Context) {
return
}

(*mutex).Lock()
defer (*mutex).Unlock()

failed := false
if err := s.deleteUnneededLocalPackages(); err != nil {
s.logger.Errorf("Cannot delete unneeded packages: %v", err)
Expand Down