Commit c3ee57ef authored by James Sedgwick's avatar James Sedgwick Committed by Dave Watson

OutputBufferingHandler

Summary:
see D1618704 and D1604575
Really need to make a decision about making a push for EventBaseManager everywhere. Getting/setting EBs when there's only ever one seems so silly, but in (e.g.) tests where one-off EBs are used without using the EBM, stuff like this will break.

Test Plan: definitely compiles and sorta works, will write unit tests if the dust settles

Reviewed By: davejwatson@fb.com

Subscribers: folly-diffs@, trunkagent, fugalh, njormrod

FB internal diff: D1618727
parent 923576ab
/*
* Copyright 2014 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 <folly/experimental/wangle/channel/ChannelHandler.h>
#include <folly/io/async/EventBase.h>
#include <folly/io/async/EventBaseManager.h>
#include <folly/io/IOBuf.h>
#include <folly/io/IOBufQueue.h>
namespace folly { namespace wangle {
/*
* OutputBufferingHandler buffers writes in order to minimize syscalls. The
* transport will be written to once per event loop instead of on every write.
*/
class OutputBufferingHandler : public BytesToBytesHandler,
protected EventBase::LoopCallback {
public:
Future<void> write(Context* ctx, std::unique_ptr<IOBuf> buf) override {
CHECK(buf);
if (!queueSends_) {
return ctx->fireWrite(std::move(buf));
} else {
ctx_ = ctx;
// Delay sends to optimize for fewer syscalls
if (!sends_) {
DCHECK(!isLoopCallbackScheduled());
// Buffer all the sends, and call writev once per event loop.
sends_ = std::move(buf);
ctx->getTransport()->getEventBase()->runInLoop(this);
} else {
DCHECK(isLoopCallbackScheduled());
sends_->prependChain(std::move(buf));
}
Promise<void> p;
auto f = p.getFuture();
promises_.push_back(std::move(p));
return f;
}
}
void runLoopCallback() noexcept override {
MoveWrapper<std::vector<Promise<void>>> promises(std::move(promises_));
ctx_->fireWrite(std::move(sends_)).then([promises](Try<void>&& t) mutable {
try {
t.throwIfFailed();
for (auto& p : *promises) {
p.setValue();
}
} catch (...) {
for (auto& p : *promises) {
p.setException(std::current_exception());
}
}
});
}
std::vector<Promise<void>> promises_;
std::unique_ptr<IOBuf> sends_{nullptr};
bool queueSends_{true};
Context* ctx_;
};
}}
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