src/ptclib/sockagg.cxx

   1 /*
   2  * sockagg.cxx
   3  *
   4  * Generalised Socket Aggregation functions
   5  *
   6  * Portable Windows Library
   7  *
   8  * Copyright (C) 2005 Post Increment
   9  *
  10  * The contents of this file are subject to the Mozilla Public License
  11  * Version 1.0 (the "License"); you may not use this file except in
  12  * compliance with the License. You may obtain a copy of the License at
  13  * http://www.mozilla.org/MPL/
  14  *
  15  * Software distributed under the License is distributed on an "AS IS"
  16  * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
  17  * the License for the specific language governing rights and limitations
  18  * under the License.
  19  *
  20  * The Original Code is Portable Windows Library.
  21  *
  22  * The Initial Developer of the Original Code is Post Increment
  23  *
  24  * Portions of this code were written with the financial assistance of
  25  * Metreos Corporation (http://www.metros.com).
  26  *
  27  * Contributor(s): ______________________________________.
  28  *
  29  * $Log$
  30  * Revision 1.19  2006/11/20 03:17:24  csoutheren
  31  * Fixed incorrect check time on trace message
  32  *
  33  * Revision 1.18  2006/07/22 06:27:58  rjongbloed
  34  * Added auo-load of Winsock v2 library required by agreggated sockets.
  35  *
  36  * Revision 1.17  2006/03/13 23:34:21  csoutheren
  37  * Added log message when handle creates aggregator
  38  *
  39  * Revision 1.16  2006/03/09 05:32:59  csoutheren
  40  * Reverted to conservative locking strategy, with OnClose
  41  *
  42  * Revision 1.15  2006/03/07 11:04:56  csoutheren
  43  * Ensure socket aggregation not used on Linux
  44  *
  45  * Revision 1.14  2006/03/07 07:38:02  csoutheren
  46  * Refine timing windows on socket handling and cleanup unused code
  47  *
  48  * Revision 1.13  2006/03/06 02:37:26  csoutheren
  49  * Change handle locking to help prevent aggregation threads from hogging list
  50  *  access
  51  *
  52  * Revision 1.12  2006/03/02 07:50:38  csoutheren
  53  * Cleanup unused code
  54  * Add OnClose function
  55  *
  56  * Revision 1.11  2006/02/28 02:26:00  csoutheren
  57  * Renamed variable to be not same as member
  58  *
  59  * Revision 1.10  2006/02/28 02:08:02  csoutheren
  60  * Modified aggregation to load balance better
  61  *
  62  * Revision 1.9  2006/02/08 04:02:25  csoutheren
  63  * Added ability to enable and disable socket aggregation
  64  *
  65  * Revision 1.8  2006/01/23 05:57:39  csoutheren
  66  * More aggegator implementation
  67  *
  68  * Revision 1.7  2006/01/18 07:16:56  csoutheren
  69  * Latest version of socket aggregation code
  70  *
  71  * Revision 1.6  2006/01/05 11:39:32  rjongbloed
  72  * Fixed DevStudio warning
  73  *
  74  * Revision 1.5  2006/01/03 04:23:32  csoutheren
  75  * Fixed Unix implementation
  76  *
  77  * Revision 1.4  2005/12/23 07:49:27  csoutheren
  78  * Fixed Unix implementation
  79  *
  80  * Revision 1.3  2005/12/23 06:44:31  csoutheren
  81  * Working implementation
  82  *
  83  * Revision 1.2  2005/12/22 07:27:36  csoutheren
  84  * More implementation
  85  *
  86  * Revision 1.1  2005/12/22 03:55:52  csoutheren
  87  * Added initial version of socket aggregation classes
  88  *
  89  */
  90
  91 #ifdef __GNUC__
  92 #pragma implementation "sockagg.h"
  93 #endif
  94
  95
  96 #include <ptlib.h>
  97 #include <ptclib/sockagg.h>
  98
  99 #include <fcntl.h>
 100
 101 ////////////////////////////////////////////////////////////////
 102
 103 #if _WIN32
 104
 105 #pragma comment(lib, "Ws2_32.lib")
 106
 107 class LocalEvent : public PHandleAggregator::EventBase
 108 {
 109   public:
 110     LocalEvent()
 111     {
 112       event = CreateEvent(NULL, TRUE, FALSE,NULL);
 113       PAssert(event != NULL, "CreateEvent failed");
 114     }
 115
 116     ~LocalEvent()
 117     { CloseHandle(event); }
 118
 119     PAggregatorFD::FD GetHandle()
 120     { return event; }
 121
 122     void Set()
 123     { SetEvent(event);  }
 124
 125     void Reset()
 126     { ResetEvent(event); }
 127
 128   protected:
 129     HANDLE event;
 130 };
 131
 132 PAggregatorFD::PAggregatorFD(SOCKET v)
 133   : socket(v)
 134 {
 135   fd = WSACreateEvent();
 136   PAssert(WSAEventSelect(socket, fd, FD_READ | FD_CLOSE) == 0, "WSAEventSelect failed");
 137 }
 138
 139 PAggregatorFD::~PAggregatorFD()
 140 {
 141   WSACloseEvent(fd);
 142 }
 143
 144 bool PAggregatorFD::IsValid()
 145 {
 146   return socket != INVALID_SOCKET;
 147 }
 148
 149 #else // #if _WIN32
 150
 151 class LocalEvent : public PHandleAggregator::EventBase
 152 {
 153   public:
 154     LocalEvent()
 155     { ::pipe(fds); }
 156
 157     virtual ~LocalEvent()
 158     {
 159       close(fds[0]);
 160       close(fds[1]);
 161     }
 162
 163     PAggregatorFD::FD GetHandle()
 164     { return fds[0]; }
 165
 166     void Set()
 167     { char ch; write(fds[1], &ch, 1); }
 168
 169     void Reset()
 170     { char ch; read(fds[0], &ch, 1); }
 171
 172   protected:
 173     int fds[2];
 174 };
 175
 176 PAggregatorFD::PAggregatorFD(int v)
 177   : fd(v)
 178 {
 179 }
 180
 181 PAggregatorFD::~PAggregatorFD()
 182 {
 183 }
 184
 185 bool PAggregatorFD::IsValid()
 186 {
 187   return fd >= 0;
 188 }
 189
 190 #endif // #endif _WIN32
 191
 192
 193 ////////////////////////////////////////////////////////////////
 194
 195 PHandleAggregator::WorkerThreadBase::WorkerThreadBase(EventBase & _event)
 196   : PThread(100, NoAutoDeleteThread, NormalPriority, "Aggregator:%0x"), event(_event), listChanged(TRUE), shutdown(FALSE)
 197 {
 198 }
 199
 200 class WorkerThread : public PHandleAggregator::WorkerThreadBase
 201 {
 202   public:
 203     WorkerThread()
 204       : WorkerThreadBase(localEvent)
 205     { }
 206
 207     ~WorkerThread()
 208     {
 209     }
 210
 211     void Trigger()
 212     { localEvent.Set(); }
 213
 214     LocalEvent localEvent;
 215 };
 216
 217
 218 ////////////////////////////////////////////////////////////////
 219
 220 PHandleAggregator::PHandleAggregator(unsigned _max)
 221   : maxWorkerSize(_max)
 222 {
 223 }
 224
 225 BOOL PHandleAggregator::AddHandle(PAggregatedHandle * handle)
 226 {
 227   // perform the handle init function
 228   if (!handle->Init())
 229     return FALSE;
 230
 231   PWaitAndSignal m(listMutex);
 232
 233   // if the maximum number of worker threads has been reached, then
 234   // use the worker thread with the minimum number of handles
 235   if (workers.size() >= maxWorkerSize) {
 236     WorkerList_t::iterator minWorker = workers.end();
 237     size_t minSizeFound = 0x7ffff;
 238     WorkerList_t::iterator r;
 239     for (r = workers.begin(); r != workers.end(); ++r) {
 240       WorkerThreadBase & worker = **r;
 241       PWaitAndSignal m2(worker.workerMutex);
 242       if (!worker.shutdown && (worker.handleList.size() <= minSizeFound)) {
 243         minSizeFound = worker.handleList.size();
 244         minWorker     = r;
 245       }
 246     }
 247
 248     // add the worker to the thread
 249     PAssert(minWorker != workers.end(), "could not find minimum worker");
 250
 251     WorkerThreadBase & worker = **minWorker;
 252     PWaitAndSignal m2(worker.workerMutex);
 253     worker.handleList.push_back(handle);
 254     PTRACE(4, "SockAgg\tAdding handle " << (void *)handle << " to aggregator - " << worker.handleList.size() << " handles");
 255     worker.listChanged = TRUE;
 256     worker.Trigger();
 257     return TRUE;
 258   }
 259
 260   PTRACE(4, "SockAgg\tCreating new aggregator for " << (void *)handle);
 261
 262   // no worker threads usable, create a new one
 263   WorkerThread * worker = new WorkerThread;
 264   worker->handleList.push_back(handle);
 265   worker->Resume();
 266   workers.push_back(worker);
 267
 268   PTRACE(4, "SockAgg\tAdding handle " << (void *)handle << " to new aggregator");
 269
 270   return TRUE;
 271 }
 272
 273 BOOL PHandleAggregator::RemoveHandle(PAggregatedHandle * handle)
 274 {
 275   listMutex.Wait();
 276
 277   // look for the thread containing the handle we need to delete
 278   WorkerList_t::iterator r;
 279   for (r = workers.begin(); r != workers.end(); ++r) {
 280     WorkerThreadBase * worker = *r;
 281
 282     // lock the worker
 283     worker->workerMutex.Wait();
 284
 285     PAggregatedHandleList_t & hList = worker->handleList;
 286
 287     // if handle is not in this thread, then continue searching
 288     PAggregatedHandleList_t::iterator s = find(hList.begin(), hList.end(), handle);
 289     if (s == hList.end()) {
 290       worker->workerMutex.Signal();
 291       continue;
 292     }
 293
 294     PAssert(*s == handle, "Found handle is not correct!");
 295
 296     PAssert(!handle->beingProcessed, "trying to close handle that is in use");
 297
 298     // remove the handle from the worker's list of handles
 299     worker->handleList.erase(s);
 300
 301     // do the de-init action
 302     handle->DeInit();
 303
 304     // delete the handle if autodelete enabled
 305     if (handle->autoDelete)
 306       delete handle;
 307
 308     // if the worker thread has enough handles to keep running, trigger it to update
 309     if (worker->handleList.size() > 0) {
 310       PTRACE(4, "SockAgg\tRemoved handle " << (void *)handle << " from aggregator - " << worker->handleList.size() << " handles remaining");
 311       worker->listChanged = TRUE;
 312       worker->Trigger();
 313       worker->workerMutex.Signal();
 314
 315       listMutex.Signal();
 316
 317       return TRUE;
 318     }
 319
 320     PTRACE(4, "SockAgg\tworker thread empty - closing down");
 321
 322     // remove the worker thread from the list of workers
 323     workers.erase(r);
 324
 325     // shutdown the thread
 326     worker->shutdown = TRUE;
 327     worker->Trigger();
 328     worker->workerMutex.Signal();
 329
 330     // unlock the list
 331     listMutex.Signal();
 332
 333     // the worker is now finished
 334     if (!worker->WaitForTermination(10000)) {
 335       PTRACE(4, "SockAgg\tWorker did not terminate promptly");
 336     }
 337
 338     delete worker;
 339
 340     return TRUE;
 341   }
 342
 343   listMutex.Signal();
 344
 345   PAssertAlways("Cannot find aggregator handle");
 346
 347   return FALSE;
 348 }
 349
 350 ////////////////////////////////////////////////////////////////
 351
 352 typedef std::vector<PAggregatorFD::FD> fdList_t;
 353
 354 #ifdef _WIN32
 355 #define FDLIST_SIZE     WSA_MAXIMUM_WAIT_EVENTS
 356 #else
 357 #define FDLIST_SIZE     64
 358 #endif
 359
 360 void PHandleAggregator::WorkerThreadBase::Main()
 361 {
 362   PTRACE(4, "SockAgg\taggregator started");
 363
 364   fdList_t                  fdList;
 365   PAggregatorFDList_t       aggregatorFdList;
 366
 367   typedef std::map<PAggregatorFD::FD, PAggregatedHandle *> PAggregatorFdToHandleMap_t;
 368   PAggregatorFdToHandleMap_t aggregatorFdToHandleMap;
 369
 370   for (;;) {
 371
 372     // create the list of fds to wait on and find minimum timeout
 373     PTimeInterval timeout(PMaxTimeInterval);
 374     PAggregatedHandle * timeoutHandle = NULL;
 375
 376 #ifndef _WIN32
 377     fd_set rfds;
 378     FD_ZERO(&rfds);
 379     int maxFd = 0;
 380 #endif
 381
 382     {
 383       PWaitAndSignal m(workerMutex);
 384
 385       // check for shutdown
 386       if (shutdown)
 387         break;
 388
 389       // if the list of handles has changed, clear the list of handles
 390       if (listChanged) {
 391         aggregatorFdList.erase       (aggregatorFdList.begin(),      aggregatorFdList.end());
 392         aggregatorFdList.reserve     (FDLIST_SIZE);
 393         fdList.erase                 (fdList.begin(),                fdList.end());
 394         fdList.reserve               (FDLIST_SIZE);
 395         aggregatorFdToHandleMap.erase(aggregatorFdToHandleMap.begin(),         aggregatorFdToHandleMap.end());
 396       }
 397
 398       PAggregatedHandleList_t::iterator r;
 399       for (r = handleList.begin(); r != handleList.end(); ++r) {
 400         PAggregatedHandle * handle = *r;
 401
 402         if (handle->closed)
 403           continue;
 404
 405         if (listChanged) {
 406           PAggregatorFDList_t fds = handle->GetFDs();
 407           PAggregatorFDList_t::iterator s;
 408           for (s = fds.begin(); s != fds.end(); ++s) {
 409             fdList.push_back          ((*s)->fd);
 410             aggregatorFdList.push_back((*s));
 411             aggregatorFdToHandleMap.insert(PAggregatorFdToHandleMap_t::value_type((*s)->fd, handle));
 412           }
 413         }
 414
 415         if (!handle->IsPreReadDone()) {
 416           handle->PreRead();
 417           handle->SetPreReadDone();
 418         }
 419
 420         PTimeInterval t = handle->GetTimeout();
 421         if (t < timeout) {
 422           timeout = t;
 423           timeoutHandle = handle;
 424         }
 425       }
 426
 427       // add in the event fd
 428       if (listChanged) {
 429         fdList.push_back(event.GetHandle());
 430         listChanged = FALSE;
 431       }
 432
 433 #ifndef _WIN32
 434       // create the list of FDs
 435       fdList_t::iterator s;
 436       for (s = fdList.begin(); s != fdList.end(); ++s) {
 437         FD_SET(*s, &rfds);
 438         maxFd = PMAX(maxFd, *s);
 439       }
 440 #endif
 441     } // workerMutex goes out of scope
 442
 443 #ifdef _WIN32
 444     DWORD nCount = fdList.size();
 445     DWORD ret = WSAWaitForMultipleEvents(nCount,
 446                                          &fdList[0],
 447                                          false,
 448                                          (timeout == PMaxTimeInterval) ? WSA_INFINITE : (DWORD)timeout.GetMilliSeconds(),
 449                                          FALSE);
 450
 451     if (ret == WAIT_FAILED) {
 452       DWORD err = WSAGetLastError();
 453       PTRACE(1, "SockAgg\tWSAWaitForMultipleEvents error " << err);
 454     }
 455
 456     {
 457       PWaitAndSignal m(workerMutex);
 458
 459       // check for shutdown
 460       if (shutdown)
 461         break;
 462
 463       if (ret == WAIT_TIMEOUT) {
 464         // make sure the handle did not disappear while we were waiting
 465         PAggregatedHandleList_t::iterator s = find(handleList.begin(), handleList.end(), timeoutHandle);
 466         if (s == handleList.end()) {
 467           PTRACE(4, "SockAgg\tHandle was removed while waiting");
 468         }
 469         else {
 470           PTime start;
 471
 472           timeoutHandle->beingProcessed = TRUE;
 473           timeoutHandle->closed = !timeoutHandle->OnRead();
 474           timeoutHandle->beingProcessed = FALSE;
 475
 476           unsigned duration = (unsigned)(PTime() - start).GetMilliSeconds();
 477           if (duration > 50) {
 478             PTRACE(4, "SockAgg\tWarning - aggregator read routine was of extended duration = " << duration << " msecs");
 479           }
 480           if (!timeoutHandle->closed)
 481             timeoutHandle->SetPreReadDone(FALSE);
 482         }
 483       }
 484
 485       else if (WAIT_OBJECT_0 <= ret && ret <= (WAIT_OBJECT_0 + nCount - 1)) {
 486         DWORD index = ret - WAIT_OBJECT_0;
 487
 488         // if the event was triggered, redo the select
 489         if (index == nCount-1) {
 490           event.Reset();
 491           continue;
 492         }
 493
 494         PAggregatorFD * fd = aggregatorFdList[index];
 495         PAssert(fdList[index] == fd->fd, "Mismatch in fd lists");
 496
 497         PAggregatorFdToHandleMap_t::iterator r = aggregatorFdToHandleMap.find(fd->fd);
 498         if (r != aggregatorFdToHandleMap.end()) {
 499           PAggregatedHandle * handle = r->second;
 500           PAggregatedHandleList_t::iterator s = find(handleList.begin(), handleList.end(), handle);
 501           if (s == handleList.end()) {
 502             PTRACE(4, "SockAgg\tHandle was removed while waiting");
 503           }
 504           else {
 505             WSANETWORKEVENTS events;
 506             WSAEnumNetworkEvents(fd->socket, fd->fd, &events);
 507             if (events.lNetworkEvents != 0) {
 508
 509               // check for read events first so we process any data that arrives before closing
 510               if ((events.lNetworkEvents & FD_READ) != 0) {
 511
 512                 PTime start;
 513
 514                 handle->beingProcessed = TRUE;
 515                 handle->closed = !handle->OnRead();
 516                 handle->beingProcessed = FALSE;
 517
 518                 unsigned duration = (unsigned)(PTime() - start).GetMilliSeconds();
 519                 PTRACE_IF(4, duration > 50, "SockAgg\tWarning - aggregator read routine was of extended duration = " << duration << " msecs");
 520               }
 521
 522               // check for socket close
 523               if ((events.lNetworkEvents & FD_CLOSE) != 0)
 524                 handle->closed = TRUE;
 525
 526               if (!handle->closed) {
 527                 // prepare for next read
 528                 handle->SetPreReadDone(FALSE);
 529               } else {
 530                 handle->beingProcessed = TRUE;
 531                 handle->OnClose();
 532                 handle->beingProcessed = FALSE;
 533
 534                 // make sure the list is refreshed without the closed socket
 535                 listChanged = TRUE;
 536               }
 537             }
 538           }
 539         }
 540       }
 541     } // workerMutex goes out of scope
 542
 543 #else
 544
 545 #error "aggregation not yet implemented on Unix"
 546
 547 #if 0
 548
 549     P_timeval pv = timeout;
 550     int ret = ::select(maxFd+1, &rfds, NULL, NULL, pv);
 551
 552     if (ret < 0) {
 553       PTRACE(1, "SockAgg\tSelect failed with error " << errno);
 554     }
 555
 556     // loop again if nothing was ready
 557     if (ret <= 0)
 558       continue;
 559
 560     {
 561       PWaitAndSignal m(workerMutex);
 562
 563       // check for shutdown
 564       if (shutdown)
 565         break;
 566
 567       if (ret == 0) {
 568         PTime start;
 569         BOOL closed = !timeoutHandle->OnRead();
 570         unsigned duration = (unsigned)(PTime() - start).GetMilliSeconds();
 571         if (duration > 50) {
 572           PTRACE(4, "SockAgg\tWarning - aggregator read routine was of extended duration = " << duration << " msecs");
 573         }
 574         if (!closed)
 575           timeoutHandle->SetPreReadDone(FALSE);
 576       }
 577
 578       // check the event first
 579       else if (FD_ISSET(event.GetHandle(), &rfds)) {
 580         event.Reset();
 581         continue;
 582       }
 583
 584       else {
 585         PAggregatorFD * fd = aggregatorFdList[ret];
 586         PAssert(fdList[ret] == fd->fd, "Mismatch in fd lists");
 587         PAggregatorFdToHandleMap_t::iterator r = aggregatorFdToHandleMap.find(fd->fd);
 588         if (r != aggregatorFdToHandleMap.end()) {
 589           PAggregatedHandle * handle = r->second;
 590           PTime start;
 591           BOOL closed = !handle->OnRead();
 592           unsigned duration = (unsigned)(PTime() - start).GetMilliSeconds();
 593           if (duration > 50) {
 594             PTRACE(4, "SockAgg\tWarning - aggregator read routine was of extended duration = " << duration << " msecs");
 595           }
 596           if (!closed)
 597             handle->SetPreReadDone(FALSE);
 598         }
 599       }
 600     } // workerMutex goes out of scope
 601 #endif // #if 0
 602 #endif
 603   }
 604
 605   PTRACE(4, "SockAgg\taggregator finished");
 606 }
 607
 608