tests: introduce pwrite-wrap test for slow I/O

pwrite can be a slow, blocking function on an overloaded
system, but a slow pwrite requires a wrapper to simulate.

This allows us to have coverage of the:

	if (mog_ioq_contended())
		return MOG_NEXT_WAIT_RD;

cases in http_put.c

5 files changed, 241 insertions, 1 deletions

diff --git a/Makefile.am b/Makefile.am
index 494bfa8..ad19560 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -169,6 +169,12 @@ test_epoll_wrap_SOURCES = $(cmogstored_SOURCES) test/epoll-wrap.c
 test_epoll_wrap_LDFLAGS = $(cmogstored_LDFLAGS) $(AM_LDFLAGS) \
                           -Wl,--wrap=epoll_ctl -Wl,--wrap=epoll_create
 endif # HAVE_EPOLL
+
+check_PROGRAMS += test/pwrite-wrap
+test_pwrite_wrap_SOURCES = $(cmogstored_SOURCES) test/pwrite-wrap.c
+test_pwrite_wrap_LDFLAGS = $(cmogstored_LDFLAGS) $(AM_LDFLAGS) \
+                          -Wl,--wrap=pwrite
+
 endif # HAVE_LD_WRAP
 
 HELP2MAN = help2man
diff --git a/test/.gitignore b/test/.gitignore
index 02f2051..824eaf7 100644
--- a/test/.gitignore
+++ b/test/.gitignore
@@ -1,4 +1,5 @@
 /.*
 slow.mk
 epoll-wrap
+pwrite-wrap
 *.so
diff --git a/test/pwrite-wrap.c b/test/pwrite-wrap.c
new file mode 100644
index 0000000..7939041
--- /dev/null
+++ b/test/pwrite-wrap.c
@@ -0,0 +1,97 @@
+/*
+ * Copyright (C) 2013, Eric Wong <normalperson@yhbt.net>
+ * License: GPLv3 or later (see COPYING for details)
+ */
+/*
+ * fault injection wrapper for pwrite
+ */
+#include "cmogstored.h"
+#include "iov_str.h"
+#include <poll.h>
+#include <sys/uio.h>
+static sig_atomic_t pwrite_wrap_flags;
+static int pwrite_slow_msec = 5000;
+#define PWRITE_WRAP_NONE        (0)
+#define PWRITE_WRAP_SLOW        (1)
+#define PWRITE_WRAP_PARTIAL        (1 << 1)
+#define PWRITE_WRAP_ENOSPC        (1 << 2)
+#define PWRITE_WRAP_MAX                4
+static const char * const pwrite_wrap_names[] = {
+        "none", "slow", "partial", "enospc"
+};
+
+#define EMIT(s) write(STDERR_FILENO, (s), sizeof(s)-1)
+
+/* test/pwrite_wrap.rb depends on the following line */
+static const char msg[] = "pwrite fault injection\n";
+
+int __real_pwrite(int fd, void *buf, size_t count, off_t offset);
+
+int __wrap_pwrite(int fd, void *buf, size_t count, off_t offset)
+{
+        if (pwrite_wrap_flags & PWRITE_WRAP_SLOW) {
+                poll(NULL, 0, pwrite_slow_msec);
+        }
+        if ((pwrite_wrap_flags & PWRITE_WRAP_PARTIAL) && (count > 0))
+                count--;
+        if (pwrite_wrap_flags & PWRITE_WRAP_ENOSPC) {
+                errno = ENOSPC;
+                return -1;
+        }
+
+        return __real_pwrite(fd, buf, count, offset);
+}
+
+static void set_wrap_pwrite(int signum)
+{
+        static sig_atomic_t pwrite_wrap_knob = 0;
+        struct iovec vec[3];
+        union { const char *in; void *out; } name;
+
+        switch (signum) {
+        case SIGVTALRM:
+                pwrite_wrap_knob = (pwrite_wrap_knob + 1) % PWRITE_WRAP_MAX;
+                IOV_STR(&vec[0], "knob set: ");
+                break;
+        case SIGTTIN:
+                pwrite_wrap_flags |= pwrite_wrap_knob;
+                IOV_STR(&vec[0], "flag set: ");
+                break;
+        case SIGTTOU:
+                pwrite_wrap_flags ^= pwrite_wrap_knob;
+                IOV_STR(&vec[0], "flag clr: ");
+                break;
+        default: assert(0 && "unknown signal caught");
+        }
+
+        name.in = pwrite_wrap_names[pwrite_wrap_knob];
+        vec[1].iov_base = name.out;
+        vec[1].iov_len = strlen(name.in);
+
+        IOV_STR(&vec[2], "\n");
+        writev(STDERR_FILENO, vec, 3);
+}
+
+__attribute__((constructor)) void pwrite_wrap_init(void)
+{
+        struct sigaction sa;
+        const char slow_env[] = "PWRITE_WRAP_SLOW_MSEC";
+        const char *msec = getenv(slow_env);
+
+        memset(&sa, 0, sizeof(struct sigaction));
+        CHECK(int, 0, sigemptyset(&sa.sa_mask) );
+        sa.sa_handler = set_wrap_pwrite;
+        CHECK(int, 0, sigaction(SIGVTALRM, &sa, NULL));
+        CHECK(int, 0, sigaction(SIGTTIN, &sa, NULL));
+        CHECK(int, 0, sigaction(SIGTTOU, &sa, NULL));
+
+        if (msec) {
+                char *end;
+                unsigned long v = strtoul(msec, &end, 10);
+
+                if (*end || v > INT_MAX)
+                        die("Invalid %s=%s", slow_env, msec);
+                pwrite_slow_msec = v;
+                fprintf(stderr, "set %s=%d\n", slow_env, pwrite_slow_msec);
+        }
+}
diff --git a/test/pwrite_wrap.rb b/test/pwrite_wrap.rb
new file mode 100644
index 0000000..34d342a
--- /dev/null
+++ b/test/pwrite_wrap.rb
@@ -0,0 +1,136 @@
+#!/usr/bin/env ruby
+# -*- encoding: binary -*-
+# Copyright (C) 2013, Eric Wong <normalperson@yhbt.net>
+# License: GPLv3 or later (see COPYING for details)
+require 'test/test_helper'
+require 'timeout'
+
+TEST_PROG = 'test/pwrite-wrap'
+class TestPwriteWrap < Test::Unit::TestCase
+  def setup
+    @tmpdir = Dir.mktmpdir('cmogstored-pwrite-wrap-test')
+    @to_close = []
+    @host = TEST_HOST
+
+    srv = TCPServer.new(@host, 0)
+    @port = srv.addr[1]
+    srv.close
+
+    srv = TCPServer.new(@host, 0)
+    @mgmt_port = srv.addr[1]
+    srv.close
+
+    Dir.mkdir("#@tmpdir/dev666")
+    Dir.mkdir("#@tmpdir/dev333")
+    File.open("#@tmpdir/dev666/get.fid", "w") { |fp|
+      fp.seek(1000)
+      fp.write("\0")
+    }
+
+    @err = Tempfile.new("stderr")
+    cmd = [ TEST_PROG, "--docroot=#@tmpdir", "--httplisten=#@host:#@port",
+            "--mgmtlisten=#@host:#@mgmt_port", "--maxconns=500" ]
+
+    @slow_time = 5000
+    if vg = ENV["VALGRIND"]
+      cmd = vg.split(/\s+/).concat(cmd)
+      # valgrind slows everything down, so increase the sleep time
+      @slow_time *= 2
+    end
+    ENV["PWRITE_WRAP_SLOW_MSEC"] ||= (@slow_time * 2).to_s
+    @pid = fork {
+      $stderr.reopen(@err.path, "a")
+      @err.close
+      exec(*cmd)
+    }
+    @client = get_client
+    @mgmt_client = get_client(300, @mgmt_port)
+    @mgmt_client.write "server aio_threads = 1\r\n"
+    @mgmt_client.gets
+    warning = "fewer aio_threads(1) than devices(2)"
+    wait_for(30, warning) do
+      buf = File.read(@err.path)
+      if buf =~ /aio_threads/
+        puts "BUF: #{buf}"
+      end
+      buf.include?(warning)
+    end
+  end
+
+  def wait_for(sec, reason, res = 0.1)
+    stop = Time.now + sec
+    begin
+      return if yield
+      sleep res
+    end while Time.now < stop
+    assert false, reason
+  end
+
+  def test_pwrite_slow_chunked
+    __test_pwrite_slow(true)
+  end
+
+  def test_pwrite_slow_identity
+    __test_pwrite_slow(false)
+  end
+
+  def __test_pwrite_slow(chunked)
+    Process.kill(:VTALRM, @pid)
+    t_yield
+    Process.kill(:TTIN, @pid)
+    wait_for(5, "enable slow knob") do
+      File.readlines(@err.path).grep(/knob set: slow/)[0] &&
+                       File.readlines(@err.path).grep(/flag set: slow/)[0]
+    end
+
+    client_2 = get_client
+
+    thr = Thread.new do
+      begin
+        size = 1024 * 1024 * 1024 * 10
+        if chunked
+          @client.write("PUT /dev666/foo.fid HTTP/1.1\r\n" \
+                        "Host: example.com\r\n" \
+                        "Transfer-Encoding: chunked\r\n\r\n" \
+                        "#{'%x' % size}\r\n")
+        else
+          @client.write("PUT /dev666/foo.fid HTTP/1.0\r\n" \
+                        "Content-Length: #{size}\r\n\r\n")
+        end
+        IO.copy_stream("/dev/zero", @client, size)
+      rescue => err
+      end
+      err
+    end
+
+    wait_for(5, "temporary file to exist", 0.05) do
+      Dir["#@tmpdir/dev666/foo.*tmp"][0]
+    end
+    t_yield
+
+    # this should cause mog_ioq_contended to return true in http_put.c
+    a = Time.now
+    Timeout.timeout(@slow_time * 4) {
+      client_2.write("GET /dev666/get.fid HTTP/1.0\r\n\r\n")
+      assert_match(%r{HTTP/1\.1 200 OK}, client_2.gets)
+    }
+    diff = Time.now - a
+    assert_operator diff, :<, (@slow_time * 2), "diff=#{diff}"
+    @client.shutdown
+    thr.join
+    Process.kill(:TTOU, @pid)
+    wait_for(5, "clear slow flag") do
+      File.readlines(@err.path).grep(/flag set: slow/)[0]
+    end
+  end
+
+  def teardown
+    Process.kill(:QUIT, @pid) rescue nil
+    _, status = Process.waitpid2(@pid)
+    @to_close.each { |io| io.close unless io.closed? }
+    FileUtils.rm_rf(@tmpdir)
+    @err.rewind
+    $stderr.write(@err.read) if $DEBUG
+    assert status.success?, status.inspect
+  end
+end if File.exist?(TEST_PROG)
diff --git a/test/ruby.mk b/test/ruby.mk
index b089db8..5e1a1ca 100644
--- a/test/ruby.mk
+++ b/test/ruby.mk
@@ -4,5 +4,5 @@ RB_TESTS_FAST = test/cmogstored-cfg.rb test/http_dav.rb test/http_range.rb \
 RB_TESTS_SLOW = test/mgmt-usage.rb test/mgmt.rb test/mgmt-iostat.rb \
  test/http.rb test/http_put_slow.rb test/http_chunked_put.rb \
  test/graceful_quit.rb test/http_idle_expire.rb \
- test/mgmt_auto_adjust.rb
+ test/mgmt_auto_adjust.rb test/pwrite_wrap.rb
 RB_TESTS = $(RB_TESTS_FAST) $(RB_TESTS_SLOW)