buffer.cpp

Go to the documentation of this file.

#ifndef BUFFER_IMPLEMENTATION_FILE
#define BUFFER_IMPLEMENTATION_FILE

/*****************************************************************************\
*                                                                             *
*  Name   : buffer                                                            *
*  Author : Chris Koeritz                                                     *
*                                                                             *
*******************************************************************************
* Copyright (c) 1992-$now By Author.  This program is free software; you can  *
* redistribute it and/or modify it under the terms of the GNU General Public  *
* License as published by the Free Software Foundation; either version 2 of   *
* the License or (at your option) any later version.  This is online at:      *
*     http://www.fsf.org/copyleft/gpl.html                                    *
* Please send any updates to: fred@gruntose.com                               *
\*****************************************************************************/

#include "buffer.h"
#include "buffer_key.h"
#include "implem_only.h"

#include <basis/byte_array.h>
#include <basis/mutex.h>
#include <basis/set.cpp>
#include <data_struct/bit_vector.h>
#include <data_struct/span_manager.h>
#include <mechanisms/time_stamp.h>


// holds onto the packet information for the buffer.

class buffer_pack_array : public array<byte_array>
{
public:
  buffer_pack_array(int num = 0) : array<byte_array>(num) {}
};


buffer::buffer(const buffer_key &key, byte_array &contents,
    int packet_size)
: _lock(new mutex),
  _owner_alerted(false),
  _alive(true),
  _maximum_packet(packet_size),
  _key(new buffer_key(key)),
  _actual_storage(new byte_array),
  _is_storage_active(true),
  _last_add(new time_stamp()),
  _attachment(new byte_array()),
  _packet_list(new buffer_pack_array(number_of_packets
       (int(contents.length()), packet_size))),
  _buffer_status(new span_manager(number_of_packets(int(contents.length()),
      packet_size))),
  _estimation(0),
  _references(new int_set)
{
  // snarf the contents out of the passed in array.
  _actual_storage->snarf(contents);

  // mark that all packets are in place already.
  int_array status_init(2);  // the range of the marking.
  status_init[0] = 0;
  status_init[1] = short(_packet_list->length() - 1);
  _buffer_status->update(status_init);  // state that whole buffer is in place.

  // loop through and point at places in the buffer.
  int offset = 0;
  for (int i = 0; i < _packet_list->length(); i++) {
    int len = packet_size;
    // last packet is a different length usually.
    if (i == _packet_list->length() - 1)
      len = last_packet_size(int(_actual_storage->length()), packet_size);
    // store the data.
    _packet_list->use(i).reset(len, _actual_storage->access() + offset);
    offset += len;  // update the place we're pointing at.
  }
}

buffer::buffer(const buffer_key &key, int total_packets, int maximum_packet)
: _lock(new mutex),
  _owner_alerted(false),
  _alive(true),
  _maximum_packet(maximum_packet),
  _key(new buffer_key(key)),
  _actual_storage(new byte_array()),
  _is_storage_active(false),
  _last_add(new time_stamp()),
  _attachment(new byte_array()),
  _packet_list(new buffer_pack_array(total_packets)),
  _buffer_status(new span_manager(total_packets)),
  _estimation(0),
  _references(new int_set)
{}

buffer::~buffer()
{
  release();
  WHACK(_attachment);
  WHACK(_buffer_status);
  WHACK(_key);
  WHACK(_last_add);
  WHACK(_packet_list);
  WHACK(_lock);
  WHACK(_references);
  WHACK(_actual_storage);  // done in release, but what the hey.
}

void buffer::reset_estimate() { _estimation = 0; }

bool buffer::did_estimate() const { return !!_estimation; }

int buffer::estimate() const { return _estimation; }

bool buffer::alive() const { return _alive; }

bool buffer::owner_alerted() const { return _owner_alerted; }

void buffer::set_alerted(bool owner_alerted)
{ _owner_alerted = owner_alerted; }

span_manager buffer::buffer_status() const
{ auto_synchronizer l(*_lock); return *_buffer_status; }

int buffer::references() const
{ auto_synchronizer l(*_lock); return _references->elements(); }

int_set buffer::get_references() const
{ auto_synchronizer l(*_lock); return *_references; }

void buffer::add_reference(int conn_id)
{ auto_synchronizer l(*_lock); *_references += conn_id; }

void buffer::remove_reference(int conn_id)
{ auto_synchronizer l(*_lock); *_references -= conn_id; }

void buffer::set_last_add(const time_stamp &new_last_add)
{ auto_synchronizer l(*_lock); *_last_add = new_last_add; }

time_stamp buffer::last_add() const
{ auto_synchronizer l(*_lock); return *_last_add; }

buffer_key buffer::key() const
{ auto_synchronizer l(*_lock); return *_key; }

void buffer::set_key(const buffer_key &to_set)
{ auto_synchronizer l(*_lock); *_key = to_set; }

void buffer::get_attachment(byte_array &storage) const
{ auto_synchronizer l(*_lock); storage = *_attachment; }

void buffer::set_attachment(const byte_array &new_attach)
{ auto_synchronizer l(*_lock); *_attachment = new_attach; }

int buffer::packets() const
{ auto_synchronizer l(*_lock); return _packet_list->length(); }

bool buffer::whole() const
{ auto_synchronizer l(*_lock); return _buffer_status->vector().whole(); }

bool buffer::empty() const
{ auto_synchronizer l(*_lock); return _buffer_status->vector().empty(); }

int_array buffer::packet_sizes() const
{
  int_array to_return;
  auto_synchronizer l(*_lock);
  for (int i = 0; i < _packet_list->length(); i++)
    to_return += _packet_list->get(i).length();
  return to_return;
}

void buffer::release()
{
  FUNCDEF("release");
  auto_synchronizer l(*_lock);
  _last_add->reset();
  for (int i = 0; i < _packet_list->length(); i++)
    _packet_list->use(i).reset();
  WHACK(_actual_storage);
  // reset all status indicators.
  _is_storage_active = false;  // no storage at all now.
  _key->connection = 0;
  _key->msg_id = 0;
  _packet_list->reset(0);
  _buffer_status->reset(0);
  _owner_alerted = false;
  _estimation = 0;  // reset our estimator.
  _alive = true;
}

istring buffer::text_form() const
{
  auto_synchronizer l(*_lock);
  return istring(istring::SPRINTF, "[%d] size=%d alive=%s conn_id=%d "
      "msg_id=%d whole?=%s mod_time=%s", _key->buff_id.raw_id(),
      (whole()? full_length() : (_estimation? _estimation : -1) ),
      _alive? "true" : "false", _key->connection, _key->msg_id,
      whole()? "true":"false", _last_add->text_form().observe());
  // note that -1 is used here as a kludgy way of saying "has not been
  // determined yet".
}

istring buffer::detailed_form(bool really_detailed) const
{
  FUNCDEF("detailed_form");
  auto_synchronizer l(*_lock);
  istring to_return(text_form() + istring(" "));
  to_return += _buffer_status->print_received_list();
  if (really_detailed) {
    to_return += " lengths: ";
    for (int i = 0; i < packets(); i++) {
      if (_buffer_status->vector().on(i))
        to_return += istring(istring::SPRINTF, "[%d=%d] ", i, 
            _packet_list->get(i).length());
      else to_return += istring(istring::SPRINTF, "[%d=empty] ", i);
    }
  }
  return to_return;
}

void buffer::release(int packet_number, release_style status_handling)
{
  FUNCDEF("release [pack/style]");
  auto_synchronizer l(*_lock);
  if (!_alive) return;
  bounds_return(packet_number, 0, _packet_list->length() - 1, );
  if (_is_storage_active) return;  // illegal to release inside large message.
  _last_add->reset();  // reset the stamp.
  // clear out any thoughts that a packet resided in this index.
  if (status_handling == RESET_STATUS)
    _buffer_status->vector().clear(packet_number);
  _packet_list->use(packet_number).reset();
}

outcome buffer::store_packet(int packet_number, const byte_array &buffer)
{
  FUNCDEF("store_packet");
  auto_synchronizer l(*_lock);
  if (!_alive) return BAD_INPUT;
  bounds_return(packet_number, 0, _packet_list->length() - 1, BAD_INPUT);
  if (_is_storage_active) return BAD_INPUT;
    // can't store into prealloced space.
  _last_add->reset();
  if (_buffer_status->vector().on(packet_number)) return IN_USE;
    // they can't overstore a packet; this is deemed an error.
  _packet_list->use(packet_number) = buffer;
  _buffer_status->vector().light(packet_number);  // this index is now in use.
  return OKAY;
}

byte_array buffer::get_packet(int packet_number) const
{
  FUNCDEF("get_packet");
  auto_synchronizer l(*_lock);
  if (!_alive) return byte_array();
  bounds_return(packet_number, 0, _packet_list->length() - 1, byte_array());
  return _packet_list->get(packet_number);
}

int buffer::packet_length(int packet_number) const
{
  FUNCDEF("packet_length");
  auto_synchronizer l(*_lock);
  if (!_alive) return 0;
  bounds_return(packet_number, 0, _packet_list->length() - 1, 0);
  return _packet_list->get(packet_number).length();
}

int buffer::full_length(int offset) const
{
  FUNCDEF("full_length");
  auto_synchronizer l(*_lock);
  if (!_alive) return 0;
  if (offset < 0) return BAD_INPUT;
  int to_return = 0;
  // make sure all packets exist.
  if (!whole()) return INCOMPLETE;
  for (int i = 0; i < _packet_list->length(); i++) {
    // compute the size of the packet by ignoring the offset portion.
    int add_in = _packet_list->get(i).length() - offset;
    if (add_in < 0) return BAD_INPUT;
    to_return += add_in;
  }
  return to_return;
}

int buffer::guess_size(int formal(offset))
{
  auto_synchronizer l(*_lock);
  // only recompute our guess if it's not already set up.
  if (!did_estimate())
    _estimation = _packet_list->length() * _maximum_packet;
  return _estimation;
}

outcome buffer::dump(byte_array &storage_location, int offset)
{
  FUNCDEF("dump");
  auto_synchronizer l(*_lock);
  if (!_alive) return BAD_INPUT;
  _last_add->reset();
  if (offset < 0) return BAD_INPUT;
  if (!whole()) return INCOMPLETE;
  int len = full_length(offset);
  if (len < 0) return outcomes(len);
  storage_location.reset(int(len));  // make the storage area the right size.
  if (!storage_location.observe()) return OUT_OF_MEMORY;

  int store_offset = 0;  // where we're at inside the storage location.
  // loop through and snag the packs in the buffer.
  for (int i = 0; i < _packet_list->length(); i++) {
    if (offset > _packet_list->get(i).length() - 1) return BAD_INPUT;
      // this is a failure in the choice of offsets.
    byte_array new_chunk = _packet_list->get(i).subarray(offset, 
        _packet_list->get(i).length() - 1);
    storage_location.overwrite(store_offset, new_chunk);
    // update the place we're pointing at.
    store_offset += new_chunk.length();
  }
  return OKAY;
}


#endif //BUFFER_IMPLEMENTATION_FILE

---