Auto publish - #11
Closed
shibd wants to merge 9 commits into
Closed
Conversation
shibd
force-pushed
the
auto_publish
branch
2 times, most recently
from
December 12, 2022 09:12
6ded62e to
28dbe1a
Compare
Fixes apache#79 ### Motivation This is the C++ implementation for apache/pulsar#12402 ### Modifications * Add ChunkMessageIdImpl * Return ChunkMessageId when the Producer produces the chunk message or when the consumer consumes the chunk message. * In cosumer.seek, use the first chunk message-id of the chunk message-id. This will solve the problem caused by seeking chunk messages. This is also the impact of this PIP on the original business logic.
### Motivation There is typo for `auxv.h` header when checking `getauxval` symbol ### Modifications `auvx.h` -> `sys/auxv.h`
) ### Motivation Currently the released libraries don't hide the symbols from the dependencies, it can be verified by the following steps on Ubuntu: ```bash curl -O -L https://archive.apache.org/dist/pulsar/pulsar-client-cpp-3.1.0/deb-x86_64/apache-pulsar-client.deb apt install ./apache-pulsar-client.deb nm -D /usr/lib/libpulsar.so | grep curl ``` You will see lots of symbols from libcurl are included in `libpulsar.so`: ``` 0000000000709f50 T curl_easy_cleanup 000000000070a000 T curl_easy_duphandle ... ``` The root cause is that `-Wl,--exclude-libs,ALL` is added as the compile option, but it should work as a link option. ### Modifications Use `add_link_options` to add `-Wl,--exclude-libs=ALL` as the link option. It seems that `=ALL` should be correct but `,ALL` also works.
…#153) * [fix] Fix MessageId serialization when it's a batched message ### Motivation The serialization and deserialization of `MessageId` became wrong after apache#132. 1. The batch size is not serialized. 2. `BatchedMessageIdImpl` could never be deserialized. The wrong behaviors could lead to a result that all MessageId objects created from deserialization does not have a batch size, which might make `ReaderTest.testReaderOnSpecificMessageWithBatches` fail when the cmake build type is `Debug`. What's worse is that a MessageId created from deserialization is always treated as a `MessageIdImpl`, on which the acknowledgment will have wrong behavior. ### Modifications Serialize the batch size if it's valid. In deserialization, create a `BatchedMessageIdImpl` when the batch index and the batch size are valid as a batched message. There is a problem that if a `MessageId` is created from deserialization, it cannot share a `BatchMessageAcker` with other `MessageId` objects. In this case, create a fake `BatchMessageAcker` that returns false for both `ackIndividual` and `ackCumulative` methods. It will make acknowledgment always fail but will fall back to batch index ACK if batch index ACK is enabled. Add the `-DCMAKE_BUILD_TYPE=Debug` for tests to enable assertions. * Add virtual destructor
shibd
pushed a commit
that referenced
this pull request
Feb 18, 2024
apache#334) * Fix possible deadlock of Future when adding a listener after completed ### Motivation There is a case that deadlock could happen for a `Future`. Assume there is a `Promise` and its `Future`. 1. Call `Future::addListener` to add a listener that tries to acquire a user-provided mutex (`lock`). 2. Thread 1: Acquire `lock` first. 3. Thread 2: Call `Promise::setValue`, the listener will be triggered first before completed. Since `lock` is held by Thread 1, the listener will be blocked. 4. Thread 1: Call `Future::addListener`, since it detects the `InternalState::completed_` is true, it will call `get` to retrieve the result and value. Then, deadlock happens: - Thread 1 waits for `lock` is released, and then complete `InternalState::future_`. - Thread 2 holds `lock` but wait for `InternalState::future_` is completed. In a real world case, if we acquire a lock before `ProducerImpl::closeAsync`, then another thread call `setValue` in `ClientConnection::handleSuccess` and the callback of `createProducerAsync` tries to acquire the lock, `handleSuccess` will be blocked. Then in `closeAsync`, the current thread will be blocked in: ```c++ cnx->sendRequestWithId(Commands::newCloseProducer(producerId_, requestId), requestId) .addListener([self, callback](Result result, const ResponseData&) { callback(result); }); ``` The stacks: ``` Thread 1: #11 0x00007fab80da2173 in pulsar::InternalState<...>::complete (this=0x3d53e7a10, result=..., value=...) at lib/Futre.h:61 #13 pulsar::ClientConnection::handleSuccess (this=this@entry=0x2214bc000, success=...) at lib/ClientConnection.cc:1552 Thread 2: #8 get (result=..., this=0x3d53e7a10) at lib/Future.h:69 #9 pulsar::InternalState<...>::addListener (this=this@entry=0x3d53e7a10, listener=...) at lib/Future.h:51 #11 0x00007fab80e8dc4e in pulsar::ProducerImpl::closeAsync at lib/ProducerImpl.cc:794 ``` There are two points that make the deadlock: 1. We use `completed_` to represent if the future is completed. However, after it's true, the future might not be completed because the value is not set and the listeners are not completed. 2. If `addListener` is called after it's completed, we still push the listener to `listeners_` so that previous listeners could be executed before the new listener. This guarantee is unnecessarily strong. ### Modifications First, complete the future before calling the listeners. Then, use an enum to represent the status: - INITIAL: `complete` has not been called - COMPLETING: when the 1st time `complete` is called, the status will change from INITIAL to COMPLETING - COMPLETED: the future is completed. Besides, implementation of `Future` is simplified. apache#299 fixes a possible mutex crash by introducing the `std::future`. However, the root cause is the conditional variable is not used correctly: > Even if the shared variable is atomic, it must be modified while owning the mutex to correctly publish the modification to the waiting thread. See https://en.cppreference.com/w/cpp/thread/condition_variable The simplest way to fix apache#298 is just adding `lock.lock()` before `state->condition.notify_all();`. * Acquire lock again * Add initial value
This file contains hidden or 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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
No description provided.