Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(71)

Side by Side Diff: net/udp/udp_socket_win.cc

Issue 861963002: UDP: Windows implementation using non-blocking IO (Closed) Base URL: https://chromium.googlesource.com/chromium/src.git@master
Patch Set: git cl try Created 5 years, 11 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch
« net/udp/udp_socket_perftest.cc ('K') | « net/udp/udp_socket_win.h ('k') | no next file » | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
OLDNEW
1 // Copyright (c) 2012 The Chromium Authors. All rights reserved. 1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
2 // Use of this source code is governed by a BSD-style license that can be 2 // Use of this source code is governed by a BSD-style license that can be
3 // found in the LICENSE file. 3 // found in the LICENSE file.
4 4
5 #include "net/udp/udp_socket_win.h" 5 #include "net/udp/udp_socket_win.h"
6 6
7 #include <mstcpip.h> 7 #include <mstcpip.h>
8 8
9 #include "base/basictypes.h" 9 #include "base/basictypes.h"
10 #include "base/callback.h" 10 #include "base/callback.h"
(...skipping 26 matching lines...) Expand all
37 37
38 namespace net { 38 namespace net {
39 39
40 // This class encapsulates all the state that has to be preserved as long as 40 // This class encapsulates all the state that has to be preserved as long as
41 // there is a network IO operation in progress. If the owner UDPSocketWin 41 // there is a network IO operation in progress. If the owner UDPSocketWin
42 // is destroyed while an operation is in progress, the Core is detached and it 42 // is destroyed while an operation is in progress, the Core is detached and it
43 // lives until the operation completes and the OS doesn't reference any resource 43 // lives until the operation completes and the OS doesn't reference any resource
44 // declared on this class anymore. 44 // declared on this class anymore.
45 class UDPSocketWin::Core : public base::RefCounted<Core> { 45 class UDPSocketWin::Core : public base::RefCounted<Core> {
46 public: 46 public:
47 explicit Core(UDPSocketWin* socket); 47 explicit Core(UDPSocketWin* socket, bool use_overlapped_io);
48 48
49 // Start watching for the end of a read or write operation. 49 // Start watching for the end of a read or write operation in overlapped mode.
50 // In non-blocking mode watch the signals to continure read/write.
50 void WatchForRead(); 51 void WatchForRead();
51 void WatchForWrite(); 52 void WatchForWrite();
52 53
53 // The UDPSocketWin is going away. 54 // The UDPSocketWin is going away.
54 void Detach() { socket_ = NULL; } 55 void Detach() { socket_ = NULL; }
55 56
56 // The separate OVERLAPPED variables for asynchronous operation. 57 bool use_overlapped_io() const { return use_overlapped_io_; }
57 OVERLAPPED read_overlapped_;
58 OVERLAPPED write_overlapped_;
59 58
60 // The buffers used in Read() and Write(). 59 OVERLAPPED* read_overlapped() { return &read_overlapped_; }
61 scoped_refptr<IOBuffer> read_iobuffer_; 60 OVERLAPPED* write_overlapped() { return &write_overlapped_; }
62 scoped_refptr<IOBuffer> write_iobuffer_;
63 61
64 // The address storage passed to WSARecvFrom(). 62 WSAEVENT read_event() const { return read_event_; }
65 SockaddrStorage recv_addr_storage_; 63 WSAEVENT write_event() const { return write_event_; }
64
65 scoped_refptr<IOBuffer> read_iobuffer() const { return read_iobuffer_; }
66 scoped_refptr<IOBuffer> write_iobuffer() const { return write_iobuffer_; }
67
68 void set_read_iobuffer(scoped_refptr<IOBuffer> buf, int len) {
69 read_iobuffer_ = buf;
70 read_iobuffer_len_ = len;
71 }
72 void set_write_iobuffer(scoped_refptr<IOBuffer> buf, int len) {
73 write_iobuffer_ = buf;
74 write_iobuffer_len_ = len;
75 }
76
77 int read_iobuffer_len() const { return read_iobuffer_len_; }
78 int write_iobuffer_len() const { return write_iobuffer_len_; }
79
80 SockaddrStorage* recv_addr_storage() { return &recv_addr_storage_; }
66 81
67 private: 82 private:
68 friend class base::RefCounted<Core>; 83 friend class base::RefCounted<Core>;
69 84
70 class ReadDelegate : public base::win::ObjectWatcher::Delegate { 85 class ReadDelegate : public base::win::ObjectWatcher::Delegate {
71 public: 86 public:
72 explicit ReadDelegate(Core* core) : core_(core) {} 87 explicit ReadDelegate(Core* core) : core_(core) {}
73 virtual ~ReadDelegate() {} 88 virtual ~ReadDelegate() {}
74 89
75 // base::ObjectWatcher::Delegate methods: 90 // base::ObjectWatcher::Delegate methods:
(...skipping 12 matching lines...) Expand all
88 virtual void OnObjectSignaled(HANDLE object); 103 virtual void OnObjectSignaled(HANDLE object);
89 104
90 private: 105 private:
91 Core* const core_; 106 Core* const core_;
92 }; 107 };
93 108
94 ~Core(); 109 ~Core();
95 110
96 // The socket that created this object. 111 // The socket that created this object.
97 UDPSocketWin* socket_; 112 UDPSocketWin* socket_;
113 const bool use_overlapped_io_;
98 114
99 // |reader_| handles the signals from |read_watcher_|. 115 // |reader_| handles the signals from |read_watcher_|.
100 ReadDelegate reader_; 116 ReadDelegate reader_;
101 // |writer_| handles the signals from |write_watcher_|. 117 // |writer_| handles the signals from |write_watcher_|.
102 WriteDelegate writer_; 118 WriteDelegate writer_;
103 119
104 // |read_watcher_| watches for events from Read(). 120 // |read_watcher_| watches for events from Read().
105 base::win::ObjectWatcher read_watcher_; 121 base::win::ObjectWatcher read_watcher_;
106 // |write_watcher_| watches for events from Write(); 122 // |write_watcher_| watches for events from Write();
107 base::win::ObjectWatcher write_watcher_; 123 base::win::ObjectWatcher write_watcher_;
108 124
125 // The separate OVERLAPPED variables for asynchronous operation.
126 OVERLAPPED read_overlapped_;
127 OVERLAPPED write_overlapped_;
128
129 // Separate events for non-blocking IO operations.
130 WSAEVENT read_event_;
131 WSAEVENT write_event_;
132
133 // The buffers used in Read() and Write().
134 scoped_refptr<IOBuffer> read_iobuffer_;
135 scoped_refptr<IOBuffer> write_iobuffer_;
136 int read_iobuffer_len_;
137 int write_iobuffer_len_;
138
139 // The address storage passed to WSARecvFrom().
140 SockaddrStorage recv_addr_storage_;
141
109 DISALLOW_COPY_AND_ASSIGN(Core); 142 DISALLOW_COPY_AND_ASSIGN(Core);
110 }; 143 };
111 144
112 UDPSocketWin::Core::Core(UDPSocketWin* socket) 145 UDPSocketWin::Core::Core(UDPSocketWin* socket, bool use_overlapped_io)
113 : socket_(socket), 146 : socket_(socket),
147 use_overlapped_io_(use_overlapped_io),
114 reader_(this), 148 reader_(this),
115 writer_(this) { 149 writer_(this),
150 read_iobuffer_len_(0),
151 write_iobuffer_len_(0) {
116 memset(&read_overlapped_, 0, sizeof(read_overlapped_)); 152 memset(&read_overlapped_, 0, sizeof(read_overlapped_));
117 memset(&write_overlapped_, 0, sizeof(write_overlapped_)); 153 memset(&write_overlapped_, 0, sizeof(write_overlapped_));
118 154
119 read_overlapped_.hEvent = WSACreateEvent(); 155 if (use_overlapped_io) {
120 write_overlapped_.hEvent = WSACreateEvent(); 156 read_overlapped_.hEvent = WSACreateEvent();
157 write_overlapped_.hEvent = WSACreateEvent();
158 } else {
159 read_event_ = WSACreateEvent();
160 write_event_ = WSACreateEvent();
161 }
121 } 162 }
122 163
123 UDPSocketWin::Core::~Core() { 164 UDPSocketWin::Core::~Core() {
124 // Make sure the message loop is not watching this object anymore. 165 // Make sure the message loop is not watching this object anymore.
125 read_watcher_.StopWatching(); 166 read_watcher_.StopWatching();
126 write_watcher_.StopWatching(); 167 write_watcher_.StopWatching();
127 168
128 WSACloseEvent(read_overlapped_.hEvent); 169 if (use_overlapped_io_) {
129 memset(&read_overlapped_, 0xaf, sizeof(read_overlapped_)); 170 WSACloseEvent(read_overlapped_.hEvent);
130 WSACloseEvent(write_overlapped_.hEvent); 171 memset(&read_overlapped_, 0, sizeof(read_overlapped_));
131 memset(&write_overlapped_, 0xaf, sizeof(write_overlapped_)); 172 WSACloseEvent(write_overlapped_.hEvent);
173 memset(&write_overlapped_, 0, sizeof(write_overlapped_));
174 } else {
175 WSACloseEvent(read_event_);
176 WSACloseEvent(write_event_);
177 }
132 } 178 }
133 179
134 void UDPSocketWin::Core::WatchForRead() { 180 void UDPSocketWin::Core::WatchForRead() {
135 // We grab an extra reference because there is an IO operation in progress. 181 // We grab an extra reference because there is an IO operation in progress.
136 // Balanced in ReadDelegate::OnObjectSignaled(). 182 // Balanced in ReadDelegate::OnObjectSignaled().
137 AddRef(); 183 AddRef();
138 read_watcher_.StartWatching(read_overlapped_.hEvent, &reader_); 184 if (use_overlapped_io_)
185 read_watcher_.StartWatching(read_overlapped_.hEvent, &reader_);
186 else
187 read_watcher_.StartWatching(read_event_, &reader_);
139 } 188 }
140 189
141 void UDPSocketWin::Core::WatchForWrite() { 190 void UDPSocketWin::Core::WatchForWrite() {
142 // We grab an extra reference because there is an IO operation in progress. 191 // We grab an extra reference because there is an IO operation in progress.
143 // Balanced in WriteDelegate::OnObjectSignaled(). 192 // Balanced in WriteDelegate::OnObjectSignaled().
144 AddRef(); 193 AddRef();
145 write_watcher_.StartWatching(write_overlapped_.hEvent, &writer_); 194 if (use_overlapped_io_)
195 write_watcher_.StartWatching(write_overlapped_.hEvent, &writer_);
196 else
197 write_watcher_.StartWatching(write_event_, &writer_);
146 } 198 }
147 199
148 void UDPSocketWin::Core::ReadDelegate::OnObjectSignaled(HANDLE object) { 200 void UDPSocketWin::Core::ReadDelegate::OnObjectSignaled(HANDLE object) {
149 // TODO(vadimt): Remove ScopedTracker below once crbug.com/418183 is fixed. 201 // TODO(vadimt): Remove ScopedTracker below once crbug.com/418183 is fixed.
150 tracked_objects::ScopedTracker tracking_profile( 202 tracked_objects::ScopedTracker tracking_profile(
151 FROM_HERE_WITH_EXPLICIT_FUNCTION( 203 FROM_HERE_WITH_EXPLICIT_FUNCTION(
152 "UDPSocketWin_Core_ReadDelegate_OnObjectSignaled")); 204 "UDPSocketWin_Core_ReadDelegate_OnObjectSignaled"));
153 205
154 DCHECK_EQ(object, core_->read_overlapped_.hEvent); 206 if (core_->use_overlapped_io()) {
155 if (core_->socket_) 207 DCHECK_EQ(object, core_->read_overlapped()->hEvent);
156 core_->socket_->DidCompleteRead(); 208 if (core_->socket_)
209 core_->socket_->DidCompleteReadOverlapped();
210 } else {
211 DCHECK_EQ(object, core_->read_event());
212 if (core_->socket_)
213 core_->socket_->OnReadSignaledNonBlocking();
214 }
157 215
158 core_->Release(); 216 core_->Release();
159 } 217 }
160 218
161 void UDPSocketWin::Core::WriteDelegate::OnObjectSignaled(HANDLE object) { 219 void UDPSocketWin::Core::WriteDelegate::OnObjectSignaled(HANDLE object) {
162 // TODO(vadimt): Remove ScopedTracker below once crbug.com/418183 is fixed. 220 // TODO(vadimt): Remove ScopedTracker below once crbug.com/418183 is fixed.
163 tracked_objects::ScopedTracker tracking_profile( 221 tracked_objects::ScopedTracker tracking_profile(
164 FROM_HERE_WITH_EXPLICIT_FUNCTION( 222 FROM_HERE_WITH_EXPLICIT_FUNCTION(
165 "UDPSocketWin_Core_WriteDelegate_OnObjectSignaled")); 223 "UDPSocketWin_Core_WriteDelegate_OnObjectSignaled"));
166 224
167 DCHECK_EQ(object, core_->write_overlapped_.hEvent); 225 if (core_->use_overlapped_io()) {
168 if (core_->socket_) 226 DCHECK_EQ(object, core_->write_overlapped()->hEvent);
169 core_->socket_->DidCompleteWrite(); 227 if (core_->socket_)
228 core_->socket_->DidCompleteWriteOverlapped();
229 } else {
230 DCHECK_EQ(object, core_->write_event());
231 if (core_->socket_)
232 core_->socket_->OnWriteSignaledNonBlocking();
233 }
170 234
171 core_->Release(); 235 core_->Release();
172 } 236 }
173 //----------------------------------------------------------------------------- 237 //-----------------------------------------------------------------------------
174 238
175 QwaveAPI::QwaveAPI() : qwave_supported_(false) { 239 QwaveAPI::QwaveAPI() : qwave_supported_(false) {
176 HMODULE qwave = LoadLibrary(L"qwave.dll"); 240 HMODULE qwave = LoadLibrary(L"qwave.dll");
177 if (!qwave) 241 if (!qwave)
178 return; 242 return;
179 create_handle_func_ = 243 create_handle_func_ =
(...skipping 74 matching lines...) Expand 10 before | Expand all | Expand 10 after
254 net::NetLog* net_log, 318 net::NetLog* net_log,
255 const net::NetLog::Source& source) 319 const net::NetLog::Source& source)
256 : socket_(INVALID_SOCKET), 320 : socket_(INVALID_SOCKET),
257 addr_family_(0), 321 addr_family_(0),
258 is_connected_(false), 322 is_connected_(false),
259 socket_options_(SOCKET_OPTION_MULTICAST_LOOP), 323 socket_options_(SOCKET_OPTION_MULTICAST_LOOP),
260 multicast_interface_(0), 324 multicast_interface_(0),
261 multicast_time_to_live_(1), 325 multicast_time_to_live_(1),
262 bind_type_(bind_type), 326 bind_type_(bind_type),
263 rand_int_cb_(rand_int_cb), 327 rand_int_cb_(rand_int_cb),
328 non_blocking_reads_initialized_(false),
329 non_blocking_writes_initialized_(false),
264 recv_from_address_(NULL), 330 recv_from_address_(NULL),
265 net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_UDP_SOCKET)), 331 net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_UDP_SOCKET)),
266 qos_handle_(NULL), 332 qos_handle_(NULL),
267 qos_flow_id_(0) { 333 qos_flow_id_(0),
334 use_overlapped_io_(true) {
268 EnsureWinsockInit(); 335 EnsureWinsockInit();
269 net_log_.BeginEvent(NetLog::TYPE_SOCKET_ALIVE, 336 net_log_.BeginEvent(NetLog::TYPE_SOCKET_ALIVE,
270 source.ToEventParametersCallback()); 337 source.ToEventParametersCallback());
271 if (bind_type == DatagramSocket::RANDOM_BIND) 338 if (bind_type == DatagramSocket::RANDOM_BIND)
272 DCHECK(!rand_int_cb.is_null()); 339 DCHECK(!rand_int_cb.is_null());
273 } 340 }
274 341
275 UDPSocketWin::~UDPSocketWin() { 342 UDPSocketWin::~UDPSocketWin() {
276 Close(); 343 Close();
277 net_log_.EndEvent(NetLog::TYPE_SOCKET_ALIVE); 344 net_log_.EndEvent(NetLog::TYPE_SOCKET_ALIVE);
278 } 345 }
279 346
280 int UDPSocketWin::Open(AddressFamily address_family) { 347 int UDPSocketWin::Open(AddressFamily address_family) {
281 DCHECK(CalledOnValidThread()); 348 DCHECK(CalledOnValidThread());
282 DCHECK_EQ(socket_, INVALID_SOCKET); 349 DCHECK_EQ(socket_, INVALID_SOCKET);
283 350
284 addr_family_ = ConvertAddressFamily(address_family); 351 addr_family_ = ConvertAddressFamily(address_family);
285 socket_ = CreatePlatformSocket(addr_family_, SOCK_DGRAM, IPPROTO_UDP); 352 socket_ = CreatePlatformSocket(addr_family_, SOCK_DGRAM, IPPROTO_UDP);
286 if (socket_ == INVALID_SOCKET) 353 if (socket_ == INVALID_SOCKET)
287 return MapSystemError(WSAGetLastError()); 354 return MapSystemError(WSAGetLastError());
288 core_ = new Core(this); 355 core_ = new Core(this, use_overlapped_io_);
289 return OK; 356 return OK;
290 } 357 }
291 358
292 void UDPSocketWin::Close() { 359 void UDPSocketWin::Close() {
293 DCHECK(CalledOnValidThread()); 360 DCHECK(CalledOnValidThread());
294 361
295 if (socket_ == INVALID_SOCKET) 362 if (socket_ == INVALID_SOCKET)
296 return; 363 return;
297 364
298 if (qos_handle_) { 365 if (qos_handle_) {
(...skipping 71 matching lines...) Expand 10 before | Expand all | Expand 10 after
370 int buf_len, 437 int buf_len,
371 IPEndPoint* address, 438 IPEndPoint* address,
372 const CompletionCallback& callback) { 439 const CompletionCallback& callback) {
373 DCHECK(CalledOnValidThread()); 440 DCHECK(CalledOnValidThread());
374 DCHECK_NE(INVALID_SOCKET, socket_); 441 DCHECK_NE(INVALID_SOCKET, socket_);
375 CHECK(read_callback_.is_null()); 442 CHECK(read_callback_.is_null());
376 DCHECK(!recv_from_address_); 443 DCHECK(!recv_from_address_);
377 DCHECK(!callback.is_null()); // Synchronous operation not supported. 444 DCHECK(!callback.is_null()); // Synchronous operation not supported.
378 DCHECK_GT(buf_len, 0); 445 DCHECK_GT(buf_len, 0);
379 446
380 int nread = InternalRecvFrom(buf, buf_len, address); 447 int nread = use_overlapped_io_
448 ? InternalRecvFromOverlapped(buf, buf_len, address)
449 : InternalRecvFromNonBlocking(buf, buf_len, address);
381 if (nread != ERR_IO_PENDING) 450 if (nread != ERR_IO_PENDING)
382 return nread; 451 return nread;
383 452
384 read_callback_ = callback; 453 read_callback_ = callback;
385 recv_from_address_ = address; 454 recv_from_address_ = address;
386 return ERR_IO_PENDING; 455 return ERR_IO_PENDING;
387 } 456 }
388 457
389 int UDPSocketWin::Write(IOBuffer* buf, 458 int UDPSocketWin::Write(IOBuffer* buf,
390 int buf_len, 459 int buf_len,
(...skipping 12 matching lines...) Expand all
403 int buf_len, 472 int buf_len,
404 const IPEndPoint* address, 473 const IPEndPoint* address,
405 const CompletionCallback& callback) { 474 const CompletionCallback& callback) {
406 DCHECK(CalledOnValidThread()); 475 DCHECK(CalledOnValidThread());
407 DCHECK_NE(INVALID_SOCKET, socket_); 476 DCHECK_NE(INVALID_SOCKET, socket_);
408 CHECK(write_callback_.is_null()); 477 CHECK(write_callback_.is_null());
409 DCHECK(!callback.is_null()); // Synchronous operation not supported. 478 DCHECK(!callback.is_null()); // Synchronous operation not supported.
410 DCHECK_GT(buf_len, 0); 479 DCHECK_GT(buf_len, 0);
411 DCHECK(!send_to_address_.get()); 480 DCHECK(!send_to_address_.get());
412 481
413 int nwrite = InternalSendTo(buf, buf_len, address); 482 int nwrite = use_overlapped_io_
483 ? InternalSendToOverlapped(buf, buf_len, address)
484 : InternalSendToNonBlocking(buf, buf_len, address);
414 if (nwrite != ERR_IO_PENDING) 485 if (nwrite != ERR_IO_PENDING)
415 return nwrite; 486 return nwrite;
416 487
417 if (address) 488 if (address)
418 send_to_address_.reset(new IPEndPoint(*address)); 489 send_to_address_.reset(new IPEndPoint(*address));
419 write_callback_ = callback; 490 write_callback_ = callback;
420 return ERR_IO_PENDING; 491 return ERR_IO_PENDING;
421 } 492 }
422 493
423 int UDPSocketWin::Connect(const IPEndPoint& address) { 494 int UDPSocketWin::Connect(const IPEndPoint& address) {
(...skipping 135 matching lines...) Expand 10 before | Expand all | Expand 10 after
559 void UDPSocketWin::DoWriteCallback(int rv) { 630 void UDPSocketWin::DoWriteCallback(int rv) {
560 DCHECK_NE(rv, ERR_IO_PENDING); 631 DCHECK_NE(rv, ERR_IO_PENDING);
561 DCHECK(!write_callback_.is_null()); 632 DCHECK(!write_callback_.is_null());
562 633
563 // since Run may result in Write being called, clear write_callback_ up front. 634 // since Run may result in Write being called, clear write_callback_ up front.
564 CompletionCallback c = write_callback_; 635 CompletionCallback c = write_callback_;
565 write_callback_.Reset(); 636 write_callback_.Reset();
566 c.Run(rv); 637 c.Run(rv);
567 } 638 }
568 639
569 void UDPSocketWin::DidCompleteRead() { 640 void UDPSocketWin::DidCompleteReadOverlapped() {
570 DWORD num_bytes, flags; 641 DWORD num_bytes, flags;
571 BOOL ok = WSAGetOverlappedResult(socket_, &core_->read_overlapped_, 642 BOOL ok = WSAGetOverlappedResult(socket_, core_->read_overlapped(),
572 &num_bytes, FALSE, &flags); 643 &num_bytes, FALSE, &flags);
573 WSAResetEvent(core_->read_overlapped_.hEvent); 644 WSAResetEvent(core_->read_overlapped()->hEvent);
574 int result = ok ? num_bytes : MapSystemError(WSAGetLastError()); 645 int result = ok ? num_bytes : MapSystemError(WSAGetLastError());
575 // Convert address. 646 // Convert address.
576 if (recv_from_address_ && result >= 0) { 647 if (recv_from_address_ && result >= 0) {
577 if (!ReceiveAddressToIPEndpoint(recv_from_address_)) 648 if (!ReceiveAddressToIPEndpoint(recv_from_address_))
578 result = ERR_ADDRESS_INVALID; 649 result = ERR_ADDRESS_INVALID;
579 } 650 }
580 LogRead(result, core_->read_iobuffer_->data()); 651 LogRead(result, core_->read_iobuffer()->data());
581 core_->read_iobuffer_ = NULL; 652 core_->set_read_iobuffer(NULL, 0);
582 recv_from_address_ = NULL; 653 recv_from_address_ = NULL;
583 DoReadCallback(result); 654 DoReadCallback(result);
584 } 655 }
585 656
657 void UDPSocketWin::DidCompleteWriteOverlapped() {
658 DWORD num_bytes, flags;
659 BOOL ok = WSAGetOverlappedResult(socket_, core_->write_overlapped(),
660 &num_bytes, FALSE, &flags);
661 WSAResetEvent(core_->write_overlapped()->hEvent);
662 int result = ok ? num_bytes : MapSystemError(WSAGetLastError());
663 LogWrite(result, core_->write_iobuffer()->data(), send_to_address_.get());
664
665 send_to_address_.reset();
666 core_->set_write_iobuffer(NULL, 0);
667 DoWriteCallback(result);
668 }
669
670 void UDPSocketWin::OnReadSignaledNonBlocking() {
671 DCHECK(core_->read_iobuffer());
672 WSANETWORKEVENTS network_events;
673 int os_error = 0;
674 int rv = WSAEnumNetworkEvents(socket_, core_->read_event(), &network_events);
675 if (rv == SOCKET_ERROR) {
676 os_error = WSAGetLastError();
677 rv = MapSystemError(os_error);
678 DoReadCallback(rv);
679 return;
680 }
681 if (network_events.lNetworkEvents) {
682 DCHECK(network_events.lNetworkEvents & FD_READ);
683 // Regardless of the error code in network_events.iErrorCode[FD_READ_BIT]
684 // we will continue to call recvfrom because it gives a more accurate
685 // error code.
686 rv = InternalRecvFromNonBlocking(core_->read_iobuffer().get(),
687 core_->read_iobuffer_len(),
688 recv_from_address_);
689 if (rv == ERR_IO_PENDING)
690 return;
691 } else {
692 core_->WatchForRead();
693 return;
694 }
695 core_->set_read_iobuffer(NULL, 0);
696 recv_from_address_ = NULL;
697 DoReadCallback(rv);
698 }
699
700 void UDPSocketWin::OnWriteSignaledNonBlocking() {
701 DCHECK(core_->write_iobuffer());
702 WSANETWORKEVENTS network_events;
703 int os_error = 0;
704 int rv = WSAEnumNetworkEvents(socket_, core_->write_event(), &network_events);
705 if (rv == SOCKET_ERROR) {
706 os_error = WSAGetLastError();
707 rv = MapSystemError(os_error);
708 DoWriteCallback(rv);
709 return;
710 } else if (network_events.lNetworkEvents) {
711 DCHECK(network_events.lNetworkEvents & FD_WRITE);
712 // Regardless of the error code in network_events.iErrorCode[FD_WRITE_BIT]
713 // we will continue to call sendto because it gives a more accurate error
714 // code.
715 rv = InternalSendToNonBlocking(core_->write_iobuffer().get(),
716 core_->write_iobuffer_len(),
717 send_to_address_.get());
718 if (rv == ERR_IO_PENDING)
719 return;
720 } else {
721 core_->WatchForWrite();
722 return;
723 }
724 core_->set_write_iobuffer(NULL, 0);
725 send_to_address_.reset();
726 DoWriteCallback(rv);
727 }
728
586 void UDPSocketWin::LogRead(int result, const char* bytes) const { 729 void UDPSocketWin::LogRead(int result, const char* bytes) const {
587 if (result < 0) { 730 if (result < 0) {
588 net_log_.AddEventWithNetErrorCode(NetLog::TYPE_UDP_RECEIVE_ERROR, result); 731 net_log_.AddEventWithNetErrorCode(NetLog::TYPE_UDP_RECEIVE_ERROR, result);
589 return; 732 return;
590 } 733 }
591 734
592 if (net_log_.IsLogging()) { 735 if (net_log_.IsLogging()) {
593 // Get address for logging, if |address| is NULL. 736 // Get address for logging, if |address| is NULL.
594 IPEndPoint address; 737 IPEndPoint address;
595 bool is_address_valid = ReceiveAddressToIPEndpoint(&address); 738 bool is_address_valid = ReceiveAddressToIPEndpoint(&address);
596 net_log_.AddEvent( 739 net_log_.AddEvent(
597 NetLog::TYPE_UDP_BYTES_RECEIVED, 740 NetLog::TYPE_UDP_BYTES_RECEIVED,
598 CreateNetLogUDPDataTranferCallback( 741 CreateNetLogUDPDataTranferCallback(
599 result, bytes, 742 result, bytes,
600 is_address_valid ? &address : NULL)); 743 is_address_valid ? &address : NULL));
601 } 744 }
602 745
603 base::StatsCounter read_bytes("udp.read_bytes"); 746 base::StatsCounter read_bytes("udp.read_bytes");
604 read_bytes.Add(result); 747 read_bytes.Add(result);
605 NetworkActivityMonitor::GetInstance()->IncrementBytesReceived(result); 748 NetworkActivityMonitor::GetInstance()->IncrementBytesReceived(result);
606 } 749 }
607 750
608 void UDPSocketWin::DidCompleteWrite() {
609 DWORD num_bytes, flags;
610 BOOL ok = WSAGetOverlappedResult(socket_, &core_->write_overlapped_,
611 &num_bytes, FALSE, &flags);
612 WSAResetEvent(core_->write_overlapped_.hEvent);
613 int result = ok ? num_bytes : MapSystemError(WSAGetLastError());
614 LogWrite(result, core_->write_iobuffer_->data(), send_to_address_.get());
615
616 send_to_address_.reset();
617 core_->write_iobuffer_ = NULL;
618 DoWriteCallback(result);
619 }
620
621 void UDPSocketWin::LogWrite(int result, 751 void UDPSocketWin::LogWrite(int result,
622 const char* bytes, 752 const char* bytes,
623 const IPEndPoint* address) const { 753 const IPEndPoint* address) const {
624 if (result < 0) { 754 if (result < 0) {
625 net_log_.AddEventWithNetErrorCode(NetLog::TYPE_UDP_SEND_ERROR, result); 755 net_log_.AddEventWithNetErrorCode(NetLog::TYPE_UDP_SEND_ERROR, result);
626 return; 756 return;
627 } 757 }
628 758
629 if (net_log_.IsLogging()) { 759 if (net_log_.IsLogging()) {
630 net_log_.AddEvent( 760 net_log_.AddEvent(
631 NetLog::TYPE_UDP_BYTES_SENT, 761 NetLog::TYPE_UDP_BYTES_SENT,
632 CreateNetLogUDPDataTranferCallback(result, bytes, address)); 762 CreateNetLogUDPDataTranferCallback(result, bytes, address));
633 } 763 }
634 764
635 base::StatsCounter write_bytes("udp.write_bytes"); 765 base::StatsCounter write_bytes("udp.write_bytes");
636 write_bytes.Add(result); 766 write_bytes.Add(result);
637 NetworkActivityMonitor::GetInstance()->IncrementBytesSent(result); 767 NetworkActivityMonitor::GetInstance()->IncrementBytesSent(result);
638 } 768 }
639 769
640 int UDPSocketWin::InternalRecvFrom(IOBuffer* buf, int buf_len, 770 int UDPSocketWin::InternalRecvFromOverlapped(IOBuffer* buf,
641 IPEndPoint* address) { 771 int buf_len,
642 DCHECK(!core_->read_iobuffer_.get()); 772 IPEndPoint* address) {
643 SockaddrStorage& storage = core_->recv_addr_storage_; 773 DCHECK(!core_->read_iobuffer());
644 storage.addr_len = sizeof(storage.addr_storage); 774 SockaddrStorage* storage = core_->recv_addr_storage();
775 storage->addr_len = sizeof(storage->addr_storage);
645 776
646 WSABUF read_buffer; 777 WSABUF read_buffer;
647 read_buffer.buf = buf->data(); 778 read_buffer.buf = buf->data();
648 read_buffer.len = buf_len; 779 read_buffer.len = buf_len;
649 780
650 DWORD flags = 0; 781 DWORD flags = 0;
651 DWORD num; 782 DWORD num;
652 CHECK_NE(INVALID_SOCKET, socket_); 783 CHECK_NE(INVALID_SOCKET, socket_);
653 AssertEventNotSignaled(core_->read_overlapped_.hEvent); 784 AssertEventNotSignaled(core_->read_overlapped()->hEvent);
654 int rv = WSARecvFrom(socket_, &read_buffer, 1, &num, &flags, storage.addr, 785 int rv = WSARecvFrom(socket_, &read_buffer, 1, &num, &flags, storage->addr,
655 &storage.addr_len, &core_->read_overlapped_, NULL); 786 &storage->addr_len, core_->read_overlapped(), NULL);
656 if (rv == 0) { 787 if (rv == 0) {
657 if (ResetEventIfSignaled(core_->read_overlapped_.hEvent)) { 788 if (ResetEventIfSignaled(core_->read_overlapped()->hEvent)) {
658 int result = num; 789 int result = num;
659 // Convert address. 790 // Convert address.
660 if (address && result >= 0) { 791 if (address && result >= 0) {
661 if (!ReceiveAddressToIPEndpoint(address)) 792 if (!ReceiveAddressToIPEndpoint(address))
662 result = ERR_ADDRESS_INVALID; 793 result = ERR_ADDRESS_INVALID;
663 } 794 }
664 LogRead(result, buf->data()); 795 LogRead(result, buf->data());
665 return result; 796 return result;
666 } 797 }
667 } else { 798 } else {
668 int os_error = WSAGetLastError(); 799 int os_error = WSAGetLastError();
669 if (os_error != WSA_IO_PENDING) { 800 if (os_error != WSA_IO_PENDING) {
670 int result = MapSystemError(os_error); 801 int result = MapSystemError(os_error);
671 LogRead(result, NULL); 802 LogRead(result, NULL);
672 return result; 803 return result;
673 } 804 }
674 } 805 }
675 core_->WatchForRead(); 806 core_->WatchForRead();
676 core_->read_iobuffer_ = buf; 807 core_->set_read_iobuffer(buf, buf_len);
677 return ERR_IO_PENDING; 808 return ERR_IO_PENDING;
678 } 809 }
679 810
680 int UDPSocketWin::InternalSendTo(IOBuffer* buf, int buf_len, 811 int UDPSocketWin::InternalSendToOverlapped(IOBuffer* buf,
681 const IPEndPoint* address) { 812 int buf_len,
682 DCHECK(!core_->write_iobuffer_.get()); 813 const IPEndPoint* address) {
814 DCHECK(!core_->write_iobuffer());
683 SockaddrStorage storage; 815 SockaddrStorage storage;
684 struct sockaddr* addr = storage.addr; 816 struct sockaddr* addr = storage.addr;
685 // Convert address. 817 // Convert address.
686 if (!address) { 818 if (!address) {
687 addr = NULL; 819 addr = NULL;
688 storage.addr_len = 0; 820 storage.addr_len = 0;
689 } else { 821 } else {
690 if (!address->ToSockAddr(addr, &storage.addr_len)) { 822 if (!address->ToSockAddr(addr, &storage.addr_len)) {
691 int result = ERR_ADDRESS_INVALID; 823 int result = ERR_ADDRESS_INVALID;
692 LogWrite(result, NULL, NULL); 824 LogWrite(result, NULL, NULL);
693 return result; 825 return result;
694 } 826 }
695 } 827 }
696 828
697 WSABUF write_buffer; 829 WSABUF write_buffer;
698 write_buffer.buf = buf->data(); 830 write_buffer.buf = buf->data();
699 write_buffer.len = buf_len; 831 write_buffer.len = buf_len;
700 832
701 DWORD flags = 0; 833 DWORD flags = 0;
702 DWORD num; 834 DWORD num;
703 AssertEventNotSignaled(core_->write_overlapped_.hEvent); 835 AssertEventNotSignaled(core_->write_overlapped()->hEvent);
704 int rv = WSASendTo(socket_, &write_buffer, 1, &num, flags, 836 int rv = WSASendTo(socket_, &write_buffer, 1, &num, flags, addr,
705 addr, storage.addr_len, &core_->write_overlapped_, NULL); 837 storage.addr_len, core_->write_overlapped(), NULL);
706 if (rv == 0) { 838 if (rv == 0) {
707 if (ResetEventIfSignaled(core_->write_overlapped_.hEvent)) { 839 if (ResetEventIfSignaled(core_->write_overlapped()->hEvent)) {
708 int result = num; 840 int result = num;
709 LogWrite(result, buf->data(), address); 841 LogWrite(result, buf->data(), address);
710 return result; 842 return result;
711 } 843 }
712 } else { 844 } else {
713 int os_error = WSAGetLastError(); 845 int os_error = WSAGetLastError();
714 if (os_error != WSA_IO_PENDING) { 846 if (os_error != WSA_IO_PENDING) {
715 int result = MapSystemError(os_error); 847 int result = MapSystemError(os_error);
716 LogWrite(result, NULL, NULL); 848 LogWrite(result, NULL, NULL);
717 return result; 849 return result;
718 } 850 }
719 } 851 }
720 852
721 core_->WatchForWrite(); 853 core_->WatchForWrite();
722 core_->write_iobuffer_ = buf; 854 core_->set_write_iobuffer(buf, 0);
723 return ERR_IO_PENDING; 855 return ERR_IO_PENDING;
724 } 856 }
725 857
858 int UDPSocketWin::InternalRecvFromNonBlocking(IOBuffer* buf,
859 int buf_len,
860 IPEndPoint* address) {
861 if (!non_blocking_reads_initialized_) {
862 // After this call the event will be signaled when asynchronous read is
863 // completed.g
864 WSAEventSelect(socket_, core_->read_event(), FD_READ);
865 non_blocking_reads_initialized_ = true;
866 }
867 SockaddrStorage* storage = core_->recv_addr_storage();
868 storage->addr_len = sizeof(storage->addr_storage);
869
870 CHECK_NE(INVALID_SOCKET, socket_);
871 int rv = recvfrom(socket_, buf->data(), buf_len, 0, storage->addr,
872 &storage->addr_len);
873 if (rv == SOCKET_ERROR) {
874 int os_error = WSAGetLastError();
875 if (os_error == WSAEWOULDBLOCK) {
876 core_->WatchForRead();
877 core_->set_read_iobuffer(buf, buf_len);
878 return ERR_IO_PENDING;
879 } else {
880 rv = MapSystemError(os_error);
881 LogRead(rv, NULL);
882 return rv;
883 }
884 } else {
885 // Convert address.
886 if (address && rv >= 0 && !ReceiveAddressToIPEndpoint(address)) {
887 rv = ERR_ADDRESS_INVALID;
888 }
889 LogRead(rv, buf->data());
890 return rv;
891 }
892 }
893
894 int UDPSocketWin::InternalSendToNonBlocking(IOBuffer* buf,
895 int buf_len,
896 const IPEndPoint* address) {
897 if (!non_blocking_writes_initialized_) {
898 // After this call the event will be signaled when asynchronous write is
899 // completed.
900 WSAEventSelect(socket_, core_->write_event(), FD_WRITE);
901 non_blocking_writes_initialized_ = true;
902 }
903 SockaddrStorage storage;
904 struct sockaddr* addr = storage.addr;
905 // Convert address.
906 if (address) {
907 if (!address->ToSockAddr(addr, &storage.addr_len)) {
908 int result = ERR_ADDRESS_INVALID;
909 LogWrite(result, NULL, NULL);
910 return result;
911 }
912 } else {
913 addr = NULL;
914 storage.addr_len = 0;
915 }
916
917 int rv = sendto(socket_, buf->data(), buf_len, 0, addr, storage.addr_len);
918 if (rv == SOCKET_ERROR) {
919 int os_error = WSAGetLastError();
920 if (os_error == WSAEWOULDBLOCK) {
921 core_->WatchForWrite();
922 core_->set_write_iobuffer(buf, buf_len);
923 return ERR_IO_PENDING;
924 } else {
925 rv = MapSystemError(os_error);
926 LogWrite(rv, NULL, NULL);
927 return rv;
928 }
929 } else {
930 LogWrite(rv, buf->data(), address);
931 return rv;
932 }
933 }
934
726 int UDPSocketWin::SetMulticastOptions() { 935 int UDPSocketWin::SetMulticastOptions() {
727 if (!(socket_options_ & SOCKET_OPTION_MULTICAST_LOOP)) { 936 if (!(socket_options_ & SOCKET_OPTION_MULTICAST_LOOP)) {
728 DWORD loop = 0; 937 DWORD loop = 0;
729 int protocol_level = 938 int protocol_level =
730 addr_family_ == AF_INET ? IPPROTO_IP : IPPROTO_IPV6; 939 addr_family_ == AF_INET ? IPPROTO_IP : IPPROTO_IPV6;
731 int option = 940 int option =
732 addr_family_ == AF_INET ? IP_MULTICAST_LOOP: IPV6_MULTICAST_LOOP; 941 addr_family_ == AF_INET ? IP_MULTICAST_LOOP: IPV6_MULTICAST_LOOP;
733 int rv = setsockopt(socket_, protocol_level, option, 942 int rv = setsockopt(socket_, protocol_level, option,
734 reinterpret_cast<const char*>(&loop), sizeof(loop)); 943 reinterpret_cast<const char*>(&loop), sizeof(loop));
735 if (rv < 0) 944 if (rv < 0)
(...skipping 65 matching lines...) Expand 10 before | Expand all | Expand 10 after
801 for (int i = 0; i < kBindRetries; ++i) { 1010 for (int i = 0; i < kBindRetries; ++i) {
802 int rv = DoBind(IPEndPoint( 1011 int rv = DoBind(IPEndPoint(
803 address, static_cast<uint16>(rand_int_cb_.Run(kPortStart, kPortEnd)))); 1012 address, static_cast<uint16>(rand_int_cb_.Run(kPortStart, kPortEnd))));
804 if (rv == OK || rv != ERR_ADDRESS_IN_USE) 1013 if (rv == OK || rv != ERR_ADDRESS_IN_USE)
805 return rv; 1014 return rv;
806 } 1015 }
807 return DoBind(IPEndPoint(address, 0)); 1016 return DoBind(IPEndPoint(address, 0));
808 } 1017 }
809 1018
810 bool UDPSocketWin::ReceiveAddressToIPEndpoint(IPEndPoint* address) const { 1019 bool UDPSocketWin::ReceiveAddressToIPEndpoint(IPEndPoint* address) const {
811 SockaddrStorage& storage = core_->recv_addr_storage_; 1020 SockaddrStorage* storage = core_->recv_addr_storage();
812 return address->FromSockAddr(storage.addr, storage.addr_len); 1021 return address->FromSockAddr(storage->addr, storage->addr_len);
813 } 1022 }
814 1023
815 int UDPSocketWin::JoinGroup( 1024 int UDPSocketWin::JoinGroup(
816 const IPAddressNumber& group_address) const { 1025 const IPAddressNumber& group_address) const {
817 DCHECK(CalledOnValidThread()); 1026 DCHECK(CalledOnValidThread());
818 if (!is_connected()) 1027 if (!is_connected())
819 return ERR_SOCKET_NOT_CONNECTED; 1028 return ERR_SOCKET_NOT_CONNECTED;
820 1029
821 switch (group_address.size()) { 1030 switch (group_address.size()) {
822 case kIPv4AddressSize: { 1031 case kIPv4AddressSize: {
(...skipping 186 matching lines...) Expand 10 before | Expand all | Expand 10 after
1009 0, 1218 0,
1010 NULL); 1219 NULL);
1011 1220
1012 return OK; 1221 return OK;
1013 } 1222 }
1014 1223
1015 void UDPSocketWin::DetachFromThread() { 1224 void UDPSocketWin::DetachFromThread() {
1016 base::NonThreadSafe::DetachFromThread(); 1225 base::NonThreadSafe::DetachFromThread();
1017 } 1226 }
1018 1227
1228 void UDPSocketWin::UseNonBlockingIO() {
1229 if (core_) {
1230 NOTREACHED() << "Cannot change to non-blocking mode after socket is used.";
1231 } else {
1232 use_overlapped_io_ = false;
1233 }
1234 }
1235
1019 } // namespace net 1236 } // namespace net
OLDNEW
« net/udp/udp_socket_perftest.cc ('K') | « net/udp/udp_socket_win.h ('k') | no next file » | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698