sys/src/cmd/webfs/buq.c

   1 #include <u.h>
   2 #include <libc.h>
   3 #include <ctype.h>
   4 #include <fcall.h>
   5 #include <thread.h>
   6 #include <9p.h>
   7
   8 #include "dat.h"
   9 #include "fns.h"
  10
  11 static void
  12 kickwqr(Buq *q, Req *r)
  13 {
  14         Buf **bb, *b;
  15         int l;
  16
  17         for(bb = &q->bh; q->nwq > 0; bb = &b->next){
  18                 if((b = *bb) == nil)
  19                         break;
  20                 if(b->wreq == nil || (b->wreq != r && r != nil))
  21                         continue;
  22                 l = b->ep - b->rp;
  23                 b = realloc(b, sizeof(*b) + l);
  24                 *bb = b;
  25                 if(b->next == nil)
  26                         q->bt = &b->next;
  27                 memmove(b->end, b->rp, l);
  28                 b->rp = b->end;
  29                 b->ep = b->rp + l;
  30                 b->wreq->ofcall.count = b->wreq->ifcall.count;
  31                 respond(b->wreq, q->error);
  32                 b->wreq = nil;
  33                 q->nwq--;
  34         }
  35 }
  36
  37 static void
  38 matchreq(Buq *q)
  39 {
  40         Req *r;
  41         Buf *b;
  42         int l;
  43
  44         while(r = q->rh){
  45                 if((b = q->bh) == nil){
  46                         if(q->closed){
  47                                 if((q->rh = r->aux) == nil)
  48                                         q->rt = &q->rh;
  49                                 if(r->ifcall.type == Tread)
  50                                         r->ofcall.count = 0;
  51                                 respond(r, q->error);
  52                                 continue;
  53                         }
  54                         break;
  55                 }
  56                 if((q->rh = r->aux) == nil)
  57                         q->rt = &q->rh;
  58                 if(r->ifcall.type == Topen){
  59                         respond(r, nil);
  60                         continue;
  61                 }
  62                 l = b->ep - b->rp;
  63                 if(l > r->ifcall.count)
  64                         l = r->ifcall.count;
  65                 memmove(r->ofcall.data, b->rp, l);
  66                 r->ofcall.count = l;
  67                 respond(r, nil);
  68                 b->rp += l;
  69                 q->size -= l;
  70                 if(b->rp >= b->ep){
  71                         if((q->bh = b->next) == nil)
  72                                 q->bt = &q->bh;
  73                         if(r = b->wreq){
  74                                 r->ofcall.count = r->ifcall.count;
  75                                 respond(r, nil);
  76                                 q->nwq--;
  77                         }
  78                         free(b);
  79                 }
  80         }
  81         if(q->closed && q->nwq > 0)
  82                 kickwqr(q, nil);
  83         rwakeupall(&q->rz);
  84 }
  85
  86 int
  87 buread(Buq *q, void *v, int l)
  88 {
  89         Req *r;
  90         Buf *b;
  91
  92         qlock(q);
  93         while((b = q->bh) == nil){
  94                 if(q->closed){
  95                         l = 0;
  96                         if(q->error){
  97                                 werrstr("%s", q->error);
  98                                 l = -1;
  99                         }
 100                         qunlock(q);
 101                         return l;
 102                 }
 103                 rsleep(&q->rz);
 104         }
 105         if(l > (b->ep - b->rp))
 106                 l = b->ep - b->rp;
 107         memmove(v, b->rp, l);
 108         b->rp += l;
 109         q->size -= l;
 110         rwakeup(&q->rz);
 111         if(b->rp < b->ep){
 112                 qunlock(q);
 113                 return l;
 114         }
 115         if((q->bh = b->next) == nil)
 116                 q->bt = &q->bh;
 117         qunlock(q);
 118         if(r = b->wreq){
 119                 r->ofcall.count = r->ifcall.count;
 120                 respond(r, nil);
 121         }
 122         free(b);
 123         return l;
 124 }
 125
 126 int
 127 buwrite(Buq *q, void *v, int l)
 128 {
 129         Buf *b;
 130
 131         b = emalloc(sizeof(*b) + l);
 132         b->wreq = nil;
 133         b->rp = b->end;
 134         b->ep = b->rp + l;
 135         memmove(b->rp, v, l);
 136         b->next = nil;
 137         qlock(q);
 138         if(q->closed){
 139                 l = 0;
 140                 if(q->error){
 141                         werrstr("%s", q->error);
 142                         l = -1;
 143                 }
 144                 qunlock(q);
 145                 free(b);
 146                 return l;
 147         }
 148         *q->bt = b;
 149         q->bt = &b->next;
 150         q->size += l;
 151         matchreq(q);
 152         while(!q->closed && q->size >= q->limit)
 153                 rsleep(&q->rz);
 154         qunlock(q);
 155         return l;
 156 }
 157
 158 void
 159 buclose(Buq *q, char *error)
 160 {
 161         if(q == nil)
 162                 return;
 163         qlock(q);
 164         if(!q->closed){
 165                 if(error)
 166                         q->error = estrdup9p(error);
 167                 q->closed = 1;
 168         }
 169         matchreq(q);
 170         qunlock(q);
 171 }
 172
 173 Buq*
 174 bualloc(int limit)
 175 {
 176         Buq *q;
 177
 178         q = emalloc(sizeof(*q));
 179         q->limit = limit;
 180         q->rt = &q->rh;
 181         q->bt = &q->bh;
 182         q->rz.l = q;
 183         incref(q);
 184         return q;
 185 }
 186
 187 void
 188 bufree(Buq *q)
 189 {
 190         Buf *b;
 191         Key *k;
 192
 193         if(q == nil || decref(q))
 194                 return;
 195         while(b = q->bh){
 196                 q->bh = b->next;
 197                 free(b);
 198         }
 199         freeurl(q->url);
 200         while(k = q->hdr){
 201                 q->hdr = k->next;
 202                 free(k);
 203         }
 204         free(q->error);
 205         free(q);
 206 }
 207
 208 void
 209 bureq(Buq *q, Req *r)
 210 {
 211         Buf *b;
 212         int l;
 213
 214         switch(r->ifcall.type){
 215         default:
 216                 respond(r, "bug in bureq");
 217                 return;
 218         case Twrite:
 219                 l = r->ifcall.count;
 220                 if(!q->closed && (q->size + l) < q->limit){
 221                         r->ofcall.count = buwrite(q, r->ifcall.data, r->ifcall.count);
 222                         respond(r, nil);
 223                         return;
 224                 }
 225                 b = emalloc(sizeof(*b));
 226                 b->wreq = r;
 227                 b->rp = (uchar*)r->ifcall.data;
 228                 b->ep = b->rp + l;
 229                 b->next = nil;
 230                 qlock(q);
 231                 *q->bt = b;
 232                 q->bt = &b->next;
 233                 q->size += l;
 234                 q->nwq++;
 235                 break;
 236         case Tread:
 237         case Topen:
 238                 r->aux = nil;
 239                 qlock(q);
 240                 *q->rt = r;
 241                 q->rt = (Req**)&r->aux;
 242                 break;
 243         }
 244         matchreq(q);
 245         qunlock(q);
 246 }
 247
 248 void
 249 buflushreq(Buq *q, Req *r)
 250 {
 251         Req **rr;
 252
 253         switch(r->ifcall.type){
 254         default:
 255                 respond(r, "bug in bufflushreq");
 256                 return;
 257         case Twrite:
 258                 qlock(q);
 259                 kickwqr(q, r);
 260                 break;
 261         case Topen:
 262         case Tread:
 263                 qlock(q);
 264                 for(rr = &q->rh; *rr; rr = (Req**)&((*rr)->aux)){
 265                         if(*rr != r)
 266                                 continue;
 267                         if((*rr = r->aux) == nil)
 268                                 q->rt = rr;
 269                         respond(r, "interrupted");
 270                         break;
 271                 }
 272                 break;
 273         }
 274         qunlock(q);
 275 }