Stxxl  1.2.1
async_schedule.h
1 /***************************************************************************
2  * include/stxxl/bits/algo/async_schedule.h
3  *
4  * Part of the STXXL. See http://stxxl.sourceforge.net
5  *
6  * Copyright (C) 2002 Roman Dementiev <dementiev@mpi-sb.mpg.de>
7  *
8  * Distributed under the Boost Software License, Version 1.0.
9  * (See accompanying file LICENSE_1_0.txt or copy at
10  * http://www.boost.org/LICENSE_1_0.txt)
11  **************************************************************************/
12 
13 #ifndef STXXL_ASYNC_SCHEDULE_HEADER
14 #define STXXL_ASYNC_SCHEDULE_HEADER
15 
16 #include <queue>
17 #include <algorithm>
18 #include <functional>
19 
20 #ifdef STXXL_BOOST_CONFIG
21  #include <boost/config.hpp>
22 #endif
23 #include <stxxl/bits/io/iobase.h>
24 
25 #include <stxxl/bits/compat_hash_map.h>
26 #include <stxxl/bits/compat_hash_set.h>
27 
28 
29 __STXXL_BEGIN_NAMESPACE
30 
31 struct sim_event // only one type of event: WRITE COMPLETED
32 {
33  int_type timestamp;
34  int_type iblock;
35  inline sim_event(int_type t, int_type b) : timestamp(t), iblock(b) { }
36 };
37 
38 struct sim_event_cmp : public std::binary_function<sim_event, sim_event, bool>
39 {
40  inline bool operator () (const sim_event & a, const sim_event & b) const
41  {
42  return a.timestamp > b.timestamp;
43  }
44 };
45 
46 int_type simulate_async_write(
47  int * disks,
48  const int_type L,
49  const int_type m_init,
50  const int_type D,
51  std::pair<int_type, int_type> * o_time);
52 
53 typedef std::pair<int_type, int_type> write_time_pair;
54 struct write_time_cmp : public std::binary_function<write_time_pair, write_time_pair, bool>
55 {
56  inline bool operator () (const write_time_pair & a, const write_time_pair & b) const
57  {
58  return a.second > b.second;
59  }
60 };
61 
62 void compute_prefetch_schedule(
63  int_type * first,
64  int_type * last,
65  int_type * out_first,
66  int_type m,
67  int_type D);
68 
69 template <typename run_type>
70 void simulate_async_write(
71  const run_type & input,
72  const int_type m_init,
73  const int_type D,
74  std::pair<int_type, int_type> * o_time)
75 {
76  typedef std::priority_queue<sim_event, std::vector<sim_event>, sim_event_cmp> event_queue_type;
77  typedef std::queue<int_type> disk_queue_type;
78 
79  const int_type L = input.size();
80  assert(L >= D);
81  disk_queue_type * disk_queues = new disk_queue_type[L];
82  event_queue_type event_queue;
83 
84  int_type m = m_init;
85  int_type i = L - 1;
86  int_type oldtime = 0;
87  bool * disk_busy = new bool[D];
88 
89  while (m && (i >= 0))
90  {
91  int_type disk = input[i].bid.storage->get_id();
92  disk_queues[disk].push(i);
93  i--;
94  m--;
95  }
96 
97  for (int_type ii = 0; ii < D; ii++)
98  if (!disk_queues[ii].empty())
99  {
100  int_type j = disk_queues[ii].front();
101  disk_queues[ii].pop();
102  event_queue.push(sim_event(1, j));
103  }
104 
105  while (!event_queue.empty())
106  {
107  sim_event cur = event_queue.top();
108  event_queue.pop();
109  if (oldtime != cur.timestamp)
110  {
111  // clear disk_busy
112  for (int_type i = 0; i < D; i++)
113  disk_busy[i] = false;
114 
115 
116  oldtime = cur.timestamp;
117  }
118  o_time[cur.iblock] = std::pair<int_type, int_type>(cur.iblock, cur.timestamp + 1);
119 
120  m++;
121  if (i >= 0)
122  {
123  m--;
124  int_type disk = input[i].bid.storage->get_id();
125  if (disk_busy[disk])
126  {
127  disk_queues[disk].push(i);
128  }
129  else
130  {
131  event_queue.push(sim_event(cur.timestamp + 1, i));
132  disk_busy[disk] = true;
133  }
134 
135  i--;
136  }
137 
138  // add next block to write
139  int_type disk = input[cur.iblock].bid.storage->get_id();
140  if (!disk_busy[disk] && !disk_queues[disk].empty())
141  {
142  event_queue.push(sim_event(cur.timestamp + 1, disk_queues[disk].front()));
143  disk_queues[disk].pop();
144  disk_busy[disk] = true;
145  }
146  }
147 
148  delete[] disk_busy;
149  delete[] disk_queues;
150 }
151 
152 
153 template <typename run_type>
154 void compute_prefetch_schedule(
155  const run_type & input,
156  int_type * out_first,
157  int_type m,
158  int_type D)
159 {
160  typedef std::pair<int_type, int_type> pair_type;
161  const int_type L = input.size();
162  if (L <= D)
163  {
164  for (int_type i = 0; i < L; ++i)
165  out_first[i] = i;
166 
167  return;
168  }
169  pair_type * write_order = new pair_type[L];
170 
171  simulate_async_write(input, m, D, write_order);
172 
173  std::stable_sort(write_order, write_order + L, write_time_cmp());
174 
175  for (int_type i = 0; i < L; i++)
176  out_first[i] = write_order[i].first;
177 
178 
179  delete[] write_order;
180 }
181 
182 
183 template <typename bid_iterator_type>
184 void simulate_async_write(
185  bid_iterator_type input,
186  const int_type L,
187  const int_type m_init,
188  const int_type /*D*/,
189  std::pair<int_type, int_type> * o_time)
190 {
191  typedef std::priority_queue<sim_event, std::vector<sim_event>, sim_event_cmp> event_queue_type;
192  typedef std::queue<int_type> disk_queue_type;
193 
194  //disk_queue_type * disk_queues = new disk_queue_type[L];
195  typedef typename compat_hash_map<int, disk_queue_type>::result disk_queues_type;
196  disk_queues_type disk_queues;
197  event_queue_type event_queue;
198 
199  int_type m = m_init;
200  int_type i = L - 1;
201  int_type oldtime = 0;
202  //bool * disk_busy = new bool [D];
203  compat_hash_set<int>::result disk_busy;
204 
205  while (m && (i >= 0))
206  {
207  int_type disk = (*(input + i)).storage->get_id();
208  disk_queues[disk].push(i);
209  i--;
210  m--;
211  }
212 
213  /*
214  for(int_type i=0;i<D;i++)
215  if(!disk_queues[i].empty())
216  {
217  int_type j = disk_queues[i].front();
218  disk_queues[i].pop();
219  event_queue.push(sim_event(1,j));
220  }
221  */
222  disk_queues_type::iterator it = disk_queues.begin();
223  for ( ; it != disk_queues.end(); ++it)
224  {
225  int_type j = (*it).second.front();
226  (*it).second.pop();
227  event_queue.push(sim_event(1, j));
228  }
229 
230  while (!event_queue.empty())
231  {
232  sim_event cur = event_queue.top();
233  event_queue.pop();
234  if (oldtime != cur.timestamp)
235  {
236  // clear disk_busy
237  //for(int_type i=0;i<D;i++)
238  // disk_busy[i] = false;
239  disk_busy.clear();
240 
241  oldtime = cur.timestamp;
242  }
243  o_time[cur.iblock] = std::pair<int_type, int_type>(cur.iblock, cur.timestamp + 1);
244 
245  m++;
246  if (i >= 0)
247  {
248  m--;
249  int_type disk = (*(input + i)).storage->get_id();
250  if (disk_busy.find(disk) != disk_busy.end())
251  {
252  disk_queues[disk].push(i);
253  }
254  else
255  {
256  event_queue.push(sim_event(cur.timestamp + 1, i));
257  disk_busy.insert(disk);
258  }
259 
260  i--;
261  }
262 
263  // add next block to write
264  int_type disk = (*(input + cur.iblock)).storage->get_id();
265  if (disk_busy.find(disk) == disk_busy.end() && !disk_queues[disk].empty())
266  {
267  event_queue.push(sim_event(cur.timestamp + 1, disk_queues[disk].front()));
268  disk_queues[disk].pop();
269  disk_busy.insert(disk);
270  }
271  }
272 
273  //delete [] disk_busy;
274  //delete [] disk_queues;
275 }
276 
277 
278 template <typename bid_iterator_type>
279 void compute_prefetch_schedule(
280  bid_iterator_type input_begin,
281  bid_iterator_type input_end,
282  int_type * out_first,
283  int_type m,
284  int_type D)
285 {
286  typedef std::pair<int_type, int_type> pair_type;
287  const int_type L = input_end - input_begin;
288  STXXL_VERBOSE1("compute_prefetch_schedule: sequence length=" << L << " disks=" << D);
289  if (L <= D)
290  {
291  for (int_type i = 0; i < L; ++i)
292  out_first[i] = i;
293 
294  return;
295  }
296  pair_type * write_order = new pair_type[L];
297 
298  simulate_async_write(input_begin, L, m, D, write_order);
299 
300  std::stable_sort(write_order, write_order + L, write_time_cmp());
301 
302  STXXL_VERBOSE1("Computed prefetch order for " << L << " blocks with " <<
303  D << " disks and " << m << " blocks");
304  for (int_type i = 0; i < L; i++)
305  {
306  out_first[i] = write_order[i].first;
307  STXXL_VERBOSE1(write_order[i].first);
308  }
309 
310  delete[] write_order;
311 }
312 
313 __STXXL_END_NAMESPACE
314 
315 #endif // !STXXL_ASYNC_SCHEDULE_HEADER