refactor: simplify journal and restore streamer cancellation NOT READY FOR REVIEW #4549
Conversation
| @@ -42,7 +42,7 @@ JournalStreamer::JournalStreamer(journal::Journal* journal, Context* cntx) | |||
| } | |||
|
|
|||
| JournalStreamer::~JournalStreamer() { | |||
| if (!cntx_->IsCancelled()) { | |||
| if (!cntx_->GetError()) { | |||
There was a problem hiding this comment.
Why GetError() here and not IsCancelled()
There was a problem hiding this comment.
because if we cancel the streamer, we should wait for WaitForInflightToComplete() and in_flight_bytes_ should be 0
otherwise not
There was a problem hiding this comment.
But it's the exact same thing? We can cancel a context in two ways:
- Calling
Contenxt::Cancelwhich submits the error codeoperation_cancelto the underline error handler via theReportError - By calling
ReportErrorwhich submits the error code passed as an argument to the underline error handler
Both (1) and (2) call Cancellation::Cancel so IsCancelled() will always be true for both paths.
Why do we need to fetch the error here via GetError and why does not !cntx_->IsCancelled() is not enough in this context ?
I could be missing something -- I just want to make sure I understand this
There was a problem hiding this comment.
Thx @kostasrim you are right. Honestly, such context interface doesn't look logical to me and I thought the functions are different. your comment helps me
| cntx_->ReportError(ec); | ||
| } else if (!pending_buf_.Empty() && !IsStopped()) { | ||
| AsyncWrite(); | ||
| if (!cntx_->IsCancelled()) { |
There was a problem hiding this comment.
Do we care loosing the error code here on cancellation ? Could it be that the AsyncWrite fails for some other reason ? Would it be useful to store this error or log it ?
There was a problem hiding this comment.
we don't care regarding error if we cancel streamer
| @@ -59,10 +59,6 @@ class JournalStreamer { | |||
| void AsyncWrite(); | |||
| void OnCompletion(std::error_code ec, size_t len); | |||
|
|
|||
| bool IsStopped() const { | |||
| return cntx_->IsCancelled(); | |||
| @@ -91,10 +87,6 @@ class RestoreStreamer : public JournalStreamer { | |||
|
|
|||
| void SendFinalize(long attempt); | |||
|
|
|||
| bool IsSnapshotFinished() const { | |||
BorysTheDev
force-pushed
the
streamer_refactoring
branch
2 times, most recently
from
4a3cabc to
3735d47
Compare
| @@ -36,7 +36,9 @@ class OutgoingMigration::SliceSlotMigration : private ProtocolClient { | |||
| SliceSlotMigration(DbSlice* slice, ServerContext server_context, SlotSet slots, | |||
| journal::Journal* journal, OutgoingMigration* om) | |||
| : ProtocolClient(server_context), streamer_(slice, std::move(slots), journal, &cntx_) { | |||
| cntx_.SwitchErrorHandler([om](auto ge) { om->Finish(std::move(ge)); }); | |||
| cntx_.SwitchErrorHandler([om](auto ge) { | |||
| om->Finish(ge != errc::operation_canceled ? std::move(ge) : GenericError()); | |||
There was a problem hiding this comment.
Why are we purging operation_cancelled and replacing it with an empty GenericError ?
|
@BorysTheDev plz do not force push once a PR has comments. It's impossible to keep track of the logical changes between different commits.
|
BorysTheDev
changed the title
BorysTheDev
force-pushed
the
streamer_refactoring
branch
from
3735d47 to
fa338a6
Compare
fixes: #4284