src/router.cc

   1 ///
   2 /// \file       router.cc
   3 ///             Support classes for the pluggable socket routing system.
   4 ///
   5
   6 /*
   7     Copyright (C) 2008-2013, 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 "i18n.h"
  23 #include "router.h"
  24 #include "scoped_lock.h"
  25 #include "data.h"
  26 #include "protostructs.h"
  27 #include "protocol.h"
  28 #include "usbwrap.h"
  29 #include "endian.h"
  30 #include "debug.h"
  31 #include <unistd.h>
  32 #include <iostream>
  33 #include <iomanip>
  34
  35 using namespace std;
  36
  37 namespace Barry {
  38
  39 ///////////////////////////////////////////////////////////////////////////////
  40 // SocketDataHandler default methods
  41
  42 void SocketRoutingQueue::SocketDataHandler::Error(Barry::Error &error)
  43 {
  44         // Just log the error
  45         eout("SocketDataHandler: Error: " << error.what());
  46         (void) error;
  47 }
  48
  49 SocketRoutingQueue::SocketDataHandler::~SocketDataHandler()
  50 {
  51         // Nothing to destroy
  52 }
  53
  54 ///////////////////////////////////////////////////////////////////////////////
  55 // SocketRoutingQueue constructors
  56
  57 SocketRoutingQueue::SocketRoutingQueue(int prealloc_buffer_count,
  58                                         int default_read_timeout)
  59         : m_dev(0)
  60         , m_writeEp(0)
  61         , m_readEp(0)
  62         , m_interest(false)
  63         , m_seen_usb_error(false)
  64         , m_timeout(default_read_timeout)
  65         , m_continue_reading(false)
  66 {
  67         pthread_mutex_init(&m_mutex, NULL);
  68
  69         pthread_mutex_init(&m_readwaitMutex, NULL);
  70         pthread_cond_init(&m_readwaitCond, NULL);
  71
  72         AllocateBuffers(prealloc_buffer_count);
  73 }
  74
  75 SocketRoutingQueue::~SocketRoutingQueue()
  76 {
  77         // thread running?
  78         if( m_continue_reading ) {
  79                 m_continue_reading = false;
  80                 pthread_join(m_usb_read_thread, NULL);
  81         }
  82
  83         // dump all unused packets to debug output
  84         SocketQueueMap::const_iterator b = m_socketQueues.begin();
  85         for( ; b != m_socketQueues.end(); ++b ) {
  86                 DumpSocketQueue(b->first, b->second->m_queue);
  87         }
  88         if( m_default.size() ) {
  89                 ddout("(Default queue is socket 0)");
  90                 DumpSocketQueue(0, m_default);
  91         }
  92 }
  93
  94 ///////////////////////////////////////////////////////////////////////////////
  95 // protected members
  96
  97 //
  98 // ReturnBuffer
  99 //
 100 /// Provides a method of returning a buffer to the free queue
 101 /// after processing.  The DataHandle class calls this automatically
 102 /// from its destructor.
 103 void SocketRoutingQueue::ReturnBuffer(Data *buf)
 104 {
 105         // don't need to lock here, since m_free handles its own locking
 106         m_free.push(buf);
 107 }
 108
 109 //
 110 // QueuePacket
 111 //
 112 /// Helper function to add a buffer to a socket queue.
 113 /// Returns false if no queue is available for that socket.
 114 //// Also empties the DataHandle on success.
 115 ///
 116 bool SocketRoutingQueue::QueuePacket(SocketId socket, DataHandle &buf)
 117 {
 118         if( m_interest ) {
 119                 bool seq = Protocol::IsSequencePacket(*buf.get());
 120
 121                 // lock so we can access the m_socketQueues map safely
 122                 scoped_lock lock(m_mutex);
 123
 124                 // search for registration of socket
 125                 SocketQueueMap::iterator qi = m_socketQueues.find(socket);
 126                 if( qi != m_socketQueues.end() ) {
 127                         if( seq ) {
 128                                 if( (qi->second->m_type & SequencePackets) == 0 )
 129                                         return false;
 130                         }
 131                         else {
 132                                 if( (qi->second->m_type & DataPackets) == 0 )
 133                                         return false;
 134                         }
 135                         qi->second->m_queue.push(buf.release());
 136                         return true;
 137                 }
 138         }
 139
 140         return false;
 141 }
 142
 143 bool SocketRoutingQueue::QueuePacket(DataQueue &queue, DataHandle &buf)
 144 {
 145         // don't need to lock here, since queue handles its own locking
 146         queue.push(buf.release());
 147         return true;
 148 }
 149
 150 //
 151 // RouteOrQueuePacket
 152 //
 153 /// Same as QueuePacket, except sends the data to the callback if
 154 /// a callback is available.
 155 ///
 156 /// This function duplicates code from QueuePacket(), in order to
 157 /// optimize the mutex locking.
 158 ///
 159 bool SocketRoutingQueue::RouteOrQueuePacket(SocketId socket, DataHandle &buf)
 160 {
 161         // search for registration of socket
 162         if( m_interest ) {
 163                 // lock so we can access the m_socketQueues map safely
 164                 scoped_lock lock(m_mutex);
 165
 166                 SocketQueueMap::iterator qi = m_socketQueues.find(socket);
 167                 if( qi != m_socketQueues.end() ) {
 168                         SocketDataHandlerPtr &sdh = qi->second->m_handler;
 169
 170                         // is there a handler?
 171                         if( sdh ) {
 172                                 // unlock & let the handler process it
 173                                 lock.unlock();
 174                                 sdh->DataReceived(*buf.get());
 175
 176                                 // no exceptions thrown, clear the
 177                                 // DataHandle, sending packet back to its
 178                                 // free list
 179                                 buf.reset();
 180                                 return true;
 181                         }
 182                         else {
 183                                 qi->second->m_queue.push(buf.release());
 184                                 return true;
 185                         }
 186                 }
 187         }
 188
 189         return false;
 190 }
 191
 192 //
 193 // SimpleReadThread()
 194 //
 195 /// Convenience thread to handle USB read activity.
 196 ///
 197 void *SocketRoutingQueue::SimpleReadThread(void *userptr)
 198 {
 199         SocketRoutingQueue *q = (SocketRoutingQueue *)userptr;
 200
 201         // read from USB and write to stdout until finished
 202         q->m_seen_usb_error = false;
 203         while( q->m_continue_reading ) {
 204                 try {
 205                         q->DoRead(1000);        // timeout in milliseconds
 206                 }
 207                 catch (std::runtime_error const &e) {
 208                         eout(_("SimpleReadThread received uncaught exception: ") <<  typeid(e).name() << _(" what: ") << e.what());
 209                 }
 210                 catch (...) {
 211                         eout(_("SimpleReadThread recevied uncaught exception of unknown type"));
 212                 }
 213         }
 214         return 0;
 215 }
 216
 217 void SocketRoutingQueue::DumpSocketQueue(SocketId socket, const DataQueue &dq)
 218 {
 219         // dump a record of any unused packets in the queue, for debugging
 220         if( dq.size() ) {
 221                 ddout(_("SocketRoutingQueue Leftovers: ")
 222                         << dec << dq.size()
 223                         << _(" packet(s) for socket: ") << "0x"
 224                         << hex << (unsigned int) socket
 225                         << "\n"
 226                         << dq);
 227         }
 228 }
 229
 230
 231 ///////////////////////////////////////////////////////////////////////////////
 232 // public API
 233
 234 // These functions connect the router to an external Usb::Device
 235 // object.  Normally this is handled automatically by the
 236 // Controller class, but are public here in case they are needed.
 237 void SocketRoutingQueue::SetUsbDevice(Usb::Device *dev, int writeEp, int readEp,
 238                                         SocketDataHandlerPtr callback)
 239 {
 240         scoped_lock lock(m_mutex);
 241         m_dev = dev;
 242         m_usb_error_dev_callback = callback;
 243         m_writeEp = writeEp;
 244         m_readEp = readEp;
 245 }
 246
 247 void SocketRoutingQueue::ClearUsbDevice()
 248 {
 249         scoped_lock lock(m_mutex);
 250         m_dev = 0;
 251         m_usb_error_dev_callback.reset();
 252         lock.unlock();
 253
 254         // wait for the DoRead cycle to finish, so the external
 255         // Usb::Device object doesn't close before we're done with it
 256         scoped_lock wait(m_readwaitMutex);
 257         pthread_cond_wait(&m_readwaitCond, &m_readwaitMutex);
 258 }
 259
 260 bool SocketRoutingQueue::UsbDeviceReady()
 261 {
 262         scoped_lock lock(m_mutex);
 263         return m_dev != 0 && !m_seen_usb_error;
 264 }
 265
 266 //
 267 // AllocateBuffers
 268 //
 269 /// This class starts out with no buffers, and will grow one buffer
 270 /// at a time if needed.  Call this to allocate count buffers
 271 /// all at once and place them on the free queue.  After calling
 272 /// this function, at least count buffers will exist in the free
 273 /// queue.  If there are already count buffers, none will be added.
 274 ///
 275 void SocketRoutingQueue::AllocateBuffers(int count)
 276 {
 277         int todo = count - m_free.size();
 278
 279         for( int i = 0; i < todo; i++ ) {
 280                 // m_free handles its own locking
 281                 m_free.push( new Data );
 282         }
 283 }
 284
 285 //
 286 // DefaultRead (both variations)
 287 //
 288 /// Returns the data for the next unregistered socket.
 289 /// Blocks until timeout or data is available.
 290 /// Returns false (or null pointer) on timeout and no data.
 291 /// With the return version of the function, there is no
 292 /// copying performed.
 293 ///
 294 /// This version performs a copy.
 295 ///
 296 bool SocketRoutingQueue::DefaultRead(Data &receive, int timeout)
 297 {
 298         DataHandle buf = DefaultRead(timeout);
 299         if( !buf.get() )
 300                 return false;
 301
 302         // copy to desired buffer
 303         receive = *buf.get();
 304         return true;
 305 }
 306
 307 ///
 308 /// This version does not perform a copy.
 309 ///
 310 DataHandle SocketRoutingQueue::DefaultRead(int timeout)
 311 {
 312         if( m_seen_usb_error && timeout == -1 ) {
 313                 // If an error has been seen and not cleared then no
 314                 // more data will be read into the queue by
 315                 // DoRead(). Forcing the timeout to zero allows any
 316                 // data already in the queue to be read, but prevents
 317                 // waiting for data which will never arrive.
 318                 timeout = 0;
 319         }
 320
 321         // m_default handles its own locking
 322         // Be careful with the queue timeout, since its -1 means "forever"
 323         Data *buf = m_default.wait_pop(timeout == -1 ? m_timeout : timeout);
 324         return DataHandle(*this, buf);
 325 }
 326
 327 //
 328 // RegisterInterestAndType
 329 //
 330 /// Register an interest in data from a certain socket.  To read
 331 /// from that socket, use the SocketRead() function from then on.
 332 ///
 333 /// Any non-registered socket goes in the default queue
 334 /// and must be read by DefaultRead()
 335 ///
 336 /// If not null, handler is called when new data is read.  It will
 337 /// be called in the same thread instance that DoRead() is called from.
 338 /// Handler is passed the DataQueue Data pointer, and so no
 339 /// copying is done.  Once the handler returns, the data is
 340 /// considered processed and not added to the interested queue,
 341 /// but instead returned to m_free.
 342 ///
 343 /// Throws std::logic_error if already registered.
 344 ///
 345 void SocketRoutingQueue::RegisterInterestAndType(SocketId socket,
 346                                                  SocketDataHandlerPtr handler,
 347                                                  InterestType type)
 348 {
 349         // modifying our own std::map, need a lock
 350         scoped_lock lock(m_mutex);
 351
 352         SocketQueueMap::iterator qi = m_socketQueues.find(socket);
 353         if( qi != m_socketQueues.end() )
 354                 throw std::logic_error(_("RegisterInterest requesting a previously registered socket."));
 355
 356         m_socketQueues[socket] = QueueEntryPtr( new QueueEntry(handler, type) );
 357         m_interest = true;
 358 }
 359
 360 //
 361 // RegisterInterest
 362 //
 363 /// This behaves like RegisterInterest(SocketId, SocketDataHandlerPtr, InterestType)
 364 /// but defaults to an InterestType of SequenceAndDataPackets
 365 ///
 366 void SocketRoutingQueue::RegisterInterest(SocketId socket,
 367                                           SocketDataHandlerPtr handler)
 368 {
 369         RegisterInterestAndType(socket, handler, SequenceAndDataPackets);
 370 }
 371
 372 //
 373 // UnregisterInterest
 374 //
 375 /// Unregisters interest in data from the given socket, and discards
 376 /// any existing data in its interest queue.  Any new incoming data
 377 /// for this socket will be placed in the default queue.
 378 ///
 379 void SocketRoutingQueue::UnregisterInterest(SocketId socket)
 380 {
 381         // modifying our own std::map, need a lock
 382         scoped_lock lock(m_mutex);
 383
 384         SocketQueueMap::iterator qi = m_socketQueues.find(socket);
 385         if( qi == m_socketQueues.end() )
 386                 return; // nothing registered, done
 387
 388         // dump a record of any unused packets in the queue, for debugging
 389         DumpSocketQueue(qi->first, qi->second->m_queue);
 390
 391         // salvage all our data buffers
 392         m_free.append_from( qi->second->m_queue );
 393
 394         // remove the QueueEntryPtr from the map
 395         m_socketQueues.erase( qi );
 396
 397         // check the interest flag
 398         m_interest = m_socketQueues.size() > 0;
 399 }
 400
 401 //
 402 //
 403 // ChangeInterest
 404 //
 405 /// Changes the type of data that a client is interested in for a certain socket.
 406 /// Interest in the socket must have previously been registered by a call
 407 /// to RegisterInterest().
 408 void SocketRoutingQueue::ChangeInterest(SocketId socket, InterestType type)
 409 {
 410         // modifying our own std::map, need a lock
 411         scoped_lock lock(m_mutex);
 412
 413         SocketQueueMap::iterator qi = m_socketQueues.find(socket);
 414         if( qi == m_socketQueues.end() )
 415                 throw std::logic_error("ChangeInterest requires a previously registered socket.");
 416
 417         qi->second->m_type = type;
 418 }
 419
 420 //
 421 // SocketRead
 422 //
 423 /// Reads data from the interested socket cache.  Can only read
 424 /// from sockets that have been previously registered.
 425 ///
 426 /// Blocks until timeout or data is available.
 427 ///
 428 /// Returns false (or null pointer) on timeout and no data.
 429 /// With the return version of the function, there is no
 430 /// copying performed.
 431 ///
 432 /// Throws std::logic_error if a socket was requested that was
 433 /// not previously registered.
 434 ///
 435 /// Copying is performed with this function.
 436 ///
 437 bool SocketRoutingQueue::SocketRead(SocketId socket, Data &receive, int timeout)
 438 {
 439         DataHandle buf = SocketRead(socket, timeout);
 440         if( !buf.get() )
 441                 return false;
 442
 443         // copy to desired buffer
 444         receive = *buf.get();
 445         return true;
 446 }
 447
 448 ///
 449 /// Copying is not performed with this function.
 450 ///
 451 /// Throws std::logic_error if a socket was requested that was
 452 /// not previously registered.
 453 ///
 454 DataHandle SocketRoutingQueue::SocketRead(SocketId socket, int timeout)
 455 {
 456         QueueEntryPtr qep;
 457         DataQueue *dq = 0;
 458
 459         // accessing our own std::map, need a lock
 460         {
 461                 scoped_lock lock(m_mutex);
 462                 SocketQueueMap::iterator qi = m_socketQueues.find(socket);
 463                 if( qi == m_socketQueues.end() )
 464                         throw std::logic_error(_("SocketRead requested data from unregistered socket."));
 465
 466                 // got our queue, save the whole QueueEntryPtr (shared_ptr),
 467                 // and unlock, since we will be waiting on the DataQueue,
 468                 // not the socketQueues map
 469                 //
 470                 // This is safe, since even if UnregisterInterest is called,
 471                 // our pointer won't be deleted until our shared_ptr
 472                 // (QueueEntryPtr) goes out of scope.
 473                 //
 474                 // The remaining problem is that wait_pop() might wait
 475                 // forever if there is no timeout... c'est la vie.
 476                 // Should'a used a timeout. :-)
 477                 qep = qi->second;
 478                 dq = &qep->m_queue;
 479         }
 480
 481         // get data from DataQueue
 482         // Be careful with the queue timeout, since its -1 means "forever"
 483         Data *buf = dq->wait_pop(timeout == -1 ? m_timeout : timeout);
 484
 485         // specifically delete our copy of shared pointer, in a locked
 486         // environment
 487         {
 488                 scoped_lock lock(m_mutex);
 489                 qep.reset();
 490         }
 491
 492         return DataHandle(*this, buf);
 493 }
 494
 495 // Returns true if data is available for that socket.
 496 bool SocketRoutingQueue::IsAvailable(SocketId socket) const
 497 {
 498         scoped_lock lock(m_mutex);
 499         SocketQueueMap::const_iterator qi = m_socketQueues.find(socket);
 500         if( qi == m_socketQueues.end() )
 501                 return false;
 502         return qi->second->m_queue.size() > 0;
 503 }
 504
 505 //
 506 // DoRead
 507 //
 508 /// Called by the application's "read thread" to read the next usb
 509 /// packet and route it to the correct queue.  Returns after every
 510 /// read, even if a handler is associated with a queue.
 511 /// Note: this function is safe to call before SetUsbDevice() is
 512 /// called... it just doesn't do anything if there is no usb
 513 /// device to work with.
 514 ///
 515 /// Timeout is in milliseconds.
 516 //  This timeout is for the USB subsystem, so no special handling
 517 //  for it is needed... just use usbwrap's default timeout.
 518 void SocketRoutingQueue::DoRead(int timeout)
 519 {
 520         class ReadWaitSignal
 521         {
 522                 pthread_mutex_t &m_Mutex;
 523                 pthread_cond_t &m_Cond;
 524         public:
 525                 ReadWaitSignal(pthread_mutex_t &mut, pthread_cond_t &cond)
 526                         : m_Mutex(mut), m_Cond(cond)
 527                         {}
 528                 ~ReadWaitSignal()
 529                 {
 530                         scoped_lock wait(m_Mutex);
 531                         pthread_cond_signal(&m_Cond);
 532                 }
 533         } readwait(m_readwaitMutex, m_readwaitCond);
 534
 535         Usb::Device * volatile dev = 0;
 536         int readEp;
 537         DataHandle buf(*this, 0);
 538
 539         // if we are not connected to a USB device yet, just wait
 540         {
 541                 scoped_lock lock(m_mutex);
 542
 543                 if( !m_dev || m_seen_usb_error ) {
 544                         lock.unlock();  // unlock early, since we're sleeping
 545                         // sleep only a short time, since things could be
 546                         // in the process of setup or teardown
 547                         usleep(125000);
 548                         return;
 549                 }
 550
 551                 dev = m_dev;
 552                 readEp = m_readEp;
 553
 554                 // fetch a free buffer
 555                 Data *raw = m_free.pop();
 556                 if( !raw )
 557                         buf = DataHandle(*this, new Data);
 558                 else
 559                         buf = DataHandle(*this, raw);
 560         }
 561
 562         // take a chance and do the read unlocked, as this has the potential
 563         // for blocking for a while
 564         try {
 565
 566                 Data &data = *buf.get();
 567
 568                 if( !dev->BulkRead(readEp, data, timeout) )
 569                         return; // no data, done!
 570
 571                 MAKE_PACKET(pack, data);
 572
 573                 // make sure the size is right
 574                 if( data.GetSize() < SB_PACKET_SOCKET_SIZE )
 575                         return; // bad size, just skip
 576
 577                 // extract the socket from the packet
 578                 uint16_t socket = btohs(pack->socket);
 579
 580                 // if this is a sequence packet, handle it specially
 581                 if( Protocol::IsSequencePacket(data) ) {
 582                         // sequence.socket is a single byte
 583                         socket = pack->u.sequence.socket;
 584
 585                         //////////////////////////////////////////////
 586                         // ALWAYS queue sequence packets, so that
 587                         // the socket code can handle SyncSend()
 588                         if( !QueuePacket(socket, buf) ) {
 589                                 // if no queue available for this
 590                                 // socket, send it to the default
 591                                 // queue
 592                                 QueuePacket(m_default, buf);
 593                         }
 594
 595                         // done with sequence packet
 596                         return;
 597                 }
 598
 599                 // we have data, now route or queue it
 600                 if( RouteOrQueuePacket(socket, buf) )
 601                         return; // done
 602
 603                 // if we get here, send to default queue
 604                 QueuePacket(m_default, buf);
 605         }
 606         catch( Usb::Timeout & ) {
 607                 // this is expected... just ignore
 608         }
 609         catch( Usb::Error &ue ) {
 610                 // set the flag first, in case any of the handlers
 611                 // are able to recover from this error
 612                 m_seen_usb_error = true;
 613
 614                 // this is unexpected, but we're in a thread here...
 615                 // Need to iterate through all the registered handlers
 616                 // calling their error callback.
 617                 // Can't be locked when calling the callback, so need
 618                 // to make a list of them first.
 619                 scoped_lock lock(m_mutex);
 620                 std::vector<SocketDataHandlerPtr> handlers;
 621                 SocketQueueMap::iterator qi = m_socketQueues.begin();
 622                 while( qi != m_socketQueues.end() ) {
 623                         SocketDataHandlerPtr &sdh = qi->second->m_handler;
 624                         // is there a handler?
 625                         if( sdh ) {
 626                                 handlers.push_back(sdh);
 627                         }
 628                         ++qi;
 629                 }
 630
 631                 SocketDataHandlerPtr usb_error_handler = m_usb_error_dev_callback;
 632
 633                 lock.unlock();
 634                 std::vector<SocketDataHandlerPtr>::iterator hi = handlers.begin();
 635                 while( hi != handlers.end() ) {
 636                         (*hi)->Error(ue);
 637                         ++hi;
 638                 }
 639
 640                 // and finally, call the specific error callback if available
 641                 if( usb_error_handler.get() ) {
 642                         usb_error_handler->Error(ue);
 643                 }
 644         }
 645 }
 646
 647 void SocketRoutingQueue::SpinoffSimpleReadThread()
 648 {
 649         // signal that it's ok to run inside the thread
 650         if( m_continue_reading )
 651                 return; // already running
 652         m_continue_reading = true;
 653
 654         // Start USB read thread, to handle all routing
 655         int ret = pthread_create(&m_usb_read_thread, NULL, &SimpleReadThread, this);
 656         if( ret ) {
 657                 m_continue_reading = false;
 658                 throw Barry::ErrnoError(_("SocketRoutingQueue: Error creating USB read thread."), ret);
 659         }
 660 }
 661
 662 } // namespace Barry
 663