c:/Git/PeerCast.root/PeerCast/core/win32/wsocket.cpp

   1 // ------------------------------------------------
   2 // File : wsocket.cpp
   3 // Date: 4-apr-2002
   4 // Author: giles
   5 // Desc:
   6 //              Windows version of ClientSocket. Handles the nitty gritty of actually
   7 //              reading and writing TCP
   8 //
   9 // (c) 2002 peercast.org
  10 // ------------------------------------------------
  11 // This program is free software; you can redistribute it and/or modify
  12 // it under the terms of the GNU General Public License as published by
  13 // the Free Software Foundation; either version 2 of the License, or
  14 // (at your option) any later version.
  15
  16 // This program is distributed in the hope that it will be useful,
  17 // but WITHOUT ANY WARRANTY; without even the implied warranty of
  18 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  19 // GNU General Public License for more details.
  20 // ------------------------------------------------
  21
  22 // TODO: fix socket closing
  23
  24
  25 //#include "stdafx.h"
  26 //#include "winsock2.h"
  27 #include <windows.h>
  28 #include <stdio.h>
  29 #include "wsocket.h"
  30 #include "stats.h"
  31 #ifdef _DEBUG
  32 #include "chkMemoryLeak.h"
  33 #define DEBUG_NEW new(__FILE__, __LINE__)
  34 #define new DEBUG_NEW
  35 #endif
  36
  37
  38 // --------------------------------------------------
  39 void WSAClientSocket::init()
  40 {
  41         WORD wVersionRequested;
  42         WSADATA wsaData;
  43         int err;
  44
  45         wVersionRequested = MAKEWORD( 2, 0 );
  46         err = WSAStartup( wVersionRequested, &wsaData );
  47         if ( err != 0 )
  48                 throw SockException("Unable to init sockets");
  49
  50     //LOG4("WSAStartup:  OK");
  51
  52 }
  53 // --------------------------------------------------
  54 bool ClientSocket::getHostname(char *str,unsigned int ip)
  55 {
  56         HOSTENT *he;
  57
  58         ip = htonl(ip);
  59
  60         he = gethostbyaddr((char *)&ip,sizeof(ip),AF_INET);
  61
  62         if (he)
  63         {
  64                 strcpy(str,he->h_name);
  65                 return true;
  66         }else
  67                 return false;
  68 }
  69
  70 unsigned int cache_ip = 0;
  71 unsigned int cache_time = 0;
  72
  73 // --------------------------------------------------
  74 unsigned int ClientSocket::getIP(char *name)
  75 {
  76         unsigned int ctime = sys->getTime();
  77         bool null_flg = (name == NULL);
  78
  79         if (null_flg){
  80                 if ((cache_time != 0) && (cache_time + 60 > ctime)){
  81                         return cache_ip;
  82                 } else {
  83                         cache_time = 0;
  84                         cache_ip = 0;
  85                 }
  86         }
  87
  88         char szHostName[256];
  89
  90         if (!name)
  91         {
  92                 if (gethostname(szHostName, sizeof(szHostName))==0)
  93                         name = szHostName;
  94                 else
  95                         return 0;
  96         }
  97
  98         HOSTENT *he = WSAClientSocket::resolveHost(name);
  99
 100         if (!he)
 101                 return 0;
 102
 103
 104         LPSTR lpAddr = he->h_addr_list[0];
 105         if (lpAddr)
 106         {
 107                 unsigned int ret;
 108                 struct in_addr  inAddr;
 109                 memmove (&inAddr, lpAddr, 4);
 110
 111                 ret =  inAddr.S_un.S_un_b.s_b1<<24 |
 112                            inAddr.S_un.S_un_b.s_b2<<16 |
 113                            inAddr.S_un.S_un_b.s_b3<<8 |
 114                            inAddr.S_un.S_un_b.s_b4;
 115
 116                 if (null_flg){
 117                         cache_ip = ret;
 118                         cache_time = ctime;
 119                 }
 120                 return ret;
 121         }
 122         return 0;
 123 }
 124 // --------------------------------------------------
 125 void WSAClientSocket::setLinger(int sec)
 126 {
 127         linger linger;
 128         linger.l_onoff = (sec>0)?1:0;
 129     linger.l_linger = sec;
 130
 131         if (setsockopt(sockNum, SOL_SOCKET, SO_LINGER, (const char *)&linger, sizeof (linger)) == -1)
 132                 throw SockException("Unable to set LINGER");
 133 }
 134
 135 // --------------------------------------------------
 136 void WSAClientSocket::setBlocking(bool yes)
 137 {
 138         unsigned long op = yes ? 0 : 1;
 139         if (ioctlsocket(sockNum, FIONBIO, &op) == SOCKET_ERROR)
 140                 throw SockException("Can`t set blocking");
 141 }
 142 // --------------------------------------------------
 143 void WSAClientSocket::setNagle(bool on)
 144 {
 145     int nodelay = (on==false);
 146         if (setsockopt(sockNum, SOL_SOCKET, TCP_NODELAY, (char *)&nodelay, sizeof nodelay) == -1)
 147                 throw SockException("Unable to set NODELAY");
 148
 149 }
 150
 151 // --------------------------------------------------
 152 void WSAClientSocket::setReuse(bool yes)
 153 {
 154         unsigned long op = yes ? 1 : 0;
 155         if (setsockopt(sockNum,SOL_SOCKET,SO_REUSEADDR,(char *)&op,sizeof(unsigned long)) == -1)
 156                 throw SockException("Unable to set REUSE");
 157 }
 158
 159 // --------------------------------------------------
 160 void WSAClientSocket::setBufSize(int size)
 161 {
 162         int oldop;
 163         int op = size;
 164         int len = sizeof(op);
 165         if (getsockopt(sockNum,SOL_SOCKET,SO_RCVBUF,(char *)&oldop,&len) == -1) {
 166                 LOG_DEBUG("Unable to get RCVBUF");
 167         } else if (oldop < size) {
 168                 if (setsockopt(sockNum,SOL_SOCKET,SO_RCVBUF,(char *)&op,len) == -1)
 169                         LOG_DEBUG("Unable to set RCVBUF");
 170                 //else
 171                 //      LOG_DEBUG("*** recvbufsize:%d -> %d", oldop, op);
 172         }
 173
 174         if (getsockopt(sockNum,SOL_SOCKET,SO_SNDBUF,(char *)&oldop,&len) == -1) {
 175                 LOG_DEBUG("Unable to get SNDBUF");
 176         } else if (oldop < size) {
 177                 if (setsockopt(sockNum,SOL_SOCKET,SO_SNDBUF,(char *)&op,len) == -1)
 178                         LOG_DEBUG("Unable to set SNDBUF");
 179                 //else
 180                 //      LOG_DEBUG("*** sendbufsize: %d -> %d", oldop, op);
 181         }
 182 }
 183
 184 // --------------------------------------------------
 185 HOSTENT *WSAClientSocket::resolveHost(const char *hostName)
 186 {
 187         HOSTENT *he;
 188
 189         if ((he = gethostbyname(hostName)) == NULL)
 190         {
 191                 // if failed, try using gethostbyaddr instead
 192
 193                 unsigned long ip = inet_addr(hostName);
 194
 195                 if (ip == INADDR_NONE)
 196                         return NULL;
 197
 198                 if ((he = gethostbyaddr((char *)&ip,sizeof(ip),AF_INET)) == NULL)
 199                         return NULL;
 200         }
 201         return he;
 202 }
 203
 204 // --------------------------------------------------
 205 void WSAClientSocket::open(Host &rh)
 206 {
 207         sockNum = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);
 208
 209         if (sockNum == INVALID_SOCKET)
 210                 throw SockException("Can`t open socket");
 211
 212         setBlocking(false);
 213 #ifdef DISABLE_NAGLE
 214         setNagle(false);
 215 #endif
 216         setBufSize(65535);
 217
 218         host = rh;
 219
 220         memset(&remoteAddr,0,sizeof(remoteAddr));
 221
 222         remoteAddr.sin_family = AF_INET;
 223         remoteAddr.sin_port = htons(host.port);
 224         remoteAddr.sin_addr.S_un.S_addr = htonl(host.ip);
 225
 226 }
 227 // --------------------------------------------------
 228 void WSAClientSocket::checkTimeout(bool r, bool w)
 229 {
 230     int err = WSAGetLastError();
 231     if (err == WSAEWOULDBLOCK)
 232     {
 233
 234                 timeval timeout;
 235                 fd_set read_fds;
 236                 fd_set write_fds;
 237
 238                 timeout.tv_sec = 0;
 239                 timeout.tv_usec = 0;
 240
 241         FD_ZERO (&write_fds);
 242                 if (w)
 243                 {
 244                         timeout.tv_sec = (int)this->writeTimeout/1000;
 245                         FD_SET (sockNum, &write_fds);
 246                 }
 247
 248         FD_ZERO (&read_fds);
 249                 if (r)
 250                 {
 251                         timeout.tv_sec = (int)this->readTimeout/1000;
 252                 FD_SET (sockNum, &read_fds);
 253                 }
 254
 255                 timeval *tp;
 256                 if (timeout.tv_sec)
 257                         tp = &timeout;
 258                 else
 259                         tp = NULL;
 260
 261
 262                 int r=select (NULL, &read_fds, &write_fds, NULL, tp);
 263
 264         if (r == 0)
 265                         throw TimeoutException();
 266                 else if (r == SOCKET_ERROR)
 267                         throw SockException("select failed.");
 268
 269         }else{
 270                 char str[32];
 271                 sprintf(str,"%d",err);
 272                 throw SockException(str);
 273         }
 274 }
 275
 276 // --------------------------------------------------
 277 void WSAClientSocket::checkTimeout2(bool r, bool w)
 278 {
 279     {
 280
 281                 timeval timeout;
 282                 fd_set read_fds;
 283                 fd_set write_fds;
 284
 285                 timeout.tv_sec = 0;
 286                 timeout.tv_usec = 0;
 287
 288         FD_ZERO (&write_fds);
 289                 if (w)
 290                 {
 291                         timeout.tv_sec = (int)this->writeTimeout/1000;
 292                         FD_SET (sockNum, &write_fds);
 293                 }
 294
 295         FD_ZERO (&read_fds);
 296                 if (r)
 297                 {
 298                         timeout.tv_sec = (int)this->readTimeout/1000;
 299                 FD_SET (sockNum, &read_fds);
 300                 }
 301
 302                 timeval *tp;
 303                 if (timeout.tv_sec)
 304                         tp = &timeout;
 305                 else
 306                         tp = NULL;
 307
 308
 309                 int r=select (NULL, &read_fds, &write_fds, NULL, tp);
 310
 311         if (r == 0)
 312                         throw TimeoutException();
 313                 else if (r == SOCKET_ERROR)
 314                         throw SockException("select failed.");
 315         }
 316 }
 317
 318 // --------------------------------------------------
 319 Host WSAClientSocket::getLocalHost()
 320 {
 321         struct sockaddr_in localAddr;
 322
 323         int len = sizeof(localAddr);
 324     if (getsockname(sockNum, (sockaddr *)&localAddr, &len) == 0)
 325                 return Host(SWAP4(localAddr.sin_addr.s_addr),0);
 326         else
 327                 return Host(0,0);
 328 }
 329
 330 // --------------------------------------------------
 331 void WSAClientSocket::connect()
 332 {
 333         if (::connect(sockNum,(struct sockaddr *)&remoteAddr,sizeof(remoteAddr)) == SOCKET_ERROR)
 334                 checkTimeout(false,true);
 335
 336 }
 337 // --------------------------------------------------
 338 int WSAClientSocket::read(void *p, int l)
 339 {
 340         int bytesRead=0;
 341
 342         while (l)
 343         {
 344                 if (rbDataSize >= l) {
 345                         memcpy(p, &apReadBuf[rbPos], l);
 346                         rbPos += l;
 347                         rbDataSize -= l;
 348                         return l;
 349                 } else if (rbDataSize > 0) {
 350                         memcpy(p, &apReadBuf[rbPos], rbDataSize);
 351                         p = (char *) p + rbDataSize;
 352                         l -= rbDataSize;
 353                         bytesRead += rbDataSize;
 354                 }
 355
 356                 rbPos = 0;
 357                 rbDataSize = 0;
 358                 //int r = recv(sockNum, (char *)p, l, 0);
 359                 int r = recv(sockNum, apReadBuf, RBSIZE, 0);
 360                 if (r == SOCKET_ERROR)
 361                 {
 362                         // non-blocking sockets always fall through to here
 363                         checkTimeout(true,false);
 364
 365                 }else if (r == 0)
 366                 {
 367                         throw EOFException("Closed on read");
 368
 369                 }else
 370                 {
 371                         stats.add(Stats::BYTESIN,r);
 372                         if (host.localIP())
 373                                 stats.add(Stats::LOCALBYTESIN,r);
 374                         updateTotals(r,0);
 375                         //bytesRead += r;
 376                         //l -= r;
 377                         //p = (char *)p+r;
 378
 379                         rbDataSize += r;
 380                 }
 381         }
 382         return bytesRead;
 383 }
 384 // --------------------------------------------------
 385 int WSAClientSocket::readUpto(void *p, int l)
 386 {
 387         int bytesRead=0;
 388         while (l)
 389         {
 390                 int r = recv(sockNum, (char *)p, l, 0);
 391                 if (r == SOCKET_ERROR)
 392                 {
 393                         // non-blocking sockets always fall through to here
 394                         //checkTimeout(true,false);
 395                         return r;
 396
 397                 }else if (r == 0)
 398                 {
 399                         break;
 400                 }else
 401                 {
 402                         stats.add(Stats::BYTESIN,r);
 403                         if (host.localIP())
 404                                 stats.add(Stats::LOCALBYTESIN,r);
 405                         updateTotals(r,0);
 406                         bytesRead += r;
 407                         l -= r;
 408                         p = (char *)p+r;
 409                 }
 410                 if (bytesRead) break;
 411         }
 412         return bytesRead;
 413 }
 414
 415
 416 // --------------------------------------------------
 417 void WSAClientSocket::write(const void *p, int l)
 418 {
 419         while (l)
 420         {
 421                 int r = send(sockNum, (char *)p, l, 0);
 422                 if (r == SOCKET_ERROR)
 423                 {
 424                         checkTimeout(false,true);
 425                 }
 426                 else if (r == 0)
 427                 {
 428                         throw SockException("Closed on write");
 429                 }
 430                 else
 431                 if (r > 0)
 432                 {
 433                         stats.add(Stats::BYTESOUT,r);
 434                         if (host.localIP())
 435                                 stats.add(Stats::LOCALBYTESOUT,r);
 436
 437                         updateTotals(0,r);
 438                         l -= r;
 439                         p = (char *)p+r;
 440                 }
 441         }
 442 }
 443
 444 // --------------------------------------------------
 445 void WSAClientSocket::bufferingWrite(const void *p, int l)
 446 {
 447         if (bufList.isNull() && p != NULL){
 448                 while(l){
 449                         int r = send(sockNum, (char *)p, l, 0);
 450                         if (r == SOCKET_ERROR){
 451                                 int err = WSAGetLastError();
 452                                 if (err == WSAEWOULDBLOCK){
 453                                         bufList.add(p, l);
 454 //                                      LOG_DEBUG("normal add");
 455                                         break;
 456                                 } else {
 457                                         char str[32];
 458                                         sprintf(str,"%d",err);
 459                                         throw SockException(str);
 460                                 }
 461                         } else if (r == 0) {
 462                                 throw SockException("Closed on write");
 463                         } else if (r > 0){
 464                                 stats.add(Stats::BYTESOUT,r);
 465                                 if (host.localIP())
 466                                         stats.add(Stats::LOCALBYTESOUT,r);
 467
 468                                 updateTotals(0,r);
 469                                 l -= r;
 470                                 p = (char *)p+r;
 471                         }
 472                 }
 473         } else {
 474 //              LOG_DEBUG("***************BufferingWrite");
 475                 if (p)
 476                         bufList.add(p,l);
 477
 478                 bool flg = true;
 479
 480                 while(flg){
 481                         SocketBuffer *tmp;
 482                         tmp = bufList.getTop();
 483
 484                         if (tmp){
 485 //                              LOG_DEBUG("tmp->pos = %d, tmp->len = %d, %d", tmp->pos, tmp->len, tmp);
 486                                 while(tmp->pos < tmp->len){
 487                                         int r = send(sockNum, (char*)(tmp->buf + tmp->pos), tmp->len - tmp->pos, 0);
 488 //                                      LOG_DEBUG("send = %d", r);
 489                                         if (r == SOCKET_ERROR){
 490                                                 int err = WSAGetLastError();
 491                                                 if (err == WSAEWOULDBLOCK){
 492                                                         flg = false;
 493                                                         break;
 494                                                 } else {
 495                                                         bufList.clear();
 496                                                         char str[32];
 497                                                         sprintf(str,"%d",err);
 498                                                         throw SockException(str);
 499                                                 }
 500                                         } else if (r == 0){
 501                                                 bufList.clear();
 502                                                 throw SockException("Closed on write");
 503                                         } else if (r > 0){
 504                                                 stats.add(Stats::BYTESOUT,r);
 505                                                 if (host.localIP())
 506                                                         stats.add(Stats::LOCALBYTESOUT,r);
 507
 508                                                 updateTotals(0,r);
 509
 510                                                 tmp->pos += r;
 511                                                 if (tmp->pos >= tmp->len){
 512 //                                                      LOG_DEBUG("deleteTop");
 513                                                         bufList.deleteTop();
 514                                                         break;
 515                                                 }
 516                                         }
 517                                 }
 518                         } else {
 519                                 flg = false;
 520                         }
 521                 }
 522 //              LOG_DEBUG("bufferingWrite end");
 523         }
 524 }
 525
 526 // --------------------------------------------------
 527 void WSAClientSocket::checkBuffering(bool r, bool w)
 528 {
 529     int err = WSAGetLastError();
 530     if (err == WSAEWOULDBLOCK)
 531     {
 532
 533                 timeval timeout;
 534                 fd_set read_fds;
 535                 fd_set write_fds;
 536
 537                 timeout.tv_sec = 0;
 538                 timeout.tv_usec = 0;
 539
 540         FD_ZERO (&write_fds);
 541                 if (w)
 542                 {
 543                         timeout.tv_sec = (int)this->writeTimeout/1000;
 544                         FD_SET (sockNum, &write_fds);
 545                 }
 546
 547         FD_ZERO (&read_fds);
 548                 if (r)
 549                 {
 550                         timeout.tv_sec = (int)this->readTimeout/1000;
 551                 FD_SET (sockNum, &read_fds);
 552                 }
 553
 554                 timeval *tp;
 555                 if (timeout.tv_sec)
 556                         tp = &timeout;
 557                 else
 558                         tp = NULL;
 559
 560
 561                 int r=select (NULL, &read_fds, &write_fds, NULL, tp);
 562
 563         if (r == 0)
 564                         throw TimeoutException();
 565                 else if (r == SOCKET_ERROR)
 566                         throw SockException("select failed.");
 567
 568         }else{
 569                 char str[32];
 570                 sprintf(str,"%d",err);
 571                 throw SockException(str);
 572         }
 573 }
 574
 575 // --------------------------------------------------
 576 void WSAClientSocket::bind(Host &h)
 577 {
 578         struct sockaddr_in localAddr;
 579
 580         if ((sockNum = socket (PF_INET, SOCK_STREAM, IPPROTO_TCP)) == -1)
 581                 throw SockException("Can`t open socket");
 582
 583         setBlocking(false);
 584         setReuse(true);
 585
 586         memset(&localAddr,0,sizeof(localAddr));
 587         localAddr.sin_family = AF_INET;
 588         localAddr.sin_port = htons(h.port);
 589         localAddr.sin_addr.s_addr = INADDR_ANY;
 590
 591         if( ::bind (sockNum, (sockaddr *)&localAddr, sizeof(localAddr)) == -1)
 592                 throw SockException("Can`t bind socket");
 593
 594         if (::listen(sockNum,SOMAXCONN))
 595                 throw SockException("Can`t listen",WSAGetLastError());
 596
 597         host = h;
 598 }
 599 // --------------------------------------------------
 600 ClientSocket *WSAClientSocket::accept()
 601 {
 602
 603         int fromSize = sizeof(sockaddr_in);
 604         sockaddr_in from;
 605
 606         int conSock = ::accept(sockNum,(sockaddr *)&from,&fromSize);
 607
 608
 609         if (conSock ==  INVALID_SOCKET)
 610                 return NULL;
 611
 612
 613     WSAClientSocket *cs = new WSAClientSocket();
 614         cs->sockNum = conSock;
 615
 616         cs->host.port = (from.sin_port & 0xff) << 8 | ((from.sin_port >> 8) & 0xff);
 617         cs->host.ip = from.sin_addr.S_un.S_un_b.s_b1<<24 |
 618                                   from.sin_addr.S_un.S_un_b.s_b2<<16 |
 619                                   from.sin_addr.S_un.S_un_b.s_b3<<8 |
 620                                   from.sin_addr.S_un.S_un_b.s_b4;
 621
 622
 623         cs->setBlocking(false);
 624 #ifdef DISABLE_NAGLE
 625         cs->setNagle(false);
 626 #endif
 627         cs->setBufSize(65535);
 628
 629         return cs;
 630 }
 631
 632 // --------------------------------------------------
 633 void WSAClientSocket::close()
 634 {
 635         sockLock.on();
 636         if (sockNum)
 637         {
 638 //              shutdown(sockNum,SD_SEND);
 639
 640                 setReadTimeout(2000);
 641                 unsigned int stime = sys->getTime();
 642                 try
 643                 {
 644                         char c[1024];
 645                         while (read(&c, sizeof(c)) > 0)
 646                                 if (sys->getTime() - stime > 5)
 647                                         break;
 648                 }catch(StreamException &) {}
 649
 650                 if (closesocket (sockNum))
 651                         LOG_ERROR("closesocket() error");
 652
 653
 654                 sockNum=0;
 655         }
 656         sockLock.off();
 657 }
 658
 659 // --------------------------------------------------
 660 bool    WSAClientSocket::readReady()
 661 {
 662         if (rbDataSize) return true;
 663
 664         timeval timeout;
 665         fd_set read_fds;
 666
 667         timeout.tv_sec = 0;
 668         timeout.tv_usec = 0;
 669
 670     FD_ZERO (&read_fds);
 671     FD_SET (sockNum, &read_fds);
 672
 673         return select (sockNum+1, &read_fds, NULL, NULL, &timeout) == 1;
 674 }
 675