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 void
   8 _reqqueueproc(void *v)
   9 {
  10         Reqqueue *q;
  11         Req *r;
  12         void (*f)(Req *);
  13         int fd;
  14         char *buf;
  15
  16         q = v;
  17         rfork(RFNOTEG);
  18
  19         buf = smprint("/proc/%lud/ctl", (ulong)getpid());
  20         fd = open(buf, OWRITE|OCEXEC);
  21         free(buf);
  22
  23         for(;;){
  24                 qlock(q);
  25                 q->flush = 0;
  26                 if(fd >= 0)
  27                         write(fd, "nointerrupt", 11);
  28                 q->cur = nil;
  29                 while(q->next == q)
  30                         rsleep(q);
  31                 r = (Req*)(((char*)q->next) - ((char*)&((Req*)0)->qu));
  32                 r->qu.next->prev = r->qu.prev;
  33                 r->qu.prev->next = r->qu.next;
  34                 f = r->qu.f;
  35                 memset(&r->qu, 0, sizeof(r->qu));
  36                 q->cur = r;
  37                 qunlock(q);
  38                 if(f == nil)
  39                         break;
  40                 f(r);
  41         }
  42
  43         if(fd >= 0)
  44                 close(fd);
  45         free(r);
  46         free(q);
  47         threadexits(nil);
  48 }
  49
  50 Reqqueue *
  51 reqqueuecreate(void)
  52 {
  53         Reqqueue *q;
  54
  55         q = emalloc9p(sizeof(*q));
  56         memset(q, 0, sizeof(*q));
  57         q->l = q;
  58         q->next = q->prev = q;
  59         q->pid = proccreate(_reqqueueproc, q, mainstacksize);
  60         return q;
  61 }
  62
  63 void
  64 reqqueuepush(Reqqueue *q, Req *r, void (*f)(Req *))
  65 {
  66         qlock(q);
  67         r->qu.f = f;
  68         r->qu.next = q;
  69         r->qu.prev = q->prev;
  70         q->prev->next = &r->qu;
  71         q->prev = &r->qu;
  72         rwakeup(q);
  73         qunlock(q);
  74 }
  75
  76 void
  77 reqqueueflush(Reqqueue *q, Req *r)
  78 {
  79         qlock(q);
  80         if(q->cur == r){
  81                 threadint(q->pid);
  82                 q->flush++;
  83                 qunlock(q);
  84         }else{
  85                 if(r->qu.next != nil){
  86                         r->qu.next->prev = r->qu.prev;
  87                         r->qu.prev->next = r->qu.next;
  88                 }
  89                 memset(&r->qu, 0, sizeof(r->qu));
  90                 qunlock(q);
  91                 respond(r, "interrupted");
  92         }
  93 }
  94
  95 void
  96 reqqueuefree(Reqqueue *q)
  97 {
  98         Req *r;
  99
 100         if(q == nil)
 101                 return;
 102         r = emalloc9p(sizeof(Req));
 103         reqqueuepush(q, r, nil);
 104 }