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

Comparing libeio/eio.c (file contents):
Revision 1.16 by root, Fri May 30 04:50:21 2008 UTC vs.
Revision 1.27 by root, Wed Oct 22 18:15:36 2008 UTC

… …
120 120
121#define EIO_TICKS ((1000000 + 1023) >> 10) 121#define EIO_TICKS ((1000000 + 1023) >> 10)
122 122
123/*****************************************************************************/ 123/*****************************************************************************/
124 124
125#if __GNUC__ >= 3
126# define expect(expr,value) __builtin_expect ((expr),(value))
127#else
128# define expect(expr,value) (expr)
129#endif
130
131#define expect_false(expr) expect ((expr) != 0, 0)
132#define expect_true(expr) expect ((expr) != 0, 1)
133
134/*****************************************************************************/
135
125#define ETP_PRI_MIN EIO_PRI_MIN 136#define ETP_PRI_MIN EIO_PRI_MIN
126#define ETP_PRI_MAX EIO_PRI_MAX 137#define ETP_PRI_MAX EIO_PRI_MAX
127 138
128struct etp_worker; 139struct etp_worker;
129 140
… …
177static mutex_t wrklock = X_MUTEX_INIT; 188static mutex_t wrklock = X_MUTEX_INIT;
178static mutex_t reslock = X_MUTEX_INIT; 189static mutex_t reslock = X_MUTEX_INIT;
179static mutex_t reqlock = X_MUTEX_INIT; 190static mutex_t reqlock = X_MUTEX_INIT;
180static cond_t reqwait = X_COND_INIT; 191static cond_t reqwait = X_COND_INIT;
181 192
193#if !HAVE_PREADWRITE
194/*
195 * make our pread/pwrite emulation safe against themselves, but not against
196 * normal read/write by using a mutex. slows down execution a lot,
197 * but that's your problem, not mine.
198 */
199static mutex_t preadwritelock = X_MUTEX_INIT;
200#endif
201
182typedef struct etp_worker 202typedef struct etp_worker
183{ 203{
184 /* locked by wrklock */ 204 /* locked by wrklock */
185 struct etp_worker *prev, *next; 205 struct etp_worker *prev, *next;
186 206
… …
330 350
331static void etp_atfork_child (void) 351static void etp_atfork_child (void)
332{ 352{
333 ETP_REQ *prv; 353 ETP_REQ *prv;
334 354
335 while (prv = reqq_shift (&req_queue)) 355 while ((prv = reqq_shift (&req_queue)))
336 ETP_DESTROY (prv); 356 ETP_DESTROY (prv);
337 357
338 while (prv = reqq_shift (&res_queue)) 358 while ((prv = reqq_shift (&res_queue)))
339 ETP_DESTROY (prv); 359 ETP_DESTROY (prv);
340 360
341 while (wrk_first.next != &wrk_first) 361 while (wrk_first.next != &wrk_first)
342 { 362 {
343 etp_worker *wrk = wrk_first.next; 363 etp_worker *wrk = wrk_first.next;
… …
371 391
372 pthread_once (&doinit, etp_once_init); 392 pthread_once (&doinit, etp_once_init);
373 393
374 want_poll_cb = want_poll; 394 want_poll_cb = want_poll;
375 done_poll_cb = done_poll; 395 done_poll_cb = done_poll;
396
397 return 0;
376} 398}
377 399
378X_THREAD_PROC (etp_proc); 400X_THREAD_PROC (etp_proc);
379 401
380static void etp_start_thread (void) 402static void etp_start_thread (void)
… …
400 X_UNLOCK (wrklock); 422 X_UNLOCK (wrklock);
401} 423}
402 424
403static void etp_maybe_start_thread (void) 425static void etp_maybe_start_thread (void)
404{ 426{
405 if (etp_nthreads () >= wanted) 427 if (expect_true (etp_nthreads () >= wanted))
406 return; 428 return;
407 429
408 /* todo: maybe use idle here, but might be less exact */ 430 /* todo: maybe use idle here, but might be less exact */
409 if (0 <= (int)etp_nthreads () + (int)etp_npending () - (int)etp_nreqs ()) 431 if (expect_true (0 <= (int)etp_nthreads () + (int)etp_npending () - (int)etp_nreqs ()))
410 return; 432 return;
411 433
412 etp_start_thread (); 434 etp_start_thread ();
413} 435}
414 436
… …
467 489
468 X_LOCK (reqlock); 490 X_LOCK (reqlock);
469 --nreqs; 491 --nreqs;
470 X_UNLOCK (reqlock); 492 X_UNLOCK (reqlock);
471 493
472 if (req->type == EIO_GROUP && req->size) 494 if (expect_false (req->type == EIO_GROUP && req->size))
473 { 495 {
474 req->int1 = 1; /* mark request as delayed */ 496 req->int1 = 1; /* mark request as delayed */
475 continue; 497 continue;
476 } 498 }
477 else 499 else
478 { 500 {
479 int res = ETP_FINISH (req); 501 int res = ETP_FINISH (req);
480 if (res) 502 if (expect_false (res))
481 return res; 503 return res;
482 } 504 }
483 505
484 if (maxreqs && !--maxreqs) 506 if (expect_false (maxreqs && !--maxreqs))
485 break; 507 break;
486 508
487 if (maxtime) 509 if (maxtime)
488 { 510 {
489 gettimeofday (&tv_now, 0); 511 gettimeofday (&tv_now, 0);
… …
508 530
509static void etp_submit (ETP_REQ *req) 531static void etp_submit (ETP_REQ *req)
510{ 532{
511 req->pri -= ETP_PRI_MIN; 533 req->pri -= ETP_PRI_MIN;
512 534
513 if (req->pri < ETP_PRI_MIN - ETP_PRI_MIN) req->pri = ETP_PRI_MIN - ETP_PRI_MIN; 535 if (expect_false (req->pri < ETP_PRI_MIN - ETP_PRI_MIN)) req->pri = ETP_PRI_MIN - ETP_PRI_MIN;
514 if (req->pri > ETP_PRI_MAX - ETP_PRI_MIN) req->pri = ETP_PRI_MAX - ETP_PRI_MIN; 536 if (expect_false (req->pri > ETP_PRI_MAX - ETP_PRI_MIN)) req->pri = ETP_PRI_MAX - ETP_PRI_MIN;
515 537
538 if (expect_false (req->type == EIO_GROUP))
539 {
540 /* I hope this is worth it :/ */
516 X_LOCK (reqlock); 541 X_LOCK (reqlock);
517 ++nreqs; 542 ++nreqs;
543 X_UNLOCK (reqlock);
544
545 X_LOCK (reslock);
546
547 ++npending;
548
549 if (!reqq_push (&res_queue, req) && want_poll_cb)
550 want_poll_cb ();
551
552 X_UNLOCK (reslock);
553 }
554 else
555 {
556 X_LOCK (reqlock);
557 ++nreqs;
518 ++nready; 558 ++nready;
519 reqq_push (&req_queue, req); 559 reqq_push (&req_queue, req);
520 X_COND_SIGNAL (reqwait); 560 X_COND_SIGNAL (reqwait);
521 X_UNLOCK (reqlock); 561 X_UNLOCK (reqlock);
522 562
523 etp_maybe_start_thread (); 563 etp_maybe_start_thread ();
564 }
524} 565}
525 566
526static void etp_set_max_poll_time (double nseconds) 567static void etp_set_max_poll_time (double nseconds)
527{ 568{
528 if (WORDACCESS_UNSAFE) X_LOCK (reslock); 569 if (WORDACCESS_UNSAFE) X_LOCK (reslock);
… …
563 604
564static void grp_try_feed (eio_req *grp) 605static void grp_try_feed (eio_req *grp)
565{ 606{
566 while (grp->size < grp->int2 && !EIO_CANCELLED (grp)) 607 while (grp->size < grp->int2 && !EIO_CANCELLED (grp))
567 { 608 {
568 int old_len = grp->size; 609 grp->flags &= ~EIO_FLAG_GROUPADD;
569 610
570 EIO_FEED (grp); 611 EIO_FEED (grp);
571 612
572 /* stop if no progress has been made */ 613 /* stop if no progress has been made */
573 if (old_len == grp->size) 614 if (!(grp->flags & EIO_FLAG_GROUPADD))
574 { 615 {
575 grp->feed = 0; 616 grp->feed = 0;
576 break; 617 break;
577 } 618 }
578 } 619 }
… …
695 736
696/*****************************************************************************/ 737/*****************************************************************************/
697/* work around various missing functions */ 738/* work around various missing functions */
698 739
699#if !HAVE_PREADWRITE 740#if !HAVE_PREADWRITE
741# undef pread
742# undef pwrite
700# define pread eio__pread 743# define pread eio__pread
701# define pwrite eio__pwrite 744# define pwrite eio__pwrite
702
703/*
704 * make our pread/pwrite safe against themselves, but not against
705 * normal read/write by using a mutex. slows down execution a lot,
706 * but that's your problem, not mine.
707 */
708static mutex_t preadwritelock = X_MUTEX_INIT;
709 745
710static ssize_t 746static ssize_t
711eio__pread (int fd, void *buf, size_t count, off_t offset) 747eio__pread (int fd, void *buf, size_t count, off_t offset)
712{ 748{
713 ssize_t res; 749 ssize_t res;
… …
740} 776}
741#endif 777#endif
742 778
743#ifndef HAVE_FUTIMES 779#ifndef HAVE_FUTIMES
744 780
781# undef utimes
782# undef futimes
745# define utimes(path,times) eio__utimes (path, times) 783# define utimes(path,times) eio__utimes (path, times)
746# define futimes(fd,times) eio__futimes (fd, times) 784# define futimes(fd,times) eio__futimes (fd, times)
747 785
748static int 786static int
749eio__utimes (const char *filename, const struct timeval times[2]) 787eio__utimes (const char *filename, const struct timeval times[2])
… …
768} 806}
769 807
770#endif 808#endif
771 809
772#if !HAVE_FDATASYNC 810#if !HAVE_FDATASYNC
811# undef fdatasync
773# define fdatasync fsync 812# define fdatasync(fd) fsync (fd)
774#endif 813#endif
814
815/* sync_file_range always needs emulation */
816int
817eio__sync_file_range (int fd, off_t offset, size_t nbytes, unsigned int flags)
818{
819#if HAVE_SYNC_FILE_RANGE
820 int res;
821
822 if (EIO_SYNC_FILE_RANGE_WAIT_BEFORE != SYNC_FILE_RANGE_WAIT_BEFORE
823 || EIO_SYNC_FILE_RANGE_WRITE != SYNC_FILE_RANGE_WRITE
824 || EIO_SYNC_FILE_RANGE_WAIT_AFTER != SYNC_FILE_RANGE_WAIT_AFTER)
825 {
826 flags = 0
827 | (flags & EIO_SYNC_FILE_RANGE_WAIT_BEFORE ? SYNC_FILE_RANGE_WAIT_BEFORE : 0)
828 | (flags & EIO_SYNC_FILE_RANGE_WRITE ? SYNC_FILE_RANGE_WRITE : 0)
829 | (flags & EIO_SYNC_FILE_RANGE_WAIT_AFTER ? SYNC_FILE_RANGE_WAIT_AFTER : 0);
830 }
831
832 res = sync_file_range (fd, offset, nbytes, flags);
833
834 if (res != ENOSYS)
835 return res;
836#endif
837
838 /* even though we could play tricks with the flags, it's better to always
839 * call fdatasync, as thta matches the expectation of it's users best */
840 return fdatasync (fd);
841}
775 842
776#if !HAVE_READAHEAD 843#if !HAVE_READAHEAD
844# undef readahead
777# define readahead(fd,offset,count) eio__readahead (fd, offset, count, self) 845# define readahead(fd,offset,count) eio__readahead (fd, offset, count, self)
778 846
779static ssize_t 847static ssize_t
780eio__readahead (int fd, off_t offset, size_t count, worker *self) 848eio__readahead (int fd, off_t offset, size_t count, etp_worker *self)
781{ 849{
782 size_t todo = count; 850 size_t todo = count;
783 dBUF; 851 dBUF;
784 852
785 while (todo > 0) 853 while (todo > 0)
… …
901 int memlen = 4096; 969 int memlen = 4096;
902 int memofs = 0; 970 int memofs = 0;
903 int res = 0; 971 int res = 0;
904 972
905 X_LOCK (wrklock); 973 X_LOCK (wrklock);
974 /* the corresponding closedir is in ETP_WORKER_CLEAR */
906 self->dirp = dirp = opendir (req->ptr1); 975 self->dirp = dirp = opendir (req->ptr1);
907 req->flags |= EIO_FLAG_PTR2_FREE; 976 req->flags |= EIO_FLAG_PTR2_FREE;
908 req->ptr2 = names = malloc (memlen); 977 req->ptr2 = names = malloc (memlen);
909 X_UNLOCK (wrklock); 978 X_UNLOCK (wrklock);
910 979
… …
943 1012
944 if (errno) 1013 if (errno)
945 res = -1; 1014 res = -1;
946 1015
947 req->result = res; 1016 req->result = res;
1017}
1018
1019#if !(_POSIX_MAPPED_FILES && _POSIX_SYNCHRONIZED_IO)
1020# undef msync
1021# define msync(a,b,c) ENOSYS
1022#endif
1023
1024int
1025eio__mtouch (void *mem, size_t len, int flags)
1026{
1027 intptr_t addr = (intptr_t)mem;
1028 intptr_t end = addr + len;
1029#ifdef PAGESIZE
1030 const intptr_t page = PAGESIZE;
1031#else
1032 static intptr_t page;
1033
1034 if (!page)
1035 page = sysconf (_SC_PAGESIZE);
1036#endif
1037
1038 addr &= ~(page - 1); /* assume page size is always a power of two */
1039
1040 if (addr < end)
1041 if (flags) /* modify */
1042 do { *((volatile sig_atomic_t *)addr) |= 0; } while ((addr += page) < len);
1043 else
1044 do { *((volatile sig_atomic_t *)addr) ; } while ((addr += page) < len);
1045
1046 return 0;
948} 1047}
949 1048
950/*****************************************************************************/ 1049/*****************************************************************************/
951 1050
952#define ALLOC(len) \ 1051#define ALLOC(len) \
… …
1039 1138
1040/*****************************************************************************/ 1139/*****************************************************************************/
1041 1140
1042int eio_init (void (*want_poll)(void), void (*done_poll)(void)) 1141int eio_init (void (*want_poll)(void), void (*done_poll)(void))
1043{ 1142{
1044 etp_init (want_poll, done_poll); 1143 return etp_init (want_poll, done_poll);
1045} 1144}
1046 1145
1047static void eio_api_destroy (eio_req *req) 1146static void eio_api_destroy (eio_req *req)
1048{ 1147{
1049 free (req); 1148 free (req);
… …
1111 case EIO_RMDIR: req->result = rmdir (req->ptr1); break; 1210 case EIO_RMDIR: req->result = rmdir (req->ptr1); break;
1112 case EIO_MKDIR: req->result = mkdir (req->ptr1, (mode_t)req->int2); break; 1211 case EIO_MKDIR: req->result = mkdir (req->ptr1, (mode_t)req->int2); break;
1113 case EIO_RENAME: req->result = rename (req->ptr1, req->ptr2); break; 1212 case EIO_RENAME: req->result = rename (req->ptr1, req->ptr2); break;
1114 case EIO_LINK: req->result = link (req->ptr1, req->ptr2); break; 1213 case EIO_LINK: req->result = link (req->ptr1, req->ptr2); break;
1115 case EIO_SYMLINK: req->result = symlink (req->ptr1, req->ptr2); break; 1214 case EIO_SYMLINK: req->result = symlink (req->ptr1, req->ptr2); break;
1116 case EIO_MKNOD: req->result = mknod (req->ptr1, (mode_t)req->int2, (dev_t)req->offs); break; 1215 case EIO_MKNOD: req->result = mknod (req->ptr1, (mode_t)req->int2, (dev_t)req->int3); break;
1117 1216
1118 case EIO_READLINK: ALLOC (NAME_MAX); 1217 case EIO_READLINK: ALLOC (NAME_MAX);
1119 req->result = readlink (req->ptr1, req->ptr2, NAME_MAX); break; 1218 req->result = readlink (req->ptr1, req->ptr2, NAME_MAX); break;
1120 1219
1121 case EIO_SYNC: req->result = 0; sync (); break; 1220 case EIO_SYNC: req->result = 0; sync (); break;
1122 case EIO_FSYNC: req->result = fsync (req->int1); break; 1221 case EIO_FSYNC: req->result = fsync (req->int1); break;
1123 case EIO_FDATASYNC: req->result = fdatasync (req->int1); break; 1222 case EIO_FDATASYNC: req->result = fdatasync (req->int1); break;
1223 case EIO_MSYNC: req->result = msync (req->ptr2, req->size, req->int1); break;
1224 case EIO_MTOUCH: req->result = eio__mtouch (req->ptr2, req->size, req->int1); break;
1225 case EIO_SYNC_FILE_RANGE: req->result = eio__sync_file_range (req->int1, req->offs, req->size, req->int2); break;
1124 1226
1125 case EIO_READDIR: eio__scandir (req, self); break; 1227 case EIO_READDIR: eio__scandir (req, self); break;
1126 1228
1127 case EIO_BUSY: 1229 case EIO_BUSY:
1128#ifdef _WIN32 1230#ifdef _WIN32
… …
1160 1262
1161 req->result = req->type == EIO_FUTIME 1263 req->result = req->type == EIO_FUTIME
1162 ? futimes (req->int1, times) 1264 ? futimes (req->int1, times)
1163 : utimes (req->ptr1, times); 1265 : utimes (req->ptr1, times);
1164 } 1266 }
1267 break;
1165 1268
1166 case EIO_GROUP: 1269 case EIO_GROUP:
1270 abort (); /* handled in eio_request */
1271
1167 case EIO_NOP: 1272 case EIO_NOP:
1168 req->result = 0; 1273 req->result = 0;
1169 break; 1274 break;
1170 1275
1171 case EIO_CUSTOM: 1276 case EIO_CUSTOM:
1172 req->feed (req); 1277 ((void (*)(eio_req *))req->feed) (req);
1173 break; 1278 break;
1174 1279
1175 default: 1280 default:
1176 req->result = -1; 1281 req->result = -1;
1177 break; 1282 break;
… …
1200eio_req *eio_fsync (int fd, int pri, eio_cb cb, void *data) 1305eio_req *eio_fsync (int fd, int pri, eio_cb cb, void *data)
1201{ 1306{
1202 REQ (EIO_FSYNC); req->int1 = fd; SEND; 1307 REQ (EIO_FSYNC); req->int1 = fd; SEND;
1203} 1308}
1204 1309
1310eio_req *eio_msync (void *addr, size_t length, int flags, int pri, eio_cb cb, void *data)
1311{
1312 REQ (EIO_MSYNC); req->ptr2 = addr; req->size = length; req->int1 = flags; SEND;
1313}
1314
1315eio_req *eio_mtouch (void *addr, size_t length, int flags, int pri, eio_cb cb, void *data)
1316{
1317 REQ (EIO_MTOUCH); req->ptr2 = addr; req->size = length; req->int1 = flags; SEND;
1318}
1319
1320eio_req *eio_sync_file_range (int fd, off_t offset, size_t nbytes, unsigned int flags, int pri, eio_cb cb, void *data)
1321{
1322 REQ (EIO_SYNC_FILE_RANGE); req->int1 = fd; req->offs = offset; req->size = nbytes; req->int2 = flags; SEND;
1323}
1324
1205eio_req *eio_fdatasync (int fd, int pri, eio_cb cb, void *data) 1325eio_req *eio_fdatasync (int fd, int pri, eio_cb cb, void *data)
1206{ 1326{
1207 REQ (EIO_FDATASYNC); req->int1 = fd; SEND; 1327 REQ (EIO_FDATASYNC); req->int1 = fd; SEND;
1208} 1328}
1209 1329
… …
1328 return eio__1path (EIO_READDIR, path, pri, cb, data); 1448 return eio__1path (EIO_READDIR, path, pri, cb, data);
1329} 1449}
1330 1450
1331eio_req *eio_mknod (const char *path, mode_t mode, dev_t dev, int pri, eio_cb cb, void *data) 1451eio_req *eio_mknod (const char *path, mode_t mode, dev_t dev, int pri, eio_cb cb, void *data)
1332{ 1452{
1333 REQ (EIO_MKNOD); PATH; req->int2 = (long)mode; req->int2 = (long)dev; SEND; 1453 REQ (EIO_MKNOD); PATH; req->int2 = (long)mode; req->int3 = (long)dev; SEND;
1334} 1454}
1335 1455
1336static eio_req * 1456static eio_req *
1337eio__2path (int type, const char *path, const char *new_path, int pri, eio_cb cb, void *data) 1457eio__2path (int type, const char *path, const char *new_path, int pri, eio_cb cb, void *data)
1338{ 1458{
… …
1364 return eio__2path (EIO_RENAME, path, new_path, pri, cb, data); 1484 return eio__2path (EIO_RENAME, path, new_path, pri, cb, data);
1365} 1485}
1366 1486
1367eio_req *eio_custom (eio_cb execute, int pri, eio_cb cb, void *data) 1487eio_req *eio_custom (eio_cb execute, int pri, eio_cb cb, void *data)
1368{ 1488{
1369 REQ (EIO_CUSTOM); req->feed = execute; SEND; 1489 REQ (EIO_CUSTOM); req->feed = (void (*)(eio_req *))execute; SEND;
1370} 1490}
1371 1491
1372#endif 1492#endif
1373 1493
1374eio_req *eio_grp (eio_cb cb, void *data) 1494eio_req *eio_grp (eio_cb cb, void *data)
… …
1402 1522
1403void eio_grp_add (eio_req *grp, eio_req *req) 1523void eio_grp_add (eio_req *grp, eio_req *req)
1404{ 1524{
1405 assert (("cannot add requests to IO::AIO::GRP after the group finished", grp->int1 != 2)); 1525 assert (("cannot add requests to IO::AIO::GRP after the group finished", grp->int1 != 2));
1406 1526
1527 grp->flags |= EIO_FLAG_GROUPADD;
1528
1407 ++grp->size; 1529 ++grp->size;
1408 req->grp = grp; 1530 req->grp = grp;
1409 1531
1410 req->grp_prev = 0; 1532 req->grp_prev = 0;
1411 req->grp_next = grp->grp_first; 1533 req->grp_next = grp->grp_first;

Diff Legend

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