STXXL  1.4.1
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Groups Pages
wbtl_file.cpp
Go to the documentation of this file.
1 /***************************************************************************
2  * lib/io/wbtl_file.cpp
3  *
4  * a write-buffered-translation-layer pseudo file
5  *
6  * Part of the STXXL. See http://stxxl.sourceforge.net
7  *
8  * Copyright (C) 2008-2009 Andreas Beckmann <[email protected]>
9  * Copyright (C) 2009 Johannes Singler <[email protected]>
10  *
11  * Distributed under the Boost Software License, Version 1.0.
12  * (See accompanying file LICENSE_1_0.txt or copy at
13  * http://www.boost.org/LICENSE_1_0.txt)
14  **************************************************************************/
15 
18 
19 #if STXXL_HAVE_WBTL_FILE
20 
21 #include <algorithm>
22 #include <iomanip>
23 #include <stxxl/bits/io/io.h>
24 #include <stxxl/bits/parallel.h>
25 #include <stxxl/aligned_alloc>
26 
27 #ifndef STXXL_VERBOSE_WBTL
28 #define STXXL_VERBOSE_WBTL STXXL_VERBOSE2
29 #endif
30 
32 
33 wbtl_file::wbtl_file(
34  file* backend_file,
35  size_type write_buffer_size,
36  int write_buffers,
37  int queue_id, int allocator_id)
38  : disk_queued_file(queue_id, allocator_id), storage(backend_file),
39  sz(0), write_block_size(write_buffer_size),
40  free_bytes(0), curbuf(1), curpos(write_block_size)
41 {
42  STXXL_UNUSED(write_buffers);
43  assert(write_buffers == 2); // currently hardcoded
44  write_buffer[0] = static_cast<char*>(stxxl::aligned_alloc<STXXL_BLOCK_ALIGN>(write_block_size));
45  write_buffer[1] = static_cast<char*>(stxxl::aligned_alloc<STXXL_BLOCK_ALIGN>(write_block_size));
46  buffer_address[0] = offset_type(-1);
47  buffer_address[1] = offset_type(-1);
48 }
49 
50 wbtl_file::~wbtl_file()
51 {
52  stxxl::aligned_dealloc<STXXL_BLOCK_ALIGN>(write_buffer[1]);
53  stxxl::aligned_dealloc<STXXL_BLOCK_ALIGN>(write_buffer[0]);
54  delete storage;
55  storage = 0;
56 }
57 
58 void wbtl_file::serve(void* buffer, offset_type offset, size_type bytes,
59  request::request_type type)
60 {
61  if (type == request::READ)
62  {
63  //stats::scoped_read_timer read_timer(size());
64  sread(buffer, offset, bytes);
65  }
66  else
67  {
68  //stats::scoped_write_timer write_timer(size());
69  swrite(buffer, offset, bytes);
70  }
71 }
72 
73 void wbtl_file::lock()
74 {
75  storage->lock();
76 }
77 
78 wbtl_file::offset_type wbtl_file::size()
79 {
80  return sz;
81 }
82 
83 void wbtl_file::set_size(offset_type newsize)
84 {
85  scoped_mutex_lock mapping_lock(mapping_mutex);
86  assert(sz <= newsize); // may not shrink
87  if (sz < newsize) {
88  _add_free_region(sz, newsize - sz);
89  storage->set_size(newsize);
90  sz = newsize;
91  assert(sz == storage->size());
92  }
93 }
94 
95 #define FMT_A_S(_addr_, _size_) "0x" << std::hex << std::setfill('0') << std::setw(8) << (_addr_) << "/0x" << std::setw(8) << (_size_)
96  #define FMT_A_C(_addr_, _size_) "0x" << std::setw(8) << (_addr_) << "(" << std::dec << (_size_) << ")"
97  #define FMT_A(_addr_) "0x" << std::setw(8) << (_addr_)
98 
99 // logical address
100 void wbtl_file::discard(offset_type offset, offset_type size)
101 {
102  scoped_mutex_lock mapping_lock(mapping_mutex);
103  sortseq::iterator physical = address_mapping.find(offset);
104  STXXL_VERBOSE_WBTL("wbtl:discard l" << FMT_A_S(offset, size) << " @ p" << FMT_A(physical != address_mapping.end() ? physical->second : 0xffffffff));
105  if (physical == address_mapping.end()) {
106  // could be OK if the block was never written ...
107  //STXXL_ERRMSG("discard: mapping not found: " << FMT_A_S(offset, size) << " ==> " << "???");
108  } else {
109  offset_type physical_offset = physical->second;
110  address_mapping.erase(physical);
111  _add_free_region(physical_offset, size);
112  place_map::iterator reverse = reverse_mapping.find(physical_offset);
113  if (reverse == reverse_mapping.end()) {
114  STXXL_ERRMSG("discard: reverse mapping not found: " << FMT_A_S(offset, size) << " ==> " << "???");
115  } else {
116  assert(offset == (reverse->second).first);
117  reverse_mapping.erase(reverse);
118  }
119  storage->discard(physical_offset, size);
120  }
121 }
122 
123 // physical address
124 void wbtl_file::_add_free_region(offset_type offset, offset_type size)
125 {
126  // mapping_lock has to be aquired by caller
127  STXXL_VERBOSE_WBTL("wbtl:addfre p" << FMT_A_S(offset, size) << " F <= f" << FMT_A_C(free_bytes, free_space.size()));
128  offset_type region_pos = offset;
129  offset_type region_size = size;
130  if (!free_space.empty())
131  {
132  sortseq::iterator succ = free_space.upper_bound(region_pos);
133  sortseq::iterator pred = succ;
134  pred--;
135  check_corruption(region_pos, region_size, pred, succ);
136  if (succ == free_space.end())
137  {
138  if (pred == free_space.end())
139  {
140  //dump();
141  assert(pred != free_space.end());
142  }
143  if ((*pred).first + (*pred).second == region_pos)
144  {
145  // coalesce with predecessor
146  region_size += (*pred).second;
147  region_pos = (*pred).first;
148  free_space.erase(pred);
149  }
150  }
151  else
152  {
153  if (free_space.size() > 1)
154  {
155  bool succ_is_not_the_first = (succ != free_space.begin());
156  if ((*succ).first == region_pos + region_size)
157  {
158  // coalesce with successor
159  region_size += (*succ).second;
160  free_space.erase(succ);
161  }
162  if (succ_is_not_the_first)
163  {
164  if (pred == free_space.end())
165  {
166  //dump();
167  assert(pred != free_space.end());
168  }
169  if ((*pred).first + (*pred).second == region_pos)
170  {
171  // coalesce with predecessor
172  region_size += (*pred).second;
173  region_pos = (*pred).first;
174  free_space.erase(pred);
175  }
176  }
177  }
178  else
179  {
180  if ((*succ).first == region_pos + region_size)
181  {
182  // coalesce with successor
183  region_size += (*succ).second;
184  free_space.erase(succ);
185  }
186  }
187  }
188  }
189 
190  free_space[region_pos] = region_size;
191  free_bytes += size;
192  STXXL_VERBOSE_WBTL("wbtl:free p" << FMT_A_S(region_pos, region_size) << " F => f" << FMT_A_C(free_bytes, free_space.size()));
193 }
194 
195 void wbtl_file::sread(void* buffer, offset_type offset, size_type bytes)
196 {
197  scoped_mutex_lock buffer_lock(buffer_mutex);
198  int cached = -1;
199  offset_type physical_offset;
200  // map logical to physical address
201  {
202  scoped_mutex_lock mapping_lock(mapping_mutex);
203  sortseq::iterator physical = address_mapping.find(offset);
204  if (physical == address_mapping.end()) {
205  STXXL_ERRMSG("wbtl_read: mapping not found: " << FMT_A_S(offset, bytes) << " ==> " << "???");
206  //STXXL_THROW_ERRNO(io_error, "wbtl_read of unmapped memory");
207  physical_offset = 0xffffffff;
208  } else {
209  physical_offset = physical->second;
210  }
211  }
212 
213  if (buffer_address[curbuf] <= physical_offset &&
214  physical_offset < buffer_address[curbuf] + write_block_size)
215  {
216  // block is in current write buffer
217  assert(physical_offset + bytes <= buffer_address[curbuf] + write_block_size);
218  memcpy(buffer, write_buffer[curbuf] + (physical_offset - buffer_address[curbuf]), bytes);
219  stats::get_instance()->read_cached(bytes);
220  cached = curbuf;
221  }
222  else if (buffer_address[1 - curbuf] <= physical_offset &&
223  physical_offset < buffer_address[1 - curbuf] + write_block_size)
224  {
225  // block is in previous write buffer
226  assert(physical_offset + bytes <= buffer_address[1 - curbuf] + write_block_size);
227  memcpy(buffer, write_buffer[1 - curbuf] + (physical_offset - buffer_address[1 - curbuf]), bytes);
228  stats::get_instance()->read_cached(bytes);
229  cached = curbuf;
230  }
231  else if (physical_offset == 0xffffffff) {
232  // block was deleted or never written before
233  char* uninitialized = (char*)malloc(sizeof(char));
234  memset(buffer, *uninitialized, bytes);
235  free(uninitialized);
236  }
237  else
238  {
239  // block is not cached
240  request_ptr req = storage->aread(buffer, physical_offset, bytes);
241  req->wait(false);
242  }
243  STXXL_VERBOSE_WBTL("wbtl:sread l" << FMT_A_S(offset, bytes) << " @ p" << FMT_A(physical_offset) << " " << std::dec << cached);
244  STXXL_UNUSED(cached);
245 }
246 
247 void wbtl_file::swrite(void* buffer, offset_type offset, size_type bytes)
248 {
249  scoped_mutex_lock buffer_lock(buffer_mutex);
250  // is the block already mapped?
251  {
252  scoped_mutex_lock mapping_lock(mapping_mutex);
253  sortseq::iterator physical = address_mapping.find(offset);
254  STXXL_VERBOSE_WBTL("wbtl:swrite l" << FMT_A_S(offset, bytes) << " @ <= p" <<
255  FMT_A_C(physical != address_mapping.end() ? physical->second : 0xffffffff, address_mapping.size()));
256  if (physical != address_mapping.end()) {
257  mapping_lock.unlock();
258  // FIXME: special case if we can replace it in the current writing block
259  discard(offset, bytes);
260  }
261  }
262 
263  if (bytes > write_block_size - curpos)
264  {
265  // not enough space in the current write buffer
266 
267  if (buffer_address[curbuf] != offset_type(-1)) {
268  STXXL_VERBOSE_WBTL("wbtl:w2disk p" << FMT_A_S(buffer_address[curbuf], write_block_size));
269 
270  // mark remaining part as free
271  if (curpos < write_block_size)
272  _add_free_region(buffer_address[curbuf] + curpos, write_block_size - curpos);
273 
274  if (backend_request.get()) {
275  backend_request->wait(false);
276  }
277 
278  backend_request = storage->awrite(write_buffer[curbuf], buffer_address[curbuf], write_block_size);
279  }
280 
281  curbuf = 1 - curbuf;
282 
283  buffer_address[curbuf] = get_next_write_block();
284  curpos = 0;
285  }
286  assert(bytes <= write_block_size - curpos);
287 
288  // write block into buffer
289  memcpy(write_buffer[curbuf] + curpos, buffer, bytes);
290  stats::get_instance()->write_cached(bytes);
291 
292  scoped_mutex_lock mapping_lock(mapping_mutex);
293  address_mapping[offset] = buffer_address[curbuf] + curpos;
294  reverse_mapping[buffer_address[curbuf] + curpos] = place(offset, bytes);
295  STXXL_VERBOSE_WBTL("wbtl:swrite l" << FMT_A_S(offset, bytes) << " @ => p" << FMT_A_C(buffer_address[curbuf] + curpos, address_mapping.size()));
296  curpos += bytes;
297 }
298 
299 wbtl_file::offset_type wbtl_file::get_next_write_block()
300 {
301  // mapping_lock has to be aquired by caller
302  sortseq::iterator space =
303  std::find_if(free_space.begin(), free_space.end(),
304  bind2nd(FirstFit(), write_block_size) _STXXL_FORCE_SEQUENTIAL);
305 
306  if (space != free_space.end())
307  {
308  offset_type region_pos = (*space).first;
309  offset_type region_size = (*space).second;
310  free_space.erase(space);
311  if (region_size > write_block_size)
312  free_space[region_pos + write_block_size] = region_size - write_block_size;
313 
314  free_bytes -= write_block_size;
315 
316  STXXL_VERBOSE_WBTL("wbtl:nextwb p" << FMT_A_S(region_pos, write_block_size) << " F f" << FMT_A_C(free_bytes, free_space.size()));
317  return region_pos;
318  }
319 
320  STXXL_THROW_ERRNO(io_error, "OutOfSpace, probably fragmented");
321 }
322 
323 void wbtl_file::check_corruption(offset_type region_pos, offset_type region_size,
324  sortseq::iterator pred, sortseq::iterator succ)
325 {
326  if (pred != free_space.end())
327  {
328  if (pred->first <= region_pos && pred->first + pred->second > region_pos)
329  {
330  STXXL_THROW(bad_ext_alloc, "Error: double deallocation of external memory " <<
331  "System info: P " << pred->first << " " << pred->second << " " << region_pos);
332  }
333  }
334  if (succ != free_space.end())
335  {
336  if (region_pos <= succ->first && region_pos + region_size > succ->first)
337  {
338  STXXL_THROW(bad_ext_alloc, "Error: double deallocation of external memory "
339  << "System info: S " << region_pos << " " << region_size << " " << succ->first);
340  }
341  }
342 }
343 
344 const char* wbtl_file::io_type() const
345 {
346  return "wbtl";
347 }
348 
350 
351 #endif // #if STXXL_HAVE_WBTL_FILE
352 // vim: et:ts=4:sw=4
#define STXXL_THROW(exception_type, error_message)
Throws exception_type with &quot;Error in [function] : [error_message]&quot;.
counting_ptr< request > request_ptr
A reference counting pointer for request.
Definition: request.h:113
virtual void wait(bool measure_time=true)=0
Suspends calling thread until completion of the request.
#define STXXL_BEGIN_NAMESPACE
Definition: namespace.h:16
void STXXL_UNUSED(const U &)
Definition: unused.h:22
#define STXXL_ERRMSG(x)
Definition: verbose.h:80
#define STXXL_THROW_ERRNO(exception_type, error_message)
Throws exception_type with &quot;Error in [function] : [error_message] : [errno message]&quot;.
static const size_t bytes
number of bytes in uint_pair
Definition: uint_types.h:96
#define _STXXL_FORCE_SEQUENTIAL
Definition: parallel.h:50
void discard(StreamAlgorithm &in)
Reads stream content and discards it. Useful where you do not need the processed stream anymore...
Definition: stream.h:640
#define STXXL_END_NAMESPACE
Definition: namespace.h:17