Commit bcbc4ff5 authored by Ameya Limaye's avatar Ameya Limaye Committed by Facebook Github Bot

Throw custom exception type when ABD fails a result because some other token did not call dispatch

Summary:
Throw custom exception type when ABD fails a result because some other token did not call dispatch:
- This is useful because the caller can catch this exception and discard it if they find some other exception
- The custom exception does not give an indication of the problem. The other exception/s thrown will do that.

Reviewed By: yfeldblum

Differential Revision: D4395742

fbshipit-source-id: be80f66b1297e9faf625a2fb087590a7d0a9335d
parent 0e4a41d0
...@@ -550,6 +550,7 @@ nobase_follyinclude_HEADERS += \ ...@@ -550,6 +550,7 @@ nobase_follyinclude_HEADERS += \
fibers/Baton-inl.h \ fibers/Baton-inl.h \
fibers/BatchDispatcher.h \ fibers/BatchDispatcher.h \
fibers/BoostContextCompatibility.h \ fibers/BoostContextCompatibility.h \
fibers/detail/AtomicBatchDispatcher.h \
fibers/EventBaseLoopController.h \ fibers/EventBaseLoopController.h \
fibers/EventBaseLoopController-inl.h \ fibers/EventBaseLoopController-inl.h \
fibers/Fiber.h \ fibers/Fiber.h \
...@@ -577,6 +578,7 @@ nobase_follyinclude_HEADERS += \ ...@@ -577,6 +578,7 @@ nobase_follyinclude_HEADERS += \
libfolly_la_SOURCES += \ libfolly_la_SOURCES += \
fibers/Baton.cpp \ fibers/Baton.cpp \
fibers/detail/AtomicBatchDispatcher.cpp \
fibers/Fiber.cpp \ fibers/Fiber.cpp \
fibers/FiberManager.cpp \ fibers/FiberManager.cpp \
fibers/FiberManagerMap.cpp \ fibers/FiberManagerMap.cpp \
......
...@@ -29,12 +29,14 @@ struct AtomicBatchDispatcher<InputT, ResultT>::DispatchBaton { ...@@ -29,12 +29,14 @@ struct AtomicBatchDispatcher<InputT, ResultT>::DispatchBaton {
optEntries_.reserve(numEntries); optEntries_.reserve(numEntries);
} }
void setError(std::string message) { void setExceptionWrapper(folly::exception_wrapper&& exWrapper) {
optErrorMessage_ = std::move(message); exceptionWrapper_ = std::move(exWrapper);
} }
void setExpectedCount(size_t expectedCount) { void setExpectedCount(size_t expectedCount) {
assert(expectedCount_ == 0 || !"expectedCount_ being set more than once");
expectedCount_ = expectedCount; expectedCount_ = expectedCount;
optEntries_.resize(expectedCount_);
} }
Future<ResultT> getFutureResult(InputT&& input, size_t sequenceNumber) { Future<ResultT> getFutureResult(InputT&& input, size_t sequenceNumber) {
...@@ -42,17 +44,13 @@ struct AtomicBatchDispatcher<InputT, ResultT>::DispatchBaton { ...@@ -42,17 +44,13 @@ struct AtomicBatchDispatcher<InputT, ResultT>::DispatchBaton {
optEntries_.resize(sequenceNumber + 1); optEntries_.resize(sequenceNumber + 1);
} }
folly::Optional<Entry>& optEntry = optEntries_[sequenceNumber]; folly::Optional<Entry>& optEntry = optEntries_[sequenceNumber];
if (optEntry) { assert(!optEntry || !"Multiple inputs have the same token sequence number");
throw std::logic_error(
"Cannot have multiple inputs with same token sequence number");
}
optEntry = Entry(std::move(input)); optEntry = Entry(std::move(input));
return optEntry->promise.getFuture(); return optEntry->promise.getFuture();
} }
private: private:
void setExceptionResults(std::exception_ptr eptr) { void setExceptionResults(const folly::exception_wrapper& exceptionWrapper) {
auto exceptionWrapper = exception_wrapper(eptr);
for (auto& optEntry : optEntries_) { for (auto& optEntry : optEntries_) {
if (optEntry) { if (optEntry) {
optEntry->promise.setException(exceptionWrapper); optEntry->promise.setException(exceptionWrapper);
...@@ -60,53 +58,56 @@ struct AtomicBatchDispatcher<InputT, ResultT>::DispatchBaton { ...@@ -60,53 +58,56 @@ struct AtomicBatchDispatcher<InputT, ResultT>::DispatchBaton {
} }
} }
void setExceptionResults(std::exception_ptr eptr) {
auto exceptionWrapper = exception_wrapper(eptr);
return setExceptionResults(exceptionWrapper);
}
template <typename TException> template <typename TException>
void setExceptionResults( void setExceptionResults(
const TException& ex, const TException& ex,
std::exception_ptr eptr = std::exception_ptr()) { std::exception_ptr eptr = std::exception_ptr()) {
auto exceptionWrapper = auto exceptionWrapper =
eptr ? exception_wrapper(eptr, ex) : exception_wrapper(ex); eptr ? exception_wrapper(eptr, ex) : exception_wrapper(ex);
for (auto& optEntry : optEntries_) { return setExceptionResults(exceptionWrapper);
if (optEntry) {
optEntry->promise.setException(exceptionWrapper);
}
}
} }
void fulfillPromises() { void fulfillPromises() {
try { try {
// If an error message is set, set all promises to exception with message // If an error message is set, set all promises to exception with message
if (optErrorMessage_) { if (exceptionWrapper_) {
auto ex = std::logic_error(*optErrorMessage_); return setExceptionResults(exceptionWrapper_);
return setExceptionResults(std::move(ex)); }
// Validate entries count same as expectedCount_
assert(
optEntries_.size() == expectedCount_ ||
!"Entries vector did not have expected size");
std::vector<size_t> vecTokensNotDispatched;
for (size_t i = 0; i < expectedCount_; ++i) {
if (!optEntries_[i]) {
vecTokensNotDispatched.push_back(i);
}
}
if (!vecTokensNotDispatched.empty()) {
return setExceptionResults(ABDTokenNotDispatchedException(
detail::createABDTokenNotDispatchedExMsg(vecTokensNotDispatched)));
} }
// Create inputs vector and validate entries count same as expectedCount_ // Create the inputs vector
std::vector<InputT> inputs; std::vector<InputT> inputs;
inputs.reserve(expectedCount_); inputs.reserve(expectedCount_);
bool allEntriesFound = (optEntries_.size() == expectedCount_);
if (allEntriesFound) {
for (auto& optEntry : optEntries_) { for (auto& optEntry : optEntries_) {
if (!optEntry) {
allEntriesFound = false;
break;
}
inputs.emplace_back(std::move(optEntry->input)); inputs.emplace_back(std::move(optEntry->input));
} }
}
if (!allEntriesFound) {
auto ex = std::logic_error(
"One or more input tokens destroyed before calling dispatch");
return setExceptionResults(std::move(ex));
}
// Call the user provided batch dispatch function to get all results // Call the user provided batch dispatch function to get all results
// and make sure that we have the expected number of results returned // and make sure that we have the expected number of results returned
auto results = dispatchFunction_(std::move(inputs)); auto results = dispatchFunction_(std::move(inputs));
if (results.size() != expectedCount_) { if (results.size() != expectedCount_) {
auto ex = std::logic_error( return setExceptionResults(
"Unexpected number of results returned from dispatch function"); ABDUsageException(detail::createUnexpectedNumResultsABDUsageExMsg(
return setExceptionResults(std::move(ex)); expectedCount_, results.size())));
} }
// Fulfill the promises with the results from the batch dispatch // Fulfill the promises with the results from the batch dispatch
...@@ -114,8 +115,10 @@ struct AtomicBatchDispatcher<InputT, ResultT>::DispatchBaton { ...@@ -114,8 +115,10 @@ struct AtomicBatchDispatcher<InputT, ResultT>::DispatchBaton {
optEntries_[i]->promise.setValue(std::move(results[i])); optEntries_[i]->promise.setValue(std::move(results[i]));
} }
} catch (const std::exception& ex) { } catch (const std::exception& ex) {
// Set exceptions thrown when executing the user provided dispatch func
return setExceptionResults(ex, std::current_exception()); return setExceptionResults(ex, std::current_exception());
} catch (...) { } catch (...) {
// Set exceptions thrown when executing the user provided dispatch func
return setExceptionResults(std::current_exception()); return setExceptionResults(std::current_exception());
} }
} }
...@@ -139,7 +142,7 @@ struct AtomicBatchDispatcher<InputT, ResultT>::DispatchBaton { ...@@ -139,7 +142,7 @@ struct AtomicBatchDispatcher<InputT, ResultT>::DispatchBaton {
size_t expectedCount_; size_t expectedCount_;
DispatchFunctionT dispatchFunction_; DispatchFunctionT dispatchFunction_;
std::vector<folly::Optional<Entry>> optEntries_; std::vector<folly::Optional<Entry>> optEntries_;
folly::Optional<std::string> optErrorMessage_; folly::exception_wrapper exceptionWrapper_;
}; };
template <typename InputT, typename ResultT> template <typename InputT, typename ResultT>
...@@ -158,7 +161,7 @@ Future<ResultT> AtomicBatchDispatcher<InputT, ResultT>::Token::dispatch( ...@@ -158,7 +161,7 @@ Future<ResultT> AtomicBatchDispatcher<InputT, ResultT>::Token::dispatch(
InputT input) { InputT input) {
auto baton = std::move(baton_); auto baton = std::move(baton_);
if (!baton) { if (!baton) {
throw std::logic_error( throw ABDUsageException(
"Dispatch called more than once on the same Token object"); "Dispatch called more than once on the same Token object");
} }
return baton->getFutureResult(std::move(input), sequenceNumber_); return baton->getFutureResult(std::move(input), sequenceNumber_);
...@@ -173,8 +176,10 @@ AtomicBatchDispatcher<InputT, ResultT>::AtomicBatchDispatcher( ...@@ -173,8 +176,10 @@ AtomicBatchDispatcher<InputT, ResultT>::AtomicBatchDispatcher(
template <typename InputT, typename ResultT> template <typename InputT, typename ResultT>
AtomicBatchDispatcher<InputT, ResultT>::~AtomicBatchDispatcher() { AtomicBatchDispatcher<InputT, ResultT>::~AtomicBatchDispatcher() {
if (baton_) { if (baton_) {
baton_->setError( // Set error here rather than throw because we do not want to throw from
"AtomicBatchDispatcher destroyed before commit() was called on it"); // the destructor of AtomicBatchDispatcher
baton_->setExceptionWrapper(
folly::make_exception_wrapper<ABDCommitNotCalledException>());
commit(); commit();
} }
} }
...@@ -182,7 +187,7 @@ AtomicBatchDispatcher<InputT, ResultT>::~AtomicBatchDispatcher() { ...@@ -182,7 +187,7 @@ AtomicBatchDispatcher<InputT, ResultT>::~AtomicBatchDispatcher() {
template <typename InputT, typename ResultT> template <typename InputT, typename ResultT>
void AtomicBatchDispatcher<InputT, ResultT>::reserve(size_t numEntries) { void AtomicBatchDispatcher<InputT, ResultT>::reserve(size_t numEntries) {
if (!baton_) { if (!baton_) {
throw std::logic_error("Cannot call reserve(....) after calling commit()"); throw ABDUsageException("Cannot call reserve(....) after calling commit()");
} }
baton_->reserve(numEntries); baton_->reserve(numEntries);
} }
...@@ -190,7 +195,7 @@ void AtomicBatchDispatcher<InputT, ResultT>::reserve(size_t numEntries) { ...@@ -190,7 +195,7 @@ void AtomicBatchDispatcher<InputT, ResultT>::reserve(size_t numEntries) {
template <typename InputT, typename ResultT> template <typename InputT, typename ResultT>
auto AtomicBatchDispatcher<InputT, ResultT>::getToken() -> Token { auto AtomicBatchDispatcher<InputT, ResultT>::getToken() -> Token {
if (!baton_) { if (!baton_) {
throw std::logic_error("Cannot issue more tokens after calling commit()"); throw ABDUsageException("Cannot issue more tokens after calling commit()");
} }
return Token(baton_, numTokensIssued_++); return Token(baton_, numTokensIssued_++);
} }
...@@ -199,7 +204,7 @@ template <typename InputT, typename ResultT> ...@@ -199,7 +204,7 @@ template <typename InputT, typename ResultT>
void AtomicBatchDispatcher<InputT, ResultT>::commit() { void AtomicBatchDispatcher<InputT, ResultT>::commit() {
auto baton = std::move(baton_); auto baton = std::move(baton_);
if (!baton) { if (!baton) {
throw std::logic_error( throw ABDUsageException(
"Cannot call commit() more than once on the same dispatcher"); "Cannot call commit() more than once on the same dispatcher");
} }
baton->setExpectedCount(numTokensIssued_); baton->setExpectedCount(numTokensIssued_);
......
...@@ -17,15 +17,50 @@ ...@@ -17,15 +17,50 @@
#include <folly/Function.h> #include <folly/Function.h>
#include <folly/Optional.h> #include <folly/Optional.h>
#include <folly/fibers/detail/AtomicBatchDispatcher.h>
#include <folly/futures/Future.h> #include <folly/futures/Future.h>
#include <folly/futures/Promise.h> #include <folly/futures/Promise.h>
#include <memory> #include <memory>
#include <stdexcept>
#include <string>
#include <utility> #include <utility>
#include <vector> #include <vector>
namespace folly { namespace folly {
namespace fibers { namespace fibers {
/**
* An exception class that gets thrown when the AtomicBatchDispatcher is used
* incorrectly. This is indicative of a bug in the user code.
* Examples are, multiple dispatch calls on the same token, trying to get more
* tokens from the dispatcher after commit has been called, etc.
*/
class ABDUsageException : public std::logic_error {
using std::logic_error::logic_error;
};
/**
* An exception class that gets set on the promise for dispatched tokens, when
* the AtomicBatchDispatcher was destroyed before commit was called on it.
*/
class ABDCommitNotCalledException : public std::runtime_error {
public:
ABDCommitNotCalledException()
: std::runtime_error(
"AtomicBatchDispatcher destroyed before commit() was called") {}
};
/**
* An exception class that gets set on the promise for dispatched tokens, when
* one or more other tokens in the batch were destroyed before dispatch was
* called on them.
* Only here so that the caller can distinguish the real failure cause
* rather than these subsequently thrown exceptions.
*/
class ABDTokenNotDispatchedException : public std::runtime_error {
using std::runtime_error::runtime_error;
};
/** /**
* AtomicBatchDispatcher should be used if you want to process fiber tasks in * AtomicBatchDispatcher should be used if you want to process fiber tasks in
* parallel, but require to synchronize them at some point. The canonical * parallel, but require to synchronize them at some point. The canonical
...@@ -72,21 +107,21 @@ namespace fibers { ...@@ -72,21 +107,21 @@ namespace fibers {
* 1) The dispatcher is destroyed before calling commit on it, for example * 1) The dispatcher is destroyed before calling commit on it, for example
* because the user forgot to call commit OR an exception was thrown * because the user forgot to call commit OR an exception was thrown
* in user code before the call to commit: * in user code before the call to commit:
* - The future ResultT has an exception of type std::logic_error set for all * - The future ResultT has an exception of type ABDCommitNotCalledException
* tokens that were issued by the dispatcher (once all tokens are either * set for all tokens that were issued by the dispatcher (once all tokens
* destroyed or have called dispatch) * are either destroyed or have called dispatch)
* 2) Calling the dispatch function more than once on the same Token object * 2) Calling the dispatch function more than once on the same Token object
* (or a moved version of the same Token): * (or a moved version of the same Token):
* - Subsequent calls to dispatch (after the first one) will throw an * - Subsequent calls to dispatch (after the first one) will throw an
* std::logic_error exception (the batch itself will not have any errors * ABDUsageException exception (the batch itself will not have any errors
* and will get processed) * and will get processed)
* 3) One/more of the Tokens issued are destroyed before calling dispatch on * 3) One/more of the Tokens issued are destroyed before calling dispatch on
* it/them: * it/them:
* - The future ResultT has an exception of type std::logic_error set for all * - The future ResultT has an ABDTokenNotDispatchedException set for all
* tokens that were issued by the dispatcher (once all tokens are either * tokens that were issued by the dispatcher (once all tokens are either
* destroyed or have called dispatch) * destroyed or have called dispatch)
* 4) dispatcher.getToken() is called after calling dispatcher.commit() * 4) dispatcher.getToken() is called after calling dispatcher.commit()
* - the call to getToken() will throw an std::logic_error exception * - the call to getToken() will throw an ABDUsageException exception
* (the batch itself will not have any errors and will get processed). * (the batch itself will not have any errors and will get processed).
* 5) All tokens were issued and called dispatch, the user provided batch * 5) All tokens were issued and called dispatch, the user provided batch
* dispatch function is called, but that function throws any exception. * dispatch function is called, but that function throws any exception.
......
/*
* Copyright 2017-present Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include "AtomicBatchDispatcher.h"
#include <folly/Format.h>
namespace folly {
namespace fibers {
namespace detail {
std::string createABDTokenNotDispatchedExMsg(
const std::vector<size_t>& vecTokensNotDispatched) {
size_t numTokensNotDispatched = vecTokensNotDispatched.size();
assert(numTokensNotDispatched > 0);
size_t numSeqNumToPrint =
(numTokensNotDispatched > 10 ? 10 : numTokensNotDispatched);
std::string strInputsNotFound =
folly::sformat("{}", vecTokensNotDispatched[0]);
for (size_t i = 1; i < numSeqNumToPrint; ++i) {
strInputsNotFound += folly::sformat(", {}", vecTokensNotDispatched[i]);
}
if (numSeqNumToPrint < numTokensNotDispatched) {
strInputsNotFound += "...";
}
return folly::sformat(
"{} input tokens (seq nums: {}) destroyed before calling dispatch",
numTokensNotDispatched,
strInputsNotFound);
}
std::string createUnexpectedNumResultsABDUsageExMsg(
size_t numExpectedResults,
size_t numActualResults) {
return folly::sformat(
"Unexpected number of results ({}) returned from dispatch function, "
"expected ({})",
numActualResults,
numExpectedResults);
}
} // namespace detail
} // namespace fibers
} // namespace folly
/*
* Copyright 2017 Facebook, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#pragma once
#include <string>
#include <vector>
namespace folly {
namespace fibers {
namespace detail {
std::string createABDTokenNotDispatchedExMsg(
const std::vector<size_t>& vecTokensNotDispatched);
std::string createUnexpectedNumResultsABDUsageExMsg(
size_t numExpectedResults,
size_t numActualResults);
} // namespace detail
} // namespace fibers
} // namespace folly
...@@ -1864,7 +1864,7 @@ void dispatchJobs( ...@@ -1864,7 +1864,7 @@ void dispatchJobs(
if (dispatchProblem == DispatchProblem::DuplicateDispatch) { if (dispatchProblem == DispatchProblem::DuplicateDispatch) {
if (i == problemIndex) { if (i == problemIndex) {
EXPECT_THROW(job.token.dispatch(job.input), std::logic_error); EXPECT_THROW(job.token.dispatch(job.input), ABDUsageException);
} }
} }
} catch (...) { } catch (...) {
...@@ -1969,7 +1969,7 @@ TEST(FiberManager, ABD_DispatcherDestroyedBeforeCallingCommit) { ...@@ -1969,7 +1969,7 @@ TEST(FiberManager, ABD_DispatcherDestroyedBeforeCallingCommit) {
/* User code handles the exception and does not exit process */ /* User code handles the exception and does not exit process */
} }
evb.loop(); evb.loop();
validateResults<std::logic_error>(results, COUNT); validateResults<ABDCommitNotCalledException>(results, COUNT);
} }
TEST(FiberManager, ABD_PreprocessingFailureTest) { TEST(FiberManager, ABD_PreprocessingFailureTest) {
...@@ -1985,7 +1985,7 @@ TEST(FiberManager, ABD_PreprocessingFailureTest) { ...@@ -1985,7 +1985,7 @@ TEST(FiberManager, ABD_PreprocessingFailureTest) {
dispatchJobs(executor, jobs, results, DispatchProblem::PreprocessThrows, 8); dispatchJobs(executor, jobs, results, DispatchProblem::PreprocessThrows, 8);
atomicBatchDispatcher.commit(); atomicBatchDispatcher.commit();
evb.loop(); evb.loop();
validateResults<std::logic_error>(results, COUNT - 1); validateResults<ABDTokenNotDispatchedException>(results, COUNT - 1);
} }
TEST(FiberManager, ABD_MultipleDispatchOnSameTokenErrorTest) { TEST(FiberManager, ABD_MultipleDispatchOnSameTokenErrorTest) {
...@@ -2014,12 +2014,12 @@ TEST(FiberManager, ABD_GetTokenCalledAfterCommitTest) { ...@@ -2014,12 +2014,12 @@ TEST(FiberManager, ABD_GetTokenCalledAfterCommitTest) {
createAtomicBatchDispatcher(std::move(dispatchFunc)); createAtomicBatchDispatcher(std::move(dispatchFunc));
createJobs(atomicBatchDispatcher, jobs, COUNT); createJobs(atomicBatchDispatcher, jobs, COUNT);
atomicBatchDispatcher.commit(); atomicBatchDispatcher.commit();
EXPECT_THROW(atomicBatchDispatcher.getToken(), std::logic_error); EXPECT_THROW(atomicBatchDispatcher.getToken(), ABDUsageException);
dispatchJobs(executor, jobs, results); dispatchJobs(executor, jobs, results);
EXPECT_THROW(atomicBatchDispatcher.getToken(), std::logic_error); EXPECT_THROW(atomicBatchDispatcher.getToken(), ABDUsageException);
evb.loop(); evb.loop();
validateResults(results, COUNT); validateResults(results, COUNT);
EXPECT_THROW(atomicBatchDispatcher.getToken(), std::logic_error); EXPECT_THROW(atomicBatchDispatcher.getToken(), ABDUsageException);
} }
TEST(FiberManager, ABD_UserProvidedBatchDispatchThrowsTest) { TEST(FiberManager, ABD_UserProvidedBatchDispatchThrowsTest) {
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment