1 /***************************************************************************
2  *  include/stxxl/bits/mng/buf_istream_reverse.h
3  *
4  *  Part of the STXXL. See http://stxxl.sourceforge.net
5  *
6  *  Copyright (C) 2002-2004 Roman Dementiev <dementiev@mpi-sb.mpg.de>
7  *  Copyright (C) 2013 Timo Bingmann <tb@panthema.net>
8  *
9  *  Distributed under the Boost Software License, Version 1.0.
10  *  (See accompanying file LICENSE_1_0.txt or copy at
11  *  http://www.boost.org/LICENSE_1_0.txt)
12  **************************************************************************/
13 
14 #ifndef STXXL_MNG_BUF_ISTREAM_REVERSE_HEADER
15 #define STXXL_MNG_BUF_ISTREAM_REVERSE_HEADER
16 
17 #include <stxxl/bits/mng/config.h>
18 #include <stxxl/bits/mng/bid.h>
19 #include <stxxl/bits/mng/block_prefetcher.h>
20 #include <stxxl/bits/noncopyable.h>
21 #include <stxxl/bits/algo/async_schedule.h>
22 
23 STXXL_BEGIN_NAMESPACE
24 
25 //! \addtogroup schedlayer
26 //! \{
27 
28 // a paranoid check
29 #define BUF_ISTREAM_CHECK_END
30 
31 //! Buffered input stream, reading the items in the blocks in reverse order.
32 //!
33 //! Reads data records from the stream of blocks in reverse order.
34 //! \remark Reading performed in the background, i.e. with overlapping of I/O and computation
35 template <typename BlockType, typename BidIteratorType>
36 class buf_istream_reverse : private noncopyable
37 {
38 public:
39     typedef BlockType block_type;
40     typedef BidIteratorType bid_iterator_type;
41 
42     //-tb note that we redefine the BID type here, because there is no way to
43     //-derive it from BidIteratorType (which is usually just a POD pointer).
44     typedef BIDArray<block_type::raw_size> bid_vector_type;
45 
46 private:
buf_istream_reverse()47     buf_istream_reverse() { }
48 
49 protected:
50     typedef block_prefetcher<block_type, typename bid_vector_type::iterator> prefetcher_type;
51     prefetcher_type* prefetcher;
52     int_type current_elem;
53     block_type* current_blk;
54     int_type* prefetch_seq;
55 #ifdef BUF_ISTREAM_CHECK_END
56     bool not_finished;
57 #endif
58     bid_vector_type bids_;
59 
60 public:
61     typedef typename block_type::reference reference;
62     typedef buf_istream_reverse<block_type, bid_iterator_type> self_type;
63 
64     //! Constructs input stream object, reading [first,last) blocks in reverse.
65     //! \param begin \c bid_iterator pointing to the first block of the stream
66     //! \param end \c bid_iterator pointing to the ( \b last + 1 ) block of the stream
67     //! \param nbuffers number of buffers for internal use
buf_istream_reverse(bid_iterator_type begin,bid_iterator_type end,int_type nbuffers)68     buf_istream_reverse(bid_iterator_type begin, bid_iterator_type end, int_type nbuffers)
69         : current_elem(0),
70 #ifdef BUF_ISTREAM_CHECK_END
71           not_finished(true),
72 #endif
73           bids_(end - begin)
74     {
75         // copy list of bids in reverse
76         std::reverse_copy(begin, end, bids_.begin());
77 
78         // calculate prefetch sequence
79         const unsigned_type ndisks = config::get_instance()->disks_number();
80         const unsigned_type mdevid = config::get_instance()->get_max_device_id();
81 
82         prefetch_seq = new int_type[bids_.size()];
83 
84         // optimal schedule
85         nbuffers = STXXL_MAX(2 * ndisks, unsigned_type(nbuffers - 1));
86         compute_prefetch_schedule(bids_.begin(), bids_.end(), prefetch_seq,
87                                   nbuffers, mdevid);
88 
89         // create stream prefetcher
90         prefetcher = new prefetcher_type(bids_.begin(), bids_.end(), prefetch_seq, nbuffers);
91 
92         // fetch block: last in sequence
93         current_blk = prefetcher->pull_block();
94         current_elem = block_type::size - 1;
95     }
96 
97     //! Input stream operator, reads in \c record.
98     //! \param record reference to the block record type,
99     //!        contains value of the next record in the stream after the call of the operator
100     //! \return reference to itself (stream object)
101     self_type& operator >> (reference record)
102     {
103 #ifdef BUF_ISTREAM_CHECK_END
104         assert(not_finished);
105 #endif
106 
107         record = current_blk->elem[current_elem--];
108 
109         if (UNLIKELY(current_elem < 0))
110         {
111             current_elem = block_type::size - 1;
112 #ifdef BUF_ISTREAM_CHECK_END
113             not_finished = prefetcher->block_consumed(current_blk);
114 #else
115             prefetcher->block_consumed(current_blk);
116 #endif
117         }
118 
119         return (*this);
120     }
121 
122     //! Returns reference to the current record in the stream.
current()123     reference current()     /* const */
124     {
125         return current_blk->elem[current_elem];
126     }
127 
128     //! Returns reference to the current record in the stream.
129     reference operator * ()     /* const */
130     {
131         return current_blk->elem[current_elem];
132     }
133 
134     //! Moves to the _previous_ record in the stream.
135     //! \return reference to itself after the advance
136     self_type& operator ++ ()
137     {
138 #ifdef BUF_ISTREAM_CHECK_END
139         assert(not_finished);
140 #endif
141 
142         current_elem--;
143 
144         if (UNLIKELY(current_elem < 0))
145         {
146             current_elem = block_type::size - 1;
147 #ifdef BUF_ISTREAM_CHECK_END
148             not_finished = prefetcher->block_consumed(current_blk);
149 #else
150             prefetcher->block_consumed(current_blk);
151 #endif
152         }
153         return *this;
154     }
155 
156     //! Frees used internal objects.
~buf_istream_reverse()157     ~buf_istream_reverse()
158     {
159         delete prefetcher;
160         delete[] prefetch_seq;
161     }
162 };
163 
164 //! \}
165 
166 STXXL_END_NAMESPACE
167 
168 #endif // !STXXL_MNG_BUF_ISTREAM_REVERSE_HEADER
169