Added unit-tests

SConstruct

@@ -22,4 +22,3 @@ Export('env satscons')
 SConscript(glob.glob("*/SConscript"))
 
 SatSCons.StandardTargets(env)
-SatSCons.AllTests(env)

Scheduler/Scheduler.cc

@@ -48,7 +48,7 @@ prefix_ satcom::lib::Scheduler::Scheduler()
         throw SystemException(errno);
 }
 
-prefix_ void satcom::lib::Scheduler::set(int fd, Callback const & cb, EventId eventMask)
+prefix_ void satcom::lib::Scheduler::add(int fd, Callback const & cb, EventId eventMask)
 {
     FdTable::iterator i (fdTable_.find(fd));
     int action (EPOLL_CTL_MOD);
@@ -57,34 +57,38 @@ prefix_ void satcom::lib::Scheduler::set(int fd, Callback const & cb, EventId ev
         i = fdTable_.insert(std::make_pair(fd, EventSpec())).first;
     }
 
-    if (eventMask | EV_READ)  i->second.cb_read = cb;
-    if (eventMask | EV_WRITE) i->second.cb_write = cb;
-    if (eventMask | EV_HUP)   i->second.cb_hup = cb;
-    if (eventMask | EV_ERR)   i->second.cb_err = cb;
+    if (eventMask & EV_READ)  i->second.cb_read = cb;
+    if (eventMask & EV_PRIO)  i->second.cb_prio = cb;
+    if (eventMask & EV_WRITE) i->second.cb_write = cb;
+    if (eventMask & EV_HUP)   i->second.cb_hup = cb;
+    if (eventMask & EV_ERR)   i->second.cb_err = cb;
 
     epoll_event ev;
     memset(&ev,0,sizeof(ev));
     ev.events = i->second.epollMask();
+    ev.data.fd = fd;
     
     if (epoll_ctl(epollFd_, action, fd, &ev)<0)
         throw SystemException(errno);
 }
 
-prefix_ void satcom::lib::Scheduler::unset(int fd, EventId eventMask)
+prefix_ void satcom::lib::Scheduler::remove(int fd, EventId eventMask)
 {
     FdTable::iterator i (fdTable_.find(fd));
     if (i == fdTable_.end()) 
         return;
 
-    if (eventMask | EV_READ)  i->second.cb_read = 0;
-    if (eventMask | EV_WRITE) i->second.cb_write = 0;
-    if (eventMask | EV_HUP)   i->second.cb_hup = 0;
-    if (eventMask | EV_ERR)   i->second.cb_err = 0;
+    if (eventMask & EV_READ)  i->second.cb_read = 0;
+    if (eventMask & EV_PRIO)  i->second.cb_prio = 0;
+    if (eventMask & EV_WRITE) i->second.cb_write = 0;
+    if (eventMask & EV_HUP)   i->second.cb_hup = 0;
+    if (eventMask & EV_ERR)   i->second.cb_err = 0;
 
     epoll_event ev;
     memset(&ev,0,sizeof(ev));
     ev.events = i->second.epollMask();
-    
+    ev.data.fd = fd;
+
     int action (EPOLL_CTL_MOD);
     if (ev.events==0) {
         action = EPOLL_CTL_DEL;
@@ -99,13 +103,53 @@ prefix_ int satcom::lib::Scheduler::EventSpec::epollMask()
     const
 {
     int mask (0);
-    if (cb_read)  mask |= EPOLLIN | EPOLLPRI;
+    if (cb_read)  mask |= EPOLLIN;
+    if (cb_prio)  mask |= EPOLLPRI;
     if (cb_write) mask |= EPOLLOUT;
     if (cb_hup)   mask |= EPOLLHUP;
     if (cb_err)   mask |= EPOLLERR;
     return mask;
 }
 
+prefix_ void satcom::lib::Scheduler::process()
+{
+    terminate_ = false;
+    while (! terminate_) {
+        struct epoll_event ev;
+        int events = epoll_wait(epollFd_, &ev, 1, 1000);
+        if (events<0)
+            throw SystemException(errno);
+        if (events==0)
+            continue;
+        
+        FdTable::iterator i = fdTable_.find(ev.data.fd);
+        BOOST_ASSERT (i != fdTable_.end() );
+        EventSpec const & spec (i->second); 
+
+        if (ev.events & EPOLLIN) {
+            BOOST_ASSERT(spec.cb_read); 
+            spec.cb_read(ev.data.fd, EV_READ);
+        }
+        else if (ev.events & EPOLLPRI) {
+            BOOST_ASSERT(spec.cb_prio);
+            spec.cb_prio (ev.data.fd, EV_PRIO);
+        }
+        else if (ev.events & EPOLLOUT) {
+            BOOST_ASSERT(spec.cb_write);
+            spec.cb_write(ev.data.fd, EV_WRITE);
+        }
+
+        else if (ev.events & EPOLLHUP) {
+            BOOST_ASSERT(spec.cb_hup);
+            spec.cb_hup(ev.data.fd, EV_HUP);
+        }
+        else if (ev.events & EPOLLERR) {
+            BOOST_ASSERT(spec.cb_err);
+            spec.cb_err(ev.data.fd, EV_ERR);
+        }
+    }
+}
+
 ///////////////////////////////cc.e////////////////////////////////////////
 #undef prefix_
 

Scheduler/Scheduler.cci

+// $Id$
+//
+// Copyright (C) 2006 
+// Fraunhofer Institut fuer offene Kommunikationssysteme (FOKUS)
+// Kompetenzzentrum fuer Satelitenkommunikation (SatCom)
+//     Stefan Bund <stefan.bund@fokus.fraunhofer.de>
+//
+// This program 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 2 of the License, or
+// (at your option) any later version.
+//
+// This program 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 this program; if not, write to the
+// Free Software Foundation, Inc.,
+// 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
+
+// Definition of inline non-template functions
+
+//#include "Scheduler.ih"
+
+// Custom includes
+
+#define prefix_ inline
+///////////////////////////////cci.p///////////////////////////////////////
+
+prefix_ void satcom::lib::Scheduler::terminate()
+{
+    terminate_ = true;
+}
+
+///////////////////////////////cci.e///////////////////////////////////////
+#undef prefix_
+
+
+// Local Variables:
+// mode: c++
+// c-file-style: "satcom"
+// End:

Scheduler/Scheduler.hh

@@ -50,8 +50,8 @@ namespace lib {
         // Types
 
         enum EventId { EV_NONE=0, 
-                       EV_READ=1, EV_WRITE=2, EV_HUP=4, EV_ERR=8, 
-                       EV_ALL=15 };
+                       EV_READ=1, EV_PRIO=2, EV_WRITE=4, EV_HUP=8, EV_ERR=16, 
+                       EV_ALL=31 };
         typedef boost::function<void (int fd, EventId event)> Callback;
 
         ///////////////////////////////////////////////////////////////////////////
@@ -69,8 +69,12 @@ namespace lib {
         ///@}
         ///////////////////////////////////////////////////////////////////////////
 
-        void set(int fd, Callback const & cb, EventId eventMask = EV_ALL);
-        void unset(int fd, EventId eventMask = EV_ALL);
+        void add(int fd, Callback const & cb, EventId eventMask = EV_ALL);
+        void remove(int fd, EventId eventMask = EV_ALL);
+
+        void process();
+
+        void terminate();
 
     protected:
 
@@ -80,6 +84,7 @@ namespace lib {
         struct EventSpec 
         {
             Callback cb_read;
+            Callback cb_prio;
             Callback cb_write;
             Callback cb_hup;
             Callback cb_err;
@@ -91,12 +96,13 @@ namespace lib {
 
         FdTable fdTable_;
         int epollFd_;
+        bool terminate_;
     };
 
 }}
 
 ///////////////////////////////hh.e////////////////////////////////////////
-//#include "Scheduler.cci"
+#include "Scheduler.cci"
 //#include "Scheduler.ct"
 //#include "Scheduler.cti"
 #endif

Scheduler/Scheduler.test.cc

@@ -26,6 +26,16 @@
 //#include "scheduler.test.ih"
 
 // Custom includes
+#include <sys/types.h>
+#include <signal.h>
+#include <sys/wait.h>
+#include <unistd.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <errno.h>
+#include <string.h>
+#include <iostream>
+
 #include "Scheduler.hh"
 
 #include <boost/test/auto_unit_test.hpp>
@@ -36,9 +46,179 @@
 
 using namespace satcom::lib;
 
+namespace {
+    
+    char const * SOCK_PATH = "/tmp/sched_test.sock";
+    
+    void error(char const * fn, char const * proc="")
+    {
+        std::cerr << "\n" << proc << fn << ": " << strerror(errno) << std::endl;
+    }
+
+    void fail(char const * fn)
+    {
+        error(fn,"server: ");
+        _exit(1);
+    }
+
+    void server()
+    {
+        int sock = socket(PF_UNIX,SOCK_STREAM,0);
+        if (sock<0) fail("socket");
+        struct sockaddr_un sun;
+        memset(&sun,0,sizeof(sun));
+        sun.sun_family = AF_UNIX;
+        strcpy(sun.sun_path,SOCK_PATH);
+        if (bind(sock,(struct sockaddr*)&sun,sizeof(sun))<0) fail("bind");
+        if (listen(sock,1)<0) fail("listen");
+        int conn = accept(sock,0,0);
+        if (conn < 0) fail("accept");
+
+        ///////////////////////////////////////////////////////////////////////////
+
+        if (write(conn,"READ",4)<0) fail("write");
+        char buffer[1024];
+        int size =  read(conn,buffer,1024);
+        if (size<0) fail("read");
+        if (size == 5) {
+            buffer[5] = 0;
+            if (strcmp(buffer,"WRITE")==0) {
+                if (write(conn,"OK",2)<0) fail("write");
+            } else
+                if (write(conn,"FAIL",4)<0) fail("write");
+        } else
+            if (write(conn,"FAIL",4)<0) fail("write");
+
+        ///////////////////////////////////////////////////////////////////////////
+
+        close(conn);
+        close(sock);
+    }
+
+    int start_server()
+    {
+        unlink(SOCK_PATH);
+        int pid = fork();
+        if (pid == 0) {
+            server();
+            _exit(0);
+        }
+        if (pid < 0) {
+            error("fork");
+            return 0;
+        }
+            
+        sleep(1); // Wait for the server socket to be opened
+        return pid;
+    }
+
+    bool stop_server(int pid)
+    {
+        sleep(1); // Wait for the server to terminate
+        if (kill(pid,SIGTERM)<0) {
+            error("kill");
+            return false;
+        }
+        int status = 0;
+        if (waitpid(pid,&status,0)<0) {
+            error("waitpid");
+            return false;
+        }
+        unlink(SOCK_PATH);
+        if (WIFSIGNALED(status)) {
+            std::cerr << "\nserver terminated with signal " << WTERMSIG(status) << std::endl;
+            return false;
+        }
+        if (WEXITSTATUS(status)!=0) {
+            std::cerr << "\nserver terminated with exit status " << WEXITSTATUS(status) << std::endl;
+            return false;
+        }
+        return true;
+    }
+
+    char buffer[1024];
+    int size;
+    int event;
+
+    void callback(int fd, Scheduler::EventId ev)
+    {
+        event = ev;
+        switch (event) {
+        case Scheduler::EV_READ:
+            size = recv(fd,buffer,1024,0);
+            break;
+        case Scheduler::EV_PRIO:
+            size = recv(fd,buffer,1024,MSG_OOB);
+            Scheduler::instance().terminate();
+            break;
+        case Scheduler::EV_WRITE:
+            size = write(fd,buffer,size);
+            Scheduler::instance().terminate();
+            break;
+        case Scheduler::EV_HUP:
+        case Scheduler::EV_ERR:
+        case Scheduler::EV_NONE:
+        case Scheduler::EV_ALL:
+            ;
+        }
+        Scheduler::instance().terminate();
+            
+    }
+        
+}
+
 BOOST_AUTO_UNIT_TEST(scheduler)
 {
+    int pid = start_server();
+    BOOST_REQUIRE (pid);
+
+    int sock = socket(PF_UNIX,SOCK_STREAM,0);
+    if (sock<0) {
+        error("socket");
+        BOOST_FAIL("socket");
+    }
+    struct sockaddr_un sun;
+    memset(&sun,0,sizeof(sun));
+    sun.sun_family = AF_UNIX;
+    strcpy(sun.sun_path,SOCK_PATH);
+    
+    if (connect(sock,(struct sockaddr*)&sun,sizeof(sun))<0) {
+        error("connect");
+        BOOST_FAIL("connect");
+    }
+
+    ///////////////////////////////////////////////////////////////////////////
+
     BOOST_CHECK_NO_THROW( Scheduler::instance() );
+
+    BOOST_CHECK_NO_THROW( Scheduler::instance().add(sock,&callback,Scheduler::EV_READ) );
+    event = Scheduler::EV_NONE;
+    BOOST_CHECK_NO_THROW( Scheduler::instance().process() );
+    BOOST_CHECK_EQUAL( event, Scheduler::EV_READ );
+    BOOST_REQUIRE_EQUAL( size, 4 );
+    buffer[size]=0;
+    BOOST_CHECK_EQUAL( buffer, "READ" );
+
+    BOOST_CHECK_NO_THROW( Scheduler::instance().add(sock,&callback,Scheduler::EV_WRITE) );
+    strcpy(buffer,"WRITE");
+    size=5;
+    event = Scheduler::EV_NONE;
+    BOOST_CHECK_NO_THROW( Scheduler::instance().process() );
+    BOOST_CHECK_EQUAL( event, Scheduler::EV_WRITE );
+
+    BOOST_CHECK_NO_THROW( Scheduler::instance().remove(sock,Scheduler::EV_WRITE) );
+    event = Scheduler::EV_NONE;
+    BOOST_CHECK_NO_THROW( Scheduler::instance().process() );
+    BOOST_CHECK_EQUAL( event, Scheduler::EV_READ );
+    BOOST_REQUIRE_EQUAL( size, 2 );
+    buffer[size]=0;
+    BOOST_CHECK_EQUAL( buffer, "OK" );
+
+    ///////////////////////////////////////////////////////////////////////////
+
+    close(sock);
+
+    BOOST_CHECK (stop_server(pid));
 }
 
 ///////////////////////////////cc.e////////////////////////////////////////