qio.c 23 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540
  1. #include "u.h"
  2. #include "../port/lib.h"
  3. #include "mem.h"
  4. #include "dat.h"
  5. #include "fns.h"
  6. #include "../port/error.h"
  7. static ulong padblockcnt;
  8. static ulong concatblockcnt;
  9. static ulong pullupblockcnt;
  10. static ulong copyblockcnt;
  11. static ulong consumecnt;
  12. static ulong producecnt;
  13. static ulong qcopycnt;
  14. static int debugging;
  15. #define QDEBUG if(0)
  16. /*
  17. * IO queues
  18. */
  19. typedef struct Queue Queue;
  20. struct Queue
  21. {
  22. Lock;
  23. Block* bfirst; /* buffer */
  24. Block* blast;
  25. int len; /* bytes allocated to queue */
  26. int dlen; /* data bytes in queue */
  27. int limit; /* max bytes in queue */
  28. int inilim; /* initial limit */
  29. int state;
  30. int noblock; /* true if writes return immediately when q full */
  31. int eof; /* number of eofs read by user */
  32. void (*kick)(void*); /* restart output */
  33. void (*bypass)(void*, Block*); /* bypass queue altogether */
  34. void* arg; /* argument to kick */
  35. QLock rlock; /* mutex for reading processes */
  36. Rendez rr; /* process waiting to read */
  37. QLock wlock; /* mutex for writing processes */
  38. Rendez wr; /* process waiting to write */
  39. char err[ERRMAX];
  40. };
  41. enum
  42. {
  43. Maxatomic = 64*1024,
  44. };
  45. uint qiomaxatomic = Maxatomic;
  46. void
  47. ixsummary(void)
  48. {
  49. debugging ^= 1;
  50. iallocsummary();
  51. print("pad %lud, concat %lud, pullup %lud, copy %lud\n",
  52. padblockcnt, concatblockcnt, pullupblockcnt, copyblockcnt);
  53. print("consume %lud, produce %lud, qcopy %lud\n",
  54. consumecnt, producecnt, qcopycnt);
  55. }
  56. /*
  57. * free a list of blocks
  58. */
  59. void
  60. freeblist(Block *b)
  61. {
  62. Block *next;
  63. for(; b != 0; b = next){
  64. next = b->next;
  65. if(b->ref == 1)
  66. b->next = nil;
  67. freeb(b);
  68. }
  69. }
  70. /*
  71. * pad a block to the front (or the back if size is negative)
  72. */
  73. Block*
  74. padblock(Block *bp, int size)
  75. {
  76. int n;
  77. Block *nbp;
  78. QDEBUG checkb(bp, "padblock 1");
  79. if(size >= 0){
  80. if(bp->rp - bp->base >= size){
  81. bp->rp -= size;
  82. return bp;
  83. }
  84. if(bp->next)
  85. panic("padblock %#p", getcallerpc(&bp));
  86. n = BLEN(bp);
  87. padblockcnt++;
  88. nbp = allocb(size+n);
  89. nbp->rp += size;
  90. nbp->wp = nbp->rp;
  91. memmove(nbp->wp, bp->rp, n);
  92. nbp->wp += n;
  93. freeb(bp);
  94. nbp->rp -= size;
  95. } else {
  96. size = -size;
  97. if(bp->next)
  98. panic("padblock %#p", getcallerpc(&bp));
  99. if(bp->lim - bp->wp >= size)
  100. return bp;
  101. n = BLEN(bp);
  102. padblockcnt++;
  103. nbp = allocb(size+n);
  104. memmove(nbp->wp, bp->rp, n);
  105. nbp->wp += n;
  106. freeb(bp);
  107. }
  108. QDEBUG checkb(nbp, "padblock 1");
  109. return nbp;
  110. }
  111. /*
  112. * return count of bytes in a string of blocks
  113. */
  114. int
  115. blocklen(Block *bp)
  116. {
  117. int len;
  118. len = 0;
  119. while(bp) {
  120. len += BLEN(bp);
  121. bp = bp->next;
  122. }
  123. return len;
  124. }
  125. /*
  126. * return count of space in blocks
  127. */
  128. int
  129. blockalloclen(Block *bp)
  130. {
  131. int len;
  132. len = 0;
  133. while(bp) {
  134. len += BALLOC(bp);
  135. bp = bp->next;
  136. }
  137. return len;
  138. }
  139. /*
  140. * copy the string of blocks into
  141. * a single block and free the string
  142. */
  143. Block*
  144. concatblock(Block *bp)
  145. {
  146. int len;
  147. Block *nb, *f;
  148. if(bp->next == 0)
  149. return bp;
  150. nb = allocb(blocklen(bp));
  151. for(f = bp; f; f = f->next) {
  152. len = BLEN(f);
  153. memmove(nb->wp, f->rp, len);
  154. nb->wp += len;
  155. }
  156. concatblockcnt += BLEN(nb);
  157. freeblist(bp);
  158. QDEBUG checkb(nb, "concatblock 1");
  159. return nb;
  160. }
  161. /*
  162. * make sure the first block has at least n bytes
  163. */
  164. Block*
  165. pullupblock(Block *bp, int n)
  166. {
  167. int i;
  168. Block *nbp;
  169. /*
  170. * this should almost always be true, it's
  171. * just to avoid every caller checking.
  172. */
  173. if(BLEN(bp) >= n)
  174. return bp;
  175. /*
  176. * if not enough room in the first block,
  177. * add another to the front of the list.
  178. */
  179. if(bp->lim - bp->rp < n){
  180. nbp = allocb(n);
  181. nbp->next = bp;
  182. bp = nbp;
  183. }
  184. /*
  185. * copy bytes from the trailing blocks into the first
  186. */
  187. n -= BLEN(bp);
  188. while(nbp = bp->next){
  189. i = BLEN(nbp);
  190. if(i > n) {
  191. memmove(bp->wp, nbp->rp, n);
  192. pullupblockcnt++;
  193. bp->wp += n;
  194. nbp->rp += n;
  195. QDEBUG checkb(bp, "pullupblock 1");
  196. return bp;
  197. } else {
  198. /* shouldn't happen but why crash if it does */
  199. if(i < 0){
  200. print("pullup negative length packet, called from %#p\n",
  201. getcallerpc(&bp));
  202. i = 0;
  203. }
  204. memmove(bp->wp, nbp->rp, i);
  205. pullupblockcnt++;
  206. bp->wp += i;
  207. bp->next = nbp->next;
  208. nbp->next = 0;
  209. freeb(nbp);
  210. n -= i;
  211. if(n == 0){
  212. QDEBUG checkb(bp, "pullupblock 2");
  213. return bp;
  214. }
  215. }
  216. }
  217. freeb(bp);
  218. return 0;
  219. }
  220. /*
  221. * make sure the first block has at least n bytes
  222. */
  223. Block*
  224. pullupqueue(Queue *q, int n)
  225. {
  226. Block *b;
  227. if(BLEN(q->bfirst) >= n)
  228. return q->bfirst;
  229. q->bfirst = pullupblock(q->bfirst, n);
  230. for(b = q->bfirst; b != nil && b->next != nil; b = b->next)
  231. ;
  232. q->blast = b;
  233. return q->bfirst;
  234. }
  235. /*
  236. * trim to len bytes starting at offset
  237. */
  238. Block *
  239. trimblock(Block *bp, int offset, int len)
  240. {
  241. ulong l;
  242. Block *nb, *startb;
  243. QDEBUG checkb(bp, "trimblock 1");
  244. if(blocklen(bp) < offset+len) {
  245. freeblist(bp);
  246. return nil;
  247. }
  248. while((l = BLEN(bp)) < offset) {
  249. offset -= l;
  250. nb = bp->next;
  251. bp->next = nil;
  252. freeb(bp);
  253. bp = nb;
  254. }
  255. startb = bp;
  256. bp->rp += offset;
  257. while((l = BLEN(bp)) < len) {
  258. len -= l;
  259. bp = bp->next;
  260. }
  261. bp->wp -= (BLEN(bp) - len);
  262. if(bp->next) {
  263. freeblist(bp->next);
  264. bp->next = nil;
  265. }
  266. return startb;
  267. }
  268. /*
  269. * copy 'count' bytes into a new block
  270. */
  271. Block*
  272. copyblock(Block *bp, int count)
  273. {
  274. int l;
  275. Block *nbp;
  276. QDEBUG checkb(bp, "copyblock 0");
  277. nbp = allocb(count);
  278. for(; count > 0 && bp != 0; bp = bp->next){
  279. l = BLEN(bp);
  280. if(l > count)
  281. l = count;
  282. memmove(nbp->wp, bp->rp, l);
  283. nbp->wp += l;
  284. count -= l;
  285. }
  286. if(count > 0){
  287. memset(nbp->wp, 0, count);
  288. nbp->wp += count;
  289. }
  290. copyblockcnt++;
  291. QDEBUG checkb(nbp, "copyblock 1");
  292. return nbp;
  293. }
  294. Block*
  295. adjustblock(Block* bp, int len)
  296. {
  297. int n;
  298. Block *nbp;
  299. if(len < 0){
  300. freeb(bp);
  301. return nil;
  302. }
  303. if(bp->rp+len > bp->lim){
  304. nbp = copyblock(bp, len);
  305. freeblist(bp);
  306. QDEBUG checkb(nbp, "adjustblock 1");
  307. return nbp;
  308. }
  309. n = BLEN(bp);
  310. if(len > n)
  311. memset(bp->wp, 0, len-n);
  312. bp->wp = bp->rp+len;
  313. QDEBUG checkb(bp, "adjustblock 2");
  314. return bp;
  315. }
  316. /*
  317. * throw away up to count bytes from a
  318. * list of blocks. Return count of bytes
  319. * thrown away.
  320. */
  321. int
  322. pullblock(Block **bph, int count)
  323. {
  324. Block *bp;
  325. int n, bytes;
  326. bytes = 0;
  327. if(bph == nil)
  328. return 0;
  329. while(*bph != nil && count != 0) {
  330. bp = *bph;
  331. n = BLEN(bp);
  332. if(count < n)
  333. n = count;
  334. bytes += n;
  335. count -= n;
  336. bp->rp += n;
  337. QDEBUG checkb(bp, "pullblock ");
  338. if(BLEN(bp) == 0) {
  339. *bph = bp->next;
  340. bp->next = nil;
  341. freeb(bp);
  342. }
  343. }
  344. return bytes;
  345. }
  346. /*
  347. * get next block from a queue, return null if nothing there
  348. */
  349. Block*
  350. qget(Queue *q)
  351. {
  352. int dowakeup;
  353. Block *b;
  354. /* sync with qwrite */
  355. ilock(q);
  356. b = q->bfirst;
  357. if(b == nil){
  358. q->state |= Qstarve;
  359. iunlock(q);
  360. return nil;
  361. }
  362. q->bfirst = b->next;
  363. b->next = 0;
  364. q->len -= BALLOC(b);
  365. q->dlen -= BLEN(b);
  366. QDEBUG checkb(b, "qget");
  367. /* if writer flow controlled, restart */
  368. if((q->state & Qflow) && q->len < q->limit/2){
  369. q->state &= ~Qflow;
  370. dowakeup = 1;
  371. } else
  372. dowakeup = 0;
  373. iunlock(q);
  374. if(dowakeup)
  375. wakeup(&q->wr);
  376. return b;
  377. }
  378. /*
  379. * throw away the next 'len' bytes in the queue
  380. */
  381. int
  382. qdiscard(Queue *q, int len)
  383. {
  384. Block *b;
  385. int dowakeup, n, sofar;
  386. ilock(q);
  387. for(sofar = 0; sofar < len; sofar += n){
  388. b = q->bfirst;
  389. if(b == nil)
  390. break;
  391. QDEBUG checkb(b, "qdiscard");
  392. n = BLEN(b);
  393. if(n <= len - sofar){
  394. q->bfirst = b->next;
  395. b->next = 0;
  396. q->len -= BALLOC(b);
  397. q->dlen -= BLEN(b);
  398. freeb(b);
  399. } else {
  400. n = len - sofar;
  401. b->rp += n;
  402. q->dlen -= n;
  403. }
  404. }
  405. /*
  406. * if writer flow controlled, restart
  407. *
  408. * This used to be
  409. * q->len < q->limit/2
  410. * but it slows down tcp too much for certain write sizes.
  411. * I really don't understand it completely. It may be
  412. * due to the queue draining so fast that the transmission
  413. * stalls waiting for the app to produce more data. - presotto
  414. */
  415. if((q->state & Qflow) && q->len < q->limit){
  416. q->state &= ~Qflow;
  417. dowakeup = 1;
  418. } else
  419. dowakeup = 0;
  420. iunlock(q);
  421. if(dowakeup)
  422. wakeup(&q->wr);
  423. return sofar;
  424. }
  425. /*
  426. * Interrupt level copy out of a queue, return # bytes copied.
  427. */
  428. int
  429. qconsume(Queue *q, void *vp, int len)
  430. {
  431. Block *b;
  432. int n, dowakeup;
  433. uchar *p = vp;
  434. Block *tofree = nil;
  435. /* sync with qwrite */
  436. ilock(q);
  437. for(;;) {
  438. b = q->bfirst;
  439. if(b == 0){
  440. q->state |= Qstarve;
  441. iunlock(q);
  442. return -1;
  443. }
  444. QDEBUG checkb(b, "qconsume 1");
  445. n = BLEN(b);
  446. if(n > 0)
  447. break;
  448. q->bfirst = b->next;
  449. q->len -= BALLOC(b);
  450. /* remember to free this */
  451. b->next = tofree;
  452. tofree = b;
  453. };
  454. if(n < len)
  455. len = n;
  456. memmove(p, b->rp, len);
  457. consumecnt += n;
  458. b->rp += len;
  459. q->dlen -= len;
  460. /* discard the block if we're done with it */
  461. if((q->state & Qmsg) || len == n){
  462. q->bfirst = b->next;
  463. b->next = 0;
  464. q->len -= BALLOC(b);
  465. q->dlen -= BLEN(b);
  466. /* remember to free this */
  467. b->next = tofree;
  468. tofree = b;
  469. }
  470. /* if writer flow controlled, restart */
  471. if((q->state & Qflow) && q->len < q->limit/2){
  472. q->state &= ~Qflow;
  473. dowakeup = 1;
  474. } else
  475. dowakeup = 0;
  476. iunlock(q);
  477. if(dowakeup)
  478. wakeup(&q->wr);
  479. if(tofree != nil)
  480. freeblist(tofree);
  481. return len;
  482. }
  483. int
  484. qpass(Queue *q, Block *b)
  485. {
  486. int dlen, len, dowakeup;
  487. /* sync with qread */
  488. dowakeup = 0;
  489. ilock(q);
  490. if(q->len >= q->limit){
  491. freeblist(b);
  492. iunlock(q);
  493. return -1;
  494. }
  495. if(q->state & Qclosed){
  496. len = BALLOC(b);
  497. freeblist(b);
  498. iunlock(q);
  499. return len;
  500. }
  501. /* add buffer to queue */
  502. if(q->bfirst)
  503. q->blast->next = b;
  504. else
  505. q->bfirst = b;
  506. len = BALLOC(b);
  507. dlen = BLEN(b);
  508. QDEBUG checkb(b, "qpass");
  509. while(b->next){
  510. b = b->next;
  511. QDEBUG checkb(b, "qpass");
  512. len += BALLOC(b);
  513. dlen += BLEN(b);
  514. }
  515. q->blast = b;
  516. q->len += len;
  517. q->dlen += dlen;
  518. if(q->len >= q->limit/2)
  519. q->state |= Qflow;
  520. if(q->state & Qstarve){
  521. q->state &= ~Qstarve;
  522. dowakeup = 1;
  523. }
  524. iunlock(q);
  525. if(dowakeup)
  526. wakeup(&q->rr);
  527. return len;
  528. }
  529. int
  530. qpassnolim(Queue *q, Block *b)
  531. {
  532. int dlen, len, dowakeup;
  533. /* sync with qread */
  534. dowakeup = 0;
  535. ilock(q);
  536. if(q->state & Qclosed){
  537. freeblist(b);
  538. iunlock(q);
  539. return BALLOC(b);
  540. }
  541. /* add buffer to queue */
  542. if(q->bfirst)
  543. q->blast->next = b;
  544. else
  545. q->bfirst = b;
  546. len = BALLOC(b);
  547. dlen = BLEN(b);
  548. QDEBUG checkb(b, "qpass");
  549. while(b->next){
  550. b = b->next;
  551. QDEBUG checkb(b, "qpass");
  552. len += BALLOC(b);
  553. dlen += BLEN(b);
  554. }
  555. q->blast = b;
  556. q->len += len;
  557. q->dlen += dlen;
  558. if(q->len >= q->limit/2)
  559. q->state |= Qflow;
  560. if(q->state & Qstarve){
  561. q->state &= ~Qstarve;
  562. dowakeup = 1;
  563. }
  564. iunlock(q);
  565. if(dowakeup)
  566. wakeup(&q->rr);
  567. return len;
  568. }
  569. /*
  570. * if the allocated space is way out of line with the used
  571. * space, reallocate to a smaller block
  572. */
  573. Block*
  574. packblock(Block *bp)
  575. {
  576. Block **l, *nbp;
  577. int n;
  578. for(l = &bp; *l; l = &(*l)->next){
  579. nbp = *l;
  580. n = BLEN(nbp);
  581. if((n<<2) < BALLOC(nbp)){
  582. *l = allocb(n);
  583. memmove((*l)->wp, nbp->rp, n);
  584. (*l)->wp += n;
  585. (*l)->next = nbp->next;
  586. freeb(nbp);
  587. }
  588. }
  589. return bp;
  590. }
  591. int
  592. qproduce(Queue *q, void *vp, int len)
  593. {
  594. Block *b;
  595. int dowakeup;
  596. uchar *p = vp;
  597. /* sync with qread */
  598. dowakeup = 0;
  599. ilock(q);
  600. /* no waiting receivers, room in buffer? */
  601. if(q->len >= q->limit){
  602. q->state |= Qflow;
  603. iunlock(q);
  604. return -1;
  605. }
  606. /* save in buffer */
  607. b = iallocb(len);
  608. if(b == 0){
  609. iunlock(q);
  610. return 0;
  611. }
  612. memmove(b->wp, p, len);
  613. producecnt += len;
  614. b->wp += len;
  615. if(q->bfirst)
  616. q->blast->next = b;
  617. else
  618. q->bfirst = b;
  619. q->blast = b;
  620. /* b->next = 0; done by iallocb() */
  621. q->len += BALLOC(b);
  622. q->dlen += BLEN(b);
  623. QDEBUG checkb(b, "qproduce");
  624. if(q->state & Qstarve){
  625. q->state &= ~Qstarve;
  626. dowakeup = 1;
  627. }
  628. if(q->len >= q->limit)
  629. q->state |= Qflow;
  630. iunlock(q);
  631. if(dowakeup)
  632. wakeup(&q->rr);
  633. return len;
  634. }
  635. /*
  636. * copy from offset in the queue
  637. */
  638. Block*
  639. qcopy(Queue *q, int len, ulong offset)
  640. {
  641. int sofar;
  642. int n;
  643. Block *b, *nb;
  644. uchar *p;
  645. nb = allocb(len);
  646. ilock(q);
  647. /* go to offset */
  648. b = q->bfirst;
  649. for(sofar = 0; ; sofar += n){
  650. if(b == nil){
  651. iunlock(q);
  652. return nb;
  653. }
  654. n = BLEN(b);
  655. if(sofar + n > offset){
  656. p = b->rp + offset - sofar;
  657. n -= offset - sofar;
  658. break;
  659. }
  660. QDEBUG checkb(b, "qcopy");
  661. b = b->next;
  662. }
  663. /* copy bytes from there */
  664. for(sofar = 0; sofar < len;){
  665. if(n > len - sofar)
  666. n = len - sofar;
  667. memmove(nb->wp, p, n);
  668. qcopycnt += n;
  669. sofar += n;
  670. nb->wp += n;
  671. b = b->next;
  672. if(b == nil)
  673. break;
  674. n = BLEN(b);
  675. p = b->rp;
  676. }
  677. iunlock(q);
  678. return nb;
  679. }
  680. /*
  681. * called by non-interrupt code
  682. */
  683. Queue*
  684. qopen(int limit, int msg, void (*kick)(void*), void *arg)
  685. {
  686. Queue *q;
  687. q = malloc(sizeof(Queue));
  688. if(q == 0)
  689. return 0;
  690. q->limit = q->inilim = limit;
  691. q->kick = kick;
  692. q->arg = arg;
  693. q->state = msg;
  694. q->state |= Qstarve;
  695. q->eof = 0;
  696. q->noblock = 0;
  697. return q;
  698. }
  699. /* open a queue to be bypassed */
  700. Queue*
  701. qbypass(void (*bypass)(void*, Block*), void *arg)
  702. {
  703. Queue *q;
  704. q = malloc(sizeof(Queue));
  705. if(q == 0)
  706. return 0;
  707. q->limit = 0;
  708. q->arg = arg;
  709. q->bypass = bypass;
  710. q->state = 0;
  711. return q;
  712. }
  713. static int
  714. notempty(void *a)
  715. {
  716. Queue *q = a;
  717. return (q->state & Qclosed) || q->bfirst != 0;
  718. }
  719. /*
  720. * wait for the queue to be non-empty or closed.
  721. * called with q ilocked.
  722. */
  723. static int
  724. qwait(Queue *q)
  725. {
  726. /* wait for data */
  727. for(;;){
  728. if(q->bfirst != nil)
  729. break;
  730. if(q->state & Qclosed){
  731. if(++q->eof > 3)
  732. return -1;
  733. if(*q->err && strcmp(q->err, Ehungup) != 0)
  734. return -1;
  735. return 0;
  736. }
  737. q->state |= Qstarve; /* flag requesting producer to wake me */
  738. iunlock(q);
  739. sleep(&q->rr, notempty, q);
  740. ilock(q);
  741. }
  742. return 1;
  743. }
  744. /*
  745. * add a block list to a queue
  746. */
  747. void
  748. qaddlist(Queue *q, Block *b)
  749. {
  750. /* queue the block */
  751. if(q->bfirst)
  752. q->blast->next = b;
  753. else
  754. q->bfirst = b;
  755. q->len += blockalloclen(b);
  756. q->dlen += blocklen(b);
  757. while(b->next)
  758. b = b->next;
  759. q->blast = b;
  760. }
  761. /*
  762. * called with q ilocked
  763. */
  764. Block*
  765. qremove(Queue *q)
  766. {
  767. Block *b;
  768. b = q->bfirst;
  769. if(b == nil)
  770. return nil;
  771. q->bfirst = b->next;
  772. b->next = nil;
  773. q->dlen -= BLEN(b);
  774. q->len -= BALLOC(b);
  775. QDEBUG checkb(b, "qremove");
  776. return b;
  777. }
  778. /*
  779. * copy the contents of a string of blocks into
  780. * memory. emptied blocks are freed. return
  781. * pointer to first unconsumed block.
  782. */
  783. Block*
  784. bl2mem(uchar *p, Block *b, int n)
  785. {
  786. int i;
  787. Block *next;
  788. for(; b != nil; b = next){
  789. i = BLEN(b);
  790. if(i > n){
  791. memmove(p, b->rp, n);
  792. b->rp += n;
  793. return b;
  794. }
  795. memmove(p, b->rp, i);
  796. n -= i;
  797. p += i;
  798. b->rp += i;
  799. next = b->next;
  800. freeb(b);
  801. }
  802. return nil;
  803. }
  804. /*
  805. * copy the contents of memory into a string of blocks.
  806. * return nil on error.
  807. */
  808. Block*
  809. mem2bl(uchar *p, int len)
  810. {
  811. int n;
  812. Block *b, *first, **l;
  813. first = nil;
  814. l = &first;
  815. if(waserror()){
  816. freeblist(first);
  817. nexterror();
  818. }
  819. do {
  820. n = len;
  821. if(n > Maxatomic)
  822. n = Maxatomic;
  823. *l = b = allocb(n);
  824. setmalloctag(b, (up->text[0]<<24)|(up->text[1]<<16)|(up->text[2]<<8)|up->text[3]);
  825. memmove(b->wp, p, n);
  826. b->wp += n;
  827. p += n;
  828. len -= n;
  829. l = &b->next;
  830. } while(len > 0);
  831. poperror();
  832. return first;
  833. }
  834. /*
  835. * put a block back to the front of the queue
  836. * called with q ilocked
  837. */
  838. void
  839. qputback(Queue *q, Block *b)
  840. {
  841. b->next = q->bfirst;
  842. if(q->bfirst == nil)
  843. q->blast = b;
  844. q->bfirst = b;
  845. q->len += BALLOC(b);
  846. q->dlen += BLEN(b);
  847. }
  848. /*
  849. * flow control, get producer going again
  850. * called with q ilocked
  851. */
  852. static void
  853. qwakeup_iunlock(Queue *q)
  854. {
  855. int dowakeup = 0;
  856. /* if writer flow controlled, restart */
  857. if((q->state & Qflow) && q->len < q->limit/2){
  858. q->state &= ~Qflow;
  859. dowakeup = 1;
  860. }
  861. iunlock(q);
  862. /* wakeup flow controlled writers */
  863. if(dowakeup){
  864. if(q->kick)
  865. q->kick(q->arg);
  866. wakeup(&q->wr);
  867. }
  868. }
  869. /*
  870. * get next block from a queue (up to a limit)
  871. */
  872. Block*
  873. qbread(Queue *q, int len)
  874. {
  875. Block *b, *nb;
  876. int n;
  877. qlock(&q->rlock);
  878. if(waserror()){
  879. qunlock(&q->rlock);
  880. nexterror();
  881. }
  882. ilock(q);
  883. switch(qwait(q)){
  884. case 0:
  885. /* queue closed */
  886. iunlock(q);
  887. qunlock(&q->rlock);
  888. poperror();
  889. return nil;
  890. case -1:
  891. /* multiple reads on a closed queue */
  892. iunlock(q);
  893. error(q->err);
  894. }
  895. /* if we get here, there's at least one block in the queue */
  896. b = qremove(q);
  897. n = BLEN(b);
  898. /* split block if it's too big and this is not a message queue */
  899. nb = b;
  900. if(n > len){
  901. if((q->state&Qmsg) == 0){
  902. n -= len;
  903. b = allocb(n);
  904. memmove(b->wp, nb->rp+len, n);
  905. b->wp += n;
  906. qputback(q, b);
  907. }
  908. nb->wp = nb->rp + len;
  909. }
  910. /* restart producer */
  911. qwakeup_iunlock(q);
  912. poperror();
  913. qunlock(&q->rlock);
  914. return nb;
  915. }
  916. /*
  917. * read a queue. if no data is queued, post a Block
  918. * and wait on its Rendez.
  919. */
  920. long
  921. qread(Queue *q, void *vp, int len)
  922. {
  923. Block *b, *first, **l;
  924. int m, n;
  925. qlock(&q->rlock);
  926. if(waserror()){
  927. qunlock(&q->rlock);
  928. nexterror();
  929. }
  930. ilock(q);
  931. again:
  932. switch(qwait(q)){
  933. case 0:
  934. /* queue closed */
  935. iunlock(q);
  936. qunlock(&q->rlock);
  937. poperror();
  938. return 0;
  939. case -1:
  940. /* multiple reads on a closed queue */
  941. iunlock(q);
  942. error(q->err);
  943. }
  944. /* if we get here, there's at least one block in the queue */
  945. if(q->state & Qcoalesce){
  946. /* when coalescing, 0 length blocks just go away */
  947. b = q->bfirst;
  948. if(BLEN(b) <= 0){
  949. freeb(qremove(q));
  950. goto again;
  951. }
  952. /* grab the first block plus as many
  953. * following blocks as will completely
  954. * fit in the read.
  955. */
  956. n = 0;
  957. l = &first;
  958. m = BLEN(b);
  959. for(;;) {
  960. *l = qremove(q);
  961. l = &b->next;
  962. n += m;
  963. b = q->bfirst;
  964. if(b == nil)
  965. break;
  966. m = BLEN(b);
  967. if(n+m > len)
  968. break;
  969. }
  970. } else {
  971. first = qremove(q);
  972. n = BLEN(first);
  973. }
  974. /* copy to user space outside of the ilock */
  975. iunlock(q);
  976. b = bl2mem(vp, first, len);
  977. ilock(q);
  978. /* take care of any left over partial block */
  979. if(b != nil){
  980. n -= BLEN(b);
  981. if(q->state & Qmsg)
  982. freeb(b);
  983. else
  984. qputback(q, b);
  985. }
  986. /* restart producer */
  987. qwakeup_iunlock(q);
  988. poperror();
  989. qunlock(&q->rlock);
  990. return n;
  991. }
  992. static int
  993. qnotfull(void *a)
  994. {
  995. Queue *q = a;
  996. return q->len < q->limit || (q->state & Qclosed);
  997. }
  998. ulong noblockcnt;
  999. /*
  1000. * add a block to a queue obeying flow control
  1001. */
  1002. long
  1003. qbwrite(Queue *q, Block *b)
  1004. {
  1005. int n, dowakeup;
  1006. Proc *p;
  1007. n = BLEN(b);
  1008. if(q->bypass){
  1009. (*q->bypass)(q->arg, b);
  1010. return n;
  1011. }
  1012. dowakeup = 0;
  1013. qlock(&q->wlock);
  1014. if(waserror()){
  1015. if(b != nil)
  1016. freeb(b);
  1017. qunlock(&q->wlock);
  1018. nexterror();
  1019. }
  1020. ilock(q);
  1021. /* give up if the queue is closed */
  1022. if(q->state & Qclosed){
  1023. iunlock(q);
  1024. error(q->err);
  1025. }
  1026. /* if nonblocking, don't queue over the limit */
  1027. if(q->len >= q->limit){
  1028. if(q->noblock){
  1029. iunlock(q);
  1030. freeb(b);
  1031. noblockcnt += n;
  1032. qunlock(&q->wlock);
  1033. poperror();
  1034. return n;
  1035. }
  1036. }
  1037. /* queue the block */
  1038. if(q->bfirst)
  1039. q->blast->next = b;
  1040. else
  1041. q->bfirst = b;
  1042. q->blast = b;
  1043. b->next = 0;
  1044. q->len += BALLOC(b);
  1045. q->dlen += n;
  1046. QDEBUG checkb(b, "qbwrite");
  1047. b = nil;
  1048. /* make sure other end gets awakened */
  1049. if(q->state & Qstarve){
  1050. q->state &= ~Qstarve;
  1051. dowakeup = 1;
  1052. }
  1053. iunlock(q);
  1054. /* get output going again */
  1055. if(q->kick && (dowakeup || (q->state&Qkick)))
  1056. q->kick(q->arg);
  1057. /* wakeup anyone consuming at the other end */
  1058. if(dowakeup){
  1059. p = wakeup(&q->rr);
  1060. /* if we just wokeup a higher priority process, let it run */
  1061. if(p != nil && p->priority > up->priority)
  1062. sched();
  1063. }
  1064. /*
  1065. * flow control, wait for queue to get below the limit
  1066. * before allowing the process to continue and queue
  1067. * more. We do this here so that postnote can only
  1068. * interrupt us after the data has been queued. This
  1069. * means that things like 9p flushes and ssl messages
  1070. * will not be disrupted by software interrupts.
  1071. *
  1072. * Note - this is moderately dangerous since a process
  1073. * that keeps getting interrupted and rewriting will
  1074. * queue infinite crud.
  1075. */
  1076. for(;;){
  1077. if(q->noblock || qnotfull(q))
  1078. break;
  1079. ilock(q);
  1080. q->state |= Qflow;
  1081. iunlock(q);
  1082. sleep(&q->wr, qnotfull, q);
  1083. }
  1084. USED(b);
  1085. qunlock(&q->wlock);
  1086. poperror();
  1087. return n;
  1088. }
  1089. /*
  1090. * write to a queue. only Maxatomic bytes at a time is atomic.
  1091. */
  1092. int
  1093. qwrite(Queue *q, void *vp, int len)
  1094. {
  1095. int n, sofar;
  1096. Block *b;
  1097. uchar *p = vp;
  1098. QDEBUG if(!islo())
  1099. print("qwrite hi %#p\n", getcallerpc(&q));
  1100. sofar = 0;
  1101. do {
  1102. n = len-sofar;
  1103. if(n > Maxatomic)
  1104. n = Maxatomic;
  1105. b = allocb(n);
  1106. setmalloctag(b, (up->text[0]<<24)|(up->text[1]<<16)|(up->text[2]<<8)|up->text[3]);
  1107. if(waserror()){
  1108. freeb(b);
  1109. nexterror();
  1110. }
  1111. memmove(b->wp, p+sofar, n);
  1112. poperror();
  1113. b->wp += n;
  1114. qbwrite(q, b);
  1115. sofar += n;
  1116. } while(sofar < len && (q->state & Qmsg) == 0);
  1117. return len;
  1118. }
  1119. /*
  1120. * used by print() to write to a queue. Since we may be splhi or not in
  1121. * a process, don't qlock.
  1122. *
  1123. * this routine merges adjacent blocks if block n+1 will fit into
  1124. * the free space of block n.
  1125. */
  1126. int
  1127. qiwrite(Queue *q, void *vp, int len)
  1128. {
  1129. int n, sofar, dowakeup;
  1130. Block *b;
  1131. uchar *p = vp;
  1132. dowakeup = 0;
  1133. sofar = 0;
  1134. do {
  1135. n = len-sofar;
  1136. if(n > Maxatomic)
  1137. n = Maxatomic;
  1138. b = iallocb(n);
  1139. if(b == nil)
  1140. break;
  1141. memmove(b->wp, p+sofar, n);
  1142. b->wp += n;
  1143. ilock(q);
  1144. /* we use an artificially high limit for kernel prints since anything
  1145. * over the limit gets dropped
  1146. */
  1147. if(q->dlen >= 16*1024){
  1148. iunlock(q);
  1149. freeb(b);
  1150. break;
  1151. }
  1152. QDEBUG checkb(b, "qiwrite");
  1153. if(q->bfirst)
  1154. q->blast->next = b;
  1155. else
  1156. q->bfirst = b;
  1157. q->blast = b;
  1158. q->len += BALLOC(b);
  1159. q->dlen += n;
  1160. if(q->state & Qstarve){
  1161. q->state &= ~Qstarve;
  1162. dowakeup = 1;
  1163. }
  1164. iunlock(q);
  1165. if(dowakeup){
  1166. if(q->kick)
  1167. q->kick(q->arg);
  1168. wakeup(&q->rr);
  1169. }
  1170. sofar += n;
  1171. } while(sofar < len && (q->state & Qmsg) == 0);
  1172. return sofar;
  1173. }
  1174. /*
  1175. * be extremely careful when calling this,
  1176. * as there is no reference accounting
  1177. */
  1178. void
  1179. qfree(Queue *q)
  1180. {
  1181. qclose(q);
  1182. free(q);
  1183. }
  1184. /*
  1185. * Mark a queue as closed. No further IO is permitted.
  1186. * All blocks are released.
  1187. */
  1188. void
  1189. qclose(Queue *q)
  1190. {
  1191. Block *bfirst;
  1192. if(q == nil)
  1193. return;
  1194. /* mark it */
  1195. ilock(q);
  1196. q->state |= Qclosed;
  1197. q->state &= ~(Qflow|Qstarve);
  1198. strcpy(q->err, Ehungup);
  1199. bfirst = q->bfirst;
  1200. q->bfirst = 0;
  1201. q->len = 0;
  1202. q->dlen = 0;
  1203. q->noblock = 0;
  1204. iunlock(q);
  1205. /* free queued blocks */
  1206. freeblist(bfirst);
  1207. /* wake up readers/writers */
  1208. wakeup(&q->rr);
  1209. wakeup(&q->wr);
  1210. }
  1211. /*
  1212. * Mark a queue as closed. Wakeup any readers. Don't remove queued
  1213. * blocks.
  1214. */
  1215. void
  1216. qhangup(Queue *q, char *msg)
  1217. {
  1218. /* mark it */
  1219. ilock(q);
  1220. q->state |= Qclosed;
  1221. if(msg == 0 || *msg == 0)
  1222. strcpy(q->err, Ehungup);
  1223. else
  1224. strncpy(q->err, msg, ERRMAX-1);
  1225. iunlock(q);
  1226. /* wake up readers/writers */
  1227. wakeup(&q->rr);
  1228. wakeup(&q->wr);
  1229. }
  1230. /*
  1231. * return non-zero if the q is hungup
  1232. */
  1233. int
  1234. qisclosed(Queue *q)
  1235. {
  1236. return q->state & Qclosed;
  1237. }
  1238. /*
  1239. * mark a queue as no longer hung up
  1240. */
  1241. void
  1242. qreopen(Queue *q)
  1243. {
  1244. ilock(q);
  1245. q->state &= ~Qclosed;
  1246. q->state |= Qstarve;
  1247. q->eof = 0;
  1248. q->limit = q->inilim;
  1249. iunlock(q);
  1250. }
  1251. /*
  1252. * return bytes queued
  1253. */
  1254. int
  1255. qlen(Queue *q)
  1256. {
  1257. return q->dlen;
  1258. }
  1259. /*
  1260. * return space remaining before flow control
  1261. */
  1262. int
  1263. qwindow(Queue *q)
  1264. {
  1265. int l;
  1266. l = q->limit - q->len;
  1267. if(l < 0)
  1268. l = 0;
  1269. return l;
  1270. }
  1271. /*
  1272. * return true if we can read without blocking
  1273. */
  1274. int
  1275. qcanread(Queue *q)
  1276. {
  1277. return q->bfirst!=0;
  1278. }
  1279. /*
  1280. * change queue limit
  1281. */
  1282. void
  1283. qsetlimit(Queue *q, int limit)
  1284. {
  1285. q->limit = limit;
  1286. }
  1287. /*
  1288. * set blocking/nonblocking
  1289. */
  1290. void
  1291. qnoblock(Queue *q, int onoff)
  1292. {
  1293. q->noblock = onoff;
  1294. }
  1295. /*
  1296. * flush the output queue
  1297. */
  1298. void
  1299. qflush(Queue *q)
  1300. {
  1301. Block *bfirst;
  1302. /* mark it */
  1303. ilock(q);
  1304. bfirst = q->bfirst;
  1305. q->bfirst = 0;
  1306. q->len = 0;
  1307. q->dlen = 0;
  1308. iunlock(q);
  1309. /* free queued blocks */
  1310. freeblist(bfirst);
  1311. /* wake up readers/writers */
  1312. wakeup(&q->wr);
  1313. }
  1314. int
  1315. qfull(Queue *q)
  1316. {
  1317. return q->state & Qflow;
  1318. }
  1319. int
  1320. qstate(Queue *q)
  1321. {
  1322. return q->state;
  1323. }