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 func init() {
  21         var d syscall.WSAData
  22         e := syscall.WSAStartup(uint32(0x202), &d)
  23         if e != nil {
  24                 initErr = os.NewSyscallError("WSAStartup", e)
  25         }
  26 }
  27
  28 func closesocket(s syscall.Handle) error {
  29         return syscall.Closesocket(s)
  30 }
  31
  32 // Interface for all io operations.
  33 type anOpIface interface {
  34         Op() *anOp
  35         Name() string
  36         Submit() error
  37 }
  38
  39 // IO completion result parameters.
  40 type ioResult struct {
  41         qty uint32
  42         err error
  43 }
  44
  45 // anOp implements functionality common to all io operations.
  46 type anOp struct {
  47         // Used by IOCP interface, it must be first field
  48         // of the struct, as our code rely on it.
  49         o syscall.Overlapped
  50
  51         resultc chan ioResult
  52         errnoc  chan error
  53         fd      *netFD
  54 }
  55
  56 func (o *anOp) Init(fd *netFD, mode int) {
  57         o.fd = fd
  58         var i int
  59         if mode == 'r' {
  60                 i = 0
  61         } else {
  62                 i = 1
  63         }
  64         if fd.resultc[i] == nil {
  65                 fd.resultc[i] = make(chan ioResult, 1)
  66         }
  67         o.resultc = fd.resultc[i]
  68         if fd.errnoc[i] == nil {
  69                 fd.errnoc[i] = make(chan error)
  70         }
  71         o.errnoc = fd.errnoc[i]
  72 }
  73
  74 func (o *anOp) Op() *anOp {
  75         return o
  76 }
  77
  78 // bufOp is used by io operations that read / write
  79 // data from / to client buffer.
  80 type bufOp struct {
  81         anOp
  82         buf syscall.WSABuf
  83 }
  84
  85 func (o *bufOp) Init(fd *netFD, buf []byte, mode int) {
  86         o.anOp.Init(fd, mode)
  87         o.buf.Len = uint32(len(buf))
  88         if len(buf) == 0 {
  89                 o.buf.Buf = nil
  90         } else {
  91                 o.buf.Buf = (*byte)(unsafe.Pointer(&buf[0]))
  92         }
  93 }
  94
  95 // resultSrv will retrieve all io completion results from
  96 // iocp and send them to the correspondent waiting client
  97 // goroutine via channel supplied in the request.
  98 type resultSrv struct {
  99         iocp syscall.Handle
 100 }
 101
 102 func (s *resultSrv) Run() {
 103         var o *syscall.Overlapped
 104         var key uint32
 105         var r ioResult
 106         for {
 107                 r.err = syscall.GetQueuedCompletionStatus(s.iocp, &(r.qty), &key, &o, syscall.INFINITE)
 108                 switch {
 109                 case r.err == nil:
 110                         // Dequeued successfully completed io packet.
 111                 case r.err == syscall.Errno(syscall.WAIT_TIMEOUT) && o == nil:
 112                         // Wait has timed out (should not happen now, but might be used in the future).
 113                         panic("GetQueuedCompletionStatus timed out")
 114                 case o == nil:
 115                         // Failed to dequeue anything -> report the error.
 116                         panic("GetQueuedCompletionStatus failed " + r.err.Error())
 117                 default:
 118                         // Dequeued failed io packet.
 119                 }
 120                 (*anOp)(unsafe.Pointer(o)).resultc <- r
 121         }
 122 }
 123
 124 // ioSrv executes net io requests.
 125 type ioSrv struct {
 126         submchan chan anOpIface // submit io requests
 127         canchan  chan anOpIface // cancel io requests
 128 }
 129
 130 // ProcessRemoteIO will execute submit io requests on behalf
 131 // of other goroutines, all on a single os thread, so it can
 132 // cancel them later. Results of all operations will be sent
 133 // back to their requesters via channel supplied in request.
 134 func (s *ioSrv) ProcessRemoteIO() {
 135         runtime.LockOSThread()
 136         defer runtime.UnlockOSThread()
 137         for {
 138                 select {
 139                 case o := <-s.submchan:
 140                         o.Op().errnoc <- o.Submit()
 141                 case o := <-s.canchan:
 142                         o.Op().errnoc <- syscall.CancelIo(syscall.Handle(o.Op().fd.sysfd))
 143                 }
 144         }
 145 }
 146
 147 // ExecIO executes a single io operation. It either executes it
 148 // inline, or, if a deadline is employed, passes the request onto
 149 // a special goroutine and waits for completion or cancels request.
 150 // deadline is unix nanos.
 151 func (s *ioSrv) ExecIO(oi anOpIface, deadline int64) (int, error) {
 152         var err error
 153         o := oi.Op()
 154         if deadline != 0 {
 155                 // Send request to a special dedicated thread,
 156                 // so it can stop the io with CancelIO later.
 157                 s.submchan <- oi
 158                 err = <-o.errnoc
 159         } else {
 160                 err = oi.Submit()
 161         }
 162         switch err {
 163         case nil:
 164                 // IO completed immediately, but we need to get our completion message anyway.
 165         case syscall.ERROR_IO_PENDING:
 166                 // IO started, and we have to wait for its completion.
 167                 err = nil
 168         default:
 169                 return 0, &OpError{oi.Name(), o.fd.net, o.fd.laddr, err}
 170         }
 171         // Wait for our request to complete.
 172         var r ioResult
 173         if deadline != 0 {
 174                 dt := deadline - time.Now().UnixNano()
 175                 if dt < 1 {
 176                         dt = 1
 177                 }
 178                 timer := time.NewTimer(time.Duration(dt) * time.Nanosecond)
 179                 defer timer.Stop()
 180                 select {
 181                 case r = <-o.resultc:
 182                 case <-timer.C:
 183                         s.canchan <- oi
 184                         <-o.errnoc
 185                         r = <-o.resultc
 186                         if r.err == syscall.ERROR_OPERATION_ABORTED { // IO Canceled
 187                                 r.err = syscall.EWOULDBLOCK
 188                         }
 189                 }
 190         } else {
 191                 r = <-o.resultc
 192         }
 193         if r.err != nil {
 194                 err = &OpError{oi.Name(), o.fd.net, o.fd.laddr, r.err}
 195         }
 196         return int(r.qty), err
 197 }
 198
 199 // Start helper goroutines.
 200 var resultsrv *resultSrv
 201 var iosrv *ioSrv
 202 var onceStartServer sync.Once
 203
 204 func startServer() {
 205         resultsrv = new(resultSrv)
 206         var err error
 207         resultsrv.iocp, err = syscall.CreateIoCompletionPort(syscall.InvalidHandle, 0, 0, 1)
 208         if err != nil {
 209                 panic("CreateIoCompletionPort: " + err.Error())
 210         }
 211         go resultsrv.Run()
 212
 213         iosrv = new(ioSrv)
 214         iosrv.submchan = make(chan anOpIface)
 215         iosrv.canchan = make(chan anOpIface)
 216         go iosrv.ProcessRemoteIO()
 217 }
 218
 219 // Network file descriptor.
 220 type netFD struct {
 221         // locking/lifetime of sysfd
 222         sysmu   sync.Mutex
 223         sysref  int
 224         closing bool
 225
 226         // immutable until Close
 227         sysfd       syscall.Handle
 228         family      int
 229         sotype      int
 230         isConnected bool
 231         net         string
 232         laddr       Addr
 233         raddr       Addr
 234         resultc     [2]chan ioResult // read/write completion results
 235         errnoc      [2]chan error    // read/write submit or cancel operation errors
 236
 237         // owned by client
 238         rdeadline int64
 239         rio       sync.Mutex
 240         wdeadline int64
 241         wio       sync.Mutex
 242 }
 243
 244 func allocFD(fd syscall.Handle, family, sotype int, net string) *netFD {
 245         netfd := &netFD{
 246                 sysfd:  fd,
 247                 family: family,
 248                 sotype: sotype,
 249                 net:    net,
 250         }
 251         runtime.SetFinalizer(netfd, (*netFD).Close)
 252         return netfd
 253 }
 254
 255 func newFD(fd syscall.Handle, family, proto int, net string) (*netFD, error) {
 256         if initErr != nil {
 257                 return nil, initErr
 258         }
 259         onceStartServer.Do(startServer)
 260         // Associate our socket with resultsrv.iocp.
 261         if _, err := syscall.CreateIoCompletionPort(syscall.Handle(fd), resultsrv.iocp, 0, 0); err != nil {
 262                 return nil, err
 263         }
 264         return allocFD(fd, family, proto, net), nil
 265 }
 266
 267 func (fd *netFD) setAddr(laddr, raddr Addr) {
 268         fd.laddr = laddr
 269         fd.raddr = raddr
 270 }
 271
 272 func (fd *netFD) connect(ra syscall.Sockaddr) error {
 273         return syscall.Connect(fd.sysfd, ra)
 274 }
 275
 276 var errClosing = errors.New("use of closed network connection")
 277
 278 // Add a reference to this fd.
 279 // If closing==true, mark the fd as closing.
 280 // Returns an error if the fd cannot be used.
 281 func (fd *netFD) incref(closing bool) error {
 282         if fd == nil {
 283                 return errClosing
 284         }
 285         fd.sysmu.Lock()
 286         if fd.closing {
 287                 fd.sysmu.Unlock()
 288                 return errClosing
 289         }
 290         fd.sysref++
 291         if closing {
 292                 fd.closing = true
 293         }
 294         closing = fd.closing
 295         fd.sysmu.Unlock()
 296         return nil
 297 }
 298
 299 // Remove a reference to this FD and close if we've been asked to do so (and
 300 // there are no references left.
 301 func (fd *netFD) decref() {
 302         fd.sysmu.Lock()
 303         fd.sysref--
 304         // NOTE(rsc): On Unix we check fd.sysref == 0 here before closing,
 305         // but on Windows we have no way to wake up the blocked I/O other
 306         // than closing the socket (or calling Shutdown, which breaks other
 307         // programs that might have a reference to the socket).  So there is
 308         // a small race here that we might close fd.sysfd and then some other
 309         // goroutine might start a read of fd.sysfd (having read it before we
 310         // write InvalidHandle to it), which might refer to some other file
 311         // if the specific handle value gets reused.  I think handle values on
 312         // Windows are not reused as aggressively as file descriptors on Unix,
 313         // so this might be tolerable.
 314         if fd.closing && fd.sysfd != syscall.InvalidHandle {
 315                 // In case the user has set linger, switch to blocking mode so
 316                 // the close blocks.  As long as this doesn't happen often, we
 317                 // can handle the extra OS processes.  Otherwise we'll need to
 318                 // use the resultsrv for Close too.  Sigh.
 319                 syscall.SetNonblock(fd.sysfd, false)
 320                 closesocket(fd.sysfd)
 321                 fd.sysfd = syscall.InvalidHandle
 322                 // no need for a finalizer anymore
 323                 runtime.SetFinalizer(fd, nil)
 324         }
 325         fd.sysmu.Unlock()
 326 }
 327
 328 func (fd *netFD) Close() error {
 329         if err := fd.incref(true); err != nil {
 330                 return err
 331         }
 332         fd.decref()
 333         return nil
 334 }
 335
 336 func (fd *netFD) shutdown(how int) error {
 337         if fd == nil || fd.sysfd == syscall.InvalidHandle {
 338                 return syscall.EINVAL
 339         }
 340         err := syscall.Shutdown(fd.sysfd, how)
 341         if err != nil {
 342                 return &OpError{"shutdown", fd.net, fd.laddr, err}
 343         }
 344         return nil
 345 }
 346
 347 func (fd *netFD) CloseRead() error {
 348         return fd.shutdown(syscall.SHUT_RD)
 349 }
 350
 351 func (fd *netFD) CloseWrite() error {
 352         return fd.shutdown(syscall.SHUT_WR)
 353 }
 354
 355 // Read from network.
 356
 357 type readOp struct {
 358         bufOp
 359 }
 360
 361 func (o *readOp) Submit() error {
 362         var d, f uint32
 363         return syscall.WSARecv(syscall.Handle(o.fd.sysfd), &o.buf, 1, &d, &f, &o.o, nil)
 364 }
 365
 366 func (o *readOp) Name() string {
 367         return "WSARecv"
 368 }
 369
 370 func (fd *netFD) Read(buf []byte) (int, error) {
 371         if fd == nil {
 372                 return 0, syscall.EINVAL
 373         }
 374         fd.rio.Lock()
 375         defer fd.rio.Unlock()
 376         if err := fd.incref(false); err != nil {
 377                 return 0, err
 378         }
 379         defer fd.decref()
 380         if fd.sysfd == syscall.InvalidHandle {
 381                 return 0, syscall.EINVAL
 382         }
 383         var o readOp
 384         o.Init(fd, buf, 'r')
 385         n, err := iosrv.ExecIO(&o, fd.rdeadline)
 386         if err == nil && n == 0 {
 387                 err = io.EOF
 388         }
 389         return n, err
 390 }
 391
 392 // ReadFrom from network.
 393
 394 type readFromOp struct {
 395         bufOp
 396         rsa  syscall.RawSockaddrAny
 397         rsan int32
 398 }
 399
 400 func (o *readFromOp) Submit() error {
 401         var d, f uint32
 402         return syscall.WSARecvFrom(o.fd.sysfd, &o.buf, 1, &d, &f, &o.rsa, &o.rsan, &o.o, nil)
 403 }
 404
 405 func (o *readFromOp) Name() string {
 406         return "WSARecvFrom"
 407 }
 408
 409 func (fd *netFD) ReadFrom(buf []byte) (n int, sa syscall.Sockaddr, err error) {
 410         if fd == nil {
 411                 return 0, nil, syscall.EINVAL
 412         }
 413         if len(buf) == 0 {
 414                 return 0, nil, nil
 415         }
 416         fd.rio.Lock()
 417         defer fd.rio.Unlock()
 418         if err := fd.incref(false); err != nil {
 419                 return 0, nil, err
 420         }
 421         defer fd.decref()
 422         var o readFromOp
 423         o.Init(fd, buf, 'r')
 424         o.rsan = int32(unsafe.Sizeof(o.rsa))
 425         n, err = iosrv.ExecIO(&o, fd.rdeadline)
 426         if err != nil {
 427                 return 0, nil, err
 428         }
 429         sa, _ = o.rsa.Sockaddr()
 430         return
 431 }
 432
 433 // Write to network.
 434
 435 type writeOp struct {
 436         bufOp
 437 }
 438
 439 func (o *writeOp) Submit() error {
 440         var d uint32
 441         return syscall.WSASend(o.fd.sysfd, &o.buf, 1, &d, 0, &o.o, nil)
 442 }
 443
 444 func (o *writeOp) Name() string {
 445         return "WSASend"
 446 }
 447
 448 func (fd *netFD) Write(buf []byte) (int, error) {
 449         if fd == nil {
 450                 return 0, syscall.EINVAL
 451         }
 452         fd.wio.Lock()
 453         defer fd.wio.Unlock()
 454         if err := fd.incref(false); err != nil {
 455                 return 0, err
 456         }
 457         defer fd.decref()
 458         var o writeOp
 459         o.Init(fd, buf, 'w')
 460         return iosrv.ExecIO(&o, fd.wdeadline)
 461 }
 462
 463 // WriteTo to network.
 464
 465 type writeToOp struct {
 466         bufOp
 467         sa syscall.Sockaddr
 468 }
 469
 470 func (o *writeToOp) Submit() error {
 471         var d uint32
 472         return syscall.WSASendto(o.fd.sysfd, &o.buf, 1, &d, 0, o.sa, &o.o, nil)
 473 }
 474
 475 func (o *writeToOp) Name() string {
 476         return "WSASendto"
 477 }
 478
 479 func (fd *netFD) WriteTo(buf []byte, sa syscall.Sockaddr) (int, error) {
 480         if fd == nil {
 481                 return 0, syscall.EINVAL
 482         }
 483         if len(buf) == 0 {
 484                 return 0, nil
 485         }
 486         fd.wio.Lock()
 487         defer fd.wio.Unlock()
 488         if err := fd.incref(false); err != nil {
 489                 return 0, err
 490         }
 491         defer fd.decref()
 492         if fd.sysfd == syscall.InvalidHandle {
 493                 return 0, syscall.EINVAL
 494         }
 495         var o writeToOp
 496         o.Init(fd, buf, 'w')
 497         o.sa = sa
 498         return iosrv.ExecIO(&o, fd.wdeadline)
 499 }
 500
 501 // Accept new network connections.
 502
 503 type acceptOp struct {
 504         anOp
 505         newsock syscall.Handle
 506         attrs   [2]syscall.RawSockaddrAny // space for local and remote address only
 507 }
 508
 509 func (o *acceptOp) Submit() error {
 510         var d uint32
 511         l := uint32(unsafe.Sizeof(o.attrs[0]))
 512         return syscall.AcceptEx(o.fd.sysfd, o.newsock,
 513                 (*byte)(unsafe.Pointer(&o.attrs[0])), 0, l, l, &d, &o.o)
 514 }
 515
 516 func (o *acceptOp) Name() string {
 517         return "AcceptEx"
 518 }
 519
 520 func (fd *netFD) accept(toAddr func(syscall.Sockaddr) Addr) (*netFD, error) {
 521         if err := fd.incref(false); err != nil {
 522                 return nil, err
 523         }
 524         defer fd.decref()
 525
 526         // Get new socket.
 527         // See ../syscall/exec.go for description of ForkLock.
 528         syscall.ForkLock.RLock()
 529         s, err := syscall.Socket(fd.family, fd.sotype, 0)
 530         if err != nil {
 531                 syscall.ForkLock.RUnlock()
 532                 return nil, err
 533         }
 534         syscall.CloseOnExec(s)
 535         syscall.ForkLock.RUnlock()
 536
 537         // Associate our new socket with IOCP.
 538         onceStartServer.Do(startServer)
 539         if _, err := syscall.CreateIoCompletionPort(s, resultsrv.iocp, 0, 0); err != nil {
 540                 return nil, &OpError{"CreateIoCompletionPort", fd.net, fd.laddr, err}
 541         }
 542
 543         // Submit accept request.
 544         var o acceptOp
 545         o.Init(fd, 'r')
 546         o.newsock = s
 547         _, err = iosrv.ExecIO(&o, 0)
 548         if err != nil {
 549                 closesocket(s)
 550                 return nil, err
 551         }
 552
 553         // Inherit properties of the listening socket.
 554         err = syscall.Setsockopt(s, syscall.SOL_SOCKET, syscall.SO_UPDATE_ACCEPT_CONTEXT, (*byte)(unsafe.Pointer(&fd.sysfd)), int32(unsafe.Sizeof(fd.sysfd)))
 555         if err != nil {
 556                 closesocket(s)
 557                 return nil, err
 558         }
 559
 560         // Get local and peer addr out of AcceptEx buffer.
 561         var lrsa, rrsa *syscall.RawSockaddrAny
 562         var llen, rlen int32
 563         l := uint32(unsafe.Sizeof(*lrsa))
 564         syscall.GetAcceptExSockaddrs((*byte)(unsafe.Pointer(&o.attrs[0])),
 565                 0, l, l, &lrsa, &llen, &rrsa, &rlen)
 566         lsa, _ := lrsa.Sockaddr()
 567         rsa, _ := rrsa.Sockaddr()
 568
 569         netfd := allocFD(s, fd.family, fd.sotype, fd.net)
 570         netfd.setAddr(toAddr(lsa), toAddr(rsa))
 571         return netfd, nil
 572 }
 573
 574 // Unimplemented functions.
 575
 576 func (fd *netFD) dup() (*os.File, error) {
 577         // TODO: Implement this
 578         return nil, os.NewSyscallError("dup", syscall.EWINDOWS)
 579 }
 580
 581 var errNoSupport = errors.New("address family not supported")
 582
 583 func (fd *netFD) ReadMsg(p []byte, oob []byte) (n, oobn, flags int, sa syscall.Sockaddr, err error) {
 584         return 0, 0, 0, nil, errNoSupport
 585 }
 586
 587 func (fd *netFD) WriteMsg(p []byte, oob []byte, sa syscall.Sockaddr) (n int, oobn int, err error) {
 588         return 0, 0, errNoSupport
 589 }