You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by an...@apache.org on 2018/06/28 18:22:20 UTC
[06/16] mesos git commit: Added `io::prepare_async` and
`io::is_async` functions.
Added `io::prepare_async` and `io::is_async` functions.
The `os::nonblock` and `os::isNonBlock` fucntions are stubs that only
do things for Windows sockets. For the IOCP implementation, we need to
associate the handles with the IOCP handel, which is created by
libprocess. Thus, we need these new functions to prepare the handle
for the async IO functions by either setting the hande to non-blocking
on POSIX systems or by associating it with an IOCP handle on Windows.
Review: https://reviews.apache.org/r/67386/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/b6f3d715
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/b6f3d715
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/b6f3d715
Branch: refs/heads/master
Commit: b6f3d715380884d2bd690508dd3ce2df323fa46d
Parents: 2310609
Author: Akash Gupta <ak...@hotmail.com>
Authored: Wed Jun 27 14:30:12 2018 -0700
Committer: Andrew Schwartzmeyer <an...@schwartzmeyer.com>
Committed: Wed Jun 27 15:06:10 2018 -0700
----------------------------------------------------------------------
3rdparty/libprocess/include/process/io.hpp | 30 +++++++-
3rdparty/libprocess/src/io.cpp | 72 +++++++++++---------
3rdparty/libprocess/src/socket.cpp | 7 +-
.../libprocess/src/tests/subprocess_tests.cpp | 4 +-
4 files changed, 75 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/b6f3d715/3rdparty/libprocess/include/process/io.hpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/include/process/io.hpp b/3rdparty/libprocess/include/process/io.hpp
index cc2caf4..5329bfe 100644
--- a/3rdparty/libprocess/include/process/io.hpp
+++ b/3rdparty/libprocess/include/process/io.hpp
@@ -46,6 +46,30 @@ const short WRITE = 0x02;
const size_t BUFFERED_READ_SIZE = 16*4096;
/**
+ * Prepares a file descriptor to be ready for asynchronous IO. On POSIX
+ * systems, this sets the file descriptor to non-blocking. On Windows, this
+ * will assign the file descriptor to an IO completion port.
+ *
+ * NOTE: Because the IO completion port is only known at the libprocess level,
+ * we need this function instead of simply using stout's `os::nonblock` and
+ * `os::isNonblock` functions like we could do for POSIX systems.
+ *
+ * @return On success, returns Nothing. On error, returns an Error.
+ */
+Try<Nothing> prepare_async(int_fd fd);
+
+
+/**
+ * Checks if `io::prepare_async` has been called on the file descriptor.
+ *
+ * @return Returns if the file descriptor is asynchronous. An asynchronous
+ * file descriptor is defined to be non-blocking on POSIX systems and
+ * overlapped and associated with an IO completion port on Windows.
+ * An error will be returned if the file descriptor is invalid.
+ */
+Try<bool> is_async(int_fd fd);
+
+/**
* Returns the events (a subset of the events specified) that can be
* performed on the specified file descriptor without blocking.
*
@@ -58,7 +82,8 @@ Future<short> poll(int_fd fd, short events);
/**
* Performs a single non-blocking read by polling on the specified
- * file descriptor until any data can be be read.
+ * file descriptor until any data can be be read. `io::prepare_async`
+ * needs to be called beforehand.
*
* The future will become ready when some data is read (may be less than
* the specified size).
@@ -85,7 +110,8 @@ Future<std::string> read(int_fd fd);
/**
* Performs a single non-blocking write by polling on the specified
- * file descriptor until data can be be written.
+ * file descriptor until data can be be written. `io::prepare_async`
+ * needs to be called beforehand.
*
* The future will become ready when some data is written (may be less than
* the specified size of the data).
http://git-wip-us.apache.org/repos/asf/mesos/blob/b6f3d715/3rdparty/libprocess/src/io.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/io.cpp b/3rdparty/libprocess/src/io.cpp
index 97f2b17..70715e2 100644
--- a/3rdparty/libprocess/src/io.cpp
+++ b/3rdparty/libprocess/src/io.cpp
@@ -133,19 +133,33 @@ Future<size_t> write(int_fd fd, const void* data, size_t size)
} // namespace internal {
+Try<Nothing> prepare_async(int_fd fd)
+{
+ // TODO(akagup): Add windows iocp.
+ return os::nonblock(fd);
+}
+
+
+Try<bool> is_async(int_fd fd)
+{
+ // TODO(akagup): Add windows iocp.
+ return os::isNonblock(fd);
+}
+
+
Future<size_t> read(int_fd fd, void* data, size_t size)
{
process::initialize();
// Check the file descriptor.
- Try<bool> nonblock = os::isNonblock(fd);
- if (nonblock.isError()) {
+ Try<bool> async = is_async(fd);
+ if (async.isError()) {
// The file descriptor is not valid (e.g., has been closed).
- return Failure("Failed to check if file descriptor was non-blocking: " +
- nonblock.error());
- } else if (!nonblock.get()) {
- // The file descriptor is not non-blocking.
- return Failure("Expected a non-blocking file descriptor");
+ return Failure(
+ "Failed to check if file descriptor was asynchronous: " +
+ async.error());
+ } else if (!async.get()) {
+ return Failure("Expected an asynchronous file descriptor.");
}
return internal::read(fd, data, size);
@@ -157,15 +171,14 @@ Future<size_t> write(int_fd fd, const void* data, size_t size)
process::initialize();
// Check the file descriptor.
- Try<bool> nonblock = os::isNonblock(fd);
- if (nonblock.isError()) {
+ Try<bool> async = is_async(fd);
+ if (async.isError()) {
// The file descriptor is not valid (e.g., has been closed).
return Failure(
- "Failed to check if file descriptor was non-blocking: " +
- nonblock.error());
- } else if (!nonblock.get()) {
- // The file descriptor is not non-blocking.
- return Failure("Expected a non-blocking file descriptor");
+ "Failed to check if file descriptor was asynchronous: " +
+ async.error());
+ } else if (!async.get()) {
+ return Failure("Expected an asynchronous file descriptor.");
}
return internal::write(fd, data, size);
@@ -237,13 +250,12 @@ Future<string> read(int_fd fd)
cloexec.error());
}
- // Make the file descriptor non-blocking.
- Try<Nothing> nonblock = os::nonblock(fd);
- if (nonblock.isError()) {
+ Try<Nothing> async = prepare_async(fd);
+ if (async.isError()) {
os::close(fd);
return Failure(
- "Failed to make duplicated file descriptor non-blocking: " +
- nonblock.error());
+ "Failed to make duplicated file descriptor asynchronous: " +
+ async.error());
}
// TODO(benh): Wrap up this data as a struct, use 'Owner'.
@@ -298,13 +310,12 @@ Future<Nothing> write(int_fd fd, const string& data)
cloexec.error());
}
- // Make the file descriptor non-blocking.
- Try<Nothing> nonblock = os::nonblock(fd);
- if (nonblock.isError()) {
+ Try<Nothing> async = prepare_async(fd);
+ if (async.isError()) {
os::close(fd);
return Failure(
- "Failed to make duplicated file descriptor non-blocking: " +
- nonblock.error());
+ "Failed to make duplicated file descriptor asynchronous: " +
+ async.error());
}
// We store `data.size()` so that we can just use `size` in the
@@ -389,19 +400,18 @@ Future<Nothing> redirect(
return Failure("Failed to set close-on-exec on 'to': " + cloexec.error());
}
- // Make the file descriptors non-blocking (no-op if already set).
- Try<Nothing> nonblock = os::nonblock(from);
- if (nonblock.isError()) {
+ Try<Nothing> async = prepare_async(from);
+ if (async.isError()) {
os::close(from);
os::close(to.get());
- return Failure("Failed to make 'from' non-blocking: " + nonblock.error());
+ return Failure("Failed to make 'from' asynchronous: " + async.error());
}
- nonblock = os::nonblock(to.get());
- if (nonblock.isError()) {
+ async = prepare_async(to.get());
+ if (async.isError()) {
os::close(from);
os::close(to.get());
- return Failure("Failed to make 'to' non-blocking: " + nonblock.error());
+ return Failure("Failed to make 'to' asynchronous: " + async.error());
}
// NOTE: We wrap `os::close` in a lambda to disambiguate on Windows.
http://git-wip-us.apache.org/repos/asf/mesos/blob/b6f3d715/3rdparty/libprocess/src/socket.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/socket.cpp b/3rdparty/libprocess/src/socket.cpp
index 504cb54..3031cd8 100644
--- a/3rdparty/libprocess/src/socket.cpp
+++ b/3rdparty/libprocess/src/socket.cpp
@@ -15,6 +15,7 @@
#include <boost/shared_array.hpp>
+#include <process/io.hpp>
#include <process/loop.hpp>
#include <process/network.hpp>
#include <process/owned.hpp>
@@ -79,10 +80,10 @@ Try<std::shared_ptr<SocketImpl>> SocketImpl::create(
return Error("Failed to create socket: " + s.error());
}
- Try<Nothing> nonblock = os::nonblock(s.get());
- if (nonblock.isError()) {
+ Try<Nothing> async = io::prepare_async(s.get());
+ if (async.isError()) {
os::close(s.get());
- return Error("Failed to create socket, nonblock: " + nonblock.error());
+ return Error("Failed to create socket, prepare_async: " + async.error());
}
Try<Nothing> cloexec = os::cloexec(s.get());
http://git-wip-us.apache.org/repos/asf/mesos/blob/b6f3d715/3rdparty/libprocess/src/tests/subprocess_tests.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/tests/subprocess_tests.cpp b/3rdparty/libprocess/src/tests/subprocess_tests.cpp
index 568d77b..b12490e 100644
--- a/3rdparty/libprocess/src/tests/subprocess_tests.cpp
+++ b/3rdparty/libprocess/src/tests/subprocess_tests.cpp
@@ -465,10 +465,10 @@ TEST_F(SubprocessTest, PipeRedirect)
S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);
ASSERT_SOME(fd);
- ASSERT_SOME(os::nonblock(fd.get()));
+ ASSERT_SOME(io::prepare_async(fd.get()));
ASSERT_SOME(s->out());
- ASSERT_SOME(os::nonblock(s->out().get()));
+ ASSERT_SOME(io::prepare_async(s->out().get()));
AWAIT_READY(io::redirect(s->out().get(), fd.get()));
// Close our copy of the fd.