adaptors/base/performance-logging/networking/network-performance-server.cpp

   1 /*
   2  * Copyright (c) 2015 Samsung Electronics Co., Ltd.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  * http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  *
  16  */
  17
  18 // CLASS HEADER
  19 #include "network-performance-server.h"
  20
  21 // EXTERNAL INCLUDES
  22 #include <boost/thread.hpp>
  23
  24 // INTERNAL INCLUDES
  25 #include <base/performance-logging/performance-marker.h>
  26
  27 namespace Dali
  28 {
  29
  30 namespace Internal
  31 {
  32
  33 namespace Adaptor
  34 {
  35
  36 namespace // un-named namespace
  37 {
  38 const unsigned int SERVER_PORT = 3031;
  39 const unsigned int MAXIMUM_PORTS_TO_TRY = 10; ///< if port in use, try up to SERVER_PORT + 10
  40 const unsigned int CONNECTION_BACKLOG = 2;   ///<  maximum length of the queue of pending connections.
  41 const unsigned int SOCKET_READ_BUFFER_SIZE = 4096;
  42 typedef Vector< NetworkPerformanceClient*> ClientList;
  43 }
  44
  45 NetworkPerformanceServer::NetworkPerformanceServer( AdaptorInternalServices& adaptorServices,
  46                                                     const EnvironmentOptions& logOptions )
  47 : mTriggerEventFactory( adaptorServices.GetTriggerEventFactoryInterface() ),
  48   mSocketFactory( adaptorServices.GetSocketFactoryInterface() ),
  49   mLogOptions( logOptions ),
  50   mServerThread( NULL ),
  51   mListeningSocket( NULL ),
  52   mClientUniqueId( 0 ),
  53   mClientCount( 0 ),
  54   mLogFunctionInstalled( false )
  55 {
  56 }
  57
  58 NetworkPerformanceServer::~NetworkPerformanceServer()
  59 {
  60   Stop();
  61
  62   if( mLogFunctionInstalled )
  63   {
  64     mLogOptions.UnInstallLogFunction();
  65   }
  66 }
  67
  68 void NetworkPerformanceServer::Start()
  69 {
  70   // start a thread to listen for incoming connection requests
  71   if (! mServerThread )
  72   {
  73     if( mListeningSocket )
  74     {
  75       mSocketFactory.DestroySocket( mListeningSocket );
  76     }
  77     mListeningSocket = mSocketFactory.NewSocket( SocketInterface::TCP);
  78     mListeningSocket->ReuseAddress( true );
  79
  80     bool bound = false;
  81     unsigned int basePort = 0;
  82
  83     // try a small range of ports, so if multiple Dali apps are running you can select
  84     // which one to connect to
  85     while( !bound && ( basePort <  MAXIMUM_PORTS_TO_TRY ))
  86     {
  87       bound = mListeningSocket->Bind( SERVER_PORT + basePort );
  88       if( !bound )
  89       {
  90         basePort++;
  91       }
  92     }
  93     if(!bound )
  94     {
  95       DALI_LOG_ERROR("Failed to bind to a port \n");
  96       return;
  97     }
  98
  99     mListeningSocket->Listen( CONNECTION_BACKLOG );
 100
 101     // start a thread which will block waiting for new connections
 102     mServerThread = new boost::thread(boost::bind(&NetworkPerformanceServer::ConnectionListener, this));
 103
 104     Dali::Integration::Log::LogMessage(Integration::Log::DebugInfo, "~~~ NetworkPerformanceServer started on port %d ~~~ \n",  SERVER_PORT + basePort);
 105
 106   }
 107 }
 108 void NetworkPerformanceServer::Stop()
 109 {
 110   if( !mServerThread )
 111   {
 112     return;
 113   }
 114   // close the server thread to prevent any new connections
 115   mListeningSocket->ExitSelect();
 116
 117   // wait for the thread to exit.
 118   mServerThread->join();
 119
 120   // close the socket
 121   mListeningSocket->CloseSocket();
 122
 123   delete mServerThread;
 124   mSocketFactory.DestroySocket( mListeningSocket );
 125
 126   mServerThread = NULL;
 127   mListeningSocket = NULL;
 128
 129   // this will tell all client threads to quit
 130   StopClients();
 131
 132   // wait for all threads to exit and the client count to hit zero
 133   {
 134     boost::mutex::scoped_lock lock(mClientListMutex);
 135
 136     while (mClientCount != 0)
 137     {
 138       mClientCountUpdated.wait(lock);
 139     }
 140   }
 141 }
 142
 143 bool NetworkPerformanceServer::IsRunning() const
 144 {
 145   if (mServerThread )
 146   {
 147     return true;
 148   }
 149   return false;
 150 }
 151
 152 void NetworkPerformanceServer::ClientThread( NetworkPerformanceClient* client )
 153 {
 154   mClientCount++;
 155
 156   SocketInterface& socket( client->GetSocket() );
 157
 158   for( ;; )
 159   {
 160     SocketInterface::SelectReturn ret = socket.Select();
 161
 162     if( ret == SocketInterface::DATA_AVAILABLE )
 163     {
 164      // Read
 165       char buffer[ SOCKET_READ_BUFFER_SIZE ];
 166       unsigned int  bytesRead;
 167
 168       bool ok  = socket.Read( buffer, sizeof( buffer ) , bytesRead);
 169       if( ok && ( bytesRead > 0) )
 170       {
 171         client->ProcessCommand( buffer, bytesRead );
 172       }
 173       else   // if bytesRead == 0, then client closed connection, if ok == false then an error
 174       {
 175         DeleteClient( client );
 176         return;
 177       }
 178     }
 179     else // ret == QUIT or ERROR
 180     {
 181       DeleteClient( client);
 182       return;
 183     }
 184   }
 185 }
 186
 187 void NetworkPerformanceServer::ConnectionListener()
 188 {
 189   // install Dali logging function for this thread
 190   if( !mLogFunctionInstalled )
 191   {
 192     mLogOptions.InstallLogFunction();
 193     mLogFunctionInstalled = true;
 194   }
 195
 196   for( ;; )
 197   {
 198     // this will block, waiting for a client to connect
 199     // or for mListeningSocket->ExitSelect() to be called
 200
 201     SocketInterface::SelectReturn ret = mListeningSocket->Select();
 202
 203     if( ret == SocketInterface::DATA_AVAILABLE )
 204     {
 205       SocketInterface* clientSocket = mListeningSocket->Accept();
 206
 207       // new connection made, spawn a thread to handle it
 208       NetworkPerformanceClient* client = AddClient( clientSocket );
 209       new boost::thread(boost::bind(&NetworkPerformanceServer::ClientThread, this, client));
 210     }
 211     else // ret == SocketInterface::QUIT or SocketInterface::ERROR
 212     {
 213       return;
 214     }
 215   }
 216 }
 217
 218 NetworkPerformanceClient* NetworkPerformanceServer::AddClient( SocketInterface* clientSocket )
 219 {
 220   // This function is only called from the listening thread
 221   NetworkPerformanceClient* client= new NetworkPerformanceClient(clientSocket,
 222                                                                  mClientUniqueId++,
 223                                                                  mTriggerEventFactory,
 224                                                                  *this,
 225                                                                  mSocketFactory);
 226
 227   // protect the mClients list which can be accessed from multiple threads.
 228   boost::mutex::scoped_lock sharedDatalock( mClientListMutex );
 229
 230   mClients.PushBack( client );
 231
 232   return client;
 233 }
 234
 235 void NetworkPerformanceServer::DeleteClient( NetworkPerformanceClient* client )
 236 {
 237   // protect the mClients list while modifying
 238   boost::mutex::scoped_lock sharedDatalock( mClientListMutex );
 239
 240   // remove from the list, and delete it
 241   for( ClientList::Iterator iter = mClients.Begin(); iter != mClients.End() ; ++iter )
 242   {
 243     if( (*iter) == client )
 244     {
 245       mClients.Erase( iter );
 246       delete client;
 247
 248       // if there server is shutting down, it waits for client count to hit zero
 249       mClientCount--;
 250
 251       // lets the server know the client count has been modified
 252       mClientCountUpdated.notify_one();
 253       return;
 254     }
 255   }
 256 }
 257
 258 void NetworkPerformanceServer::SendData( const char* const data, unsigned int bufferSizeInBytes,unsigned int clientId )
 259 {
 260   if( ! mClientCount )
 261   {
 262     return;
 263   }
 264
 265   // prevent clients been added / deleted while transmiting data
 266   boost::mutex::scoped_lock sharedDatalock( mClientListMutex );
 267
 268   for( ClientList::Iterator iter = mClients.Begin(); iter != mClients.End() ; ++iter )
 269   {
 270     NetworkPerformanceClient* client = (*iter);
 271     if( client->GetId() == clientId )
 272     {
 273       client->WriteSocket(data ,bufferSizeInBytes);
 274       return;
 275     }
 276   }
 277 }
 278
 279 void NetworkPerformanceServer::TransmitMarker( const PerformanceMarker& marker, const char* const description )
 280 {
 281   if( ! IsRunning() )
 282   {
 283     return;
 284   }
 285   // prevent clients been added / deleted while transmiting data
 286   boost::mutex::scoped_lock sharedDatalock( mClientListMutex );
 287
 288   for( ClientList::Iterator iter = mClients.Begin(); iter != mClients.End() ; ++iter )
 289   {
 290     NetworkPerformanceClient* client = (*iter);
 291     client->TransmitMarker( marker, description );
 292   }
 293 }
 294
 295
 296 void NetworkPerformanceServer::StopClients()
 297 {
 298   // prevent clients been added / deleted while stopping all clients
 299   boost::mutex::scoped_lock sharedDatalock( mClientListMutex );
 300
 301   for( ClientList::Iterator iter = mClients.Begin(); iter != mClients.End() ; ++iter )
 302   {
 303     NetworkPerformanceClient* client = (*iter);
 304     // stop the client from waiting for new commands, and exit from it's thread
 305     client->ExitSelect();
 306   }
 307 }
 308
 309 } // namespace Internal
 310
 311 } // namespace Adaptor
 312
 313 } // namespace Dali