Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
3 changes: 1 addition & 2 deletions src/groups/mqb/mqbblp/mqbblp_clusterqueuehelper.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -2433,8 +2433,7 @@ void ClusterQueueHelper::onHandleReleasedDispatched(

// Delete the stream if fully released.

if (result.hasNoHandleClients() || (result.hasNoHandleStreamConsumers() &&
result.hasNoHandleStreamProducers())) {
if (result.hasNoHandleClients() || result.isHandleStreamEmpty()) {
// Expect `handle` to be non-empty with these values in `result`
BSLS_ASSERT_SAFE(handle);

Expand Down
68 changes: 24 additions & 44 deletions src/groups/mqb/mqbblp/mqbblp_queueengineutil.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -566,9 +566,7 @@ int QueueEngineUtil_ReleaseHandleProctor::releaseHandle(
mqbi::QueueHandle* handle,
const bmqp_ctrlmsg::QueueHandleParameters& params)
{
bsls::Types::Uint64 lostFlags = 0;
int rc = d_queueState_p->handleCatalog().releaseHandleHelper(&d_handleSp,
&lostFlags,
handle,
params,
d_isFinal);
Expand All @@ -587,40 +585,14 @@ int QueueEngineUtil_ReleaseHandleProctor::releaseHandle(
d_result.makeNoHandleClients();
}

if (bmqt::QueueFlagsUtil::isReader(lostFlags)) {
// no more consumers for any subStreams for this handle
d_result.makeNoHandleStreamConsumers();
}

if (bmqt::QueueFlagsUtil::isWriter(lostFlags)) {
// no more producers for any subStreams for this handle
d_result.makeNoHandleStreamProducers();
if (0 == handle->handleParameters().writeCount()) {
// no more producers for this handle
d_result.makeNoHandleProducers();
}

return 0;
}

mqbi::QueueHandleReleaseResult
QueueEngineUtil_ReleaseHandleProctor::releaseQueueStream(
const bmqp_ctrlmsg::QueueHandleParameters& params)
{
BSLS_ASSERT_SAFE(d_handleSp);

mqbi::QueueHandleReleaseResult result;
// Update queue's aggregated parameters.
mqbi::QueueCounts cumulative = d_queueState_p->subtract(params);

if (cumulative.d_readCount == 0) {
// no more consumers for this subStream across all handles
result.makeNoQueueStreamConsumers();
}
if (cumulative.d_writeCount == 0) {
// no more producers for this subStream across all handles
result.makeNoQueueStreamProducers();
}
return result;
}

mqbi::QueueHandleReleaseResult
QueueEngineUtil_ReleaseHandleProctor::releaseStream(
const bmqp_ctrlmsg::QueueHandleParameters& params)
Expand All @@ -632,17 +604,8 @@ QueueEngineUtil_ReleaseHandleProctor::releaseStream(
params.readCount(),
params.writeCount());

// `d_result` is accumulated result in the case of wildcard.
// `result` is about the given stream only.
mqbi::QueueHandleReleaseResult result;

if (0 == diff.d_readCount) {
result.makeNoHandleStreamConsumers();
}
if (0 == diff.d_writeCount) {
result.makeNoHandleStreamProducers();
}
result.apply(releaseQueueStream(params));
// Update queue's aggregated parameters.
mqbi::QueueCounts cumulative = d_queueState_p->subtract(params);

if (d_isFinal && (diff.d_readCount || diff.d_writeCount)) {
// mismatch between isFinal and actual counts in the closeQueue
Expand All @@ -657,10 +620,27 @@ QueueEngineUtil_ReleaseHandleProctor::releaseStream(
copy.readCount() = diff.d_readCount;
copy.writeCount() = diff.d_writeCount;

result.apply(releaseQueueStream(copy));
cumulative = d_queueState_p->subtract(copy);
diff = mqbi::QueueCounts(0, 0);
}

// `d_result` is accumulated result in the case of wildcard.
// `result` is about the given stream only.
mqbi::QueueHandleReleaseResult result;

if (0 == diff.d_readCount) {
// no more consumers for this subStream for this handle
result.makeNoHandleStreamConsumers();
result.makeNoHandleStreamProducers();
if (0 == diff.d_writeCount) {
result.makeHandleStreamEmpty();
}
}
if (0 == cumulative.d_readCount) {
// no more consumers for this subStream across all handles
result.makeNoQueueStreamConsumers();
if (0 == cumulative.d_writeCount) {
result.makeQueueStreamEmpty();
}
}
d_result.apply(result);

Expand Down
13 changes: 4 additions & 9 deletions src/groups/mqb/mqbblp/mqbblp_queueengineutil.h
Original file line number Diff line number Diff line change
Expand Up @@ -159,9 +159,11 @@ struct QueueEngineUtil_ReleaseHandleProctor {
// Discern the following (partially orthogonal) cases:
//
// - no more consumers or producers for all subStream for this handle
// - no more consumers for this subStream across all handles
// - no more producers for this subStream across all handles
// - no more producers for this handle
// - no more consumers for this subStream for this handle
// - no more clients for this subStream for this handle
// - no more consumers for this subStream across all handles
// - no more clients for this subStream across all handles

// DATA

Expand Down Expand Up @@ -203,13 +205,6 @@ struct QueueEngineUtil_ReleaseHandleProctor {
int releaseHandle(mqbi::QueueHandle* handle,
const bmqp_ctrlmsg::QueueHandleParameters& params);

/// Decrement cumulative queue's readCount and writeCount of the stream
/// identified by the specified `params` by readCount and writeCount in
/// the `params`. Return result indicating producer/consumer absence
/// for the given queue.
mqbi::QueueHandleReleaseResult
releaseQueueStream(const bmqp_ctrlmsg::QueueHandleParameters& params);

/// Decrement handle's readCount and writeCount of the stream identified
/// by the specified `params` by readCount and writeCount in the
/// `params`. Decrement cumulative queue's readCount and writeCount of
Expand Down
12 changes: 2 additions & 10 deletions src/groups/mqb/mqbblp/mqbblp_queuehandlecatalog.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -197,7 +197,6 @@ mqbi::QueueHandle* QueueHandleCatalog::createHandle(

int QueueHandleCatalog::releaseHandleHelper(
bsl::shared_ptr<mqbi::QueueHandle>* handleSp,
bsls::Types::Uint64* lostFlags,
mqbi::QueueHandle* handle,
const bmqp_ctrlmsg::QueueHandleParameters& handleParameters,
bool isFinal)
Expand All @@ -219,7 +218,6 @@ int QueueHandleCatalog::releaseHandleHelper(

bmqp_ctrlmsg::QueueHandleParameters currentHandleParameters =
handle->handleParameters();
bsls::Types::Uint64 oldFlags = currentHandleParameters.flags();

// Update queue's aggregated parameters
bmqp::QueueUtil::subtractHandleParameters(&currentHandleParameters,
Expand All @@ -228,10 +226,7 @@ int QueueHandleCatalog::releaseHandleHelper(

handle->setHandleParameters(currentHandleParameters);

*handleSp = handleIt->value();
*lostFlags = bmqt::QueueFlagsUtil::removals(
oldFlags,
currentHandleParameters.flags());
*handleSp = handleIt->value();

const bool allCountsZero = bmqp::QueueUtil::isEmpty(
currentHandleParameters);
Expand All @@ -248,7 +243,6 @@ int QueueHandleCatalog::releaseHandleHelper(
<< ", lastReleased handle parameters: "
<< handleParameters << "]";
d_handles.erase(handleIt);
*lostFlags = oldFlags; // Handle is deleted so all flags are lost.
}

BALL_LOG_INFO_BLOCK
Expand All @@ -260,9 +254,7 @@ int QueueHandleCatalog::releaseHandleHelper(
<< ", removed handle parameters: " << handleParameters
<< ", isFinal: " << isFinal
<< ", remaining handle parameters: " << currentHandleParameters
<< "], lostFlags: '";
bmqt::QueueFlagsUtil::prettyPrint(BALL_LOG_OUTPUT_STREAM, *lostFlags);
BALL_LOG_OUTPUT_STREAM << "'";
<< "]";
}

return 0;
Expand Down
7 changes: 2 additions & 5 deletions src/groups/mqb/mqbblp/mqbblp_queuehandlecatalog.h
Original file line number Diff line number Diff line change
Expand Up @@ -195,13 +195,10 @@ class QueueHandleCatalog {
/// longer representing any resource, it is removed from the internal
/// map, the specified `deleted` is set to true and the specified
/// `handleSp` is populated with it allowing the caller to control at
/// which time destruction will occur. The specified `lostFlags` is set
/// to the flags that were set on the queue prior to the release, but
/// are no longer after. Return 0 on success, or a non-zero value on
/// error.
/// which time destruction will occur. Return 0 on success, or a
/// non-zero value on error.
int releaseHandleHelper(
bsl::shared_ptr<mqbi::QueueHandle>* handleSp,
bsls::Types::Uint64* lostFlags,
mqbi::QueueHandle* handle,
const bmqp_ctrlmsg::QueueHandleParameters& handleParameters,
bool isFinal);
Expand Down
3 changes: 1 addition & 2 deletions src/groups/mqb/mqbblp/mqbblp_queuesessionmanager.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -488,8 +488,7 @@ void QueueSessionManager::onHandleReleasedDispatched(
handle->queue(),
mqbi::DispatcherEventType::e_DISPATCHER);
}
else if (result.hasNoQueueStreamConsumers() &&
result.hasNoQueueStreamProducers()) {
else if (result.isHandleStreamEmpty()) {
QueueState::StreamsMap::const_iterator sqiIter =
it->second.d_subQueueInfosMap.findBySubId(queueId.subId());
BSLS_ASSERT_SAFE(sqiIter != it->second.d_subQueueInfosMap.end());
Expand Down
4 changes: 2 additions & 2 deletions src/groups/mqb/mqbblp/mqbblp_relayqueueengine.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -762,8 +762,8 @@ void RelayQueueEngine::onHandleReleasedDispatched(
d_pushStream.removeApp(upstreamSubQueueId);
app->clear();
}
if (streamResult.hasNoHandleStreamProducers()) {
// Remove the producer handle from cached producers/consumers
if (streamResult.isHandleStreamEmpty()) {
// Remove the handle from this App's cached parameters
app->d_cache.erase(handle);
}
}
Expand Down
2 changes: 1 addition & 1 deletion src/groups/mqb/mqbblp/mqbblp_remotequeue.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -759,7 +759,7 @@ void RemoteQueue::onHandleReleased(
}
}
}
if (result.hasNoHandleStreamProducers()) {
if (result.hasNoHandleProducers()) {
// Leaving it to the downstream to NACK pending messages.
cleanPendingMessages(handle.get());
}
Expand Down
7 changes: 3 additions & 4 deletions src/groups/mqb/mqbi/mqbi_queue.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -29,12 +29,11 @@ bsl::ostream& operator<<(bsl::ostream& stream,
stream << "["
<< (rhs.hasNoHandleStreamConsumers() ? " hasNoHandleStreamConsumers"
: "")
<< (rhs.hasNoHandleStreamProducers() ? " hasNoHandleStreamProducers"
: "")
<< (rhs.hasNoHandleProducers() ? " hasNoHandleProducers" : "")
<< (rhs.hasNoQueueStreamConsumers() ? " hasNoQueueStreamConsumers"
: "")
<< (rhs.hasNoQueueStreamProducers() ? " hasNoQueueStreamProducers"
: "")
<< (rhs.isHandleStreamEmpty() ? " emptyHandleStream" : "")
<< (rhs.isQueueStreamEmpty() ? " emptyQueueStream" : "")
<< (rhs.hasNoHandleClients() ? " hasNoHandleClients" : "") << " ]";
return stream;
}
Expand Down
68 changes: 38 additions & 30 deletions src/groups/mqb/mqbi/mqbi_queue.h
Original file line number Diff line number Diff line change
Expand Up @@ -370,14 +370,17 @@ struct QueueHandleReleaseResult {
/// no more consumers for this subStream for this handle
e_NO_HANDLE_STREAM_CONSUMERS = (1 << 1),

/// no more producers for this subStream for this handle
e_NO_HANDLE_STREAM_PRODUCERS = (1 << 2),
/// no more producers for this handle
e_NO_HANDLE_PRODUCERS = (1 << 2),

/// no more consumers for this subStream across all handles
e_NO_QUEUE_STREAM_CONSUMERS = (1 << 3),

/// no more producers for this subStream across all handles
e_NO_QUEUE_STREAM_PRODUCERS = (1 << 4)
/// no more clients for this subStream for this handle
e_HANDLE_STREAM_EMPTY = (1 << 4),

/// no more clients for this subStream across all handles
e_QUEUE_STREAM_EMPTY = (1 << 5)
};

// DATA
Expand All @@ -392,17 +395,18 @@ struct QueueHandleReleaseResult {
/// this queue.
void makeNoHandleStreamConsumers();

/// Indicate absence of producers for this subStream for this handle in
/// this queue.
void makeNoHandleStreamProducers();
/// Indicate absence of producers for this handle in this queue.
void makeNoHandleProducers();

/// Indicate absence of consumers for this subStream for all handles in
/// this queue.
void makeNoQueueStreamConsumers();

/// Indicate absence of producers for this subStream for all handles in
/// this queue.
void makeNoQueueStreamProducers();
/// Indicate that this subStream for this handle is empty.
void makeHandleStreamEmpty();

/// Indicate that this subStream for all handles is empty.
void makeQueueStreamEmpty();

/// Indicate absence of consumers and producers for all subStreams in
/// in this handle in this queue.
Expand All @@ -417,20 +421,19 @@ struct QueueHandleReleaseResult {
/// handle in this queue.
bool hasNoHandleStreamConsumers() const;

/// Return true if there are no producers for this subStream for this
/// handle in this queue.
bool hasNoHandleStreamProducers() const;
/// Return true if there are no producers for this handle in this queue.
bool hasNoHandleProducers() const;

/// Return true if there are no consumers for this subStream for all
/// handles in this queue.
bool hasNoQueueStreamConsumers() const;

/// Return true if there are no producers for this subStream for all
/// handles in this queue.
bool hasNoQueueStreamProducers() const;
/// Return true if there are no clients for this subStream for this
/// handle in this queue.
bool isHandleStreamEmpty() const;

/// Return true if there are no consumers and producers for this
/// subStream for all handles in this queue.
/// Return true if there are no clients for this subStream for all
/// handles in this queue.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I found it confusing that a handle knows something about all other handles. Can this information become false by manipulating other handles? Can we avoid storing it at all on this level?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

This does come not from a handle but from a QueueEngine. As a result of (handle initiated) action.
The result is not cached, it is a return value of the operation.

bool isQueueStreamEmpty() const;

/// Return true if there are no consumers and producers for all
Expand Down Expand Up @@ -1194,19 +1197,24 @@ inline void QueueHandleReleaseResult::makeNoHandleStreamConsumers()
d_result |= e_NO_HANDLE_STREAM_CONSUMERS;
}

inline void QueueHandleReleaseResult::makeNoHandleStreamProducers()
inline void QueueHandleReleaseResult::makeNoHandleProducers()
{
d_result |= e_NO_HANDLE_STREAM_PRODUCERS;
d_result |= e_NO_HANDLE_PRODUCERS;
}

inline void QueueHandleReleaseResult::makeNoQueueStreamConsumers()
{
d_result |= e_NO_QUEUE_STREAM_CONSUMERS;
}

inline void QueueHandleReleaseResult::makeNoQueueStreamProducers()
inline void QueueHandleReleaseResult::makeHandleStreamEmpty()
{
d_result |= e_NO_QUEUE_STREAM_PRODUCERS;
d_result |= e_HANDLE_STREAM_EMPTY;
}

inline void QueueHandleReleaseResult::makeQueueStreamEmpty()
{
d_result |= e_QUEUE_STREAM_EMPTY;
}

inline void QueueHandleReleaseResult::makeNoHandleClients()
Expand All @@ -1230,24 +1238,24 @@ inline bool QueueHandleReleaseResult::hasNoHandleStreamConsumers() const
return (d_result & e_NO_HANDLE_STREAM_CONSUMERS);
}

inline bool QueueHandleReleaseResult::hasNoHandleStreamProducers() const
inline bool QueueHandleReleaseResult::hasNoHandleProducers() const
{
return (d_result & e_NO_HANDLE_STREAM_PRODUCERS);
return (d_result & e_NO_HANDLE_PRODUCERS);
}

inline bool QueueHandleReleaseResult::isQueueStreamEmpty() const
inline bool QueueHandleReleaseResult::isHandleStreamEmpty() const
{
return hasNoQueueStreamConsumers() && hasNoQueueStreamProducers();
return (d_result & e_HANDLE_STREAM_EMPTY);
}

inline bool QueueHandleReleaseResult::hasNoQueueStreamConsumers() const
inline bool QueueHandleReleaseResult::isQueueStreamEmpty() const
{
return (d_result & e_NO_QUEUE_STREAM_CONSUMERS);
return (d_result & e_QUEUE_STREAM_EMPTY);
}

inline bool QueueHandleReleaseResult::hasNoQueueStreamProducers() const
inline bool QueueHandleReleaseResult::hasNoQueueStreamConsumers() const
{
return (d_result & e_NO_QUEUE_STREAM_PRODUCERS);
return (d_result & e_NO_QUEUE_STREAM_CONSUMERS);
}

// ---------------------------------
Expand Down
Loading