common/bidirconn.cc

Go to the documentation of this file.

//
//
//
// ***** 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 <typeinfo>

#include "consmgr.h"
#include "bidirconn.h"

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

#if SHOULD_NOT_BE_HERE
// XXX - Move to Connection::connect?  Maybe a wrapper in Connection
// to call the relevant subclass member?

// This is called to start the reader thread for an outgoing connection.
// It invokes Outlet::establish which will make the writer thread, then
// hands over control to BiDirConn::establish, which will maintain
// and monitor the connection.
void *
BiDirConn::establish_wrapper(void *instance)
{
    BiDirConn   *me = static_cast<BiDirConn*>(instance);

    // Sub-class initialization/connection done by Outlet::establish()
    Outlet::establish(me);

    me->establish();

    // (Generally, this is unlikely to return...)
    return(NULL);
}
#endif  /* SHOULD_NOT_BE_HERE */

void *
BiDirConn::reader_thread(void *instance)
{
    void        *ret;
    BiDirConn   *self = static_cast<BiDirConn*>(instance);

    // BiDirConn::reader() won't return, unless something odd has
    // happened.  It just keeps reading and passing data until it can't.
    try {
        ret = self->reader();
    } catch (pthread_exception &pe) {
        cerr << "* reader thread (" << pthread_self() << ") terminating; ";
        cerr << pe.what() << endl;
    } catch (fireball &fb) {
        cerr << "* reader thread (" << pthread_self() << ") terminating; ";
        fb.bail(cerr);
    } catch (std::exception &e) {
        cerr << "reader thread (" << pthread_self() <<
            ") caught std::exception: " << e.what() << endl;
    } catch (string &e) {
        cerr << "reader thread (" << pthread_self() <<
            ") caught string exception: " << e << endl;
    } catch (char *e) {
        cerr << "reader thread (" << pthread_self() <<
            ") caught char* exception: " << e << endl;
    } catch (long l) {
        cerr << "reader thread (" << pthread_self() <<
            ") caught long exception: " << l << endl;
    } catch (unsigned long l) {
        cerr << "reader thread (" << pthread_self() <<
            ") caught unsigned long exception: " << l << endl;
    } catch (short l) {
        cerr << "reader thread (" << pthread_self() <<
            ") caught short exception: " << l << endl;
    } catch (unsigned short l) {
        cerr << "reader thread (" << pthread_self() <<
            ") caught unsigned short exception: " << l << endl;
    } catch (int l) {
        cerr << "reader thread (" << pthread_self() <<
            ") caught int exception: " << l << endl;
    } catch (unsigned int l) {
        cerr << "reader thread (" << pthread_self() <<
            ") caught unsigned int exception: " << l << endl;
    } catch (char l) {
        cerr << "reader thread (" << pthread_self() <<
            ") caught char exception: " << l << endl;
    } catch (unsigned char l) {
        cerr << "reader thread (" << pthread_self() <<
            ") caught unsigned char exception: " << l << endl;
#ifndef __linux__
    } catch (...) {
        // XXX - How the hell do I figure out what this is?!?
        cerr << "reader thread (" << pthread_self() <<
            ") caught unknown exception" << endl;
#endif
    }

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

    return(ret);
}

/*
 * Real member functions...
 */

BiDirConn::BiDirConn(ConnectorP &ourConnector)
        : Outlet::Outlet(ourConnector), tid_reader((pthread_t)NULL)
{
    // Don't create/invoke/configure things here.  Our sub-classes need
    // to take care of configuration, because Connection::Connection will
    // invoke the reader thread implemented here.....
 
    return;
}

BiDirConn::~BiDirConn(void) throw()
{
// XXX - The destructor shouldn't do any of this.  shutdown() should've
// been called.  Should perhaps we call shutdown from the destructor?
// Would that even make sense?
#if SHOULD_NOT_BE_HERE
    int         ret;

    // Lock the mutex...
#if 0
    MUTEX_LOG("Checking the lock on " << &mutex);
    ret = pthread_mutex_trylock(&mutex);
    if (ret == EBUSY) {
        MUTEX_LOG(&mutex << " is already locked,"
#ifdef __NetBSD__
             << " locked by thread " << pthread__id(mutex.ptm_owner) << ";"
#endif
             << " my thread id is " << pthread_self());
        // This is wrong.  Someone's got this mutex locked, and we're
        // about to destroy it?  I think a thread exited without unlocking
        // the mutex.  Can that happen?
        MUTEX_LOG("Waiting for lock on " << &mutex);
        ret = pthread_mutex_lock(&mutex);
    }
#else
    MUTEX_LOG("Locking " << &mutex);
    ret = pthread_mutex_lock(&mutex);
#endif  /* 0 */
    PTHREAD_CHECK_AND_THROW(ret, "mutex_lock(mutex)");
    MUTEX_LOG(&mutex << " is locked.");

#if 0
    // Close down the reader fd
    if ((is_connected()) && (read_fd >= 0)) {
        close(read_fd);
        read_fd = -1;
    }
#endif  /* 0 */

    if (tid_reader) {
        // Flush the reader thread.  It's probably already exited, but
        // how can we tell?  pthread_kill(0) is unreliable...  :-(
        clog << "In BiDirConn::~BiDirConn, joining tid_reader ("
             << tid_reader << ")" << endl;
        ret = pthread_join(tid_reader, NULL);
        clog << "BiDirConn::~BiDirConn, pthread_join of reader returned "
             << ret << endl;
    }

    // And unlock the mutex...
    ret = pthread_mutex_unlock(&mutex);
    PTHREAD_CHECK_AND_THROW(ret, "mutex_unlock(mutex)");
    MUTEX_LOG(&mutex << " is unlocked again.");
#endif  /* SHOULD_NOT_BE_HERE */

    clog << "BiDirConn::~BiDirConn returning" << endl;
    return;
}

// XXX - The management section of the below code should be in
// Connection::manage(), and the "bring up a connection" part
// into sub-class implentations of the pure-virtual Connection::connect().
#if SHOULD_NOT_BE_HERE
// This is the bulk of the establish() functionality.  It will establish
// the reader thread, then stick around to monitor the connection and
// manage anything that needs to be managed.
void
BiDirConn::establish(void)
{
    int         ret;

    // I don't think we can not be connected here, but...
    if (is_connected()) {
        // Start the reader thread in addition to the writer thread
        // created by Outlet::establish()
        clog << "** Creating the reader thread..." << endl;
        ret = pthread_create(&tid_reader, NULL, reader_wrapper, this);
        PTHREAD_CHECK_AND_THROW(ret, "pthread_create(tid_reader)");
        clog << "BiDirConn::establish created reader thread ("
             << tid_reader << ")" << endl;
    } else {
        clog << "** Bogon!  \"" << get_id() << ";" << this <<
            "\" not connected!" << endl;
        return;
    }

    // Now, hang out watching for messages, so we can manage these
    // threads, much like Connection::manage() does...
    MUTEX_LOG("Locking mutex " << &mutex);
    ret = pthread_mutex_lock(&mutex);
    PTHREAD_CHECK_AND_THROW(ret, "mutex_lock(mutex)");
    MUTEX_LOG("Mutex " << &mutex << " locked");
    try {
        while (is_connected(true)) {
            // It seems that if we're in pthread_cond_wait(), and we're
            // canceled, we may leave the mutex locked, and be exited.
            // XXX - the right solution is likely to avoid any need of
            // ever pthread_cancel()ing anything, but.....
            pthread_cleanup_push(cleanup_unlock, &mutex);

            MUTEX_LOG("cond_wait'ing with " << &mutex);
            ret = pthread_cond_wait(&state_change, &mutex);

            PTHREAD_CHECK_AND_THROW(ret, "cond_wait(state_change)");
            // Remove the mutex-unlocking cleanup function without calling it
            pthread_cleanup_pop(0);

#if 0
            // XXX - When called from a destructor, things can be
            // "screwed up" here, and get_id(), amongst others, could
            // be a pure virtual function, because the relevant sub-
            // classes destructor has already run.  How can we catch
            // this here?
            clog << "BiDirConn(" << this << ")::establish(" << get_id()
                 << "; " << pthread_self() << "): State has changed ";
#else
            clog << "BiDirConn(" << this << ")::establish("
                 << pthread_self() << "): State has changed ";
#endif

            if (is_connected(true)) {
                // Still connected?  Must mean we have control-message(s).
                clog << "(still connected)" << endl;

                // XXX - Do stuff.
                clog << "BiDirConn::establish: got a control message?!?" << endl;
            }
        }
    } catch (...) {
        MUTEX_LOG("Unlocking " << &mutex);
        pthread_mutex_unlock(&mutex);
        throw;
    }

    clog << "(disconnected; nuking threads)" << endl;

    // Destroy each of the child threads (reader & writer), and
    // detach them so that their resources can be recycled...
    try {
        ret = pthread_mutex_unlock(&mutex);
        PTHREAD_CHECK_AND_THROW(ret, "mutex_unlock(mutex)");
        shutdown();
    } CATCH_STR_OR_DIE(72) {
        cerr << "Umm, unknown error at " << __FILE__ << ":"
         << (__LINE__ - 2) << endl;
        exit(73);
    }

    return;
}
#endif /* SHOULD_NOT_BE_HERE */

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

    clog << "In BiDirConn::stop_engine() of (" << this->get_id() << ")" << endl;

    clog << "(my thread_id is " << pthread_self() << ", gonna kill " << tid_reader << ")" << endl;

    try {
        // Disconnect before shutting down the reader [and writer] threads
        // XXX - Should we be calling disconnect on the Connector?  or otherwise
        // just destroying it?
        if (base->connected())
            base->disconnect();
        
        // XXX - Damn.  pthread_kill()/pthread_cancel() seem to not
        // necessarily work for me, at least not portably.  I'm gonna
        // have to make my own way to figure out if they thread has
        // exited.  Mur...
        if (tid_reader && (pthread_kill(tid_reader, 0) == 0)) {
            // Try to cancel it, but it may already have exited, in
            // which case, pthread_cancel() will return ESRCH...
            clog << "Canceling reader thread (" << tid_reader << ")" << endl;
            ret = pthread_cancel(tid_reader);
            clog << "pthread_cancel of reader (" << tid_reader << ") returned " << ret << endl;
            if (ret == 0) {
                ret = pthread_detach(tid_reader);
                if (ret == EINVAL) {
                    // The cancel worked, but the thread isn't "joinable".
                    clog << "stop_engine(" << this->get_id() << "): reader thread "
                        << "pthread_detach() failed after pthread_cancel().  "
                        << "Assuming adequate death." << endl;
                } else
                    PTHREAD_CHECK_AND_THROW(ret, "detach(tid_reader)");
                clog << "BiDirConn::stop_engine() of (" << this->get_id()
                    << "); canceled reader thread" << endl;
            } else if (ret != ESRCH)
                PTHREAD_CHECK_AND_THROW(ret, "cancel(tid_reader)");
        } else
            clog << "tid_reader is " << tid_reader << ", but kill returned 0" << endl;
    } catch (CMException &e) {
        clog << "Consmgr exception in BiDirConn::stop_engine(): " << e.what() << endl;
    } catch (...) {
        cerr << "BiDirConn produced unexpected exception!" << endl;
    }
    
    clog << "BiDirConn::stop_engine() successful, calling Outlet::stop_engine()" << endl;

    Outlet::stop_engine();

    return;

}

void
BiDirConn::start_engine(void)
{
    int         ret;

    ret = pthread_create(&tid_reader, NULL, reader_thread, this);
    PTHREAD_CHECK_AND_THROW(ret, "pthread_create(tid_reader)");

    clog << "** BiDirConn::reader() started up, thread id " << tid_reader
         << std::endl;

    // Now call our super-classes Outlet::start_engine()
    Outlet::start_engine();

    return;
}

void *
BiDirConn::reader()
{
    int         ret;

    clog << "Reader thread..." << endl;

    clog << "Reader (" << this->get_id() << "; " << pthread_self() << ") about to read..." << endl;
    /* Now, with mutexes unlocked, call our Connector's read method... */
    do {
        try {
            /* This will block waiting for available data */
            Data        *d = base->read();

            clog << "read " << d->len() << " bytes (" << base->get_id()
                << "): " << *d << std::endl;
            
            // Now put this bit of data onto the queues of everyone in the
            // list of listeners...
            
            // We should process this data before sending it to our receivers,
            // in case it's not really data that needs to be passed on...
            process_data(d);
            
            /* If there's data left after processing, send it to our recipients */
            if (d->len() > 0) {
                MessageP        msg(new Data(*d));
                
                clog << "Reader (" << this->get_id() << ") about to send message (" << msg << ") to all our recipients" << endl;
                // Add this data Message to all of the recipients data queue's...
                this->send_message(msg);
            }
        } catch (CMStateChange &e) {
            cerr << "Got a state change (disconnect?) from Connector::read()"
                 << "; sending message to MasterController." << endl;
            note_disconnect();
            // XXXXXXXXX
            break;
        } catch (...) {
            cerr << "Connector::read() failed within " << base->get_id()
                 << ", rethrowing." << endl;
            throw;
        }


        clog << "Reader (" << this->get_id() << ") about to read..." << endl;
    } while (1);
clog << "Reader: How'd I get here?" << endl;

#if SHOULD_NOT_BE_HERE
    /* XXX - This code needs to be moved up into the catch() cases */
    /* from the call to Connector::read().  That's where these */
    /* conditions will need to be dealt with.  */
    if (len < 0) {
        if ((len == -1) && (errno == EBADF)) {
            clog << "Reader (" << this->get_id() << ") read -1 (EBADF)" << endl;
            // Umm, this happens when things are closed down.  Just exit.
        } else
            throw(fireball(9, "BiDirConn::reader - read(2)", true));
    } else if (len == 0) {
        cerr << "EOF on BiDirConn::reader() (" << this->get_id() << "; " << pthread_self() << ")" << endl;
        // Disconnect everything, or in this case, just note that it's
        // disconnected itself and clean up...
        this->disconnect();
    }
#endif  /* SHOULD_NOT_BE_HERE */

    clog << "Reader thread, exiting..." << endl;

    return(NULL);
}

void
BiDirConn::process_data(Data *d)
{
    /* XXX What should we do here?  Check for a Message, or would 
     * that belong in Messageable?  Protocol handling? */
    clog << "BiDirConn::process_data(): we're " << get_id() << " (" << typeid(this).name() << "): Doing nothing" << endl;
    
    return;
}

// Add an Encoder to the set of encoders with which to process all
// data before it is sent to our receivers...
void
BiDirConn::add_encoder(Encoder *e)
{
    encoders.insert(e);

    return;
}

std::string
BiDirConn::get_id(void) const
{
    return(string("BiDirConn above ")+base->get_id());
}

---