dali/internal/network/common/network-performance-server.cpp

   1 /*
   2  * Copyright (c) 2022 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 <dali/internal/network/common/network-performance-server.h>
  20
  21 // INTERNAL INCLUDES
  22 #include <dali/internal/system/common/performance-marker.h>
  23
  24 namespace Dali
  25 {
  26 namespace Internal
  27 {
  28 namespace Adaptor
  29 {
  30 namespace // un-named namespace
  31 {
  32 const unsigned int                        SERVER_PORT             = 3031;
  33 const unsigned int                        MAXIMUM_PORTS_TO_TRY    = 10; ///< if port in use, try up to SERVER_PORT + 10
  34 const unsigned int                        CONNECTION_BACKLOG      = 2;  ///<  maximum length of the queue of pending connections.
  35 const unsigned int                        SOCKET_READ_BUFFER_SIZE = 4096;
  36 typedef Vector<NetworkPerformanceClient*> ClientList;
  37
  38 /**
  39  * POD passed to client thread on startup
  40  */
  41 struct ClientThreadInfo
  42 {
  43   NetworkPerformanceServer* server;
  44   NetworkPerformanceClient* client;
  45 };
  46 } // namespace
  47
  48 NetworkPerformanceServer::NetworkPerformanceServer(AdaptorInternalServices&  adaptorServices,
  49                                                    const EnvironmentOptions& logOptions)
  50 : mSocketFactory(adaptorServices.GetSocketFactoryInterface()),
  51   mLogOptions(logOptions),
  52   mServerThread(0),
  53   mListeningSocket(NULL),
  54   mClientUniqueId(0),
  55   mClientCount(0),
  56   mLogFunctionInstalled(false)
  57 {
  58 }
  59
  60 NetworkPerformanceServer::~NetworkPerformanceServer()
  61 {
  62   Stop();
  63
  64   if(mLogFunctionInstalled)
  65   {
  66     mLogOptions.UnInstallLogFunction();
  67   }
  68 }
  69
  70 void NetworkPerformanceServer::Start()
  71 {
  72   // start a thread to listen for incoming connection requests
  73   if(!mServerThread)
  74   {
  75     if(mListeningSocket)
  76     {
  77       mSocketFactory.DestroySocket(mListeningSocket);
  78     }
  79     mListeningSocket = mSocketFactory.NewSocket(SocketInterface::TCP);
  80     mListeningSocket->ReuseAddress(true);
  81
  82     bool         bound    = false;
  83     unsigned int basePort = 0;
  84
  85     // try a small range of ports, so if multiple Dali apps are running you can select
  86     // which one to connect to
  87     while(!bound && (basePort < MAXIMUM_PORTS_TO_TRY))
  88     {
  89       bound = mListeningSocket->Bind(SERVER_PORT + basePort);
  90       if(!bound)
  91       {
  92         basePort++;
  93       }
  94     }
  95     if(!bound)
  96     {
  97       DALI_LOG_ERROR("Failed to bind to a port \n");
  98       return;
  99     }
 100
 101     mListeningSocket->Listen(CONNECTION_BACKLOG);
 102
 103     // start a thread which will block waiting for new connections
 104     int error = pthread_create(&mServerThread, NULL, ConnectionListenerFunc, this);
 105     DALI_ASSERT_ALWAYS(!error && "pthread create failed");
 106
 107     Dali::Integration::Log::LogMessage(Integration::Log::INFO, "~~~ NetworkPerformanceServer started on port %d ~~~ \n", SERVER_PORT + basePort);
 108   }
 109 }
 110 void NetworkPerformanceServer::Stop()
 111 {
 112   if(!mServerThread)
 113   {
 114     return;
 115   }
 116
 117   if(mListeningSocket)
 118   {
 119     // close the server thread to prevent any new connections
 120     mListeningSocket->ExitSelect();
 121   }
 122
 123   // wait for the thread to exit.
 124   void* exitValue;
 125   pthread_join(mServerThread, &exitValue);
 126
 127   if(mListeningSocket)
 128   {
 129     // close the socket
 130     mListeningSocket->CloseSocket();
 131   }
 132
 133   mSocketFactory.DestroySocket(mListeningSocket);
 134
 135   mListeningSocket = NULL;
 136
 137   // this will tell all client threads to quit
 138   StopClients();
 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) && (bytesRead <= SOCKET_READ_BUFFER_SIZE))
 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     else // ret == SocketInterface::QUIT or SocketInterface::ERROR
 218     {
 219       return;
 220     }
 221   }
 222 }
 223
 224 void* NetworkPerformanceServer::ClientThreadFunc(void* data)
 225 {
 226   ClientThreadInfo* info = static_cast<ClientThreadInfo*>(data);
 227   info->server->ClientThread(info->client);
 228   delete info;
 229   return NULL;
 230 }
 231
 232 NetworkPerformanceClient* NetworkPerformanceServer::AddClient(SocketInterface* clientSocket, pthread_t* clientThread)
 233 {
 234   // This function is only called from the listening thread
 235   NetworkPerformanceClient* client = new NetworkPerformanceClient(clientThread,
 236                                                                   clientSocket,
 237                                                                   mClientUniqueId++,
 238                                                                   *this,
 239                                                                   mSocketFactory);
 240
 241   // protect the mClients list which can be accessed from multiple threads.
 242   Mutex::ScopedLock lock(mClientListMutex);
 243
 244   mClients.PushBack(client);
 245
 246   return client;
 247 }
 248
 249 void NetworkPerformanceServer::DeleteClient(NetworkPerformanceClient* client)
 250 {
 251   // protect the mClients list while modifying
 252   Mutex::ScopedLock lock(mClientListMutex);
 253
 254   // remove from the list, and delete it
 255   for(ClientList::Iterator iter = mClients.Begin(); iter != mClients.End(); ++iter)
 256   {
 257     if((*iter) == client)
 258     {
 259       mClients.Erase(iter);
 260       delete client;
 261
 262       // if there server is shutting down, it waits for client count to hit zero
 263       mClientCount--;
 264
 265       return;
 266     }
 267   }
 268 }
 269
 270 void NetworkPerformanceServer::SendData(const char* const data, unsigned int bufferSizeInBytes, unsigned int clientId)
 271 {
 272   if(!mClientCount)
 273   {
 274     return;
 275   }
 276
 277   // prevent clients been added / deleted while transmiting data
 278   Mutex::ScopedLock lock(mClientListMutex);
 279
 280   for(ClientList::Iterator iter = mClients.Begin(); iter != mClients.End(); ++iter)
 281   {
 282     NetworkPerformanceClient* client = (*iter);
 283     if(client->GetId() == clientId)
 284     {
 285       client->WriteSocket(data, bufferSizeInBytes);
 286       return;
 287     }
 288   }
 289 }
 290
 291 void NetworkPerformanceServer::TransmitMarker(const PerformanceMarker& marker, const char* const description)
 292 {
 293   if(!IsRunning())
 294   {
 295     return;
 296   }
 297   // prevent clients been added / deleted while transmiting data
 298   Mutex::ScopedLock lock(mClientListMutex);
 299
 300   for(ClientList::Iterator iter = mClients.Begin(); iter != mClients.End(); ++iter)
 301   {
 302     NetworkPerformanceClient* client = (*iter);
 303     client->TransmitMarker(marker, description);
 304   }
 305 }
 306
 307 void NetworkPerformanceServer::StopClients()
 308 {
 309   // prevent clients been added / deleted while stopping all clients
 310   Mutex::ScopedLock lock(mClientListMutex);
 311
 312   for(ClientList::Iterator iter = mClients.Begin(); iter != mClients.End(); ++iter)
 313   {
 314     NetworkPerformanceClient* client = (*iter);
 315     // stop the client from waiting for new commands, and exit from it's thread
 316     client->ExitSelect();
 317   }
 318 }
 319
 320 } // namespace Adaptor
 321
 322 } // namespace Internal
 323
 324 } // namespace Dali