S3FC project page

S3FC home page

---

               

---

s3_txport_tcp.tcc

Go to the documentation of this file.

/*
 * Stone Three Foundation Class (s3fc) provides a number of utility classes.
 * Copyright (C) 2001 by Stone Three Signal Processing (Pty) Ltd.
 *
 * Authored by Stone Three Signal Processing (Pty) Ltd.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library 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
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 *
 * Please see the file 'COPYING' in the source root directory.
 */

#include <s3fc/s3_periodic_notifier.h>
#include <s3fc/s3_event.h>
#include <s3fc/s3_txport_base.h>
#include <s3fc/s3_socket_tcp.h>
#include <cassert>
#include <memory>

#ifndef _WIN32
#include <netinet/in.h>
#endif

template<class T_Data, class T_Socket>
class s3_txport_tcp_tx_task : public s3_thread_base
{
   s3_semaphore sem;
   s3_fifo_queue<s3_txport_data<T_Data> >& in_q;
   s3_pack_buffer b;
   s3_txport_tcp<T_Data, T_Socket> *parent;
   s3_semaphore terminate;
   s3_periodic_notifier pn;
   std::list<s3_txport_client_id> id_list;
   
   s3_txport_tcp_tx_task(s3_fifo_queue<s3_txport_data<T_Data> >& a_q,
          s3_txport_tcp<T_Data, T_Socket> *a_parent);
   
   ~s3_txport_tcp_tx_task()
   {
      pn.unsubscribe(sem);
   }
   
   void handle_failure(s3_txport_client_id id, std::string err_msg)
   {
      // we just report errors
      parent->dispatch_event(
    s3_txport_event(s3_txport_event::error, id, err_msg));
      parent->disconnect_by_id(id);
   }

   void cleanup();
   
 public:
   void main_loop();

   void set_terminate()
   {
      terminate.post();
   }

   friend class s3_txport_tcp<T_Data, T_Socket>;
};

template<class T_Data, class T_Socket>
s3_txport_tcp_tx_task<T_Data, T_Socket>::s3_txport_tcp_tx_task(
   s3_fifo_queue<s3_txport_data<T_Data> >& a_q,
   s3_txport_tcp<T_Data, T_Socket> *a_parent) :
   in_q(a_q), parent(a_parent)
{
   // subscribe the queues we read to the notification semaphore.
   pn.set_period(500);
   pn.subscribe(sem);
}

template<class T_Data, class T_Socket>
void s3_txport_tcp_tx_task<T_Data, T_Socket>::cleanup()
{
   pn.disable();
   parent->meta_queue.unsubscribe_consumer(sem);
   in_q.unsubscribe_consumer(sem);
}

template<class T_Data, class T_Socket>
void s3_txport_tcp_tx_task<T_Data, T_Socket>::main_loop()
{
   typename s3_fifo_queue<s3_txport_data<T_Data> >::s3_queue_data_t *d;
   unsigned int sz = 0;

   pn.enable();
   parent->meta_queue.subscribe_consumer(sem);
   in_q.subscribe_consumer(sem);
   while (1)
   {
      if (terminate.try_wait()) return;
      b.reset();
      // Weed out any dead tasks
      parent->reap();
      // Now wait for something to happen
      sem.wait();
      
      // If we have metadata, send it. No while loop as a
      // semaphore will have been posted for each event.
      if (!parent->meta_queue.empty())
      {
    typename s3_txport_tcp<T_Data, T_Socket>::metadata_type m =
       parent->meta_queue.pop();
    S3FC_DBG( std::cout << "Sending metadata to client " << m.id
         << " " << m.metadata << std::endl );
         if (parent->is_connected_by_id(m.id))
         {
            parent->set_critical_by_id(m.id);
            s3_streamable::pack(m.metadata,b);
            // Shift size left and set metadata bit
            sz = b.get_size() << 1 | 1;
       T_Socket* sock = parent->get_socket_by_id(m.id);
       assert( sock != 0 );
       if ( sock->is_closed() )
            {
               parent->clear_critical_by_id(m.id);
               continue;
            }
            if ( !sock->write(&sz, sizeof(sz), -1, &terminate) )
            {
               handle_failure(m.id, sock->get_error());
               parent->clear_critical_by_id(m.id);
               if (sock->get_errno() == -2) return;
               continue;
            }
            if ( !sock->write(b.get_buffer(), sz >> 1, -1, &terminate) )
            {
               handle_failure(m.id, sock->get_error());
               parent->clear_critical_by_id(m.id);
               if (sock->get_errno() == -2) return;
               continue;
            }
            parent->clear_critical_by_id(m.id);
         }
         else
         {
             std::ostringstream msg;
             msg << " Metadata sent to non-existant client ID";
             parent->dispatch_event(
      s3_txport_event(s3_txport_event::error, m.id, msg.str()));
         }
         continue;
      }
      
      // Check the data queue for real stuff to transmit
      d = in_q.nbl_open_output();
      if (d == 0) continue; // Loop because the hit was something else
      
      s3_streamable::pack(d->data,b);
      id_list = d->get_id_list();
      in_q.close_output(d);
      
      if (!id_list.empty())
      {
         if (terminate.try_wait()) return;
         // Now send the same data to each client in the list
         for ( std::list<s3_txport_client_id>::iterator ii = id_list.begin();
               ii != id_list.end(); ii++) //foreach destination
         {
            s3_txport_client_id id = *ii;
            // Check that the connection still exists before sending data.
            // If the set_critical_by_id returns true, it implies that the
            // connection is present
            parent->set_critical_by_id(id);
            if (parent->is_connected_by_id(id))
            {
               sz = b.get_size();
               if (sz >= (1u << (static_cast<unsigned int>(sizeof(sz))*8u-1u)))
               {
        throw s3_generic_exception(
           "s3_txport_tcp_tx_task::main_loop()",
           "Maximum packet size exceeded");
               }
               // Shift size over by one bit. If the LSB is zero then the
               // data is regular payload and not metadata.
               sz <<= 1;
          T_Socket* sock = parent->get_socket_by_id(id);
          assert( sock != 0 );
          if ( sock->is_closed() )
               {
                  parent->clear_critical_by_id(id);
                  continue;
               }
               if ( !sock->write(&sz, sizeof(sz), -1, &terminate) )
               {
                  handle_failure(id, sock->get_error());
                  parent->clear_critical_by_id(id);
                  if (sock->get_errno() == -2) return;
                  continue;
               }
          if ( !sock->write(b.get_buffer(), sz >> 1, -1, &terminate) )
               {
                  handle_failure(id, sock->get_error());
                  parent->clear_critical_by_id(id);
                  if (sock->get_errno() == -2) return;
                  continue;
               }
               parent->clear_critical_by_id(id);
            }
            else // ignore data and inform event handlers
            {
               std::ostringstream msg;
               msg << " Data sent to non-existant client ID";
               parent->dispatch_event(
        s3_txport_event(s3_txport_event::error, id, msg.str()));
          parent->clear_critical_by_id(id);
            }
         }
      }
      else
      {
    std::ostringstream msg;
    msg << " No connected clients to send to";
    parent->dispatch_event(
       s3_txport_event(s3_txport_event::error, -1, msg.str()));
      }
   }
   // we will never get here.
   throw (s3_generic_exception("s3_txport_tx_task::main_loop()",
                "Instruction pointer cannot get here!"));
}

template<class T_Data, class T_Socket>
class s3_txport_tcp_rx_task : public s3_thread_base
{
   s3_semaphore sem;
   s3_pack_buffer b;
   T_Socket* sock;
   s3_txport_client_id id;
   std::string client_ip;
   std::string message;
   bool is_connected;
   s3_txport_tcp<T_Data, T_Socket> *parent;
   s3_semaphore terminate;
   s3_periodic_notifier pn;
   s3_mutex c_lock;

   void set_message(const std::string m)
   {
      message = m;
   }
   
  protected:
   void cleanup()
   {
#if 0      
      S3FC_DBG( std::cout << "s3_txport_tcp_rx_task::cleanup(): "
      << "Cleaning up thread with socket "
      << sock.get_fd() << " and ID " << int(id)
      << std::endl << std::flush );
#endif      
      
      c_lock.lock();
      is_connected = false;
      c_lock.unlock();
      
      lock_state();
      sock->close();
      delete sock;
      sock = 0;
      unlock_state();
      
      parent->dying_tasklets.push(id);
      parent->rx_queue.unsubscribe_producer(sem);
      pn.disable();
   }
   
   // Constructor
   s3_txport_tcp_rx_task(
      s3_txport_client_id a_id, T_Socket* a_sock,
      const std::string& a_client_ip,
      s3_txport_tcp<T_Data,T_Socket> *a_parent) :
      sock(a_sock), id(a_id), client_ip(a_client_ip), message(""),
      is_connected(true), parent(a_parent), pn(100)
   {
      pn.subscribe(sem);
   }

   ~s3_txport_tcp_rx_task()
   {
      pn.unsubscribe(sem);
   }

 public:
   void set_terminate()
   {
      terminate.post();
   }

   std::string get_message() const
   {
      return message;
   }

   std::string get_client_ip() const
   {
      return client_ip;
   }

   bool connected() const
   {
      bool ret;
      c_lock.lock();
      ret = is_connected;
      c_lock.unlock();
      return ret;
   }

   T_Socket* get_socket()
   {
       return sock;
   }

   void main_loop()
   {
      pn.enable();
      parent->rx_queue.subscribe_producer(sem);

      typename s3_fifo_queue<s3_txport_data<T_Data> >::s3_queue_data_t *d;
#if 0      
      S3FC_DBG( std::cout << "s3_txport_tcp_rx_task::main_loop(): "
      << "Receive thread for socket " << sock.get_fd()
      << " with client ID " << int(id) << " started"
      << std::endl << std::flush );
#endif      
      bool meta = false;
      while (1)
      {
         int size = 0;
         meta = false;
    
         if (terminate.try_wait())
    {
       S3FC_DBG( std::cout << "s3_txport_tcp_rx_task::main_loop(): "
            << "termination semaphore posted, terminating"
            << std::endl << std::flush );
       return;
    }
#if 0  
    S3FC_DBG( std::cout << "survived termination semaphore"
         << " (socket " << sock.get_fd() << ")"
         << std::endl << std::flush );
#endif    
    if ( !sock->read(&size, sizeof(size), &terminate) )
    {
            set_message(sock->get_error());
            return;
         }

    S3FC_DBG( std::cout << "read succeeded" << std::endl << std::flush );

         if (size & 1) meta = true;
         size >>= 1; // Shift right to remove meta bit.

         // Plump up the streamable for our next move
         b.set_capacity(size);
         if ( !sock->read(b.get_buffer(), size, &terminate) )
         {
            set_message(sock->get_error());
            return;
         }
         if (!meta)
         {
            // Send data to the queue, checking if
            do
            {
               d = parent->rx_queue.nbl_open_input();
               if (d == 0 ) // Blocking...
               {
                  if (terminate.try_wait()) return;
                  // Just wait for something to happen
                  sem.wait();
               }
            }
            while (d == 0);
            s3_streamable::unpack(d->data,b);
            // Tag data with our socket (client ID)
            d->set_id(id);
            parent->rx_queue.close_input(d);
         }
         else
         {
            std::string m;
            s3_streamable::unpack(m,b);
            S3FC_DBG( std::cout << "s3_txport_tcp_rx_task::main_loop(): "
            << "Metadata " << m << " received for id "
            << int(id) << std::endl << std::flush );
            parent->dispatch_event(
          s3_txport_event(s3_txport_event::metadata, id, m));
    }
         b.reset();
      }
   }

   friend class s3_txport_tcp<T_Data,T_Socket>;
};

template<class T_Data, class T_Socket>
s3_txport_tcp<T_Data, T_Socket>::s3_txport_tcp(
   int queue_slots, std::string network, std::string mask) :
   rx_queue(queue_slots), tx_queue(queue_slots), connect_id(0),
   sock(INVALID_SOCKET), tx_thread(tx_queue, this),  network_mask(0),
   network_number(0)
{
// initialise WS2_32.DLL as per MS manual
#ifdef _WIN32
   WSADATA wsaData;
   /*
    * If multiple s3_txport_tcp objects are created in a single
    * process space the WSAStartup call will be called multiple times.
    * This is legal, you just have to call WSACleanup() an equal
    * number of times.`
    * 
    * See ~s3_txport_tcp() for more comments on this.
    * 
    * [jpool]
    */
   WSAStartup(0x0101, &wsaData);
#endif
   struct in_addr ip_addr_temp;
#ifdef _WIN32
   /*
    * inet_pton does not exist in Win32 API. The older version inet_addr
    * does, so I'm using that instead.
    *
    * [jpool] 09/09/2002
    */
   ip_addr_temp.s_addr = inet_addr( network.c_str() );   
   if ( ip_addr_temp.s_addr == INADDR_NONE )
   {
      throw s3_generic_exception("s3_txport_tcp::s3_txport_tcp()",
             "Invalid network number string");
   }
#else
   int ret = inet_pton(AF_INET,network.c_str(),&ip_addr_temp);
   if (ret <= 0)
   {
      throw s3_generic_exception("s3_txport_tcp::s3_txport_tcp()",
             "Invalid network number string");
   }
#endif
   network_number = ntohl(ip_addr_temp.s_addr);
#ifdef _WIN32
   ip_addr_temp.s_addr = inet_addr( mask.c_str() );   
   if ( ip_addr_temp.s_addr == INADDR_NONE )
   {
      throw s3_generic_exception("s3_txport_tcp::s3_txport_tcp()",
             "Invalid network mask string");
   }
#else
   ret = inet_pton(AF_INET,mask.c_str(),&ip_addr_temp);
   if (ret <= 0)
   {
      throw s3_generic_exception("s3_txport_tcp::s3_txport_tcp()",
             "Invalid network mask string");
   }
#endif
   network_mask = ntohl(ip_addr_temp.s_addr);

   tx_thread.start();
}

template<class T_Data, class T_Socket>
s3_txport_tcp<T_Data, T_Socket>::~s3_txport_tcp()
{
   stop();
   // de-initialise WS2_32.DLL
#ifdef _WIN32
   /*
    * For each WSAStartup() we must have a WSACleanup().  Each time
    * WSACleanup() is called it will just decrement an internal
    * counter. Only the last call will free the resources.
    *
    * [jpool]
    */
   WSACleanup();
#endif   
}

template<class T_Data, class T_Socket>
void s3_txport_tcp<T_Data, T_Socket>::restart()
{
   if (!tx_thread.is_running())
   {
      tx_thread.start();
   }
}

template<class T_Data, class T_Socket>
bool s3_txport_tcp<T_Data, T_Socket>::disconnect_by_id(s3_txport_client_id id)
{
   S3FC_DBG( std::cout << "s3_txport_tcp::disconnect_by_id(id="
        << int(id) << ") called" << std::endl << std::flush );
   
   bool retval = false;
   
   // This will cause the thread to die, and trigger its removal from
   // the list of active clients.
   connection_mutex.lock();
   if (connections.count(id) != 0)
   {
      if (connections[id]->is_running())
      {
         connections[id]->set_terminate();
         retval = true;

    S3FC_DBG( std::cout << "s3_txport_tcp::disconnect_by_id(): "
         "set_terminate() on connection done" << std::endl
         << std::flush );
      }
   }
   connection_mutex.unlock();
   
   return retval;
}

template<class T_Data, class T_Socket>
void s3_txport_tcp<T_Data, T_Socket>::disconnect_all()
{
   while (num_connections() != 0)
   {
      std::list<s3_txport_client_id> id_list = get_client_id_list();
      
      for (std::list<s3_txport_client_id>::iterator ii = id_list.begin();
      ii != id_list.end(); ii++)
      {
         disconnect_by_id(*ii);
      }
      S3FC_DBG(std::cout << "s3_txport_tcp::disconnect_all() reaping"
          << std::endl);
      reap();
      if (num_connections() != 0)
      {
         S3FC_DBG(std::cout
        << "s3_txport_tcp::disconnect_all() num_connections()="
        << num_connections() << std::endl);
         S3FC_DBG(std::cout
        << "s3_txport_tcp::disconnect_all() dying task count="
        << dying_tasklets.size() << std::endl);
#ifdef _WIN32
    // Sleep in ms. Must include <winbase.h>. [rfanner]
    Sleep(1000);
#else
    // POSIX sleep in. Must include <netinet/in.h>. [rfanner]
    sleep(1);
#endif
      }
   }
}

template<class T_Data, class T_Socket>
void s3_txport_tcp<T_Data, T_Socket>::stop()
{
   S3FC_DBG( std::cout << "s3_txport_tcp::stop() on TID " << get_thread_id()
        << std::endl << std::flush );
   
   // Terminate the server to client tx thread.
   if (tx_thread.is_running())
   {
      tx_thread.set_terminate();
      tx_thread.wait_on_exit();
   }

   // Kill the listener.
   if (is_running())
   {
      set_terminate();
      wait_on_exit();
   }
   disconnect_all();
   
   S3FC_DBG( std::cout << "s3_txport_tcp::stop(): num_connections()="
        << num_connections() << std::endl );
   S3FC_DBG( std::cout << "s3_txport_tcp::stop(): dying task count="
        << dying_tasklets.size() << std::endl );
   S3FC_DBG( std::cout << "s3_txport_tcp::stop(): thread " << get_thread_id()
        << " is done" << std::endl << std::flush );
}

template<class T_Data, class T_Socket>
void 
s3_txport_tcp<T_Data, T_Socket>::set_critical_by_id(s3_txport_client_id id)
{
   connection_mutex.lock();
   // If the connection is registered and it is alive, make it critical
   if (connections.count(id) != 0 )
   {
      connections[id]->lock_state();
   }
   connection_mutex.unlock();
}

template<class T_Data, class T_Socket>
void 
s3_txport_tcp<T_Data, T_Socket>::clear_critical_by_id(s3_txport_client_id id)
{
   connection_mutex.lock();
   if (connections.count(id) != 0)
   {
      connections[id]->unlock_state();
   }
   connection_mutex.unlock();
}

template<class T_Data, class T_Socket>
T_Socket* s3_txport_tcp<T_Data, T_Socket>::get_socket_by_id(
   s3_txport_client_id id)
{
   T_Socket* ret = 0;
   connection_mutex.lock();
   if (connections.count(id) != 0)
   {
      ret = connections[id]->get_socket();
   }
   connection_mutex.unlock();
   
   return ret;
}

template<class T_Data, class T_Socket>
bool 
s3_txport_tcp<T_Data, T_Socket>::is_connected_by_id(s3_txport_client_id id)
{
   bool ret = false;
   
   connection_mutex.lock();
   if (connections.count(id) != 0)
   {
      ret = connections[id]->connected();
   }
   connection_mutex.unlock();
   
   return ret;
}

template<class T_Data, class T_Socket>
std::string 
s3_txport_tcp<T_Data, T_Socket>::get_ip_address_by_id(s3_txport_client_id id)
{
   std::string ret;
   
   connection_mutex.lock();
   if (connections.count(id) != 0)
   {
      ret = connections[id]->get_client_ip();
   }
   connection_mutex.unlock();
   
   return ret;
}

template<class T_Data, class T_Socket>
std::list<s3_txport_client_id>
s3_txport_tcp<T_Data, T_Socket>::get_client_id_list() const
{
   typename thread_map::const_iterator ii;
   std::list<s3_txport_client_id> current_list;
   
   connection_mutex.lock();
   for (ii = connections.begin(); ii != connections.end(); ii++)
   {
      current_list.push_back(ii->first);
   }
   connection_mutex.unlock();
   
   return current_list;
}

template<class T_Data, class T_Socket>
void 
s3_txport_tcp<T_Data, T_Socket>::send_metadata(s3_txport_client_id id, const std::string& mdata)
{
   meta_queue.push(metadata_type(id, mdata));
}

template<class T_Data, class T_Socket>
bool s3_txport_tcp<T_Data, T_Socket>::listen(const unsigned int a_port)
{
   port = a_port;
   
   if (!is_running())
   {
      if ( !sock.is_valid() )
      {
#if 0  
    S3FC_DBG2_("s3_txport_tcp::listen()", "socket=" << sock.get_fd());
#endif    

    throw s3_generic_exception(
       "s3_txport_tcp::listen()",
       "Failed to open a socket: " + sock.get_error());
      }
      if ( !sock.bind(INADDR_ANY, port) )
      {
         std::ostringstream msg;
         msg << "s3_txport_tcp::listen():"
             << " Failed to connect to port " << port
             << " because " << sock.get_error();
         s3_event_dispatcher<s3_txport_event>::dispatch_event(
       s3_txport_event(s3_txport_event::error, -1, msg.str()));
         return false;
      }
      if ( !sock.listen(SOMAXCONN) )
      {
         std::ostringstream msg;
         msg << " Failed to listen on socket"  << port
             << " because " << sock.get_error();
         throw s3_generic_exception("s3_txport_tcp::listen()", msg.str());
      }

      start();
   }
   if (!is_running()) // If we get here, the thread must be running.
   {
      throw s3_generic_exception(
    "s3_txport_tcp::listen()", "Listener thread failed to start.");
   }
   
   return true;
}

template<class T_Data, class T_Socket>
void s3_txport_tcp<T_Data, T_Socket>::cleanup()
{
   // Close the socket we have bound to.
   sock.close();
}

template<class T_Data, class T_Socket>
void s3_txport_tcp<T_Data, T_Socket>::reap()
{
   reap_mutex.lock();
   while (!dying_tasklets.empty())
   {
      S3FC_DBG(std::cout << "s3_txport_tcp::reap(): Reaping dead tasks"
          << std::endl << std::flush );
      s3_txport_client_id tid = dying_tasklets.pop();
      
      connection_mutex.lock();
      S3FC_DBG( std::cout << "Reaper has lock" << std::endl << std::flush );
      if (connections.count(tid) == 0) // See if the connection is still here
      {
    connection_mutex.unlock();
    S3FC_DBG( std::cout << "Reaper released lock (client vanished)"
         << std::endl << std::flush );
    continue; // It has gone so just ignore it.
      }
      if (!connections[tid]->test_state()) // will return false if critical
      {
         // Must be critical so just push the id back into the queue
         dying_tasklets.push(tid);
         connection_mutex.unlock();
         S3FC_DBG( std::cout << "Reaper released lock (rx was critical)"
         << std::endl << std::flush );
         sched_yield();
         continue;
      }
      // If we get here, we have a lock we don't need, so ditch it.
      connections[tid]->unlock_state();
      S3FC_DBG( std::cout << "Waiting on client thread " << int(tid)
           << std::endl << std::flush );
      // Some tasks may not have been terminated, so we do it if required.
      //if (connections[tid]->is_running()) connections[tid]->terminate();
      connections[tid]->wait_on_exit();
      S3FC_DBG( std::cout << "Client " << int(tid) << " terminated"
      << std::endl << std::flush );
      // We need to store the reason for this death for use when we
      // generate an event, as it would be unwise to dispatch the
      // event holding the mutex...
      std::string reason = connections[tid]->get_client_ip() + " " +
            connections[tid]->get_message();
      delete connections[tid];
      // Wipe the thread out of the map
      connections.erase(tid);
      connection_mutex.unlock();
      S3FC_DBG(std::cout << "Reaper released lock" << std::endl);
      s3_event_dispatcher<s3_txport_event>::dispatch_event(
    s3_txport_event(s3_txport_event::disconnect, tid, reason));
   }
   reap_mutex.unlock();
}

template<class T_Data, class T_Socket>
s3_txport_client_id 
s3_txport_tcp<T_Data, T_Socket>::connect(const std::string& IP_address, const unsigned int port)
{
   // Create a TCP socket
   std::auto_ptr<T_Socket> lsock(new T_Socket(sock));

   if ( !lsock->is_valid() ) // This is also fatal, so throw
   {
      throw s3_generic_exception(
    "s3_txport_tcp::connect()", "Failed to open a socket");
   }

   S3FC_DBG2_("s3_txport_tcp::connect()", "connecting with lsock");
   
   if ( !lsock->connect(IP_address, port) )
   {
      std::ostringstream msg;
      msg << "s3_txport_tcp::connect():"
     << " Failed to connect to " << IP_address << ":" << port
     << " because " << lsock->get_error();
      s3_event_dispatcher<s3_txport_event>::dispatch_event(
    s3_txport_event(s3_txport_event::error, -1, msg.str()));
      lsock->close();
      return static_cast<s3_txport_client_id>(0);
   }
   S3FC_DBG( std::cout << "s3_txport_client::s3_txport_client(): "
        << "connected, spawning handler" << std::endl << std::flush );
   // If we get here, we are connected. All comms takes place via
   // the socket.  Post two semaphores and wait until both have
   // been waited on.
   return spawn_handler(lsock.release(), IP_address);
}

// The thread that will handle the data from the connections.
template<class T_Data, class T_Socket>
void s3_txport_tcp<T_Data, T_Socket>::main_loop()
{
   S3FC_DBG(std::cout
       << "Daemon configured. Awaiting client connections..." 
       << std::endl << std::flush);
   while (1)
   {
      if (terminate.try_wait()) return;
      
      //test_cancel();
      if ( sock.select_rd(1) )
      {
    S3FC_DBG2_("s3_txport_tcp::main_loop()",
          "incoming connection. making server-side socket.");

    struct sockaddr_in clientname;
    std::auto_ptr<T_Socket> active_sock(new T_Socket());
    if ( sock.accept(*active_sock, &clientname) )
         {
       S3FC_DBG2_("s3_txport_tcp::main_loop()", "active_sock ok");

       int ip_no = ntohl(clientname.sin_addr.s_addr);
       // Record the name of the connecting producer for the logger.
       char *tmp = (char *)&ip_no;
       std::ostringstream hostname;
       hostname << static_cast<unsigned int>(tmp[3]) << "."
           << static_cast<unsigned int>(tmp[2]) << "."
           << static_cast<unsigned int>(tmp[1]) << "."
           << static_cast<unsigned int>(tmp[0]);
       
       if ( (ip_no & network_mask) == network_number )
       {
          S3FC_DBG( std::cout << "s3_txport_tcp::main_loop(): "
          << "Connection established, spawning handler"
          << std::endl << std::flush );
          spawn_handler(active_sock.release(), hostname.str());
       }
       else
       {
          active_sock->close();
          S3FC_DBG( std::cout << "s3_txport_tcp::main_loop(): "
          << "Connection from " << hostname << " rejected"
          << std::endl << std::flush );
          
s3_event_dispatcher<s3_txport_event>::		  dispatch_event(s3_txport_event(s3_txport_event::reject,
                   -1, hostname.str()));
       }
         }
    else
    {
       
s3_event_dispatcher<s3_txport_event>::	       dispatch_event(s3_txport_event(s3_txport_event::error,
                     -1, sock.get_error()));
    }
      }
   }
}

template<class T_Data, class T_Socket>
s3_txport_client_id 
s3_txport_tcp<T_Data, T_Socket>::spawn_handler(T_Socket* active_sock, std::string hostname)
{
   connection_mutex.lock();
   s3_txport_client_id id = next_id();
   connections[id] = new s3_txport_tcp_rx_task<T_Data, T_Socket>(
      connect_id, active_sock, hostname, this);
   // now start the thread to receive the client's data.
   connections[id]->start();
   connection_mutex.unlock();

   s3_event_dispatcher<s3_txport_event>::dispatch_event(
      s3_txport_event(s3_txport_event::connect, id, hostname));
#if 0   
   S3FC_DBG( std::cout << "Socket " << active_sock.get_fd()
        << " created for ID " << int(id) << std::endl << std::flush );
#endif   
   return id;
}

template<class T_Data, class T_Socket>
int s3_txport_tcp<T_Data, T_Socket>::num_connections() const
{
   int retval = 0;

   connection_mutex.lock();
   retval = connections.size();
   connection_mutex.unlock();
   
   return retval;
}

---

Send comments to: s3fc@stonethree.com