Producer/Consumer 큐 구현하기(std::vector와 Template를 함께 사용하기)

최근 개발중 Producer(생산자) / Consumer(소비자) Queue를 만들었다.

매우 간단하다.

두개의 뮤텍스로 각자의 큐를 관리하고 일정 시간마다 Productor에 쌓인 데이터를 Consumer로 전달한다.

이렇게 나누는 중요한 이유는 생산자와 소비자를 별도로 관리하기 위함이다.

생산자는 빠른 시간에 데이터를 생산자 큐에 전달하고 다음 작업을 진행한다.

소비자는 소비자 큐의 작업 목록을 순서대로 처리한다.

두 작업이 서로 간섭하지 않게 하기 위해 두개의 큐와 두개의 뮤텍스로 이 기능을 제공한다.

1초마다 생산자 큐를 비우고 소비자 큐에 추가하도록 작업 하였다.

문제는 해당 기능의 큐에 std::vector를 사용하였다.

이와 함께 Job을 다양화 하기 위해 Templete를 사용하였다.

개인적으로 Template는 기초적인 사용외에는 해본적이 없다.

결국 사방에서 문제가 발생했는데 현재까지 걸렸다 해결한 문제는 아래와 같다.

---

1. iterator 사용

• vector<T*>::iterator _iter = q.begin() 을 사용하면 아래와 같은 에러를 발생한다
• error: expected `;' before '_iter'
• 여기저기 뒤지다가 찾은 정답은 요기 에 가면 나온다
• typename vector<T*>::iterator _iter = q.begin()  <= typename 사용시
• class vector<T*>::iterator _iter = q.begin()  <= class 사용시

---

2. 매우 허접한 코드이지만 그래도 누군가에게 조금의 도움이라도 되길 바라며 올려 본다.

//---------------------------------------------------------------------------------------------------------------------
// pcqueue.hpp
//
// Asker's product consumer queue
//---------------------------------------------------------------------------------------------------------------------
#pragma once

#include <iostream>
#include <vector>
#include <pthread.h>

#include "errno.h"

#include "common.hpp"

using namespace std;
//---------------------------------------------------------------------------------------------------------------------
//---------------------------------------------------------------------------------------------------------------------
//---------------------------------------------------------------------------------------------------------------------
template <class T>
class CPCQueue
{
private:
	pthread_mutex_t ProducerLock;
	pthread_mutex_t ConsumerLock;

	vector<T*> ProducerQueue;
	vector<T*> ConsumerQueue;

private:
	bool LockProducer()
	{
		int _nError = pthread_mutex_lock( &ProducerLock );
		if ( _nError )
		{
			cout << "error in lock producer. return : " << _nError << ", err :" << strerror(errno) << endl;
			formatlog( LOG_CRIT, "error in lock producer. return(%d) ErrNo(%d) ErrMsg(%s)", _nError, errno, strerror(errno));
			return false;
		}
	
		return true;
	}

	bool UnlockProducer()
	{
		int _nError = pthread_mutex_unlock( &ProducerLock );
		if ( _nError )
		{
			cout << "error in unlock producer. return : " << _nError << ", err :" << strerror(errno) << endl;
			formatlog( LOG_CRIT, "error in unlock producer. return(%d) ErrNo(%d) ErrMsg(%s)", _nError, errno, strerror(errno));
			return false;
		}
	
		return true;
	}

	bool LockConsumer()
	{
		int _nError = pthread_mutex_lock( &ConsumerLock );
		if ( _nError )
		{
			cout << "error in lock consumer. return : " << _nError << ", err :" << strerror(errno) << endl;
			formatlog( LOG_CRIT, "error in lock consumer. return(%d) ErrNo(%d) ErrMsg(%s)", _nError, errno, strerror(errno));
			return false;
		}

		return true;
	}

	bool UnlockConsumer()
	{
		int _nError = pthread_mutex_unlock( &ConsumerLock );
		if ( _nError )
		{
			cout << "error in unlock consumer. return : " << _nError << ", err :" << strerror(errno) << endl;
			formatlog( LOG_CRIT, "error in unlock consumer. return(%d) ErrNo(%d) ErrMsg(%s)", _nError, errno, strerror(errno));
			return false;
		}

		return true;
	}


public:
	CPCQueue(){}
	virtual ~CPCQueue()
	{
		for ( class vector<T*>::iterator _iter = ProducerQueue.begin(); _iter != ProducerQueue.end(); _iter++)
			delete *_iter;

		for ( class vector<T*>::iterator _iter = ConsumerQueue.begin(); _iter != ConsumerQueue.end(); _iter++)
			delete *_iter;

		ProducerQueue.clear();
		ConsumerQueue.clear();
	}

	bool Init()
	{
		int _ErrorNo = pthread_mutex_init( &ProducerLock, NULL);
		if (_ErrorNo)
			return false;
	
		_ErrorNo = pthread_mutex_init( &ConsumerLock, NULL);
		if (_ErrorNo)
			return false;
	
		return true;
	}

	// Product에 하나를 추가한다
	bool AddProduct( T* _Job )
	{
		if ( false == LockProducer())
		{
			// lock 실패
			formatlog( LOG_CRIT, "CPCQueue::%s(%d) Lock Fail. fatal issue.", __func__, __LINE__);
			return false;
		}

		ProducerQueue.push_back(_Job);
	 
		if ( false == UnlockProducer())
		{
			// unlock 실패
			formatlog( LOG_CRIT, "CPCQueue::%s(%d) Unlock Fail. fatal issue.", __func__, __LINE__);
			return false;
		}

		return true;
	}

	// Consumer 쪽에서 하나씩 빼낸다.
	bool PopFromConsumer(T** _Value)
	{
		*_Value = NULL;

		if ( false == LockConsumer())
		{
			// lock 실패
			formatlog( LOG_CRIT, "CPCQueue::%s(%d) Lock Fail. fatal issue.", __func__, __LINE__);
			return false;
		}

		if (!ConsumerQueue.empty())
		{
			*_Value = ConsumerQueue.front();
			ConsumerQueue.erase(ConsumerQueue.begin());
		}

		if ( false == UnlockConsumer())
		{
			// unlock 실패
			formatlog( LOG_CRIT, "CPCQueue::%s(%d) Unlock Fail. fatal issue.", __func__, __LINE__);
			return false;
		}

		return true;
	}

	// 합할 것이 하나도 없는 경우 false를 반환한다
	bool MoveProducerToConsumer(int& _MoveCount)
	{
		if ( false == LockConsumer())
		{
			// lock 실패
			formatlog( LOG_CRIT, "CPCQueue::%s(%d) Lock Fail. fatal issue.", __func__, __LINE__);
			return false;
		}

		if ( false == LockProducer())
		{
			// lock 실패
			formatlog( LOG_CRIT, "CPCQueue::%s(%d) Lock Fail. fatal issue.", __func__, __LINE__);
			return false;
		}

		_MoveCount = ProducerQueue.size();
		if (!ProducerQueue.empty())
		{
			for ( class vector<T*>::const_iterator _iter = ProducerQueue.begin(); _iter != ProducerQueue.end(); _iter++)
				ConsumerQueue.push_back(*_iter);
	
			ProducerQueue.clear();
		}

		if ( false == UnlockProducer())
		{
			// unlock 실패
			formatlog( LOG_CRIT, "CPCQueue::%s(%d) Unlock Fail. fatal issue.", __func__, __LINE__);
			return false;
		}

		if ( false == UnlockConsumer())
		{
			// unlock 실패
			formatlog( LOG_CRIT, "CPCQueue::%s(%d) Unlock Fail. fatal issue.", __func__, __LINE__);
			return false;
		}

		return true;
	}

	int GetConsumerSize() const
	{
		return ConsumerQueue.size();
	}

};
//---------------------------------------------------------------------------------------------------------------------