greenplumn CSyncPool 源码

greenplumn CSyncPool 代码

文件路径：/src/backend/gporca/libgpos/include/gpos/common/CSyncPool.h

//---------------------------------------------------------------------------
//	Greenplum Database
//	Copyright (C) 2011 EMC Corp.
//
//	@filename:
//		CSyncPool.h
//
//	@doc:
//		Template-based synchronized object pool class with minimum synchronization
//		overhead; it provides thread-safe object retrieval and release through
//		atomic primitives (lock-free);
//
//		Object pool is dynamically created during construction and released at
//		destruction; users retrieve objects without incurring the construction
//		cost (memory allocation, constructor invocation)
//
//		In order for the objects to be used in lock-free lists, the class uses
//		the clock algorithm to recycle objects.
//---------------------------------------------------------------------------
#ifndef GPOS_CSyncPool_H
#define GPOS_CSyncPool_H

#include "gpos/common/CAutoP.h"
#include "gpos/task/ITask.h"
#include "gpos/types.h"
#include "gpos/utils.h"

#define BYTES_PER_ULONG (GPOS_SIZEOF(ULONG))
#define BITS_PER_ULONG (BYTES_PER_ULONG * 8)

namespace gpos
{
//---------------------------------------------------------------------------
//	@class:
//		CSyncPool<class T>
//
//	@doc:
//		Object pool class (not thread-safe, despite the name)
//
//---------------------------------------------------------------------------
template <class T>
class CSyncPool
{
private:
	// memory pool
	CMemoryPool *m_mp;

	// array of preallocated objects
	T *m_objects;

	// bitmap indicating object reservation
	ULONG *m_objs_reserved;

	// bitmap indicating object recycle
	ULONG *m_objs_recycled;

	// number of allocated objects
	ULONG m_numobjs;

	// number of elements (ULONG) in bitmap
	ULONG m_bitmap_size;

	// offset of last lookup - clock index
	ULONG_PTR m_last_lookup_idx;

	// offset of id inside the object
	ULONG m_id_offset;

	// atomically set bit if it is unset
	BOOL
	SetBit(ULONG *dest, ULONG bit_val)
	{
		GPOS_ASSERT(nullptr != dest);

		ULONG old_val = *dest;

		// keep trying while the bit is unset
		while (0 == (bit_val & old_val))
		{
			ULONG new_val = bit_val | old_val;

			// attempt to set the bit
			if (*dest == old_val)
			{
				*dest = new_val;
				return true;
			}

			old_val = *dest;
		}

		return false;
	}

	// atomically unset bit if it is set
	BOOL
	UnsetBit(ULONG *dest, ULONG bit_val)
	{
		GPOS_ASSERT(nullptr != dest);

		ULONG old_val = *dest;

		// keep trying while the bit is set
		while (bit_val == (bit_val & old_val))
		{
			ULONG new_val = bit_val ^ old_val;

			// attempt to set the bit
			if (*dest == old_val)
			{
				*dest = new_val;
				return true;
			}

			old_val = *dest;
		}

		return false;
	}

public:
	CSyncPool(const CSyncPool &) = delete;

	// ctor
	CSyncPool(CMemoryPool *mp, ULONG size)
		: m_mp(mp),
		  m_objects(nullptr),
		  m_objs_reserved(nullptr),
		  m_objs_recycled(nullptr),
		  m_numobjs(size),
		  m_bitmap_size(size / BITS_PER_ULONG + 1),
		  m_last_lookup_idx(0),
		  m_id_offset(gpos::ulong_max)
	{
	}

	// dtor
	~CSyncPool()
	{
		if (gpos::ulong_max != m_id_offset)
		{
			GPOS_ASSERT(nullptr != m_objects);
			GPOS_ASSERT(nullptr != m_objs_reserved);
			GPOS_ASSERT(nullptr != m_objs_recycled);

#ifdef GPOS_DEBUG
			if (!ITask::Self()->HasPendingExceptions())
			{
				for (ULONG i = 0; i < m_numobjs; i++)
				{
					ULONG elem_offset = i / BITS_PER_ULONG;
					ULONG bit_offset = i % BITS_PER_ULONG;
					ULONG bit_val = 1 << bit_offset;

					BOOL reserved =
						(bit_val == (m_objs_reserved[elem_offset] & bit_val));
					BOOL recycled =
						(bit_val == (m_objs_recycled[elem_offset] & bit_val));

					GPOS_ASSERT((!reserved || recycled) &&
								"Object is still in use");
				}
			}
#endif	// GPOS_DEBUG

			GPOS_DELETE_ARRAY(m_objects);
			GPOS_DELETE_ARRAY(m_objs_reserved);
			GPOS_DELETE_ARRAY(m_objs_recycled);
		}
	}

	// init function to facilitate arrays
	void
	Init(ULONG id_offset)
	{
		GPOS_ASSERT(ALIGNED_32(id_offset));

		m_objects = GPOS_NEW_ARRAY(m_mp, T, m_numobjs);
		m_objs_reserved = GPOS_NEW_ARRAY(m_mp, ULONG, m_bitmap_size);
		m_objs_recycled = GPOS_NEW_ARRAY(m_mp, ULONG, m_bitmap_size);

		m_id_offset = id_offset;

		// initialize object ids
		for (ULONG i = 0; i < m_numobjs; i++)
		{
			ULONG *id = (ULONG *) (((BYTE *) &m_objects[i]) + m_id_offset);
			*id = i;
		}

		// initialize bitmaps
		for (ULONG i = 0; i < m_bitmap_size; i++)
		{
			m_objs_reserved[i] = 0;
			m_objs_recycled[i] = 0;
		}
	}

	// find unreserved object and reserve it
	T *
	PtRetrieve()
	{
		GPOS_ASSERT(gpos::ulong_max != m_id_offset &&
					"Id offset not initialized.");

		// iterate over all objects twice (two full clock rotations);
		// objects marked as recycled cannot be reserved on the first round;
		for (ULONG i = 0; i < 2 * m_numobjs; i++)
		{
			// move clock index
			ULONG_PTR index = (m_last_lookup_idx++) % m_numobjs;

			ULONG elem_offset = (ULONG) index / BITS_PER_ULONG;
			ULONG bit_offset = (ULONG) index % BITS_PER_ULONG;
			ULONG bit_val = 1 << bit_offset;

			// attempt to reserve object
			if (SetBit(&m_objs_reserved[elem_offset], bit_val))
			{
				// set id in corresponding object
				T *elem = &m_objects[index];

#ifdef GPOS_DEBUG
				ULONG *id = (ULONG *) (((BYTE *) elem) + m_id_offset);
				GPOS_ASSERT(index == *id);
#endif	// GPOS_DEBUG

				return elem;
			}

			// object is reserved, check if it has been marked for recycling
			if (bit_val == (bit_val & m_objs_recycled[elem_offset]))
			{
				// attempt to unset the recycle bit
				if (UnsetBit(&m_objs_recycled[elem_offset], bit_val))
				{
#ifdef GPOS_DEBUG
					BOOL recycled =
#endif	// GPOS_DEBUG

						// unset the reserve bit - must succeed
						UnsetBit(&m_objs_reserved[elem_offset], bit_val);

					GPOS_ASSERT(recycled &&
								"Object was reserved before being recycled");
				}
			}
		}

		// no object is currently available, create a new one
		T *elem = GPOS_NEW(m_mp) T();
		*(ULONG *) (((BYTE *) elem) + m_id_offset) = gpos::ulong_max;

		return elem;
	}

	// recycle reserved object
	void
	Recycle(T *elem)
	{
		GPOS_ASSERT(gpos::ulong_max != m_id_offset &&
					"Id offset not initialized.");

		ULONG offset = *(ULONG *) (((BYTE *) elem) + m_id_offset);
		if (gpos::ulong_max == offset)
		{
			// object does not belong to the array, delete it
			GPOS_DELETE(elem);
			return;
		}

		GPOS_ASSERT(offset < m_numobjs);

		ULONG elem_offset = offset / BITS_PER_ULONG;
		ULONG bit_offset = offset % BITS_PER_ULONG;
		ULONG bit_val = 1 << bit_offset;

#ifdef GPOS_DEBUG
		ULONG reserved = m_objs_reserved[elem_offset];
		GPOS_ASSERT((bit_val == (bit_val & reserved)) &&
					"Object is not reserved");

		BOOL mark_recycled =
#endif	// GPOS_DEBUG
			SetBit(&m_objs_recycled[elem_offset], bit_val);

		GPOS_ASSERT(mark_recycled &&
					"Object has already been marked for recycling");
	}

};	// class CSyncPool
}  // namespace gpos

#endif	// !GPOS_CSyncPool_H


// EOF

相关信息

greenplumn 源码目录

相关文章

greenplumn CAutoP 源码

greenplumn CAutoRef 源码

greenplumn CAutoRg 源码

greenplumn CAutoTimer 源码

greenplumn CBitSet 源码

greenplumn CBitSetIter 源码

greenplumn CBitVector 源码

greenplumn CDebugCounter 源码

greenplumn CDouble 源码

greenplumn CDynamicPtrArray 源码