streams.cc 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389
  1. // Copyright (c) 2011 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. // Streams classes.
  5. //
  6. // These memory-resident streams are used for serializing data into a sequential
  7. // region of memory.
  8. //
  9. // Streams are divided into SourceStreams for reading and SinkStreams for
  10. // writing. Streams are aggregated into Sets which allows several streams to be
  11. // used at once. Example: we can write A1, B1, A2, B2 but achieve the memory
  12. // layout A1 A2 B1 B2 by writing 'A's to one stream and 'B's to another.
  13. //
  14. // The aggregated streams are important to Courgette's compression efficiency,
  15. // we use it to cluster similar kinds of data which helps to generate longer
  16. // common subsequences and repeated sequences.
  17. #include "courgette/streams.h"
  18. #include <memory.h>
  19. #include <stddef.h>
  20. #include <stdint.h>
  21. #include "base/logging.h"
  22. namespace courgette {
  23. // Update this version number if the serialization format of a StreamSet
  24. // changes.
  25. static const unsigned int kStreamsSerializationFormatVersion = 20090218;
  26. //
  27. // This is a cut down Varint implementation, implementing only what we use for
  28. // streams.
  29. //
  30. class Varint {
  31. public:
  32. // Maximum lengths of varint encoding of uint32_t
  33. static const int kMax32 = 5;
  34. // Parses a Varint32 encoded value from |source| and stores it in |output|,
  35. // and returns a pointer to the following byte. Returns nullptr if a valid
  36. // varint value was not found before |limit|.
  37. static const uint8_t* Parse32WithLimit(const uint8_t* source,
  38. const uint8_t* limit,
  39. uint32_t* output);
  40. // Writes the Varint32 encoded representation of |value| to buffer
  41. // |destination|. |destination| must have sufficient length to hold kMax32
  42. // bytes. Returns a pointer to the byte just past the last encoded byte.
  43. static uint8_t* Encode32(uint8_t* destination, uint32_t value);
  44. };
  45. // Parses a Varint32 encoded unsigned number from |source|. The Varint32
  46. // encoding is a little-endian sequence of bytes containing base-128 digits,
  47. // with the high order bit set to indicate if there are more digits.
  48. //
  49. // For each byte, we mask out the digit and 'or' it into the right place in the
  50. // result.
  51. //
  52. // The digit loop is unrolled for performance. It usually exits after the first
  53. // one or two digits.
  54. const uint8_t* Varint::Parse32WithLimit(const uint8_t* source,
  55. const uint8_t* limit,
  56. uint32_t* output) {
  57. uint32_t digit, result;
  58. if (source >= limit)
  59. return nullptr;
  60. digit = *(source++);
  61. result = digit & 127;
  62. if (digit < 128) {
  63. *output = result;
  64. return source;
  65. }
  66. if (source >= limit)
  67. return nullptr;
  68. digit = *(source++);
  69. result |= (digit & 127) << 7;
  70. if (digit < 128) {
  71. *output = result;
  72. return source;
  73. }
  74. if (source >= limit)
  75. return nullptr;
  76. digit = *(source++);
  77. result |= (digit & 127) << 14;
  78. if (digit < 128) {
  79. *output = result;
  80. return source;
  81. }
  82. if (source >= limit)
  83. return nullptr;
  84. digit = *(source++);
  85. result |= (digit & 127) << 21;
  86. if (digit < 128) {
  87. *output = result;
  88. return source;
  89. }
  90. if (source >= limit)
  91. return nullptr;
  92. digit = *(source++);
  93. result |= (digit & 127) << 28;
  94. if (digit < 128) {
  95. *output = result;
  96. return source;
  97. }
  98. return nullptr; // Value is too long to be a Varint32.
  99. }
  100. // Write the base-128 digits in little-endian order. All except the last digit
  101. // have the high bit set to indicate more digits.
  102. inline uint8_t* Varint::Encode32(uint8_t* destination, uint32_t value) {
  103. while (value >= 128) {
  104. *(destination++) = static_cast<uint8_t>(value) | 128;
  105. value = value >> 7;
  106. }
  107. *(destination++) = static_cast<uint8_t>(value);
  108. return destination;
  109. }
  110. void SourceStream::Init(const SinkStream& sink) {
  111. Init(sink.Buffer(), sink.Length());
  112. }
  113. bool SourceStream::Read(void* destination, size_t count) {
  114. if (current_ + count > end_)
  115. return false;
  116. memcpy(destination, current_, count);
  117. current_ += count;
  118. return true;
  119. }
  120. bool SourceStream::ReadVarint32(uint32_t* output_value) {
  121. const uint8_t* after = Varint::Parse32WithLimit(current_, end_, output_value);
  122. if (!after)
  123. return false;
  124. current_ = after;
  125. return true;
  126. }
  127. bool SourceStream::ReadVarint32Signed(int32_t* output_value) {
  128. // Signed numbers are encoded as unsigned numbers so that numbers nearer zero
  129. // have shorter varint encoding.
  130. // 0000xxxx encoded as 000xxxx0.
  131. // 1111xxxx encoded as 000yyyy1 where yyyy is complement of xxxx.
  132. uint32_t unsigned_value;
  133. if (!ReadVarint32(&unsigned_value))
  134. return false;
  135. if (unsigned_value & 1)
  136. *output_value = ~static_cast<int32_t>(unsigned_value >> 1);
  137. else
  138. *output_value = (unsigned_value >> 1);
  139. return true;
  140. }
  141. bool SourceStream::ShareSubstream(size_t offset, size_t length,
  142. SourceStream* substream) {
  143. if (offset > Remaining())
  144. return false;
  145. if (length > Remaining() - offset)
  146. return false;
  147. substream->Init(current_ + offset, length);
  148. return true;
  149. }
  150. bool SourceStream::ReadSubstream(size_t length, SourceStream* substream) {
  151. if (!ShareSubstream(0, length, substream))
  152. return false;
  153. current_ += length;
  154. return true;
  155. }
  156. bool SourceStream::Skip(size_t byte_count) {
  157. if (current_ + byte_count > end_)
  158. return false;
  159. current_ += byte_count;
  160. return true;
  161. }
  162. CheckBool SinkStream::Write(const void* data, size_t byte_count) {
  163. return buffer_.append(static_cast<const char*>(data), byte_count);
  164. }
  165. CheckBool SinkStream::WriteVarint32(uint32_t value) {
  166. uint8_t buffer[Varint::kMax32];
  167. uint8_t* end = Varint::Encode32(buffer, value);
  168. return Write(buffer, end - buffer);
  169. }
  170. CheckBool SinkStream::WriteVarint32Signed(int32_t value) {
  171. // Encode signed numbers so that numbers nearer zero have shorter
  172. // varint encoding.
  173. // 0000xxxx encoded as 000xxxx0.
  174. // 1111xxxx encoded as 000yyyy1 where yyyy is complement of xxxx.
  175. bool ret;
  176. if (value < 0)
  177. ret = WriteVarint32(~value * 2 + 1);
  178. else
  179. ret = WriteVarint32(value * 2);
  180. return ret;
  181. }
  182. CheckBool SinkStream::WriteSizeVarint32(size_t value) {
  183. uint32_t narrowed_value = static_cast<uint32_t>(value);
  184. // On 32-bit, the compiler should figure out this test always fails.
  185. LOG_ASSERT(value == narrowed_value);
  186. return WriteVarint32(narrowed_value);
  187. }
  188. CheckBool SinkStream::Append(SinkStream* other) {
  189. bool ret = Write(other->buffer_.data(), other->buffer_.size());
  190. if (ret)
  191. other->Retire();
  192. return ret;
  193. }
  194. void SinkStream::Retire() {
  195. buffer_.clear();
  196. }
  197. ////////////////////////////////////////////////////////////////////////////////
  198. SourceStreamSet::SourceStreamSet()
  199. : count_(kMaxStreams) {
  200. }
  201. SourceStreamSet::~SourceStreamSet() = default;
  202. // Initializes from |source|.
  203. // The stream set for N streams is serialized as a header
  204. // <version><N><length1><length2>...<lengthN>
  205. // followed by the stream contents
  206. // <bytes1><bytes2>...<bytesN>
  207. //
  208. bool SourceStreamSet::Init(const void* source, size_t byte_count) {
  209. const uint8_t* start = static_cast<const uint8_t*>(source);
  210. const uint8_t* end = start + byte_count;
  211. unsigned int version;
  212. const uint8_t* finger = Varint::Parse32WithLimit(start, end, &version);
  213. if (finger == nullptr)
  214. return false;
  215. if (version != kStreamsSerializationFormatVersion)
  216. return false;
  217. unsigned int count;
  218. finger = Varint::Parse32WithLimit(finger, end, &count);
  219. if (finger == nullptr)
  220. return false;
  221. if (count > kMaxStreams)
  222. return false;
  223. count_ = count;
  224. unsigned int lengths[kMaxStreams];
  225. size_t accumulated_length = 0;
  226. for (size_t i = 0; i < count_; ++i) {
  227. finger = Varint::Parse32WithLimit(finger, end, &lengths[i]);
  228. if (finger == nullptr)
  229. return false;
  230. accumulated_length += lengths[i];
  231. }
  232. // Remaining bytes should add up to sum of lengths.
  233. if (static_cast<size_t>(end - finger) != accumulated_length)
  234. return false;
  235. accumulated_length = finger - start;
  236. for (size_t i = 0; i < count_; ++i) {
  237. stream(i)->Init(start + accumulated_length, lengths[i]);
  238. accumulated_length += lengths[i];
  239. }
  240. return true;
  241. }
  242. bool SourceStreamSet::Init(SourceStream* source) {
  243. // TODO(sra): consume the rest of |source|.
  244. return Init(source->Buffer(), source->Remaining());
  245. }
  246. bool SourceStreamSet::ReadSet(SourceStreamSet* set) {
  247. uint32_t stream_count = 0;
  248. SourceStream* control_stream = this->stream(0);
  249. if (!control_stream->ReadVarint32(&stream_count))
  250. return false;
  251. uint32_t lengths[kMaxStreams] = {}; // i.e. all zero.
  252. for (size_t i = 0; i < stream_count; ++i) {
  253. if (!control_stream->ReadVarint32(&lengths[i]))
  254. return false;
  255. }
  256. for (size_t i = 0; i < stream_count; ++i) {
  257. if (!this->stream(i)->ReadSubstream(lengths[i], set->stream(i)))
  258. return false;
  259. }
  260. return true;
  261. }
  262. bool SourceStreamSet::Empty() const {
  263. for (size_t i = 0; i < count_; ++i) {
  264. if (streams_[i].Remaining() != 0)
  265. return false;
  266. }
  267. return true;
  268. }
  269. ////////////////////////////////////////////////////////////////////////////////
  270. SinkStreamSet::SinkStreamSet()
  271. : count_(kMaxStreams) {
  272. }
  273. SinkStreamSet::~SinkStreamSet() = default;
  274. void SinkStreamSet::Init(size_t stream_index_limit) {
  275. count_ = stream_index_limit;
  276. }
  277. // The header for a stream set for N streams is serialized as
  278. // <version><N><length1><length2>...<lengthN>
  279. CheckBool SinkStreamSet::CopyHeaderTo(SinkStream* header) {
  280. bool ret = header->WriteVarint32(kStreamsSerializationFormatVersion);
  281. if (ret) {
  282. ret = header->WriteSizeVarint32(count_);
  283. for (size_t i = 0; ret && i < count_; ++i) {
  284. ret = header->WriteSizeVarint32(stream(i)->Length());
  285. }
  286. }
  287. return ret;
  288. }
  289. // Writes |this| to |combined_stream|. See SourceStreamSet::Init for the layout
  290. // of the stream metadata and contents.
  291. CheckBool SinkStreamSet::CopyTo(SinkStream *combined_stream) {
  292. SinkStream header;
  293. bool ret = CopyHeaderTo(&header);
  294. if (!ret)
  295. return ret;
  296. // Reserve the correct amount of storage.
  297. size_t length = header.Length();
  298. for (size_t i = 0; i < count_; ++i) {
  299. length += stream(i)->Length();
  300. }
  301. ret = combined_stream->Reserve(length);
  302. if (ret) {
  303. ret = combined_stream->Append(&header);
  304. for (size_t i = 0; ret && i < count_; ++i) {
  305. ret = combined_stream->Append(stream(i));
  306. }
  307. }
  308. return ret;
  309. }
  310. CheckBool SinkStreamSet::WriteSet(SinkStreamSet* set) {
  311. uint32_t lengths[kMaxStreams];
  312. // 'stream_count' includes all non-empty streams and all empty stream numbered
  313. // lower than a non-empty stream.
  314. size_t stream_count = 0;
  315. for (size_t i = 0; i < kMaxStreams; ++i) {
  316. SinkStream* stream = set->stream(i);
  317. lengths[i] = static_cast<uint32_t>(stream->Length());
  318. if (lengths[i] > 0)
  319. stream_count = i + 1;
  320. }
  321. SinkStream* control_stream = this->stream(0);
  322. bool ret = control_stream->WriteSizeVarint32(stream_count);
  323. for (size_t i = 0; ret && i < stream_count; ++i) {
  324. ret = control_stream->WriteSizeVarint32(lengths[i]);
  325. }
  326. for (size_t i = 0; ret && i < stream_count; ++i) {
  327. ret = this->stream(i)->Append(set->stream(i));
  328. }
  329. return ret;
  330. }
  331. } // namespace