ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/IO-AIO/AIO.xs
(Generate patch)

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.41 by root, Sat Jun 24 19:14:04 2006 UTC vs.
Revision 1.44 by root, Sat Oct 21 23:20:29 2006 UTC

… …
53 REQ_STAT, REQ_LSTAT, REQ_FSTAT, 53 REQ_STAT, REQ_LSTAT, REQ_FSTAT,
54 REQ_FSYNC, REQ_FDATASYNC, 54 REQ_FSYNC, REQ_FDATASYNC,
55 REQ_UNLINK, REQ_RMDIR, REQ_RENAME, 55 REQ_UNLINK, REQ_RMDIR, REQ_RENAME,
56 REQ_READDIR, 56 REQ_READDIR,
57 REQ_LINK, REQ_SYMLINK, 57 REQ_LINK, REQ_SYMLINK,
58 REQ_GROUP,
58}; 59};
59 60
61#define AIO_REQ_KLASS "IO::AIO::REQ"
62#define AIO_GRP_KLASS "IO::AIO::GRP"
63
60typedef struct aio_cb { 64typedef struct aio_cb
65{
66 struct aio_cb *grp, *grp_prev, *grp_next;
67
61 struct aio_cb *volatile next; 68 struct aio_cb *volatile next;
62 69
63 int type; 70 SV *self; /* the perl counterpart of this request, if any */
64 71
65 /* should receive a cleanup, with unions */ 72 SV *data, *callback;
66 int fd, fd2; 73 SV *fh, *fh2;
74 void *dataptr, *data2ptr;
75 Stat_t *statdata;
67 off_t offset; 76 off_t offset;
68 size_t length; 77 size_t length;
69 ssize_t result; 78 ssize_t result;
79
80 int type;
81 int fd, fd2;
82 int errorno;
83 STRLEN dataoffset;
70 mode_t mode; /* open */ 84 mode_t mode; /* open */
71 int errorno; 85 unsigned char cancelled;
72 SV *data, *callback;
73 SV *fh, *fh2;
74 void *dataptr, *data2ptr;
75 STRLEN dataoffset;
76
77 Stat_t *statdata;
78} aio_cb; 86} aio_cb;
79 87
80typedef aio_cb *aio_req; 88typedef aio_cb *aio_req;
89typedef aio_cb *aio_req_ornot;
90typedef aio_cb *aio_group;
81 91
82static int started, wanted; 92static int started, wanted;
83static volatile int nreqs; 93static volatile int nreqs;
84static int max_outstanding = 1<<30; 94static int max_outstanding = 1<<30;
85static int respipe [2]; 95static int respipe [2];
… …
89static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER; 99static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
90 100
91static volatile aio_req reqs, reqe; /* queue start, queue end */ 101static volatile aio_req reqs, reqe; /* queue start, queue end */
92static volatile aio_req ress, rese; /* queue start, queue end */ 102static volatile aio_req ress, rese; /* queue start, queue end */
93 103
104/* must be called at most once */
105static SV *req_sv (aio_req req, const char *klass)
106{
107 req->self = (SV *)newHV ();
108 sv_magic (req->self, 0, PERL_MAGIC_ext, (char *)req, 0);
109
110 return sv_2mortal (sv_bless (newRV_noinc (req->self), gv_stashpv (klass, 1)));
111}
112
113static aio_req SvAIO_REQ (SV *sv, const char *klass)
114{
115 if (!sv_derived_from (sv, klass) || !SvROK (sv))
116 croak ("object of class %s expected", klass);
117
118 MAGIC *mg = mg_find (SvRV (sv), PERL_MAGIC_ext);
119
120 return mg ? (aio_req)mg->mg_ptr : 0;
121}
122
94static void free_req (aio_req req) 123static void req_free (aio_req req)
95{ 124{
125 if (req->self)
126 {
127 sv_unmagic (req->self, PERL_MAGIC_ext);
128 SvREFCNT_dec (req->self);
129 }
130
96 if (req->data) 131 if (req->data)
97 SvREFCNT_dec (req->data); 132 SvREFCNT_dec (req->data);
98 133
99 if (req->fh) 134 if (req->fh)
100 SvREFCNT_dec (req->fh); 135 SvREFCNT_dec (req->fh);
… …
226 PUSHMARK (SP); 261 PUSHMARK (SP);
227 XPUSHs (sv_2mortal (fh)); 262 XPUSHs (sv_2mortal (fh));
228 } 263 }
229 } 264 }
230 265
231 if (SvOK (req->callback)) 266 if (SvOK (req->callback) && !req->cancelled)
232 { 267 {
233 PUTBACK; 268 PUTBACK;
234 call_sv (req->callback, G_VOID | G_EVAL); 269 call_sv (req->callback, G_VOID | G_EVAL);
235 SPAGAIN; 270 SPAGAIN;
236 271
237 if (SvTRUE (ERRSV)) 272 if (SvTRUE (ERRSV))
238 { 273 {
239 free_req (req); 274 req_free (req);
240 croak (0); 275 croak (0);
241 } 276 }
242 } 277 }
243 278
244 LEAVE; 279 LEAVE;
245 280
246 errno = errorno; 281 errno = errorno;
247 count++; 282 count++;
248 } 283 }
249 284
250 free_req (req); 285 req_free (req);
251 } 286 }
252 287
253 return count; 288 return count;
254} 289}
255 290
… …
274 309
275 sigprocmask (SIG_SETMASK, &oldsigset, 0); 310 sigprocmask (SIG_SETMASK, &oldsigset, 0);
276} 311}
277 312
278static void 313static void
279send_req (aio_req req) 314req_send (aio_req req)
280{ 315{
281 while (started < wanted && nreqs >= started) 316 while (started < wanted && nreqs >= started)
282 start_thread (); 317 start_thread ();
283 318
284 nreqs++; 319 nreqs++;
… …
315{ 350{
316 aio_req req; 351 aio_req req;
317 Newz (0, req, 1, aio_cb); 352 Newz (0, req, 1, aio_cb);
318 req->type = REQ_QUIT; 353 req->type = REQ_QUIT;
319 354
320 send_req (req); 355 req_send (req);
321} 356}
322 357
323static void min_parallel (int nthreads) 358static void min_parallel (int nthreads)
324{ 359{
325 if (wanted < nthreads) 360 if (wanted < nthreads)
… …
648 683
649 pthread_mutex_unlock (&reqlock); 684 pthread_mutex_unlock (&reqlock);
650 685
651 errno = 0; /* strictly unnecessary */ 686 errno = 0; /* strictly unnecessary */
652 687
653 type = req->type; 688 if (!req->cancelled)
654
655 switch (type) 689 switch (req->type)
656 { 690 {
657 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break; 691 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
658 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break; 692 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
659 693
660 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break; 694 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
661 case REQ_SENDFILE: req->result = sendfile_ (req->fd, req->fd2, req->offset, req->length); break; 695 case REQ_SENDFILE: req->result = sendfile_ (req->fd, req->fd2, req->offset, req->length); break;
662 696
663 case REQ_STAT: req->result = stat (req->dataptr, req->statdata); break; 697 case REQ_STAT: req->result = stat (req->dataptr, req->statdata); break;
664 case REQ_LSTAT: req->result = lstat (req->dataptr, req->statdata); break; 698 case REQ_LSTAT: req->result = lstat (req->dataptr, req->statdata); break;
665 case REQ_FSTAT: req->result = fstat (req->fd , req->statdata); break; 699 case REQ_FSTAT: req->result = fstat (req->fd , req->statdata); break;
666 700
667 case REQ_OPEN: req->result = open (req->dataptr, req->fd, req->mode); break; 701 case REQ_OPEN: req->result = open (req->dataptr, req->fd, req->mode); break;
668 case REQ_CLOSE: req->result = close (req->fd); break; 702 case REQ_CLOSE: req->result = close (req->fd); break;
669 case REQ_UNLINK: req->result = unlink (req->dataptr); break; 703 case REQ_UNLINK: req->result = unlink (req->dataptr); break;
670 case REQ_RMDIR: req->result = rmdir (req->dataptr); break; 704 case REQ_RMDIR: req->result = rmdir (req->dataptr); break;
671 case REQ_RENAME: req->result = rename (req->data2ptr, req->dataptr); break; 705 case REQ_RENAME: req->result = rename (req->data2ptr, req->dataptr); break;
672 case REQ_LINK: req->result = link (req->data2ptr, req->dataptr); break; 706 case REQ_LINK: req->result = link (req->data2ptr, req->dataptr); break;
673 case REQ_SYMLINK: req->result = symlink (req->data2ptr, req->dataptr); break; 707 case REQ_SYMLINK: req->result = symlink (req->data2ptr, req->dataptr); break;
674 708
675 case REQ_FDATASYNC: req->result = fdatasync (req->fd); break; 709 case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
676 case REQ_FSYNC: req->result = fsync (req->fd); break; 710 case REQ_FSYNC: req->result = fsync (req->fd); break;
677 case REQ_READDIR: req->result = scandir_ (req->dataptr, &req->data2ptr); break; 711 case REQ_READDIR: req->result = scandir_ (req->dataptr, &req->data2ptr); break;
678 712
679 case REQ_QUIT: 713 case REQ_QUIT:
680 break; 714 break;
681 715
682 default: 716 default:
683 req->result = ENOSYS; 717 req->result = ENOSYS;
684 break; 718 break;
685 } 719 }
686 720
687 req->errorno = errno; 721 req->errorno = errno;
688 722
689 pthread_mutex_lock (&reslock); 723 pthread_mutex_lock (&reslock);
690 724
… …
744 778
745 while (reqs) 779 while (reqs)
746 { 780 {
747 prv = reqs; 781 prv = reqs;
748 reqs = prv->next; 782 reqs = prv->next;
749 free_req (prv); 783 req_free (prv);
750 } 784 }
751 785
752 reqs = reqe = 0; 786 reqs = reqe = 0;
753 787
754 while (ress) 788 while (ress)
755 { 789 {
756 prv = ress; 790 prv = ress;
757 ress = prv->next; 791 ress = prv->next;
758 free_req (prv); 792 req_free (prv);
759 } 793 }
760 794
761 ress = rese = 0; 795 ress = rese = 0;
762 796
763 close (respipe [0]); 797 close (respipe [0]);
… …
769 803
770#define dREQ \ 804#define dREQ \
771 aio_req req; \ 805 aio_req req; \
772 \ 806 \
773 if (SvOK (callback) && !SvROK (callback)) \ 807 if (SvOK (callback) && !SvROK (callback)) \
774 croak ("clalback must be undef or of reference type"); \ 808 croak ("callback must be undef or of reference type"); \
775 \ 809 \
776 Newz (0, req, 1, aio_cb); \ 810 Newz (0, req, 1, aio_cb); \
777 if (!req) \ 811 if (!req) \
778 croak ("out of memory during aio_req allocation"); \ 812 croak ("out of memory during aio_req allocation"); \
779 \ 813 \
780 req->callback = newSVsv (callback); 814 req->callback = newSVsv (callback)
815
816#define REQ_SEND \
817 req_send (req); \
818 \
819 if (GIMME_V != G_VOID) \
820 XPUSHs (req_sv (req, AIO_REQ_KLASS));
781 821
782MODULE = IO::AIO PACKAGE = IO::AIO 822MODULE = IO::AIO PACKAGE = IO::AIO
783 823
784PROTOTYPES: ENABLE 824PROTOTYPES: ENABLE
785 825
… …
817 SV * pathname 857 SV * pathname
818 int flags 858 int flags
819 int mode 859 int mode
820 SV * callback 860 SV * callback
821 PROTOTYPE: $$$;$ 861 PROTOTYPE: $$$;$
822 CODE: 862 PPCODE:
823{ 863{
824 dREQ; 864 dREQ;
825 865
826 req->type = REQ_OPEN; 866 req->type = REQ_OPEN;
827 req->data = newSVsv (pathname); 867 req->data = newSVsv (pathname);
828 req->dataptr = SvPVbyte_nolen (req->data); 868 req->dataptr = SvPVbyte_nolen (req->data);
829 req->fd = flags; 869 req->fd = flags;
830 req->mode = mode; 870 req->mode = mode;
831 871
832 send_req (req); 872 REQ_SEND;
833} 873}
834 874
835void 875void
836aio_close (fh,callback=&PL_sv_undef) 876aio_close (fh,callback=&PL_sv_undef)
837 SV * fh 877 SV * fh
… …
839 PROTOTYPE: $;$ 879 PROTOTYPE: $;$
840 ALIAS: 880 ALIAS:
841 aio_close = REQ_CLOSE 881 aio_close = REQ_CLOSE
842 aio_fsync = REQ_FSYNC 882 aio_fsync = REQ_FSYNC
843 aio_fdatasync = REQ_FDATASYNC 883 aio_fdatasync = REQ_FDATASYNC
844 CODE: 884 PPCODE:
845{ 885{
846 dREQ; 886 dREQ;
847 887
848 req->type = ix; 888 req->type = ix;
849 req->fh = newSVsv (fh); 889 req->fh = newSVsv (fh);
850 req->fd = PerlIO_fileno (IoIFP (sv_2io (fh))); 890 req->fd = PerlIO_fileno (IoIFP (sv_2io (fh)));
851 891
852 send_req (req); 892 REQ_SEND (req);
853} 893}
854 894
855void 895void
856aio_read (fh,offset,length,data,dataoffset,callback=&PL_sv_undef) 896aio_read (fh,offset,length,data,dataoffset,callback=&PL_sv_undef)
857 SV * fh 897 SV * fh
… …
862 SV * callback 902 SV * callback
863 ALIAS: 903 ALIAS:
864 aio_read = REQ_READ 904 aio_read = REQ_READ
865 aio_write = REQ_WRITE 905 aio_write = REQ_WRITE
866 PROTOTYPE: $$$$$;$ 906 PROTOTYPE: $$$$$;$
867 CODE: 907 PPCODE:
868{ 908{
869 aio_req req; 909 aio_req req;
870 STRLEN svlen; 910 STRLEN svlen;
871 char *svptr = SvPVbyte (data, svlen); 911 char *svptr = SvPVbyte (data, svlen);
872 912
… …
910 { 950 {
911 SvREADONLY_on (data); 951 SvREADONLY_on (data);
912 req->data2ptr = (void *)data; 952 req->data2ptr = (void *)data;
913 } 953 }
914 954
915 send_req (req); 955 REQ_SEND;
916 } 956 }
917} 957}
918 958
919void 959void
920aio_sendfile (out_fh,in_fh,in_offset,length,callback=&PL_sv_undef) 960aio_sendfile (out_fh,in_fh,in_offset,length,callback=&PL_sv_undef)
… …
922 SV * in_fh 962 SV * in_fh
923 UV in_offset 963 UV in_offset
924 UV length 964 UV length
925 SV * callback 965 SV * callback
926 PROTOTYPE: $$$$;$ 966 PROTOTYPE: $$$$;$
927 CODE: 967 PPCODE:
928{ 968{
929 dREQ; 969 dREQ;
930 970
931 req->type = REQ_SENDFILE; 971 req->type = REQ_SENDFILE;
932 req->fh = newSVsv (out_fh); 972 req->fh = newSVsv (out_fh);
… …
934 req->fh2 = newSVsv (in_fh); 974 req->fh2 = newSVsv (in_fh);
935 req->fd2 = PerlIO_fileno (IoIFP (sv_2io (in_fh))); 975 req->fd2 = PerlIO_fileno (IoIFP (sv_2io (in_fh)));
936 req->offset = in_offset; 976 req->offset = in_offset;
937 req->length = length; 977 req->length = length;
938 978
939 send_req (req); 979 REQ_SEND;
940} 980}
941 981
942void 982void
943aio_readahead (fh,offset,length,callback=&PL_sv_undef) 983aio_readahead (fh,offset,length,callback=&PL_sv_undef)
944 SV * fh 984 SV * fh
945 UV offset 985 UV offset
946 IV length 986 IV length
947 SV * callback 987 SV * callback
948 PROTOTYPE: $$$;$ 988 PROTOTYPE: $$$;$
949 CODE: 989 PPCODE:
950{ 990{
951 dREQ; 991 dREQ;
952 992
953 req->type = REQ_READAHEAD; 993 req->type = REQ_READAHEAD;
954 req->fh = newSVsv (fh); 994 req->fh = newSVsv (fh);
955 req->fd = PerlIO_fileno (IoIFP (sv_2io (fh))); 995 req->fd = PerlIO_fileno (IoIFP (sv_2io (fh)));
956 req->offset = offset; 996 req->offset = offset;
957 req->length = length; 997 req->length = length;
958 998
959 send_req (req); 999 REQ_SEND;
960} 1000}
961 1001
962void 1002void
963aio_stat (fh_or_path,callback=&PL_sv_undef) 1003aio_stat (fh_or_path,callback=&PL_sv_undef)
964 SV * fh_or_path 1004 SV * fh_or_path
965 SV * callback 1005 SV * callback
966 ALIAS: 1006 ALIAS:
967 aio_stat = REQ_STAT 1007 aio_stat = REQ_STAT
968 aio_lstat = REQ_LSTAT 1008 aio_lstat = REQ_LSTAT
969 CODE: 1009 PPCODE:
970{ 1010{
971 dREQ; 1011 dREQ;
972 1012
973 New (0, req->statdata, 1, Stat_t); 1013 New (0, req->statdata, 1, Stat_t);
974 if (!req->statdata) 1014 if (!req->statdata)
975 { 1015 {
976 free_req (req); 1016 req_free (req);
977 croak ("out of memory during aio_req->statdata allocation"); 1017 croak ("out of memory during aio_req->statdata allocation");
978 } 1018 }
979 1019
980 if (SvPOK (fh_or_path)) 1020 if (SvPOK (fh_or_path))
981 { 1021 {
… …
988 req->type = REQ_FSTAT; 1028 req->type = REQ_FSTAT;
989 req->fh = newSVsv (fh_or_path); 1029 req->fh = newSVsv (fh_or_path);
990 req->fd = PerlIO_fileno (IoIFP (sv_2io (fh_or_path))); 1030 req->fd = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
991 } 1031 }
992 1032
993 send_req (req); 1033 REQ_SEND;
994} 1034}
995 1035
996void 1036void
997aio_unlink (pathname,callback=&PL_sv_undef) 1037aio_unlink (pathname,callback=&PL_sv_undef)
998 SV * pathname 1038 SV * pathname
999 SV * callback 1039 SV * callback
1000 ALIAS: 1040 ALIAS:
1001 aio_unlink = REQ_UNLINK 1041 aio_unlink = REQ_UNLINK
1002 aio_rmdir = REQ_RMDIR 1042 aio_rmdir = REQ_RMDIR
1003 aio_readdir = REQ_READDIR 1043 aio_readdir = REQ_READDIR
1004 CODE: 1044 PPCODE:
1005{ 1045{
1006 dREQ; 1046 dREQ;
1007 1047
1008 req->type = ix; 1048 req->type = ix;
1009 req->data = newSVsv (pathname); 1049 req->data = newSVsv (pathname);
1010 req->dataptr = SvPVbyte_nolen (req->data); 1050 req->dataptr = SvPVbyte_nolen (req->data);
1011 1051
1012 send_req (req); 1052 REQ_SEND;
1013} 1053}
1014 1054
1015void 1055void
1016aio_link (oldpath,newpath,callback=&PL_sv_undef) 1056aio_link (oldpath,newpath,callback=&PL_sv_undef)
1017 SV * oldpath 1057 SV * oldpath
… …
1019 SV * callback 1059 SV * callback
1020 ALIAS: 1060 ALIAS:
1021 aio_link = REQ_LINK 1061 aio_link = REQ_LINK
1022 aio_symlink = REQ_SYMLINK 1062 aio_symlink = REQ_SYMLINK
1023 aio_rename = REQ_RENAME 1063 aio_rename = REQ_RENAME
1024 CODE: 1064 PPCODE:
1025{ 1065{
1026 dREQ; 1066 dREQ;
1027 1067
1028 req->type = ix; 1068 req->type = ix;
1029 req->fh = newSVsv (oldpath); 1069 req->fh = newSVsv (oldpath);
1030 req->data2ptr = SvPVbyte_nolen (req->fh); 1070 req->data2ptr = SvPVbyte_nolen (req->fh);
1031 req->data = newSVsv (newpath); 1071 req->data = newSVsv (newpath);
1032 req->dataptr = SvPVbyte_nolen (req->data); 1072 req->dataptr = SvPVbyte_nolen (req->data);
1033 1073
1034 send_req (req); 1074 REQ_SEND;
1075}
1076
1077void
1078aio_group (callback=&PL_sv_undef)
1079 SV * callback
1080 PROTOTYPE: ;&
1081 PPCODE:
1082{
1083 dREQ;
1084 req->type = REQ_GROUP;
1085 XPUSHs (req_sv (req, AIO_GRP_KLASS));
1035} 1086}
1036 1087
1037void 1088void
1038flush () 1089flush ()
1039 PROTOTYPE: 1090 PROTOTYPE:
… …
1083 CODE: 1134 CODE:
1084 RETVAL = nreqs; 1135 RETVAL = nreqs;
1085 OUTPUT: 1136 OUTPUT:
1086 RETVAL 1137 RETVAL
1087 1138
1139MODULE = IO::AIO PACKAGE = IO::AIO::REQ
1140
1141void
1142cancel (aio_req_ornot req)
1143 PROTOTYPE:
1144 CODE:
1145 req->cancelled = 1;
1146

Diff Legend

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