iOS/Pods/Realm/include/core/realm/impl/input_stream.hpp

   1 /*************************************************************************
   2  *
   3  * Copyright 2016 Realm Inc.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   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, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  **************************************************************************/
  18
  19 #ifndef REALM_IMPL_INPUT_STREAM_HPP
  20 #define REALM_IMPL_INPUT_STREAM_HPP
  21
  22 #include <algorithm>
  23
  24 #include <realm/binary_data.hpp>
  25 #include <realm/impl/cont_transact_hist.hpp>
  26 #include <realm/util/buffer.hpp>
  27
  28
  29 namespace realm {
  30 namespace _impl {
  31
  32
  33 class InputStream {
  34 public:
  35     /// Read bytes from this input stream and place them in the specified
  36     /// buffer. The returned value is the actual number of bytes that were read,
  37     /// and this is some number `n` such that `n <= min(size, m)` where `m` is
  38     /// the number of bytes that could have been read from this stream before
  39     /// reaching its end. Also, `n` cannot be zero unless `m` or `size` is
  40     /// zero. The intention is that `size` should be non-zero, a the return
  41     /// value used as the end-of-input indicator.
  42     ///
  43     /// Implementations are only allowed to block (put the calling thread to
  44     /// sleep) up until the point in time where the first byte can be made
  45     /// availble.
  46     virtual size_t read(char* buffer, size_t size) = 0;
  47
  48     virtual ~InputStream() noexcept
  49     {
  50     }
  51 };
  52
  53
  54 class SimpleInputStream : public InputStream {
  55 public:
  56     SimpleInputStream(const char* data, size_t size) noexcept
  57         : m_ptr(data)
  58         , m_end(data + size)
  59     {
  60     }
  61     size_t read(char* buffer, size_t size) override
  62     {
  63         size_t n = std::min(size, size_t(m_end - m_ptr));
  64         const char* begin = m_ptr;
  65         m_ptr += n;
  66         realm::safe_copy_n(begin, n, buffer);
  67         return n;
  68     }
  69
  70 private:
  71     const char* m_ptr;
  72     const char* const m_end;
  73 };
  74
  75
  76 class NoCopyInputStream {
  77 public:
  78     /// \return if any bytes was read.
  79     /// A value of false indicates end-of-input.
  80     /// If return value is true, \a begin and \a end are
  81     /// updated to reflect the start and limit of a
  82     /// contiguous memory chunk.
  83     virtual bool next_block(const char*& begin, const char*& end) = 0;
  84
  85     virtual ~NoCopyInputStream() noexcept
  86     {
  87     }
  88 };
  89
  90
  91 class NoCopyInputStreamAdaptor : public NoCopyInputStream {
  92 public:
  93     NoCopyInputStreamAdaptor(InputStream& in, char* buffer, size_t buffer_size) noexcept
  94         : m_in(in)
  95         , m_buffer(buffer)
  96         , m_buffer_size(buffer_size)
  97     {
  98     }
  99     bool next_block(const char*& begin, const char*& end) override
 100     {
 101         size_t n = m_in.read(m_buffer, m_buffer_size);
 102         begin = m_buffer;
 103         end = m_buffer + n;
 104         return n;
 105     }
 106
 107 private:
 108     InputStream& m_in;
 109     char* m_buffer;
 110     size_t m_buffer_size;
 111 };
 112
 113
 114 class SimpleNoCopyInputStream : public NoCopyInputStream {
 115 public:
 116     SimpleNoCopyInputStream(const char* data, size_t size)
 117         : m_data(data)
 118         , m_size(size)
 119     {
 120     }
 121
 122     bool next_block(const char*& begin, const char*& end) override
 123     {
 124         if (m_size == 0)
 125             return 0;
 126         size_t size = m_size;
 127         begin = m_data;
 128         end = m_data + size;
 129         m_size = 0;
 130         return size;
 131     }
 132
 133 private:
 134     const char* m_data;
 135     size_t m_size;
 136 };
 137
 138 class MultiLogNoCopyInputStream : public NoCopyInputStream {
 139 public:
 140     MultiLogNoCopyInputStream(const BinaryData* logs_begin, const BinaryData* logs_end)
 141         : m_logs_begin(logs_begin)
 142         , m_logs_end(logs_end)
 143     {
 144         if (m_logs_begin != m_logs_end)
 145             m_curr_buf_remaining_size = m_logs_begin->size();
 146     }
 147
 148     size_t read(char* buffer, size_t size)
 149     {
 150         if (m_logs_begin == m_logs_end)
 151             return 0;
 152         for (;;) {
 153             if (m_curr_buf_remaining_size > 0) {
 154                 size_t offset = m_logs_begin->size() - m_curr_buf_remaining_size;
 155                 const char* data = m_logs_begin->data() + offset;
 156                 size_t size_2 = std::min(m_curr_buf_remaining_size, size);
 157                 m_curr_buf_remaining_size -= size_2;
 158                 // FIXME: Eliminate the need for copying by changing the API of
 159                 // Replication::InputStream such that blocks can be handed over
 160                 // without copying. This is a straight forward change, but the
 161                 // result is going to be more complicated and less conventional.
 162                 realm::safe_copy_n(data, size_2, buffer);
 163                 return size_2;
 164             }
 165
 166             ++m_logs_begin;
 167             if (m_logs_begin == m_logs_end)
 168                 return 0;
 169             m_curr_buf_remaining_size = m_logs_begin->size();
 170         }
 171     }
 172
 173     bool next_block(const char*& begin, const char*& end) override
 174     {
 175         while (m_logs_begin < m_logs_end) {
 176             size_t result = m_logs_begin->size();
 177             const char* data = m_logs_begin->data();
 178             m_logs_begin++;
 179             if (result == 0)
 180                 continue; // skip empty blocks
 181             begin = data;
 182             end = data + result;
 183             return result;
 184         }
 185         return 0;
 186     }
 187
 188 private:
 189     const BinaryData* m_logs_begin;
 190     const BinaryData* m_logs_end;
 191     size_t m_curr_buf_remaining_size;
 192 };
 193
 194
 195 class ChangesetInputStream : public NoCopyInputStream {
 196 public:
 197     using version_type = History::version_type;
 198     static constexpr unsigned NB_BUFFERS = 8;
 199
 200     ChangesetInputStream(History& hist, version_type begin_version, version_type end_version)
 201         : m_history(hist)
 202         , m_begin_version(begin_version)
 203         , m_end_version(end_version)
 204     {
 205         get_changeset();
 206     }
 207
 208     bool next_block(const char*& begin, const char*& end) override
 209     {
 210         while (m_valid) {
 211             BinaryData actual = m_changesets_begin->get_next();
 212
 213             if (actual.size() > 0) {
 214                 begin = actual.data();
 215                 end = actual.data() + actual.size();
 216                 return true;
 217             }
 218
 219             m_changesets_begin++;
 220
 221             if (REALM_UNLIKELY(m_changesets_begin == m_changesets_end)) {
 222                 get_changeset();
 223             }
 224         }
 225         return false; // End of input
 226     }
 227
 228 private:
 229     History& m_history;
 230     version_type m_begin_version, m_end_version;
 231     BinaryIterator m_changesets[NB_BUFFERS]; // Buffer
 232     BinaryIterator* m_changesets_begin = nullptr;
 233     BinaryIterator* m_changesets_end = nullptr;
 234     bool m_valid;
 235
 236     void get_changeset()
 237     {
 238         auto versions_to_get = m_end_version - m_begin_version;
 239         m_valid = versions_to_get > 0;
 240         if (m_valid) {
 241             if (versions_to_get > NB_BUFFERS)
 242                 versions_to_get = NB_BUFFERS;
 243             version_type end_version = m_begin_version + versions_to_get;
 244             m_history.get_changesets(m_begin_version, end_version, m_changesets);
 245             m_begin_version = end_version;
 246             m_changesets_begin = m_changesets;
 247             m_changesets_end = m_changesets_begin + versions_to_get;
 248         }
 249     }
 250 };
 251
 252 } // namespace _impl
 253 } // namespace realm
 254
 255 #endif // REALM_IMPL_INPUT_STREAM_HPP