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
  22 // INTERNAL INCLUDES
  23 #include <base/performance-logging/performance-marker.h>
  24
  25 namespace Dali
  26 {
  27
  28 namespace Internal
  29 {
  30
  31 namespace Adaptor
  32 {
  33
  34 namespace // un-named namespace
  35 {
  36 const unsigned int SERVER_PORT = 3031;
  37 const unsigned int MAXIMUM_PORTS_TO_TRY = 10; ///< if port in use, try up to SERVER_PORT + 10
  38 const unsigned int CONNECTION_BACKLOG = 2;   ///<  maximum length of the queue of pending connections.
  39 const unsigned int SOCKET_READ_BUFFER_SIZE = 4096;
  40 typedef Vector< NetworkPerformanceClient*> ClientList;
  41
  42 /**
  43  * POD passed to client thread on startup
  44  */
  45 struct ClientThreadInfo
  46 {
  47   NetworkPerformanceServer* server;
  48   NetworkPerformanceClient* client;
  49 };
  50 }
  51
  52 NetworkPerformanceServer::NetworkPerformanceServer( AdaptorInternalServices& adaptorServices,
  53                                                     const EnvironmentOptions& logOptions )
  54 : mTriggerEventFactory( adaptorServices.GetTriggerEventFactoryInterface() ),
  55   mSocketFactory( adaptorServices.GetSocketFactoryInterface() ),
  56   mLogOptions( logOptions ),
  57   mServerThread( 0 ),
  58   mListeningSocket( NULL ),
  59   mClientUniqueId( 0 ),
  60   mClientCount( 0 ),
  61   mLogFunctionInstalled( false )
  62 {
  63 }
  64
  65 NetworkPerformanceServer::~NetworkPerformanceServer()
  66 {
  67   Stop();
  68
  69   if( mLogFunctionInstalled )
  70   {
  71     mLogOptions.UnInstallLogFunction();
  72   }
  73 }
  74
  75 void NetworkPerformanceServer::Start()
  76 {
  77   // start a thread to listen for incoming connection requests
  78   if (! mServerThread )
  79   {
  80     if( mListeningSocket )
  81     {
  82       mSocketFactory.DestroySocket( mListeningSocket );
  83     }
  84     mListeningSocket = mSocketFactory.NewSocket( SocketInterface::TCP);
  85     mListeningSocket->ReuseAddress( true );
  86
  87     bool bound = false;
  88     unsigned int basePort = 0;
  89
  90     // try a small range of ports, so if multiple Dali apps are running you can select
  91     // which one to connect to
  92     while( !bound && ( basePort <  MAXIMUM_PORTS_TO_TRY ))
  93     {
  94       bound = mListeningSocket->Bind( SERVER_PORT + basePort );
  95       if( !bound )
  96       {
  97         basePort++;
  98       }
  99     }
 100     if(!bound )
 101     {
 102       DALI_LOG_ERROR("Failed to bind to a port \n");
 103       return;
 104     }
 105
 106     mListeningSocket->Listen( CONNECTION_BACKLOG );
 107
 108     // start a thread which will block waiting for new connections
 109     int error = pthread_create( &mServerThread, NULL, ConnectionListenerFunc, this );
 110     DALI_ASSERT_ALWAYS( !error && "pthread create failed" );
 111
 112     Dali::Integration::Log::LogMessage(Integration::Log::DebugInfo, "~~~ NetworkPerformanceServer started on port %d ~~~ \n",  SERVER_PORT + basePort);
 113
 114   }
 115 }
 116 void NetworkPerformanceServer::Stop()
 117 {
 118   if( !mServerThread )
 119   {
 120     return;
 121   }
 122   // close the server thread to prevent any new connections
 123   mListeningSocket->ExitSelect();
 124
 125   // wait for the thread to exit.
 126   void* exitValue;
 127   pthread_join( mServerThread, &exitValue );
 128
 129   // close the socket
 130   mListeningSocket->CloseSocket();
 131
 132   mSocketFactory.DestroySocket( mListeningSocket );
 133
 134   mListeningSocket = NULL;
 135
 136   // this will tell all client threads to quit
 137   StopClients();
 138
 139 }
 140
 141 bool NetworkPerformanceServer::IsRunning() const
 142 {
 143   if (mServerThread )
 144   {
 145     return true;
 146   }
 147   return false;
 148 }
 149
 150 void NetworkPerformanceServer::ClientThread( NetworkPerformanceClient* client )
 151 {
 152   mClientCount++;
 153
 154   SocketInterface& socket( client->GetSocket() );
 155
 156   for( ;; )
 157   {
 158     SocketInterface::SelectReturn ret = socket.Select();
 159
 160     if( ret == SocketInterface::DATA_AVAILABLE )
 161     {
 162      // Read
 163       char buffer[ SOCKET_READ_BUFFER_SIZE ];
 164       unsigned int  bytesRead;
 165
 166       bool ok  = socket.Read( buffer, sizeof( buffer ) , bytesRead);
 167       if( ok && ( bytesRead > 0) )
 168       {
 169         client->ProcessCommand( buffer, bytesRead );
 170       }
 171       else   // if bytesRead == 0, then client closed connection, if ok == false then an error
 172       {
 173         DeleteClient( client );
 174         return;
 175       }
 176     }
 177     else // ret == QUIT or ERROR
 178     {
 179       DeleteClient( client);
 180       return;
 181     }
 182   }
 183 }
 184
 185 void NetworkPerformanceServer::ConnectionListener()
 186 {
 187   // install Dali logging function for this thread
 188   if( !mLogFunctionInstalled )
 189   {
 190     mLogOptions.InstallLogFunction();
 191     mLogFunctionInstalled = true;
 192   }
 193
 194   for( ;; )
 195   {
 196     // this will block, waiting for a client to connect
 197     // or for mListeningSocket->ExitSelect() to be called
 198
 199     SocketInterface::SelectReturn ret = mListeningSocket->Select();
 200
 201     if( ret == SocketInterface::DATA_AVAILABLE )
 202     {
 203       SocketInterface* clientSocket = mListeningSocket->Accept();
 204
 205       // new connection made, spawn a thread to handle it
 206       pthread_t* clientThread = new pthread_t();
 207
 208       NetworkPerformanceClient* client = AddClient( clientSocket, clientThread );
 209
 210       ClientThreadInfo* info = new ClientThreadInfo;
 211       info->client = client;
 212       info->server = this;
 213
 214       int error = pthread_create( clientThread, NULL, ClientThreadFunc, info );
 215       DALI_ASSERT_ALWAYS( !error && "pthread create failed" );
 216
 217     }
 218     else // ret == SocketInterface::QUIT or SocketInterface::ERROR
 219     {
 220       return;
 221     }
 222   }
 223 }
 224
 225 void* NetworkPerformanceServer::ClientThreadFunc( void* data )
 226 {
 227   ClientThreadInfo* info = static_cast<ClientThreadInfo*>( data );
 228   info->server->ClientThread( info->client );
 229   delete info;
 230   return NULL;
 231 }
 232
 233 NetworkPerformanceClient* NetworkPerformanceServer::AddClient( SocketInterface* clientSocket, pthread_t* clientThread )
 234 {
 235   // This function is only called from the listening thread
 236   NetworkPerformanceClient* client= new NetworkPerformanceClient( clientThread,
 237                                                                   clientSocket,
 238                                                                   mClientUniqueId++,
 239                                                                   mTriggerEventFactory,
 240                                                                   *this,
 241                                                                   mSocketFactory);
 242
 243   // protect the mClients list which can be accessed from multiple threads.
 244   Mutex::ScopedLock lock( mClientListMutex );
 245
 246   mClients.PushBack( client );
 247
 248   return client;
 249 }
 250
 251 void NetworkPerformanceServer::DeleteClient( NetworkPerformanceClient* client )
 252 {
 253   // protect the mClients list while modifying
 254   Mutex::ScopedLock lock( mClientListMutex );
 255
 256   // remove from the list, and delete it
 257   for( ClientList::Iterator iter = mClients.Begin(); iter != mClients.End() ; ++iter )
 258   {
 259     if( (*iter) == client )
 260     {
 261       mClients.Erase( iter );
 262       delete client;
 263
 264       // if there server is shutting down, it waits for client count to hit zero
 265       mClientCount--;
 266
 267       return;
 268     }
 269   }
 270 }
 271
 272 void NetworkPerformanceServer::SendData( const char* const data, unsigned int bufferSizeInBytes,unsigned int clientId )
 273 {
 274   if( ! mClientCount )
 275   {
 276     return;
 277   }
 278
 279   // prevent clients been added / deleted while transmiting data
 280   Mutex::ScopedLock lock( mClientListMutex );
 281
 282   for( ClientList::Iterator iter = mClients.Begin(); iter != mClients.End() ; ++iter )
 283   {
 284     NetworkPerformanceClient* client = (*iter);
 285     if( client->GetId() == clientId )
 286     {
 287       client->WriteSocket(data ,bufferSizeInBytes);
 288       return;
 289     }
 290   }
 291 }
 292
 293 void NetworkPerformanceServer::TransmitMarker( const PerformanceMarker& marker, const char* const description )
 294 {
 295   if( ! IsRunning() )
 296   {
 297     return;
 298   }
 299   // prevent clients been added / deleted while transmiting data
 300   Mutex::ScopedLock lock( mClientListMutex );
 301
 302   for( ClientList::Iterator iter = mClients.Begin(); iter != mClients.End() ; ++iter )
 303   {
 304     NetworkPerformanceClient* client = (*iter);
 305     client->TransmitMarker( marker, description );
 306   }
 307 }
 308
 309
 310 void NetworkPerformanceServer::StopClients()
 311 {
 312   // prevent clients been added / deleted while stopping all clients
 313   Mutex::ScopedLock lock( mClientListMutex );
 314
 315   for( ClientList::Iterator iter = mClients.Begin(); iter != mClients.End() ; ++iter )
 316   {
 317     NetworkPerformanceClient* client = (*iter);
 318     // stop the client from waiting for new commands, and exit from it's thread
 319     client->ExitSelect();
 320
 321     void* exitValue;
 322     pthread_t* thread = client->GetThread();
 323     pthread_join( *thread, &exitValue );
 324     delete thread;
 325
 326   }
 327 }
 328
 329 } // namespace Internal
 330
 331 } // namespace Adaptor
 332
 333 } // namespace Dali