1 //  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
2 //  This source code is licensed under both the GPLv2 (found in the
3 //  COPYING file in the root directory) and Apache 2.0 License
4 //  (found in the LICENSE.Apache file in the root directory).
5 #pragma once
6 
7 #ifndef ROCKSDB_LITE
8 
9 #include <cstdint>
10 #include <memory>
11 #include <string>
12 
13 #include "rocksdb/env.h"
14 #include "rocksdb/slice.h"
15 #include "rocksdb/statistics.h"
16 #include "rocksdb/status.h"
17 #include "rocksdb/types.h"
18 #include "utilities/blob_db/blob_log_format.h"
19 
20 namespace ROCKSDB_NAMESPACE {
21 
22 class WritableFileWriter;
23 
24 namespace blob_db {
25 
26 /**
27  * Writer is the blob log stream writer. It provides an append-only
28  * abstraction for writing blob data.
29  *
30  *
31  * Look at blob_db_format.h to see the details of the record formats.
32  */
33 
34 class Writer {
35  public:
36   // Create a writer that will append data to "*dest".
37   // "*dest" must be initially empty.
38   // "*dest" must remain live while this Writer is in use.
39   Writer(std::unique_ptr<WritableFileWriter>&& dest, Env* env,
40          Statistics* statistics, uint64_t log_number, uint64_t bpsync,
41          bool use_fsync, uint64_t boffset = 0);
42   // No copying allowed
43   Writer(const Writer&) = delete;
44   Writer& operator=(const Writer&) = delete;
45 
46   ~Writer() = default;
47 
48   static void ConstructBlobHeader(std::string* buf, const Slice& key,
49                                   const Slice& val, uint64_t expiration);
50 
51   Status AddRecord(const Slice& key, const Slice& val, uint64_t* key_offset,
52                    uint64_t* blob_offset);
53 
54   Status AddRecord(const Slice& key, const Slice& val, uint64_t expiration,
55                    uint64_t* key_offset, uint64_t* blob_offset);
56 
57   Status EmitPhysicalRecord(const std::string& headerbuf, const Slice& key,
58                             const Slice& val, uint64_t* key_offset,
59                             uint64_t* blob_offset);
60 
61   Status AppendFooter(BlobLogFooter& footer);
62 
63   Status WriteHeader(BlobLogHeader& header);
64 
file()65   WritableFileWriter* file() { return dest_.get(); }
66 
file()67   const WritableFileWriter* file() const { return dest_.get(); }
68 
get_log_number()69   uint64_t get_log_number() const { return log_number_; }
70 
ShouldSync()71   bool ShouldSync() const { return block_offset_ > next_sync_offset_; }
72 
73   Status Sync();
74 
ResetSyncPointer()75   void ResetSyncPointer() { next_sync_offset_ += bytes_per_sync_; }
76 
77  private:
78   std::unique_ptr<WritableFileWriter> dest_;
79   Env* env_;
80   Statistics* statistics_;
81   uint64_t log_number_;
82   uint64_t block_offset_;  // Current offset in block
83   uint64_t bytes_per_sync_;
84   uint64_t next_sync_offset_;
85   bool use_fsync_;
86 
87  public:
88   enum ElemType { kEtNone, kEtFileHdr, kEtRecord, kEtFileFooter };
89   ElemType last_elem_type_;
90 };
91 
92 }  // namespace blob_db
93 }  // namespace ROCKSDB_NAMESPACE
94 #endif  // ROCKSDB_LITE
95