ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/libeio/eio.c
(Generate patch)

Comparing libeio/eio.c (file contents):
Revision 1.23 by root, Sun Sep 21 00:23:45 2008 UTC vs.
Revision 1.30 by root, Wed Jun 3 12:24:49 2009 UTC

… …
36 * provisions above, a recipient may use your version of this file under 36 * provisions above, a recipient may use your version of this file under
37 * either the BSD or the GPL. 37 * either the BSD or the GPL.
38 */ 38 */
39 39
40#include "eio.h" 40#include "eio.h"
41
42#ifdef EIO_STACKSIZE
43# define XTHREAD_STACKSIZE EIO_STACKSIZE
44#endif
41#include "xthread.h" 45#include "xthread.h"
42 46
43#include <errno.h> 47#include <errno.h>
44#include <stddef.h> 48#include <stddef.h>
45#include <stdlib.h> 49#include <stdlib.h>
… …
120 124
121#define EIO_TICKS ((1000000 + 1023) >> 10) 125#define EIO_TICKS ((1000000 + 1023) >> 10)
122 126
123/*****************************************************************************/ 127/*****************************************************************************/
124 128
129#if __GNUC__ >= 3
130# define expect(expr,value) __builtin_expect ((expr),(value))
131#else
132# define expect(expr,value) (expr)
133#endif
134
135#define expect_false(expr) expect ((expr) != 0, 0)
136#define expect_true(expr) expect ((expr) != 0, 1)
137
138/*****************************************************************************/
139
125#define ETP_PRI_MIN EIO_PRI_MIN 140#define ETP_PRI_MIN EIO_PRI_MIN
126#define ETP_PRI_MAX EIO_PRI_MAX 141#define ETP_PRI_MAX EIO_PRI_MAX
127 142
128struct etp_worker; 143struct etp_worker;
129 144
… …
411 X_UNLOCK (wrklock); 426 X_UNLOCK (wrklock);
412} 427}
413 428
414static void etp_maybe_start_thread (void) 429static void etp_maybe_start_thread (void)
415{ 430{
416 if (etp_nthreads () >= wanted) 431 if (expect_true (etp_nthreads () >= wanted))
417 return; 432 return;
418 433
419 /* todo: maybe use idle here, but might be less exact */ 434 /* todo: maybe use idle here, but might be less exact */
420 if (0 <= (int)etp_nthreads () + (int)etp_npending () - (int)etp_nreqs ()) 435 if (expect_true (0 <= (int)etp_nthreads () + (int)etp_npending () - (int)etp_nreqs ()))
421 return; 436 return;
422 437
423 etp_start_thread (); 438 etp_start_thread ();
424} 439}
425 440
… …
478 493
479 X_LOCK (reqlock); 494 X_LOCK (reqlock);
480 --nreqs; 495 --nreqs;
481 X_UNLOCK (reqlock); 496 X_UNLOCK (reqlock);
482 497
483 if (req->type == EIO_GROUP && req->size) 498 if (expect_false (req->type == EIO_GROUP && req->size))
484 { 499 {
485 req->int1 = 1; /* mark request as delayed */ 500 req->int1 = 1; /* mark request as delayed */
486 continue; 501 continue;
487 } 502 }
488 else 503 else
489 { 504 {
490 int res = ETP_FINISH (req); 505 int res = ETP_FINISH (req);
491 if (res) 506 if (expect_false (res))
492 return res; 507 return res;
493 } 508 }
494 509
495 if (maxreqs && !--maxreqs) 510 if (expect_false (maxreqs && !--maxreqs))
496 break; 511 break;
497 512
498 if (maxtime) 513 if (maxtime)
499 { 514 {
500 gettimeofday (&tv_now, 0); 515 gettimeofday (&tv_now, 0);
… …
519 534
520static void etp_submit (ETP_REQ *req) 535static void etp_submit (ETP_REQ *req)
521{ 536{
522 req->pri -= ETP_PRI_MIN; 537 req->pri -= ETP_PRI_MIN;
523 538
524 if (req->pri < ETP_PRI_MIN - ETP_PRI_MIN) req->pri = ETP_PRI_MIN - ETP_PRI_MIN; 539 if (expect_false (req->pri < ETP_PRI_MIN - ETP_PRI_MIN)) req->pri = ETP_PRI_MIN - ETP_PRI_MIN;
525 if (req->pri > ETP_PRI_MAX - ETP_PRI_MIN) req->pri = ETP_PRI_MAX - ETP_PRI_MIN; 540 if (expect_false (req->pri > ETP_PRI_MAX - ETP_PRI_MIN)) req->pri = ETP_PRI_MAX - ETP_PRI_MIN;
526 541
542 if (expect_false (req->type == EIO_GROUP))
543 {
544 /* I hope this is worth it :/ */
527 X_LOCK (reqlock); 545 X_LOCK (reqlock);
528 ++nreqs; 546 ++nreqs;
547 X_UNLOCK (reqlock);
548
549 X_LOCK (reslock);
550
551 ++npending;
552
553 if (!reqq_push (&res_queue, req) && want_poll_cb)
554 want_poll_cb ();
555
556 X_UNLOCK (reslock);
557 }
558 else
559 {
560 X_LOCK (reqlock);
561 ++nreqs;
529 ++nready; 562 ++nready;
530 reqq_push (&req_queue, req); 563 reqq_push (&req_queue, req);
531 X_COND_SIGNAL (reqwait); 564 X_COND_SIGNAL (reqwait);
532 X_UNLOCK (reqlock); 565 X_UNLOCK (reqlock);
533 566
534 etp_maybe_start_thread (); 567 etp_maybe_start_thread ();
568 }
535} 569}
536 570
537static void etp_set_max_poll_time (double nseconds) 571static void etp_set_max_poll_time (double nseconds)
538{ 572{
539 if (WORDACCESS_UNSAFE) X_LOCK (reslock); 573 if (WORDACCESS_UNSAFE) X_LOCK (reslock);
… …
574 608
575static void grp_try_feed (eio_req *grp) 609static void grp_try_feed (eio_req *grp)
576{ 610{
577 while (grp->size < grp->int2 && !EIO_CANCELLED (grp)) 611 while (grp->size < grp->int2 && !EIO_CANCELLED (grp))
578 { 612 {
579 int old_len = grp->size; 613 grp->flags &= ~EIO_FLAG_GROUPADD;
580 614
581 EIO_FEED (grp); 615 EIO_FEED (grp);
582 616
583 /* stop if no progress has been made */ 617 /* stop if no progress has been made */
584 if (old_len == grp->size) 618 if (!(grp->flags & EIO_FLAG_GROUPADD))
585 { 619 {
586 grp->feed = 0; 620 grp->feed = 0;
587 break; 621 break;
588 } 622 }
589 } 623 }
… …
706 740
707/*****************************************************************************/ 741/*****************************************************************************/
708/* work around various missing functions */ 742/* work around various missing functions */
709 743
710#if !HAVE_PREADWRITE 744#if !HAVE_PREADWRITE
745# undef pread
746# undef pwrite
711# define pread eio__pread 747# define pread eio__pread
712# define pwrite eio__pwrite 748# define pwrite eio__pwrite
713 749
714static ssize_t 750static ssize_t
715eio__pread (int fd, void *buf, size_t count, off_t offset) 751eio__pread (int fd, void *buf, size_t count, off_t offset)
… …
735 771
736 X_LOCK (preadwritelock); 772 X_LOCK (preadwritelock);
737 ooffset = lseek (fd, 0, SEEK_CUR); 773 ooffset = lseek (fd, 0, SEEK_CUR);
738 lseek (fd, offset, SEEK_SET); 774 lseek (fd, offset, SEEK_SET);
739 res = write (fd, buf, count); 775 res = write (fd, buf, count);
740 lseek (fd, offset, SEEK_SET); 776 lseek (fd, ooffset, SEEK_SET);
741 X_UNLOCK (preadwritelock); 777 X_UNLOCK (preadwritelock);
742 778
743 return res; 779 return res;
744} 780}
745#endif 781#endif
746 782
747#ifndef HAVE_FUTIMES 783#ifndef HAVE_FUTIMES
748 784
785# undef utimes
786# undef futimes
749# define utimes(path,times) eio__utimes (path, times) 787# define utimes(path,times) eio__utimes (path, times)
750# define futimes(fd,times) eio__futimes (fd, times) 788# define futimes(fd,times) eio__futimes (fd, times)
751 789
752static int 790static int
753eio__utimes (const char *filename, const struct timeval times[2]) 791eio__utimes (const char *filename, const struct timeval times[2])
… …
772} 810}
773 811
774#endif 812#endif
775 813
776#if !HAVE_FDATASYNC 814#if !HAVE_FDATASYNC
815# undef fdatasync
777# define fdatasync fsync 816# define fdatasync(fd) fsync (fd)
778#endif 817#endif
818
819/* sync_file_range always needs emulation */
820int
821eio__sync_file_range (int fd, off_t offset, size_t nbytes, unsigned int flags)
822{
823#if HAVE_SYNC_FILE_RANGE
824 int res;
825
826 if (EIO_SYNC_FILE_RANGE_WAIT_BEFORE != SYNC_FILE_RANGE_WAIT_BEFORE
827 || EIO_SYNC_FILE_RANGE_WRITE != SYNC_FILE_RANGE_WRITE
828 || EIO_SYNC_FILE_RANGE_WAIT_AFTER != SYNC_FILE_RANGE_WAIT_AFTER)
829 {
830 flags = 0
831 | (flags & EIO_SYNC_FILE_RANGE_WAIT_BEFORE ? SYNC_FILE_RANGE_WAIT_BEFORE : 0)
832 | (flags & EIO_SYNC_FILE_RANGE_WRITE ? SYNC_FILE_RANGE_WRITE : 0)
833 | (flags & EIO_SYNC_FILE_RANGE_WAIT_AFTER ? SYNC_FILE_RANGE_WAIT_AFTER : 0);
834 }
835
836 res = sync_file_range (fd, offset, nbytes, flags);
837
838 if (!res || errno != ENOSYS)
839 return res;
840#endif
841
842 /* even though we could play tricks with the flags, it's better to always
843 * call fdatasync, as thta matches the expectation of it's users best */
844 return fdatasync (fd);
845}
779 846
780#if !HAVE_READAHEAD 847#if !HAVE_READAHEAD
848# undef readahead
781# define readahead(fd,offset,count) eio__readahead (fd, offset, count, self) 849# define readahead(fd,offset,count) eio__readahead (fd, offset, count, self)
782 850
783static ssize_t 851static ssize_t
784eio__readahead (int fd, off_t offset, size_t count, etp_worker *self) 852eio__readahead (int fd, off_t offset, size_t count, etp_worker *self)
785{ 853{
… …
951 1019
952 req->result = res; 1020 req->result = res;
953} 1021}
954 1022
955#if !(_POSIX_MAPPED_FILES && _POSIX_SYNCHRONIZED_IO) 1023#if !(_POSIX_MAPPED_FILES && _POSIX_SYNCHRONIZED_IO)
1024# undef msync
956# define msync(a,b,c) ENOSYS 1025# define msync(a,b,c) ((errno = ENOSYS), -1)
957#endif 1026#endif
958 1027
959int 1028int
960eio__mtouch (void *mem, size_t len, int flags) 1029eio__mtouch (void *mem, size_t len, int flags)
961{ 1030{
… …
1155 case EIO_SYNC: req->result = 0; sync (); break; 1224 case EIO_SYNC: req->result = 0; sync (); break;
1156 case EIO_FSYNC: req->result = fsync (req->int1); break; 1225 case EIO_FSYNC: req->result = fsync (req->int1); break;
1157 case EIO_FDATASYNC: req->result = fdatasync (req->int1); break; 1226 case EIO_FDATASYNC: req->result = fdatasync (req->int1); break;
1158 case EIO_MSYNC: req->result = msync (req->ptr2, req->size, req->int1); break; 1227 case EIO_MSYNC: req->result = msync (req->ptr2, req->size, req->int1); break;
1159 case EIO_MTOUCH: req->result = eio__mtouch (req->ptr2, req->size, req->int1); break; 1228 case EIO_MTOUCH: req->result = eio__mtouch (req->ptr2, req->size, req->int1); break;
1229 case EIO_SYNC_FILE_RANGE: req->result = eio__sync_file_range (req->int1, req->offs, req->size, req->int2); break;
1160 1230
1161 case EIO_READDIR: eio__scandir (req, self); break; 1231 case EIO_READDIR: eio__scandir (req, self); break;
1162 1232
1163 case EIO_BUSY: 1233 case EIO_BUSY:
1164#ifdef _WIN32 1234#ifdef _WIN32
… …
1196 1266
1197 req->result = req->type == EIO_FUTIME 1267 req->result = req->type == EIO_FUTIME
1198 ? futimes (req->int1, times) 1268 ? futimes (req->int1, times)
1199 : utimes (req->ptr1, times); 1269 : utimes (req->ptr1, times);
1200 } 1270 }
1271 break;
1201 1272
1202 case EIO_GROUP: 1273 case EIO_GROUP:
1274 abort (); /* handled in eio_request */
1275
1203 case EIO_NOP: 1276 case EIO_NOP:
1204 req->result = 0; 1277 req->result = 0;
1205 break; 1278 break;
1206 1279
1207 case EIO_CUSTOM: 1280 case EIO_CUSTOM:
… …
1244} 1317}
1245 1318
1246eio_req *eio_mtouch (void *addr, size_t length, int flags, int pri, eio_cb cb, void *data) 1319eio_req *eio_mtouch (void *addr, size_t length, int flags, int pri, eio_cb cb, void *data)
1247{ 1320{
1248 REQ (EIO_MTOUCH); req->ptr2 = addr; req->size = length; req->int1 = flags; SEND; 1321 REQ (EIO_MTOUCH); req->ptr2 = addr; req->size = length; req->int1 = flags; SEND;
1322}
1323
1324eio_req *eio_sync_file_range (int fd, off_t offset, size_t nbytes, unsigned int flags, int pri, eio_cb cb, void *data)
1325{
1326 REQ (EIO_SYNC_FILE_RANGE); req->int1 = fd; req->offs = offset; req->size = nbytes; req->int2 = flags; SEND;
1249} 1327}
1250 1328
1251eio_req *eio_fdatasync (int fd, int pri, eio_cb cb, void *data) 1329eio_req *eio_fdatasync (int fd, int pri, eio_cb cb, void *data)
1252{ 1330{
1253 REQ (EIO_FDATASYNC); req->int1 = fd; SEND; 1331 REQ (EIO_FDATASYNC); req->int1 = fd; SEND;
… …
1448 1526
1449void eio_grp_add (eio_req *grp, eio_req *req) 1527void eio_grp_add (eio_req *grp, eio_req *req)
1450{ 1528{
1451 assert (("cannot add requests to IO::AIO::GRP after the group finished", grp->int1 != 2)); 1529 assert (("cannot add requests to IO::AIO::GRP after the group finished", grp->int1 != 2));
1452 1530
1531 grp->flags |= EIO_FLAG_GROUPADD;
1532
1453 ++grp->size; 1533 ++grp->size;
1454 req->grp = grp; 1534 req->grp = grp;
1455 1535
1456 req->grp_prev = 0; 1536 req->grp_prev = 0;
1457 req->grp_next = grp->grp_first; 1537 req->grp_next = grp->grp_first;

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines