Added Seek & Tell functionality to ThreadedIoFile.

Change-Id: I3c714014e961d08110ce194ad865322e0b487fcb

packager/media/file/file.gyp

@@ -51,6 +51,7 @@
       'dependencies': [
         '../../testing/gtest.gyp:gtest',
         '../../testing/gtest.gyp:gtest_main',
+        '../../third_party/gflags/gflags.gyp:gflags',
         'file',
       ],
     },

packager/media/file/file_unittest.cc

@@ -4,11 +4,15 @@
 // license that can be found in the LICENSE file or at
 // https://developers.google.com/open-source/licenses/bsd
 
+#include <gflags/gflags.h>
 #include <gtest/gtest.h>
 
 #include "packager/base/files/file_util.h"
 #include "packager/media/file/file.h"
 
+DECLARE_uint64(io_cache_size);
+DECLARE_uint64(io_block_size);
+
 namespace {
 const int kDataSize = 1024;
 }
@@ -169,5 +173,111 @@ TEST_F(LocalFileTest, WriteFlushCheckSize) {
   }
 }
 
+class ParamLocalFileTest : public LocalFileTest,
+                           public ::testing::WithParamInterface<uint8_t> {
+};
+
+TEST_P(ParamLocalFileTest, SeekWriteAndSeekRead) {
+  const uint32_t kBlockSize(10);
+  const uint32_t kInitialWriteSize(100);
+  const uint32_t kFinalFileSize(200);
+
+  google::FlagSaver flag_saver;
+  FLAGS_io_block_size = kBlockSize;
+  FLAGS_io_cache_size = GetParam();
+
+  std::vector<uint8_t> buffer(kInitialWriteSize);
+  File* file = File::Open(local_file_name_no_prefix_.c_str(), "w");
+  ASSERT_TRUE(file != nullptr);
+  ASSERT_EQ(kInitialWriteSize, file->Write(buffer.data(), kInitialWriteSize));
+  EXPECT_EQ(kInitialWriteSize, file->Size());
+  uint64_t position;
+  ASSERT_TRUE(file->Tell(&position));
+  ASSERT_EQ(kInitialWriteSize, position);
+
+  for (uint8_t offset = 0; offset < kFinalFileSize; ++offset) {
+    EXPECT_TRUE(file->Seek(offset));
+    ASSERT_TRUE(file->Tell(&position));
+    EXPECT_EQ(offset, position);
+    EXPECT_EQ(2u, file->Write(buffer.data(), 2u));
+    ASSERT_TRUE(file->Tell(&position));
+    EXPECT_EQ(offset + 2u, position);
+    ++offset;
+    EXPECT_TRUE(file->Seek(offset));
+    ASSERT_TRUE(file->Tell(&position));
+    EXPECT_EQ(offset, position);
+    EXPECT_EQ(1, file->Write(&offset, 1));
+    ASSERT_TRUE(file->Tell(&position));
+    EXPECT_EQ(offset + 1u, position);
+  }
+  EXPECT_EQ(kFinalFileSize, file->Size());
+  ASSERT_TRUE(file->Close());
+
+  file = File::Open(local_file_name_no_prefix_.c_str(), "r");
+  ASSERT_TRUE(file != nullptr);
+  for (uint8_t offset = 1; offset < kFinalFileSize; offset += 2) {
+    uint8_t read_byte;
+    EXPECT_TRUE(file->Seek(offset));
+    ASSERT_TRUE(file->Tell(&position));
+    EXPECT_EQ(offset, position);
+    EXPECT_EQ(1, file->Read(&read_byte, 1));
+    ASSERT_TRUE(file->Tell(&position));
+    EXPECT_EQ(offset + 1u, position);
+    EXPECT_EQ(offset, read_byte);
+  }
+  EXPECT_EQ(0, file->Read(buffer.data(), 1));
+  ASSERT_TRUE(file->Seek(0));
+  EXPECT_EQ(1, file->Read(buffer.data(), 1));
+  EXPECT_TRUE(file->Close());
+}
+INSTANTIATE_TEST_CASE_P(TestSeekWithDifferentCacheSizes,
+                        ParamLocalFileTest,
+                        ::testing::Values(20u, 1000u));
+
+// This test should only be enabled for filesystems which do not allow seeking
+// past EOF.
+TEST_F(LocalFileTest, DISABLED_WriteSeekOutOfBounds) {
+  const uint32_t kFileSize(100);
+
+  std::vector<uint8_t> buffer(kFileSize);
+  File* file = File::Open(local_file_name_no_prefix_.c_str(), "w");
+  ASSERT_TRUE(file != nullptr);
+  ASSERT_EQ(kFileSize, file->Write(buffer.data(), kFileSize));
+  ASSERT_EQ(kFileSize, file->Size());
+  EXPECT_FALSE(file->Seek(kFileSize + 1));
+  EXPECT_TRUE(file->Seek(kFileSize));
+  EXPECT_EQ(1, file->Write(buffer.data(), 1));
+  EXPECT_TRUE(file->Seek(kFileSize + 1));
+  EXPECT_EQ(kFileSize + 1, file->Size());
+}
+
+// This test should only be enabled for filesystems which do not allow seeking
+// past EOF.
+TEST_F(LocalFileTest, DISABLED_ReadSeekOutOfBounds) {
+  const uint32_t kFileSize(100);
+
+  File::Delete(local_file_name_no_prefix_.c_str());
+  std::vector<uint8_t> buffer(kFileSize);
+  File* file = File::Open(local_file_name_no_prefix_.c_str(), "w");
+  ASSERT_TRUE(file != nullptr);
+  ASSERT_EQ(kFileSize, file->Write(buffer.data(), kFileSize));
+  ASSERT_EQ(kFileSize, file->Size());
+  ASSERT_TRUE(file->Close());
+  file = File::Open(local_file_name_no_prefix_.c_str(), "r");
+  ASSERT_TRUE(file != nullptr);
+  EXPECT_FALSE(file->Seek(kFileSize + 1));
+  EXPECT_TRUE(file->Seek(kFileSize));
+  uint64_t position;
+  EXPECT_TRUE(file->Tell(&position));
+  EXPECT_EQ(kFileSize, position);
+  EXPECT_EQ(0u, file->Read(buffer.data(), 1));
+  EXPECT_TRUE(file->Seek(0));
+  EXPECT_TRUE(file->Tell(&position));
+  EXPECT_EQ(0u, position);
+  EXPECT_EQ(kFileSize, file->Read(buffer.data(), kFileSize));
+  EXPECT_EQ(0u, file->Read(buffer.data(), 1));
+  EXPECT_TRUE(file->Close());
+}
+
 }  // namespace media
 }  // namespace edash_packager

packager/media/file/threaded_io_file.cc

@@ -26,6 +26,7 @@ ThreadedIoFile::ThreadedIoFile(scoped_ptr<File, FileCloser> internal_file,
       mode_(mode),
       cache_(io_cache_size),
       io_buffer_(io_block_size),
+      position_(0),
       size_(0),
       eof_(false),
       flushing_(false),
@@ -42,6 +43,7 @@ bool ThreadedIoFile::Open() {
   if (!internal_file_->Open())
     return false;
 
+  position_ = 0;
   size_ = internal_file_->Size();
 
   thread_.reset(new ClosureThread("ThreadedIoFile",
@@ -79,7 +81,11 @@ int64_t ThreadedIoFile::Read(void* buffer, uint64_t length) {
   if (NoBarrier_Load(&internal_file_error_))
     return NoBarrier_Load(&internal_file_error_);
 
-  return cache_.Read(buffer, length);
+
+  uint64_t bytes_read = cache_.Read(buffer, length);
+  position_ += bytes_read;
+
+  return bytes_read;
 }
 
 int64_t ThreadedIoFile::Write(const void* buffer, uint64_t length) {
@@ -90,8 +96,12 @@ int64_t ThreadedIoFile::Write(const void* buffer, uint64_t length) {
   if (NoBarrier_Load(&internal_file_error_))
     return NoBarrier_Load(&internal_file_error_);
 
-  size_ += length;
+  uint64_t bytes_written = cache_.Write(buffer, length);
-  return cache_.Write(buffer, length);
+  position_ += bytes_written;
+  if (position_ > size_)
+    size_ = position_;
+
+  return bytes_written;
 }
 
 int64_t ThreadedIoFile::Size() {
@@ -112,6 +122,42 @@ bool ThreadedIoFile::Flush() {
   return internal_file_->Flush();
 }
 
+bool ThreadedIoFile::Seek(uint64_t position) {
+  if (mode_ == kOutputMode) {
+    // Writing. Just flush the cache and seek.
+    if (!Flush()) return false;
+    if (!internal_file_->Seek(position)) return false;
+  } else {
+    // Reading. Close cache, wait for I/O thread to exit, seek, and restart
+    // I/O thread.
+    cache_.Close();
+    thread_->Join();
+    bool result = internal_file_->Seek(position);
+    if (!result) {
+      // Seek failed. Seek to logical position instead.
+      if (!internal_file_->Seek(position_) && (position != position_)) {
+        LOG(WARNING) << "Seek failed. ThreadedIoFile left in invalid state.";
+      }
+    }
+    cache_.Reopen();
+    eof_ = false;
+    thread_.reset(new ClosureThread("ThreadedIoFile",
+                                    base::Bind(&ThreadedIoFile::RunInInputMode,
+                                               base::Unretained(this))));
+    thread_->Start();
+    if (!result) return false;
+  }
+  position_ = position;
+  return true;
+}
+
+bool ThreadedIoFile::Tell(uint64_t* position) {
+  DCHECK(position);
+
+  *position = position_;
+  return true;
+}
+
 void ThreadedIoFile::RunInInputMode() {
   DCHECK(internal_file_);
   DCHECK(thread_);
@@ -126,20 +172,12 @@ void ThreadedIoFile::RunInInputMode() {
       cache_.Close();
       return;
     }
-    cache_.Write(&io_buffer_[0], read_result);
+    if (cache_.Write(&io_buffer_[0], read_result) == 0) {
+      return;
+    }
   }
 }
 
-bool ThreadedIoFile::Seek(uint64_t position) {
-  NOTIMPLEMENTED();
-  return false;
-}
-
-bool ThreadedIoFile::Tell(uint64_t* position) {
-  NOTIMPLEMENTED();
-  return false;
-}
-
 void ThreadedIoFile::RunInOutputMode() {
   DCHECK(internal_file_);
   DCHECK(thread_);

packager/media/file/threaded_io_file.h

@@ -58,6 +58,7 @@ class ThreadedIoFile : public File {
   IoCache cache_;
   std::vector<uint8_t> io_buffer_;
   scoped_ptr<ClosureThread> thread_;
+  uint64_t position_;
   uint64_t size_;
   base::subtle::Atomic32 eof_;
   bool flushing_;