update ceph source to reef 18.1.2

[ceph.git] / ceph / src / log / test.cc

diff --git a/ceph/src/log/test.cc b/ceph/src/log/test.cc
index f1ea107518ed1efd17bccd476b530c183baf214e..08ba5bff6afb75550c80370451d61b29255e556c 100644 (file)
--- a/ceph/src/log/test.cc
+++ b/ceph/src/log/test.cc
@@ -10,6 +10,10 @@
 #include "global/global_context.h"
 #include "common/dout.h"
 
+#include <unistd.h>
+
+#include <limits.h>
+
 using namespace std;
 using namespace ceph::logging;
 
@@ -166,6 +170,125 @@ TEST(Log, ManyGather)
   log.stop();
 }
 
+static void readpipe(int fd, int verify)
+{
+  while (1) {
+    /* Use larger buffer on receiver as Linux will allow pipes buffers to
+     * exceed PIPE_BUF. We can't avoid tearing due to small read buffers from
+     * the Ceph side.
+     */
+
+    char buf[65536] = "";
+    int rc = read(fd, buf, (sizeof buf) - 1);
+    if (rc == 0) {
+      _exit(0);
+    } else if (rc == -1) {
+      _exit(1);
+    } else if (rc > 0) {
+      if (verify) {
+        char* p = strrchr(buf, '\n');
+        /* verify no torn writes */
+        if (p == NULL) {
+          _exit(2);
+        } else if (p[1] != '\0') {
+          write(2, buf, strlen(buf));
+          _exit(3);
+        }
+      }
+    } else _exit(100);
+    usleep(500);
+  }
+}
+
+TEST(Log, StderrPipeAtomic)
+{
+  int pfd[2] = {-1, -1};
+  int rc = pipe(pfd);
+  ASSERT_EQ(rc, 0);
+  pid_t pid = fork();
+  if (pid == 0) {
+    close(pfd[1]);
+    readpipe(pfd[0], 1);
+  } else if (pid == (pid_t)-1) {
+    ASSERT_EQ(0, 1);
+  }
+  close(pfd[0]);
+
+  SubsystemMap subs;
+  subs.set_log_level(1, 20);
+  subs.set_gather_level(1, 10);
+  Log log(&subs);
+  log.start();
+  log.set_log_file("");
+  log.reopen_log_file();
+  log.set_stderr_fd(pfd[1]);
+  log.set_stderr_level(1, 20);
+  /* -128 for prefix space */
+  for (int i = 0; i < PIPE_BUF-128; i++) {
+    MutableEntry e(1, 1);
+    auto& s = e.get_ostream();
+    for (int j = 0; j < i; j++) {
+      char c = 'a';
+      c += (j % 26);
+      s << c;
+    }
+    log.submit_entry(std::move(e));
+  }
+  log.flush();
+  log.stop();
+  close(pfd[1]);
+  int status;
+  pid_t waited = waitpid(pid, &status, 0);
+  ASSERT_EQ(pid, waited);
+  ASSERT_NE(WIFEXITED(status), 0);
+  ASSERT_EQ(WEXITSTATUS(status), 0);
+}
+
+TEST(Log, StderrPipeBig)
+{
+  int pfd[2] = {-1, -1};
+  int rc = pipe(pfd);
+  ASSERT_EQ(rc, 0);
+  pid_t pid = fork();
+  if (pid == 0) {
+    /* no verification as some reads will be torn due to size > PIPE_BUF */
+    close(pfd[1]);
+    readpipe(pfd[0], 0);
+  } else if (pid == (pid_t)-1) {
+    ASSERT_EQ(0, 1);
+  }
+  close(pfd[0]);
+
+  SubsystemMap subs;
+  subs.set_log_level(1, 20);
+  subs.set_gather_level(1, 10);
+  Log log(&subs);
+  log.start();
+  log.set_log_file("");
+  log.reopen_log_file();
+  log.set_stderr_fd(pfd[1]);
+  log.set_stderr_level(1, 20);
+  /* -128 for prefix space */
+  for (int i = 0; i < PIPE_BUF*2; i++) {
+    MutableEntry e(1, 1);
+    auto& s = e.get_ostream();
+    for (int j = 0; j < i; j++) {
+      char c = 'a';
+      c += (j % 26);
+      s << c;
+    }
+    log.submit_entry(std::move(e));
+  }
+  log.flush();
+  log.stop();
+  close(pfd[1]);
+  int status;
+  pid_t waited = waitpid(pid, &status, 0);
+  ASSERT_EQ(pid, waited);
+  ASSERT_NE(WIFEXITED(status), 0);
+  ASSERT_EQ(WEXITSTATUS(status), 0);
+}
+
 void do_segv()
 {
   SubsystemMap subs;