sys/src/lib9p/queue.c

   1 #include <u.h>
   2 #include <libc.h>
   3 #include <thread.h>
   4 #include <fcall.h>
   5 #include <9p.h>
   6
   7 static int
   8 _reqqueuenote(void *, char *note)
   9 {
  10         return strcmp(note, "flush") == 0;
  11 }
  12
  13 static void
  14 _reqqueueproc(void *v)
  15 {
  16         Reqqueue *q;
  17         Req *r;
  18         void (*f)(Req *);
  19
  20         q = v;
  21         rfork(RFNOTEG);
  22         threadnotify(_reqqueuenote, 1);
  23         for(;;){
  24                 qlock(q);
  25                 q->cur = nil;
  26                 while(q->next == q)
  27                         rsleep(q);
  28                 r = (Req*)(((char*)q->next) - ((char*)&((Req*)0)->qu));
  29                 r->qu.next->prev = r->qu.prev;
  30                 r->qu.prev->next = r->qu.next;
  31                 f = r->qu.f;
  32                 memset(&r->qu, 0, sizeof(r->qu));
  33                 q->cur = r;
  34                 qunlock(q);
  35                 f(r);
  36         }
  37 }
  38
  39 Reqqueue *
  40 reqqueuecreate(void)
  41 {
  42         Reqqueue *q;
  43
  44         q = emalloc9p(sizeof(*q));
  45         memset(q, 0, sizeof(*q));
  46         q->l = q;
  47         q->next = q->prev = q;
  48         q->pid = threadpid(proccreate(_reqqueueproc, q, mainstacksize));
  49         return q;
  50 }
  51
  52 void
  53 reqqueuepush(Reqqueue *q, Req *r, void (*f)(Req *))
  54 {
  55         qlock(q);
  56         r->qu.f = f;
  57         r->qu.next = q;
  58         r->qu.prev = q->prev;
  59         q->prev->next = &r->qu;
  60         q->prev = &r->qu;
  61         rwakeup(q);
  62         qunlock(q);
  63 }
  64
  65 void
  66 reqqueueflush(Reqqueue *q, Req *r)
  67 {
  68         qlock(q);
  69         if(q->cur == r){
  70                 postnote(PNPROC, q->pid, "flush");
  71                 qunlock(q);
  72         }else{
  73                 if(r->qu.next != nil){
  74                         r->qu.next->prev = r->qu.prev;
  75                         r->qu.prev->next = r->qu.next;
  76                 }
  77                 memset(&r->qu, 0, sizeof(r->qu));
  78                 qunlock(q);
  79                 respond(r, "interrupted");
  80         }
  81 }