ceph/src/seastar/tests/unit/file_io_test.cc

   1 /*
   2  * This file is open source software, licensed to you under the terms
   3  * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
   4  * distributed with this work for additional information regarding copyright
   5  * ownership.  You may not use this file except in compliance with the License.
   6  *
   7  * You may obtain a copy of the License at
   8  *
   9  *   http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing,
  12  * software distributed under the License is distributed on an
  13  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  14  * KIND, either express or implied.  See the License for the
  15  * specific language governing permissions and limitations
  16  * under the License.
  17  */
  18 /*
  19  * Copyright (C) 2014-2015 Cloudius Systems, Ltd.
  20  */
  21
  22 #include <seastar/testing/test_case.hh>
  23
  24 #include <seastar/core/semaphore.hh>
  25 #include <seastar/core/condition-variable.hh>
  26 #include <seastar/core/file.hh>
  27 #include <seastar/core/reactor.hh>
  28 #include <seastar/core/thread.hh>
  29 #include <seastar/core/stall_sampler.hh>
  30 #include <iostream>
  31
  32 using namespace seastar;
  33
  34 SEASTAR_TEST_CASE(open_flags_test) {
  35     open_flags flags = open_flags::rw | open_flags::create  | open_flags::exclusive;
  36     BOOST_REQUIRE(std::underlying_type_t<open_flags>(flags) ==
  37                   (std::underlying_type_t<open_flags>(open_flags::rw) |
  38                    std::underlying_type_t<open_flags>(open_flags::create) |
  39                    std::underlying_type_t<open_flags>(open_flags::exclusive)));
  40
  41     open_flags mask = open_flags::create  | open_flags::exclusive;
  42     BOOST_REQUIRE((flags & mask) == mask);
  43     return make_ready_future<>();
  44 }
  45
  46 struct file_test {
  47     file_test(file&& f) : f(std::move(f)) {}
  48     file f;
  49     semaphore sem = { 0 };
  50     semaphore par = { 1000 };
  51 };
  52
  53 SEASTAR_TEST_CASE(test1) {
  54     // Note: this tests generates a file "testfile.tmp" with size 4096 * max (= 40 MB).
  55     static constexpr auto max = 10000;
  56     return open_file_dma("testfile.tmp", open_flags::rw | open_flags::create).then([] (file f) {
  57         auto ft = new file_test{std::move(f)};
  58         for (size_t i = 0; i < max; ++i) {
  59             ft->par.wait().then([ft, i] {
  60                 auto wbuf = allocate_aligned_buffer<unsigned char>(4096, 4096);
  61                 std::fill(wbuf.get(), wbuf.get() + 4096, i);
  62                 auto wb = wbuf.get();
  63                 ft->f.dma_write(i * 4096, wb, 4096).then(
  64                         [ft, i, wbuf = std::move(wbuf)] (size_t ret) mutable {
  65                     BOOST_REQUIRE(ret == 4096);
  66                     auto rbuf = allocate_aligned_buffer<unsigned char>(4096, 4096);
  67                     auto rb = rbuf.get();
  68                     ft->f.dma_read(i * 4096, rb, 4096).then(
  69                             [ft, rbuf = std::move(rbuf), wbuf = std::move(wbuf)] (size_t ret) mutable {
  70                         BOOST_REQUIRE(ret == 4096);
  71                         BOOST_REQUIRE(std::equal(rbuf.get(), rbuf.get() + 4096, wbuf.get()));
  72                         ft->sem.signal(1);
  73                         ft->par.signal();
  74                     });
  75                 });
  76             });
  77         }
  78         return ft->sem.wait(max).then([ft] () mutable {
  79             return ft->f.flush();
  80         }).then([ft] {
  81             return ft->f.close();
  82         }).then([ft] () mutable {
  83             std::cout << "done\n";
  84             delete ft;
  85         });
  86     });
  87 }
  88
  89 SEASTAR_TEST_CASE(parallel_write_fsync) {
  90     return internal::report_reactor_stalls([] {
  91         return async([] {
  92             // Plan: open a file and write to it like crazy. In parallel fsync() it all the time.
  93             auto fname = "testfile.tmp";
  94             auto sz = uint64_t(32*1024*1024);
  95             auto buffer_size = 32768;
  96             auto write_concurrency = 16;
  97             auto fsync_every = 1024*1024;
  98             auto max_write_ahead_of_fsync = 4*1024*1024; // ensures writes don't complete too quickly
  99             auto written = uint64_t(0);
 100             auto fsynced_at = uint64_t(0);
 101
 102             file f = open_file_dma(fname, open_flags::rw | open_flags::create | open_flags::truncate).get0();
 103             // Avoid filesystem problems with size-extending operations
 104             f.truncate(sz).get();
 105
 106             auto fsync_semaphore = semaphore(0);
 107             auto may_write_condvar = condition_variable();
 108             auto fsync_thread = thread([&] {
 109                 auto fsynced = uint64_t(0);
 110                 while (fsynced < sz) {
 111                     fsync_semaphore.wait(fsync_every).get();
 112                     fsynced_at = written;
 113                     // Signal the condition variable now so that writes proceed
 114                     // in parallel with the fsync
 115                     may_write_condvar.broadcast();
 116                     f.flush().get();
 117                     fsynced += fsync_every;
 118                 }
 119             });
 120
 121             auto write_semaphore = semaphore(write_concurrency);
 122             while (written < sz) {
 123                 write_semaphore.wait().get();
 124                 may_write_condvar.wait([&] {
 125                     return written <= fsynced_at + max_write_ahead_of_fsync;
 126                 }).get();
 127                 auto buf = temporary_buffer<char>::aligned(f.memory_dma_alignment(), buffer_size);
 128                 f.dma_write(written, buf.get(), buf.size()).then([&fsync_semaphore, &write_semaphore, buf = std::move(buf)] (size_t w) {
 129                     fsync_semaphore.signal(buf.size());
 130                     write_semaphore.signal();
 131                 });
 132                 written += buffer_size;
 133             }
 134             write_semaphore.wait(write_concurrency).get();
 135
 136             fsync_thread.join().get();
 137             f.close().get();
 138             remove_file(fname).get();
 139         });
 140     }).then([] (internal::stall_report sr) {
 141         std::cout << "parallel_write_fsync: " << sr << "\n";
 142     });
 143 }