libgo/go/net/fd_windows.go

   1 // Copyright 2010 The Go Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style
   3 // license that can be found in the LICENSE file.
   4
   5 package net
   6
   7 import (
   8         "errors"
   9         "io"
  10         "os"
  11         "runtime"
  12         "sync"
  13         "syscall"
  14         "time"
  15         "unsafe"
  16 )
  17
  18 var initErr error
  19
  20 // CancelIo Windows API cancels all outstanding IO for a particular
  21 // socket on current thread. To overcome that limitation, we run
  22 // special goroutine, locked to OS single thread, that both starts
  23 // and cancels IO. It means, there are 2 unavoidable thread switches
  24 // for every IO.
  25 // Some newer versions of Windows has new CancelIoEx API, that does
  26 // not have that limitation and can be used from any thread. This
  27 // package uses CancelIoEx API, if present, otherwise it fallback
  28 // to CancelIo.
  29
  30 var canCancelIO bool // determines if CancelIoEx API is present
  31
  32 func sysInit() {
  33         var d syscall.WSAData
  34         e := syscall.WSAStartup(uint32(0x202), &d)
  35         if e != nil {
  36                 initErr = os.NewSyscallError("WSAStartup", e)
  37         }
  38         canCancelIO = syscall.LoadCancelIoEx() == nil
  39         if syscall.LoadGetAddrInfo() == nil {
  40                 lookupIP = newLookupIP
  41         }
  42 }
  43
  44 func closesocket(s syscall.Handle) error {
  45         return syscall.Closesocket(s)
  46 }
  47
  48 func canUseConnectEx(net string) bool {
  49         if net == "udp" || net == "udp4" || net == "udp6" {
  50                 // ConnectEx windows API does not support connectionless sockets.
  51                 return false
  52         }
  53         return syscall.LoadConnectEx() == nil
  54 }
  55
  56 func dialTimeout(net, addr string, timeout time.Duration) (Conn, error) {
  57         if !canUseConnectEx(net) {
  58                 // Use the relatively inefficient goroutine-racing
  59                 // implementation of DialTimeout.
  60                 return dialTimeoutRace(net, addr, timeout)
  61         }
  62         deadline := time.Now().Add(timeout)
  63         _, addri, err := resolveNetAddr("dial", net, addr, deadline)
  64         if err != nil {
  65                 return nil, err
  66         }
  67         return dialAddr(net, addr, addri, deadline)
  68 }
  69
  70 // Interface for all IO operations.
  71 type anOpIface interface {
  72         Op() *anOp
  73         Name() string
  74         Submit() error
  75 }
  76
  77 // IO completion result parameters.
  78 type ioResult struct {
  79         qty uint32
  80         err error
  81 }
  82
  83 // anOp implements functionality common to all IO operations.
  84 type anOp struct {
  85         // Used by IOCP interface, it must be first field
  86         // of the struct, as our code rely on it.
  87         o syscall.Overlapped
  88
  89         resultc chan ioResult
  90         errnoc  chan error
  91         fd      *netFD
  92 }
  93
  94 func (o *anOp) Init(fd *netFD, mode int) {
  95         o.fd = fd
  96         var i int
  97         if mode == 'r' {
  98                 i = 0
  99         } else {
 100                 i = 1
 101         }
 102         if fd.resultc[i] == nil {
 103                 fd.resultc[i] = make(chan ioResult, 1)
 104         }
 105         o.resultc = fd.resultc[i]
 106         if fd.errnoc[i] == nil {
 107                 fd.errnoc[i] = make(chan error)
 108         }
 109         o.errnoc = fd.errnoc[i]
 110 }
 111
 112 func (o *anOp) Op() *anOp {
 113         return o
 114 }
 115
 116 // bufOp is used by IO operations that read / write
 117 // data from / to client buffer.
 118 type bufOp struct {
 119         anOp
 120         buf syscall.WSABuf
 121 }
 122
 123 func (o *bufOp) Init(fd *netFD, buf []byte, mode int) {
 124         o.anOp.Init(fd, mode)
 125         o.buf.Len = uint32(len(buf))
 126         if len(buf) == 0 {
 127                 o.buf.Buf = nil
 128         } else {
 129                 o.buf.Buf = (*byte)(unsafe.Pointer(&buf[0]))
 130         }
 131 }
 132
 133 // resultSrv will retrieve all IO completion results from
 134 // iocp and send them to the correspondent waiting client
 135 // goroutine via channel supplied in the request.
 136 type resultSrv struct {
 137         iocp syscall.Handle
 138 }
 139
 140 func (s *resultSrv) Run() {
 141         var o *syscall.Overlapped
 142         var key uint32
 143         var r ioResult
 144         for {
 145                 r.err = syscall.GetQueuedCompletionStatus(s.iocp, &(r.qty), &key, &o, syscall.INFINITE)
 146                 switch {
 147                 case r.err == nil:
 148                         // Dequeued successfully completed IO packet.
 149                 case r.err == syscall.Errno(syscall.WAIT_TIMEOUT) && o == nil:
 150                         // Wait has timed out (should not happen now, but might be used in the future).
 151                         panic("GetQueuedCompletionStatus timed out")
 152                 case o == nil:
 153                         // Failed to dequeue anything -> report the error.
 154                         panic("GetQueuedCompletionStatus failed " + r.err.Error())
 155                 default:
 156                         // Dequeued failed IO packet.
 157                 }
 158                 (*anOp)(unsafe.Pointer(o)).resultc <- r
 159         }
 160 }
 161
 162 // ioSrv executes net IO requests.
 163 type ioSrv struct {
 164         submchan chan anOpIface // submit IO requests
 165         canchan  chan anOpIface // cancel IO requests
 166 }
 167
 168 // ProcessRemoteIO will execute submit IO requests on behalf
 169 // of other goroutines, all on a single os thread, so it can
 170 // cancel them later. Results of all operations will be sent
 171 // back to their requesters via channel supplied in request.
 172 // It is used only when the CancelIoEx API is unavailable.
 173 func (s *ioSrv) ProcessRemoteIO() {
 174         runtime.LockOSThread()
 175         defer runtime.UnlockOSThread()
 176         for {
 177                 select {
 178                 case o := <-s.submchan:
 179                         o.Op().errnoc <- o.Submit()
 180                 case o := <-s.canchan:
 181                         o.Op().errnoc <- syscall.CancelIo(syscall.Handle(o.Op().fd.sysfd))
 182                 }
 183         }
 184 }
 185
 186 // ExecIO executes a single IO operation oi. It submits and cancels
 187 // IO in the current thread for systems where Windows CancelIoEx API
 188 // is available. Alternatively, it passes the request onto
 189 // a special goroutine and waits for completion or cancels request.
 190 // deadline is unix nanos.
 191 func (s *ioSrv) ExecIO(oi anOpIface, deadline int64) (int, error) {
 192         var err error
 193         o := oi.Op()
 194         // Calculate timeout delta.
 195         var delta int64
 196         if deadline != 0 {
 197                 delta = deadline - time.Now().UnixNano()
 198                 if delta <= 0 {
 199                         return 0, &OpError{oi.Name(), o.fd.net, o.fd.laddr, errTimeout}
 200                 }
 201         }
 202         // Start IO.
 203         if canCancelIO {
 204                 err = oi.Submit()
 205         } else {
 206                 // Send request to a special dedicated thread,
 207                 // so it can stop the IO with CancelIO later.
 208                 s.submchan <- oi
 209                 err = <-o.errnoc
 210         }
 211         switch err {
 212         case nil:
 213                 // IO completed immediately, but we need to get our completion message anyway.
 214         case syscall.ERROR_IO_PENDING:
 215                 // IO started, and we have to wait for its completion.
 216                 err = nil
 217         default:
 218                 return 0, &OpError{oi.Name(), o.fd.net, o.fd.laddr, err}
 219         }
 220         // Setup timer, if deadline is given.
 221         var timer <-chan time.Time
 222         if delta > 0 {
 223                 t := time.NewTimer(time.Duration(delta) * time.Nanosecond)
 224                 defer t.Stop()
 225                 timer = t.C
 226         }
 227         // Wait for our request to complete.
 228         var r ioResult
 229         var cancelled, timeout bool
 230         select {
 231         case r = <-o.resultc:
 232         case <-timer:
 233                 cancelled = true
 234                 timeout = true
 235         case <-o.fd.closec:
 236                 cancelled = true
 237         }
 238         if cancelled {
 239                 // Cancel it.
 240                 if canCancelIO {
 241                         err := syscall.CancelIoEx(syscall.Handle(o.Op().fd.sysfd), &o.o)
 242                         // Assuming ERROR_NOT_FOUND is returned, if IO is completed.
 243                         if err != nil && err != syscall.ERROR_NOT_FOUND {
 244                                 // TODO(brainman): maybe do something else, but panic.
 245                                 panic(err)
 246                         }
 247                 } else {
 248                         s.canchan <- oi
 249                         <-o.errnoc
 250                 }
 251                 // Wait for IO to be canceled or complete successfully.
 252                 r = <-o.resultc
 253                 if r.err == syscall.ERROR_OPERATION_ABORTED { // IO Canceled
 254                         if timeout {
 255                                 r.err = errTimeout
 256                         } else {
 257                                 r.err = errClosing
 258                         }
 259                 }
 260         }
 261         if r.err != nil {
 262                 err = &OpError{oi.Name(), o.fd.net, o.fd.laddr, r.err}
 263         }
 264         return int(r.qty), err
 265 }
 266
 267 // Start helper goroutines.
 268 var resultsrv *resultSrv
 269 var iosrv *ioSrv
 270 var onceStartServer sync.Once
 271
 272 func startServer() {
 273         resultsrv = new(resultSrv)
 274         var err error
 275         resultsrv.iocp, err = syscall.CreateIoCompletionPort(syscall.InvalidHandle, 0, 0, 1)
 276         if err != nil {
 277                 panic("CreateIoCompletionPort: " + err.Error())
 278         }
 279         go resultsrv.Run()
 280
 281         iosrv = new(ioSrv)
 282         if !canCancelIO {
 283                 // Only CancelIo API is available. Lets start special goroutine
 284                 // locked to an OS thread, that both starts and cancels IO.
 285                 iosrv.submchan = make(chan anOpIface)
 286                 iosrv.canchan = make(chan anOpIface)
 287                 go iosrv.ProcessRemoteIO()
 288         }
 289 }
 290
 291 // Network file descriptor.
 292 type netFD struct {
 293         // locking/lifetime of sysfd
 294         sysmu   sync.Mutex
 295         sysref  int
 296         closing bool
 297
 298         // immutable until Close
 299         sysfd       syscall.Handle
 300         family      int
 301         sotype      int
 302         isConnected bool
 303         net         string
 304         laddr       Addr
 305         raddr       Addr
 306         resultc     [2]chan ioResult // read/write completion results
 307         errnoc      [2]chan error    // read/write submit or cancel operation errors
 308         closec      chan bool        // used by Close to cancel pending IO
 309
 310         // serialize access to Read and Write methods
 311         rio, wio sync.Mutex
 312
 313         // read and write deadlines
 314         rdeadline, wdeadline deadline
 315 }
 316
 317 func allocFD(fd syscall.Handle, family, sotype int, net string) *netFD {
 318         netfd := &netFD{
 319                 sysfd:  fd,
 320                 family: family,
 321                 sotype: sotype,
 322                 net:    net,
 323                 closec: make(chan bool),
 324         }
 325         return netfd
 326 }
 327
 328 func newFD(fd syscall.Handle, family, proto int, net string) (*netFD, error) {
 329         if initErr != nil {
 330                 return nil, initErr
 331         }
 332         onceStartServer.Do(startServer)
 333         // Associate our socket with resultsrv.iocp.
 334         if _, err := syscall.CreateIoCompletionPort(syscall.Handle(fd), resultsrv.iocp, 0, 0); err != nil {
 335                 return nil, err
 336         }
 337         return allocFD(fd, family, proto, net), nil
 338 }
 339
 340 func (fd *netFD) setAddr(laddr, raddr Addr) {
 341         fd.laddr = laddr
 342         fd.raddr = raddr
 343         runtime.SetFinalizer(fd, (*netFD).closesocket)
 344 }
 345
 346 // Make new connection.
 347
 348 type connectOp struct {
 349         anOp
 350         ra syscall.Sockaddr
 351 }
 352
 353 func (o *connectOp) Submit() error {
 354         return syscall.ConnectEx(o.fd.sysfd, o.ra, nil, 0, nil, &o.o)
 355 }
 356
 357 func (o *connectOp) Name() string {
 358         return "ConnectEx"
 359 }
 360
 361 func (fd *netFD) connect(ra syscall.Sockaddr) error {
 362         if !canUseConnectEx(fd.net) {
 363                 return syscall.Connect(fd.sysfd, ra)
 364         }
 365         // ConnectEx windows API requires an unconnected, previously bound socket.
 366         var la syscall.Sockaddr
 367         switch ra.(type) {
 368         case *syscall.SockaddrInet4:
 369                 la = &syscall.SockaddrInet4{}
 370         case *syscall.SockaddrInet6:
 371                 la = &syscall.SockaddrInet6{}
 372         default:
 373                 panic("unexpected type in connect")
 374         }
 375         if err := syscall.Bind(fd.sysfd, la); err != nil {
 376                 return err
 377         }
 378         // Call ConnectEx API.
 379         var o connectOp
 380         o.Init(fd, 'w')
 381         o.ra = ra
 382         _, err := iosrv.ExecIO(&o, fd.wdeadline.value())
 383         if err != nil {
 384                 return err
 385         }
 386         // Refresh socket properties.
 387         return syscall.Setsockopt(fd.sysfd, syscall.SOL_SOCKET, syscall.SO_UPDATE_CONNECT_CONTEXT, (*byte)(unsafe.Pointer(&fd.sysfd)), int32(unsafe.Sizeof(fd.sysfd)))
 388 }
 389
 390 // Add a reference to this fd.
 391 // If closing==true, mark the fd as closing.
 392 // Returns an error if the fd cannot be used.
 393 func (fd *netFD) incref(closing bool) error {
 394         if fd == nil {
 395                 return errClosing
 396         }
 397         fd.sysmu.Lock()
 398         if fd.closing {
 399                 fd.sysmu.Unlock()
 400                 return errClosing
 401         }
 402         fd.sysref++
 403         if closing {
 404                 fd.closing = true
 405         }
 406         closing = fd.closing
 407         fd.sysmu.Unlock()
 408         return nil
 409 }
 410
 411 // Remove a reference to this FD and close if we've been asked to do so (and
 412 // there are no references left.
 413 func (fd *netFD) decref() {
 414         if fd == nil {
 415                 return
 416         }
 417         fd.sysmu.Lock()
 418         fd.sysref--
 419         if fd.closing && fd.sysref == 0 && fd.sysfd != syscall.InvalidHandle {
 420                 closesocket(fd.sysfd)
 421                 fd.sysfd = syscall.InvalidHandle
 422                 // no need for a finalizer anymore
 423                 runtime.SetFinalizer(fd, nil)
 424         }
 425         fd.sysmu.Unlock()
 426 }
 427
 428 func (fd *netFD) Close() error {
 429         if err := fd.incref(true); err != nil {
 430                 return err
 431         }
 432         defer fd.decref()
 433         // unblock pending reader and writer
 434         close(fd.closec)
 435         // wait for both reader and writer to exit
 436         fd.rio.Lock()
 437         defer fd.rio.Unlock()
 438         fd.wio.Lock()
 439         defer fd.wio.Unlock()
 440         return nil
 441 }
 442
 443 func (fd *netFD) shutdown(how int) error {
 444         if err := fd.incref(false); err != nil {
 445                 return err
 446         }
 447         defer fd.decref()
 448         err := syscall.Shutdown(fd.sysfd, how)
 449         if err != nil {
 450                 return &OpError{"shutdown", fd.net, fd.laddr, err}
 451         }
 452         return nil
 453 }
 454
 455 func (fd *netFD) CloseRead() error {
 456         return fd.shutdown(syscall.SHUT_RD)
 457 }
 458
 459 func (fd *netFD) CloseWrite() error {
 460         return fd.shutdown(syscall.SHUT_WR)
 461 }
 462
 463 func (fd *netFD) closesocket() error {
 464         return closesocket(fd.sysfd)
 465 }
 466
 467 // Read from network.
 468
 469 type readOp struct {
 470         bufOp
 471 }
 472
 473 func (o *readOp) Submit() error {
 474         var d, f uint32
 475         return syscall.WSARecv(syscall.Handle(o.fd.sysfd), &o.buf, 1, &d, &f, &o.o, nil)
 476 }
 477
 478 func (o *readOp) Name() string {
 479         return "WSARecv"
 480 }
 481
 482 func (fd *netFD) Read(buf []byte) (int, error) {
 483         if err := fd.incref(false); err != nil {
 484                 return 0, err
 485         }
 486         defer fd.decref()
 487         fd.rio.Lock()
 488         defer fd.rio.Unlock()
 489         var o readOp
 490         o.Init(fd, buf, 'r')
 491         n, err := iosrv.ExecIO(&o, fd.rdeadline.value())
 492         if err == nil && n == 0 {
 493                 err = io.EOF
 494         }
 495         return n, err
 496 }
 497
 498 // ReadFrom from network.
 499
 500 type readFromOp struct {
 501         bufOp
 502         rsa  syscall.RawSockaddrAny
 503         rsan int32
 504 }
 505
 506 func (o *readFromOp) Submit() error {
 507         var d, f uint32
 508         return syscall.WSARecvFrom(o.fd.sysfd, &o.buf, 1, &d, &f, &o.rsa, &o.rsan, &o.o, nil)
 509 }
 510
 511 func (o *readFromOp) Name() string {
 512         return "WSARecvFrom"
 513 }
 514
 515 func (fd *netFD) ReadFrom(buf []byte) (n int, sa syscall.Sockaddr, err error) {
 516         if len(buf) == 0 {
 517                 return 0, nil, nil
 518         }
 519         if err := fd.incref(false); err != nil {
 520                 return 0, nil, err
 521         }
 522         defer fd.decref()
 523         fd.rio.Lock()
 524         defer fd.rio.Unlock()
 525         var o readFromOp
 526         o.Init(fd, buf, 'r')
 527         o.rsan = int32(unsafe.Sizeof(o.rsa))
 528         n, err = iosrv.ExecIO(&o, fd.rdeadline.value())
 529         if err != nil {
 530                 return 0, nil, err
 531         }
 532         sa, _ = o.rsa.Sockaddr()
 533         return
 534 }
 535
 536 // Write to network.
 537
 538 type writeOp struct {
 539         bufOp
 540 }
 541
 542 func (o *writeOp) Submit() error {
 543         var d uint32
 544         return syscall.WSASend(o.fd.sysfd, &o.buf, 1, &d, 0, &o.o, nil)
 545 }
 546
 547 func (o *writeOp) Name() string {
 548         return "WSASend"
 549 }
 550
 551 func (fd *netFD) Write(buf []byte) (int, error) {
 552         if err := fd.incref(false); err != nil {
 553                 return 0, err
 554         }
 555         defer fd.decref()
 556         fd.wio.Lock()
 557         defer fd.wio.Unlock()
 558         var o writeOp
 559         o.Init(fd, buf, 'w')
 560         return iosrv.ExecIO(&o, fd.wdeadline.value())
 561 }
 562
 563 // WriteTo to network.
 564
 565 type writeToOp struct {
 566         bufOp
 567         sa syscall.Sockaddr
 568 }
 569
 570 func (o *writeToOp) Submit() error {
 571         var d uint32
 572         return syscall.WSASendto(o.fd.sysfd, &o.buf, 1, &d, 0, o.sa, &o.o, nil)
 573 }
 574
 575 func (o *writeToOp) Name() string {
 576         return "WSASendto"
 577 }
 578
 579 func (fd *netFD) WriteTo(buf []byte, sa syscall.Sockaddr) (int, error) {
 580         if len(buf) == 0 {
 581                 return 0, nil
 582         }
 583         if err := fd.incref(false); err != nil {
 584                 return 0, err
 585         }
 586         defer fd.decref()
 587         fd.wio.Lock()
 588         defer fd.wio.Unlock()
 589         var o writeToOp
 590         o.Init(fd, buf, 'w')
 591         o.sa = sa
 592         return iosrv.ExecIO(&o, fd.wdeadline.value())
 593 }
 594
 595 // Accept new network connections.
 596
 597 type acceptOp struct {
 598         anOp
 599         newsock syscall.Handle
 600         attrs   [2]syscall.RawSockaddrAny // space for local and remote address only
 601 }
 602
 603 func (o *acceptOp) Submit() error {
 604         var d uint32
 605         l := uint32(unsafe.Sizeof(o.attrs[0]))
 606         return syscall.AcceptEx(o.fd.sysfd, o.newsock,
 607                 (*byte)(unsafe.Pointer(&o.attrs[0])), 0, l, l, &d, &o.o)
 608 }
 609
 610 func (o *acceptOp) Name() string {
 611         return "AcceptEx"
 612 }
 613
 614 func (fd *netFD) accept(toAddr func(syscall.Sockaddr) Addr) (*netFD, error) {
 615         if err := fd.incref(false); err != nil {
 616                 return nil, err
 617         }
 618         defer fd.decref()
 619
 620         // Get new socket.
 621         // See ../syscall/exec_unix.go for description of ForkLock.
 622         syscall.ForkLock.RLock()
 623         s, err := syscall.Socket(fd.family, fd.sotype, 0)
 624         if err != nil {
 625                 syscall.ForkLock.RUnlock()
 626                 return nil, &OpError{"socket", fd.net, fd.laddr, err}
 627         }
 628         syscall.CloseOnExec(s)
 629         syscall.ForkLock.RUnlock()
 630
 631         // Associate our new socket with IOCP.
 632         onceStartServer.Do(startServer)
 633         if _, err := syscall.CreateIoCompletionPort(s, resultsrv.iocp, 0, 0); err != nil {
 634                 closesocket(s)
 635                 return nil, &OpError{"CreateIoCompletionPort", fd.net, fd.laddr, err}
 636         }
 637
 638         // Submit accept request.
 639         var o acceptOp
 640         o.Init(fd, 'r')
 641         o.newsock = s
 642         _, err = iosrv.ExecIO(&o, fd.rdeadline.value())
 643         if err != nil {
 644                 closesocket(s)
 645                 return nil, err
 646         }
 647
 648         // Inherit properties of the listening socket.
 649         err = syscall.Setsockopt(s, syscall.SOL_SOCKET, syscall.SO_UPDATE_ACCEPT_CONTEXT, (*byte)(unsafe.Pointer(&fd.sysfd)), int32(unsafe.Sizeof(fd.sysfd)))
 650         if err != nil {
 651                 closesocket(s)
 652                 return nil, &OpError{"Setsockopt", fd.net, fd.laddr, err}
 653         }
 654
 655         // Get local and peer addr out of AcceptEx buffer.
 656         var lrsa, rrsa *syscall.RawSockaddrAny
 657         var llen, rlen int32
 658         l := uint32(unsafe.Sizeof(*lrsa))
 659         syscall.GetAcceptExSockaddrs((*byte)(unsafe.Pointer(&o.attrs[0])),
 660                 0, l, l, &lrsa, &llen, &rrsa, &rlen)
 661         lsa, _ := lrsa.Sockaddr()
 662         rsa, _ := rrsa.Sockaddr()
 663
 664         netfd := allocFD(s, fd.family, fd.sotype, fd.net)
 665         netfd.setAddr(toAddr(lsa), toAddr(rsa))
 666         return netfd, nil
 667 }
 668
 669 // Unimplemented functions.
 670
 671 func (fd *netFD) dup() (*os.File, error) {
 672         // TODO: Implement this
 673         return nil, os.NewSyscallError("dup", syscall.EWINDOWS)
 674 }
 675
 676 var errNoSupport = errors.New("address family not supported")
 677
 678 func (fd *netFD) ReadMsg(p []byte, oob []byte) (n, oobn, flags int, sa syscall.Sockaddr, err error) {
 679         return 0, 0, 0, nil, errNoSupport
 680 }
 681
 682 func (fd *netFD) WriteMsg(p []byte, oob []byte, sa syscall.Sockaddr) (n int, oobn int, err error) {
 683         return 0, 0, errNoSupport
 684 }