- static long s_buffer_count = 0; // counts for debugging storage mgmt

- static long s_buffer_reader_count = 0;

-

- /* ----------------------------------------------------------------------------

- Notes on storage management

-

- Pretty much all the fundamental classes are now using the

- shared_ptr stuff for automatic reference counting. To ensure that

- no mistakes are made, we make the constructors for classes private,

- and then provide a free factory function that returns a smart

- pointer to the desired class.

-

- gr::buffer and gr::buffer_reader are no exceptions. However, they

- both want pointers to each other, and unless we do something, we'll

- never delete any of them because of the circular structure.

- They'll always have a reference count of at least one. We could

- use boost::weak_ptr's from gr::buffer to gr::buffer_reader but that

- introduces it's own problems. (gr::buffer_reader's destructor needs

- to call gr::buffer::drop_reader, but has no easy way to get a

- shared_ptr to itself.)

-

- Instead, we solve this problem by having gr::buffer hold a raw

- pointer to gr::buffer_reader in its d_reader vector.

- gr::buffer_reader's destructor calls gr::buffer::drop_reader, so

- we're never left with an dangling pointer. gr::buffer_reader still

- has a shared_ptr to the buffer ensuring that the buffer doesn't go

- away under it. However, when the reference count of a

- gr::buffer_reader goes to zero, we can successfully reclaim it.

- ---------------------------------------------------------------------------- */

-

- /*

- * Compute the minimum number of buffer items that work (i.e.,

- * address space wrap-around works). To work is to satisfy this

- * constraint for integer buffer_size and k:

- *

- * type_size * nitems == k * page_size

- */

- static long

- minimum_buffer_items(long type_size, long page_size)

- {

- return page_size / boost::math::gcd (type_size, page_size);

- return page_size / boost::integer::gcd (type_size, page_size);

- }

-

-

- buffer::buffer(int nitems, size_t sizeof_item, block_sptr link)

- : d_base(0), d_bufsize(0), d_max_reader_delay(0), d_vmcircbuf(0),

- d_sizeof_item(sizeof_item), d_link(link),

- d_write_index(0), d_abs_write_offset(0), d_done(false),

- {

- if(!allocate_buffer (nitems, sizeof_item))

- throw std::bad_alloc ();

- }

- buffer_sptr

- make_buffer(int nitems, size_t sizeof_item, block_sptr link)

- {

- }

- buffer::~buffer()

- {

- }

-

- /*!

- * sets d_vmcircbuf, d_base, d_bufsize.

- * returns true iff successful.

- */

- bool

- buffer::allocate_buffer(int nitems, size_t sizeof_item)

- {

- int min_nitems = minimum_buffer_items(sizeof_item, granularity);

- if(nitems % min_nitems != 0)

- nitems = ((nitems / min_nitems) + 1) * min_nitems;

- if(nitems > 2 * orig_nitems && nitems * (int) sizeof_item > granularity){

- std::cerr << "gr::buffer::allocate_buffer: warning: tried to allocate\n"

- << " " << orig_nitems << " items of size "

- << sizeof_item << ". Due to alignment requirements\n"

- << " " << nitems << " were allocated. If this isn't OK, consider padding\n"

- << " your structure to a power-of-two bytes.\n"

- << " On this platform, our allocation granularity is " << granularity << " bytes.\n";

- if(d_vmcircbuf == 0){

- std::cerr << "gr::buffer::allocate_buffer: failed to allocate buffer of size "

- << d_bufsize * d_sizeof_item / 1024 << " KB\n";

- return false;

- }

- int

- buffer::space_available()

- {

- if(d_readers.empty())

- return d_bufsize - 1; // See comment below

- // Find out the maximum amount of data available to our readers

-

- int most_data = d_readers[0]->items_available();

- uint64_t min_items_read = d_readers[0]->nitems_read();

- for(size_t i = 1; i < d_readers.size (); i++) {

- most_data = std::max(most_data, d_readers[i]->items_available());

- min_items_read = std::min(min_items_read, d_readers[i]->nitems_read());

- }

-

- if(min_items_read != d_last_min_items_read) {

- prune_tags(d_last_min_items_read);

- d_last_min_items_read = min_items_read;

- }

-

- // The -1 ensures that the case d_write_index == d_read_index is

- // unambiguous. It indicates that there is no data for the reader

- return d_bufsize - most_data - 1;

- }

- void *

- buffer::write_pointer()

- {

- return &d_base[d_write_index * d_sizeof_item];

- }

- void

- buffer::update_write_pointer(int nitems)

- {

- }

- void

- buffer::set_done(bool done)

- {

- }

-

- buffer_reader_sptr

- buffer_add_reader(buffer_sptr buf, int nzero_preload, block_sptr link, int delay)

- {

- if(nzero_preload < 0)

- throw std::invalid_argument("buffer_add_reader: nzero_preload must be >= 0");

-

- buffer_reader_sptr r(new buffer_reader(buf,

- buf->index_sub(buf->d_write_index,

- nzero_preload),

- link));

- buf->d_readers.push_back(r.get ());

- }

- void

- buffer::drop_reader(buffer_reader *reader)

- {

- std::vector<buffer_reader *>::iterator result =

- std::find(d_readers.begin(), d_readers.end(), reader);

- if(result == d_readers.end())

- throw std::invalid_argument("buffer::drop_reader"); // we didn't find it...

- }

- void

- buffer::add_item_tag(const tag_t &tag)

- {

- d_item_tags.insert(std::pair<uint64_t,tag_t>(tag.offset,tag));

- }

- void

- buffer::remove_item_tag(const tag_t &tag, long id)

- {

- for(std::multimap<uint64_t,tag_t>::iterator it = d_item_tags.lower_bound(tag.offset); it != d_item_tags.upper_bound(tag.offset); ++it) {

- if((*it).second == tag) {

- (*it).second.marked_deleted.push_back(id);

- }

- }

- void

- buffer::prune_tags(uint64_t max_time)

- {

- std::multimap<uint64_t,tag_t>::iterator itr(d_item_tags.begin()), tmp;

- while(itr != d_item_tags.end()) {

- item_time = (*itr).second.offset;

- if(item_time+d_max_reader_delay + bufsize() < max_time) {

- tmp = itr;

- itr++;

- d_item_tags.erase(tmp);

- }

- else {

- // d_item_tags is a map sorted by offset, so when the if

- // condition above fails, all future tags in the map must also

- // fail. So just break here.

- break;

- }

- }

- long

- buffer_ncurrently_allocated()

- {

- return s_buffer_count;

- }

- // ----------------------------------------------------------------------------

- buffer_reader::buffer_reader(buffer_sptr buffer, unsigned int read_index,

- block_sptr link)

- : d_buffer(buffer), d_read_index(read_index), d_abs_read_offset(0), d_link(link),

- {

- }

- buffer_reader::~buffer_reader()

- {

- }

- void

- buffer_reader::declare_sample_delay(unsigned delay)

- {

- d_buffer->d_max_reader_delay = std::max(d_attr_delay,

- d_buffer->d_max_reader_delay);

- }

-

- unsigned

- buffer_reader::sample_delay() const

- {

- return d_attr_delay;

- }

-

- int

- buffer_reader::items_available() const

- {

- }

- const void *

- buffer_reader::read_pointer()

- {

- }

- void

- buffer_reader::update_read_pointer(int nitems)

- {

- d_read_index = d_buffer->index_add (d_read_index, nitems);

- }

-

- void

- buffer_reader::get_tags_in_range(std::vector<tag_t> &v,

- uint64_t abs_start,

- uint64_t abs_end,

- long id)

- {

- std::multimap<uint64_t,tag_t>::iterator itr = d_buffer->get_tags_lower_bound(std::min(abs_start, abs_start - d_attr_delay));

- std::multimap<uint64_t,tag_t>::iterator itr_end = d_buffer->get_tags_upper_bound(std::min(abs_end, abs_end - d_attr_delay));

- while(itr != itr_end) {

- item_time = (*itr).second.offset + d_attr_delay;

- if((item_time >= abs_start) && (item_time < abs_end)) {

- std::vector<long>::iterator id_itr;

- id_itr = std::find(itr->second.marked_deleted.begin(), itr->second.marked_deleted.end(), id);

- // If id is not in the vector of marked blocks

- if(id_itr == itr->second.marked_deleted.end()) {

- tag_t t = (*itr).second;

- t.offset += d_attr_delay;

- v.push_back(t);

- v.back().marked_deleted.clear();

- }

- itr++;

- }

- long

- buffer_reader_ncurrently_allocated()

- {

- return s_buffer_reader_count;

- }