net: FileStream cleanup

net/base/file_stream_context_posix.cc

Index: net/base/file_stream_context_posix.cc
diff --git a/net/base/file_stream_context_posix.cc b/net/base/file_stream_context_posix.cc
index 9f3d060822ffd19629658d209d19474463c2df0d..677246922cb53496c84f801a664a162e3c85f041 100644
--- a/net/base/file_stream_context_posix.cc
+++ b/net/base/file_stream_context_posix.cc
@@ -2,16 +2,9 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-// For 64-bit file access (off_t = off64_t, lseek64, etc).
-#define _FILE_OFFSET_BITS 64
-
 #include "net/base/file_stream_context.h"
 
 #include <errno.h>
-#include <fcntl.h>
-#include <sys/stat.h>
-#include <sys/types.h>
-#include <unistd.h>
 
 #include "base/basictypes.h"
 #include "base/bind.h"
@@ -27,24 +20,8 @@
 #include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
 
-#if defined(OS_ANDROID)
-// Android's bionic libc only supports the LFS transitional API.
-#define off_t off64_t
-#define lseek lseek64
-#define stat stat64
-#define fstat fstat64
-#endif
-
 namespace net {
 
-// We cast back and forth, so make sure it's the size we're expecting.
-COMPILE_ASSERT(sizeof(int64) == sizeof(off_t), off_t_64_bit);
-
-// Make sure our Whence mappings match the system headers.
-COMPILE_ASSERT(FROM_BEGIN   == SEEK_SET &&
-               FROM_CURRENT == SEEK_CUR &&
-               FROM_END     == SEEK_END, whence_matches_system);
-
 FileStream::Context::Context(const scoped_refptr<base::TaskRunner>& task_runner)
     : async_in_progress_(false),
       orphaned_(false),

[hashimoto, 2014/06/12 13:03:37]

nit: Please don't leave buf_len_ uninitialized.

[rvargas (doing something else), 2014/06/13 02:49:10]

Done.

@@ -62,12 +39,13 @@ FileStream::Context::Context(base::File file,
 FileStream::Context::~Context() {
 }
 
-int FileStream::Context::ReadAsync(IOBuffer* in_buf,
-                                   int buf_len,
-                                   const CompletionCallback& callback) {
+int FileStream::Context::Read(IOBuffer* in_buf,
+                              int buf_len,
+                              const CompletionCallback& callback) {
   DCHECK(!async_in_progress_);
 
   scoped_refptr<IOBuffer> buf = in_buf;
+  buf_ = NULL;
   const bool posted = base::PostTaskAndReplyWithResult(
       task_runner_.get(),
       FROM_HERE,
@@ -81,12 +59,30 @@ int FileStream::Context::ReadAsync(IOBuffer* in_buf,
   return ERR_IO_PENDING;
 }
 
-int FileStream::Context::WriteAsync(IOBuffer* in_buf,
-                                    int buf_len,
-                                    const CompletionCallback& callback) {
+int FileStream::Context::ReadNoBlocking(IOBuffer* buf,
+                                        int buf_len,
+                                        const CompletionCallback& callback) {
+  DCHECK(!async_in_progress_);
+
+  buf_ = buf;
+  buf_len_ = buf_len;
+  callback_ = callback;
+
+  base::MessageLoopForIO::current()->WatchFileDescriptor(
+      file_.GetPlatformFile(), false /* persistent */,
+      base::MessageLoopForIO::WATCH_READ, &file_watcher_, this);

[hashimoto, 2014/06/12 13:03:37]

I'm quite unfamiliar with libevent nor details of message pump.
Isn't there any case where something goes wrong (e.g. file get deleted, disk
gets removed physically) and we end up waiting for the event forever?

[rvargas (doing something else), 2014/06/13 02:49:10]

Good point.

I don't know if there is something to do in the case of critical failures. I
know that we can wait forever if this is used for a regular file. I know some
types of fd will just trigger the notification on "critical" failures, but I
suspect that some devices won't do that.

I added code to allow proper cleanup when the consumer gets tired of waiting (or
detect a failure by some other means).

I guess it is up to the caller to decide if it is safe to use non-blocking io.

[hashimoto, 2014/06/13 06:10:42]

Makes sense.
Hope the comment in file_stream.h prevents users from carelessly choosing the
non-blocking variant when inappropriate.

+
+  async_in_progress_ = true;
+  return ERR_IO_PENDING;
+}
+
+int FileStream::Context::Write(IOBuffer* in_buf,
+                               int buf_len,
+                               const CompletionCallback& callback) {
   DCHECK(!async_in_progress_);
 
   scoped_refptr<IOBuffer> buf = in_buf;
+  buf_ = NULL;
   const bool posted = base::PostTaskAndReplyWithResult(
       task_runner_.get(),
       FROM_HERE,
@@ -100,18 +96,27 @@ int FileStream::Context::WriteAsync(IOBuffer* in_buf,
   return ERR_IO_PENDING;
 }
 
-FileStream::Context::IOResult FileStream::Context::SeekFileImpl(Whence whence,
-                                                                int64 offset) {
-  off_t res = lseek(file_.GetPlatformFile(), static_cast<off_t>(offset),
-                    static_cast<int>(whence));
-  if (res == static_cast<off_t>(-1))
-    return IOResult::FromOSError(errno);
+int FileStream::Context::WriteNoBlocking(IOBuffer* buf,
+                                         int buf_len,
+                                         const CompletionCallback& callback) {
+  DCHECK(!async_in_progress_);
 
-  return IOResult(res, 0);
+  buf_ = buf;
+  buf_len_ = buf_len;
+  callback_ = callback;
+
+  base::MessageLoopForIO::current()->WatchFileDescriptor(
+      file_.GetPlatformFile(), false /* persistent */,
+      base::MessageLoopForIO::WATCH_WRITE, &file_watcher_, this);
+
+  async_in_progress_ = true;
+  return ERR_IO_PENDING;
 }
 
-FileStream::Context::IOResult FileStream::Context::FlushFileImpl() {
-  ssize_t res = HANDLE_EINTR(fsync(file_.GetPlatformFile()));
+FileStream::Context::IOResult FileStream::Context::SeekFileImpl(
+    base::File::Whence whence,
+    int64 offset) {
+  int64 res = file_.Seek(whence, offset);
   if (res == -1)
     return IOResult::FromOSError(errno);
 
@@ -121,9 +126,7 @@ FileStream::Context::IOResult FileStream::Context::FlushFileImpl() {
 FileStream::Context::IOResult FileStream::Context::ReadFileImpl(
     scoped_refptr<IOBuffer> buf,
     int buf_len) {
-  // Loop in the case of getting interrupted by a signal.
-  ssize_t res = HANDLE_EINTR(read(file_.GetPlatformFile(), buf->data(),
-                                  static_cast<size_t>(buf_len)));
+  int res = file_.ReadAtCurrentPosNoBestEffort(buf->data(), buf_len);
   if (res == -1)
     return IOResult::FromOSError(errno);
 
@@ -133,12 +136,25 @@ FileStream::Context::IOResult FileStream::Context::ReadFileImpl(
 FileStream::Context::IOResult FileStream::Context::WriteFileImpl(
     scoped_refptr<IOBuffer> buf,
     int buf_len) {
-  ssize_t res = HANDLE_EINTR(write(file_.GetPlatformFile(), buf->data(),
-                                   buf_len));
+  int res = file_.WriteAtCurrentPosNoBestEffort(buf->data(), buf_len);
   if (res == -1)
     return IOResult::FromOSError(errno);
 
   return IOResult(res, 0);
 }
 
+void FileStream::Context::OnFileCanReadWithoutBlocking(int fd) {
+  DCHECK_EQ(file_.GetPlatformFile(), fd);
+  DCHECK(async_in_progress_);
+  async_in_progress_ = false;
+  Read(buf_.get(), buf_len_, callback_);
+}
+
+void FileStream::Context::OnFileCanWriteWithoutBlocking(int fd) {
+  DCHECK_EQ(file_.GetPlatformFile(), fd);
+  DCHECK(async_in_progress_);
+  async_in_progress_ = false;
+  Write(buf_.get(), buf_len_, callback_);
+}
+
 }  // namespace net