src/router.cc

   1 ///
   2 /// \file       router.cc
   3 ///             Support classes for the pluggable socket routing system.
   4 ///
   5
   6 /*
   7     Copyright (C) 2008-2009, Net Direct Inc. (http://www.netdirect.ca/)
   8
   9     This program is free software; you can redistribute it and/or modify
  10     it under the terms of the GNU General Public License as published by
  11     the Free Software Foundation; either version 2 of the License, or
  12     (at your option) any later version.
  13
  14     This program is distributed in the hope that it will be useful,
  15     but WITHOUT ANY WARRANTY; without even the implied warranty of
  16     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
  17
  18     See the GNU General Public License in the COPYING file at the
  19     root directory of this project for more details.
  20 */
  21
  22 #include "router.h"
  23 #include "scoped_lock.h"
  24 #include "data.h"
  25 #include "protostructs.h"
  26 #include "usbwrap.h"
  27 #include "endian.h"
  28 #include "debug.h"
  29 #include <unistd.h>
  30
  31 namespace Barry {
  32
  33 ///////////////////////////////////////////////////////////////////////////////
  34 // SocketRoutingQueue constructors
  35
  36 SocketRoutingQueue::SocketRoutingQueue(int prealloc_buffer_count)
  37         : m_dev(0)
  38         , m_writeEp(0)
  39         , m_readEp(0)
  40         , m_interest(false)
  41         , m_continue_reading(false)
  42 {
  43         pthread_mutex_init(&m_mutex, NULL);
  44
  45         pthread_mutex_init(&m_readwaitMutex, NULL);
  46         pthread_cond_init(&m_readwaitCond, NULL);
  47
  48         AllocateBuffers(prealloc_buffer_count);
  49 }
  50
  51 SocketRoutingQueue::~SocketRoutingQueue()
  52 {
  53         // thread running?
  54         if( m_continue_reading ) {
  55                 m_continue_reading = false;
  56                 pthread_join(m_usb_read_thread, NULL);
  57         }
  58 }
  59
  60 ///////////////////////////////////////////////////////////////////////////////
  61 // protected members
  62
  63 //
  64 // ReturnBuffer
  65 //
  66 /// Provides a method of returning a buffer to the free queue
  67 /// after processing.  The DataHandle class calls this automatically
  68 /// from its destructor.
  69 void SocketRoutingQueue::ReturnBuffer(Data *buf)
  70 {
  71         // don't need to lock here, since m_free handles its own locking
  72         m_free.push(buf);
  73 }
  74
  75 //
  76 // SimpleReadThread()
  77 //
  78 /// Convenience thread to handle USB read activity.
  79 ///
  80 void *SocketRoutingQueue::SimpleReadThread(void *userptr)
  81 {
  82         SocketRoutingQueue *q = (SocketRoutingQueue *)userptr;
  83
  84         // read from USB and write to stdout until finished
  85         std::string msg;
  86         while( q->m_continue_reading ) {
  87                 if( !q->DoRead(msg, 1000) ) {   // timeout in milliseconds
  88                         eout("Error in SimpleReadThread: " << msg);
  89                 }
  90         }
  91         return 0;
  92 }
  93
  94
  95 ///////////////////////////////////////////////////////////////////////////////
  96 // public API
  97
  98 // These functions connect the router to an external Usb::Device
  99 // object.  Normally this is handled automatically by the
 100 // Controller class, but are public here in case they are needed.
 101 void SocketRoutingQueue::SetUsbDevice(Usb::Device *dev, int writeEp, int readEp)
 102 {
 103         scoped_lock lock(m_mutex);
 104         m_dev = dev;
 105         m_writeEp = writeEp;
 106         m_readEp = readEp;
 107 }
 108
 109 void SocketRoutingQueue::ClearUsbDevice()
 110 {
 111         scoped_lock lock(m_mutex);
 112         m_dev = 0;
 113         lock.unlock();
 114
 115         // wait for the DoRead cycle to finish, so the external
 116         // Usb::Device object doesn't close before we're done with it
 117         scoped_lock wait(m_readwaitMutex);
 118         pthread_cond_wait(&m_readwaitCond, &m_readwaitMutex);
 119 }
 120
 121 bool SocketRoutingQueue::UsbDeviceReady()
 122 {
 123         scoped_lock lock(m_mutex);
 124         return m_dev != 0;
 125 }
 126
 127 //
 128 // AllocateBuffers
 129 //
 130 /// This class starts out with no buffers, and will grow one buffer
 131 /// at a time if needed.  Call this to allocate count buffers
 132 /// all at once and place them on the free queue.  After calling
 133 /// this function, at least count buffers will exist in the free
 134 /// queue.  If there are already count buffers, none will be added.
 135 ///
 136 void SocketRoutingQueue::AllocateBuffers(int count)
 137 {
 138         int todo = count - m_free.size();
 139
 140         for( int i = 0; i < todo; i++ ) {
 141                 // m_free handles its own locking
 142                 m_free.push( new Data );
 143         }
 144 }
 145
 146 //
 147 // DefaultRead (both variations)
 148 //
 149 /// Returns the data for the next unregistered socket.
 150 /// Blocks until timeout or data is available.
 151 /// Returns false (or null pointer) on timeout and no data.
 152 /// With the return version of the function, there is no
 153 /// copying performed.
 154 ///
 155 /// This version performs a copy.
 156 ///
 157 bool SocketRoutingQueue::DefaultRead(Data &receive, int timeout)
 158 {
 159         DataHandle buf = DefaultRead(timeout);
 160         if( !buf.get() )
 161                 return false;
 162
 163         // copy to desired buffer
 164         receive = *buf.get();
 165         return true;
 166 }
 167
 168 ///
 169 /// This version does not perform a copy.
 170 ///
 171 DataHandle SocketRoutingQueue::DefaultRead(int timeout)
 172 {
 173         // m_default handles its own locking
 174         Data *buf = m_default.wait_pop(timeout);
 175         return DataHandle(*this, buf);
 176 }
 177
 178 //
 179 // RegisterInterest
 180 //
 181 /// Register an interest in data from a certain socket.  To read
 182 /// from that socket, use the SocketRead() function from then on.
 183 ///
 184 /// Any non-registered socket goes in the default queue
 185 /// and must be read by DefaultRead()
 186 ///
 187 /// If not null, handler is called when new data is read.  It will
 188 /// be called in the same thread instance that DoRead() is called from.
 189 /// Handler is passed the DataQueue Data pointer, and so no
 190 /// copying is done.  Once the handler returns, the data is
 191 /// considered processed and not added to the interested queue,
 192 /// but instead returned to m_free.
 193 ///
 194 /// Throws std::logic_error if already registered.
 195 ///
 196 void SocketRoutingQueue::RegisterInterest(SocketId socket,
 197                                           SocketDataHandler handler,
 198                                           void *context)
 199 {
 200         // modifying our own std::map, need a lock
 201         scoped_lock lock(m_mutex);
 202
 203         SocketQueueMap::iterator qi = m_socketQueues.find(socket);
 204         if( qi != m_socketQueues.end() )
 205                 throw std::logic_error("RegisterInterest requesting a previously registered socket.");
 206
 207         m_socketQueues[socket] = QueueEntryPtr( new QueueEntry(handler, context) );
 208         m_interest = true;
 209 }
 210
 211 //
 212 // UnregisterInterest
 213 //
 214 /// Unregisters interest in data from the given socket, and discards
 215 /// any existing data in its interest queue.  Any new incoming data
 216 /// for this socket will be placed in the default queue.
 217 ///
 218 void SocketRoutingQueue::UnregisterInterest(SocketId socket)
 219 {
 220         // modifying our own std::map, need a lock
 221         scoped_lock lock(m_mutex);
 222
 223         SocketQueueMap::iterator qi = m_socketQueues.find(socket);
 224         if( qi == m_socketQueues.end() )
 225                 return; // nothing registered, done
 226
 227         // salvage all our data buffers
 228         m_free.append_from( qi->second->m_queue );
 229
 230         // remove the QueueEntryPtr from the map
 231         m_socketQueues.erase( qi );
 232
 233         // check the interest flag
 234         m_interest = m_socketQueues.size() > 0;
 235 }
 236
 237 //
 238 // SocketRead
 239 //
 240 /// Reads data from the interested socket cache.  Can only read
 241 /// from sockets that have been previously registered.
 242 ///
 243 /// Blocks until timeout or data is available.
 244 ///
 245 /// Returns false (or null pointer) on timeout and no data.
 246 /// With the return version of the function, there is no
 247 /// copying performed.
 248 ///
 249 /// Throws std::logic_error if a socket was requested that was
 250 /// not previously registered.
 251 ///
 252 /// Copying is performed with this function.
 253 ///
 254 bool SocketRoutingQueue::SocketRead(SocketId socket, Data &receive, int timeout)
 255 {
 256         DataHandle buf = SocketRead(socket, timeout);
 257         if( !buf.get() )
 258                 return false;
 259
 260         // copy to desired buffer
 261         receive = *buf.get();
 262         return true;
 263 }
 264
 265 ///
 266 /// Copying is not performed with this function.
 267 ///
 268 /// Throws std::logic_error if a socket was requested that was
 269 /// not previously registered.
 270 ///
 271 DataHandle SocketRoutingQueue::SocketRead(SocketId socket, int timeout)
 272 {
 273         QueueEntryPtr qep;
 274         DataQueue *dq = 0;
 275
 276         // accessing our own std::map, need a lock
 277         {
 278                 scoped_lock lock(m_mutex);
 279                 SocketQueueMap::iterator qi = m_socketQueues.find(socket);
 280                 if( qi == m_socketQueues.end() )
 281                         throw std::logic_error("SocketRead requested data from unregistered socket.");
 282
 283                 // got our queue, save the whole QueueEntryPtr (shared_ptr),
 284                 // and unlock, since we will be waiting on the DataQueue,
 285                 // not the socketQueues map
 286                 //
 287                 // This is safe, since even if UnregisterInterest is called,
 288                 // our pointer won't be deleted until our shared_ptr
 289                 // (QueueEntryPtr) goes out of scope.
 290                 //
 291                 // The remaining problem is that wait_pop() might wait
 292                 // forever if there is no timeout... c'est la vie.
 293                 // Should'a used a timeout. :-)
 294                 qep = qi->second;
 295                 dq = &qep->m_queue;
 296         }
 297
 298         // get data from DataQueue
 299         Data *buf = dq->wait_pop(timeout);
 300
 301         // specifically delete our copy of shared pointer, in a locked
 302         // environment
 303         {
 304                 scoped_lock lock(m_mutex);
 305                 qep.reset();
 306         }
 307
 308         return DataHandle(*this, buf);
 309 }
 310
 311 // Returns true if data is available for that socket.
 312 bool SocketRoutingQueue::IsAvailable(SocketId socket) const
 313 {
 314         scoped_lock lock(m_mutex);
 315         SocketQueueMap::const_iterator qi = m_socketQueues.find(socket);
 316         if( qi == m_socketQueues.end() )
 317                 return false;
 318         return qi->second->m_queue.size() > 0;
 319 }
 320
 321 //
 322 // DoRead
 323 //
 324 /// Called by the application's "read thread" to read the next usb
 325 /// packet and route it to the correct queue.  Returns after every
 326 /// read, even if a handler is associated with a queue.
 327 /// Note: this function is safe to call before SetUsbDevice() is
 328 /// called... it just doesn't do anything if there is no usb
 329 /// device to work with.
 330 ///
 331 /// Timeout is in milliseconds.
 332 ///
 333 /// Returns false in the case of USB errors and puts the error message
 334 /// in msg.
 335 ///
 336 bool SocketRoutingQueue::DoRead(std::string &msg, int timeout)
 337 {
 338         class ReadWaitSignal
 339         {
 340                 pthread_mutex_t &m_Mutex;
 341                 pthread_cond_t &m_Cond;
 342         public:
 343                 ReadWaitSignal(pthread_mutex_t &mut, pthread_cond_t &cond)
 344                         : m_Mutex(mut), m_Cond(cond)
 345                         {}
 346                 ~ReadWaitSignal()
 347                 {
 348                         scoped_lock wait(m_Mutex);
 349                         pthread_cond_signal(&m_Cond);
 350                 }
 351         } readwait(m_readwaitMutex, m_readwaitCond);
 352
 353         Usb::Device * volatile dev = 0;
 354         int readEp;
 355         DataHandle buf(*this, 0);
 356
 357         // if we are not connected to a USB device yet, just wait
 358         {
 359                 scoped_lock lock(m_mutex);
 360
 361                 if( !m_dev ) {
 362                         lock.unlock();  // unlock early, since we're sleeping
 363                         // sleep only a short time, since things could be
 364                         // in the process of setup or teardown
 365                         usleep(125000);
 366                         return true;
 367                 }
 368
 369                 dev = m_dev;
 370                 readEp = m_readEp;
 371
 372                 // fetch a free buffer
 373                 Data *raw = m_free.pop();
 374                 if( !raw )
 375                         buf = DataHandle(*this, new Data);
 376                 else
 377                         buf = DataHandle(*this, raw);
 378         }
 379
 380         // take a chance and do the read unlocked, as this has the potential
 381         // for blocking for a while
 382         try {
 383
 384                 Data &data = *buf.get();
 385
 386                 if( !dev->BulkRead(readEp, data, timeout) )
 387                         return true;    // no data, done!
 388
 389                 MAKE_PACKET(pack, data);
 390
 391                 // make sure the size is right
 392                 if( data.GetSize() < sizeof(pack->socket) )
 393                         return true;    // bad size, just skip
 394
 395                 // extract the socket from the packet
 396                 uint16_t socket = btohs(pack->socket);
 397
 398                 // we have data, now lock up again to place it
 399                 // in the right queue
 400                 scoped_lock lock(m_mutex);
 401
 402                 // search for registration of socket
 403                 if( m_interest ) {
 404                         SocketQueueMap::iterator qi = m_socketQueues.find(socket);
 405                         if( qi != m_socketQueues.end() ) {
 406                                 SocketDataHandler sdh = qi->second->m_handler;
 407                                 void *ctx = qi->second->m_context;
 408
 409                                 // is there a handler?
 410                                 if( sdh ) {
 411                                         // unlock & let the handler process it
 412                                         lock.unlock();
 413                                         (*sdh)(ctx, buf.get());
 414                                         return true;
 415                                 }
 416                                 else {
 417                                         qi->second->m_queue.push(buf.release());
 418                                         return true;
 419                                 }
 420                         }
 421
 422                         // fall through
 423                 }
 424
 425                 // safe to unlock now, we are done with the map
 426                 lock.unlock();
 427
 428                 // if we get here, send to default queue
 429                 m_default.push(buf.release());
 430                 return true;
 431
 432         }
 433         catch( Usb::Timeout & ) {
 434                 // this is expected... just ignore
 435         }
 436         catch( Usb::Error &ue ) {
 437                 // this is unexpected, but we're in a thread here...
 438                 // return false and the caller decide how to handle it
 439                 msg = ue.what();
 440                 return false;
 441         }
 442
 443         return true;
 444 }
 445
 446 void SocketRoutingQueue::SpinoffSimpleReadThread()
 447 {
 448         // signal that it's ok to run inside the thread
 449         if( m_continue_reading )
 450                 return; // already running
 451         m_continue_reading = true;
 452
 453         // Start USB read thread, to handle all routing
 454         int ret = pthread_create(&m_usb_read_thread, NULL, &SimpleReadThread, this);
 455         if( ret ) {
 456                 m_continue_reading = false;
 457                 throw Barry::ErrnoError("SocketRoutingQueue: Error creating USB read thread.", ret);
 458         }
 459 }
 460
 461 } // namespace Barry
 462