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 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 };
  67
  68 static struct
  69 {
  70         Lock;
  71         PollDesc*       first;
  72         // PollDesc objects must be type-stable,
  73         // because we can get ready notification from epoll/kqueue
  74         // after the descriptor is closed/reused.
  75         // Stale notifications are detected using seq variable,
  76         // seq is incremented when deadlines are changed or descriptor is reused.
  77 } pollcache;
  78
  79 static bool     netpollblock(PollDesc*, int32, bool);
  80 static G*       netpollunblock(PollDesc*, int32, bool);
  81 static void     deadline(int64, Eface);
  82 static void     readDeadline(int64, Eface);
  83 static void     writeDeadline(int64, Eface);
  84 static PollDesc*        allocPollDesc(void);
  85 static intgo    checkerr(PollDesc *pd, int32 mode);
  86
  87 static FuncVal deadlineFn       = {(void(*)(void))deadline};
  88 static FuncVal readDeadlineFn   = {(void(*)(void))readDeadline};
  89 static FuncVal writeDeadlineFn  = {(void(*)(void))writeDeadline};
  90
  91 func runtime_pollServerInit() {
  92         runtime_netpollinit();
  93 }
  94
  95 func runtime_pollOpen(fd uintptr) (pd *PollDesc, errno int) {
  96         pd = allocPollDesc();
  97         runtime_lock(pd);
  98         if(pd->wg != nil && pd->wg != READY)
  99                 runtime_throw("runtime_pollOpen: blocked write on free descriptor");
 100         if(pd->rg != nil && pd->rg != READY)
 101                 runtime_throw("runtime_pollOpen: blocked read on free descriptor");
 102         pd->fd = fd;
 103         pd->closing = false;
 104         pd->seq++;
 105         pd->rg = nil;
 106         pd->rd = 0;
 107         pd->wg = nil;
 108         pd->wd = 0;
 109         runtime_unlock(pd);
 110
 111         errno = runtime_netpollopen(fd, pd);
 112 }
 113
 114 func runtime_pollClose(pd *PollDesc) {
 115         if(!pd->closing)
 116                 runtime_throw("runtime_pollClose: close w/o unblock");
 117         if(pd->wg != nil && pd->wg != READY)
 118                 runtime_throw("runtime_pollClose: blocked write on closing descriptor");
 119         if(pd->rg != nil && pd->rg != READY)
 120                 runtime_throw("runtime_pollClose: blocked read on closing descriptor");
 121         runtime_netpollclose(pd->fd);
 122         runtime_lock(&pollcache);
 123         pd->link = pollcache.first;
 124         pollcache.first = pd;
 125         runtime_unlock(&pollcache);
 126 }
 127
 128 func runtime_pollReset(pd *PollDesc, mode int) (err int) {
 129         err = checkerr(pd, mode);
 130         if(err)
 131                 goto ret;
 132         if(mode == 'r')
 133                 pd->rg = nil;
 134         else if(mode == 'w')
 135                 pd->wg = nil;
 136 ret:
 137 }
 138
 139 func runtime_pollWait(pd *PollDesc, mode int) (err int) {
 140         err = checkerr(pd, mode);
 141         if(err == 0) {
 142                 // As for now only Solaris uses level-triggered IO.
 143                 if(Solaris)
 144                         runtime_netpollarm(pd->fd, mode);
 145                 while(!netpollblock(pd, mode, false)) {
 146                         err = checkerr(pd, mode);
 147                         if(err != 0)
 148                                 break;
 149                         // Can happen if timeout has fired and unblocked us,
 150                         // but before we had a chance to run, timeout has been reset.
 151                         // Pretend it has not happened and retry.
 152                 }
 153         }
 154 }
 155
 156 func runtime_pollWaitCanceled(pd *PollDesc, mode int) {
 157         // This function is used only on windows after a failed attempt to cancel
 158         // a pending async IO operation. Wait for ioready, ignore closing or timeouts.
 159         while(!netpollblock(pd, mode, true))
 160                 ;
 161 }
 162
 163 func runtime_pollSetDeadline(pd *PollDesc, d int64, mode int) {
 164         G *rg, *wg;
 165
 166         runtime_lock(pd);
 167         if(pd->closing) {
 168                 runtime_unlock(pd);
 169                 return;
 170         }
 171         pd->seq++;  // invalidate current timers
 172         // Reset current timers.
 173         if(pd->rt.fv) {
 174                 runtime_deltimer(&pd->rt);
 175                 pd->rt.fv = nil;
 176         }
 177         if(pd->wt.fv) {
 178                 runtime_deltimer(&pd->wt);
 179                 pd->wt.fv = nil;
 180         }
 181         // Setup new timers.
 182         if(d != 0 && d <= runtime_nanotime())
 183                 d = -1;
 184         if(mode == 'r' || mode == 'r'+'w')
 185                 pd->rd = d;
 186         if(mode == 'w' || mode == 'r'+'w')
 187                 pd->wd = d;
 188         if(pd->rd > 0 && pd->rd == pd->wd) {
 189                 pd->rt.fv = &deadlineFn;
 190                 pd->rt.when = pd->rd;
 191                 // Copy current seq into the timer arg.
 192                 // Timer func will check the seq against current descriptor seq,
 193                 // if they differ the descriptor was reused or timers were reset.
 194                 pd->rt.arg.type = (Type*)pd->seq;
 195                 pd->rt.arg.data = pd;
 196                 runtime_addtimer(&pd->rt);
 197         } else {
 198                 if(pd->rd > 0) {
 199                         pd->rt.fv = &readDeadlineFn;
 200                         pd->rt.when = pd->rd;
 201                         pd->rt.arg.type = (Type*)pd->seq;
 202                         pd->rt.arg.data = pd;
 203                         runtime_addtimer(&pd->rt);
 204                 }
 205                 if(pd->wd > 0) {
 206                         pd->wt.fv = &writeDeadlineFn;
 207                         pd->wt.when = pd->wd;
 208                         pd->wt.arg.type = (Type*)pd->seq;
 209                         pd->wt.arg.data = pd;
 210                         runtime_addtimer(&pd->wt);
 211                 }
 212         }
 213         // If we set the new deadline in the past, unblock currently pending IO if any.
 214         rg = nil;
 215         runtime_atomicstorep(&wg, nil);  // full memory barrier between stores to rd/wd and load of rg/wg in netpollunblock
 216         if(pd->rd < 0)
 217                 rg = netpollunblock(pd, 'r', false);
 218         if(pd->wd < 0)
 219                 wg = netpollunblock(pd, 'w', false);
 220         runtime_unlock(pd);
 221         if(rg)
 222                 runtime_ready(rg);
 223         if(wg)
 224                 runtime_ready(wg);
 225 }
 226
 227 func runtime_pollUnblock(pd *PollDesc) {
 228         G *rg, *wg;
 229
 230         runtime_lock(pd);
 231         if(pd->closing)
 232                 runtime_throw("runtime_pollUnblock: already closing");
 233         pd->closing = true;
 234         pd->seq++;
 235         runtime_atomicstorep(&rg, nil);  // full memory barrier between store to closing and read of rg/wg in netpollunblock
 236         rg = netpollunblock(pd, 'r', false);
 237         wg = netpollunblock(pd, 'w', false);
 238         if(pd->rt.fv) {
 239                 runtime_deltimer(&pd->rt);
 240                 pd->rt.fv = nil;
 241         }
 242         if(pd->wt.fv) {
 243                 runtime_deltimer(&pd->wt);
 244                 pd->wt.fv = nil;
 245         }
 246         runtime_unlock(pd);
 247         if(rg)
 248                 runtime_ready(rg);
 249         if(wg)
 250                 runtime_ready(wg);
 251 }
 252
 253 uintptr
 254 runtime_netpollfd(PollDesc *pd)
 255 {
 256         return pd->fd;
 257 }
 258
 259 // make pd ready, newly runnable goroutines (if any) are enqueued info gpp list
 260 void
 261 runtime_netpollready(G **gpp, PollDesc *pd, int32 mode)
 262 {
 263         G *rg, *wg;
 264
 265         rg = wg = nil;
 266         if(mode == 'r' || mode == 'r'+'w')
 267                 rg = netpollunblock(pd, 'r', true);
 268         if(mode == 'w' || mode == 'r'+'w')
 269                 wg = netpollunblock(pd, 'w', true);
 270         if(rg) {
 271                 rg->schedlink = *gpp;
 272                 *gpp = rg;
 273         }
 274         if(wg) {
 275                 wg->schedlink = *gpp;
 276                 *gpp = wg;
 277         }
 278 }
 279
 280 static intgo
 281 checkerr(PollDesc *pd, int32 mode)
 282 {
 283         if(pd->closing)
 284                 return 1;  // errClosing
 285         if((mode == 'r' && pd->rd < 0) || (mode == 'w' && pd->wd < 0))
 286                 return 2;  // errTimeout
 287         return 0;
 288 }
 289
 290 static bool
 291 blockcommit(G *gp, G **gpp)
 292 {
 293         return runtime_casp(gpp, WAIT, gp);
 294 }
 295
 296 // returns true if IO is ready, or false if timedout or closed
 297 // waitio - wait only for completed IO, ignore errors
 298 static bool
 299 netpollblock(PollDesc *pd, int32 mode, bool waitio)
 300 {
 301         G **gpp, *old;
 302
 303         gpp = &pd->rg;
 304         if(mode == 'w')
 305                 gpp = &pd->wg;
 306
 307         // set the gpp semaphore to WAIT
 308         for(;;) {
 309                 old = *gpp;
 310                 if(old == READY) {
 311                         *gpp = nil;
 312                         return true;
 313                 }
 314                 if(old != nil)
 315                         runtime_throw("netpollblock: double wait");
 316                 if(runtime_casp(gpp, nil, WAIT))
 317                         break;
 318         }
 319
 320         // need to recheck error states after setting gpp to WAIT
 321         // this is necessary because runtime_pollUnblock/runtime_pollSetDeadline/deadlineimpl
 322         // do the opposite: store to closing/rd/wd, membarrier, load of rg/wg
 323         if(waitio || checkerr(pd, mode) == 0)
 324                 runtime_park((bool(*)(G*, void*))blockcommit, gpp, "IO wait");
 325         // be careful to not lose concurrent READY notification
 326         old = runtime_xchgp(gpp, nil);
 327         if(old > WAIT)
 328                 runtime_throw("netpollblock: corrupted state");
 329         return old == READY;
 330 }
 331
 332 static G*
 333 netpollunblock(PollDesc *pd, int32 mode, bool ioready)
 334 {
 335         G **gpp, *old, *new;
 336
 337         gpp = &pd->rg;
 338         if(mode == 'w')
 339                 gpp = &pd->wg;
 340
 341         for(;;) {
 342                 old = *gpp;
 343                 if(old == READY)
 344                         return nil;
 345                 if(old == nil && !ioready) {
 346                         // Only set READY for ioready. runtime_pollWait
 347                         // will check for timeout/cancel before waiting.
 348                         return nil;
 349                 }
 350                 new = nil;
 351                 if(ioready)
 352                         new = READY;
 353                 if(runtime_casp(gpp, old, new))
 354                         break;
 355         }
 356         if(old > WAIT)
 357                 return old;  // must be G*
 358         return nil;
 359 }
 360
 361 static void
 362 deadlineimpl(int64 now, Eface arg, bool read, bool write)
 363 {
 364         PollDesc *pd;
 365         uint32 seq;
 366         G *rg, *wg;
 367
 368         USED(now);
 369         pd = (PollDesc*)arg.data;
 370         // This is the seq when the timer was set.
 371         // If it's stale, ignore the timer event.
 372         seq = (uintptr)arg.type;
 373         rg = wg = nil;
 374         runtime_lock(pd);
 375         if(seq != pd->seq) {
 376                 // The descriptor was reused or timers were reset.
 377                 runtime_unlock(pd);
 378                 return;
 379         }
 380         if(read) {
 381                 if(pd->rd <= 0 || pd->rt.fv == nil)
 382                         runtime_throw("deadlineimpl: inconsistent read deadline");
 383                 pd->rd = -1;
 384                 runtime_atomicstorep(&pd->rt.fv, nil);  // full memory barrier between store to rd and load of rg in netpollunblock
 385                 rg = netpollunblock(pd, 'r', false);
 386         }
 387         if(write) {
 388                 if(pd->wd <= 0 || (pd->wt.fv == nil && !read))
 389                         runtime_throw("deadlineimpl: inconsistent write deadline");
 390                 pd->wd = -1;
 391                 runtime_atomicstorep(&pd->wt.fv, nil);  // full memory barrier between store to wd and load of wg in netpollunblock
 392                 wg = netpollunblock(pd, 'w', false);
 393         }
 394         runtime_unlock(pd);
 395         if(rg)
 396                 runtime_ready(rg);
 397         if(wg)
 398                 runtime_ready(wg);
 399 }
 400
 401 static void
 402 deadline(int64 now, Eface arg)
 403 {
 404         deadlineimpl(now, arg, true, true);
 405 }
 406
 407 static void
 408 readDeadline(int64 now, Eface arg)
 409 {
 410         deadlineimpl(now, arg, true, false);
 411 }
 412
 413 static void
 414 writeDeadline(int64 now, Eface arg)
 415 {
 416         deadlineimpl(now, arg, false, true);
 417 }
 418
 419 static PollDesc*
 420 allocPollDesc(void)
 421 {
 422         PollDesc *pd;
 423         uint32 i, n;
 424
 425         runtime_lock(&pollcache);
 426         if(pollcache.first == nil) {
 427                 n = PollBlockSize/sizeof(*pd);
 428                 if(n == 0)
 429                         n = 1;
 430                 // Must be in non-GC memory because can be referenced
 431                 // only from epoll/kqueue internals.
 432                 pd = runtime_persistentalloc(n*sizeof(*pd), 0, &mstats.other_sys);
 433                 for(i = 0; i < n; i++) {
 434                         pd[i].link = pollcache.first;
 435                         pollcache.first = &pd[i];
 436                 }
 437         }
 438         pd = pollcache.first;
 439         pollcache.first = pd->link;
 440         runtime_unlock(&pollcache);
 441         return pd;
 442 }