src/server/main/socket-manager.cpp

   1 /*
   2  *  Copyright (c) 2000 - 2014 Samsung Electronics Co., Ltd All Rights Reserved
   3  *
   4  *  Contact: Rafal Krypa <r.krypa@samsung.com>
   5  *
   6  *  Licensed under the Apache License, Version 2.0 (the "License");
   7  *  you may not use this file except in compliance with the License.
   8  *  You may obtain a copy of the License at
   9  *
  10  *      http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  *  Unless required by applicable law or agreed to in writing, software
  13  *  distributed under the License is distributed on an "AS IS" BASIS,
  14  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  *  See the License for the specific language governing permissions and
  16  *  limitations under the License
  17  */
  18 /*
  19  * @file        socket-manager.cpp
  20  * @author      Bartlomiej Grzelewski (b.grzelewski@samsung.com)
  21  * @version     1.0
  22  * @brief       Implementation of SocketManager.
  23  */
  24
  25 #include <set>
  26
  27 #include <signal.h>
  28 #include <sys/select.h>
  29 #include <sys/signalfd.h>
  30 #include <sys/types.h>
  31 #include <sys/socket.h>
  32 #include <sys/smack.h>
  33 #include <sys/un.h>
  34 #include <sys/stat.h>
  35 #include <unistd.h>
  36 #include <fcntl.h>
  37 #include <signal.h>
  38 #include <errno.h>
  39 #include <time.h>
  40
  41 #include <systemd/sd-daemon.h>
  42
  43 #include <dpl/log/log.h>
  44 #include <dpl/assert.h>
  45
  46 #include <smack-check.h>
  47 #include <socket-manager.h>
  48
  49 namespace {
  50
  51 const time_t SOCKET_TIMEOUT = 20;
  52
  53 } // namespace anonymous
  54
  55 namespace SecurityManager {
  56
  57 struct DummyService : public GenericSocketService {
  58     ServiceDescriptionVector GetServiceDescription() {
  59         return ServiceDescriptionVector();
  60     }
  61     void Event(const AcceptEvent &event) { (void)event; }
  62     void Event(const WriteEvent &event) { (void)event; }
  63     void Event(const ReadEvent &event) { (void)event; }
  64     void Event(const CloseEvent &event) { (void)event; }
  65 };
  66
  67 struct SignalService : public GenericSocketService {
  68     int GetDescriptor() {
  69         LogInfo("set up");
  70         sigset_t mask;
  71         sigemptyset(&mask);
  72         sigaddset(&mask, SIGTERM);
  73         if (-1 == pthread_sigmask(SIG_BLOCK, &mask, NULL))
  74             return -1;
  75         return signalfd(-1, &mask, 0);
  76     }
  77
  78     ServiceDescriptionVector GetServiceDescription() {
  79         return ServiceDescriptionVector();
  80     }
  81
  82     void Event(const AcceptEvent &event) { (void)event; } // not supported
  83     void Event(const WriteEvent &event) { (void)event; }  // not supported
  84     void Event(const CloseEvent &event) { (void)event; }  // not supported
  85
  86     void Event(const ReadEvent &event) {
  87         LogDebug("Get signal information");
  88
  89         if(sizeof(struct signalfd_siginfo) != event.rawBuffer.size()) {
  90             LogError("Wrong size of signalfd_siginfo struct. Expected: "
  91                 << sizeof(signalfd_siginfo) << " Get: "
  92                 << event.rawBuffer.size());
  93             return;
  94         }
  95
  96         signalfd_siginfo *siginfo = (signalfd_siginfo*)(&(event.rawBuffer[0]));
  97
  98         if (siginfo->ssi_signo == SIGTERM) {
  99             LogInfo("Got signal: SIGTERM");
 100             static_cast<SocketManager*>(m_serviceManager)->MainLoopStop();
 101             return;
 102         }
 103
 104         LogInfo("This should not happend. Got signal: " << siginfo->ssi_signo);
 105     }
 106 };
 107
 108 SocketManager::SocketDescription&
 109 SocketManager::CreateDefaultReadSocketDescription(int sock, bool timeout)
 110 {
 111     if ((int)m_socketDescriptionVector.size() <= sock)
 112         m_socketDescriptionVector.resize(sock+20);
 113
 114     auto &desc = m_socketDescriptionVector[sock];
 115     desc.isListen = false;
 116     desc.isOpen = true;
 117     desc.interfaceID = 0;
 118     desc.service = NULL;
 119     desc.counter = ++m_counter;
 120
 121     if (timeout) {
 122         desc.timeout = time(NULL) + SOCKET_TIMEOUT;
 123         if (false == desc.isTimeout) {
 124             Timeout tm;
 125             tm.time = desc.timeout;
 126             tm.sock = sock;
 127             m_timeoutQueue.push(tm);
 128         }
 129     }
 130
 131     desc.isTimeout = timeout;
 132
 133     FD_SET(sock, &m_readSet);
 134     m_maxDesc = sock > m_maxDesc ? sock : m_maxDesc;
 135     return desc;
 136 }
 137
 138 SocketManager::SocketManager()
 139   : m_maxDesc(0)
 140   , m_counter(0)
 141 {
 142     FD_ZERO(&m_readSet);
 143     FD_ZERO(&m_writeSet);
 144     if (-1 == pipe(m_notifyMe)) {
 145         int err = errno;
 146         ThrowMsg(Exception::InitFailed, "Error in pipe: " << strerror(err));
 147     }
 148     LogInfo("Pipe: Read desc: " << m_notifyMe[0] << " Write desc: " << m_notifyMe[1]);
 149
 150     auto &desc = CreateDefaultReadSocketDescription(m_notifyMe[0], false);
 151     desc.service = new DummyService;
 152
 153     // std::thread bases on pthread so this should work fine
 154     sigset_t set;
 155     sigemptyset(&set);
 156     sigaddset(&set, SIGPIPE);
 157     pthread_sigmask(SIG_BLOCK, &set, NULL);
 158
 159     // add support for TERM signal (passed from systemd)
 160     auto *signalService = new SignalService;
 161     signalService->SetSocketManager(this);
 162     int filefd = signalService->GetDescriptor();
 163     if (-1 == filefd) {
 164         LogError("Error in SignalService.GetDescriptor()");
 165         delete signalService;
 166     } else {
 167         auto &desc2 = CreateDefaultReadSocketDescription(filefd, false);
 168         desc2.service = signalService;
 169         LogInfo("SignalService mounted on " << filefd << " descriptor");
 170     }
 171 }
 172
 173 SocketManager::~SocketManager() {
 174     std::set<GenericSocketService*> serviceMap;
 175
 176     // Find all services. Set is used to remove duplicates.
 177     // In this implementation, services are not able to react in any way.
 178     for (size_t i=0; i < m_socketDescriptionVector.size(); ++i)
 179         if (m_socketDescriptionVector[i].isOpen)
 180             serviceMap.insert(m_socketDescriptionVector[i].service);
 181
 182     // Time to destroy all services.
 183     for(auto it = serviceMap.begin(); it != serviceMap.end(); ++it) {
 184         LogDebug("delete " << (void*)(*it));
 185         delete *it;
 186     }
 187
 188     for (size_t i = 0; i < m_socketDescriptionVector.size(); ++i)
 189         if (m_socketDescriptionVector[i].isOpen)
 190             close(i);
 191
 192     // All socket except one were closed. Now pipe input must be closed.
 193     close(m_notifyMe[1]);
 194 }
 195
 196 void SocketManager::ReadyForAccept(int sock) {
 197     struct sockaddr_un clientAddr;
 198     unsigned int clientLen = sizeof(clientAddr);
 199     int client = accept4(sock, (struct sockaddr*) &clientAddr, &clientLen, SOCK_NONBLOCK);
 200 //    LogInfo("Accept on sock: " << sock << " Socket opended: " << client);
 201     if (-1 == client) {
 202         int err = errno;
 203         LogError("Error in accept: " << strerror(err));
 204         return;
 205     }
 206
 207     auto &desc = CreateDefaultReadSocketDescription(client, true);
 208     desc.interfaceID = m_socketDescriptionVector[sock].interfaceID;
 209     desc.service = m_socketDescriptionVector[sock].service;
 210     desc.useSendMsg = m_socketDescriptionVector[sock].useSendMsg;
 211
 212     GenericSocketService::AcceptEvent event;
 213     event.connectionID.sock = client;
 214     event.connectionID.counter = desc.counter;
 215     event.interfaceID = desc.interfaceID;
 216     desc.service->Event(event);
 217 }
 218
 219 void SocketManager::ReadyForRead(int sock) {
 220     if (m_socketDescriptionVector[sock].isListen) {
 221         ReadyForAccept(sock);
 222         return;
 223     }
 224
 225     GenericSocketService::ReadEvent event;
 226     event.connectionID.sock = sock;
 227     event.connectionID.counter = m_socketDescriptionVector[sock].counter;
 228     event.rawBuffer.resize(4096);
 229
 230     auto &desc = m_socketDescriptionVector[sock];
 231     desc.timeout = time(NULL) + SOCKET_TIMEOUT;
 232
 233     ssize_t size = read(sock, &event.rawBuffer[0], 4096);
 234
 235     if (size == 0) {
 236         CloseSocket(sock);
 237     } else if (size >= 0) {
 238         event.rawBuffer.resize(size);
 239         desc.service->Event(event);
 240     } else if (size == -1) {
 241         int err = errno;
 242         switch(err) {
 243             case EAGAIN:
 244             case EINTR:
 245                 break;
 246             default:
 247                 LogError("Reading sock error: " << strerror(err));
 248                 CloseSocket(sock);
 249         }
 250     }
 251 }
 252
 253 void SocketManager::ReadyForSendMsg(int sock) {
 254     auto &desc = m_socketDescriptionVector[sock];
 255
 256     if (desc.sendMsgDataQueue.empty()) {
 257          FD_CLR(sock, &m_writeSet);
 258          return;
 259     }
 260
 261     auto data = desc.sendMsgDataQueue.front();
 262     ssize_t result = sendmsg(sock, data.getMsghdr(), data.flags());
 263
 264     if (result == -1) {
 265         int err = errno;
 266         switch(err) {
 267         case EAGAIN:
 268         case EINTR:
 269             break;
 270         case EPIPE:
 271         default:
 272             LogError("Error during send: " << strerror(err));
 273             CloseSocket(sock);
 274             break;
 275         }
 276         return;
 277     } else {
 278         desc.sendMsgDataQueue.pop();
 279     }
 280
 281     if (desc.sendMsgDataQueue.empty()) {
 282         FD_CLR(sock, &m_writeSet);
 283     }
 284
 285     desc.timeout = time(NULL) + SOCKET_TIMEOUT;
 286
 287     GenericSocketService::WriteEvent event;
 288     event.connectionID.sock = sock;
 289     event.connectionID.counter = desc.counter;
 290     event.size = result;
 291     event.left = desc.sendMsgDataQueue.size();
 292
 293     desc.service->Event(event);
 294 }
 295
 296 void SocketManager::ReadyForWriteBuffer(int sock) {
 297     auto &desc = m_socketDescriptionVector[sock];
 298     size_t size = desc.rawBuffer.size();
 299     ssize_t result = write(sock, &desc.rawBuffer[0], size);
 300     if (result == -1) {
 301         int err = errno;
 302         switch(err) {
 303         case EAGAIN:
 304         case EINTR:
 305             // select will trigger write once again, nothing to do
 306             break;
 307         case EPIPE:
 308         default:
 309             LogError("Error during write: " << strerror(err));
 310             CloseSocket(sock);
 311             break;
 312         }
 313         return; // We do not want to propagate error to next layer
 314     }
 315
 316     desc.rawBuffer.erase(desc.rawBuffer.begin(), desc.rawBuffer.begin()+result);
 317
 318     desc.timeout = time(NULL) + SOCKET_TIMEOUT;
 319
 320     if (desc.rawBuffer.empty())
 321         FD_CLR(sock, &m_writeSet);
 322
 323     GenericSocketService::WriteEvent event;
 324     event.connectionID.sock = sock;
 325     event.connectionID.counter = desc.counter;
 326     event.size = result;
 327     event.left = desc.rawBuffer.size();
 328
 329     desc.service->Event(event);
 330 }
 331
 332 void SocketManager::ReadyForWrite(int sock) {
 333     m_socketDescriptionVector[sock].useSendMsg ?
 334         ReadyForSendMsg(sock) : ReadyForWriteBuffer(sock);
 335 }
 336
 337 void SocketManager::MainLoop() {
 338     // remove evironment values passed by systemd
 339     sd_listen_fds(1);
 340
 341     // Daemon is ready to work.
 342     sd_notify(0, "READY=1");
 343
 344     m_working = true;
 345     while(m_working) {
 346         fd_set readSet = m_readSet;
 347         fd_set writeSet = m_writeSet;
 348
 349         timeval localTempTimeout;
 350         timeval *ptrTimeout = &localTempTimeout;
 351
 352         // I need to extract timeout from priority_queue.
 353         // Timeout in priority_queue may be deprecated.
 354         // I need to find some actual one.
 355         while(!m_timeoutQueue.empty()) {
 356             auto &top = m_timeoutQueue.top();
 357             auto &desc = m_socketDescriptionVector[top.sock];
 358
 359             if (top.time == desc.timeout) {
 360                 // This timeout matches timeout from socket.
 361                 // It can be used.
 362                 break;
 363             } else {
 364                 // This socket was used after timeout in priority queue was set up.
 365                 // We need to update timeout and find some useable one.
 366                 Timeout tm = { desc.timeout , top.sock};
 367                 m_timeoutQueue.pop();
 368                 m_timeoutQueue.push(tm);
 369             }
 370         }
 371
 372         if (m_timeoutQueue.empty()) {
 373             LogDebug("No usaable timeout found.");
 374             ptrTimeout = NULL; // select will wait without timeout
 375         } else {
 376             time_t currentTime = time(NULL);
 377             auto &pqTimeout = m_timeoutQueue.top();
 378
 379             // 0 means that select won't block and socket will be closed ;-)
 380             ptrTimeout->tv_sec =
 381               currentTime < pqTimeout.time ? pqTimeout.time - currentTime : 0;
 382             ptrTimeout->tv_usec = 0;
 383 //            LogDebug("Set up timeout: " << (int)ptrTimeout->tv_sec
 384 //                << " seconds. Socket: " << pqTimeout.sock);
 385         }
 386
 387         int ret = select(m_maxDesc+1, &readSet, &writeSet, NULL, ptrTimeout);
 388
 389         if (0 == ret) { // timeout
 390             Assert(!m_timeoutQueue.empty());
 391
 392             Timeout pqTimeout = m_timeoutQueue.top();
 393             m_timeoutQueue.pop();
 394
 395             auto &desc = m_socketDescriptionVector[pqTimeout.sock];
 396
 397             if (!desc.isTimeout || !desc.isOpen) {
 398                 // Connection was closed. Timeout is useless...
 399                 desc.isTimeout = false;
 400                 continue;
 401             }
 402
 403             if (pqTimeout.time < desc.timeout) {
 404                 // Is it possible?
 405                 // This socket was used after timeout. We need to update timeout.
 406                 pqTimeout.time = desc.timeout;
 407                 m_timeoutQueue.push(pqTimeout);
 408                 continue;
 409             }
 410
 411             // timeout from m_timeoutQueue matches with socket.timeout
 412             // and connection is open. Time to close it!
 413             // Putting new timeout in queue here is pointless.
 414             desc.isTimeout = false;
 415             CloseSocket(pqTimeout.sock);
 416
 417             // All done. Now we should process next select ;-)
 418             continue;
 419         }
 420
 421         if (-1 == ret) {
 422             switch(errno) {
 423             case EINTR:
 424                 LogDebug("EINTR in select");
 425                 break;
 426             default:
 427                 int err = errno;
 428                 LogError("Error in select: " << strerror(err));
 429                 return;
 430             }
 431             continue;
 432         }
 433         for(int i = 0; i<m_maxDesc+1 && ret; ++i) {
 434             if (FD_ISSET(i, &readSet)) {
 435                 ReadyForRead(i);
 436                 --ret;
 437             }
 438             if (FD_ISSET(i, &writeSet)) {
 439                 ReadyForWrite(i);
 440                 --ret;
 441             }
 442         }
 443         ProcessQueue();
 444     }
 445 }
 446
 447 void SocketManager::MainLoopStop()
 448 {
 449     m_working = false;
 450     NotifyMe();
 451 }
 452
 453 int SocketManager::GetSocketFromSystemD(
 454     const GenericSocketService::ServiceDescription &desc)
 455 {
 456     int fd;
 457
 458     // TODO optimalization - do it once in object constructor
 459     //                       and remember all information path->sockfd
 460     int n = sd_listen_fds(0);
 461
 462     LogInfo("sd_listen_fds returns: " << n);
 463
 464     if (n < 0) {
 465         LogError("Error in sd_listend_fds");
 466         ThrowMsg(Exception::InitFailed, "Error in sd_listend_fds");
 467     }
 468
 469     for(fd = SD_LISTEN_FDS_START; fd < SD_LISTEN_FDS_START+n; ++fd) {
 470         if (0 < sd_is_socket_unix(fd, SOCK_STREAM, 1,
 471                                   desc.serviceHandlerPath.c_str(), 0))
 472         {
 473             LogInfo("Useable socket " << desc.serviceHandlerPath <<
 474                 " was passed by SystemD under descriptor " << fd);
 475             return fd;
 476         }
 477     }
 478     LogInfo("No useable sockets were passed by systemd.");
 479     return -1;
 480 }
 481
 482 int SocketManager::CreateDomainSocketHelp(
 483     const GenericSocketService::ServiceDescription &desc)
 484 {
 485     int sockfd;
 486
 487     if (-1 == (sockfd = socket(AF_UNIX, SOCK_STREAM, 0))) {
 488         int err = errno;
 489         LogError("Error in socket: " << strerror(err));
 490         ThrowMsg(Exception::InitFailed, "Error in socket: " << strerror(err));
 491     }
 492
 493     if (smack_check()) {
 494         LogInfo("Set up smack label: " << desc.smackLabel);
 495
 496         if (0 != smack_fsetlabel(sockfd, desc.smackLabel.c_str(), SMACK_LABEL_IPIN)) {
 497             LogError("Error in smack_fsetlabel");
 498             ThrowMsg(Exception::InitFailed, "Error in smack_fsetlabel");
 499         }
 500     } else {
 501         LogInfo("No smack on platform. Socket won't be securied with smack label!");
 502     }
 503
 504     int flags;
 505     if (-1 == (flags = fcntl(sockfd, F_GETFL, 0)))
 506         flags = 0;
 507
 508     if (-1 == fcntl(sockfd, F_SETFL, flags | O_NONBLOCK)) {
 509         int err = errno;
 510         close(sockfd);
 511         LogError("Error in fcntl: " << strerror(err));
 512         ThrowMsg(Exception::InitFailed, "Error in fcntl: " << strerror(err));
 513     }
 514
 515     sockaddr_un serverAddress;
 516     memset(&serverAddress, 0, sizeof(serverAddress));
 517     serverAddress.sun_family = AF_UNIX;
 518     strcpy(serverAddress.sun_path, desc.serviceHandlerPath.c_str());
 519     unlink(serverAddress.sun_path);
 520
 521     mode_t originalUmask;
 522     originalUmask = umask(0);
 523
 524     if (-1 == bind(sockfd, (struct sockaddr*)&serverAddress, sizeof(serverAddress))) {
 525         int err = errno;
 526         close(sockfd);
 527         LogError("Error in bind: " << strerror(err));
 528         ThrowMsg(Exception::InitFailed, "Error in bind: " << strerror(err));
 529     }
 530
 531     umask(originalUmask);
 532
 533     if (-1 == listen(sockfd, 5)) {
 534         int err = errno;
 535         close(sockfd);
 536         LogError("Error in listen: " << strerror(err));
 537         ThrowMsg(Exception::InitFailed, "Error in listen: " << strerror(err));
 538     }
 539
 540     return sockfd;
 541 }
 542
 543 void SocketManager::CreateDomainSocket(
 544     GenericSocketService *service,
 545     const GenericSocketService::ServiceDescription &desc)
 546 {
 547     int sockfd = GetSocketFromSystemD(desc);
 548     if (-1 == sockfd)
 549         sockfd = CreateDomainSocketHelp(desc);
 550
 551     auto &description = CreateDefaultReadSocketDescription(sockfd, false);
 552
 553     description.isListen = true;
 554     description.interfaceID = desc.interfaceID;
 555     description.useSendMsg = desc.useSendMsg;
 556     description.service = service;
 557
 558     LogDebug("Listen on socket: " << sockfd <<
 559         " Handler: " << desc.serviceHandlerPath.c_str());
 560 }
 561
 562 void SocketManager::RegisterSocketService(GenericSocketService *service) {
 563     service->SetSocketManager(this);
 564     auto serviceVector = service->GetServiceDescription();
 565     Try {
 566         for (auto iter = serviceVector.begin(); iter != serviceVector.end(); ++iter)
 567             CreateDomainSocket(service, *iter);
 568     } Catch (Exception::Base) {
 569         for (int i =0; i < (int)m_socketDescriptionVector.size(); ++i)
 570         {
 571             auto &desc = m_socketDescriptionVector[i];
 572             if (desc.service == service && desc.isOpen) {
 573                 close(i);
 574                 desc.isOpen = false;
 575             }
 576         }
 577         ReThrow(Exception::Base);
 578     }
 579 }
 580
 581 void SocketManager::Close(ConnectionID connectionID) {
 582     {
 583         std::lock_guard<std::mutex> ulock(m_eventQueueMutex);
 584         m_closeQueue.push(connectionID);
 585     }
 586     NotifyMe();
 587 }
 588
 589 void SocketManager::Write(ConnectionID connectionID, const RawBuffer &rawBuffer) {
 590     WriteBuffer buffer;
 591     buffer.connectionID = connectionID;
 592     buffer.rawBuffer = rawBuffer;
 593     {
 594         std::lock_guard<std::mutex> ulock(m_eventQueueMutex);
 595         m_writeBufferQueue.push(buffer);
 596     }
 597     NotifyMe();
 598 }
 599
 600 void SocketManager::Write(ConnectionID connectionID, const SendMsgData &sendMsgData) {
 601     WriteData data;
 602     data.connectionID = connectionID;
 603     data.sendMsgData = sendMsgData;
 604     {
 605         std::lock_guard<std::mutex> ulock(m_eventQueueMutex);
 606         m_writeDataQueue.push(data);
 607     }
 608     NotifyMe();
 609 }
 610
 611 void SocketManager::NotifyMe() {
 612     TEMP_FAILURE_RETRY(write(m_notifyMe[1], "You have message ;-)", 1));
 613 }
 614
 615 void SocketManager::ProcessQueue() {
 616     WriteBuffer buffer;
 617     WriteData data;
 618     {
 619         std::lock_guard<std::mutex> ulock(m_eventQueueMutex);
 620         while (!m_writeBufferQueue.empty()) {
 621             buffer = m_writeBufferQueue.front();
 622             m_writeBufferQueue.pop();
 623
 624             auto &desc = m_socketDescriptionVector[buffer.connectionID.sock];
 625
 626             if (!desc.isOpen) {
 627                 LogDebug("Received packet for write but connection is closed. Packet ignored!");
 628                 continue;
 629             }
 630
 631             if (desc.counter != buffer.connectionID.counter)
 632             {
 633                 LogDebug("Received packet for write but counter is broken. Packet ignored!");
 634                 continue;
 635             }
 636
 637             if (desc.useSendMsg) {
 638                 LogError("Some service tried to push rawdata to socket that usees sendmsg!");
 639                 continue;
 640             }
 641
 642             std::copy(
 643                 buffer.rawBuffer.begin(),
 644                 buffer.rawBuffer.end(),
 645                 std::back_inserter(desc.rawBuffer));
 646
 647             FD_SET(buffer.connectionID.sock, &m_writeSet);
 648         }
 649
 650         while(!m_writeDataQueue.empty()) {
 651             data = m_writeDataQueue.front();
 652             m_writeDataQueue.pop();
 653
 654             auto &desc = m_socketDescriptionVector[data.connectionID.sock];
 655
 656             if (!desc.isOpen) {
 657                 LogDebug("Received packet for sendmsg but connection is closed. Packet ignored!");
 658                 continue;
 659             }
 660
 661             if (desc.counter != data.connectionID.counter)
 662             {
 663                 LogDebug("Received packet for write but counter is broken. Packet ignored!");
 664                 continue;
 665             }
 666
 667             if (!desc.useSendMsg) {
 668                 LogError("Some service tries to push SendMsgData to socket that uses write!");
 669                 continue;
 670             }
 671
 672             desc.sendMsgDataQueue.push(data.sendMsgData);
 673
 674             FD_SET(data.connectionID.sock, &m_writeSet);
 675         }
 676     }
 677
 678     while (1) {
 679         ConnectionID connection;
 680         {
 681             std::lock_guard<std::mutex> ulock(m_eventQueueMutex);
 682             if (m_closeQueue.empty())
 683                 return;
 684             connection = m_closeQueue.front();
 685             m_closeQueue.pop();
 686         }
 687
 688         if (!m_socketDescriptionVector[connection.sock].isOpen)
 689             continue;
 690
 691         if (connection.counter != m_socketDescriptionVector[connection.sock].counter)
 692             continue;
 693
 694         CloseSocket(connection.sock);
 695     }
 696 }
 697
 698 void SocketManager::CloseSocket(int sock) {
 699 //    LogInfo("Closing socket: " << sock);
 700     auto &desc = m_socketDescriptionVector[sock];
 701
 702     if (!(desc.isOpen)) {
 703         // This may happend when some information was waiting for write to the
 704         // socket and in the same time socket was closed by the client.
 705         LogError("Socket " << sock << " is not open. Nothing to do!");
 706         return;
 707     }
 708
 709     GenericSocketService::CloseEvent event;
 710     event.connectionID.sock = sock;
 711     event.connectionID.counter = desc.counter;
 712     auto service = desc.service;
 713
 714     desc.isOpen = false;
 715     desc.service = NULL;
 716     desc.interfaceID = -1;
 717     desc.rawBuffer.clear();
 718     while(!desc.sendMsgDataQueue.empty())
 719         desc.sendMsgDataQueue.pop();
 720
 721     if (service)
 722         service->Event(event);
 723     else
 724         LogError("Critical! Service is NULL! This should never happend!");
 725
 726     TEMP_FAILURE_RETRY(close(sock));
 727     FD_CLR(sock, &m_readSet);
 728     FD_CLR(sock, &m_writeSet);
 729 }
 730
 731 } // namespace SecurityManager