Defragment.h

Go to the documentation of this file.

#ifndef CASTOR_NET_DEFRAGMENT_H
#define CASTOR_NET_DEFRAGMENT_H 1

/*
 * $Id: Defragment.h 1679 2006-11-16 15:42:23Z kb $
 *
 *
 * Copyright 2008 Carpe Noctem, Distributed Systems Group,
 * University of Kassel. All right reserved.
 *
 * The code is derived from the software contributed to Carpe Noctem by
 * the Carpe Noctem Team.
 *
 * The code is licensed under the Carpe Noctem Userfriendly BSD-Based
 * License (CNUBBL). Redistribution and use in source and binary forms,
 * with or without modification, are permitted provided that the
 * conditions of the CNUBBL are met.
 *
 * You should have received a copy of the CNUBBL along with this
 * software. The license is also available on our website:
 * http://carpenoctem.das-lab.net/license.txt
 *
 *
 * The Castor socket class
 */

#include "Messages.h"
#include "DateTime.h"

#include <boost/shared_ptr.hpp>
#include <boost/thread/mutex.hpp>
#include <map>

namespace castor { namespace net {

        template<class Type, class Decoder>
                class Defragment {

                        protected:

                                typedef boost::shared_ptr<Type> TypePtr;
                                typedef boost::shared_ptr<Decoder> DecoderPtr;

                                struct FragmentInfo {
                                   
                                        TypePtr fragment;
                                        unsigned long long timestamp;
           
                                        FragmentInfo(TypePtr fragment) :
                                                fragment(fragment), timestamp(DateTime::getUtcNow().getTicks())
                                        {
                                        }

                                        FragmentInfo(const FragmentInfo &other) :
                                                fragment(other.fragment), timestamp(other.timestamp)
                                        {
                                        }
                                };
   
                                struct KeyInfo {
 
                                        int fragId;
                                        NetAddress address;
 
                                        KeyInfo(int fragId, NetAddress address) :
                                                fragId(fragId), address(address)
                                        {
                                        }
 
                                        KeyInfo(const KeyInfo &other) :
                                                fragId(other.fragId), address(other.address)
                                        {
                                        }
 
                                        friend bool operator<(const KeyInfo &one, const KeyInfo &other) {
                                                if (one.fragId < other.fragId) return true;
                                                if (one.fragId != other.fragId) return false;
                                                
                                                return (one.address < other.address);
                                        }
                                };

                                typedef std::map<KeyInfo, FragmentInfo> FragmentMap;

                                FragmentMap fragments;
                                TypePtr reference;

                                boost::mutex monitor;
                                
                                unsigned long long invPeriod;
                                unsigned long long lastInvalidated;

                        public:

                                Defragment(unsigned long long invPeriod) :
                                        fragments(), reference(TypePtr(new Type)),
                                        monitor(), invPeriod(invPeriod), lastInvalidated(0)
                                {
                                }

                                TypePtr getMessage(const char *data, size_t offset, size_t count, const NetAddress &address) {
        
                                        invalidate();
 
                                        DecoderPtr dec = DecoderPtr(new Decoder(data, offset, count));
 
                                        int fragId = this->reference->getFragmentId(dec);
 
                                        // If the message is not fragmented, return it immediately
                                        if (fragId == -1) {
                                                return boost::static_pointer_cast<Type>(this->reference->decodeAny(dec));
                                        }
                
                                        // Otherwise, try to locate and reassemble it
                                        boost::mutex::scoped_lock lock(this->monitor);
        
                                        KeyInfo ki(fragId, address);
        
                                        // If there is a fragment with this id already, apply the new fragment
                                        typename FragmentMap::iterator itr = this->fragments.find(ki);

                                        if (itr != this->fragments.end()) {

                                                TypePtr result = itr->second.fragment;
        
                                                if (result->decodePart(dec)) {
        
                                                        // If the fragment was successfully reassembled, remove and return it
                                                        this->fragments.erase(itr);
        
                                                        return result;
                                                }
        
                                        } else {
        
                                                // Add a new fragment
                                                TypePtr m = boost::static_pointer_cast<Type>(this->reference->decodeAnyPart(dec));
        
                                                this->fragments.insert(std::pair<KeyInfo, FragmentInfo>(ki, FragmentInfo(m)));
                                        }

                                        return TypePtr();
                                }


                        protected:
                                
                                void invalidate() {
                
                                        unsigned long long now = DateTime::getUtcNow().getTicks();
                
                                        if (((now - this->lastInvalidated) / 10000) > this->invPeriod) {
                
                                                boost::mutex::scoped_lock lock(this->monitor);

                                                for (typename FragmentMap::iterator itr = this->fragments.begin();
                                                         itr != this->fragments.end(); itr++)
                                                {
                                                        if (((now - itr->second.timestamp) / 10000) > this->invPeriod) {

                                                                typename FragmentMap::iterator itrNext = itr;
                                                                itrNext++;
                                                                
                                                                this->fragments.erase(itr);
                                                                
                                                                if (itrNext == this->fragments.end()) {
                                                                        break;
                                                                }

                                                                itr = itrNext;
                                                        }
                                                }
                                        }

                                        this->lastInvalidated = now;
                                }
                        };
} }

#endif /* CASTOR_SPICA_DEFRAGMENT_H */