From: henrique Date: Tue, 24 Jun 2014 14:08:22 +0000 (+0200) Subject: THRIFT-2577 C++ TFileTransport missuse of closesocket on windows platform X-Git-Url: https://source.supwisdom.com/gerrit/gitweb?a=commitdiff_plain;h=58993d120578ce36fb88f7af06c06d2eba213b65;p=common%2Fthrift.git THRIFT-2577 C++ TFileTransport missuse of closesocket on windows platform Patch: suuyaoo This closes #142 --- --- diff --git a/lib/cpp/src/thrift/transport/PlatformSocket.h b/lib/cpp/src/thrift/transport/PlatformSocket.h index 58d68a47..019f4cb4 100644 --- a/lib/cpp/src/thrift/transport/PlatformSocket.h +++ b/lib/cpp/src/thrift/transport/PlatformSocket.h @@ -22,6 +22,7 @@ #ifdef _WIN32 # define THRIFT_GET_SOCKET_ERROR ::WSAGetLastError() +# define THRIFT_ERRNO (*_errno()) # define THRIFT_EINPROGRESS WSAEINPROGRESS # define THRIFT_EAGAIN WSAEWOULDBLOCK # define THRIFT_EINTR WSAEINTR @@ -40,6 +41,15 @@ # define THRIFT_F_SETFL 1 # define THRIFT_GETTIMEOFDAY thrift_gettimeofday # define THRIFT_CLOSESOCKET closesocket +# define THRIFT_CLOSE _close +# define THRIFT_OPEN _open +# define THRIFT_FTRUNCATE _chsize_s +# define THRIFT_FSYNC _commit +# define THRIFT_LSEEK _lseek +# define THRIFT_WRITE _write +# define THRIFT_READ _read +# define THRIFT_FSTAT _fstat +# define THRIFT_STAT _stat # define THRIFT_GAI_STRERROR gai_strerrorA # define THRIFT_SSIZET ptrdiff_t # define THRIFT_SNPRINTF _snprintf @@ -61,6 +71,7 @@ #else //not _WIN32 # include # define THRIFT_GET_SOCKET_ERROR errno +# define THRIFT_ERRNO errno # define THRIFT_EINTR EINTR # define THRIFT_EINPROGRESS EINPROGRESS # define THRIFT_ECONNRESET ECONNRESET @@ -79,6 +90,15 @@ # define THRIFT_F_SETFL F_SETFL # define THRIFT_GETTIMEOFDAY gettimeofday # define THRIFT_CLOSESOCKET close +# define THRIFT_CLOSE close +# define THRIFT_OPEN open +# define THRIFT_FTRUNCATE ftruncate +# define THRIFT_FSYNC fsync +# define THRIFT_LSEEK lseek +# define THRIFT_WRITE write +# define THRIFT_READ read +# define THRIFT_STAT stat +# define THRIFT_FSTAT fstat # define THRIFT_GAI_STRERROR gai_strerror # define THRIFT_SSIZET ssize_t # define THRIFT_SNPRINTF snprintf diff --git a/lib/cpp/src/thrift/transport/TFDTransport.cpp b/lib/cpp/src/thrift/transport/TFDTransport.cpp index 3b72de52..26365f08 100644 --- a/lib/cpp/src/thrift/transport/TFDTransport.cpp +++ b/lib/cpp/src/thrift/transport/TFDTransport.cpp @@ -40,8 +40,8 @@ void TFDTransport::close() { return; } - int rv = ::THRIFT_CLOSESOCKET(fd_); - int errno_copy = THRIFT_GET_SOCKET_ERROR; + int rv = ::THRIFT_CLOSE(fd_); + int errno_copy = THRIFT_ERRNO; fd_ = -1; // Have to check uncaught_exception because this is called in the destructor. if (rv < 0 && !std::uncaught_exception()) { @@ -55,14 +55,14 @@ uint32_t TFDTransport::read(uint8_t* buf, uint32_t len) { unsigned int maxRetries = 5; // same as the TSocket default unsigned int retries = 0; while (true) { - THRIFT_SSIZET rv = ::read(fd_, buf, len); + THRIFT_SSIZET rv = ::THRIFT_READ(fd_, buf, len); if (rv < 0) { - if (THRIFT_GET_SOCKET_ERROR == THRIFT_EINTR && retries < maxRetries) { + if (THRIFT_ERRNO == THRIFT_EINTR && retries < maxRetries) { // If interrupted, try again ++retries; continue; } - int errno_copy = THRIFT_GET_SOCKET_ERROR; + int errno_copy = THRIFT_ERRNO; throw TTransportException(TTransportException::UNKNOWN, "TFDTransport::read()", errno_copy); @@ -75,10 +75,10 @@ uint32_t TFDTransport::read(uint8_t* buf, uint32_t len) { void TFDTransport::write(const uint8_t* buf, uint32_t len) { while (len > 0) { - THRIFT_SSIZET rv = ::write(fd_, buf, len); + THRIFT_SSIZET rv = ::THRIFT_WRITE(fd_, buf, len); if (rv < 0) { - int errno_copy = THRIFT_GET_SOCKET_ERROR; + int errno_copy = THRIFT_ERRNO; throw TTransportException(TTransportException::UNKNOWN, "TFDTransport::write()", errno_copy); diff --git a/lib/cpp/src/thrift/transport/TFileTransport.cpp b/lib/cpp/src/thrift/transport/TFileTransport.cpp index c94ecd20..c267f2ea 100644 --- a/lib/cpp/src/thrift/transport/TFileTransport.cpp +++ b/lib/cpp/src/thrift/transport/TFileTransport.cpp @@ -99,8 +99,8 @@ void TFileTransport::resetOutputFile(int fd, string filename, off_t offset) { // flush any events in the queue flush(); GlobalOutput.printf("error, current file (%s) not closed", filename_.c_str()); - if (-1 == ::THRIFT_CLOSESOCKET(fd_)) { - int errno_copy = THRIFT_GET_SOCKET_ERROR; + if (-1 == ::THRIFT_CLOSE(fd_)) { + int errno_copy = THRIFT_ERRNO; GlobalOutput.perror("TFileTransport: resetOutputFile() ::close() ", errno_copy); throw TTransportException(TTransportException::UNKNOWN, "TFileTransport: error in file close", errno_copy); } else { @@ -154,8 +154,8 @@ TFileTransport::~TFileTransport() { // close logfile if (fd_ > 0) { - if(-1 == ::THRIFT_CLOSESOCKET(fd_)) { - GlobalOutput.perror("TFileTransport: ~TFileTransport() ::close() ", THRIFT_GET_SOCKET_ERROR); + if(-1 == ::THRIFT_CLOSE(fd_)) { + GlobalOutput.perror("TFileTransport: ~TFileTransport() ::close() ", THRIFT_ERRNO); } else { //successfully closed fd fd_ = 0; @@ -300,7 +300,7 @@ void TFileTransport::writerThread() { try { openLogFile(); } catch (...) { - int errno_copy = THRIFT_GET_SOCKET_ERROR; + int errno_copy = THRIFT_ERRNO; GlobalOutput.perror("TFileTransport: writerThread() openLogFile() ", errno_copy); fd_ = 0; hasIOError = true; @@ -313,14 +313,10 @@ void TFileTransport::writerThread() { seekToEnd(); // throw away any partial events offset_ += readState_.lastDispatchPtr_; -#ifndef _WIN32 - ftruncate(fd_, offset_); -#else - _chsize_s(fd_, offset_); -#endif + THRIFT_FTRUNCATE(fd_, offset_); readState_.resetAllValues(); } catch (...) { - int errno_copy = THRIFT_GET_SOCKET_ERROR; + int errno_copy = THRIFT_ERRNO; GlobalOutput.perror("TFileTransport: writerThread() initialization ", errno_copy); hasIOError = true; } @@ -340,11 +336,9 @@ void TFileTransport::writerThread() { // Try to empty buffers before exit if (enqueueBuffer_->isEmpty() && dequeueBuffer_->isEmpty()) { -#ifndef _WIN32 - fsync(fd_); -#endif - if (-1 == ::THRIFT_CLOSESOCKET(fd_)) { - int errno_copy = THRIFT_GET_SOCKET_ERROR; + ::THRIFT_FSYNC(fd_); + if (-1 == ::THRIFT_CLOSE(fd_)) { + int errno_copy = THRIFT_ERRNO; GlobalOutput.perror("TFileTransport: writerThread() ::close() ", errno_copy); } else { //fd successfully closed @@ -369,7 +363,7 @@ void TFileTransport::writerThread() { return; } if (!fd_) { - ::THRIFT_CLOSESOCKET(fd_); + ::THRIFT_CLOSE(fd_); fd_ = 0; } try { @@ -403,14 +397,14 @@ void TFileTransport::writerThread() { // if adding this event will cross a chunk boundary, pad the chunk with zeros if (chunk1 != chunk2) { // refetch the offset to keep in sync - offset_ = lseek(fd_, 0, SEEK_CUR); + offset_ = THRIFT_LSEEK(fd_, 0, SEEK_CUR); int32_t padding = (int32_t)((offset_ / chunkSize_ + 1) * chunkSize_ - offset_); uint8_t* zeros = new uint8_t[padding]; memset(zeros, '\0', padding); boost::scoped_array array(zeros); if (-1 == ::write(fd_, zeros, padding)) { - int errno_copy = THRIFT_GET_SOCKET_ERROR; + int errno_copy = THRIFT_ERRNO; GlobalOutput.perror("TFileTransport: writerThread() error while padding zeros ", errno_copy); hasIOError = true; continue; @@ -422,8 +416,8 @@ void TFileTransport::writerThread() { // write the dequeued event to the file if (outEvent->eventSize_ > 0) { - if (-1 == ::write(fd_, outEvent->eventBuff_, outEvent->eventSize_)) { - int errno_copy = THRIFT_GET_SOCKET_ERROR; + if (-1 == ::THRIFT_WRITE(fd_, outEvent->eventBuff_, outEvent->eventSize_)) { + int errno_copy = THRIFT_ERRNO; GlobalOutput.perror("TFileTransport: error while writing event ", errno_copy); hasIOError = true; continue; @@ -487,9 +481,7 @@ void TFileTransport::writerThread() { if (flush) { // sync (force flush) file to disk -#ifndef _WIN32 - fsync(fd_); -#endif + THRIFT_FSYNC(fd_); unflushed = 0; getNextFlushTime(&ts_next_flush); @@ -600,7 +592,7 @@ eventInfo* TFileTransport::readEvent() { if (readState_.bufferPtr_ == readState_.bufferLen_) { // advance the offset pointer offset_ += readState_.bufferLen_; - readState_.bufferLen_ = static_cast(::read(fd_, readBuff_, readBuffSize_)); + readState_.bufferLen_ = static_cast(::THRIFT_READ(fd_, readBuff_, readBuffSize_)); // if (readState_.bufferLen_) { // T_DEBUG_L(1, "Amount read: %u (offset: %lu)", readState_.bufferLen_, offset_); // } @@ -809,11 +801,11 @@ void TFileTransport::seekToChunk(int32_t chunk) { seekToEnd = true; chunk = numChunks - 1; // this is the min offset to process events till - minEndOffset = lseek(fd_, 0, SEEK_END); + minEndOffset = ::THRIFT_LSEEK(fd_, 0, SEEK_END); } off_t newOffset = off_t(chunk) * chunkSize_; - offset_ = lseek(fd_, newOffset, SEEK_SET); + offset_ = ::THRIFT_LSEEK(fd_, newOffset, SEEK_SET); readState_.resetAllValues(); currentEvent_ = NULL; if (offset_ == -1) { @@ -847,11 +839,11 @@ uint32_t TFileTransport::getNumChunks() { return 0; } - struct stat f_info; - int rv = fstat(fd_, &f_info); + struct THRIFT_STAT f_info; + int rv = ::THRIFT_FSTAT(fd_, &f_info); if (rv < 0) { - int errno_copy = THRIFT_GET_SOCKET_ERROR; + int errno_copy = THRIFT_ERRNO; throw TTransportException(TTransportException::UNKNOWN, "TFileTransport::getNumChunks() (fstat)", errno_copy); @@ -877,17 +869,16 @@ void TFileTransport::openLogFile() { #ifndef _WIN32 mode_t mode = readOnly_ ? S_IRUSR | S_IRGRP | S_IROTH : S_IRUSR | S_IWUSR| S_IRGRP | S_IROTH; int flags = readOnly_ ? O_RDONLY : O_RDWR | O_CREAT | O_APPEND; - fd_ = ::open(filename_.c_str(), flags, mode); #else int mode = readOnly_ ? _S_IREAD : _S_IREAD | _S_IWRITE; int flags = readOnly_ ? _O_RDONLY : _O_RDWR | _O_CREAT | _O_APPEND; - fd_ = ::_open(filename_.c_str(), flags, mode); #endif + fd_ = ::THRIFT_OPEN(filename_.c_str(), flags, mode); offset_ = 0; // make sure open call was successful if(fd_ == -1) { - int errno_copy = THRIFT_GET_SOCKET_ERROR; + int errno_copy = THRIFT_ERRNO; GlobalOutput.perror("TFileTransport: openLogFile() ::open() file: " + filename_, errno_copy); throw TTransportException(TTransportException::NOT_OPEN, filename_, errno_copy); } diff --git a/lib/cpp/src/thrift/transport/TSimpleFileTransport.cpp b/lib/cpp/src/thrift/transport/TSimpleFileTransport.cpp index 9af1445e..6bd716e4 100644 --- a/lib/cpp/src/thrift/transport/TSimpleFileTransport.cpp +++ b/lib/cpp/src/thrift/transport/TSimpleFileTransport.cpp @@ -54,7 +54,7 @@ TSimpleFileTransport(const std::string& path, bool read, bool write) #else int mode = _S_IREAD | _S_IWRITE; #endif - int fd = ::open(path.c_str(), + int fd = ::THRIFT_OPEN(path.c_str(), flags, mode); if (fd < 0) { diff --git a/lib/cpp/test/TFDTransportTest.cpp b/lib/cpp/test/TFDTransportTest.cpp index e30d9c0b..7b962d83 100644 --- a/lib/cpp/test/TFDTransportTest.cpp +++ b/lib/cpp/test/TFDTransportTest.cpp @@ -19,6 +19,7 @@ #include #include +#include #include #include using apache::thrift::transport::TTransportException; @@ -37,8 +38,10 @@ int main() { TFDTransport t(256, TFDTransport::CLOSE_ON_DESTROY); t.close(); } + std::cout << "NOT OK 0!" << std::endl; std::abort(); } catch (TTransportException) { + std::cout << "OK!" << std::endl; } try {