async_reception.cpp

/*   
 *   (c) Copyright 2008 Philipp Skadorov (philipp_s@users.sourceforge.net)
 *
 *   This file is part of FREESECS.
 *
 *   FREESECS is free software: you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation, either version 3 of the License, or
 *   (at your option) any later version.
 *
 *   FREESECS 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 General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with FREESECS, see COPYING.
 *   If not, see <http://www.gnu.org/licenses/>.
 */
#include <sys/socket.h>
#include <sys/un.h>
#include <errno.h>
#include <string.h>
#include <algorithm>
#include "hsms_signums.h"
#include "async_reception.h"
#include "tsqueue.h"

#include "log.h"

#define FSM_NAME "ASYNC READ"

#define ASYNC_RECEPTION_LOG_LEVEL 1000

#define max(a,b) ((a)>(b)?(a):(b))

using namespace freesecs;

int                         async_reception_t::_subscr_count    = 0;
int                         async_reception_t::_update_fd[2]    = {0,0};
freesecs::async_reception_t::thisv_t  async_reception_t::_thisv;
pthread_mutex_t             async_reception_t::_thisv_mx        = PTHREAD_MUTEX_INITIALIZER;
pthread_t                   async_reception_t::_sigthread_id    = 0;

freesecs::async_reception_t::async_reception_t(int fd, const char *name)
:_fd(fd)
,_buf(NULL)
,_bytes_to_read(0)
,_bytes_read(0)
,_rx_error(0)
,_state(IDLE)
{
    snprintf(_name, MAX_FSM_NAME, "ASYNC READ (%s)", name);
    if(0 == _subscr_count)
    {
        lock_t l(_thisv_mx);
        //open local socket here
        if(socketpair(PF_UNIX, SOCK_DGRAM, 0, _update_fd))
        {
            throw;
        }
    }

    if(1)
    {
        lock_t l(_thisv_mx);
        _thisv.push_back(this);
        _subscr_count++;


        if(0 == _sigthread_id)
        {
            if(pthread_create(&_sigthread_id, NULL, signalthread, (void*)this))
            {
                throw;
            }
        }

        set_need_update_fdset();
    }
}

freesecs::async_reception_t::~async_reception_t()
{
    if(0 == (_subscr_count--))
    {
        set_need_exit();
        pthread_join(_sigthread_id, NULL);
        
        _thisv.clear();
        shutdown(_update_fd[0], SHUT_RDWR);
        shutdown(_update_fd[1], SHUT_RDWR);
        close(_update_fd[0]);
        close(_update_fd[1]);
    }
    else
    {
        lock_t l(_thisv_mx);
        
        thisv_t::iterator it = std::find(_thisv.begin(), _thisv.end(), this);
        if(it != _thisv.end())
        {
            _thisv.erase(it);
        }
        set_need_update_fdset();
    }
}

void 
freesecs::async_reception_t::start_recv(void *buf, size_t count)
{
    event_data_t data;

    data.e0.buf     = buf;
    data.e0.count   = count;

    process(0, data);
}

void 
freesecs::async_reception_t::stop_recv()
{
    event_data_t data;
    process(2, data);
}

freesecs::async_reception_t::state_t
freesecs::async_reception_t::get_state()
{
    return _state;
}

size_t 
freesecs::async_reception_t::get_bytes_read()
{
    return _bytes_read;
}

void 
freesecs::async_reception_t::process(int e, event_data_t data)
{
    state_t state_old = _state;
    TRACE_FSM_BEGIN(ASYNC_RECEPTION_LOG_LEVEL, _name, _state, e);

    switch(_state)
    {
        case IDLE:
            if(0 == e)
            {
                _state = WTG_FOR_DATA;
            }
            else
            {
                goto sm_wrong_ev;
            }
        break;
        case WTG_FOR_DATA:
            if(1 == e)
            {
                _rx_error           =   data.e1.rx_error;
                _bytes_read         +=  data.e1.bytes_read;
                if(x1())
                {
                    z1(data);
                    _state = IDLE;
                }
                else if(x2())
                {
                    _state = RECEIVED;
                }
                else
                {
                    z2(data);
                }
            }
            else if(2 == e)
            {
                z3(data);
                _state = IDLE;
            }
            else
            {
                goto sm_wrong_ev;
            }
        break;
        case RECEIVED:
            if(0 == e)
            {
                _state = WTG_FOR_DATA;
            }
            else
            {
                goto sm_wrong_ev;
            }
        break;
        default:
            goto sm_wrong_ev;
        break;
    }

    if(state_old == _state)
    {
        goto sm_end;
    }

    TRACE_FSM_TRANS(ASYNC_RECEPTION_LOG_LEVEL, _name, state_old, _state);
    
    switch(_state)
    {
        case IDLE:
        break;
        case WTG_FOR_DATA:
            z1_0(data);
        break;
        case RECEIVED:
            z2_0(data);
        break;
        default:
        break;
    }

sm_end:
    TRACE_FSM_END(ASYNC_RECEPTION_LOG_LEVEL, _name, _state, e);
    return;

sm_wrong_ev:
    TRACE_FSM_ERROR(ASYNC_RECEPTION_LOG_LEVEL, _name, "event %d not handled in current state: %d", e, _state);
    TRACE_FSM_END(ASYNC_RECEPTION_LOG_LEVEL, _name, _state, e);
    return;
    
}

void 
freesecs::async_reception_t::z1_0(event_data_t data)
{
    _bytes_read = 0;

    _buf = data.e0.buf;
    _bytes_to_read = data.e0.count;
}

#define UPDATE_VAL  0xff
#define EXIT_VAL    0xfe

void 
freesecs::async_reception_t::set_need_update_fdset()
{
    char    data = UPDATE_VAL;
    send(_update_fd[0], &data, 1, 0);
}

void 
freesecs::async_reception_t::set_need_exit()
{
    char    data = EXIT_VAL;
    send(_update_fd[0], &data, 1, 0);
}

bool 
freesecs::async_reception_t::get_need_update_fdset()
{
    unsigned char    data;
    bool    need_update = false;

    if(1 == recv(_update_fd[1], &data, 1, MSG_DONTWAIT))
    {
        if(EXIT_VAL == data)
        {
            TRACE_DEBUG(ASYNC_RECEPTION_LOG_LEVEL, "async rcv: got exit command. Exiting...");
            pthread_exit(NULL);
        }
        need_update = true;
    }

    return need_update;
}


void 
freesecs::async_reception_t::update_fdset(int *fdn, fd_set *fds, thisv_t *thisv, bool clear_cache)
{
    typedef std::list<int> fdl_t;
    static fdl_t fdl;

    if(true == clear_cache)
    {
        lock_t l(_thisv_mx);

        fdl.clear();
        
        for(thisv_t::iterator it = _thisv.begin(); it != _thisv.end(); ++it)
        {
            fdl.push_back((*it)->_fd);
        }
    
        fdl.push_back(_update_fd[1]);
        fdl.sort();
    
        thisv->assign(_thisv.begin(), _thisv.end());
    }

    FD_ZERO (fds);
    *fdn = fdl.back();

    for(fdl_t::iterator it = fdl.begin(); it != fdl.end(); ++it)
    {
        FD_SET(*it, fds);
    }
}

void 
freesecs::async_reception_t::z2_0(event_data_t)
{
    data_recvd_signal(_bytes_read);
}
void 
freesecs::async_reception_t::z1(event_data_t)
{
    recv_error_signal(_rx_error);
}
void 
freesecs::async_reception_t::z2(event_data_t data)
{
    data_partially_recvd_signal(_bytes_read);
}
void 
freesecs::async_reception_t::z3(event_data_t)
{
}
bool 
freesecs::async_reception_t::x1()
{
    return 0 != _rx_error;
}
bool 
freesecs::async_reception_t::x2()
{
    return _bytes_read >= _bytes_to_read;
}

void *
freesecs::async_reception_t::signalthread(void *arg)
{
    int                 r, nfds = 0;
    fd_set              rd;
    bool                force_update = true;
    thisv_t             local_thisv;


    for(;;)
    {
        update_fdset(&nfds, &rd, &local_thisv, force_update);

        r = select(nfds + 1, &rd, NULL, NULL, NULL);

        if(0 == r)
        {
            continue;
        }
        else if(EBADF == r)
        {
            TRACE_DEBUG(ASYNC_RECEPTION_LOG_LEVEL, 
                        "async rcv thread: bad file "
                        "descriptor. Cleanup...");
            //TODO: remove bad fd from the list
        }
        else if(0 > r)
        {
            //TRACE_DEBUG("async rcv thread: pselect failed. Exiting...");
            //return NULL;
            TRACE_DEBUG(ASYNC_RECEPTION_LOG_LEVEL,
                        "async rcv thread: pselect "
                        "failed: %s. Renew fdset and restart..", strerror(errno));
            force_update = true;
            continue;
        }
        else
        {
            if(FD_ISSET(_update_fd[1], &rd))
            {
                force_update = true;
                TRACE_DEBUG(ASYNC_RECEPTION_LOG_LEVEL,
                            "async rcv thread: update signal.");
                if(true == get_need_update_fdset() && 1 == r)
                {
                    //only update sock has been written to,
                    //so just update fdset and continue polling
                    continue;
                }
            }
            else
            {
                force_update = false;
            }

            //use a copy of thisv for the case
            //thisv is updated when new async recv
            //objects are created from async_reception_t::check_fd_process
            //thisv copy is updated each update_fdset call
            for_each(local_thisv.begin(), local_thisv.end(), 
                    std::bind2nd(std::mem_fun(&async_reception_t::check_fd_process),&rd));
        }
    }

}

bool 
freesecs::async_reception_t::check_fd_process(const fd_set* fds)
{
    int             r = 0;
    event_data_t    event_data;

    bool can_read = 0 < FD_ISSET(_fd, fds);

    if(can_read)
    {
        TRACE_DEBUG(ASYNC_RECEPTION_LOG_LEVEL,
                    "%s: shall read %d bytes, read %d bytes", _name, _bytes_to_read, _bytes_read);
        if(_bytes_to_read)
        {
            if(_bytes_to_read > _bytes_read)
            {
                r = read(_fd, _buf, _bytes_to_read - _bytes_read);

                if(r < 0)
                {
                    TRACE_DEBUG(ASYNC_RECEPTION_LOG_LEVEL,
                        "%s: read error: %s", _name, strerror(errno));
                    event_data.e1.rx_error = errno;
                    event_data.e1.bytes_read = 0;
                }
                if(r == 0)//file empty/socket closed
                {
                    TRACE_DEBUG(ASYNC_RECEPTION_LOG_LEVEL,
                        "%s: read 0 bytes, errno: %s", _name, strerror(errno));
                    event_data.e1.rx_error = ECANCELED;
                    event_data.e1.bytes_read = 0;
                }
                else
                {
                    event_data.e1.rx_error = 0;
                    event_data.e1.bytes_read = r;
                }
            }
        }
        else
        {
            event_data.e1.rx_error = 0;
            event_data.e1.bytes_read = 0;
        }
        TRACE_DEBUG(ASYNC_RECEPTION_LOG_LEVEL,
                    "%s: read %d bytes", _name, r);
        process(1, event_data);
    }    
    else
    {
        //TRACE_DEBUG(ASYNC_RECEPTION_LOG_LEVEL, "%s: nothing to read", _name);
    }
    return can_read;
}

---