@ -23,6 +23,7 @@
# include <ares.h>
# include "absl/functional/any_invocable.h"
# include "absl/log/check.h"
# include "absl/status/status.h"
# include "absl/strings/str_format.h"
@ -132,8 +133,8 @@ class GrpcPolledFdWindows : public GrpcPolledFd {
shutdown_called_ ) ;
grpc_core : : CSliceUnref ( read_buf_ ) ;
grpc_core : : CSliceUnref ( write_buf_ ) ;
GPR_ASSERT ( read_closure_ = = nullptr ) ;
GPR_ASSERT ( write_closure_ = = nullptr ) ;
CHECK ( read_closure_ = = nullptr ) ;
CHECK ( write_closure_ = = nullptr ) ;
if ( ! shutdown_called_ ) {
winsocket_ - > Shutdown ( DEBUG_LOCATION , " ~GrpcPolledFdWindows " ) ;
}
@ -141,15 +142,15 @@ class GrpcPolledFdWindows : public GrpcPolledFd {
void RegisterForOnReadableLocked (
absl : : AnyInvocable < void ( absl : : Status ) > read_closure ) override {
GPR_ASSERT ( read_closure_ = = nullptr ) ;
CHECK ( read_closure_ = = nullptr ) ;
read_closure_ = std : : move ( read_closure ) ;
grpc_core : : CSliceUnref ( read_buf_ ) ;
GPR_ASSERT ( ! read_buf_has_data_ ) ;
CHECK ( ! read_buf_has_data_ ) ;
read_buf_ = GRPC_SLICE_MALLOC ( kReadBufferSize ) ;
if ( connect_done_ ) {
ContinueRegisterForOnReadableLocked ( ) ;
} else {
GPR_ASSERT ( pending_continue_register_for_on_readable_locked_ = = false ) ;
CHECK ( pending_continue_register_for_on_readable_locked_ = = false ) ;
pending_continue_register_for_on_readable_locked_ = true ;
}
}
@ -160,16 +161,16 @@ class GrpcPolledFdWindows : public GrpcPolledFd {
GRPC_ARES_RESOLVER_TRACE_LOG (
" fd:|%s| RegisterForOnWriteableLocked called " , GetName ( ) ) ;
} else {
GPR_ASSERT ( socket_type_ = = SOCK_STREAM ) ;
CHECK ( socket_type_ = = SOCK_STREAM ) ;
GRPC_ARES_RESOLVER_TRACE_LOG (
" fd:|%s| RegisterForOnWriteableLocked called tcp_write_state_: %d "
" connect_done_: %d " ,
GetName ( ) , tcp_write_state_ , connect_done_ ) ;
}
GPR_ASSERT ( write_closure_ = = nullptr ) ;
CHECK ( write_closure_ = = nullptr ) ;
write_closure_ = std : : move ( write_closure ) ;
if ( ! connect_done_ ) {
GPR_ASSERT ( ! pending_continue_register_for_on_writeable_locked_ ) ;
CHECK ( ! pending_continue_register_for_on_writeable_locked_ ) ;
pending_continue_register_for_on_writeable_locked_ = true ;
} else {
ContinueRegisterForOnWriteableLocked ( ) ;
@ -179,7 +180,7 @@ class GrpcPolledFdWindows : public GrpcPolledFd {
bool IsFdStillReadableLocked ( ) override { return read_buf_has_data_ ; }
bool ShutdownLocked ( absl : : Status error ) override {
GPR_ASSERT ( ! shutdown_called_ ) ;
CHECK ( ! shutdown_called_ ) ;
if ( ! absl : : IsCancelled ( error ) ) {
return false ;
}
@ -223,7 +224,7 @@ class GrpcPolledFdWindows : public GrpcPolledFd {
// c-ares overloads this recv_from virtual socket function to receive
// data on both UDP and TCP sockets, and from is nullptr for TCP.
if ( from ! = nullptr ) {
GPR_ASSERT ( * from_len < = recv_from_source_addr_len_ ) ;
CHECK ( * from_len < = recv_from_source_addr_len_ ) ;
memcpy ( from , & recv_from_source_addr_ , recv_from_source_addr_len_ ) ;
* from_len = recv_from_source_addr_len_ ;
}
@ -291,7 +292,7 @@ class GrpcPolledFdWindows : public GrpcPolledFd {
" fd:|%s| ContinueRegisterForOnReadableLocked "
" wsa_connect_error_:%d " ,
GetName ( ) , wsa_connect_error_ ) ;
GPR_ASSERT ( connect_done_ ) ;
CHECK ( connect_done_ ) ;
if ( wsa_connect_error_ ! = 0 ) {
ScheduleAndNullReadClosure ( GRPC_WSA_ERROR ( wsa_connect_error_ , " connect " ) ) ;
return ;
@ -328,7 +329,7 @@ class GrpcPolledFdWindows : public GrpcPolledFd {
" fd:|%s| ContinueRegisterForOnWriteableLocked "
" wsa_connect_error_:%d " ,
GetName ( ) , wsa_connect_error_ ) ;
GPR_ASSERT ( connect_done_ ) ;
CHECK ( connect_done_ ) ;
if ( wsa_connect_error_ ! = 0 ) {
ScheduleAndNullWriteClosure (
GRPC_WSA_ERROR ( wsa_connect_error_ , " connect " ) ) ;
@ -338,7 +339,7 @@ class GrpcPolledFdWindows : public GrpcPolledFd {
ScheduleAndNullWriteClosure ( absl : : OkStatus ( ) ) ;
return ;
}
GPR_ASSERT ( socket_type_ = = SOCK_STREAM ) ;
CHECK ( socket_type_ = = SOCK_STREAM ) ;
int wsa_error_code = 0 ;
switch ( tcp_write_state_ ) {
case WRITE_IDLE :
@ -386,7 +387,7 @@ class GrpcPolledFdWindows : public GrpcPolledFd {
// Therefore, the sendv handler for UDP sockets must only attempt
// to write everything inline.
GRPC_ARES_RESOLVER_TRACE_LOG ( " fd:|%s| SendVUDP called " , GetName ( ) ) ;
GPR_ASSERT ( GRPC_SLICE_LENGTH ( write_buf_ ) = = 0 ) ;
CHECK_EQ ( GRPC_SLICE_LENGTH ( write_buf_ ) , 0 ) ;
grpc_core : : CSliceUnref ( write_buf_ ) ;
write_buf_ = FlattenIovec ( iov , iov_count ) ;
DWORD bytes_sent = 0 ;
@ -434,12 +435,12 @@ class GrpcPolledFdWindows : public GrpcPolledFd {
// send again. If c-ares still needs to send even more data, we'll get
// to it eventually.
grpc_slice currently_attempted = FlattenIovec ( iov , iov_count ) ;
GPR_ASSERT ( GRPC_SLICE_LENGTH ( currently_attempted ) > =
GRPC_SLICE_LENGTH ( write_buf_ ) ) ;
CHECK ( GRPC_SLICE_LENGTH ( currently_attempted ) > =
GRPC_SLICE_LENGTH ( write_buf_ ) ) ;
ares_ssize_t total_sent = 0 ;
for ( size_t i = 0 ; i < GRPC_SLICE_LENGTH ( write_buf_ ) ; i + + ) {
GPR_ASSERT ( GRPC_SLICE_START_PTR ( currently_attempted ) [ i ] = =
GRPC_SLICE_START_PTR ( write_buf_ ) [ i ] ) ;
CHECK ( GRPC_SLICE_START_PTR ( currently_attempted ) [ i ] = =
GRPC_SLICE_START_PTR ( write_buf_ ) [ i ] ) ;
total_sent + + ;
}
grpc_core : : CSliceUnref ( currently_attempted ) ;
@ -458,9 +459,9 @@ class GrpcPolledFdWindows : public GrpcPolledFd {
" pending_register_for_writeable:%d " ,
GetName ( ) , pending_continue_register_for_on_readable_locked_ ,
pending_continue_register_for_on_writeable_locked_ ) ;
GPR_ASSERT ( ! connect_done_ ) ;
CHECK ( ! connect_done_ ) ;
connect_done_ = true ;
GPR_ASSERT ( wsa_connect_error_ = = 0 ) ;
CHECK_EQ ( wsa_connect_error_ , 0 ) ;
if ( shutdown_called_ ) {
wsa_connect_error_ = WSA_OPERATION_ABORTED ;
} else {
@ -469,7 +470,7 @@ class GrpcPolledFdWindows : public GrpcPolledFd {
BOOL wsa_success = WSAGetOverlappedResult (
winsocket_ - > raw_socket ( ) , winsocket_ - > write_info ( ) - > overlapped ( ) ,
& transferred_bytes , FALSE , & flags ) ;
GPR_ASSERT ( transferred_bytes = = 0 ) ;
CHECK_EQ ( transferred_bytes , 0 ) ;
if ( ! wsa_success ) {
wsa_connect_error_ = WSAGetLastError ( ) ;
char * msg = gpr_format_message ( wsa_connect_error_ ) ;
@ -491,8 +492,8 @@ class GrpcPolledFdWindows : public GrpcPolledFd {
int ConnectUDP ( WSAErrorContext * wsa_error_ctx , const struct sockaddr * target ,
ares_socklen_t target_len ) {
GRPC_ARES_RESOLVER_TRACE_LOG ( " fd:%s ConnectUDP " , GetName ( ) ) ;
GPR_ASSERT ( ! connect_done_ ) ;
GPR_ASSERT ( wsa_connect_error_ = = 0 ) ;
CHECK ( ! connect_done_ ) ;
CHECK_EQ ( wsa_connect_error_ , 0 ) ;
SOCKET s = winsocket_ - > raw_socket ( ) ;
int out =
WSAConnect ( s , target , target_len , nullptr , nullptr , nullptr , nullptr ) ;
@ -620,7 +621,7 @@ class GrpcPolledFdWindows : public GrpcPolledFd {
void OnIocpWriteable ( ) {
grpc_core : : MutexLock lock ( mu_ ) ;
GRPC_ARES_RESOLVER_TRACE_LOG ( " OnIocpWriteableInner. fd:|%s| " , GetName ( ) ) ;
GPR_ASSERT ( socket_type_ = = SOCK_STREAM ) ;
CHECK ( socket_type_ = = SOCK_STREAM ) ;
absl : : Status error ;
if ( winsocket_ - > write_info ( ) - > result ( ) . wsa_error ! = 0 ) {
error = GRPC_WSA_ERROR ( winsocket_ - > write_info ( ) - > result ( ) . wsa_error ,
@ -631,7 +632,7 @@ class GrpcPolledFdWindows : public GrpcPolledFd {
GetName ( ) , winsocket_ - > write_info ( ) - > result ( ) . wsa_error ,
grpc_core : : StatusToString ( error ) . c_str ( ) ) ;
}
GPR_ASSERT ( tcp_write_state_ = = WRITE_PENDING ) ;
CHECK ( tcp_write_state_ = = WRITE_PENDING ) ;
if ( error . ok ( ) ) {
tcp_write_state_ = WRITE_WAITING_FOR_VERIFICATION_UPON_RETRY ;
write_buf_ = grpc_slice_sub_no_ref (
@ -712,7 +713,7 @@ class CustomSockFuncs {
GRPC_ARES_RESOLVER_TRACE_LOG (
" fd:|%s| created with params af:%d type:%d protocol:%d " ,
polled_fd - > GetName ( ) , af , type , protocol ) ;
GPR_ASSERT ( self - > sockets_ . insert ( { s , std : : move ( polled_fd ) } ) . second ) ;
CHECK ( self - > sockets_ . insert ( { s , std : : move ( polled_fd ) } ) . second ) ;
return s ;
}
@ -722,7 +723,7 @@ class CustomSockFuncs {
GrpcPolledFdFactoryWindows * self =
static_cast < GrpcPolledFdFactoryWindows * > ( user_data ) ;
auto it = self - > sockets_ . find ( as ) ;
GPR_ASSERT ( it ! = self - > sockets_ . end ( ) ) ;
CHECK ( it ! = self - > sockets_ . end ( ) ) ;
return it - > second - > Connect ( & wsa_error_ctx , target , target_len ) ;
}
@ -732,7 +733,7 @@ class CustomSockFuncs {
GrpcPolledFdFactoryWindows * self =
static_cast < GrpcPolledFdFactoryWindows * > ( user_data ) ;
auto it = self - > sockets_ . find ( as ) ;
GPR_ASSERT ( it ! = self - > sockets_ . end ( ) ) ;
CHECK ( it ! = self - > sockets_ . end ( ) ) ;
return it - > second - > SendV ( & wsa_error_ctx , iov , iovec_count ) ;
}
@ -743,7 +744,7 @@ class CustomSockFuncs {
GrpcPolledFdFactoryWindows * self =
static_cast < GrpcPolledFdFactoryWindows * > ( user_data ) ;
auto it = self - > sockets_ . find ( as ) ;
GPR_ASSERT ( it ! = self - > sockets_ . end ( ) ) ;
CHECK ( it ! = self - > sockets_ . end ( ) ) ;
return it - > second - > RecvFrom ( & wsa_error_ctx , data , data_len , flags , from ,
from_len ) ;
}
@ -802,7 +803,7 @@ void GrpcPolledFdFactoryWindows::Initialize(grpc_core::Mutex* mutex,
std : : unique_ptr < GrpcPolledFd > GrpcPolledFdFactoryWindows : : NewGrpcPolledFdLocked (
ares_socket_t as ) {
auto it = sockets_ . find ( as ) ;
GPR_ASSERT ( it ! = sockets_ . end ( ) ) ;
CHECK ( it ! = sockets_ . end ( ) ) ;
return std : : make_unique < GrpcPolledFdWrapper > ( it - > second . get ( ) ) ;
}
Tanvi Jagtap
7 months ago
committed by
Copybara-Service