History

Added parallel write support to stdio. Now, both read and write operations on stdio are parallel.

shindepravin committed 16 years ago

Commit 95751f21 parent 485a8c81 file · +67 −128Browse files

Unified · Split

usr/brasil/emu/port/devtask.c+67−128
@@ -534,19 +534,19 @@ proc_splice (void *param) · 533 unmodified lines
534 534
535 535
536static void536static void
537remoteopen (Chan *c, int omode, char *fname, int resno )537remoteopen (Chan *c, int omode, char *fname, int filetype, int resno )
538{538{
539 Conv *cv;539 Conv *cv;
540 int tmpfileuse;540 int tmpfileuse;
541 RemoteResource *tmprr;541 RemoteResource *tmprr;
542 int filetype;542 int stdin, stdout;
543 RemoteFile *rf;543 RemoteFile *rf;
544 char buff[KNAMELEN*3];544 char buff[KNAMELEN*3];
545 Chan *tmpchan;545 Chan *tmpchan;
546 int i;546 int i;
547 547
548 cv = cmd.conv[CONV (c->qid)];548 cv = cmd.conv[CONV (c->qid)];
549 filetype = TYPE(c->qid) - Qdata;549
550 tmpfileuse = getfileopencount (cv->rjob, filetype);550 tmpfileuse = getfileopencount (cv->rjob, filetype);
551 551
552 if ( tmpfileuse < 1) {552 if ( tmpfileuse < 1) {
@@ -564,11 +564,14 @@ remoteopen (Chan *c, int omode, char *fname, int resno ) · 11 unmodified lines
564 rf = &tmprr->remotefiles[filetype];564 rf = &tmprr->remotefiles[filetype];
565 rf->cfile = tmpchan;565 rf->cfile = tmpchan;
566 566
567 if ( (TYPE (c->qid) == Qdata) && (omode == OREAD || omode == ORDWR)){567 if ((TYPE (c->qid) == Qdata)) {
568 /* start the reader thread */568
569 if (vflag) print ("\nthread starting" );569 if (omode == OREAD || omode == ORDWR) {
570// sprint (buff, "PRR[%s]", c->name->s);570 /* start the reader thread */
571 kproc (buff, procreadremote, rf, 0 );571 if (vflag) print ("\nthread starting" );
572// sprint (buff, "PRR[%s]", c->name->s);
573 kproc (buff, procreadremote, rf, 0 );
574 }
572 }575 }
573 tmprr = tmprr->next; /* this can be RO lock*/576 tmprr = tmprr->next; /* this can be RO lock*/
574 wunlock (&cv->rjob->l);577 wunlock (&cv->rjob->l);
@@ -628,6 +631,8 @@ cmdopen (Chan *c, int omode) · 53 unmodified lines
628 case Qenv:631 case Qenv:
629 case Qargs:632 case Qargs:
630 case Qns:633 case Qns:
634 case Qstdin:
635 case Qstdout:
631 error (Egreg); /* TODO: IMPLEMENT */636 error (Egreg); /* TODO: IMPLEMENT */
632 break; 637 break;
633 638
@@ -661,7 +666,6 @@ cmdopen (Chan *c, int omode) · 27 unmodified lines
661 if (vflag) print ("clone open successful [%s]\n", c->name->s);666 if (vflag) print ("clone open successful [%s]\n", c->name->s);
662 break;667 break;
663 668
664 case Qstdout:
665 case Qstatus:669 case Qstatus:
666 case Qstderr:670 case Qstderr:
667 case Qwait:671 case Qwait:
@@ -670,14 +674,7 @@ cmdopen (Chan *c, int omode) · 2 unmodified lines
670 674
671 case Qdata: 675 case Qdata:
672 case Qctl:676 case Qctl:
673 case Qstdin: /* FIXME: make sure that file is open in write only mode */
674 677
675 if (TYPE(c->qid) == Qstdin ) {
676 if (omode != OWRITE ) {
677 error (Eperm);
678 }
679 }
680
681 qlock (&cmd.l);678 qlock (&cmd.l);
682 cv = cmd.conv[CONV (c->qid)];679 cv = cmd.conv[CONV (c->qid)];
683 qlock (&cv->l);680 qlock (&cv->l);
@@ -727,16 +724,6 @@ cmdopen (Chan *c, int omode) · 43 unmodified lines
727 }724 }
728 break;725 break;
729 726
730 case Qstdin:
731 fname = "stdin";
732 cv->count[0]++;
733 break;
734
735 case Qstdout:
736 fname = "stdout";
737 cv->count[1]++;
738 break;
739
740 case Qstderr:727 case Qstderr:
741 fname = "stderr";728 fname = "stderr";
742 if(omode != OREAD)729 if(omode != OREAD)
@@ -757,8 +744,20 @@ cmdopen (Chan *c, int omode) · 14 unmodified lines
757 fname = nil;744 fname = nil;
758 }745 }
759 746
747 /* do remote open when there are sub-sessions */
760 if ( fname != nil && tmpjc > 0 ) {748 if ( fname != nil && tmpjc > 0 ) {
761 remoteopen (c, omode, fname, tmpjc);749 if (TYPE(c->qid) == Qdata) {
750 /* special case of */
751 if(omode == OWRITE || omode == ORDWR) {
752 /* open stdin part */
753 remoteopen (c, omode, fname, (Qstdin-Qdata), tmpjc);
754 }
755 if(omode == OREAD || omode == ORDWR) {
756 remoteopen (c, omode, fname, (Qstdout-Qdata), tmpjc);
757 }
758 } else {
759 remoteopen (c, omode, fname, filetype, tmpjc);
760 }
762 } /* end if : fname != nil */761 } /* end if : fname != nil */
763 762
764 break;763 break;
@@ -862,19 +861,18 @@ closeconv (Conv *c) · 97 unmodified lines
862}861}
863 862
864static void863static void
865remoteclose (Chan *c, int rjcount)864remoteclose (Chan *c, int filetype, int rjcount)
866{865{
867 Conv *cc;866 Conv *cc;
868 RemoteResource *tmprr;867 RemoteResource *tmprr;
869 int i;868 int i;
870 int filetype;
871 int tmpfilec;869 int tmpfilec;
872 870
873 871
874 872
875 if (vflag) print ("remote closing [%s]\n", c->name->s);873 if (vflag) print ("remote closing [%s]\n", c->name->s);
876 cc = cmd.conv[CONV(c->qid)];874 cc = cmd.conv[CONV(c->qid)];
877 filetype = TYPE(c->qid) - Qdata;875
878 tmpfilec = getfileopencount (cc->rjob, filetype);876 tmpfilec = getfileopencount (cc->rjob, filetype);
879 877
880 /* close ctl files only when everything else is closed */878 /* close ctl files only when everything else is closed */
@@ -926,6 +924,7 @@ cmdclose (Chan *c) · 45 unmodified lines
926{924{
927 Conv *cc;925 Conv *cc;
928 int r;926 int r;
927 int filetype;
929 int tmpjc;928 int tmpjc;
930 929
931 if ( (c->flag & COPEN) == 0)930 if ( (c->flag & COPEN) == 0)
@@ -947,7 +946,8 @@ cmdclose (Chan *c) · 15 unmodified lines
947 946
948 cc = cmd.conv[CONV (c->qid)];947 cc = cmd.conv[CONV (c->qid)];
949 tmpjc = getrjobcount (cc->rjob);948 tmpjc = getrjobcount (cc->rjob);
950 949 filetype = TYPE(c->qid) - Qdata;
950
951 if (vflag) print ("rjobcount is [%d]\n", tmpjc);951 if (vflag) print ("rjobcount is [%d]\n", tmpjc);
952 if (tmpjc < 0) {952 if (tmpjc < 0) {
953 /* FIXME: you can only close status file953 /* FIXME: you can only close status file
@@ -974,7 +974,17 @@ cmdclose (Chan *c) · 20 unmodified lines
974 974
975 if (tmpjc > 0) {975 if (tmpjc > 0) {
976 /* close for remote resources also */976 /* close for remote resources also */
977 remoteclose (c, tmpjc);977
978 if(TYPE(c->qid) == Qdata){
979 if(c->mode == OWRITE || c->mode == ORDWR) {
980 remoteclose (c, (Qstdin-Qdata), tmpjc);
981 }
982 if(c->mode == OREAD || c->mode == ORDWR) {
983 remoteclose (c, (Qstdout-Qdata), tmpjc);
984 }
985 } else {
986 remoteclose (c, filetype, tmpjc);
987 }
978 }988 }
979 989
980 /* This code is only for CTL file */990 /* This code is only for CTL file */
@@ -1025,8 +1035,12 @@ readfromallasync (Chan *ch, void *a, long n, vlong offset) · 44 unmodified lines
1025 }1035 }
1026 1036
1027 if (vflag) print ("readfromallasync cname [%s]\n", ch->name->s);1037 if (vflag) print ("readfromallasync cname [%s]\n", ch->name->s);
1028 1038
1029 filetype = TYPE(ch->qid) - Qdata;1039 if (TYPE(ch->qid) == Qdata) {
1040 filetype = Qstdout - Qdata;
1041 } else {
1042 filetype = TYPE(ch->qid) - Qdata;
1043 }
1030 1044
1031 while (1) {1045 while (1) {
1032 rlock (&c->rjob->l);1046 rlock (&c->rjob->l);
@@ -1062,94 +1076,6 @@ readfromallasync (Chan *ch, void *a, long n, vlong offset) · 29 unmodified lines
1062} /* end function : readfromallasync */1076} /* end function : readfromallasync */
1063 1077
1064 1078
1065static long
1066readfromallasyncold (Chan *ch, void *a, long n, vlong offset)
1067{
1068 void *c_a;
1069 vlong c_offset;
1070 long ret, c_ret, c_n, i;
1071 Conv *c;
1072 long tmpjc;
1073 long ccount, bcount;
1074 int filetype;
1075 RemoteResource *tmprr;
1076 RemoteFile *rf;
1077
1078 USED (offset);
1079 c = cmd.conv[CONV (ch->qid)];
1080
1081 tmpjc = getrjobcount (c->rjob);
1082 /* making sure that remote resources are allocated */
1083 if (tmpjc < 0) {
1084 if (vflag) print ("resources not reserved\n");
1085 error (ENOReservation);
1086 }
1087
1088 if (vflag) print ("*****readfromallasync cname [%s]\n", ch->name->s);
1089
1090 filetype = TYPE(ch->qid) - Qdata;
1091
1092
1093 /* read from multiple remote files */
1094 c_a = a;
1095 c_n = n;
1096 c_offset = offset;
1097 ret = 0;
1098
1099 while (1) {
1100
1101 rlock (&c->rjob->l);
1102 tmprr = c->rjob->first;
1103 runlock (&c->rjob->l);
1104
1105 ccount = 0;
1106 bcount = 0;
1107 for (i = 0 ; i < tmpjc; ++i) {
1108 rf = &tmprr->remotefiles[filetype];
1109
1110 /* read from queue */
1111 c_ret = qconsume (rf->async_queue, c_a, c_n);
1112 if (vflag) print ("******%ldth read gave [%ld] data, state[%d]\n", i, c_ret, rf->state);
1113 if (c_ret >= 0 ) {
1114 ret = ret + c_ret;
1115 c_a = (uchar *)c_a + c_ret;
1116 c_n = c_n - c_ret;
1117 }
1118 else {
1119 if (rf->state == 2 ) {
1120 /* This thread is done */
1121 ++ccount;
1122 } else {
1123 ++bcount;
1124 }
1125
1126 }
1127 if ( c_n <= 0 ) {
1128 break;
1129 }
1130
1131 /* FIXME: should I lock following also in read mode ?? */
1132 tmprr = tmprr->next;
1133 } /* end for : */
1134
1135 if (ret > 0 ) {
1136 break;
1137 }
1138
1139 if (ccount >= tmpjc) {
1140 if (vflag) print ("----readfromallasync fileclose [%ld]\n",
1141 ccount );
1142 /* as all nodes are close, return NOW */
1143 break;
1144 }
1145 if (vflag) print ("---readfromallasync repeating with [%ld][%ld]\n", ccount, bcount);
1146
1147 } /* end while : infinite */
1148
1149 if (vflag) print ("******readfromallasync [%ld]\n", ret);
1150 return ret;
1151} /* end function : readfromallasync */
1152
1153static long 1079static long
1154readfromall (Chan *ch, void *a, long n, vlong offset)1080readfromall (Chan *ch, void *a, long n, vlong offset)
1155{1081{
@@ -1172,8 +1098,13 @@ readfromall (Chan *ch, void *a, long n, vlong offset) · 16 unmodified lines
1172 }1098 }
1173 1099
1174 if (vflag) print ("readfromall cname [%s]\n", ch->name->s);1100 if (vflag) print ("readfromall cname [%s]\n", ch->name->s);
1175 1101
1176 filetype = TYPE(ch->qid) - Qdata;1102 if (TYPE(ch->qid) == Qdata) {
1103 filetype = Qstdout - Qdata;
1104 } else {
1105 filetype = TYPE(ch->qid) - Qdata;
1106 }
1107
1177 /* read from multiple remote files */1108 /* read from multiple remote files */
1178 c_a = a;1109 c_a = a;
1179 c_n = n;1110 c_n = n;
@@ -2166,7 +2097,11 @@ static long p_send2all (Chan *ch, void *a, long n, vlong offset) · 986 unmodified lines
2166 error (ENOReservation);2097 error (ENOReservation);
2167 }2098 }
2168 2099
2169 filetype = TYPE(ch->qid) - Qdata;2100 if (TYPE(ch->qid) == Qdata) {
2101 filetype = Qstdin - Qdata;
2102 } else {
2103 filetype = TYPE(ch->qid) - Qdata;
2104 }
2170 2105
2171 /* data should be sent to all resources */2106 /* data should be sent to all resources */
2172 wlock (&c->rjob->l);2107 wlock (&c->rjob->l);
@@ -2256,8 +2191,12 @@ sendtoall (Chan *ch, void *a, long n, vlong offset) · 83 unmodified lines
2256 if (vflag) print ("resources not reserved\n");2191 if (vflag) print ("resources not reserved\n");
2257 error (ENOReservation);2192 error (ENOReservation);
2258 }2193 }
2259 2194
2260 filetype = TYPE(ch->qid) - Qdata;2195 if (TYPE(ch->qid) == Qdata) {
2196 filetype = Qstdin - Qdata;
2197 } else {
2198 filetype = TYPE(ch->qid) - Qdata;
2199 }
2261 2200
2262 /* data should be sent to all resources */2201 /* data should be sent to all resources */
2263 rlock (&c->rjob->l);2202 rlock (&c->rjob->l);
@@ -2613,7 +2552,7 @@ cmdwrite (Chan *ch, void *a, long n, vlong offset) · 349 unmodified lines
2613 } /* end if : local file write request */2552 } /* end if : local file write request */
2614 2553
2615 /* sending to remote resources */2554 /* sending to remote resources */
2616 return sendtoall (ch, a, n, offset);2555 return p_send2all (ch, a, n, offset);
2617 } /* end switch : on filename */2556 } /* end switch : on filename */
2618 2557
2619 return ret;2558 return ret;