ceph/src/rocksdb/env/posix_logger.h

   1 //  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
   2 //  This source code is licensed under the BSD-style license found in the
   3 //  LICENSE file in the root directory of this source tree. An additional grant
   4 //  of patent rights can be found in the PATENTS file in the same directory.
   5 //
   6 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
   7 // Use of this source code is governed by a BSD-style license that can be
   8 // found in the LICENSE file. See the AUTHORS file for names of contributors.
   9 //
  10 // Logger implementation that can be shared by all environments
  11 // where enough posix functionality is available.
  12
  13 #pragma once
  14 #include <algorithm>
  15 #include <stdio.h>
  16 #include "port/sys_time.h"
  17 #include <time.h>
  18 #include <fcntl.h>
  19
  20 #ifdef OS_LINUX
  21 #ifndef FALLOC_FL_KEEP_SIZE
  22 #include <linux/falloc.h>
  23 #endif
  24 #endif
  25
  26 #include <atomic>
  27 #include "monitoring/iostats_context_imp.h"
  28 #include "rocksdb/env.h"
  29 #include "util/sync_point.h"
  30
  31 namespace rocksdb {
  32
  33 class PosixLogger : public Logger {
  34  private:
  35   FILE* file_;
  36   uint64_t (*gettid_)();  // Return the thread id for the current thread
  37   std::atomic_size_t log_size_;
  38   int fd_;
  39   const static uint64_t flush_every_seconds_ = 5;
  40   std::atomic_uint_fast64_t last_flush_micros_;
  41   Env* env_;
  42   bool flush_pending_;
  43  public:
  44   PosixLogger(FILE* f, uint64_t (*gettid)(), Env* env,
  45               const InfoLogLevel log_level = InfoLogLevel::ERROR_LEVEL)
  46       : Logger(log_level),
  47         file_(f),
  48         gettid_(gettid),
  49         log_size_(0),
  50         fd_(fileno(f)),
  51         last_flush_micros_(0),
  52         env_(env),
  53         flush_pending_(false) {}
  54   virtual ~PosixLogger() {
  55     fclose(file_);
  56   }
  57   virtual void Flush() override {
  58     TEST_SYNC_POINT("PosixLogger::Flush:Begin1");
  59     TEST_SYNC_POINT("PosixLogger::Flush:Begin2");
  60     if (flush_pending_) {
  61       flush_pending_ = false;
  62       fflush(file_);
  63     }
  64     last_flush_micros_ = env_->NowMicros();
  65   }
  66
  67   using Logger::Logv;
  68   virtual void Logv(const char* format, va_list ap) override {
  69     IOSTATS_TIMER_GUARD(logger_nanos);
  70
  71     const uint64_t thread_id = (*gettid_)();
  72
  73     // We try twice: the first time with a fixed-size stack allocated buffer,
  74     // and the second time with a much larger dynamically allocated buffer.
  75     char buffer[500];
  76     for (int iter = 0; iter < 2; iter++) {
  77       char* base;
  78       int bufsize;
  79       if (iter == 0) {
  80         bufsize = sizeof(buffer);
  81         base = buffer;
  82       } else {
  83         bufsize = 65536;
  84         base = new char[bufsize];
  85       }
  86       char* p = base;
  87       char* limit = base + bufsize;
  88
  89       struct timeval now_tv;
  90       gettimeofday(&now_tv, nullptr);
  91       const time_t seconds = now_tv.tv_sec;
  92       struct tm t;
  93       localtime_r(&seconds, &t);
  94       p += snprintf(p, limit - p,
  95                     "%04d/%02d/%02d-%02d:%02d:%02d.%06d %llx ",
  96                     t.tm_year + 1900,
  97                     t.tm_mon + 1,
  98                     t.tm_mday,
  99                     t.tm_hour,
 100                     t.tm_min,
 101                     t.tm_sec,
 102                     static_cast<int>(now_tv.tv_usec),
 103                     static_cast<long long unsigned int>(thread_id));
 104
 105       // Print the message
 106       if (p < limit) {
 107         va_list backup_ap;
 108         va_copy(backup_ap, ap);
 109         p += vsnprintf(p, limit - p, format, backup_ap);
 110         va_end(backup_ap);
 111       }
 112
 113       // Truncate to available space if necessary
 114       if (p >= limit) {
 115         if (iter == 0) {
 116           continue;       // Try again with larger buffer
 117         } else {
 118           p = limit - 1;
 119         }
 120       }
 121
 122       // Add newline if necessary
 123       if (p == base || p[-1] != '\n') {
 124         *p++ = '\n';
 125       }
 126
 127       assert(p <= limit);
 128       const size_t write_size = p - base;
 129
 130 #ifdef ROCKSDB_FALLOCATE_PRESENT
 131       const int kDebugLogChunkSize = 128 * 1024;
 132
 133       // If this write would cross a boundary of kDebugLogChunkSize
 134       // space, pre-allocate more space to avoid overly large
 135       // allocations from filesystem allocsize options.
 136       const size_t log_size = log_size_;
 137       const size_t last_allocation_chunk =
 138         ((kDebugLogChunkSize - 1 + log_size) / kDebugLogChunkSize);
 139       const size_t desired_allocation_chunk =
 140         ((kDebugLogChunkSize - 1 + log_size + write_size) /
 141            kDebugLogChunkSize);
 142       if (last_allocation_chunk != desired_allocation_chunk) {
 143         fallocate(
 144             fd_, FALLOC_FL_KEEP_SIZE, 0,
 145             static_cast<off_t>(desired_allocation_chunk * kDebugLogChunkSize));
 146       }
 147 #endif
 148
 149       size_t sz = fwrite(base, 1, write_size, file_);
 150       flush_pending_ = true;
 151       assert(sz == write_size);
 152       if (sz > 0) {
 153         log_size_ += write_size;
 154       }
 155       uint64_t now_micros = static_cast<uint64_t>(now_tv.tv_sec) * 1000000 +
 156         now_tv.tv_usec;
 157       if (now_micros - last_flush_micros_ >= flush_every_seconds_ * 1000000) {
 158         Flush();
 159       }
 160       if (base != buffer) {
 161         delete[] base;
 162       }
 163       break;
 164     }
 165   }
 166   size_t GetLogFileSize() const override { return log_size_; }
 167 };
 168
 169 }  // namespace rocksdb