You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by be...@apache.org on 2017/01/09 03:27:18 UTC
[1/5] mesos git commit: Used process::loop to avoid stack overflow
due to recursion.
Repository: mesos
Updated Branches:
refs/heads/master 9942fa774 -> 2d28c198a
Used process::loop to avoid stack overflow due to recursion.
Review: https://reviews.apache.org/r/54840
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/99844492
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/99844492
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/99844492
Branch: refs/heads/master
Commit: 998444928b9e5c46403af9a6c38fc3d05b660539
Parents: ab2565c
Author: Benjamin Hindman <be...@gmail.com>
Authored: Sat Dec 17 13:29:27 2016 -0800
Committer: Benjamin Hindman <be...@gmail.com>
Committed: Sat Jan 7 23:22:24 2017 -0800
----------------------------------------------------------------------
3rdparty/libprocess/src/http.cpp | 110 ++++++++++++++++++----------------
1 file changed, 58 insertions(+), 52 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/99844492/3rdparty/libprocess/src/http.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/http.cpp b/3rdparty/libprocess/src/http.cpp
index 689a14d..131f7ae 100644
--- a/3rdparty/libprocess/src/http.cpp
+++ b/3rdparty/libprocess/src/http.cpp
@@ -1396,34 +1396,36 @@ namespace internal {
Future<Nothing> send(network::Socket socket, Encoder* encoder)
{
- size_t* size = new size_t();
- return [=]() {
- switch (encoder->kind()) {
- case Encoder::DATA: {
- const char* data = static_cast<DataEncoder*>(encoder)->next(size);
- return socket.send(data, *size);
- }
- case Encoder::FILE: {
- off_t offset = 0;
- int fd = static_cast<FileEncoder*>(encoder)->next(&offset, size);
- return socket.sendfile(fd, offset, *size);
- }
- }
- }()
- .then([=](size_t length) -> Future<Nothing> {
- // Update the encoder with the amount sent.
- encoder->backup(*size - length);
-
- // See if there is any more of the message to send.
- if (encoder->remaining() != 0) {
- return send(socket, encoder);
- }
+ size_t* size = new size_t(0);
+ return loop(
+ None(),
+ [=]() {
+ switch (encoder->kind()) {
+ case Encoder::DATA: {
+ const char* data = static_cast<DataEncoder*>(encoder)->next(size);
+ return socket.send(data, *size);
+ }
+ case Encoder::FILE: {
+ off_t offset = 0;
+ int fd = static_cast<FileEncoder*>(encoder)->next(&offset, size);
+ return socket.sendfile(fd, offset, *size);
+ }
+ }
+ },
+ [=](size_t length) -> ControlFlow<Nothing> {
+ // Update the encoder with the amount sent.
+ encoder->backup(*size - length);
- return Nothing();
- })
- .onAny([=]() {
- delete size;
- });
+ // See if there is any more of the message to send.
+ if (encoder->remaining() != 0) {
+ return Continue();
+ }
+
+ return Break();
+ })
+ .onAny([=]() {
+ delete size;
+ });
}
@@ -1515,36 +1517,40 @@ Future<Nothing> stream(
const network::Socket& socket,
http::Pipe::Reader reader)
{
- return reader.read()
- .then([=](const string& data) mutable {
- bool finished = false;
+ return loop(
+ None(),
+ [=]() mutable {
+ return reader.read();
+ },
+ [=](const string& data) mutable {
+ bool finished = false;
- ostringstream out;
+ ostringstream out;
- if (data.empty()) {
- // Finished reading.
- out << "0\r\n" << "\r\n";
- finished = true;
- } else {
- out << std::hex << data.size() << "\r\n";
- out << data;
- out << "\r\n";
- }
+ if (data.empty()) {
+ // Finished reading.
+ out << "0\r\n" << "\r\n";
+ finished = true;
+ } else {
+ out << std::hex << data.size() << "\r\n";
+ out << data;
+ out << "\r\n";
+ }
- Encoder* encoder = new DataEncoder(out.str());
+ Encoder* encoder = new DataEncoder(out.str());
- return send(socket, encoder)
- .onAny([=]() {
- delete encoder;
- })
- .then([=]() mutable -> Future<Nothing> {
- if (!finished) {
- return stream(socket, reader);
- }
+ return send(socket, encoder)
+ .onAny([=]() {
+ delete encoder;
+ })
+ .then([=]() mutable -> ControlFlow<Nothing> {
+ if (!finished) {
+ return Continue();
+ }
- return Nothing();
- });
- });
+ return Break();
+ });
+ });
}
Re: [5/5] mesos git commit: Used `loop` in implementation of io::read
and io::write.
Posted by Michael Park <mp...@apache.org>.
Pushed a temporary fix in:
```
commit b492d4458c97b5e025e96ca229bc87a6e2500b40
Author: Jan Schlicht <ja...@mesosphere.io>
Date: Mon Jan 9 15:39:21 2017 -0800
Removed unsupported `friend` declaration.
The `friend` class declaration of nested, templated classes will raise
an error when compiling with Clang. This is a temporary fix, resolved by
making the constructor of `ControlFlow` public.
Review: https://reviews.apache.org/r/55339/
```
On Mon, Jan 9, 2017 at 2:52 PM, Benjamin Mahler <bm...@apache.org> wrote:
> Looks like there are some warnings for some of the loop changes:
>
> ../../../3rdparty/libprocess/src/io.cpp: In lambda function:
> ../../../3rdparty/libprocess/src/io.cpp:75:9: warning: control reaches
> end of non-void function [-Wreturn-type]
> }();
> ^
> mv -f .deps/libprocess_la-io.Tpo .deps/libprocess_la-io.Plo
> mv -f .deps/libprocess_la-metrics.Tpo .deps/libprocess_la-metrics.Plo
> ../../../3rdparty/libprocess/src/http.cpp: In lambda function:
> ../../../3rdparty/libprocess/src/http.cpp:1414:7: warning: control
> reaches end of non-void function [-Wreturn-type]
> },
> ^
> ../../../3rdparty/libprocess/src/http.cpp: In lambda function:
> ../../../3rdparty/libprocess/src/http.cpp:1646:13: warning: control
> reaches end of non-void function [-Wreturn-type]
> }()
> ^
>
> On Sun, Jan 8, 2017 at 7:27 PM, <be...@apache.org> wrote:
>
>> Used `loop` in implementation of io::read and io::write.
>>
>> Review: https://reviews.apache.org/r/54841
>>
>>
>> Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
>> Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/2d28c198
>> Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/2d28c198
>> Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/2d28c198
>>
>> Branch: refs/heads/master
>> Commit: 2d28c198a5c09308825b2771483b70ac42839d16
>> Parents: 9984449
>> Author: Benjamin Hindman <be...@gmail.com>
>> Authored: Mon Dec 5 10:48:56 2016 -0800
>> Committer: Benjamin Hindman <be...@gmail.com>
>> Committed: Sun Jan 8 19:27:02 2017 -0800
>>
>> ----------------------------------------------------------------------
>> 3rdparty/libprocess/src/io.cpp | 289 ++++++++++++++----------------
>> ------
>> 1 file changed, 114 insertions(+), 175 deletions(-)
>> ----------------------------------------------------------------------
>>
>>
>> http://git-wip-us.apache.org/repos/asf/mesos/blob/2d28c198/3
>> rdparty/libprocess/src/io.cpp
>> ----------------------------------------------------------------------
>> diff --git a/3rdparty/libprocess/src/io.cpp
>> b/3rdparty/libprocess/src/io.cpp
>> index d0b3ba1..8aa3576 100644
>> --- a/3rdparty/libprocess/src/io.cpp
>> +++ b/3rdparty/libprocess/src/io.cpp
>> @@ -42,132 +42,114 @@ enum ReadFlags
>> };
>>
>>
>> -void read(
>> - int fd,
>> - void* data,
>> - size_t size,
>> - ReadFlags flags,
>> - const std::shared_ptr<Promise<size_t>>& promise,
>> - const Future<short>& future)
>> +Future<size_t> read(int fd, void* data, size_t size, ReadFlags flags =
>> NONE)
>> {
>> - // Ignore this function if the read operation has been discarded.
>> - if (promise->future().hasDiscard()) {
>> - CHECK(!future.isPending());
>> - promise->discard();
>> - return;
>> - }
>> -
>> + // TODO(benh): Let the system calls do what ever they're supposed to
>> + // rather than return 0 here?
>> if (size == 0) {
>> - promise->set(0);
>> - return;
>> + return 0;
>> }
>>
>> - if (future.isDiscarded()) {
>> - promise->fail("Failed to poll: discarded future");
>> - } else if (future.isFailed()) {
>> - promise->fail(future.failure());
>> - } else {
>> - ssize_t length;
>> - if (flags == NONE) {
>> - length = os::read(fd, data, size);
>> - } else { // PEEK.
>> - // In case 'fd' is not a socket ::recv() will fail with ENOTSOCK
>> and the
>> - // error will be propagted out.
>> - // NOTE: We cast to `char*` here because the function prototypes on
>> - // Windows use `char*` instead of `void*`.
>> - length = net::recv(fd, (char*) data, size, MSG_PEEK);
>> - }
>> -
>> + return loop(
>> + None(),
>> + [=]() -> Future<Option<size_t>> {
>> + // Because the file descriptor is non-blocking, we call
>> + // read()/recv() immediately. If no data is available than
>> + // we'll call `poll` and block. We also observed that for some
>> + // combination of libev and Linux kernel versions, the poll
>> + // would block for non-deterministically long periods of
>> + // time. This may be fixed in a newer version of libev (we use
>> + // 3.8 at the time of writing this comment).
>> + ssize_t length = [=]() {
>> + switch (flags) {
>> + case PEEK:
>> + // In case `fd` is not a socket os::recv() will fail
>> + // with ENOTSOCK and the error will be returned.
>> + //
>> + // NOTE: We cast to `char*` here because the function
>> + // prototypes on Windows use `char*` instead of `void*`.
>> + return net::recv(fd, (char*) data, size, MSG_PEEK);
>> + case NONE:
>> + return os::read(fd, data, size);
>> + }
>> + }();
>> +
>> + if (length < 0) {
>> #ifdef __WINDOWS__
>> - int error = WSAGetLastError();
>> + int error = WSAGetLastError();
>> #else
>> - int error = errno;
>> + int error = errno;
>> #endif // __WINDOWS__
>>
>> - if (length < 0) {
>> - if (net::is_restartable_error(error) ||
>> net::is_retryable_error(error)) {
>> - // Restart the read operation.
>> - Future<short> future =
>> - io::poll(fd, process::io::READ).onAny(
>> - lambda::bind(&internal::read,
>> - fd,
>> - data,
>> - size,
>> - flags,
>> - promise,
>> - lambda::_1));
>> -
>> - // Stop polling if a discard occurs on our future.
>> - promise->future().onDiscard(
>> - lambda::bind(&process::internal::discard<short>,
>> - WeakFuture<short>(future)));
>> - } else {
>> - // Error occurred.
>> - promise->fail(os::strerror(errno));
>> - }
>> - } else {
>> - promise->set(length);
>> - }
>> - }
>> + if (!net::is_restartable_error(error) &&
>> + !net::is_retryable_error(error)) {
>> + // TODO(benh): Confirm that `os::strerror` does the right
>> + // thing for `error` on Windows.
>> + return Failure(os::strerror(error));
>> + }
>> +
>> + return None();
>> + }
>> +
>> + return length;
>> + },
>> + [=](const Option<size_t>& length) -> Future<ControlFlow<size_t>> {
>> + // Restart/retry if we don't yet have a result.
>> + if (length.isNone()) {
>> + return io::poll(fd, io::READ)
>> + .then([](short event) -> ControlFlow<size_t> {
>> + CHECK_EQ(io::READ, event);
>> + return Continue();
>> + });
>> + }
>> + return Break(length.get());
>> + });
>> }
>>
>>
>> -void write(
>> - int fd,
>> - const void* data,
>> - size_t size,
>> - const std::shared_ptr<Promise<size_t>>& promise,
>> - const Future<short>& future)
>> +Future<size_t> write(int fd, const void* data, size_t size)
>> {
>> - // Ignore this function if the write operation has been discarded.
>> - if (promise->future().hasDiscard()) {
>> - promise->discard();
>> - return;
>> - }
>> -
>> + // TODO(benh): Let the system calls do what ever they're supposed to
>> + // rather than return 0 here?
>> if (size == 0) {
>> - promise->set(0);
>> - return;
>> + return 0;
>> }
>>
>> - if (future.isDiscarded()) {
>> - promise->fail("Failed to poll: discarded future");
>> - } else if (future.isFailed()) {
>> - promise->fail(future.failure());
>> - } else {
>> - ssize_t length = os::write(fd, data, size);
>> + return loop(
>> + None(),
>> + [=]() -> Future<Option<size_t>> {
>> + ssize_t length = os::write(fd, data, size);
>>
>> + if (length < 0) {
>> #ifdef __WINDOWS__
>> - int error = WSAGetLastError();
>> + int error = WSAGetLastError();
>> #else
>> - int error = errno;
>> + int error = errno;
>> #endif // __WINDOWS__
>>
>> - if (length < 0) {
>> - if (net::is_restartable_error(error) ||
>> net::is_retryable_error(error)) {
>> - // Restart the write operation.
>> - Future<short> future =
>> - io::poll(fd, process::io::WRITE).onAny(
>> - lambda::bind(&internal::write,
>> - fd,
>> - data,
>> - size,
>> - promise,
>> - lambda::_1));
>> -
>> - // Stop polling if a discard occurs on our future.
>> - promise->future().onDiscard(
>> - lambda::bind(&process::internal::discard<short>,
>> - WeakFuture<short>(future)));
>> - } else {
>> - // Error occurred.
>> - promise->fail(os::strerror(errno));
>> - }
>> - } else {
>> - // TODO(benh): Retry if 'length' is 0?
>> - promise->set(length);
>> - }
>> - }
>> + if (!net::is_restartable_error(error) &&
>> + !net::is_retryable_error(error)) {
>> + // TODO(benh): Confirm that `os::strerror` does the right
>> + // thing for `error` on Windows.
>> + return Failure(os::strerror(error));
>> + }
>> +
>> + return None();
>> + }
>> +
>> + return length;
>> + },
>> + [=](const Option<size_t>& length) -> Future<ControlFlow<size_t>> {
>> + // Restart/retry if we don't yet have a result.
>> + if (length.isNone()) {
>> + return io::poll(fd, io::WRITE)
>> + .then([](short event) -> ControlFlow<size_t> {
>> + CHECK_EQ(io::WRITE, event);
>> + return Continue();
>> + });
>> + }
>> + return Break(length.get());
>> + });
>> }
>>
>> } // namespace internal {
>> @@ -177,32 +159,18 @@ Future<size_t> read(int fd, void* data, size_t size)
>> {
>> process::initialize();
>>
>> - std::shared_ptr<Promise<size_t>> promise(new Promise<size_t>());
>> -
>> // Check the file descriptor.
>> Try<bool> nonblock = os::isNonblock(fd);
>> if (nonblock.isError()) {
>> // The file descriptor is not valid (e.g., has been closed).
>> - promise->fail(
>> - "Failed to check if file descriptor was non-blocking: " +
>> - nonblock.error());
>> - return promise->future();
>> + return Failure("Failed to check if file descriptor was non-blocking:
>> " +
>> + nonblock.error());
>> } else if (!nonblock.get()) {
>> // The file descriptor is not non-blocking.
>> - promise->fail("Expected a non-blocking file descriptor");
>> - return promise->future();
>> + return Failure("Expected a non-blocking file descriptor");
>> }
>>
>> - // Because the file descriptor is non-blocking, we call read()
>> - // immediately. The read may in turn call poll if necessary,
>> - // avoiding unnecessary polling. We also observed that for some
>> - // combination of libev and Linux kernel versions, the poll would
>> - // block for non-deterministically long periods of time. This may be
>> - // fixed in a newer version of libev (we use 3.8 at the time of
>> - // writing this comment).
>> - internal::read(fd, data, size, internal::NONE, promise, io::READ);
>> -
>> - return promise->future();
>> + return internal::read(fd, data, size);
>> }
>>
>>
>> @@ -210,32 +178,19 @@ Future<size_t> write(int fd, const void* data,
>> size_t size)
>> {
>> process::initialize();
>>
>> - std::shared_ptr<Promise<size_t>> promise(new Promise<size_t>());
>> -
>> // Check the file descriptor.
>> Try<bool> nonblock = os::isNonblock(fd);
>> if (nonblock.isError()) {
>> // The file descriptor is not valid (e.g., has been closed).
>> - promise->fail(
>> + return Failure(
>> "Failed to check if file descriptor was non-blocking: " +
>> nonblock.error());
>> - return promise->future();
>> } else if (!nonblock.get()) {
>> // The file descriptor is not non-blocking.
>> - promise->fail("Expected a non-blocking file descriptor");
>> - return promise->future();
>> + return Failure("Expected a non-blocking file descriptor");
>> }
>>
>> - // Because the file descriptor is non-blocking, we call write()
>> - // immediately. The write may in turn call poll if necessary,
>> - // avoiding unnecessary polling. We also observed that for some
>> - // combination of libev and Linux kernel versions, the poll would
>> - // block for non-deterministically long periods of time. This may be
>> - // fixed in a newer version of libev (we use 3.8 at the time of
>> - // writing this comment).
>> - internal::write(fd, data, size, promise, io::WRITE);
>> -
>> - return promise->future();
>> + return internal::write(fd, data, size);
>> }
>>
>>
>> @@ -280,43 +235,15 @@ Future<size_t> peek(int fd, void* data, size_t
>> size, size_t limit)
>> nonblock.error());
>> }
>>
>> - std::shared_ptr<Promise<size_t>> promise(new Promise<size_t>());
>> -
>> - // Because the file descriptor is non-blocking, we call read()
>> - // immediately. The read may in turn call poll if necessary,
>> - // avoiding unnecessary polling. We also observed that for some
>> - // combination of libev and Linux kernel versions, the poll would
>> - // block for non-deterministically long periods of time. This may be
>> - // fixed in a newer version of libev (we use 3.8 at the time of
>> - // writing this comment).
>> - internal::read(fd, data, limit, internal::PEEK, promise, io::READ);
>> -
>> - // NOTE: We wrap `os::close` in a lambda to disambiguate on Windows.
>> - promise->future().onAny([fd]() { os::close(fd); });
>> -
>> - return promise->future();
>> + return internal::read(fd, data, limit, internal::PEEK)
>> + .onAny([fd]() {
>> + os::close(fd);
>> + });
>> }
>>
>>
>> namespace internal {
>>
>> -Future<string> _read(
>> - int fd,
>> - const std::shared_ptr<string>& buffer,
>> - const boost::shared_array<char>& data,
>> - size_t length)
>> -{
>> - return io::read(fd, data.get(), length)
>> - .then([=](size_t size) -> Future<string> {
>> - if (size == 0) { // EOF.
>> - return string(*buffer);
>> - }
>> - buffer->append(data.get(), size);
>> - return _read(fd, buffer, data, length);
>> - });
>> -}
>> -
>> -
>> Future<Nothing> splice(
>> int from,
>> int to,
>> @@ -392,9 +319,21 @@ Future<string> read(int fd)
>> std::shared_ptr<string> buffer(new string());
>> boost::shared_array<char> data(new char[BUFFERED_READ_SIZE]);
>>
>> - // NOTE: We wrap `os::close` in a lambda to disambiguate on Windows.
>> - return internal::_read(fd, buffer, data, BUFFERED_READ_SIZE)
>> - .onAny([fd]() { os::close(fd); });
>> + return loop(
>> + None(),
>> + [=]() {
>> + return io::read(fd, data.get(), BUFFERED_READ_SIZE);
>> + },
>> + [=](size_t length) -> ControlFlow<string> {
>> + if (length == 0) { // EOF.
>> + return Break(std::move(*buffer));
>> + }
>> + buffer->append(data.get(), length);
>> + return Continue();
>> + })
>> + .onAny([fd]() {
>> + os::close(fd);
>> + });
>> }
>>
>>
>>
>>
>
Re: [5/5] mesos git commit: Used `loop` in implementation of io::read
and io::write.
Posted by Michael Park <mp...@apache.org>.
Pushed a temporary fix in:
```
commit b492d4458c97b5e025e96ca229bc87a6e2500b40
Author: Jan Schlicht <ja...@mesosphere.io>
Date: Mon Jan 9 15:39:21 2017 -0800
Removed unsupported `friend` declaration.
The `friend` class declaration of nested, templated classes will raise
an error when compiling with Clang. This is a temporary fix, resolved by
making the constructor of `ControlFlow` public.
Review: https://reviews.apache.org/r/55339/
```
On Mon, Jan 9, 2017 at 2:52 PM, Benjamin Mahler <bm...@apache.org> wrote:
> Looks like there are some warnings for some of the loop changes:
>
> ../../../3rdparty/libprocess/src/io.cpp: In lambda function:
> ../../../3rdparty/libprocess/src/io.cpp:75:9: warning: control reaches
> end of non-void function [-Wreturn-type]
> }();
> ^
> mv -f .deps/libprocess_la-io.Tpo .deps/libprocess_la-io.Plo
> mv -f .deps/libprocess_la-metrics.Tpo .deps/libprocess_la-metrics.Plo
> ../../../3rdparty/libprocess/src/http.cpp: In lambda function:
> ../../../3rdparty/libprocess/src/http.cpp:1414:7: warning: control
> reaches end of non-void function [-Wreturn-type]
> },
> ^
> ../../../3rdparty/libprocess/src/http.cpp: In lambda function:
> ../../../3rdparty/libprocess/src/http.cpp:1646:13: warning: control
> reaches end of non-void function [-Wreturn-type]
> }()
> ^
>
> On Sun, Jan 8, 2017 at 7:27 PM, <be...@apache.org> wrote:
>
>> Used `loop` in implementation of io::read and io::write.
>>
>> Review: https://reviews.apache.org/r/54841
>>
>>
>> Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
>> Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/2d28c198
>> Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/2d28c198
>> Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/2d28c198
>>
>> Branch: refs/heads/master
>> Commit: 2d28c198a5c09308825b2771483b70ac42839d16
>> Parents: 9984449
>> Author: Benjamin Hindman <be...@gmail.com>
>> Authored: Mon Dec 5 10:48:56 2016 -0800
>> Committer: Benjamin Hindman <be...@gmail.com>
>> Committed: Sun Jan 8 19:27:02 2017 -0800
>>
>> ----------------------------------------------------------------------
>> 3rdparty/libprocess/src/io.cpp | 289 ++++++++++++++----------------
>> ------
>> 1 file changed, 114 insertions(+), 175 deletions(-)
>> ----------------------------------------------------------------------
>>
>>
>> http://git-wip-us.apache.org/repos/asf/mesos/blob/2d28c198/3
>> rdparty/libprocess/src/io.cpp
>> ----------------------------------------------------------------------
>> diff --git a/3rdparty/libprocess/src/io.cpp
>> b/3rdparty/libprocess/src/io.cpp
>> index d0b3ba1..8aa3576 100644
>> --- a/3rdparty/libprocess/src/io.cpp
>> +++ b/3rdparty/libprocess/src/io.cpp
>> @@ -42,132 +42,114 @@ enum ReadFlags
>> };
>>
>>
>> -void read(
>> - int fd,
>> - void* data,
>> - size_t size,
>> - ReadFlags flags,
>> - const std::shared_ptr<Promise<size_t>>& promise,
>> - const Future<short>& future)
>> +Future<size_t> read(int fd, void* data, size_t size, ReadFlags flags =
>> NONE)
>> {
>> - // Ignore this function if the read operation has been discarded.
>> - if (promise->future().hasDiscard()) {
>> - CHECK(!future.isPending());
>> - promise->discard();
>> - return;
>> - }
>> -
>> + // TODO(benh): Let the system calls do what ever they're supposed to
>> + // rather than return 0 here?
>> if (size == 0) {
>> - promise->set(0);
>> - return;
>> + return 0;
>> }
>>
>> - if (future.isDiscarded()) {
>> - promise->fail("Failed to poll: discarded future");
>> - } else if (future.isFailed()) {
>> - promise->fail(future.failure());
>> - } else {
>> - ssize_t length;
>> - if (flags == NONE) {
>> - length = os::read(fd, data, size);
>> - } else { // PEEK.
>> - // In case 'fd' is not a socket ::recv() will fail with ENOTSOCK
>> and the
>> - // error will be propagted out.
>> - // NOTE: We cast to `char*` here because the function prototypes on
>> - // Windows use `char*` instead of `void*`.
>> - length = net::recv(fd, (char*) data, size, MSG_PEEK);
>> - }
>> -
>> + return loop(
>> + None(),
>> + [=]() -> Future<Option<size_t>> {
>> + // Because the file descriptor is non-blocking, we call
>> + // read()/recv() immediately. If no data is available than
>> + // we'll call `poll` and block. We also observed that for some
>> + // combination of libev and Linux kernel versions, the poll
>> + // would block for non-deterministically long periods of
>> + // time. This may be fixed in a newer version of libev (we use
>> + // 3.8 at the time of writing this comment).
>> + ssize_t length = [=]() {
>> + switch (flags) {
>> + case PEEK:
>> + // In case `fd` is not a socket os::recv() will fail
>> + // with ENOTSOCK and the error will be returned.
>> + //
>> + // NOTE: We cast to `char*` here because the function
>> + // prototypes on Windows use `char*` instead of `void*`.
>> + return net::recv(fd, (char*) data, size, MSG_PEEK);
>> + case NONE:
>> + return os::read(fd, data, size);
>> + }
>> + }();
>> +
>> + if (length < 0) {
>> #ifdef __WINDOWS__
>> - int error = WSAGetLastError();
>> + int error = WSAGetLastError();
>> #else
>> - int error = errno;
>> + int error = errno;
>> #endif // __WINDOWS__
>>
>> - if (length < 0) {
>> - if (net::is_restartable_error(error) ||
>> net::is_retryable_error(error)) {
>> - // Restart the read operation.
>> - Future<short> future =
>> - io::poll(fd, process::io::READ).onAny(
>> - lambda::bind(&internal::read,
>> - fd,
>> - data,
>> - size,
>> - flags,
>> - promise,
>> - lambda::_1));
>> -
>> - // Stop polling if a discard occurs on our future.
>> - promise->future().onDiscard(
>> - lambda::bind(&process::internal::discard<short>,
>> - WeakFuture<short>(future)));
>> - } else {
>> - // Error occurred.
>> - promise->fail(os::strerror(errno));
>> - }
>> - } else {
>> - promise->set(length);
>> - }
>> - }
>> + if (!net::is_restartable_error(error) &&
>> + !net::is_retryable_error(error)) {
>> + // TODO(benh): Confirm that `os::strerror` does the right
>> + // thing for `error` on Windows.
>> + return Failure(os::strerror(error));
>> + }
>> +
>> + return None();
>> + }
>> +
>> + return length;
>> + },
>> + [=](const Option<size_t>& length) -> Future<ControlFlow<size_t>> {
>> + // Restart/retry if we don't yet have a result.
>> + if (length.isNone()) {
>> + return io::poll(fd, io::READ)
>> + .then([](short event) -> ControlFlow<size_t> {
>> + CHECK_EQ(io::READ, event);
>> + return Continue();
>> + });
>> + }
>> + return Break(length.get());
>> + });
>> }
>>
>>
>> -void write(
>> - int fd,
>> - const void* data,
>> - size_t size,
>> - const std::shared_ptr<Promise<size_t>>& promise,
>> - const Future<short>& future)
>> +Future<size_t> write(int fd, const void* data, size_t size)
>> {
>> - // Ignore this function if the write operation has been discarded.
>> - if (promise->future().hasDiscard()) {
>> - promise->discard();
>> - return;
>> - }
>> -
>> + // TODO(benh): Let the system calls do what ever they're supposed to
>> + // rather than return 0 here?
>> if (size == 0) {
>> - promise->set(0);
>> - return;
>> + return 0;
>> }
>>
>> - if (future.isDiscarded()) {
>> - promise->fail("Failed to poll: discarded future");
>> - } else if (future.isFailed()) {
>> - promise->fail(future.failure());
>> - } else {
>> - ssize_t length = os::write(fd, data, size);
>> + return loop(
>> + None(),
>> + [=]() -> Future<Option<size_t>> {
>> + ssize_t length = os::write(fd, data, size);
>>
>> + if (length < 0) {
>> #ifdef __WINDOWS__
>> - int error = WSAGetLastError();
>> + int error = WSAGetLastError();
>> #else
>> - int error = errno;
>> + int error = errno;
>> #endif // __WINDOWS__
>>
>> - if (length < 0) {
>> - if (net::is_restartable_error(error) ||
>> net::is_retryable_error(error)) {
>> - // Restart the write operation.
>> - Future<short> future =
>> - io::poll(fd, process::io::WRITE).onAny(
>> - lambda::bind(&internal::write,
>> - fd,
>> - data,
>> - size,
>> - promise,
>> - lambda::_1));
>> -
>> - // Stop polling if a discard occurs on our future.
>> - promise->future().onDiscard(
>> - lambda::bind(&process::internal::discard<short>,
>> - WeakFuture<short>(future)));
>> - } else {
>> - // Error occurred.
>> - promise->fail(os::strerror(errno));
>> - }
>> - } else {
>> - // TODO(benh): Retry if 'length' is 0?
>> - promise->set(length);
>> - }
>> - }
>> + if (!net::is_restartable_error(error) &&
>> + !net::is_retryable_error(error)) {
>> + // TODO(benh): Confirm that `os::strerror` does the right
>> + // thing for `error` on Windows.
>> + return Failure(os::strerror(error));
>> + }
>> +
>> + return None();
>> + }
>> +
>> + return length;
>> + },
>> + [=](const Option<size_t>& length) -> Future<ControlFlow<size_t>> {
>> + // Restart/retry if we don't yet have a result.
>> + if (length.isNone()) {
>> + return io::poll(fd, io::WRITE)
>> + .then([](short event) -> ControlFlow<size_t> {
>> + CHECK_EQ(io::WRITE, event);
>> + return Continue();
>> + });
>> + }
>> + return Break(length.get());
>> + });
>> }
>>
>> } // namespace internal {
>> @@ -177,32 +159,18 @@ Future<size_t> read(int fd, void* data, size_t size)
>> {
>> process::initialize();
>>
>> - std::shared_ptr<Promise<size_t>> promise(new Promise<size_t>());
>> -
>> // Check the file descriptor.
>> Try<bool> nonblock = os::isNonblock(fd);
>> if (nonblock.isError()) {
>> // The file descriptor is not valid (e.g., has been closed).
>> - promise->fail(
>> - "Failed to check if file descriptor was non-blocking: " +
>> - nonblock.error());
>> - return promise->future();
>> + return Failure("Failed to check if file descriptor was non-blocking:
>> " +
>> + nonblock.error());
>> } else if (!nonblock.get()) {
>> // The file descriptor is not non-blocking.
>> - promise->fail("Expected a non-blocking file descriptor");
>> - return promise->future();
>> + return Failure("Expected a non-blocking file descriptor");
>> }
>>
>> - // Because the file descriptor is non-blocking, we call read()
>> - // immediately. The read may in turn call poll if necessary,
>> - // avoiding unnecessary polling. We also observed that for some
>> - // combination of libev and Linux kernel versions, the poll would
>> - // block for non-deterministically long periods of time. This may be
>> - // fixed in a newer version of libev (we use 3.8 at the time of
>> - // writing this comment).
>> - internal::read(fd, data, size, internal::NONE, promise, io::READ);
>> -
>> - return promise->future();
>> + return internal::read(fd, data, size);
>> }
>>
>>
>> @@ -210,32 +178,19 @@ Future<size_t> write(int fd, const void* data,
>> size_t size)
>> {
>> process::initialize();
>>
>> - std::shared_ptr<Promise<size_t>> promise(new Promise<size_t>());
>> -
>> // Check the file descriptor.
>> Try<bool> nonblock = os::isNonblock(fd);
>> if (nonblock.isError()) {
>> // The file descriptor is not valid (e.g., has been closed).
>> - promise->fail(
>> + return Failure(
>> "Failed to check if file descriptor was non-blocking: " +
>> nonblock.error());
>> - return promise->future();
>> } else if (!nonblock.get()) {
>> // The file descriptor is not non-blocking.
>> - promise->fail("Expected a non-blocking file descriptor");
>> - return promise->future();
>> + return Failure("Expected a non-blocking file descriptor");
>> }
>>
>> - // Because the file descriptor is non-blocking, we call write()
>> - // immediately. The write may in turn call poll if necessary,
>> - // avoiding unnecessary polling. We also observed that for some
>> - // combination of libev and Linux kernel versions, the poll would
>> - // block for non-deterministically long periods of time. This may be
>> - // fixed in a newer version of libev (we use 3.8 at the time of
>> - // writing this comment).
>> - internal::write(fd, data, size, promise, io::WRITE);
>> -
>> - return promise->future();
>> + return internal::write(fd, data, size);
>> }
>>
>>
>> @@ -280,43 +235,15 @@ Future<size_t> peek(int fd, void* data, size_t
>> size, size_t limit)
>> nonblock.error());
>> }
>>
>> - std::shared_ptr<Promise<size_t>> promise(new Promise<size_t>());
>> -
>> - // Because the file descriptor is non-blocking, we call read()
>> - // immediately. The read may in turn call poll if necessary,
>> - // avoiding unnecessary polling. We also observed that for some
>> - // combination of libev and Linux kernel versions, the poll would
>> - // block for non-deterministically long periods of time. This may be
>> - // fixed in a newer version of libev (we use 3.8 at the time of
>> - // writing this comment).
>> - internal::read(fd, data, limit, internal::PEEK, promise, io::READ);
>> -
>> - // NOTE: We wrap `os::close` in a lambda to disambiguate on Windows.
>> - promise->future().onAny([fd]() { os::close(fd); });
>> -
>> - return promise->future();
>> + return internal::read(fd, data, limit, internal::PEEK)
>> + .onAny([fd]() {
>> + os::close(fd);
>> + });
>> }
>>
>>
>> namespace internal {
>>
>> -Future<string> _read(
>> - int fd,
>> - const std::shared_ptr<string>& buffer,
>> - const boost::shared_array<char>& data,
>> - size_t length)
>> -{
>> - return io::read(fd, data.get(), length)
>> - .then([=](size_t size) -> Future<string> {
>> - if (size == 0) { // EOF.
>> - return string(*buffer);
>> - }
>> - buffer->append(data.get(), size);
>> - return _read(fd, buffer, data, length);
>> - });
>> -}
>> -
>> -
>> Future<Nothing> splice(
>> int from,
>> int to,
>> @@ -392,9 +319,21 @@ Future<string> read(int fd)
>> std::shared_ptr<string> buffer(new string());
>> boost::shared_array<char> data(new char[BUFFERED_READ_SIZE]);
>>
>> - // NOTE: We wrap `os::close` in a lambda to disambiguate on Windows.
>> - return internal::_read(fd, buffer, data, BUFFERED_READ_SIZE)
>> - .onAny([fd]() { os::close(fd); });
>> + return loop(
>> + None(),
>> + [=]() {
>> + return io::read(fd, data.get(), BUFFERED_READ_SIZE);
>> + },
>> + [=](size_t length) -> ControlFlow<string> {
>> + if (length == 0) { // EOF.
>> + return Break(std::move(*buffer));
>> + }
>> + buffer->append(data.get(), length);
>> + return Continue();
>> + })
>> + .onAny([fd]() {
>> + os::close(fd);
>> + });
>> }
>>
>>
>>
>>
>
Re: [5/5] mesos git commit: Used `loop` in implementation of io::read
and io::write.
Posted by Benjamin Mahler <bm...@apache.org>.
Looks like there are some warnings for some of the loop changes:
../../../3rdparty/libprocess/src/io.cpp: In lambda function:
../../../3rdparty/libprocess/src/io.cpp:75:9: warning: control reaches end
of non-void function [-Wreturn-type]
}();
^
mv -f .deps/libprocess_la-io.Tpo .deps/libprocess_la-io.Plo
mv -f .deps/libprocess_la-metrics.Tpo .deps/libprocess_la-metrics.Plo
../../../3rdparty/libprocess/src/http.cpp: In lambda function:
../../../3rdparty/libprocess/src/http.cpp:1414:7: warning: control reaches
end of non-void function [-Wreturn-type]
},
^
../../../3rdparty/libprocess/src/http.cpp: In lambda function:
../../../3rdparty/libprocess/src/http.cpp:1646:13: warning: control reaches
end of non-void function [-Wreturn-type]
}()
^
On Sun, Jan 8, 2017 at 7:27 PM, <be...@apache.org> wrote:
> Used `loop` in implementation of io::read and io::write.
>
> Review: https://reviews.apache.org/r/54841
>
>
> Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
> Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/2d28c198
> Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/2d28c198
> Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/2d28c198
>
> Branch: refs/heads/master
> Commit: 2d28c198a5c09308825b2771483b70ac42839d16
> Parents: 9984449
> Author: Benjamin Hindman <be...@gmail.com>
> Authored: Mon Dec 5 10:48:56 2016 -0800
> Committer: Benjamin Hindman <be...@gmail.com>
> Committed: Sun Jan 8 19:27:02 2017 -0800
>
> ----------------------------------------------------------------------
> 3rdparty/libprocess/src/io.cpp | 289 ++++++++++++++----------------------
> 1 file changed, 114 insertions(+), 175 deletions(-)
> ----------------------------------------------------------------------
>
>
> http://git-wip-us.apache.org/repos/asf/mesos/blob/2d28c198/
> 3rdparty/libprocess/src/io.cpp
> ----------------------------------------------------------------------
> diff --git a/3rdparty/libprocess/src/io.cpp b/3rdparty/libprocess/src/io.
> cpp
> index d0b3ba1..8aa3576 100644
> --- a/3rdparty/libprocess/src/io.cpp
> +++ b/3rdparty/libprocess/src/io.cpp
> @@ -42,132 +42,114 @@ enum ReadFlags
> };
>
>
> -void read(
> - int fd,
> - void* data,
> - size_t size,
> - ReadFlags flags,
> - const std::shared_ptr<Promise<size_t>>& promise,
> - const Future<short>& future)
> +Future<size_t> read(int fd, void* data, size_t size, ReadFlags flags =
> NONE)
> {
> - // Ignore this function if the read operation has been discarded.
> - if (promise->future().hasDiscard()) {
> - CHECK(!future.isPending());
> - promise->discard();
> - return;
> - }
> -
> + // TODO(benh): Let the system calls do what ever they're supposed to
> + // rather than return 0 here?
> if (size == 0) {
> - promise->set(0);
> - return;
> + return 0;
> }
>
> - if (future.isDiscarded()) {
> - promise->fail("Failed to poll: discarded future");
> - } else if (future.isFailed()) {
> - promise->fail(future.failure());
> - } else {
> - ssize_t length;
> - if (flags == NONE) {
> - length = os::read(fd, data, size);
> - } else { // PEEK.
> - // In case 'fd' is not a socket ::recv() will fail with ENOTSOCK
> and the
> - // error will be propagted out.
> - // NOTE: We cast to `char*` here because the function prototypes on
> - // Windows use `char*` instead of `void*`.
> - length = net::recv(fd, (char*) data, size, MSG_PEEK);
> - }
> -
> + return loop(
> + None(),
> + [=]() -> Future<Option<size_t>> {
> + // Because the file descriptor is non-blocking, we call
> + // read()/recv() immediately. If no data is available than
> + // we'll call `poll` and block. We also observed that for some
> + // combination of libev and Linux kernel versions, the poll
> + // would block for non-deterministically long periods of
> + // time. This may be fixed in a newer version of libev (we use
> + // 3.8 at the time of writing this comment).
> + ssize_t length = [=]() {
> + switch (flags) {
> + case PEEK:
> + // In case `fd` is not a socket os::recv() will fail
> + // with ENOTSOCK and the error will be returned.
> + //
> + // NOTE: We cast to `char*` here because the function
> + // prototypes on Windows use `char*` instead of `void*`.
> + return net::recv(fd, (char*) data, size, MSG_PEEK);
> + case NONE:
> + return os::read(fd, data, size);
> + }
> + }();
> +
> + if (length < 0) {
> #ifdef __WINDOWS__
> - int error = WSAGetLastError();
> + int error = WSAGetLastError();
> #else
> - int error = errno;
> + int error = errno;
> #endif // __WINDOWS__
>
> - if (length < 0) {
> - if (net::is_restartable_error(error) ||
> net::is_retryable_error(error)) {
> - // Restart the read operation.
> - Future<short> future =
> - io::poll(fd, process::io::READ).onAny(
> - lambda::bind(&internal::read,
> - fd,
> - data,
> - size,
> - flags,
> - promise,
> - lambda::_1));
> -
> - // Stop polling if a discard occurs on our future.
> - promise->future().onDiscard(
> - lambda::bind(&process::internal::discard<short>,
> - WeakFuture<short>(future)));
> - } else {
> - // Error occurred.
> - promise->fail(os::strerror(errno));
> - }
> - } else {
> - promise->set(length);
> - }
> - }
> + if (!net::is_restartable_error(error) &&
> + !net::is_retryable_error(error)) {
> + // TODO(benh): Confirm that `os::strerror` does the right
> + // thing for `error` on Windows.
> + return Failure(os::strerror(error));
> + }
> +
> + return None();
> + }
> +
> + return length;
> + },
> + [=](const Option<size_t>& length) -> Future<ControlFlow<size_t>> {
> + // Restart/retry if we don't yet have a result.
> + if (length.isNone()) {
> + return io::poll(fd, io::READ)
> + .then([](short event) -> ControlFlow<size_t> {
> + CHECK_EQ(io::READ, event);
> + return Continue();
> + });
> + }
> + return Break(length.get());
> + });
> }
>
>
> -void write(
> - int fd,
> - const void* data,
> - size_t size,
> - const std::shared_ptr<Promise<size_t>>& promise,
> - const Future<short>& future)
> +Future<size_t> write(int fd, const void* data, size_t size)
> {
> - // Ignore this function if the write operation has been discarded.
> - if (promise->future().hasDiscard()) {
> - promise->discard();
> - return;
> - }
> -
> + // TODO(benh): Let the system calls do what ever they're supposed to
> + // rather than return 0 here?
> if (size == 0) {
> - promise->set(0);
> - return;
> + return 0;
> }
>
> - if (future.isDiscarded()) {
> - promise->fail("Failed to poll: discarded future");
> - } else if (future.isFailed()) {
> - promise->fail(future.failure());
> - } else {
> - ssize_t length = os::write(fd, data, size);
> + return loop(
> + None(),
> + [=]() -> Future<Option<size_t>> {
> + ssize_t length = os::write(fd, data, size);
>
> + if (length < 0) {
> #ifdef __WINDOWS__
> - int error = WSAGetLastError();
> + int error = WSAGetLastError();
> #else
> - int error = errno;
> + int error = errno;
> #endif // __WINDOWS__
>
> - if (length < 0) {
> - if (net::is_restartable_error(error) ||
> net::is_retryable_error(error)) {
> - // Restart the write operation.
> - Future<short> future =
> - io::poll(fd, process::io::WRITE).onAny(
> - lambda::bind(&internal::write,
> - fd,
> - data,
> - size,
> - promise,
> - lambda::_1));
> -
> - // Stop polling if a discard occurs on our future.
> - promise->future().onDiscard(
> - lambda::bind(&process::internal::discard<short>,
> - WeakFuture<short>(future)));
> - } else {
> - // Error occurred.
> - promise->fail(os::strerror(errno));
> - }
> - } else {
> - // TODO(benh): Retry if 'length' is 0?
> - promise->set(length);
> - }
> - }
> + if (!net::is_restartable_error(error) &&
> + !net::is_retryable_error(error)) {
> + // TODO(benh): Confirm that `os::strerror` does the right
> + // thing for `error` on Windows.
> + return Failure(os::strerror(error));
> + }
> +
> + return None();
> + }
> +
> + return length;
> + },
> + [=](const Option<size_t>& length) -> Future<ControlFlow<size_t>> {
> + // Restart/retry if we don't yet have a result.
> + if (length.isNone()) {
> + return io::poll(fd, io::WRITE)
> + .then([](short event) -> ControlFlow<size_t> {
> + CHECK_EQ(io::WRITE, event);
> + return Continue();
> + });
> + }
> + return Break(length.get());
> + });
> }
>
> } // namespace internal {
> @@ -177,32 +159,18 @@ Future<size_t> read(int fd, void* data, size_t size)
> {
> process::initialize();
>
> - std::shared_ptr<Promise<size_t>> promise(new Promise<size_t>());
> -
> // Check the file descriptor.
> Try<bool> nonblock = os::isNonblock(fd);
> if (nonblock.isError()) {
> // The file descriptor is not valid (e.g., has been closed).
> - promise->fail(
> - "Failed to check if file descriptor was non-blocking: " +
> - nonblock.error());
> - return promise->future();
> + return Failure("Failed to check if file descriptor was non-blocking:
> " +
> + nonblock.error());
> } else if (!nonblock.get()) {
> // The file descriptor is not non-blocking.
> - promise->fail("Expected a non-blocking file descriptor");
> - return promise->future();
> + return Failure("Expected a non-blocking file descriptor");
> }
>
> - // Because the file descriptor is non-blocking, we call read()
> - // immediately. The read may in turn call poll if necessary,
> - // avoiding unnecessary polling. We also observed that for some
> - // combination of libev and Linux kernel versions, the poll would
> - // block for non-deterministically long periods of time. This may be
> - // fixed in a newer version of libev (we use 3.8 at the time of
> - // writing this comment).
> - internal::read(fd, data, size, internal::NONE, promise, io::READ);
> -
> - return promise->future();
> + return internal::read(fd, data, size);
> }
>
>
> @@ -210,32 +178,19 @@ Future<size_t> write(int fd, const void* data,
> size_t size)
> {
> process::initialize();
>
> - std::shared_ptr<Promise<size_t>> promise(new Promise<size_t>());
> -
> // Check the file descriptor.
> Try<bool> nonblock = os::isNonblock(fd);
> if (nonblock.isError()) {
> // The file descriptor is not valid (e.g., has been closed).
> - promise->fail(
> + return Failure(
> "Failed to check if file descriptor was non-blocking: " +
> nonblock.error());
> - return promise->future();
> } else if (!nonblock.get()) {
> // The file descriptor is not non-blocking.
> - promise->fail("Expected a non-blocking file descriptor");
> - return promise->future();
> + return Failure("Expected a non-blocking file descriptor");
> }
>
> - // Because the file descriptor is non-blocking, we call write()
> - // immediately. The write may in turn call poll if necessary,
> - // avoiding unnecessary polling. We also observed that for some
> - // combination of libev and Linux kernel versions, the poll would
> - // block for non-deterministically long periods of time. This may be
> - // fixed in a newer version of libev (we use 3.8 at the time of
> - // writing this comment).
> - internal::write(fd, data, size, promise, io::WRITE);
> -
> - return promise->future();
> + return internal::write(fd, data, size);
> }
>
>
> @@ -280,43 +235,15 @@ Future<size_t> peek(int fd, void* data, size_t size,
> size_t limit)
> nonblock.error());
> }
>
> - std::shared_ptr<Promise<size_t>> promise(new Promise<size_t>());
> -
> - // Because the file descriptor is non-blocking, we call read()
> - // immediately. The read may in turn call poll if necessary,
> - // avoiding unnecessary polling. We also observed that for some
> - // combination of libev and Linux kernel versions, the poll would
> - // block for non-deterministically long periods of time. This may be
> - // fixed in a newer version of libev (we use 3.8 at the time of
> - // writing this comment).
> - internal::read(fd, data, limit, internal::PEEK, promise, io::READ);
> -
> - // NOTE: We wrap `os::close` in a lambda to disambiguate on Windows.
> - promise->future().onAny([fd]() { os::close(fd); });
> -
> - return promise->future();
> + return internal::read(fd, data, limit, internal::PEEK)
> + .onAny([fd]() {
> + os::close(fd);
> + });
> }
>
>
> namespace internal {
>
> -Future<string> _read(
> - int fd,
> - const std::shared_ptr<string>& buffer,
> - const boost::shared_array<char>& data,
> - size_t length)
> -{
> - return io::read(fd, data.get(), length)
> - .then([=](size_t size) -> Future<string> {
> - if (size == 0) { // EOF.
> - return string(*buffer);
> - }
> - buffer->append(data.get(), size);
> - return _read(fd, buffer, data, length);
> - });
> -}
> -
> -
> Future<Nothing> splice(
> int from,
> int to,
> @@ -392,9 +319,21 @@ Future<string> read(int fd)
> std::shared_ptr<string> buffer(new string());
> boost::shared_array<char> data(new char[BUFFERED_READ_SIZE]);
>
> - // NOTE: We wrap `os::close` in a lambda to disambiguate on Windows.
> - return internal::_read(fd, buffer, data, BUFFERED_READ_SIZE)
> - .onAny([fd]() { os::close(fd); });
> + return loop(
> + None(),
> + [=]() {
> + return io::read(fd, data.get(), BUFFERED_READ_SIZE);
> + },
> + [=](size_t length) -> ControlFlow<string> {
> + if (length == 0) { // EOF.
> + return Break(std::move(*buffer));
> + }
> + buffer->append(data.get(), length);
> + return Continue();
> + })
> + .onAny([fd]() {
> + os::close(fd);
> + });
> }
>
>
>
>
Re: [5/5] mesos git commit: Used `loop` in implementation of io::read
and io::write.
Posted by Benjamin Mahler <bm...@apache.org>.
Looks like there are some warnings for some of the loop changes:
../../../3rdparty/libprocess/src/io.cpp: In lambda function:
../../../3rdparty/libprocess/src/io.cpp:75:9: warning: control reaches end
of non-void function [-Wreturn-type]
}();
^
mv -f .deps/libprocess_la-io.Tpo .deps/libprocess_la-io.Plo
mv -f .deps/libprocess_la-metrics.Tpo .deps/libprocess_la-metrics.Plo
../../../3rdparty/libprocess/src/http.cpp: In lambda function:
../../../3rdparty/libprocess/src/http.cpp:1414:7: warning: control reaches
end of non-void function [-Wreturn-type]
},
^
../../../3rdparty/libprocess/src/http.cpp: In lambda function:
../../../3rdparty/libprocess/src/http.cpp:1646:13: warning: control reaches
end of non-void function [-Wreturn-type]
}()
^
On Sun, Jan 8, 2017 at 7:27 PM, <be...@apache.org> wrote:
> Used `loop` in implementation of io::read and io::write.
>
> Review: https://reviews.apache.org/r/54841
>
>
> Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
> Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/2d28c198
> Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/2d28c198
> Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/2d28c198
>
> Branch: refs/heads/master
> Commit: 2d28c198a5c09308825b2771483b70ac42839d16
> Parents: 9984449
> Author: Benjamin Hindman <be...@gmail.com>
> Authored: Mon Dec 5 10:48:56 2016 -0800
> Committer: Benjamin Hindman <be...@gmail.com>
> Committed: Sun Jan 8 19:27:02 2017 -0800
>
> ----------------------------------------------------------------------
> 3rdparty/libprocess/src/io.cpp | 289 ++++++++++++++----------------------
> 1 file changed, 114 insertions(+), 175 deletions(-)
> ----------------------------------------------------------------------
>
>
> http://git-wip-us.apache.org/repos/asf/mesos/blob/2d28c198/
> 3rdparty/libprocess/src/io.cpp
> ----------------------------------------------------------------------
> diff --git a/3rdparty/libprocess/src/io.cpp b/3rdparty/libprocess/src/io.
> cpp
> index d0b3ba1..8aa3576 100644
> --- a/3rdparty/libprocess/src/io.cpp
> +++ b/3rdparty/libprocess/src/io.cpp
> @@ -42,132 +42,114 @@ enum ReadFlags
> };
>
>
> -void read(
> - int fd,
> - void* data,
> - size_t size,
> - ReadFlags flags,
> - const std::shared_ptr<Promise<size_t>>& promise,
> - const Future<short>& future)
> +Future<size_t> read(int fd, void* data, size_t size, ReadFlags flags =
> NONE)
> {
> - // Ignore this function if the read operation has been discarded.
> - if (promise->future().hasDiscard()) {
> - CHECK(!future.isPending());
> - promise->discard();
> - return;
> - }
> -
> + // TODO(benh): Let the system calls do what ever they're supposed to
> + // rather than return 0 here?
> if (size == 0) {
> - promise->set(0);
> - return;
> + return 0;
> }
>
> - if (future.isDiscarded()) {
> - promise->fail("Failed to poll: discarded future");
> - } else if (future.isFailed()) {
> - promise->fail(future.failure());
> - } else {
> - ssize_t length;
> - if (flags == NONE) {
> - length = os::read(fd, data, size);
> - } else { // PEEK.
> - // In case 'fd' is not a socket ::recv() will fail with ENOTSOCK
> and the
> - // error will be propagted out.
> - // NOTE: We cast to `char*` here because the function prototypes on
> - // Windows use `char*` instead of `void*`.
> - length = net::recv(fd, (char*) data, size, MSG_PEEK);
> - }
> -
> + return loop(
> + None(),
> + [=]() -> Future<Option<size_t>> {
> + // Because the file descriptor is non-blocking, we call
> + // read()/recv() immediately. If no data is available than
> + // we'll call `poll` and block. We also observed that for some
> + // combination of libev and Linux kernel versions, the poll
> + // would block for non-deterministically long periods of
> + // time. This may be fixed in a newer version of libev (we use
> + // 3.8 at the time of writing this comment).
> + ssize_t length = [=]() {
> + switch (flags) {
> + case PEEK:
> + // In case `fd` is not a socket os::recv() will fail
> + // with ENOTSOCK and the error will be returned.
> + //
> + // NOTE: We cast to `char*` here because the function
> + // prototypes on Windows use `char*` instead of `void*`.
> + return net::recv(fd, (char*) data, size, MSG_PEEK);
> + case NONE:
> + return os::read(fd, data, size);
> + }
> + }();
> +
> + if (length < 0) {
> #ifdef __WINDOWS__
> - int error = WSAGetLastError();
> + int error = WSAGetLastError();
> #else
> - int error = errno;
> + int error = errno;
> #endif // __WINDOWS__
>
> - if (length < 0) {
> - if (net::is_restartable_error(error) ||
> net::is_retryable_error(error)) {
> - // Restart the read operation.
> - Future<short> future =
> - io::poll(fd, process::io::READ).onAny(
> - lambda::bind(&internal::read,
> - fd,
> - data,
> - size,
> - flags,
> - promise,
> - lambda::_1));
> -
> - // Stop polling if a discard occurs on our future.
> - promise->future().onDiscard(
> - lambda::bind(&process::internal::discard<short>,
> - WeakFuture<short>(future)));
> - } else {
> - // Error occurred.
> - promise->fail(os::strerror(errno));
> - }
> - } else {
> - promise->set(length);
> - }
> - }
> + if (!net::is_restartable_error(error) &&
> + !net::is_retryable_error(error)) {
> + // TODO(benh): Confirm that `os::strerror` does the right
> + // thing for `error` on Windows.
> + return Failure(os::strerror(error));
> + }
> +
> + return None();
> + }
> +
> + return length;
> + },
> + [=](const Option<size_t>& length) -> Future<ControlFlow<size_t>> {
> + // Restart/retry if we don't yet have a result.
> + if (length.isNone()) {
> + return io::poll(fd, io::READ)
> + .then([](short event) -> ControlFlow<size_t> {
> + CHECK_EQ(io::READ, event);
> + return Continue();
> + });
> + }
> + return Break(length.get());
> + });
> }
>
>
> -void write(
> - int fd,
> - const void* data,
> - size_t size,
> - const std::shared_ptr<Promise<size_t>>& promise,
> - const Future<short>& future)
> +Future<size_t> write(int fd, const void* data, size_t size)
> {
> - // Ignore this function if the write operation has been discarded.
> - if (promise->future().hasDiscard()) {
> - promise->discard();
> - return;
> - }
> -
> + // TODO(benh): Let the system calls do what ever they're supposed to
> + // rather than return 0 here?
> if (size == 0) {
> - promise->set(0);
> - return;
> + return 0;
> }
>
> - if (future.isDiscarded()) {
> - promise->fail("Failed to poll: discarded future");
> - } else if (future.isFailed()) {
> - promise->fail(future.failure());
> - } else {
> - ssize_t length = os::write(fd, data, size);
> + return loop(
> + None(),
> + [=]() -> Future<Option<size_t>> {
> + ssize_t length = os::write(fd, data, size);
>
> + if (length < 0) {
> #ifdef __WINDOWS__
> - int error = WSAGetLastError();
> + int error = WSAGetLastError();
> #else
> - int error = errno;
> + int error = errno;
> #endif // __WINDOWS__
>
> - if (length < 0) {
> - if (net::is_restartable_error(error) ||
> net::is_retryable_error(error)) {
> - // Restart the write operation.
> - Future<short> future =
> - io::poll(fd, process::io::WRITE).onAny(
> - lambda::bind(&internal::write,
> - fd,
> - data,
> - size,
> - promise,
> - lambda::_1));
> -
> - // Stop polling if a discard occurs on our future.
> - promise->future().onDiscard(
> - lambda::bind(&process::internal::discard<short>,
> - WeakFuture<short>(future)));
> - } else {
> - // Error occurred.
> - promise->fail(os::strerror(errno));
> - }
> - } else {
> - // TODO(benh): Retry if 'length' is 0?
> - promise->set(length);
> - }
> - }
> + if (!net::is_restartable_error(error) &&
> + !net::is_retryable_error(error)) {
> + // TODO(benh): Confirm that `os::strerror` does the right
> + // thing for `error` on Windows.
> + return Failure(os::strerror(error));
> + }
> +
> + return None();
> + }
> +
> + return length;
> + },
> + [=](const Option<size_t>& length) -> Future<ControlFlow<size_t>> {
> + // Restart/retry if we don't yet have a result.
> + if (length.isNone()) {
> + return io::poll(fd, io::WRITE)
> + .then([](short event) -> ControlFlow<size_t> {
> + CHECK_EQ(io::WRITE, event);
> + return Continue();
> + });
> + }
> + return Break(length.get());
> + });
> }
>
> } // namespace internal {
> @@ -177,32 +159,18 @@ Future<size_t> read(int fd, void* data, size_t size)
> {
> process::initialize();
>
> - std::shared_ptr<Promise<size_t>> promise(new Promise<size_t>());
> -
> // Check the file descriptor.
> Try<bool> nonblock = os::isNonblock(fd);
> if (nonblock.isError()) {
> // The file descriptor is not valid (e.g., has been closed).
> - promise->fail(
> - "Failed to check if file descriptor was non-blocking: " +
> - nonblock.error());
> - return promise->future();
> + return Failure("Failed to check if file descriptor was non-blocking:
> " +
> + nonblock.error());
> } else if (!nonblock.get()) {
> // The file descriptor is not non-blocking.
> - promise->fail("Expected a non-blocking file descriptor");
> - return promise->future();
> + return Failure("Expected a non-blocking file descriptor");
> }
>
> - // Because the file descriptor is non-blocking, we call read()
> - // immediately. The read may in turn call poll if necessary,
> - // avoiding unnecessary polling. We also observed that for some
> - // combination of libev and Linux kernel versions, the poll would
> - // block for non-deterministically long periods of time. This may be
> - // fixed in a newer version of libev (we use 3.8 at the time of
> - // writing this comment).
> - internal::read(fd, data, size, internal::NONE, promise, io::READ);
> -
> - return promise->future();
> + return internal::read(fd, data, size);
> }
>
>
> @@ -210,32 +178,19 @@ Future<size_t> write(int fd, const void* data,
> size_t size)
> {
> process::initialize();
>
> - std::shared_ptr<Promise<size_t>> promise(new Promise<size_t>());
> -
> // Check the file descriptor.
> Try<bool> nonblock = os::isNonblock(fd);
> if (nonblock.isError()) {
> // The file descriptor is not valid (e.g., has been closed).
> - promise->fail(
> + return Failure(
> "Failed to check if file descriptor was non-blocking: " +
> nonblock.error());
> - return promise->future();
> } else if (!nonblock.get()) {
> // The file descriptor is not non-blocking.
> - promise->fail("Expected a non-blocking file descriptor");
> - return promise->future();
> + return Failure("Expected a non-blocking file descriptor");
> }
>
> - // Because the file descriptor is non-blocking, we call write()
> - // immediately. The write may in turn call poll if necessary,
> - // avoiding unnecessary polling. We also observed that for some
> - // combination of libev and Linux kernel versions, the poll would
> - // block for non-deterministically long periods of time. This may be
> - // fixed in a newer version of libev (we use 3.8 at the time of
> - // writing this comment).
> - internal::write(fd, data, size, promise, io::WRITE);
> -
> - return promise->future();
> + return internal::write(fd, data, size);
> }
>
>
> @@ -280,43 +235,15 @@ Future<size_t> peek(int fd, void* data, size_t size,
> size_t limit)
> nonblock.error());
> }
>
> - std::shared_ptr<Promise<size_t>> promise(new Promise<size_t>());
> -
> - // Because the file descriptor is non-blocking, we call read()
> - // immediately. The read may in turn call poll if necessary,
> - // avoiding unnecessary polling. We also observed that for some
> - // combination of libev and Linux kernel versions, the poll would
> - // block for non-deterministically long periods of time. This may be
> - // fixed in a newer version of libev (we use 3.8 at the time of
> - // writing this comment).
> - internal::read(fd, data, limit, internal::PEEK, promise, io::READ);
> -
> - // NOTE: We wrap `os::close` in a lambda to disambiguate on Windows.
> - promise->future().onAny([fd]() { os::close(fd); });
> -
> - return promise->future();
> + return internal::read(fd, data, limit, internal::PEEK)
> + .onAny([fd]() {
> + os::close(fd);
> + });
> }
>
>
> namespace internal {
>
> -Future<string> _read(
> - int fd,
> - const std::shared_ptr<string>& buffer,
> - const boost::shared_array<char>& data,
> - size_t length)
> -{
> - return io::read(fd, data.get(), length)
> - .then([=](size_t size) -> Future<string> {
> - if (size == 0) { // EOF.
> - return string(*buffer);
> - }
> - buffer->append(data.get(), size);
> - return _read(fd, buffer, data, length);
> - });
> -}
> -
> -
> Future<Nothing> splice(
> int from,
> int to,
> @@ -392,9 +319,21 @@ Future<string> read(int fd)
> std::shared_ptr<string> buffer(new string());
> boost::shared_array<char> data(new char[BUFFERED_READ_SIZE]);
>
> - // NOTE: We wrap `os::close` in a lambda to disambiguate on Windows.
> - return internal::_read(fd, buffer, data, BUFFERED_READ_SIZE)
> - .onAny([fd]() { os::close(fd); });
> + return loop(
> + None(),
> + [=]() {
> + return io::read(fd, data.get(), BUFFERED_READ_SIZE);
> + },
> + [=](size_t length) -> ControlFlow<string> {
> + if (length == 0) { // EOF.
> + return Break(std::move(*buffer));
> + }
> + buffer->append(data.get(), length);
> + return Continue();
> + })
> + .onAny([fd]() {
> + os::close(fd);
> + });
> }
>
>
>
>
[5/5] mesos git commit: Used `loop` in implementation of io::read and
io::write.
Posted by be...@apache.org.
Used `loop` in implementation of io::read and io::write.
Review: https://reviews.apache.org/r/54841
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/2d28c198
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/2d28c198
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/2d28c198
Branch: refs/heads/master
Commit: 2d28c198a5c09308825b2771483b70ac42839d16
Parents: 9984449
Author: Benjamin Hindman <be...@gmail.com>
Authored: Mon Dec 5 10:48:56 2016 -0800
Committer: Benjamin Hindman <be...@gmail.com>
Committed: Sun Jan 8 19:27:02 2017 -0800
----------------------------------------------------------------------
3rdparty/libprocess/src/io.cpp | 289 ++++++++++++++----------------------
1 file changed, 114 insertions(+), 175 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/2d28c198/3rdparty/libprocess/src/io.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/io.cpp b/3rdparty/libprocess/src/io.cpp
index d0b3ba1..8aa3576 100644
--- a/3rdparty/libprocess/src/io.cpp
+++ b/3rdparty/libprocess/src/io.cpp
@@ -42,132 +42,114 @@ enum ReadFlags
};
-void read(
- int fd,
- void* data,
- size_t size,
- ReadFlags flags,
- const std::shared_ptr<Promise<size_t>>& promise,
- const Future<short>& future)
+Future<size_t> read(int fd, void* data, size_t size, ReadFlags flags = NONE)
{
- // Ignore this function if the read operation has been discarded.
- if (promise->future().hasDiscard()) {
- CHECK(!future.isPending());
- promise->discard();
- return;
- }
-
+ // TODO(benh): Let the system calls do what ever they're supposed to
+ // rather than return 0 here?
if (size == 0) {
- promise->set(0);
- return;
+ return 0;
}
- if (future.isDiscarded()) {
- promise->fail("Failed to poll: discarded future");
- } else if (future.isFailed()) {
- promise->fail(future.failure());
- } else {
- ssize_t length;
- if (flags == NONE) {
- length = os::read(fd, data, size);
- } else { // PEEK.
- // In case 'fd' is not a socket ::recv() will fail with ENOTSOCK and the
- // error will be propagted out.
- // NOTE: We cast to `char*` here because the function prototypes on
- // Windows use `char*` instead of `void*`.
- length = net::recv(fd, (char*) data, size, MSG_PEEK);
- }
-
+ return loop(
+ None(),
+ [=]() -> Future<Option<size_t>> {
+ // Because the file descriptor is non-blocking, we call
+ // read()/recv() immediately. If no data is available than
+ // we'll call `poll` and block. We also observed that for some
+ // combination of libev and Linux kernel versions, the poll
+ // would block for non-deterministically long periods of
+ // time. This may be fixed in a newer version of libev (we use
+ // 3.8 at the time of writing this comment).
+ ssize_t length = [=]() {
+ switch (flags) {
+ case PEEK:
+ // In case `fd` is not a socket os::recv() will fail
+ // with ENOTSOCK and the error will be returned.
+ //
+ // NOTE: We cast to `char*` here because the function
+ // prototypes on Windows use `char*` instead of `void*`.
+ return net::recv(fd, (char*) data, size, MSG_PEEK);
+ case NONE:
+ return os::read(fd, data, size);
+ }
+ }();
+
+ if (length < 0) {
#ifdef __WINDOWS__
- int error = WSAGetLastError();
+ int error = WSAGetLastError();
#else
- int error = errno;
+ int error = errno;
#endif // __WINDOWS__
- if (length < 0) {
- if (net::is_restartable_error(error) || net::is_retryable_error(error)) {
- // Restart the read operation.
- Future<short> future =
- io::poll(fd, process::io::READ).onAny(
- lambda::bind(&internal::read,
- fd,
- data,
- size,
- flags,
- promise,
- lambda::_1));
-
- // Stop polling if a discard occurs on our future.
- promise->future().onDiscard(
- lambda::bind(&process::internal::discard<short>,
- WeakFuture<short>(future)));
- } else {
- // Error occurred.
- promise->fail(os::strerror(errno));
- }
- } else {
- promise->set(length);
- }
- }
+ if (!net::is_restartable_error(error) &&
+ !net::is_retryable_error(error)) {
+ // TODO(benh): Confirm that `os::strerror` does the right
+ // thing for `error` on Windows.
+ return Failure(os::strerror(error));
+ }
+
+ return None();
+ }
+
+ return length;
+ },
+ [=](const Option<size_t>& length) -> Future<ControlFlow<size_t>> {
+ // Restart/retry if we don't yet have a result.
+ if (length.isNone()) {
+ return io::poll(fd, io::READ)
+ .then([](short event) -> ControlFlow<size_t> {
+ CHECK_EQ(io::READ, event);
+ return Continue();
+ });
+ }
+ return Break(length.get());
+ });
}
-void write(
- int fd,
- const void* data,
- size_t size,
- const std::shared_ptr<Promise<size_t>>& promise,
- const Future<short>& future)
+Future<size_t> write(int fd, const void* data, size_t size)
{
- // Ignore this function if the write operation has been discarded.
- if (promise->future().hasDiscard()) {
- promise->discard();
- return;
- }
-
+ // TODO(benh): Let the system calls do what ever they're supposed to
+ // rather than return 0 here?
if (size == 0) {
- promise->set(0);
- return;
+ return 0;
}
- if (future.isDiscarded()) {
- promise->fail("Failed to poll: discarded future");
- } else if (future.isFailed()) {
- promise->fail(future.failure());
- } else {
- ssize_t length = os::write(fd, data, size);
+ return loop(
+ None(),
+ [=]() -> Future<Option<size_t>> {
+ ssize_t length = os::write(fd, data, size);
+ if (length < 0) {
#ifdef __WINDOWS__
- int error = WSAGetLastError();
+ int error = WSAGetLastError();
#else
- int error = errno;
+ int error = errno;
#endif // __WINDOWS__
- if (length < 0) {
- if (net::is_restartable_error(error) || net::is_retryable_error(error)) {
- // Restart the write operation.
- Future<short> future =
- io::poll(fd, process::io::WRITE).onAny(
- lambda::bind(&internal::write,
- fd,
- data,
- size,
- promise,
- lambda::_1));
-
- // Stop polling if a discard occurs on our future.
- promise->future().onDiscard(
- lambda::bind(&process::internal::discard<short>,
- WeakFuture<short>(future)));
- } else {
- // Error occurred.
- promise->fail(os::strerror(errno));
- }
- } else {
- // TODO(benh): Retry if 'length' is 0?
- promise->set(length);
- }
- }
+ if (!net::is_restartable_error(error) &&
+ !net::is_retryable_error(error)) {
+ // TODO(benh): Confirm that `os::strerror` does the right
+ // thing for `error` on Windows.
+ return Failure(os::strerror(error));
+ }
+
+ return None();
+ }
+
+ return length;
+ },
+ [=](const Option<size_t>& length) -> Future<ControlFlow<size_t>> {
+ // Restart/retry if we don't yet have a result.
+ if (length.isNone()) {
+ return io::poll(fd, io::WRITE)
+ .then([](short event) -> ControlFlow<size_t> {
+ CHECK_EQ(io::WRITE, event);
+ return Continue();
+ });
+ }
+ return Break(length.get());
+ });
}
} // namespace internal {
@@ -177,32 +159,18 @@ Future<size_t> read(int fd, void* data, size_t size)
{
process::initialize();
- std::shared_ptr<Promise<size_t>> promise(new Promise<size_t>());
-
// Check the file descriptor.
Try<bool> nonblock = os::isNonblock(fd);
if (nonblock.isError()) {
// The file descriptor is not valid (e.g., has been closed).
- promise->fail(
- "Failed to check if file descriptor was non-blocking: " +
- nonblock.error());
- return promise->future();
+ return Failure("Failed to check if file descriptor was non-blocking: " +
+ nonblock.error());
} else if (!nonblock.get()) {
// The file descriptor is not non-blocking.
- promise->fail("Expected a non-blocking file descriptor");
- return promise->future();
+ return Failure("Expected a non-blocking file descriptor");
}
- // Because the file descriptor is non-blocking, we call read()
- // immediately. The read may in turn call poll if necessary,
- // avoiding unnecessary polling. We also observed that for some
- // combination of libev and Linux kernel versions, the poll would
- // block for non-deterministically long periods of time. This may be
- // fixed in a newer version of libev (we use 3.8 at the time of
- // writing this comment).
- internal::read(fd, data, size, internal::NONE, promise, io::READ);
-
- return promise->future();
+ return internal::read(fd, data, size);
}
@@ -210,32 +178,19 @@ Future<size_t> write(int fd, const void* data, size_t size)
{
process::initialize();
- std::shared_ptr<Promise<size_t>> promise(new Promise<size_t>());
-
// Check the file descriptor.
Try<bool> nonblock = os::isNonblock(fd);
if (nonblock.isError()) {
// The file descriptor is not valid (e.g., has been closed).
- promise->fail(
+ return Failure(
"Failed to check if file descriptor was non-blocking: " +
nonblock.error());
- return promise->future();
} else if (!nonblock.get()) {
// The file descriptor is not non-blocking.
- promise->fail("Expected a non-blocking file descriptor");
- return promise->future();
+ return Failure("Expected a non-blocking file descriptor");
}
- // Because the file descriptor is non-blocking, we call write()
- // immediately. The write may in turn call poll if necessary,
- // avoiding unnecessary polling. We also observed that for some
- // combination of libev and Linux kernel versions, the poll would
- // block for non-deterministically long periods of time. This may be
- // fixed in a newer version of libev (we use 3.8 at the time of
- // writing this comment).
- internal::write(fd, data, size, promise, io::WRITE);
-
- return promise->future();
+ return internal::write(fd, data, size);
}
@@ -280,43 +235,15 @@ Future<size_t> peek(int fd, void* data, size_t size, size_t limit)
nonblock.error());
}
- std::shared_ptr<Promise<size_t>> promise(new Promise<size_t>());
-
- // Because the file descriptor is non-blocking, we call read()
- // immediately. The read may in turn call poll if necessary,
- // avoiding unnecessary polling. We also observed that for some
- // combination of libev and Linux kernel versions, the poll would
- // block for non-deterministically long periods of time. This may be
- // fixed in a newer version of libev (we use 3.8 at the time of
- // writing this comment).
- internal::read(fd, data, limit, internal::PEEK, promise, io::READ);
-
- // NOTE: We wrap `os::close` in a lambda to disambiguate on Windows.
- promise->future().onAny([fd]() { os::close(fd); });
-
- return promise->future();
+ return internal::read(fd, data, limit, internal::PEEK)
+ .onAny([fd]() {
+ os::close(fd);
+ });
}
namespace internal {
-Future<string> _read(
- int fd,
- const std::shared_ptr<string>& buffer,
- const boost::shared_array<char>& data,
- size_t length)
-{
- return io::read(fd, data.get(), length)
- .then([=](size_t size) -> Future<string> {
- if (size == 0) { // EOF.
- return string(*buffer);
- }
- buffer->append(data.get(), size);
- return _read(fd, buffer, data, length);
- });
-}
-
-
Future<Nothing> splice(
int from,
int to,
@@ -392,9 +319,21 @@ Future<string> read(int fd)
std::shared_ptr<string> buffer(new string());
boost::shared_array<char> data(new char[BUFFERED_READ_SIZE]);
- // NOTE: We wrap `os::close` in a lambda to disambiguate on Windows.
- return internal::_read(fd, buffer, data, BUFFERED_READ_SIZE)
- .onAny([fd]() { os::close(fd); });
+ return loop(
+ None(),
+ [=]() {
+ return io::read(fd, data.get(), BUFFERED_READ_SIZE);
+ },
+ [=](size_t length) -> ControlFlow<string> {
+ if (length == 0) { // EOF.
+ return Break(std::move(*buffer));
+ }
+ buffer->append(data.get(), length);
+ return Continue();
+ })
+ .onAny([fd]() {
+ os::close(fd);
+ });
}
[2/5] mesos git commit: Updated Mesos process::loop uses with
process::ControlFlow.
Posted by be...@apache.org.
Updated Mesos process::loop uses with process::ControlFlow.
Review: https://reviews.apache.org/r/54839
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/ab2565c1
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/ab2565c1
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/ab2565c1
Branch: refs/heads/master
Commit: ab2565c18dfac57440d6d296eff814a0076cdc8c
Parents: bbb4058
Author: Benjamin Hindman <be...@gmail.com>
Authored: Sat Dec 17 13:33:15 2016 -0800
Committer: Benjamin Hindman <be...@gmail.com>
Committed: Sat Jan 7 23:22:24 2017 -0800
----------------------------------------------------------------------
src/common/recordio.hpp | 7 +-
.../containerizer/mesos/io/switchboard.cpp | 73 ++++++++++----------
src/slave/http.cpp | 9 ++-
3 files changed, 45 insertions(+), 44 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/ab2565c1/src/common/recordio.hpp
----------------------------------------------------------------------
diff --git a/src/common/recordio.hpp b/src/common/recordio.hpp
index 5a22d06..3783634 100644
--- a/src/common/recordio.hpp
+++ b/src/common/recordio.hpp
@@ -119,10 +119,11 @@ process::Future<Nothing> transform(
[=]() {
return reader->read();
},
- [=](const Result<T>& record) mutable -> process::Future<bool> {
+ [=](const Result<T>& record) mutable
+ -> process::Future<process::ControlFlow<Nothing>> {
// This could happen if EOF is sent by the writer.
if (record.isNone()) {
- return false;
+ return process::Break();
}
// This could happen if there is a de-serialization error.
@@ -136,7 +137,7 @@ process::Future<Nothing> transform(
return process::Failure("Write failed to the pipe");
}
- return true;
+ return process::Continue();
});
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/ab2565c1/src/slave/containerizer/mesos/io/switchboard.cpp
----------------------------------------------------------------------
diff --git a/src/slave/containerizer/mesos/io/switchboard.cpp b/src/slave/containerizer/mesos/io/switchboard.cpp
index 6d2e9fd..67c5167 100644
--- a/src/slave/containerizer/mesos/io/switchboard.cpp
+++ b/src/slave/containerizer/mesos/io/switchboard.cpp
@@ -79,6 +79,9 @@ using std::string;
using std::vector;
using process::await;
+using process::Break;
+using process::Continue;
+using process::ControlFlow;
using process::Clock;
using process::ErrnoFailure;
using process::Failure;
@@ -683,8 +686,11 @@ Future<http::Connection> IOSwitchboard::connect(
[=]() {
return limiter->acquire();
},
- [=](const Nothing&) {
- return infos.contains(containerId) && !os::exists(address->path());
+ [=](const Nothing&) -> ControlFlow<Nothing> {
+ if (infos.contains(containerId) && !os::exists(address->path())) {
+ return Continue();
+ }
+ return Break();
})
.then(defer(self(), [=]() -> Future<http::Connection> {
if (!infos.contains(containerId)) {
@@ -1482,22 +1488,19 @@ Future<http::Response> IOSwitchboardServerProcess::attachContainerInput(
// Loop through each record and process it. Return a proper
// response once the last record has been fully processed.
- Owned<http::Response> response(new http::Response());
-
return loop(
self(),
[=]() {
return reader->read();
},
- [=](const Result<agent::Call>& record) -> Future<bool> {
+ [=](const Result<agent::Call>& record)
+ -> Future<ControlFlow<http::Response>> {
if (record.isNone()) {
- *response = http::OK();
- return false;
+ return Break(http::OK());
}
if (record.isError()) {
- *response = http::BadRequest(record.error());
- return false;
+ return Break(http::BadRequest(record.error()));
}
// Should have already been validated by the agent.
@@ -1508,8 +1511,7 @@ Future<http::Response> IOSwitchboardServerProcess::attachContainerInput(
// Validate the rest of the `AttachContainerInput` message.
Option<Error> error = validate(record->attach_container_input());
if (error.isSome()) {
- *response = http::BadRequest(error->message);
- return false;
+ return Break(http::BadRequest(error->message));
}
const agent::ProcessIO& message =
@@ -1529,17 +1531,16 @@ Future<http::Response> IOSwitchboardServerProcess::attachContainerInput(
message.control().tty_info().window_size().columns());
if (window.isError()) {
- *response = http::BadRequest(
- "Unable to set the window size: " + window.error());
- return false;
+ return Break(http::BadRequest(
+ "Unable to set the window size: " + window.error()));
}
- return true;
+ return Continue();
}
case agent::ProcessIO::Control::HEARTBEAT: {
// For now, we ignore any interval information
// sent along with the heartbeat.
- return true;
+ return Continue();
}
default: {
UNREACHABLE();
@@ -1552,7 +1553,7 @@ Future<http::Response> IOSwitchboardServerProcess::attachContainerInput(
// If tty is enabled, the client is expected to send `EOT` instead.
if (!tty && message.data().data().length() == 0) {
os::close(stdinToFd);
- return true;
+ return Continue();
}
// Write the STDIN data to `stdinToFd`. If there is a
@@ -1561,34 +1562,30 @@ Future<http::Response> IOSwitchboardServerProcess::attachContainerInput(
// terminate the process. We don't terminate the process
// here because we want to propagate an `InternalServerError`
// back to the client.
- Owned<Promise<bool>> condition(new Promise<bool>());
-
- process::io::write(stdinToFd, message.data().data())
- .onAny(defer(self(), [this, condition, response](
- const Future<Nothing>& future) {
- if (future.isReady()) {
- condition->set(true);
- return;
- }
-
+ return process::io::write(stdinToFd, message.data().data())
+ .then(defer(self(), [=](const Nothing&)
+ -> ControlFlow<http::Response> {
+ return Continue();
+ }))
+ // TODO(benh):
+ // .recover(defer(self(), [=](...) {
+ // failure = Failure("Failed writing to stdin: discarded");
+ // return Break(http::InternalServerError(failure->message));
+ // }))
+ .repair(defer(self(), [=](
+ const Future<ControlFlow<http::Response>>& future)
+ -> ControlFlow<http::Response> {
failure = Failure(
- "Failed writing to stdin:"
- " " + (future.isFailed() ? future.failure() : "discarded"));
-
- *response = http::InternalServerError(failure->message);
-
- condition->set(false);
+ "Failed writing to stdin: " + future.failure());
+ return Break(http::InternalServerError(failure->message));
}));
-
- return condition->future();
}
default: {
UNREACHABLE();
}
}
})
- .then(defer(self(), [this, response](
- const Future<Nothing> future) -> http::Response {
+ .then(defer(self(), [=](const http::Response& response) {
// Reset `inputConnected` to allow future input connections.
inputConnected = false;
@@ -1599,7 +1596,7 @@ Future<http::Response> IOSwitchboardServerProcess::attachContainerInput(
terminate(self(), false);
}
- return *response;
+ return response;
}));
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/ab2565c1/src/slave/http.cpp
----------------------------------------------------------------------
diff --git a/src/slave/http.cpp b/src/slave/http.cpp
index b6e2d4f..24fc23b 100644
--- a/src/slave/http.cpp
+++ b/src/slave/http.cpp
@@ -80,6 +80,9 @@ using mesos::slave::ContainerTermination;
using process::AUTHENTICATION;
using process::AUTHORIZATION;
+using process::Break;
+using process::Continue;
+using process::ControlFlow;
using process::Clock;
using process::DESCRIPTION;
using process::Failure;
@@ -2587,17 +2590,17 @@ Future<Nothing> connect(Pipe::Reader reader, Pipe::Writer writer)
[=]() mutable {
return reader.read();
},
- [=](const string& chunk) mutable -> Future<bool> {
+ [=](const string& chunk) mutable -> Future<ControlFlow<Nothing>> {
if (chunk.empty()) {
// EOF case.
- return false;
+ return Break();
}
if (!writer.write(chunk)) {
return Failure("Write failed to the pipe");
}
- return true;
+ return Continue();
});
}
[3/5] mesos git commit: Used process::io::BUFFERED_READ_SIZE instead
of constant.
Posted by be...@apache.org.
Used process::io::BUFFERED_READ_SIZE instead of constant.
Review: https://reviews.apache.org/r/54838
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/63fe4b03
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/63fe4b03
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/63fe4b03
Branch: refs/heads/master
Commit: 63fe4b0362902f00773b1e5655a566a95e29267b
Parents: 9942fa7
Author: Benjamin Hindman <be...@gmail.com>
Authored: Sat Dec 17 13:32:11 2016 -0800
Committer: Benjamin Hindman <be...@gmail.com>
Committed: Sat Jan 7 23:22:24 2017 -0800
----------------------------------------------------------------------
src/slave/containerizer/mesos/io/switchboard.cpp | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/63fe4b03/src/slave/containerizer/mesos/io/switchboard.cpp
----------------------------------------------------------------------
diff --git a/src/slave/containerizer/mesos/io/switchboard.cpp b/src/slave/containerizer/mesos/io/switchboard.cpp
index c806279..6d2e9fd 100644
--- a/src/slave/containerizer/mesos/io/switchboard.cpp
+++ b/src/slave/containerizer/mesos/io/switchboard.cpp
@@ -1111,7 +1111,7 @@ Future<Nothing> IOSwitchboardServerProcess::run()
Future<Nothing> stdoutRedirect = process::io::redirect(
stdoutFromFd,
stdoutToFd,
- 4096,
+ process::io::BUFFERED_READ_SIZE,
{defer(self(),
&Self::outputHook,
lambda::_1,
@@ -1129,7 +1129,7 @@ Future<Nothing> IOSwitchboardServerProcess::run()
stderrRedirect = process::io::redirect(
stderrFromFd,
stderrToFd,
- 4096,
+ process::io::BUFFERED_READ_SIZE,
{defer(self(),
&Self::outputHook,
lambda::_1,
[4/5] mesos git commit: Introduced ControlFlow for process::loop.
Posted by be...@apache.org.
Introduced ControlFlow for process::loop.
Review: https://reviews.apache.org/r/54358
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/bbb4058d
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/bbb4058d
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/bbb4058d
Branch: refs/heads/master
Commit: bbb4058d60b50b54bcc626c25285c993ae4d8a3e
Parents: 63fe4b0
Author: Benjamin Hindman <be...@gmail.com>
Authored: Sun Dec 4 18:42:58 2016 -0800
Committer: Benjamin Hindman <be...@gmail.com>
Committed: Sat Jan 7 23:22:24 2017 -0800
----------------------------------------------------------------------
3rdparty/libprocess/include/process/loop.hpp | 214 ++++++++++++++++++----
3rdparty/libprocess/src/http.cpp | 23 ++-
3rdparty/libprocess/src/io.cpp | 21 ++-
3rdparty/libprocess/src/tests/loop_tests.cpp | 39 ++--
4 files changed, 233 insertions(+), 64 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/bbb4058d/3rdparty/libprocess/include/process/loop.hpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/include/process/loop.hpp b/3rdparty/libprocess/include/process/loop.hpp
index b35f7e6..53f6243 100644
--- a/3rdparty/libprocess/include/process/loop.hpp
+++ b/3rdparty/libprocess/include/process/loop.hpp
@@ -91,21 +91,48 @@ namespace process {
// And now what this looks like using `loop`:
//
// loop(pid,
-// []() { return iterate(); },
+// []() {
+// return iterate();
+// },
// [](T t) {
// return body(t);
// });
-template <typename Iterate, typename Body>
-Future<Nothing> loop(const Option<UPID>& pid, Iterate&& iterate, Body&& body);
+//
+// One difference between the `loop` version of the "body" versus the
+// other non-loop examples above is the return value is not `bool` or
+// `Future<bool>` but rather `ControlFlow<V>` or
+// `Future<ControlFlow<V>>`. This enables you to return values out of
+// the loop via a `Break(...)`, for example:
+//
+// loop(pid,
+// []() {
+// return iterate();
+// },
+// [](T t) {
+// if (finished(t)) {
+// return Break(SomeValue());
+// }
+// return Continue();
+// });
+template <typename Iterate,
+ typename Body,
+ typename T = typename internal::unwrap<typename result_of<Iterate()>::type>::type, // NOLINT(whitespace/line_length)
+ typename CF = typename internal::unwrap<typename result_of<Body(T)>::type>::type, // NOLINT(whitespace/line_length)
+ typename V = typename CF::ValueType>
+Future<V> loop(const Option<UPID>& pid, Iterate&& iterate, Body&& body);
// A helper for `loop` which creates a Process for us to provide an
// execution context for running the loop.
-template <typename Iterate, typename Body>
-Future<Nothing> loop(Iterate&& iterate, Body&& body)
+template <typename Iterate,
+ typename Body,
+ typename T = typename internal::unwrap<typename result_of<Iterate()>::type>::type, // NOLINT(whitespace/line_length)
+ typename CF = typename internal::unwrap<typename result_of<Body(T)>::type>::type, // NOLINT(whitespace/line_length)
+ typename V = typename CF::ValueType>
+Future<V> loop(Iterate&& iterate, Body&& body)
{
ProcessBase* process = new ProcessBase();
- return loop(
+ return loop<Iterate, Body, T, CF, V>(
spawn(process, true), // Have libprocess free `process`.
std::forward<Iterate>(iterate),
std::forward<Body>(body))
@@ -117,10 +144,117 @@ Future<Nothing> loop(Iterate&& iterate, Body&& body)
}
+// Generic "control flow" construct that is leveraged by
+// implementations such as `loop`. At a high-level a `ControlFlow`
+// represents some control flow statement such as `continue` or
+// `break`, however, these statements can both have values or be
+// value-less (i.e., these are meant to be composed "functionally" so
+// the representation of `break` captures a value that "exits the
+// current function" but the representation of `continue` does not).
+//
+// The pattern here is to define the type/representation of control
+// flow statements within the `ControlFlow` class (e.g.,
+// `ControlFlow::Continue` and `ControlFlow::Break`) but also provide
+// "syntactic sugar" to make it easier to use at the call site (e.g.,
+// the functions `Continue()` and `Break(...)`).
+template <typename T>
+class ControlFlow
+{
+public:
+ using ValueType = T;
+
+ enum class Statement
+ {
+ CONTINUE,
+ BREAK
+ };
+
+ class Continue
+ {
+ public:
+ Continue() = default;
+
+ template <typename U>
+ operator ControlFlow<U>() const
+ {
+ return ControlFlow<U>(ControlFlow<U>::Statement::CONTINUE, None());
+ }
+ };
+
+ class Break
+ {
+ public:
+ Break(T t) : t(std::move(t)) {}
+
+ template <typename U>
+ operator ControlFlow<U>() const &
+ {
+ return ControlFlow<U>(ControlFlow<U>::Statement::BREAK, t);
+ }
+
+ template <typename U>
+ operator ControlFlow<U>() &&
+ {
+ return ControlFlow<U>(ControlFlow<U>::Statement::BREAK, std::move(t));
+ }
+
+ private:
+ T t;
+ };
+
+ Statement statement() const { return s; }
+
+ T& value() & { return t.get(); }
+ const T& value() const & { return t.get(); }
+ T&& value() && { return t.get(); }
+ const T&& value() const && { return t.get(); }
+
+private:
+ template <typename U>
+ friend class ControlFlow<U>::Continue;
+ template <typename U>
+ friend class ControlFlow<U>::Break;
+
+ ControlFlow(Statement s, Option<T> t)
+ : s(s), t(std::move(t)) {}
+
+ Statement s;
+ Option<T> t;
+};
+
+
+// Provides "syntactic sugar" for creating a `ControlFlow::Continue`.
+struct Continue
+{
+ Continue() = default;
+
+ template <typename T>
+ operator ControlFlow<T>() const
+ {
+ return typename ControlFlow<T>::Continue();
+ }
+};
+
+
+// Provides "syntactic sugar" for creating a `ControlFlow::Break`.
+template <typename T>
+typename ControlFlow<typename std::decay<T>::type>::Break Break(T&& t)
+{
+ return typename ControlFlow<typename std::decay<T>::type>::Break(
+ std::forward<T>(t));
+}
+
+
+inline ControlFlow<Nothing>::Break Break()
+{
+ return ControlFlow<Nothing>::Break(Nothing());
+}
+
+
namespace internal {
-template <typename Iterate, typename Body, typename T>
-class Loop : public std::enable_shared_from_this<Loop<Iterate, Body, T>>
+template <typename Iterate, typename Body, typename T, typename R>
+class Loop : public std::enable_shared_from_this<Loop<Iterate, Body, T, R>>
{
public:
Loop(const Option<UPID>& pid, const Iterate& iterate, const Body& body)
@@ -140,7 +274,7 @@ public:
return std::weak_ptr<Loop>(shared());
}
- Future<Nothing> start()
+ Future<R> start()
{
auto self = shared();
auto weak_self = weak();
@@ -200,39 +334,47 @@ public:
discard = []() {};
while (next.isReady()) {
- Future<bool> condition = body(next.get());
- if (condition.isReady()) {
- if (condition.get()) {
- next = iterate();
- continue;
- } else {
- promise.set(Nothing());
- return;
+ Future<ControlFlow<R>> flow = body(next.get());
+ if (flow.isReady()) {
+ switch (flow->statement()) {
+ case ControlFlow<R>::Statement::CONTINUE: {
+ next = iterate();
+ continue;
+ }
+ case ControlFlow<R>::Statement::BREAK: {
+ promise.set(flow->value());
+ return;
+ }
}
} else {
- auto continuation = [self](const Future<bool>& condition) {
- if (condition.isReady()) {
- if (condition.get()) {
- self->run(self->iterate());
- } else {
- self->promise.set(Nothing());
+ auto continuation = [self](const Future<ControlFlow<R>>& flow) {
+ if (flow.isReady()) {
+ switch (flow->statement()) {
+ case ControlFlow<R>::Statement::CONTINUE: {
+ self->run(self->iterate());
+ break;
+ }
+ case ControlFlow<R>::Statement::BREAK: {
+ self->promise.set(flow->value());
+ break;
+ }
}
- } else if (condition.isFailed()) {
- self->promise.fail(condition.failure());
- } else if (condition.isDiscarded()) {
+ } else if (flow.isFailed()) {
+ self->promise.fail(flow.failure());
+ } else if (flow.isDiscarded()) {
self->promise.discard();
}
};
if (pid.isSome()) {
- condition.onAny(defer(pid.get(), continuation));
+ flow.onAny(defer(pid.get(), continuation));
} else {
- condition.onAny(continuation);
+ flow.onAny(continuation);
}
if (!promise.future().hasDiscard()) {
synchronized (mutex) {
- self->discard = [=]() mutable { condition.discard(); };
+ self->discard = [=]() mutable { flow.discard(); };
}
}
@@ -242,7 +384,7 @@ public:
// discard occurs we'll need to explicitly do discards for
// each new future that blocks.
if (promise.future().hasDiscard()) {
- condition.discard();
+ flow.discard();
}
return;
@@ -282,7 +424,7 @@ private:
const Option<UPID> pid;
Iterate iterate;
Body body;
- Promise<Nothing> promise;
+ Promise<R> promise;
// In order to discard the loop safely we capture the future that
// needs to be discarded within the `discard` function and reading
@@ -294,16 +436,14 @@ private:
} // namespace internal {
-template <typename Iterate, typename Body>
-Future<Nothing> loop(const Option<UPID>& pid, Iterate&& iterate, Body&& body)
+template <typename Iterate, typename Body, typename T, typename CF, typename V>
+Future<V> loop(const Option<UPID>& pid, Iterate&& iterate, Body&& body)
{
- using T =
- typename internal::unwrap<typename result_of<Iterate()>::type>::type;
-
using Loop = internal::Loop<
typename std::decay<Iterate>::type,
typename std::decay<Body>::type,
- T>;
+ T,
+ V>;
std::shared_ptr<Loop> loop(
new Loop(pid, std::forward<Iterate>(iterate), std::forward<Body>(body)));
http://git-wip-us.apache.org/repos/asf/mesos/blob/bbb4058d/3rdparty/libprocess/src/http.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/http.cpp b/3rdparty/libprocess/src/http.cpp
index 97d1424..689a14d 100644
--- a/3rdparty/libprocess/src/http.cpp
+++ b/3rdparty/libprocess/src/http.cpp
@@ -1607,9 +1607,9 @@ Future<Nothing> send(
[=]() mutable {
return pipeline.get();
},
- [=](const Option<Item>& item) -> Future<bool> {
+ [=](const Option<Item>& item) -> Future<ControlFlow<Nothing>> {
if (item.isNone()) {
- return false;
+ return Break();
}
Request* request = item->request;
@@ -1638,7 +1638,7 @@ Future<Nothing> send(
case Response::NONE: return send(socket, response, request);
}
}()
- .then([=]() {
+ .then([=]() -> ControlFlow<Nothing> {
// Persist the connection if the request expects it and
// the response doesn't include 'Connection: close'.
bool persist = request->keepAlive;
@@ -1647,7 +1647,10 @@ Future<Nothing> send(
persist = false;
}
}
- return persist;
+ if (persist) {
+ return Continue();
+ }
+ return Break();
});
})
.onAny([=]() {
@@ -1678,9 +1681,9 @@ Future<Nothing> receive(
[=]() {
return socket.recv(data, size);
},
- [=](size_t length) mutable -> Future<bool> {
+ [=](size_t length) mutable -> Future<ControlFlow<Nothing>> {
if (length == 0) {
- return false;
+ return Break();
}
// Decode as much of the data as possible into HTTP requests.
@@ -1706,7 +1709,7 @@ Future<Nothing> receive(
pipeline.put(Item{request, f(*request)});
}
- return true; // Keep looping!
+ return Continue(); // Keep looping!
})
.onAny([=]() {
delete decoder;
@@ -1794,15 +1797,15 @@ Future<Nothing> serve(
[=]() mutable {
return pipeline.get();
},
- [=](Option<Item> item) {
+ [=](Option<Item> item) -> ControlFlow<Nothing> {
if (item.isNone()) {
- return false;
+ return Break();
}
delete item->request;
if (promise->future().hasDiscard()) {
item->response.discard();
}
- return true;
+ return Continue();
});
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/bbb4058d/3rdparty/libprocess/src/io.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/io.cpp b/3rdparty/libprocess/src/io.cpp
index 27da897..d0b3ba1 100644
--- a/3rdparty/libprocess/src/io.cpp
+++ b/3rdparty/libprocess/src/io.cpp
@@ -329,9 +329,9 @@ Future<Nothing> splice(
[=]() {
return io::read(from, data.get(), chunk);
},
- [=](size_t length) -> Future<bool> {
+ [=](size_t length) -> Future<ControlFlow<Nothing>> {
if (length == 0) { // EOF.
- return false;
+ return Break();
}
// Send the data to the redirect hooks.
@@ -341,8 +341,8 @@ Future<Nothing> splice(
}
return io::write(to, s)
- .then([]() {
- return true;
+ .then([]() -> Future<ControlFlow<Nothing>> {
+ return Continue();
});
});
}
@@ -444,7 +444,13 @@ Future<Nothing> write(int fd, const string& data)
nonblock.error());
}
+ // We store `data.size()` so that we can just use `size` in the
+ // second lambda below versus having to make a copy of `data` in
+ // both lambdas since `data` might be very big and two copies could
+ // be expensive!
const size_t size = data.size();
+
+ // We need to share the `index` between both lambdas below.
std::shared_ptr<size_t> index(new size_t(0));
return loop(
@@ -452,8 +458,11 @@ Future<Nothing> write(int fd, const string& data)
[=]() {
return io::write(fd, data.data() + *index, size - *index);
},
- [=](size_t length) {
- return (*index += length) != size;
+ [=](size_t length) -> ControlFlow<Nothing> {
+ if ((*index += length) != size) {
+ return Continue();
+ }
+ return Break();
})
.onAny([fd]() {
os::close(fd);
http://git-wip-us.apache.org/repos/asf/mesos/blob/bbb4058d/3rdparty/libprocess/src/tests/loop_tests.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/tests/loop_tests.cpp b/3rdparty/libprocess/src/tests/loop_tests.cpp
index 8435ba8..8d1837a 100644
--- a/3rdparty/libprocess/src/tests/loop_tests.cpp
+++ b/3rdparty/libprocess/src/tests/loop_tests.cpp
@@ -13,17 +13,23 @@
#include <gmock/gmock.h>
#include <atomic>
+#include <string>
#include <process/future.hpp>
#include <process/gtest.hpp>
#include <process/loop.hpp>
#include <process/queue.hpp>
+using process::Break;
+using process::Continue;
+using process::ControlFlow;
using process::Future;
using process::loop;
using process::Promise;
using process::Queue;
+using std::string;
+
TEST(LoopTest, Sync)
{
@@ -33,8 +39,11 @@ TEST(LoopTest, Sync)
[&]() {
return value.load();
},
- [](int i) {
- return i != 0;
+ [](int i) -> ControlFlow<Nothing> {
+ if (i != 0) {
+ return Continue();
+ }
+ return Break();
});
EXPECT_TRUE(future.isPending());
@@ -50,15 +59,18 @@ TEST(LoopTest, Async)
Queue<int> queue;
Promise<int> promise1;
- Promise<bool> promise2;
+ Promise<string> promise2;
- Future<Nothing> future = loop(
+ Future<string> future = loop(
[&]() {
return queue.get();
},
[&](int i) {
promise1.set(i);
- return promise2.future();
+ return promise2.future()
+ .then([](const string& s) -> ControlFlow<string> {
+ return Break(s);
+ });
});
EXPECT_TRUE(future.isPending());
@@ -69,9 +81,11 @@ TEST(LoopTest, Async)
EXPECT_TRUE(future.isPending());
- promise2.set(false);
+ string s = "Hello world!";
- AWAIT_READY(future);
+ promise2.set(s);
+
+ AWAIT_EQ(s, future);
}
@@ -85,8 +99,8 @@ TEST(LoopTest, DiscardIterate)
[&]() {
return promise.future();
},
- [&](int i) {
- return false;
+ [&](int i) -> ControlFlow<Nothing> {
+ return Break();
});
EXPECT_TRUE(future.isPending());
@@ -100,7 +114,7 @@ TEST(LoopTest, DiscardIterate)
TEST(LoopTest, DiscardBody)
{
- Promise<bool> promise;
+ Promise<Nothing> promise;
promise.future().onDiscard([&]() { promise.discard(); });
@@ -109,7 +123,10 @@ TEST(LoopTest, DiscardBody)
return 42;
},
[&](int i) {
- return promise.future();
+ return promise.future()
+ .then([]() -> ControlFlow<Nothing> {
+ return Break();
+ });
});
EXPECT_TRUE(future.isPending());