FIFOBuffer.h 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556
  1. //
  2. // FIFOBuffer.h
  3. //
  4. // Library: Foundation
  5. // Package: Core
  6. // Module: FIFOBuffer
  7. //
  8. // Definition of the FIFOBuffer class.
  9. //
  10. // Copyright (c) 2006, Applied Informatics Software Engineering GmbH.
  11. // and Contributors.
  12. //
  13. // SPDX-License-Identifier: BSL-1.0
  14. //
  15. #ifndef Foundation_FIFOBuffer_INCLUDED
  16. #define Foundation_FIFOBuffer_INCLUDED
  17. #include "Poco/Foundation.h"
  18. #include "Poco/Exception.h"
  19. #include "Poco/Buffer.h"
  20. #include "Poco/BasicEvent.h"
  21. #include "Poco/Mutex.h"
  22. #include "Poco/Format.h"
  23. namespace Poco {
  24. template <class T>
  25. class BasicFIFOBuffer
  26. /// A simple buffer class with support for re-entrant,
  27. /// FIFO-style read/write operations, as well as (optional)
  28. /// empty/non-empty/full (i.e. writable/readable) transition
  29. /// notifications. Buffer can be flagged with end-of-file and
  30. /// error flags, which renders it un-readable/writable.
  31. ///
  32. /// Critical portions of code are protected by a recursive mutex.
  33. /// However, to achieve thread-safety in cases where multiple
  34. /// member function calls are involved and have to be atomic,
  35. /// the mutex must be locked externally.
  36. ///
  37. /// Buffer size, as well as amount of unread data and
  38. /// available space introspections are supported as well.
  39. ///
  40. /// This class is useful anywhere where a FIFO functionality
  41. /// is needed.
  42. {
  43. public:
  44. typedef T Type;
  45. mutable Poco::BasicEvent<bool> writable;
  46. /// Event indicating "writability" of the buffer,
  47. /// triggered as follows:
  48. ///
  49. /// * when buffer transitions from non-full to full,
  50. /// Writable event observers are notified, with
  51. /// false value as the argument
  52. ///
  53. /// * when buffer transitions from full to non-full,
  54. /// Writable event observers are notified, with
  55. /// true value as the argument
  56. mutable Poco::BasicEvent<bool> readable;
  57. /// Event indicating "readability" of the buffer,
  58. /// triggered as follows:
  59. ///
  60. /// * when buffer transitions from non-empty to empty,
  61. /// Readable event observers are notified, with false
  62. /// value as the argument
  63. ///
  64. /// * when FIFOBuffer transitions from empty to non-empty,
  65. /// Readable event observers are notified, with true value
  66. /// as the argument
  67. BasicFIFOBuffer(std::size_t size, bool notify = false):
  68. _buffer(size),
  69. _begin(0),
  70. _used(0),
  71. _notify(notify),
  72. _eof(false),
  73. _error(false)
  74. /// Creates the FIFOBuffer.
  75. {
  76. }
  77. BasicFIFOBuffer(T* pBuffer, std::size_t size, bool notify = false):
  78. _buffer(pBuffer, size),
  79. _begin(0),
  80. _used(0),
  81. _notify(notify),
  82. _eof(false),
  83. _error(false)
  84. /// Creates the FIFOBuffer.
  85. {
  86. }
  87. BasicFIFOBuffer(const T* pBuffer, std::size_t size, bool notify = false):
  88. _buffer(pBuffer, size),
  89. _begin(0),
  90. _used(size),
  91. _notify(notify),
  92. _eof(false),
  93. _error(false)
  94. /// Creates the FIFOBuffer.
  95. {
  96. }
  97. ~BasicFIFOBuffer()
  98. /// Destroys the FIFOBuffer.
  99. {
  100. }
  101. void resize(std::size_t newSize, bool preserveContent = true)
  102. /// Resizes the buffer. If preserveContent is true,
  103. /// the content of the old buffer is preserved.
  104. /// New size can be larger or smaller than
  105. /// the current size, but it must not be 0.
  106. /// Additionally, if the new length is smaller
  107. /// than currently used length and preserveContent
  108. /// is true, InvalidAccessException is thrown.
  109. {
  110. Mutex::ScopedLock lock(_mutex);
  111. if (preserveContent && (newSize < _used))
  112. throw InvalidAccessException("Can not resize FIFO without data loss.");
  113. std::size_t usedBefore = _used;
  114. _buffer.resize(newSize, preserveContent);
  115. if (!preserveContent) _used = 0;
  116. if (_notify) notify(usedBefore);
  117. }
  118. std::size_t peek(T* pBuffer, std::size_t length) const
  119. /// Peeks into the data currently in the FIFO
  120. /// without actually extracting it.
  121. /// If length is zero, the return is immediate.
  122. /// If length is greater than used length,
  123. /// it is substituted with the the current FIFO
  124. /// used length.
  125. ///
  126. /// Returns the number of elements copied in the
  127. /// supplied buffer.
  128. {
  129. if (0 == length) return 0;
  130. Mutex::ScopedLock lock(_mutex);
  131. if (!isReadable()) return 0;
  132. if (length > _used) length = _used;
  133. std::memcpy(pBuffer, _buffer.begin() + _begin, length * sizeof(T));
  134. return length;
  135. }
  136. std::size_t peek(Poco::Buffer<T>& buffer, std::size_t length = 0) const
  137. /// Peeks into the data currently in the FIFO
  138. /// without actually extracting it.
  139. /// Resizes the supplied buffer to the size of
  140. /// data written to it. If length is not
  141. /// supplied by the caller or is greater than length
  142. /// of currently used data, the current FIFO used
  143. /// data length is substituted for it.
  144. ///
  145. /// Returns the number of elements copied in the
  146. /// supplied buffer.
  147. {
  148. Mutex::ScopedLock lock(_mutex);
  149. if (!isReadable()) return 0;
  150. if (0 == length || length > _used) length = _used;
  151. buffer.resize(length);
  152. return peek(buffer.begin(), length);
  153. }
  154. std::size_t read(T* pBuffer, std::size_t length)
  155. /// Copies the data currently in the FIFO
  156. /// into the supplied buffer, which must be
  157. /// preallocated to at least the length size
  158. /// before calling this function.
  159. ///
  160. /// Returns the size of the copied data.
  161. {
  162. if (0 == length) return 0;
  163. Mutex::ScopedLock lock(_mutex);
  164. if (!isReadable()) return 0;
  165. std::size_t usedBefore = _used;
  166. std::size_t readLen = peek(pBuffer, length);
  167. poco_assert (_used >= readLen);
  168. _used -= readLen;
  169. if (0 == _used) _begin = 0;
  170. else _begin += length;
  171. if (_notify) notify(usedBefore);
  172. return readLen;
  173. }
  174. std::size_t read(Poco::Buffer<T>& buffer, std::size_t length = 0)
  175. /// Copies the data currently in the FIFO
  176. /// into the supplied buffer.
  177. /// Resizes the supplied buffer to the size of
  178. /// data written to it.
  179. ///
  180. /// Returns the size of the copied data.
  181. {
  182. Mutex::ScopedLock lock(_mutex);
  183. if (!isReadable()) return 0;
  184. std::size_t usedBefore = _used;
  185. std::size_t readLen = peek(buffer, length);
  186. poco_assert (_used >= readLen);
  187. _used -= readLen;
  188. if (0 == _used) _begin = 0;
  189. else _begin += length;
  190. if (_notify) notify(usedBefore);
  191. return readLen;
  192. }
  193. std::size_t write(const T* pBuffer, std::size_t length)
  194. /// Writes data from supplied buffer to the FIFO buffer.
  195. /// If there is no sufficient space for the whole
  196. /// buffer to be written, data up to available
  197. /// length is written.
  198. /// The length of data to be written is determined from the
  199. /// length argument. Function does nothing and returns zero
  200. /// if length argument is equal to zero.
  201. ///
  202. /// Returns the length of data written.
  203. {
  204. if (0 == length) return 0;
  205. Mutex::ScopedLock lock(_mutex);
  206. if (!isWritable()) return 0;
  207. if (_buffer.size() - (_begin + _used) < length)
  208. {
  209. std::memmove(_buffer.begin(), begin(), _used * sizeof(T));
  210. _begin = 0;
  211. }
  212. std::size_t usedBefore = _used;
  213. std::size_t available = _buffer.size() - _used - _begin;
  214. std::size_t len = length > available ? available : length;
  215. std::memcpy(begin() + _used, pBuffer, len * sizeof(T));
  216. _used += len;
  217. poco_assert (_used <= _buffer.size());
  218. if (_notify) notify(usedBefore);
  219. return len;
  220. }
  221. std::size_t write(const Buffer<T>& buffer, std::size_t length = 0)
  222. /// Writes data from supplied buffer to the FIFO buffer.
  223. /// If there is no sufficient space for the whole
  224. /// buffer to be written, data up to available
  225. /// length is written.
  226. /// The length of data to be written is determined from the
  227. /// length argument or buffer size (when length argument is
  228. /// default zero or greater than buffer size).
  229. ///
  230. /// Returns the length of data written.
  231. {
  232. if (length == 0 || length > buffer.size())
  233. length = buffer.size();
  234. return write(buffer.begin(), length);
  235. }
  236. std::size_t size() const
  237. /// Returns the size of the buffer.
  238. {
  239. return _buffer.size();
  240. }
  241. std::size_t used() const
  242. /// Returns the size of the used portion of the buffer.
  243. {
  244. return _used;
  245. }
  246. std::size_t available() const
  247. /// Returns the size of the available portion of the buffer.
  248. {
  249. return size() - _used;
  250. }
  251. void drain(std::size_t length = 0)
  252. /// Drains length number of elements from the buffer.
  253. /// If length is zero or greater than buffer current
  254. /// content length, buffer is emptied.
  255. {
  256. Mutex::ScopedLock lock(_mutex);
  257. std::size_t usedBefore = _used;
  258. if (0 == length || length >= _used)
  259. {
  260. _begin = 0;
  261. _used = 0;
  262. }
  263. else
  264. {
  265. _begin += length;
  266. _used -= length;
  267. }
  268. if (_notify) notify(usedBefore);
  269. }
  270. void copy(const T* ptr, std::size_t length)
  271. /// Copies the supplied data to the buffer and adjusts
  272. /// the used buffer size.
  273. {
  274. poco_check_ptr(ptr);
  275. if (0 == length) return;
  276. Mutex::ScopedLock lock(_mutex);
  277. if (length > available())
  278. throw Poco::InvalidAccessException("Cannot extend buffer.");
  279. if (!isWritable())
  280. throw Poco::InvalidAccessException("Buffer not writable.");
  281. std::memcpy(begin() + _used, ptr, length * sizeof(T));
  282. std::size_t usedBefore = _used;
  283. _used += length;
  284. if (_notify) notify(usedBefore);
  285. }
  286. void advance(std::size_t length)
  287. /// Advances buffer by length elements.
  288. /// Should be called AFTER the data
  289. /// was copied into the buffer.
  290. {
  291. Mutex::ScopedLock lock(_mutex);
  292. if (length > available())
  293. throw Poco::InvalidAccessException("Cannot extend buffer.");
  294. if (!isWritable())
  295. throw Poco::InvalidAccessException("Buffer not writable.");
  296. if (_buffer.size() - (_begin + _used) < length)
  297. {
  298. std::memmove(_buffer.begin(), begin(), _used * sizeof(T));
  299. _begin = 0;
  300. }
  301. std::size_t usedBefore = _used;
  302. _used += length;
  303. if (_notify) notify(usedBefore);
  304. }
  305. T* begin()
  306. /// Returns the pointer to the beginning of the buffer.
  307. {
  308. Mutex::ScopedLock lock(_mutex);
  309. if (_begin != 0)
  310. {
  311. // Move the data to the start of the buffer so begin() and next()
  312. // always return consistent pointers with each other and allow writing
  313. // to the end of the buffer.
  314. std::memmove(_buffer.begin(), _buffer.begin() + _begin, _used * sizeof(T));
  315. _begin = 0;
  316. }
  317. return _buffer.begin();
  318. }
  319. T* next()
  320. /// Returns the pointer to the next available position in the buffer.
  321. {
  322. Mutex::ScopedLock lock(_mutex);
  323. return begin() + _used;
  324. }
  325. T& operator [] (std::size_t index)
  326. /// Returns value at index position.
  327. /// Throws InvalidAccessException if index is larger than
  328. /// the last valid (used) buffer position.
  329. {
  330. Mutex::ScopedLock lock(_mutex);
  331. if (index >= _used)
  332. throw InvalidAccessException(format("Index out of bounds: %z (max index allowed: %z)", index, _used - 1));
  333. return _buffer[_begin + index];
  334. }
  335. const T& operator [] (std::size_t index) const
  336. /// Returns value at index position.
  337. /// Throws InvalidAccessException if index is larger than
  338. /// the last valid (used) buffer position.
  339. {
  340. Mutex::ScopedLock lock(_mutex);
  341. if (index >= _used)
  342. throw InvalidAccessException(format("Index out of bounds: %z (max index allowed: %z)", index, _used - 1));
  343. return _buffer[_begin + index];
  344. }
  345. const Buffer<T>& buffer() const
  346. /// Returns const reference to the underlying buffer.
  347. {
  348. return _buffer;
  349. }
  350. void setError(bool error = true)
  351. /// Sets the error flag on the buffer and empties it.
  352. /// If notifications are enabled, they will be triggered
  353. /// if appropriate.
  354. ///
  355. /// Setting error flag to true prevents reading and writing
  356. /// to the buffer; to re-enable FIFOBuffer for reading/writing,
  357. /// the error flag must be set to false.
  358. {
  359. if (error)
  360. {
  361. bool f = false;
  362. Mutex::ScopedLock lock(_mutex);
  363. if (error && isReadable() && _notify) readable.notify(this, f);
  364. if (error && isWritable() && _notify) writable.notify(this, f);
  365. _error = error;
  366. _used = 0;
  367. }
  368. else
  369. {
  370. bool t = true;
  371. Mutex::ScopedLock lock(_mutex);
  372. _error = false;
  373. if (_notify && !_eof) writable.notify(this, t);
  374. }
  375. }
  376. bool isValid() const
  377. /// Returns true if error flag is not set on the buffer,
  378. /// otherwise returns false.
  379. {
  380. return !_error;
  381. }
  382. void setEOF(bool eof = true)
  383. /// Sets end-of-file flag on the buffer.
  384. ///
  385. /// Setting EOF flag to true prevents writing to the
  386. /// buffer; reading from the buffer will still be
  387. /// allowed until all data present in the buffer at the
  388. /// EOF set time is drained. After that, to re-enable
  389. /// FIFOBuffer for reading/writing, EOF must be
  390. /// set to false.
  391. ///
  392. /// Setting EOF flag to false clears EOF state if it
  393. /// was previously set. If EOF was not set, it has no
  394. /// effect.
  395. {
  396. Mutex::ScopedLock lock(_mutex);
  397. bool flag = !eof;
  398. if (_notify) writable.notify(this, flag);
  399. _eof = eof;
  400. }
  401. bool hasEOF() const
  402. /// Returns true if EOF flag has been set.
  403. {
  404. return _eof;
  405. }
  406. bool isEOF() const
  407. /// Returns true if EOF flag has been set and buffer is empty.
  408. {
  409. return isEmpty() && _eof;
  410. }
  411. bool isEmpty() const
  412. /// Returns true is buffer is empty, false otherwise.
  413. {
  414. return 0 == _used;
  415. }
  416. bool isFull() const
  417. /// Returns true is buffer is full, false otherwise.
  418. {
  419. return size() == _used;
  420. }
  421. bool isReadable() const
  422. /// Returns true if buffer contains data and is not
  423. /// in error state.
  424. {
  425. return !isEmpty() && isValid();
  426. }
  427. bool isWritable() const
  428. /// Returns true if buffer is not full and is not
  429. /// in error state.
  430. {
  431. return !isFull() && isValid() && !_eof;
  432. }
  433. void setNotify(bool notify = true)
  434. /// Enables/disables notifications.
  435. {
  436. _notify = notify;
  437. }
  438. bool getNotify() const
  439. /// Returns true if notifications are enabled, false otherwise.
  440. {
  441. return _notify;
  442. }
  443. Mutex& mutex()
  444. /// Returns reference to mutex.
  445. {
  446. return _mutex;
  447. }
  448. private:
  449. void notify(std::size_t usedBefore)
  450. {
  451. bool t = true, f = false;
  452. if (usedBefore == 0 && _used > 0)
  453. readable.notify(this, t);
  454. else if (usedBefore > 0 && 0 == _used)
  455. readable.notify(this, f);
  456. if (usedBefore == _buffer.size() && _used < _buffer.size())
  457. writable.notify(this, t);
  458. else if (usedBefore < _buffer.size() && _used == _buffer.size())
  459. writable.notify(this, f);
  460. }
  461. BasicFIFOBuffer();
  462. BasicFIFOBuffer(const BasicFIFOBuffer&);
  463. BasicFIFOBuffer& operator = (const BasicFIFOBuffer&);
  464. Buffer<T> _buffer;
  465. std::size_t _begin;
  466. std::size_t _used;
  467. bool _notify;
  468. mutable Mutex _mutex;
  469. bool _eof;
  470. bool _error;
  471. };
  472. //
  473. // We provide an instantiation for char
  474. //
  475. typedef BasicFIFOBuffer<char> FIFOBuffer;
  476. } // namespace Poco
  477. #endif // Foundation_FIFOBuffer_INCLUDED