A class that manages the details of persistent storage. More...

#include <Persistent_File_Allocator.h>

Collaboration diagram for TAO_Notify::Persistent_File_Allocator:

Public Member Functions
	Persistent_File_Allocator ()
	The constructor.
	~Persistent_File_Allocator ()
	The destructor.
bool	open (const ACE_TCHAR *filename, const size_t block_size=512)
void	shutdown ()
	Wait for pending I/O and terminate our work thread.
Persistent_Storage_Block *	allocate ()
Persistent_Storage_Block *	allocate_at (size_t block_number)
	Allocate a new Persistent_Storage_Block at a given address.
Persistent_Storage_Block *	allocate_nowrite ()
	Allocate a PSB that is marked to not persist.
void	used (size_t block_number)
	Mark a block as used, removing it from the free list.
void	free (size_t block_number)
	Mark a block number as able to be used again.
size_t	block_size () const
	Access block size.
bool	read (Persistent_Storage_Block *psb)
	Read data into a PSB.
bool	write (Persistent_Storage_Block *psb)
	Write this block to the file,.
ACE_OFF_T	file_size () const
	for information (unit test) only.
Private Member Functions
void	free_block (const size_t block_number)
	Free a previously assigned block.
bool	allocate_block (size_t &block_number)
	Find and allocate a free block.
void	shutdown_thread ()
	Wait for pending I/O to complete and shut our worker thread down safely.
void	run ()
	The worker's execution thread.
Static Private Member Functions
static ACE_THR_FUNC_RETURN	thr_func (void *arg)
Private Attributes
ACE_Thread_Manager	thread_manager_
Random_File	pstore_
Bit_Vector	free_blocks_
ACE_Unbounded_Queue < Persistent_Storage_Block * >	block_queue_
TAO_SYNCH_MUTEX	lock_
TAO_SYNCH_MUTEX	free_blocks_lock_
TAO_SYNCH_MUTEX	queue_lock_
bool	terminate_thread_
bool	thread_active_
ACE_SYNCH_CONDITION	wake_up_thread_

Detailed Description

A class that manages the details of persistent storage.

Maintains a free list, write queue, allocations of new blocks, reads, and writes. This class also manages a thread that performs background updating of a Random_File.

Todo:

this is too much for one class to do. It should be refactored.

we shouldn't arbitrarily use a thread.

Definition at line 127 of file Persistent_File_Allocator.h.

Constructor & Destructor Documentation

TAO_Notify::Persistent_File_Allocator::Persistent_File_Allocator ( )

The constructor.

Definition at line 127 of file Persistent_File_Allocator.cpp.

  : pstore_()
  , terminate_thread_(false)
  , thread_active_(false)
  , wake_up_thread_(queue_lock_)
{
}

TAO_Notify::Persistent_File_Allocator::~Persistent_File_Allocator ( )

The destructor.

Definition at line 135 of file Persistent_File_Allocator.cpp.

{
  this->shutdown_thread();
}

Member Function Documentation

Persistent_Storage_Block * TAO_Notify::Persistent_File_Allocator::allocate ( void )

Allocate a new Persistent_Storage_Block and initialize it to an unused block of storage.

Definition at line 160 of file Persistent_File_Allocator.cpp.

{
  Persistent_Storage_Block* result = 0;
  size_t block_number = 0;
  ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ace_mon, this->lock_, 0);
  if (!this->allocate_block(block_number))
  {
    //@@todo: this should never happen
    // why not.  What if the disk is full?  Oh, I see we
    // allocate non-existent blocks.  FIX this
  }
  if (DEBUG_LEVEL > 0) ACE_DEBUG ((LM_DEBUG,
    ACE_TEXT ("(%P|%t) Persistent_File_Allocator::allocate: %d\n"),
    static_cast<int> (block_number)
    ));
  result = this->allocate_at(block_number);
  return result;
}

Persistent_Storage_Block * TAO_Notify::Persistent_File_Allocator::allocate_at ( size_t block_number )

Allocate a new Persistent_Storage_Block at a given address.

Definition at line 180 of file Persistent_File_Allocator.cpp.

{
  Persistent_Storage_Block* result = 0;
  this->used(block_number);
  if (DEBUG_LEVEL > 0) ACE_DEBUG ((LM_DEBUG,
    ACE_TEXT ("(%P|%t) Persistent_File_Allocator::allocate at : %d\n"),
    static_cast<int> (block_number)
    ));
  ACE_NEW_RETURN(result, Persistent_Storage_Block(
    block_number,
    this->block_size()),
    0);
  return result;
}

bool TAO_Notify::Persistent_File_Allocator::allocate_block ( size_t & block_number ) [private]

Find and allocate a free block.

Definition at line 307 of file Persistent_File_Allocator.cpp.

{
  ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ace_mon, this->free_blocks_lock_, 0);
  block_number = this->free_blocks_.find_first_bit(false);
  return true;
}

Persistent_Storage_Block * TAO_Notify::Persistent_File_Allocator::allocate_nowrite ( )

Allocate a PSB that is marked to not persist.

Definition at line 196 of file Persistent_File_Allocator.cpp.

{
  Persistent_Storage_Block* result = 0;
  ACE_NEW_RETURN (result,
                  Persistent_Storage_Block (static_cast<size_t> (~0), 0),
                  0);
  result->set_no_write();

  return result;
}

size_t TAO_Notify::Persistent_File_Allocator::block_size ( ) const

Access block size.

Definition at line 231 of file Persistent_File_Allocator.cpp.

{
  return pstore_.block_size();
}

ACE_OFF_T TAO_Notify::Persistent_File_Allocator::file_size ( ) const

for information (unit test) only.

Definition at line 323 of file Persistent_File_Allocator.cpp.

{
  return this->pstore_.size ();
}

void TAO_Notify::Persistent_File_Allocator::free ( size_t block_number )

Mark a block number as able to be used again.

Definition at line 220 of file Persistent_File_Allocator.cpp.

{
  if (DEBUG_LEVEL > 0) ACE_DEBUG ((LM_DEBUG,
    ACE_TEXT ("(%P|%t) Persistent_File_Allocator::free: %d\n"),
    static_cast<int> (block_number)
    ));
  ACE_ASSERT (this->free_blocks_.is_set (block_number));
  this->free_block(block_number);
}

void TAO_Notify::Persistent_File_Allocator::free_block ( const size_t block_number ) [private]

Free a previously assigned block.

Definition at line 299 of file Persistent_File_Allocator.cpp.

{
  ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->free_blocks_lock_);
  ACE_ASSERT (this->free_blocks_.is_set (block_number));
  this->free_blocks_.set_bit(block_number, false);
}

bool TAO_Notify::Persistent_File_Allocator::open	(	const ACE_TCHAR *	filename,
		const size_t	block_size = `512`
	)

Definition at line 141 of file Persistent_File_Allocator.cpp.

{
  bool file_opened = this->pstore_.open(filename, block_size);
  if (file_opened)
  {
    this->thread_active_ = true;
    this->thread_manager_.spawn(this->thr_func, this);
  }
  return file_opened;
}

bool TAO_Notify::Persistent_File_Allocator::read ( Persistent_Storage_Block * psb )

Read data into a PSB.

Data will come either from the queue of blocks to be written, or it will be read from the file if there are no queued write requests for this block.

Definition at line 237 of file Persistent_File_Allocator.cpp.

{
  bool result = this->thread_active_;
  bool cached = false;
  if (result)
  {
    Persistent_Storage_Block** psbtemp = 0;
    {
      ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ace_mon, this->queue_lock_, false);
      size_t queue_size = this->block_queue_.size();
      for (size_t idx = 0; !cached && (idx < queue_size); ++idx)
      {
        // We want to start at the end of the queue and work backwards...
        size_t actual_block = (queue_size - idx) - 1;
        if ((0 == this->block_queue_.get(psbtemp, actual_block))
            && (psbtemp != 0))
        {
          cached = ((*psbtemp)->block_number() == psb->block_number());
        }
      }
      // this needs to be done in the guarded section
      if (cached && (0 != psbtemp))
      {
        ACE_OS::memcpy(psb->data(), (*psbtemp)->data(), this->block_size());
      }
    }
    if (!cached)
    {
      result = pstore_.read(psb->block_number(), psb->data());
    }
  }
  return result;
}

void TAO_Notify::Persistent_File_Allocator::run ( void ) [private]

The worker's execution thread.

Definition at line 345 of file Persistent_File_Allocator.cpp.

{
  // We need this because we could be working on writing data
  // when a call to terminate comes in!
  bool do_more_work = true;
  while (do_more_work)
  {
    do_more_work = false;
    Persistent_Storage_Block * blk = 0;
    {
      ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->queue_lock_);
      while (this->block_queue_.is_empty() && !terminate_thread_)
      {
        this->wake_up_thread_.wait();
      }
      // Awkward interface to peek at head of unbounded queue
      Persistent_Storage_Block ** pblk = 0;
      if (0 == this->block_queue_.get(pblk))
      {
        do_more_work = true;
        blk = *pblk;
      }
    }
    if (0 != blk)
    {
      Persistent_Callback *callback = blk->get_callback();
      if (!blk->get_no_write())
      {
        pstore_.write(blk->block_number(), blk->data(), blk->get_sync());
      }
      {
        Persistent_Storage_Block * blk2 = 0;
        ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->queue_lock_);
        this->block_queue_.dequeue_head (blk2);
        // if this triggers, someone pushed onto the head of the queue
        // or removed the head from the queue without telling ME.
        ACE_ASSERT (blk2 == blk);
      }
      // If we own the block, then delete it.
      if (blk->get_allocator_owns())
      {
        delete blk;
        blk = 0;
      }
      if (0 != callback)
      {
        callback->persist_complete();
      }
    }
  }
  this->terminate_thread_ = false;
  this->thread_active_ = false;
}

void TAO_Notify::Persistent_File_Allocator::shutdown ( void )

Wait for pending I/O and terminate our work thread.

Definition at line 154 of file Persistent_File_Allocator.cpp.

{
  this->shutdown_thread();
}

void TAO_Notify::Persistent_File_Allocator::shutdown_thread ( ) [private]

Wait for pending I/O to complete and shut our worker thread down safely.

Definition at line 329 of file Persistent_File_Allocator.cpp.

{
  if (this->thread_active_)
  {
    {
      ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->queue_lock_);
      this->terminate_thread_ = true;
      this->wake_up_thread_.signal();
    }
    this->thread_manager_.close();
    ACE_ASSERT (!this->terminate_thread_);
    ACE_ASSERT (!this->thread_active_);
  }
}

ACE_THR_FUNC_RETURN TAO_Notify::Persistent_File_Allocator::thr_func ( void * arg ) [static, private]

Used during thread startup to cast us back to ourselves and call the run() method.

Definition at line 315 of file Persistent_File_Allocator.cpp.

{
  Persistent_File_Allocator* pfa = static_cast<Persistent_File_Allocator*> (arg);
  pfa->run();
  return 0;
}

void TAO_Notify::Persistent_File_Allocator::used ( size_t block_number )

Mark a block as used, removing it from the free list.

Definition at line 208 of file Persistent_File_Allocator.cpp.

{
  ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->free_blocks_lock_);
  if (DEBUG_LEVEL > 0) ACE_DEBUG ((LM_DEBUG,
    ACE_TEXT ("(%P|%t) Persistent_File_Allocator::used: %d\n"),
    static_cast<int> (block_number)
    ));
  ACE_ASSERT (!this->free_blocks_.is_set (block_number));
  this->free_blocks_.set_bit(block_number, true);
}

bool TAO_Notify::Persistent_File_Allocator::write ( Persistent_Storage_Block * psb )

Write this block to the file,.

Add the Persistent_Storage_Block to our write queue and let the worker thread handle writing this to the Random_File.

Definition at line 272 of file Persistent_File_Allocator.cpp.

{
  bool result = this->thread_active_;
  if (result)
  {
    Persistent_Storage_Block* ourpsb = psb;
    if (!psb->get_allocator_owns())
    {
      if (DEBUG_LEVEL) ACE_DEBUG ((LM_DEBUG,
        ACE_TEXT ("(%P|%t) Copy PSB %d\n")
        , static_cast<int> (psb->block_number ())
        ));
      ACE_NEW_RETURN(ourpsb, Persistent_Storage_Block(*psb), false);
      ourpsb->set_allocator_owns(true);
    }
    ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ace_mon, this->queue_lock_, false);
    if (DEBUG_LEVEL) ACE_DEBUG ((LM_DEBUG,
      ACE_TEXT ("(%P|%t) Queueing PSB to write block %d\n")
      , static_cast<int> (psb->block_number ())
      ));
    result = (0 == this->block_queue_.enqueue_tail(ourpsb));
    this->wake_up_thread_.signal();
  }
  return result;
}