apple/foundationdb
1
0
Fork 0
mirror of https://github.com/apple/foundationdb.git synced 2025-05-25 08:40:05 +08:00
Code Issues Packages Projects Releases Wiki Activity

catch exceptions

Browse Source
This commit is contained in:
Xiaoxi Wang 2021-07-09 22:46:16 +00:00
parent 5a43a8c367
commit 6d1c12899d
4 changed files with 15 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
fdbrpc/fdbrpc.h
View File

@ -406,8 +406,6 @@ struct NetNotifiedQueueWithAcknowledgements final : NotifiedQueue<T>,
template <class T> template <class T>
class ReplyPromiseStream { class ReplyPromiseStream {
public: public:
// The endpoints of a ReplyPromiseStream must be initialized at Task::ReadSocket, because with lower priorities a
// delay(0) in FlowTransport deliver can cause out of order delivery.
// stream.send( request ) // stream.send( request )
// Unreliable at most once delivery: Delivers request unless there is a connection failure (zero or one times) // Unreliable at most once delivery: Delivers request unless there is a connection failure (zero or one times)
@ -477,6 +475,8 @@ public:
errors->delPromiseRef(); errors->delPromiseRef();
} }
// The endpoints of a ReplyPromiseStream must be initialized at Task::ReadSocket, because with lower priorities a
// delay(0) in FlowTransport deliver can cause out of order delivery.
const Endpoint& getEndpoint() const { return queue->getEndpoint(TaskPriority::ReadSocket); } const Endpoint& getEndpoint() const { return queue->getEndpoint(TaskPriority::ReadSocket); }
bool operator==(const ReplyPromiseStream<T>& rhs) const { return queue == rhs.queue; } bool operator==(const ReplyPromiseStream<T>& rhs) const { return queue == rhs.queue; }

12
fdbserver/LogSystemPeekCursor.actor.cpp
View File

@ -386,13 +386,15 @@ Future<Void> ILogSystem::ServerPeekCursor::getMore(TaskPriority taskID) {
if (hasMessage() && !parallelGetMore) if (hasMessage() && !parallelGetMore)
return Void(); return Void();
if (!more.isValid() || more.isReady()) { if (!more.isValid() || more.isReady()) {
if (usePeekStream) { if (usePeekStream && taskID == TaskPriority::TLogPeekReply) {
more = serverPeekStreamGetMore(this, taskID); more = serverPeekStreamGetMore(this, taskID);
} else if (parallelGetMore || onlySpilled || futureResults.size()) {
more = serverPeekParallelGetMore(this, taskID);
} else {
more = serverPeekGetMore(this, taskID);
} }
// if (parallelGetMore || onlySpilled || futureResults.size()) {
// more = serverPeekParallelGetMore(this, taskID);
// }
else {
more = serverPeekGetMore(this, taskID);
}
} }
return more; return more;
} }

2
fdbserver/TLogInterface.h
View File

@ -83,7 +83,7 @@ struct TLogInterface {
streams.push_back(disablePopRequest.getReceiver()); streams.push_back(disablePopRequest.getReceiver());
streams.push_back(enablePopRequest.getReceiver()); streams.push_back(enablePopRequest.getReceiver());
streams.push_back(snapRequest.getReceiver()); streams.push_back(snapRequest.getReceiver());
streams.push_back(peekStreamMessages.getReceiver(TaskPriority::ReadSocket)); streams.push_back(peekStreamMessages.getReceiver(TaskPriority::TLogPeek));
FlowTransport::transport().addEndpoints(streams); FlowTransport::transport().addEndpoints(streams);
} }

8
fdbserver/TLogServer.actor.cpp
View File

@ -1916,15 +1916,17 @@ ACTOR Future<Void> tLogPeekStream(TLogData* self, TLogPeekStreamRequest req, Ref
req.reply.send(reply); req.reply.send(reply);
begin = reply.rep.end; begin = reply.rep.end;
onlySpilled = reply.rep.onlySpilled; onlySpilled = reply.rep.onlySpilled;
wait(delay(0.005, TaskPriority::TLogPeekReply));
// return Void(); // return Void();
} catch (Error& e) { } catch (Error& e) {
if (e.code() == error_code_end_of_stream) { if (e.code() == error_code_end_of_stream) {
req.reply.sendError(e); req.reply.sendError(e);
return Void(); return Void();
}
else if (e.code() == error_code_operation_obsolete) {
// reply stream is cancelled on the client
return Void();
} else { } else {
throw; throw;
} }
} }
} }