-
Notifications
You must be signed in to change notification settings - Fork 3.5k
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.
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
Make readModifyUpdate in MetadataCacheImpl thread-safe #9900
Conversation
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can we fall into some kind of deadlock?
It will become like when we are doing sync calls to Zookeeper, we are going to depend implicitly to an external (not under our control) execution serializer
I don't quite understand what you mean, could you please help explain? |
This might not be a sufficient solution for ensuring thread safety. Considering read operations is equally important. How does the solution ensure that read operations are thread safe? |
I have added a snapshot, I don’t know if there is any omission Visibility: A snapshot is generated for each modification. After the modification is completed, the entire snapshot is written to the cache, and the value of caffeine is immediately visible to all threads |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
My concern is that we are falling back in using a synchronous pattern to access the metadata service, and this may lead to deadlocks
@@ -166,7 +166,7 @@ private MetadataCacheImpl(MetadataStore store, MetadataSerde<T> serde) { | |||
T newValueObj; | |||
byte[] newValue; | |||
try { | |||
newValueObj = modifyFunction.apply(currentValue); | |||
newValueObj = modifyFunction.apply(serde.deserialize(serde.serialize(currentValue))); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can we add a comment that explains that we are cloning the value?
Adding an explicit static method cloneValue will help in understanding better what's going on
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
A comment has been added.
Now I just let the write operations of the same namespace be executed by the same thread, and will not lock, so there will be no deadlock problem
If the problem is about modifying concurrently the Policies structure, pulsar/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/MetadataCacheImpl.java Line 155 in cdef644
MetadataCacheImpl already handles versioning in respect to the MetadataService, but the problem is that we are modifying the stored value returned from objCache.get(). |
MetadataCacheImpl is just one of the implementations of the Metadatacache, and we cannot control the subsequent implementations. The cache implementation should not care about these things. So I moved clone to namespaceBase. |
IMHO we should fix MetadataCacheImpl, the signature and the semantics of the method is to be documented. |
You will have this problem with any local cache, such as GuavaCache, Caffeine. We cannot require these caches to guarantee. |
@315157973 IMHO there is no need to change the name to readClonedModifyUpdate because it is expected that entries in the cache are to be managed as immutable. |
@315157973 Good work. Please also update the description of this PR to cover the considerations about thread safety in accessing objects. |
@eolivelli @315157973 I agree that the wider problem should be addressed. It seems that all classes that extend org.apache.pulsar.broker.resources.BaseResources are impacted. |
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BaseResources.java
Outdated
Show resolved
Hide resolved
@315157973 pulsar/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/MetadataCacheImpl.java Line 163 in 5f77745
then we write to ZK pulsar/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/MetadataCacheImpl.java Line 175 in 5f77745
so there is no need to add a new version variable to Policies IMHO |
Yes, you are right, now it is optimistic lock |
We can only modify the set of |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I believe we are on our way.
I left one last comment.
thanks
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
Outdated
Show resolved
Hide resolved
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Looks great!
Probably we should change the title and the description because now the fix is no more strictly related only to Policies
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/BaseResources.java
Outdated
Show resolved
Hide resolved
@@ -166,6 +165,8 @@ private MetadataCacheImpl(MetadataStore store, MetadataSerde<T> serde) { | |||
T newValueObj; | |||
byte[] newValue; | |||
try { | |||
// Use clone and CAS zk to ensure thread safety |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We should do the same operation on the readModifyUpdateOrCreate
method as well
This error in a test case looks related to this patch:
|
/pulsarbot run-failure-checks |
### Motivation Now that the modification of Namespace Policies is not locked, multiple threads may modify the same local Policies object at the same time, which may cause thread safety issues. The Policies object also contains non-thread-safe collections such as HashMap and HashSet. Concurrent operations on these objects also have thread-safety issues. E.g apache#9711 ### Modifications Use clone+CAS method to ensure thread safety Visibility: The cache will be cloned, only the cloned object will be modified, and the entire object will be written back eventually. The cached object in caffeine will be immediately visible to other threads Atomicity: Modification refers to the replacement of the entire cloned object. There will be no intermediate state where some attribute values are modified, so it is atomic Orderliness: Guaranteed by the implementation class of MetadataCache. Now the cached object corresponds to a version number. Use CAS to write back. If the writeback fails, it will read -> clone -> modify -> update again to ensure order.
This is a fix based on the new feature which should be released in 2.8.0. So I will remove the release/2.7.2 label. |
There's a severe thread safety issue in 2.7.1, #9711, which this PR fixes. The issue is broader than authorization configuration and this PR addresses the fix for policy updates in a general way. |
We cannot pick this patch as it is. |
Motivation
Now that the modification of Namespace Policies is not locked, multiple threads may modify the same local Policies object at the same time, which may cause thread safety issues.
The Policies object also contains non-thread-safe collections such as HashMap and HashSet. Concurrent operations on these objects also have thread-safety issues. E.g #9711
Modifications
Use clone+CAS method to ensure thread safety
Visibility: The cache will be cloned, only the cloned object will be modified, and the entire object will be written back eventually. The cached object in caffeine will be immediately visible to other threads
Atomicity: Modification refers to the replacement of the entire cloned object. There will be no intermediate state where some attribute values are modified, so it is atomic
Orderliness: Guaranteed by the implementation class of MetadataCache. Now the cached object corresponds to a version number. Use CAS to write back. If the writeback fails, it will read -> clone -> modify -> update again to ensure order.
Verifying this change