Commit 207b9160 authored by Marco De Marco's avatar Marco De Marco
Browse files

Worker thread, event buffer, event thread and Configuration added

parent d246fe24

Makefile

+2 −1
Original line number Diff line number Diff line
@@ -16,7 +16,8 @@ CC=g++ 
CXX_DEBUG_FLAGS=-g -DVERBOSE_DEBUG

CXX_RELEASE_FLAGS=-O3

CXX_DEFAULT_FLAGS=-c -Wall -Wextra -std=c++11 -std=gnu++11

LDFLAGS=-Wall -lomniORB4 -lomniDynamic4 -lCOS4 -lomnithread -ltango -llog4tango

LDFLAGS=-Wall -lomniORB4 -lomniDynamic4 -lCOS4 -lomnithread -ltango -llog4tango \

	-lboost_thread -lboost_filesystem -lboost_system

INC_PARM=$(foreach d, $(INC_DIR), -I$d)

LIB_PARM=$(foreach d, $(LIB_DIR), -L$d)

#================================================================================

src/ClassFactory.cpp

+0 −1
Original line number Diff line number Diff line 
/*----- PROTECTED REGION ID(PreProcessor::ClassFactory.cpp) ENABLED START -----*/

static const char *RcsId = "$Id:  $";

//=============================================================================

//

// file :        ClassFactory.cpp

src/Configuration.h

0 → 100644
+83 −0
Original line number Diff line number Diff line 
#ifndef CONFIGURATION_H

#define CONFIGURATION_H



#include <iostream>

#include <stdint.h>

#include <vector>

#include <boost/shared_ptr.hpp>



namespace PreProcessor_ns

{



class Configuration

{

public:

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

//	[Public] Shared pointer typedef

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

	typedef boost::shared_ptr<Configuration> SP;



private:

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

//	[Private] Constructor destructor deleter

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

	Configuration(std::string watchPath, int workerNumber, int sleepTime, int waitTime,

		int connectionNumber, uint32_t iNotifyMask): m_watchPath(watchPath),

		m_workerNumber(workerNumber), m_sleepTime(sleepTime), m_waitTime(waitTime),

		m_connectionNumber(connectionNumber), m_iNotifyMask(iNotifyMask) {}

	virtual ~Configuration() {}



	class Deleter;

	friend class Deleter;

	class Deleter

	{

	public:

		void operator()(Configuration* c) { delete c; }

	};



public:

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

//	[Public] User methods

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

	static Configuration::SP create(std::string watchPath, int workerNumber,

        int sleepTime, int waitTime, int connectionNumber, uint32_t iNotifyMask)

	{

		Configuration::SP c_sp(new Configuration(watchPath, workerNumber, sleepTime,

			 waitTime, connectionNumber, iNotifyMask), Configuration::Deleter());



		return c_sp;

	}



	std::string getWatchPath() const { return m_watchPath; }

	unsigned int getWorkerNumber() const { return m_workerNumber; }

    unsigned int getSleepTime() const { return m_sleepTime; }

	unsigned int getWaitTime() const { return m_waitTime; }

	unsigned int getConnectionNumber() const { return m_connectionNumber; }

	uint32_t getINotifyMask() const { return m_iNotifyMask; }



private:

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

//	[Private] class variables

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

	//INotify watch path

	const std::string m_watchPath;



	//Worker thread number

	const unsigned int m_workerNumber;



	//Event thread sleep time

	const unsigned int m_sleepTime;



	//Worker thread wait time

	const unsigned int m_waitTime;



	//Number of connection per destination

	const unsigned int m_connectionNumber;



	//INotify mask

	const uint32_t m_iNotifyMask;

};



}   //End of namespace



#endif /*!CONFIGURATION_H*/

src/EventBuffer.cpp

0 → 100644
+166 −0
Original line number Diff line number Diff line 
#include <EventBuffer.h>



#include <boost/thread/locks.hpp>



namespace PreProcessor_ns

{



//==============================================================================

//	EventBuffer::EventBuffer()

//==============================================================================

EventBuffer::EventBuffer(Tango::DeviceImpl* deviceImpl_p) :

	Tango::LogAdapter(deviceImpl_p)

{

		DEBUG_STREAM << "EventBuffer::EventBuffer()" << endl;

}



//==============================================================================

//	EventBuffer::~EventBuffer()

//==============================================================================

EventBuffer::~EventBuffer()

{

	DEBUG_STREAM << "EventBuffer::~EventBuffer()" << endl;

}



//==============================================================================

//	EventBuffer::insertNew()

//==============================================================================

EventBuffer::SP EventBuffer::create(Tango::DeviceImpl* deviceImpl_p)

{

    EventBuffer::SP e_sp(new EventBuffer(deviceImpl_p),

        EventBuffer::Deleter());



    return e_sp;

}



//==============================================================================

//	EventBuffer::insertNew()

//==============================================================================

void EventBuffer::insertNew(boost::filesystem::path path)

{

    DEBUG_STREAM << "EventBuffer::insertNew()" << endl;



	boost::mutex::scoped_lock lock(m_mutex);



    bool inserted = m_buffer.insert(

        std::pair<boost::filesystem::path, EventStatus>(path, UNPROCESSED) ).second;



	if(inserted)

    {

        DEBUG_STREAM << "EventBuffer::insertNew() element "

            << path.string() << " inserted" << endl;



        lock.unlock();



        m_conditionVariable.notify_all();

    }

    else

        WARN_STREAM << "EventBuffer::insertNew() element "

            << path.string() << " duplicated" << endl;

}



//==============================================================================

//	EventBuffer::waitNew()

//==============================================================================

boost::filesystem::path EventBuffer::waitNew()

{

    DEBUG_STREAM << "EventBuffer::waitNew()" << endl;



	boost::mutex::scoped_lock lock(m_mutex);



	do

	{

		std::map<boost::filesystem::path, EventStatus>::iterator it;

		for(it=m_buffer.begin(); it!=m_buffer.end(); it++)

		{

			if(it->second == UNPROCESSED)

			{

                DEBUG_STREAM << "EventBuffer::waitNew() found new element:"

                    << it->first.string() << endl;



				it->second = ASSIGNED;

				return it->first;

			}

		}



        DEBUG_STREAM << "EventBuffer::waitNew() waiting new element" << endl;



		m_conditionVariable.wait(lock);

	}

	while(true);

}



//==============================================================================

//	EventBuffer::markAsProcessed()

//==============================================================================

void EventBuffer::markAsProcessed(boost::filesystem::path path)

{

    DEBUG_STREAM << "EventBuffer::markAsProcessed()" << endl;



    boost::mutex::scoped_lock lock(m_mutex);



    std::map<boost::filesystem::path, EventStatus>::iterator it;



	it = m_buffer.find(path);



	if(it != m_buffer.end())

    {

        switch(it->second)

        {

            case UNPROCESSED:

                ERROR_STREAM << "EventBuffer::markAsProcessed() element "

                    << path.string() << " is marked not processed" << endl;

                break;

            case ASSIGNED:

                it->second = PROCESSED;

                DEBUG_STREAM << "EventBuffer::markAsProcessed() element "

                    << path.string() << " marked as processed" << endl;

                break;

            case PROCESSED:

                ERROR_STREAM << "EventBuffer::markAsProcessed() element "

                    << path.string() << " already marked as processed" << endl;

                break;

        }

    }

    else

        ERROR_STREAM << "EventBuffer::markAsProcessed() element"

            << path.string() << " not found" << endl;

}



//==============================================================================

//	EventBuffer::removeAllProcessed()

//==============================================================================

void EventBuffer::removeAllProcessed()

{

    DEBUG_STREAM << "EventBuffer::removeAllProcessed()" << endl;



    boost::mutex::scoped_lock lock(m_mutex);



    std::map<boost::filesystem::path, EventStatus>::iterator it;

    for(it=m_buffer.begin(); it!=m_buffer.end(); ++it)

    {

        if(it->second == PROCESSED)

        {

            DEBUG_STREAM << "EventBuffer::removeAllProcessed() element "

                << it->first << "will be removed" << endl;



            std::map<boost::filesystem::path, EventStatus>::iterator to_delete = it;



            m_buffer.erase(to_delete);

        }

    }

}



//==============================================================================

//	EventBuffer::size()

//==============================================================================

std::size_t EventBuffer::size()

{

    DEBUG_STREAM << "EventBuffer::size()" << endl;



	boost::mutex::scoped_lock lock(m_mutex);



	return m_buffer.size();

}



}   //namespace

src/EventBuffer.h

0 → 100644
+80 −0
Original line number Diff line number Diff line 
#ifndef EVENT_BUFFER_H

#define EVENT_BUFFER_H



#include <tango.h>



#include <map>

#include <vector>

#include <iostream>

#include <stdexcept>



#include <boost/shared_ptr.hpp>

#include <boost/thread/mutex.hpp>

#include <boost/thread/condition_variable.hpp>

#include <boost/filesystem.hpp>



namespace PreProcessor_ns

{



class EventBuffer : public Tango::LogAdapter

{

public:

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

//	[Public] Shared pointer typedef

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

	typedef boost::shared_ptr<EventBuffer> SP;



protected:

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

//	[Protected] Constructor destructor deleter

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

	EventBuffer(Tango::DeviceImpl*);



	virtual ~EventBuffer();



	class Deleter;

	friend class Deleter;

	class Deleter

	{

	public:

		void operator()(EventBuffer* e) { delete e; }

	};



public:

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

//	[Public] Users methods

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

	static EventBuffer::SP create(Tango::DeviceImpl*);



	virtual void insertNew(boost::filesystem::path);



	virtual boost::filesystem::path waitNew();



    virtual void markAsProcessed(boost::filesystem::path);



    virtual void removeAllProcessed();



	virtual std::size_t size();



protected:

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

//	[Protected] Event status enumeration

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

    enum EventStatus { UNPROCESSED=0, ASSIGNED=1, PROCESSED=2 };



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

//	[Protected] Class variables

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

	//Access synchronization mutex

	boost::mutex m_mutex;



	//Access synchronization condition variable

	boost::condition_variable m_conditionVariable;



	//File buffer

	std::map<boost::filesystem::path, EventStatus> m_buffer;

};



}   //End of namespace



#endif /*!EVENT_BUFFER_H*/