libgo/runtime/netpoll.goc

   1 // Copyright 2013 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 // +build darwin dragonfly freebsd linux nacl netbsd openbsd solaris windows
   6
   7 package net
   8
   9 #include "runtime.h"
  10 #include "defs.h"
  11 #include "arch.h"
  12 #include "malloc.h"
  13
  14 // Map gccgo field names to gc field names.
  15 // Eface aka __go_empty_interface.
  16 #define type __type_descriptor
  17 #define data __object
  18
  19 // Integrated network poller (platform-independent part).
  20 // A particular implementation (epoll/kqueue) must define the following functions:
  21 // void runtime_netpollinit(void);                      // to initialize the poller
  22 // int32 runtime_netpollopen(uintptr fd, PollDesc *pd); // to arm edge-triggered notifications
  23                                                         // and associate fd with pd.
  24 // An implementation must call the following function to denote that the pd is ready.
  25 // void runtime_netpollready(G **gpp, PollDesc *pd, int32 mode);
  26
  27 // PollDesc contains 2 binary semaphores, rg and wg, to park reader and writer
  28 // goroutines respectively. The semaphore can be in the following states:
  29 // READY - io readiness notification is pending;
  30 //         a goroutine consumes the notification by changing the state to nil.
  31 // WAIT - a goroutine prepares to park on the semaphore, but not yet parked;
  32 //        the goroutine commits to park by changing the state to G pointer,
  33 //        or, alternatively, concurrent io notification changes the state to READY,
  34 //        or, alternatively, concurrent timeout/close changes the state to nil.
  35 // G pointer - the goroutine is blocked on the semaphore;
  36 //             io notification or timeout/close changes the state to READY or nil respectively
  37 //             and unparks the goroutine.
  38 // nil - nothing of the above.
  39 #define READY ((G*)1)
  40 #define WAIT  ((G*)2)
  41
  42 enum
  43 {
  44         PollBlockSize   = 4*1024,
  45 };
  46
  47 struct PollDesc
  48 {
  49         PollDesc* link; // in pollcache, protected by pollcache.Lock
  50
  51         // The lock protects pollOpen, pollSetDeadline, pollUnblock and deadlineimpl operations.
  52         // This fully covers seq, rt and wt variables. fd is constant throughout the PollDesc lifetime.
  53         // pollReset, pollWait, pollWaitCanceled and runtime_netpollready (IO rediness notification)
  54         // proceed w/o taking the lock. So closing, rg, rd, wg and wd are manipulated
  55         // in a lock-free way by all operations.
  56         Lock;           // protectes the following fields
  57         uintptr fd;
  58         bool    closing;
  59         uintptr seq;    // protects from stale timers and ready notifications
  60         G*      rg;     // READY, WAIT, G waiting for read or nil
  61         Timer   rt;     // read deadline timer (set if rt.fv != nil)
  62         int64   rd;     // read deadline
  63         G*      wg;     // READY, WAIT, G waiting for write or nil
  64         Timer   wt;     // write deadline timer
  65         int64   wd;     // write deadline
  66         void*   user;   // user settable cookie
  67 };
  68
  69 static struct
  70 {
  71         Lock;
  72         PollDesc*       first;
  73         // PollDesc objects must be type-stable,
  74         // because we can get ready notification from epoll/kqueue
  75         // after the descriptor is closed/reused.
  76         // Stale notifications are detected using seq variable,
  77         // seq is incremented when deadlines are changed or descriptor is reused.
  78 } pollcache;
  79
  80 static bool     netpollblock(PollDesc*, int32, bool);
  81 static G*       netpollunblock(PollDesc*, int32, bool);
  82 static void     deadline(int64, Eface);
  83 static void     readDeadline(int64, Eface);
  84 static void     writeDeadline(int64, Eface);
  85 static PollDesc*        allocPollDesc(void);
  86 static intgo    checkerr(PollDesc *pd, int32 mode);
  87
  88 static FuncVal deadlineFn       = {(void(*)(void))deadline};
  89 static FuncVal readDeadlineFn   = {(void(*)(void))readDeadline};
  90 static FuncVal writeDeadlineFn  = {(void(*)(void))writeDeadline};
  91
  92 // runtimeNano returns the current value of the runtime clock in nanoseconds.
  93 func runtimeNano() (ns int64) {
  94         ns = runtime_nanotime();
  95 }
  96
  97 func runtime_pollServerInit() {
  98         runtime_netpollinit();
  99 }
 100
 101 func runtime_pollOpen(fd uintptr) (pd *PollDesc, errno int) {
 102         pd = allocPollDesc();
 103         runtime_lock(pd);
 104         if(pd->wg != nil && pd->wg != READY)
 105                 runtime_throw("runtime_pollOpen: blocked write on free descriptor");
 106         if(pd->rg != nil && pd->rg != READY)
 107                 runtime_throw("runtime_pollOpen: blocked read on free descriptor");
 108         pd->fd = fd;
 109         pd->closing = false;
 110         pd->seq++;
 111         pd->rg = nil;
 112         pd->rd = 0;
 113         pd->wg = nil;
 114         pd->wd = 0;
 115         runtime_unlock(pd);
 116
 117         errno = runtime_netpollopen(fd, pd);
 118 }
 119
 120 func runtime_pollClose(pd *PollDesc) {
 121         if(!pd->closing)
 122                 runtime_throw("runtime_pollClose: close w/o unblock");
 123         if(pd->wg != nil && pd->wg != READY)
 124                 runtime_throw("runtime_pollClose: blocked write on closing descriptor");
 125         if(pd->rg != nil && pd->rg != READY)
 126                 runtime_throw("runtime_pollClose: blocked read on closing descriptor");
 127         runtime_netpollclose(pd->fd);
 128         runtime_lock(&pollcache);
 129         pd->link = pollcache.first;
 130         pollcache.first = pd;
 131         runtime_unlock(&pollcache);
 132 }
 133
 134 func runtime_pollReset(pd *PollDesc, mode int) (err int) {
 135         err = checkerr(pd, mode);
 136         if(err)
 137                 goto ret;
 138         if(mode == 'r')
 139                 pd->rg = nil;
 140         else if(mode == 'w')
 141                 pd->wg = nil;
 142 ret:
 143 }
 144
 145 func runtime_pollWait(pd *PollDesc, mode int) (err int) {
 146         err = checkerr(pd, mode);
 147         if(err == 0) {
 148                 // As for now only Solaris uses level-triggered IO.
 149                 if(Solaris)
 150                         runtime_netpollarm(pd, mode);
 151                 while(!netpollblock(pd, mode, false)) {
 152                         err = checkerr(pd, mode);
 153                         if(err != 0)
 154                                 break;
 155                         // Can happen if timeout has fired and unblocked us,
 156                         // but before we had a chance to run, timeout has been reset.
 157                         // Pretend it has not happened and retry.
 158                 }
 159         }
 160 }
 161
 162 func runtime_pollWaitCanceled(pd *PollDesc, mode int) {
 163         // This function is used only on windows after a failed attempt to cancel
 164         // a pending async IO operation. Wait for ioready, ignore closing or timeouts.
 165         while(!netpollblock(pd, mode, true))
 166                 ;
 167 }
 168
 169 func runtime_pollSetDeadline(pd *PollDesc, d int64, mode int) {
 170         G *rg, *wg;
 171
 172         runtime_lock(pd);
 173         if(pd->closing) {
 174                 runtime_unlock(pd);
 175                 return;
 176         }
 177         pd->seq++;  // invalidate current timers
 178         // Reset current timers.
 179         if(pd->rt.fv) {
 180                 runtime_deltimer(&pd->rt);
 181                 pd->rt.fv = nil;
 182         }
 183         if(pd->wt.fv) {
 184                 runtime_deltimer(&pd->wt);
 185                 pd->wt.fv = nil;
 186         }
 187         // Setup new timers.
 188         if(d != 0 && d <= runtime_nanotime())
 189                 d = -1;
 190         if(mode == 'r' || mode == 'r'+'w')
 191                 pd->rd = d;
 192         if(mode == 'w' || mode == 'r'+'w')
 193                 pd->wd = d;
 194         if(pd->rd > 0 && pd->rd == pd->wd) {
 195                 pd->rt.fv = &deadlineFn;
 196                 pd->rt.when = pd->rd;
 197                 // Copy current seq into the timer arg.
 198                 // Timer func will check the seq against current descriptor seq,
 199                 // if they differ the descriptor was reused or timers were reset.
 200                 pd->rt.arg.type = (Type*)pd->seq;
 201                 pd->rt.arg.data = pd;
 202                 runtime_addtimer(&pd->rt);
 203         } else {
 204                 if(pd->rd > 0) {
 205                         pd->rt.fv = &readDeadlineFn;
 206                         pd->rt.when = pd->rd;
 207                         pd->rt.arg.type = (Type*)pd->seq;
 208                         pd->rt.arg.data = pd;
 209                         runtime_addtimer(&pd->rt);
 210                 }
 211                 if(pd->wd > 0) {
 212                         pd->wt.fv = &writeDeadlineFn;
 213                         pd->wt.when = pd->wd;
 214                         pd->wt.arg.type = (Type*)pd->seq;
 215                         pd->wt.arg.data = pd;
 216                         runtime_addtimer(&pd->wt);
 217                 }
 218         }
 219         // If we set the new deadline in the past, unblock currently pending IO if any.
 220         rg = nil;
 221         runtime_atomicstorep(&wg, nil);  // full memory barrier between stores to rd/wd and load of rg/wg in netpollunblock
 222         if(pd->rd < 0)
 223                 rg = netpollunblock(pd, 'r', false);
 224         if(pd->wd < 0)
 225                 wg = netpollunblock(pd, 'w', false);
 226         runtime_unlock(pd);
 227         if(rg)
 228                 runtime_ready(rg);
 229         if(wg)
 230                 runtime_ready(wg);
 231 }
 232
 233 func runtime_pollUnblock(pd *PollDesc) {
 234         G *rg, *wg;
 235
 236         runtime_lock(pd);
 237         if(pd->closing)
 238                 runtime_throw("runtime_pollUnblock: already closing");
 239         pd->closing = true;
 240         pd->seq++;
 241         runtime_atomicstorep(&rg, nil);  // full memory barrier between store to closing and read of rg/wg in netpollunblock
 242         rg = netpollunblock(pd, 'r', false);
 243         wg = netpollunblock(pd, 'w', false);
 244         if(pd->rt.fv) {
 245                 runtime_deltimer(&pd->rt);
 246                 pd->rt.fv = nil;
 247         }
 248         if(pd->wt.fv) {
 249                 runtime_deltimer(&pd->wt);
 250                 pd->wt.fv = nil;
 251         }
 252         runtime_unlock(pd);
 253         if(rg)
 254                 runtime_ready(rg);
 255         if(wg)
 256                 runtime_ready(wg);
 257 }
 258
 259 uintptr
 260 runtime_netpollfd(PollDesc *pd)
 261 {
 262         return pd->fd;
 263 }
 264
 265 void**
 266 runtime_netpolluser(PollDesc *pd)
 267 {
 268         return &pd->user;
 269 }
 270
 271 bool
 272 runtime_netpollclosing(PollDesc *pd)
 273 {
 274         return pd->closing;
 275 }
 276
 277 void
 278 runtime_netpolllock(PollDesc *pd)
 279 {
 280         runtime_lock(pd);
 281 }
 282
 283 void
 284 runtime_netpollunlock(PollDesc *pd)
 285 {
 286         runtime_unlock(pd);
 287 }
 288
 289 // make pd ready, newly runnable goroutines (if any) are enqueued info gpp list
 290 void
 291 runtime_netpollready(G **gpp, PollDesc *pd, int32 mode)
 292 {
 293         G *rg, *wg;
 294
 295         rg = wg = nil;
 296         if(mode == 'r' || mode == 'r'+'w')
 297                 rg = netpollunblock(pd, 'r', true);
 298         if(mode == 'w' || mode == 'r'+'w')
 299                 wg = netpollunblock(pd, 'w', true);
 300         if(rg) {
 301                 rg->schedlink = *gpp;
 302                 *gpp = rg;
 303         }
 304         if(wg) {
 305                 wg->schedlink = *gpp;
 306                 *gpp = wg;
 307         }
 308 }
 309
 310 static intgo
 311 checkerr(PollDesc *pd, int32 mode)
 312 {
 313         if(pd->closing)
 314                 return 1;  // errClosing
 315         if((mode == 'r' && pd->rd < 0) || (mode == 'w' && pd->wd < 0))
 316                 return 2;  // errTimeout
 317         return 0;
 318 }
 319
 320 static bool
 321 blockcommit(G *gp, G **gpp)
 322 {
 323         return runtime_casp(gpp, WAIT, gp);
 324 }
 325
 326 // returns true if IO is ready, or false if timedout or closed
 327 // waitio - wait only for completed IO, ignore errors
 328 static bool
 329 netpollblock(PollDesc *pd, int32 mode, bool waitio)
 330 {
 331         G **gpp, *old;
 332
 333         gpp = &pd->rg;
 334         if(mode == 'w')
 335                 gpp = &pd->wg;
 336
 337         // set the gpp semaphore to WAIT
 338         for(;;) {
 339                 old = *gpp;
 340                 if(old == READY) {
 341                         *gpp = nil;
 342                         return true;
 343                 }
 344                 if(old != nil)
 345                         runtime_throw("netpollblock: double wait");
 346                 if(runtime_casp(gpp, nil, WAIT))
 347                         break;
 348         }
 349
 350         // need to recheck error states after setting gpp to WAIT
 351         // this is necessary because runtime_pollUnblock/runtime_pollSetDeadline/deadlineimpl
 352         // do the opposite: store to closing/rd/wd, membarrier, load of rg/wg
 353         if(waitio || checkerr(pd, mode) == 0)
 354                 runtime_park((bool(*)(G*, void*))blockcommit, gpp, "IO wait");
 355         // be careful to not lose concurrent READY notification
 356         old = runtime_xchgp(gpp, nil);
 357         if(old > WAIT)
 358                 runtime_throw("netpollblock: corrupted state");
 359         return old == READY;
 360 }
 361
 362 static G*
 363 netpollunblock(PollDesc *pd, int32 mode, bool ioready)
 364 {
 365         G **gpp, *old, *new;
 366
 367         gpp = &pd->rg;
 368         if(mode == 'w')
 369                 gpp = &pd->wg;
 370
 371         for(;;) {
 372                 old = *gpp;
 373                 if(old == READY)
 374                         return nil;
 375                 if(old == nil && !ioready) {
 376                         // Only set READY for ioready. runtime_pollWait
 377                         // will check for timeout/cancel before waiting.
 378                         return nil;
 379                 }
 380                 new = nil;
 381                 if(ioready)
 382                         new = READY;
 383                 if(runtime_casp(gpp, old, new))
 384                         break;
 385         }
 386         if(old > WAIT)
 387                 return old;  // must be G*
 388         return nil;
 389 }
 390
 391 static void
 392 deadlineimpl(int64 now, Eface arg, bool read, bool write)
 393 {
 394         PollDesc *pd;
 395         uint32 seq;
 396         G *rg, *wg;
 397
 398         USED(now);
 399         pd = (PollDesc*)arg.data;
 400         // This is the seq when the timer was set.
 401         // If it's stale, ignore the timer event.
 402         seq = (uintptr)arg.type;
 403         rg = wg = nil;
 404         runtime_lock(pd);
 405         if(seq != pd->seq) {
 406                 // The descriptor was reused or timers were reset.
 407                 runtime_unlock(pd);
 408                 return;
 409         }
 410         if(read) {
 411                 if(pd->rd <= 0 || pd->rt.fv == nil)
 412                         runtime_throw("deadlineimpl: inconsistent read deadline");
 413                 pd->rd = -1;
 414                 runtime_atomicstorep(&pd->rt.fv, nil);  // full memory barrier between store to rd and load of rg in netpollunblock
 415                 rg = netpollunblock(pd, 'r', false);
 416         }
 417         if(write) {
 418                 if(pd->wd <= 0 || (pd->wt.fv == nil && !read))
 419                         runtime_throw("deadlineimpl: inconsistent write deadline");
 420                 pd->wd = -1;
 421                 runtime_atomicstorep(&pd->wt.fv, nil);  // full memory barrier between store to wd and load of wg in netpollunblock
 422                 wg = netpollunblock(pd, 'w', false);
 423         }
 424         runtime_unlock(pd);
 425         if(rg)
 426                 runtime_ready(rg);
 427         if(wg)
 428                 runtime_ready(wg);
 429 }
 430
 431 static void
 432 deadline(int64 now, Eface arg)
 433 {
 434         deadlineimpl(now, arg, true, true);
 435 }
 436
 437 static void
 438 readDeadline(int64 now, Eface arg)
 439 {
 440         deadlineimpl(now, arg, true, false);
 441 }
 442
 443 static void
 444 writeDeadline(int64 now, Eface arg)
 445 {
 446         deadlineimpl(now, arg, false, true);
 447 }
 448
 449 static PollDesc*
 450 allocPollDesc(void)
 451 {
 452         PollDesc *pd;
 453         uint32 i, n;
 454
 455         runtime_lock(&pollcache);
 456         if(pollcache.first == nil) {
 457                 n = PollBlockSize/sizeof(*pd);
 458                 if(n == 0)
 459                         n = 1;
 460                 // Must be in non-GC memory because can be referenced
 461                 // only from epoll/kqueue internals.
 462                 pd = runtime_persistentalloc(n*sizeof(*pd), 0, &mstats.other_sys);
 463                 for(i = 0; i < n; i++) {
 464                         pd[i].link = pollcache.first;
 465                         pollcache.first = &pd[i];
 466                 }
 467         }
 468         pd = pollcache.first;
 469         pollcache.first = pd->link;
 470         runtime_unlock(&pollcache);
 471         return pd;
 472 }