src/manager/main/socket-manager.cpp

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