netrq.cpp

Go to the documentation of this file.

/*
 * netrq.cpp
 *
 * Copyright (c) 2009 Thomas A. Vaughan
 * All rights reserved.
 *
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *     * Redistributions of source code must retain the above copyright
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright
 *       notice, this list of conditions and the following disclaimer in the
 *       documentation and/or other materials provided with the distribution.
 *     * Neither the name of the <organization> nor the
 *       names of its contributors may be used to endorse or promote products
 *       derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THOMAS A. VAUGHAN ''AS IS'' AND ANY
 * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL THOMAS A. VAUGHAN BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * Support for network request queues.  See netrq.h
 */

// includes --------------------------------------------------------------------
#include "netrq.h"              // always include our own header first

#include <list>

#include "common/wave_ex.h"
#include "perf/perf.h"
#include "threadsafe/smart_mutex.h"


namespace netrq {

/// \ingroup netrq
/*@{*/


// interface destructors
Request::~Request(void) throw() { }
Queue::~Queue(void) throw() { }



////////////////////////////////////////////////////////////////////////////////
//
//      static helper methods
//
////////////////////////////////////////////////////////////////////////////////


////////////////////////////////////////////////////////////////////////////////
//
//      NetRQ -- class that implements the Queue interface
//
////////////////////////////////////////////////////////////////////////////////

class NetRQ : public Queue {
public:
        // constructor, destructor ---------------------------------------------
        NetRQ(void) throw() { }
        ~NetRQ(void) throw() { }

        // public class methods ------------------------------------------------
        void initialize(void) { }

        // netrq::Queue class interface methods --------------------------------
        int size(void);
        eInsertResult addRequest(IN dword_t clock,
                                IN smart_ptr<Request>& request);
        bool getNextRequest(IN dword_t clock,
                                OUT smart_ptr<Request>& request);
        bool containsRequest(IN const char * id);
        bool removeRequest(IN const char * id);

private:
        // private typedefs ----------------------------------------------------
        struct request_data_t {
                // constructor, manipulators
                request_data_t(void) throw() { this->clear(); }
                void clear(void) throw() {
                                request = NULL;
                                clock = 0;
                        }

                // data fields
                smart_ptr<Request>              request;
                dword_t                         clock;
        };


        typedef std::map<std::string, smart_ptr<request_data_t> > id_map_t;
        typedef std::list<smart_ptr<request_data_t> > req_queue_t;
        typedef std::map<dword_t, smart_ptr<req_queue_t> > clock_map_t;

        // private helper methods ----------------------------------------------
        smart_ptr<request_data_t> getRequestById(IN const char * id);
        req_queue_t * getQueueForClock(IN dword_t clock);

        // private data members ------------------------------------------------
        id_map_t                m_idMap;        // look up requests by id
        clock_map_t             m_clockMap;     // requests sorted by clock
        smart_mutex             m_mutex;
};



////////////////////////////////////////////////////////////////////////////////
//
//      NetRQ -- netrq::Queue class interface methods
//
////////////////////////////////////////////////////////////////////////////////

int
NetRQ::size
(
void
)
{
        // lock!
        mlock l(m_mutex);

        // assume id map and clock map are in sync!
        return m_idMap.size();
}



NetRQ::eInsertResult
NetRQ::addRequest
(
IN dword_t clock,
IN smart_ptr<Request>& request
)
{
        ASSERT(clock > 0, "bad clock value: %lu", (long) clock);
        ASSERT(request, "null");

        const char * id = request->getId();
        ASSERT(id, "null ID from request?");

        // lock!
        mlock l(m_mutex);
        smart_ptr<request_data_t> prd = this->getRequestById(id);
        if (prd) {
                // already have this request!
                if (clock < prd->clock) {
                        return eQueue_Failed_CurrentEntryHasLaterClock;
                } else {
                        return eQueue_Failed_CurrentEntryHasEarlierClock;
                }
        }

        // allocate request data structure
        prd = new request_data_t;
        ASSERT(prd, "out of memory");
        prd->clock = clock;
        prd->request = request;

        // look up the current clock, and add a queue if necessary
        req_queue_t * prq = this->getQueueForClock(clock);
        ASSERT(prq, "null");

        prq->push_back(prd);
        m_idMap[id] = prd;
        ASSERT(prd.get_ref_count() > 1, "bad ref count");

        // okay!
        return eQueue_Success;
}



bool
NetRQ::getNextRequest
(
IN dword_t clock,
OUT smart_ptr<Request>& request
)
{
        ASSERT(clock > 0, "Bad clock: %lu", (long) clock);
        request = NULL;

        // lock!
        mlock l(m_mutex);

        // get whatever is at the head of the clock queue
        clock_map_t::iterator head = m_clockMap.begin();
        if (m_clockMap.end() == head) {
                return false;   // queue is empty!
        }
        if (head->first > clock) {
                return false;   // next request can wait
        }

        // there is something in the map!
        req_queue_t * prq = head->second;       // sorry about the names...
        ASSERT(prq, "null queue in clock map?");

        req_queue_t::iterator i = prq->begin();
        ASSERT(prq->end() != i, "empty queue in clock map?");

        smart_ptr<request_data_t> prd = *i;
        ASSERT(prd, "null request data in queue");

        // give data to caller
        request = prd->request;
        return true;
}



bool
NetRQ::containsRequest
(
IN const char * id
)
{
        ASSERT(id, "null");

        // see if we recognize this ID
        mlock l(m_mutex);
        id_map_t::iterator i = m_idMap.find(id);
        return (m_idMap.end() != i);
}



bool
NetRQ::removeRequest
(
IN const char * id
)
{
        ASSERT(id, "null");

        // lock!
        mlock l(m_mutex);

        // do we even have this request?
        id_map_t::iterator iditer = m_idMap.find(id);
        if (m_idMap.end() == iditer) {
                return false;           // don't know about this id!
        }

        // got it, so remove it
        request_data_t * prd = iditer->second;
        ASSERT(prd, "null request data in id map");

        // look for this clock
        clock_map_t::iterator ic = m_clockMap.find(prd->clock);
        ASSERT(m_clockMap.end() != ic,
            "request in ID map but no entry for clock");
        req_queue_t * prq = ic->second;
        ASSERT(prq, "null queue in clock map");

        // have to walk the queue...
        req_queue_t::iterator irq;
        for (irq = prq->begin(); irq != prq->end(); ++irq) {
                request_data_t * iprd = *irq;
                ASSERT(iprd, "null request data in queue");
                ASSERT(iprd->request, "null");
                if (!strcmp(iprd->request->getId(), id))
                        break;
        }
        ASSERT(prq->end() != irq,
            "request in ID map but not found in queue, id=%s", id);
        prq->erase(irq);
        if (!prq->size()) {
                // queue is now empty, so delete this clock entry
                m_clockMap.erase(ic);
        }

        // finally, nuke from ID map
        m_idMap.erase(iditer);
        return true;    // found it, nuked it
}



////////////////////////////////////////////////////////////////////////////////
//
//      Public APIs
//
////////////////////////////////////////////////////////////////////////////////

smart_ptr<NetRQ::request_data_t>
NetRQ::getRequestById
(
IN const char * id
)
{
        ASSERT(id, "null");

        // NOTE: this routine is private, and assumes the caller has locked!
        id_map_t::iterator i = m_idMap.find(id);
        if (m_idMap.end() == i)
                return NULL;            // id not found

        smart_ptr<request_data_t> prd = i->second;
        ASSERT(prd, "null request data in id map");
        return prd;
}



NetRQ::req_queue_t *
NetRQ::getQueueForClock
(
IN dword_t clock
)
{
        // NOTE: this routine is private, and assumes the caller has locked!
        clock_map_t::iterator i = m_clockMap.find(clock);
        if (m_clockMap.end() != i) {
                // yes, we have a queue for this clock!
                req_queue_t * prq = i->second;
                ASSERT(prq, "null queue in clock map");
                return prq;
        }

        // no queue found for clock!  create one
        smart_ptr<req_queue_t> rq = new req_queue_t;
        ASSERT(rq, "out of memory");

        // give the clock map a reference, and return to caller
        m_clockMap[clock] = rq;
        ASSERT(2 == rq.get_ref_count(), "bad ref count");
        return rq;
}



////////////////////////////////////////////////////////////////////////////////
//
//      Public APIs
//
////////////////////////////////////////////////////////////////////////////////

smart_ptr<Queue>
Queue::create
(
void
)
{
        smart_ptr<NetRQ> local = new NetRQ;
        ASSERT(local, "out of memory");

        local->initialize();

        return local;
}



int
sendMessagesFromQueue
(
IO xdrbuf::Output * output,
IN dword_t clock,
IO Queue * queue
)
{
        ASSERT(output, "null");
        // ASSERT(clock) -- we don't care!
        ASSERT(queue, "null");

        // xdr seems to pad added buffers?
        const int xdrPadding = 8;

        // keep looping while we have messages and room
        int count = 0;
        smart_ptr<Request> request;
        while (queue->getNextRequest(clock, request)) {
                ASSERT(request, "null");
                int remain = output->getRemainingBytes();
        //      DPRINTF("%d bytes remaining in output buffer", remain);
                int req = request->getMaxBytes();
                if (remain - req < xdrPadding) {
                        break;          // can't add any more...
                }

                // okay, write this out
        //      int left = remain - req;
        //      DPRINTF("  Adding request: %s", request->getId());
        //      DPRINTF("    request bytes: %d (leaves %d?)", req, left);
                request->write(output);

                // remove request
                queue->removeRequest(request->getId());
                ++count;

                // re-add?
                int retry = request->getRetryInterval();
                if (retry > 0)
                        queue->addRequest(clock + retry, request);
        }

        return count;
}



};      // netrq namespace