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

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