common/outlet.cc

Go to the documentation of this file.

//
// Outlet - This file implements the output-only portion of Connections.
// Methods and data here should be related to the receipt of data to be
// writen, and the writing of such data.
//
// $Id: outlet.cc 384 2008-05-08 18:09:56Z cross $
//
// ***** BEGIN LICENSE BLOCK *****
// Version: MPL 1.1/GPL 2.0/LGPL 2.1
// 
// The contents of this file are subject to the Mozilla Public License
// Version 1.1 (the "License"); you may not use this file except in
// compliance with the License. You may obtain a copy of the License at
// http://www.mozilla.org/MPL/
// 
// Software distributed under the License is distributed on an "AS IS"
// basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
// the License for the specific language governing rights and limitations
// under the License.
// 
// The Original Code is the consmgr network/serial-line monitoring package.
// 
// The Initial Developer of the Original Code is Chris P. Ross.
// Portions created by the Initial Developer are Copyright (C) 2000-2008
// the Initial Developer.  All Rights Reserved.
// 
// Contributor(s):
// 
// Alternatively, the contents of this file may be used under the terms of
// either the GNU General Public License Version 2 or later (the "GPL"), or
// the GNU Lesser General Public License Version 2.1 or later (the "LGPL"),
// in which case the provisions of the GPL or the LGPL are applicable instead
// of those above. If you wish to allow use of your version of this file only
// under the terms of either the GPL or the LGPL, and not to allow others to
// use your version of this file under the terms of the MPL, indicate your
// decision by deleting the provisions above and replace them with the notice
// and other provisions required by the GPL or the LGPL. If you do not delete
// the provisions above, a recipient may use your version of this file under
// the terms of any one of the MPL, the GPL or the LGPL.
// 
// ***** END LICENSE BLOCK *****

extern "C" {
#include <unistd.h>
}

#include <iostream>
#include <sstream>

#include "consmgr.h"
#include "outlet.h"
#include "guard.h"

#if SHOULD_NOT_BE_HERE
// The location of the static OutletList defined in the Outlet class
OutletList      Outlet::outlet_list;
#endif

/*
 * These static wrapper functions exist separate from any individual
 * instantiation of the Outlet class, or it's subclasses.  They are
 * here so that pthread_create() can call them, passing the relevant
 * Outlet* as the argument, and they can simply invoke the underlying
 * member functions.
 */

void *
Outlet::writer_thread(void *instance)
{
    Outlet      *me = (Outlet *)instance;

    // me->writer shouldn't ever really return, unless it was shut
    // down.  So, we really just hang out here...
    try {
        me->writer();
    } catch (pthread_exception &pe) {
        cerr << "* Writer thread (" << pthread_self() << ") terminating; ";
        cerr << pe.what() << endl;
        // XXX - Should I exit the whole program here, or just the thread?
    } catch (fireball &fb) {
        fb.bail(cerr);
#ifndef __linux__
    } catch (...) {
        cerr << "writer thread (" << pthread_self() <<
            ") caught unknown exception" << endl;
        // XXX - Should I exit the whole program here, or just the thread?
#endif
    }

    clog << instance << "::writer_wrapper (thread id " << pthread_self() << ") exiting..." << endl;

    return(NULL);
}


/*
 * Real member functions...
 */

Outlet::Outlet(ConnectorP &ourConnector)
        : Connection::Connection(ourConnector), tid_writer((pthread_t)NULL)
{
    int         ret;

    // Instantiate the conditioned wait element
    ret = pthread_cond_init(&data_avail, NULL);
    PTHREAD_CHECK_AND_THROW(ret, "cond_init(data_avail)");

    // Instantiate the data deque mutex
    ret = pthread_mutex_init(&data_mutex, NULL);
    DMUTEX_LOG("Mutex " << &data_mutex << " initialized, in object " << this);
    PTHREAD_CHECK_AND_THROW(ret, "mutex_init(data_mutex)");

    // Don't create/invoke/configure things here.  Our sub-classes need
    // to take care of configuration, and Connection::Connection will
    // invoke the writer thread implemented here.....
 
    return;
}

Outlet::~Outlet(void) throw()
{
    int         ret;

    clog << "In Outlet::~Outlet (thread " << pthread_self() << ")" << endl;

#if 0
    // Lock the mutex...
    MUTEX_LOG("Outlet::~Outlet locking mutex " << &mutex);
    ret = pthread_mutex_lock(&mutex);
    PTHREAD_CHECK_AND_THROW(ret, "mutex_lock(mutex)");
    MUTEX_LOG("Mutex " << &mutex << " locked.");

#if 0
    // Close down the writer fd
    if (write_fd >= 0) {
        close(write_fd);
        write_fd = -1;
    }
#endif

#warning "Need to figure out how to remove ourself from the other folkses recipient list"
#if SHOULD_NOT_BE_HERE
    // Remove myself from the master list of Outlet*'s
    Outlet::outlet_list.del_item(this);
#endif
    
    // Now, unlock our mutex...
    MUTEX_LOG("Outlet::~Outlet unlocking mutex " << &mutex);
    ret = pthread_mutex_unlock(&mutex);
    PTHREAD_CHECK_AND_THROW(ret, "mutex_unlock(mutex)");

#endif

    // After it's unlocked, we'll assume that it and all other mutex's are
    // unlocked, and can be destroyed.  But first make sure other threads
    // have a chance to close down.
    pthread_yield();
    usleep(5000);

    DMUTEX_LOG("Destroying data_mutex " << &data_mutex);
    ret = pthread_mutex_destroy(&data_mutex);
    DMUTEX_LOG("mutex_destroy(" << &data_mutex << ") returned " << ret);
    if (ret) {
        cerr << "Unable to destroy data_mutex " << &data_mutex << ": "
             << strerror(ret) << endl;
        mutex_info(std::cerr,&data_mutex);
    }

    return;
}

void
Outlet::writer(void)
{
    int         ret;

    DMUTEX_LOG("locking data mutex " << &data_mutex);
    Guard       locker(&data_mutex);
    DMUTEX_LOG("data mutex " << &data_mutex << " locked.");

    clog << "Writer thread..." << endl;

    // First, iterate through the queue, writing out any data already queued
    this->write_data();

    // Now that that's done, wait for the queue to have more data in it...
    // We don't need to verify that the deque is empty, since we just
    // emptied it without ever unlocking the mutex...
    clog << "Writer (" << this->get_id() << "; " << pthread_self() << ") waiting for queued data..." << endl;

    do {
        DMUTEX_LOG("cond_wait'ing with " << &data_mutex);
        ret = pthread_cond_wait(&data_avail, &data_mutex);
        PTHREAD_CHECK_AND_THROW(ret, "cond_wait(data_avail, data_mutex)");

        DMUTEX_LOG("back from cond_wait with " << &data_mutex << " locked");
        // Do we have more data?  We should.  If we don't, it's
        // because multiple people were waiting on the data, and
        // someone else got to it before we did...  Just re-wait.
        if (! data.empty()) {
            clog << "Writer (" << this->get_id() << "; " << pthread_self() << ") writing data..." << endl;
            this->write_data();
        } else {
            // Signaled, but no data available?  Maybe we've been
            // disconnected?  If not, just wait for more data...
            if (please_exit)
                break;
        }
    } while (ret == 0);

    clog << "Writer (" << this->get_id() << "; " << pthread_self() << ") exiting..." << endl;
    DMUTEX_LOG("Unlocking mutex " << &data_mutex << " upon method end");
    return;
}

// Actually do the writing.  This will iterate the data queue, writing
// all of the data to the fd passed as the argument...
// XXX - Note: the deque's mutex (this->data_mutex) must be
// locked when we're called, and is left that way when we return...
void
Outlet::write_data()
{
    int                 ret;
    queue_iterator      q_iter;

    for (q_iter = data.begin() ; q_iter != data.end() ;
         q_iter = data.erase(q_iter)) {
        /* Retrieve the pointer for the contained Message object, then */
        /* dereference it so that we can cast that reference to a */
        /* Data&.  XXX - We should be catch()'ing std::bad_cast here... */
        Data    &d = dynamic_cast<Data&>(*MessageP::GetPointer(*q_iter));

#if 0
        set<Decoder*>::iterator d_iter = decoders.begin();
        Decoder *d = *d_iter;

        if (d_iter == decoders.end()) {
            clog << "** Empty decoders set: d == " << d << endl;
        } else {
            // Set b up to be the new one, but save the old one to dec_ref
            b2 = b;
            b = (*d_iter)->decode(b2);
            b->inc_ref();
        }
#endif

        // Now, unlock the deque, so we don't keep it locked while in write(2)
        DMUTEX_LOG("Unlocking " << &data_mutex);
        ret = pthread_mutex_unlock(&data_mutex);
        PTHREAD_CHECK_AND_THROW(ret, "mutex_unlock(data_mutex)");

        clog << "gonna write Data " << d << " (len " << d.len() << ")" << endl;
        // Now that we're all unlocked, write the data...

        try {
            ret = base->write(d);
            clog << "Outlet::write of Data " << d << " returned " << ret << endl;
            if (ret < 0)
                throw(fireball(19, "write failed", true));
        } catch (CMStateChange &e) {
            stop_engine();
            note_disconnect();
        } catch (...) {
            throw;
        }
             
        clog << "written Data " << d << "(len " << d.len() << ")" << endl;

        // And lock again, letting the for() erase the element...
        DMUTEX_LOG("Locking " << &data_mutex << " for next for()");
        ret = pthread_mutex_lock(&data_mutex);
        PTHREAD_CHECK_AND_THROW(ret, "mutex_lock(data_mutex)");
    }

    DMUTEX_LOG("write_data returning, leaving " << &data_mutex << " locked");
    // And leave data_mutex locked when we return...
    return;
}

void
Outlet::stop_engine(void)
{
    int         ret;

    // XXX - hack
    // If tid_writer == 0, it may be because we're already most of
    // the way through the shutdown process.  Just note it and
    // return...
    if (tid_writer == 0) {
        clog << "In Outlet::stop_engine(); tid_writer == 0" << endl;
        return;
    }

    clog << "In Outlet::stop_engine() (" << this->get_id() << "; "
         << tid_writer << ")" << endl;

    //please_exit = true;

    if (tid_writer && !pthread_kill(tid_writer, 0)) {
        ret = pthread_cancel(tid_writer);
        PTHREAD_CHECK_AND_THROW(ret, "pthread_cancel(tid_writer)");
        ret = pthread_detach(tid_writer);
        PTHREAD_CHECK_AND_THROW(ret, "pthread_detach(tid_writer)");
        clog << "Canceled and detached writer thread (" << tid_writer
             << ")" << endl;
        tid_writer = (pthread_t)NULL;
    }

    clog << "Outlet::stop_engine() successful, returning..." << endl;

    return;
}

// Invoke the writer thread with pthread_create() and return.
void
Outlet::start_engine(void)
{
    int         ret;

    ret = pthread_create(&tid_writer, NULL, Outlet::writer_thread, this);
    PTHREAD_CHECK_AND_THROW(ret, "pthread_create(tid_writer)");
    clog << "Outlet writer wrapper thread created, id " << tid_writer << endl;

    return;
}

std::string
Outlet::get_id(void) const
{
    std::ostringstream  id_str;

    id_str << class_name() << " using " << base->get_id();

    return(id_str.str());
}

#if 0
// Add a Decoder to the set of decoders which will be called on
// each Blob of data after we receive it...
void
Outlet::add_decoder(Decoder *d)
{

    clog << "Inserting Decoder " << d << endl;

    decoders.insert(d);

    clog << "Testing..." << endl;

    set<Decoder*>::iterator d_iter = decoders.begin();

    clog << "*d_iter is " << *d_iter << endl;

    return;
}
#endif

void
Outlet::queue_data(MessageP &dmsg)
{
    int         ret;

    clog << "Trying to enqueue Message (type " << dmsg->name() << ")" << endl;

    // Ensure that this is data, not a message.  Otherwise just ignore it.
    // XXX - Should I throw an exception here?
    Data        *d = dynamic_cast<Data*>(MessageP::GetPointer(dmsg));

    if (d) {
        Guard   locker(&data_mutex);

        // Push the SmartPtr onto the queue...
        data.push_back(dmsg);

        clog << "Data queue'd and condition being broadcast on " << get_id() << endl;
        ret = pthread_cond_broadcast(&this->data_avail);
        PTHREAD_CHECK_AND_THROW(ret, "signal_broadcast(data_avail)");
    } else {
        clog << "Oops, Outlet::queue_data got a non-Data message..." << endl;
    }

    return;
}

---