FsfCell.h

Go to the documentation of this file.

/*
  FSF: Flow Scheduling Framework library
  
  Version 1.0 Copyright (C) 2010 Alexandre R.J. Francois

  Previous versions Copyright (C) 2001-2005 University of Southern California

  Author: Alexandre R.J. Francois - alexandrefrancois@yahoo.com
  www.alexandrefrancois.org
  
  Modular Flow Scheduling Middleware
  mfsm.sourceForge.net
  
  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.
  
  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  Lesser General Public License for more details.
  
  You should have received a copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/

#ifndef FSF_CELL_H
#define FSF_CELL_H

#include "Fsf.h"
#include "FsfFilter.h"
#include "FsfRepository.h"


namespace fsf{


    
    class CCell {
    private:
        bool m_bActive; 
        bool m_bReset;  

        
        CMutex m_csOutputName; 
        std::string m_strOutputName; 
        bool m_bPassThrough; 

        
        // Active path
        CCondition m_condActiveThreads; 
        long m_nNbActiveThreads; 
        // Filter
        fsf::CMutex m_csActiveFilter; 
        CActiveFilterBase *m_pActiveFilter; 
        // Connections
        CCell *m_pUpstreamCell; 
        fsf::CMutex m_csDownstream; 
        std::list<CCell*> m_listDownstream; 
        
        // Passive path
        CCondition m_condPassiveThreads; 
        long m_nNbPassiveThreads; 
        // Filter
        fsf::CMutex m_csPassiveFilter; 
        CPassiveFilterBase *m_pPassiveFilter; 
        // Handle: results of the filtering
        fsf::CMutex m_csPassiveHandle; 
        CPassiveHandle *m_pPassiveHandle; 
        // Connection
        CRepository *m_pRepository; 
        

    protected:


        void lockOutputName() { m_csOutputName.lock(); }
        void unlockOutputName() { m_csOutputName.unlock(); }
        const std::string &getOutputName() const { return m_strOutputName; }


        void setActive(bool bActive) { m_bActive=bActive; }

        void lockDownstream() { m_csDownstream.lock(); }
        void unlockDownstream() { m_csDownstream.unlock(); }
        std::list<CCell*> &getListDownstream() { return m_listDownstream; }     

        void lockActiveFilter() { m_csActiveFilter.lock(); }
        void unlockActiveFilter() { m_csActiveFilter.unlock(); }
        CActiveFilterBase *getActiveFilter() { return m_pActiveFilter; }
        
        void lockPassiveFilter() { m_csPassiveFilter.lock(); }
        void unlockPassiveFilter() { m_csPassiveFilter.unlock(); }

        void lockPassiveHandle() { m_csPassiveHandle.lock(); }
        void unlockPassiveHandle() { m_csPassiveHandle.unlock(); }
        CPassiveHandle *getPassiveHandle() { return m_pPassiveHandle; }
        void deletePassiveHandle(){
            m_csPassiveHandle.lock();
            delete m_pPassiveHandle;
            m_pPassiveHandle=NULL;
            m_csPassiveHandle.unlock();
        }



    public:


        CCell(); 
        virtual ~CCell(); 




        void setRepository(CRepository *pRepository) { m_pRepository=pRepository; }
        void setActiveFilter(CActiveFilterBase *pFilter);
        void setPassiveFilter(CPassiveFilterBase *pFilter);
        void setOutputName(const std::string &strOutputName){
            lockOutputName();
            m_strOutputName.assign(strOutputName);
            unlockOutputName();
        }


        bool connectDownstreamCell(CCell *pCell);
        bool disconnectDownstreamCell(CCell *pCell=NULL);

        // name Flow control
        virtual bool switchCell(bool bOn); 
        bool on(); 
        bool off(); 
        virtual void setReset(bool b=true) { m_bReset=b; } 

        CActiveFilterBase *cloneActiveFilter();
        CPassiveFilterBase *clonePassiveFilter();
        void getOutputName(std::string &strOutputName){
            lockOutputName();
            strOutputName.assign(m_strOutputName);
            unlockOutputName();
        }





        virtual void getTypeID(std::string &str) const { str.assign("FSF_CELL"); }
        
        CRepository *getRepository() const { return m_pRepository; }
        // Stream connections
        CCell *getUpstreamCell() const { return m_pUpstreamCell; }

        bool isActive() const { return m_bActive; }

        bool doReset() const { return m_bReset; }
        //*}
        



        bool connectRepository(fsf::CRepository *pRepository);
        bool disconnectRepository();
        bool connectUpstreamCell(fsf::CCell *pUpstreamCell);
        bool disconnectStream();



        virtual void process(CPassiveHandle *pPassiveHandle, CActiveHandle *pActiveHandle, CActivePulse *pActivePulse) {}





        void lockActiveThreads() { m_condActiveThreads.lock(); }
        void unlockActiveThreads() { m_condActiveThreads.unlock(); }
        void waitActiveThreads() { m_condActiveThreads.wait(); }
        void broadcastActiveThreads() { m_condActiveThreads.broadcast(); }


        void lockPassiveThreads() { m_condPassiveThreads.lock(); }
        void unlockPassiveThreads() { m_condPassiveThreads.unlock(); }
        void waitPassiveThreads() { m_condPassiveThreads.wait(); }
        void broadcastPassiveThreads() { m_condPassiveThreads.broadcast(); }



        inline void startActiveThread(CActivePulse *pPulse=NULL);

        virtual void activeThread(CActivePulse *pPulse=NULL);
        
        inline void startPassiveThread(CPassivePulse *pPulse=NULL);

        virtual bool passiveThread(CPassivePulse *pPulse=NULL);

        inline static void* activeThreadProc(void* pParam);
        //. Routes the pulse if necessary and calls <code>passiveThread</code>
        inline static void* passiveThreadProc(void* pParam);


    };

    inline void* CCell::activeThreadProc(void* pParam){
        void **pThreadParam=(void**)pParam;
        CCell *pCell=(CCell*)pThreadParam[0];
            
        CActivePulse *pPulse=(CActivePulse*)pThreadParam[1];
        pCell->activeThread((CActivePulse*)pPulse);
            
        // cleanup
        delete [] pThreadParam;
            
        return NULL;
    }

    inline void CCell::startActiveThread(CActivePulse *pPulse){
        if(m_bActive){
            // start active thread
            void **pParam=new void*[2];
            pParam[0]=this;
            pParam[1]=pPulse;

            pthread_t thread;
            pthread_create(&thread,NULL,&activeThreadProc,pParam);
            pthread_detach(thread);
        }
        else{
            // complete pulse routing
            if(m_bPassThrough && !m_listDownstream.empty()){
                std::list<CCell*>::iterator it;
                lockDownstream();
                pPulse->incNbPaths(m_listDownstream.size()-1);
                for(it=m_listDownstream.begin();it!=m_listDownstream.end();++it){
                    (*it)->startActiveThread(pPulse);
                }
                unlockDownstream();
            }
            else{
                // release the Pulse and delete if no longer needed
                if(pPulse->decNbPaths()==0)
                    delete pPulse;
            }
        }
    }

    inline void* CCell::passiveThreadProc(void* pParam){
        // restore calling instance pointer
        void **pThreadParam=reinterpret_cast<void**>(pParam);
        CCell *pCell=static_cast<CCell*>(pThreadParam[0]);
        CPassivePulse *pPulse=static_cast<CPassivePulse*>(pThreadParam[1]);
            
        // call thread function
        pCell->passiveThread(pPulse);

        // cleanup
        delete [] pThreadParam;
            
        return NULL;
    }

    inline void CCell::startPassiveThread(CPassivePulse *pPulse){
        void **pParam=new void*[2];
        pParam[0]=this;
        pParam[1]=pPulse;
        
        pthread_t thread;
        pthread_create(&thread,NULL,&passiveThreadProc,pParam);
        pthread_detach(thread);
    }
    
} // namespace fsf

#endif // FSF_CELL_H

---