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/%d/ctl", getpid());
  20         fd = open(buf, OWRITE);
  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                 f(r);
  39         }
  40 }
  41
  42 Reqqueue *
  43 reqqueuecreate(void)
  44 {
  45         Reqqueue *q;
  46
  47         q = emalloc9p(sizeof(*q));
  48         memset(q, 0, sizeof(*q));
  49         q->l = q;
  50         q->next = q->prev = q;
  51         q->pid = threadpid(proccreate(_reqqueueproc, q, mainstacksize));
  52         return q;
  53 }
  54
  55 void
  56 reqqueuepush(Reqqueue *q, Req *r, void (*f)(Req *))
  57 {
  58         qlock(q);
  59         r->qu.f = f;
  60         r->qu.next = q;
  61         r->qu.prev = q->prev;
  62         q->prev->next = &r->qu;
  63         q->prev = &r->qu;
  64         rwakeup(q);
  65         qunlock(q);
  66 }
  67
  68 void
  69 reqqueueflush(Reqqueue *q, Req *r)
  70 {
  71         char buf[128];
  72         int fd;
  73
  74         qlock(q);
  75         if(q->cur == r){
  76                 sprint(buf, "/proc/%d/ctl", q->pid);
  77                 fd = open(buf, OWRITE);
  78                 if(fd >= 0){
  79                         write(fd, "interrupt", 9);
  80                         close(fd);
  81                 }
  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 }