Unified · Split
| @@ -48,14 +48,15 @@ enum · 47 unmodified lines | ||
| 48 | 48 | char ENoRemoteResources[] = "No remote resources available"; |
| 49 | 49 | char ENoResourceMatch[] = "No resources matching request"; |
| 50 | 50 | char ENOReservation[] = "No remote reservation done"; |
| 51 | −static int vflag = 0; /* for debugging messages: control prints */ | |
| 51 | +static int vflag = 1; /* for debugging messages: control prints */ | |
| 52 | 52 | |
| 53 | 53 | long lastrrselected = 0; |
| 54 | 54 | |
| 55 | 55 | struct RemoteFile |
| 56 | 56 | { |
| 57 | − RWlock l; | |
| 57 | + int state; | |
| 58 | 58 | Chan *cfile; |
| 59 | + Queue *buffer; | |
| 59 | 60 | }; |
| 60 | 61 | typedef struct RemoteFile RemoteFile; |
| 61 | 62 | |
| @@ -80,6 +81,7 @@ struct RemoteJob · 18 unmodified lines | ||
| 80 | 81 | RemoteResource *first; |
| 81 | 82 | RemoteResource *last; |
| 82 | 83 | }; |
| 84 | + | |
| 83 | 85 | typedef struct RemoteJob RemoteJob; |
| 84 | 86 | |
| 85 | 87 | #define PLATFORMCOUNT 7 |
| @@ -411,6 +413,46 @@ cmdstat (Chan *c, uchar *db, int n) · 325 unmodified lines | ||
| 411 | 413 | return devstat (c, db, n, 0, 0, cmdgen); |
| 412 | 414 | } |
| 413 | 415 | |
| 416 | +static void | |
| 417 | +procreadremote (void *a) | |
| 418 | +{ | |
| 419 | + RemoteFile *rf; | |
| 420 | + Block *content = nil; | |
| 421 | + | |
| 422 | + rf = a; | |
| 423 | + rf->state = 1; /* thread running */ | |
| 424 | + | |
| 425 | + if (waserror ()){ | |
| 426 | + if (vflag) print ("#### PRR[%s]:No more data to read\n", | |
| 427 | + rf->cfile->name->s); | |
| 428 | + freeb (content); | |
| 429 | + rf->state = 2; /* thread dead */ | |
| 430 | + pexit ("", 0); | |
| 431 | + } | |
| 432 | + while (1) { | |
| 433 | + | |
| 434 | + if (vflag) print ("#### PRR: reading from [%s]\n", | |
| 435 | + rf->cfile->name->s); | |
| 436 | + content = devbread (rf->cfile, 1024, 0); | |
| 437 | + if (BLEN (content) == 0) { | |
| 438 | + /* check the status of this child, | |
| 439 | + if it is done, then exit | |
| 440 | + else read again */ | |
| 441 | + break; | |
| 442 | + } /* end if */ | |
| 443 | + | |
| 444 | + /* Write data to queue */ | |
| 445 | + qbwrite (rf->buffer, content); | |
| 446 | + } /* end while : infinite */ | |
| 447 | + | |
| 448 | + if (vflag) print ("#### PRR[%s]:No more data to read\n", | |
| 449 | + rf->cfile->name->s); | |
| 450 | + freeb (content); | |
| 451 | + rf->state = 2; /* thread dead */ | |
| 452 | + pexit ("", 0); | |
| 453 | +} /* end function : procreadremote () */ | |
| 454 | + | |
| 455 | + | |
| 414 | 456 | /* Opens file/dir |
| 415 | 457 | It checks if specified channel is allowed to open, |
| 416 | 458 | if yes, it marks channel as open and returns it. |
| @@ -428,6 +470,7 @@ cmdopen (Chan *c, int omode) · 11 unmodified lines | ||
| 428 | 470 | int i; |
| 429 | 471 | int tmpfileuse; |
| 430 | 472 | char *fname; |
| 473 | + RemoteFile *rf; | |
| 431 | 474 | |
| 432 | 475 | |
| 433 | 476 | if (vflag) print ("trying to open [%s]\n", c->name->s); |
| @@ -580,7 +623,15 @@ cmdopen (Chan *c, int omode) · 146 unmodified lines | ||
| 580 | 623 | tmpchan = namec (buff,Aopen,omode,0); |
| 581 | 624 | |
| 582 | 625 | wlock (&cv->rjob->l); |
| 583 | − tmprr->remotefiles[filetype].cfile = tmpchan; | |
| 626 | + rf = &tmprr->remotefiles[filetype]; | |
| 627 | + rf->cfile = tmpchan; | |
| 628 | + | |
| 629 | + | |
| 630 | + if (TYPE (c->qid) == Qdata ){ | |
| 631 | + /* FIXME: start the reader thread */ | |
| 632 | + if (vflag) print ("\n######thread starting" ); | |
| 633 | + kproc ("procreadremote", procreadremote, rf, 0 ); | |
| 634 | + } | |
| 584 | 635 | tmprr = tmprr->next; /* this can be RO lock*/ |
| 585 | 636 | wunlock (&cv->rjob->l); |
| 586 | 637 | } /* end for : */ |
| @@ -615,6 +666,7 @@ freeremoteresource (RemoteResource *rr) · 28 unmodified lines | ||
| 615 | 666 | if (rr->remotefiles[i].cfile != nil) { |
| 616 | 667 | cclose (rr->remotefiles[i].cfile); |
| 617 | 668 | rr->remotefiles[i].cfile = nil; |
| 669 | + qfree (rr->remotefiles[i].buffer); | |
| 618 | 670 | /* unbind ?? |
| 619 | 671 | You can't unbind here as rjobcount is already |
| 620 | 672 | reset to zero. |
| @@ -795,6 +847,71 @@ cmdclose (Chan *c) · 174 unmodified lines | ||
| 795 | 847 | if (vflag) print ("close complete\n"); |
| 796 | 848 | } |
| 797 | 849 | |
| 850 | +static long | |
| 851 | +readfromallasync (Chan *ch, void *a, long n, vlong offset) | |
| 852 | +{ | |
| 853 | + void *c_a; | |
| 854 | + vlong c_offset; | |
| 855 | + long ret, c_ret, c_n, i; | |
| 856 | + Conv *c; | |
| 857 | + long tmpjc; | |
| 858 | + int filetype; | |
| 859 | + RemoteResource *tmprr; | |
| 860 | + RemoteFile *rf; | |
| 861 | + | |
| 862 | + USED (offset); | |
| 863 | + c = cmd.conv[CONV (ch->qid)]; | |
| 864 | + | |
| 865 | + tmpjc = getrjobcount (c->rjob); | |
| 866 | + /* making sure that remote resources are allocated */ | |
| 867 | + if (tmpjc < 0) { | |
| 868 | + if (vflag) print ("resources not reserved\n"); | |
| 869 | + error (ENOReservation); | |
| 870 | + } | |
| 871 | + | |
| 872 | + if (vflag) print ("readfromallasync cname [%s]\n", ch->name->s); | |
| 873 | + | |
| 874 | + filetype = TYPE(ch->qid) - Qdata; | |
| 875 | + /* read from multiple remote files */ | |
| 876 | + c_a = a; | |
| 877 | + c_n = n; | |
| 878 | + c_offset = offset; | |
| 879 | + ret = 0; | |
| 880 | + | |
| 881 | + rlock (&c->rjob->l); | |
| 882 | + tmprr = c->rjob->first ; | |
| 883 | + runlock (&c->rjob->l); | |
| 884 | + | |
| 885 | + for (i = 0 ; i < tmpjc; ++i) { | |
| 886 | + rf = &tmprr->remotefiles[filetype]; | |
| 887 | +/* if ( qcanread (rf->buffer)) { | |
| 888 | + len = qlen (rf->buffer); | |
| 889 | + if (len <= c_n ) { | |
| 890 | + cc_n = len; | |
| 891 | + } else { | |
| 892 | + cc_n = c_n; | |
| 893 | + } | |
| 894 | +*/ | |
| 895 | + /* read from queue */ | |
| 896 | + c_ret = qconsume (rf->buffer, c_a, c_n); | |
| 897 | + if (vflag) print ("%ldth read gave [%ld] data\n", i, c_ret); | |
| 898 | + if (c_ret > 0 ) { | |
| 899 | + ret = ret + c_ret; | |
| 900 | + c_a = (uchar *)c_a + c_ret; | |
| 901 | + c_n = c_n - c_ret; | |
| 902 | + } | |
| 903 | + if ( c_n <= 0 ) { | |
| 904 | + break; | |
| 905 | + } | |
| 906 | + | |
| 907 | + /* FIXME: should I lock following also in read mode ?? */ | |
| 908 | + tmprr = tmprr->next; | |
| 909 | + } /* end for : */ | |
| 910 | + | |
| 911 | + if (vflag) print ("readfromallasync [%ld]\n", ret); | |
| 912 | + return ret; | |
| 913 | +} /* end function : readfromallasync */ | |
| 914 | + | |
| 798 | 915 | static long |
| 799 | 916 | readfromall (Chan *ch, void *a, long n, vlong offset) |
| 800 | 917 | { |
| @@ -1191,6 +1308,8 @@ addrr (char *localpath, char *selected, RemoteJob *rjob, int jcount) · 390 unmodified lines | ||
| 1191 | 1308 | pp->next = nil; |
| 1192 | 1309 | for (i = 0; i < RCHANCOUNT; ++i ) { |
| 1193 | 1310 | pp->remotefiles[i].cfile = nil; |
| 1311 | + pp->remotefiles[i].buffer = qopen (1024, 0, nil, 0); | |
| 1312 | + pp->remotefiles[i].state = 0; | |
| 1194 | 1313 | } |
| 1195 | 1314 | |
| 1196 | 1315 | wlock (&rjob->l); |
| @@ -1480,7 +1599,11 @@ cmdread (Chan *ch, void *a, long n, vlong offset) · 283 unmodified lines | ||
| 1480 | 1599 | if (tmpjc < 0) error(ENOReservation); |
| 1481 | 1600 | |
| 1482 | 1601 | if (tmpjc > 0 ) { |
| 1483 | − return readfromall (ch, a, n, offset); | |
| 1602 | + if (TYPE (ch->qid) == Qdata ) { | |
| 1603 | + return readfromallasync (ch, a, n, offset); | |
| 1604 | + } else { | |
| 1605 | + return readfromall (ch, a, n, offset); | |
| 1606 | + } | |
| 1484 | 1607 | } |
| 1485 | 1608 | |
| 1486 | 1609 | /* getting data locally */ |
| @@ -1672,13 +1795,13 @@ cmdwrite (Chan *ch, void *a, long n, vlong offset) · 185 unmodified lines | ||
| 1672 | 1795 | nexterror (); |
| 1673 | 1796 | } |
| 1674 | 1797 | |
| 1675 | − if (vflag) print ("### cmd arg[%d] done [%s]\n", cb->nf, a); | |
| 1798 | + if (vflag) print ("cmd arg[%d] done [%s]\n", cb->nf, a); | |
| 1676 | 1799 | ct = lookupcmd (cb, cmdtab, nelem (cmdtab)); |
| 1677 | 1800 | switch (ct->index){ |
| 1678 | 1801 | case CMres : |
| 1679 | − if (vflag) print("###reserving resources %s \n",cb->f[1]); | |
| 1802 | + if (vflag) print("reserving resources %s \n",cb->f[1]); | |
| 1680 | 1803 | resNo = atol (cb->f[1]); |
| 1681 | − if (vflag) print ("###reserving resources %s [%ld]\n", | |
| 1804 | + if (vflag) print ("reserving resources %s [%ld]\n", | |
| 1682 | 1805 | cb->f[1], resNo); |
| 1683 | 1806 | |
| 1684 | 1807 | if (resNo < 0) { |
| @@ -1705,8 +1828,8 @@ cmdwrite (Chan *ch, void *a, long n, vlong offset) · 20 unmodified lines | ||
| 1705 | 1828 | preparelocalexecution (c, os, arch); |
| 1706 | 1829 | break; |
| 1707 | 1830 | } |
| 1708 | − groupres (ch, resNo, os, arch); | |
| 1709 | − if (vflag) print ("####Reservation done\n" ); | |
| 1831 | + groupres (ch, resNo, os, arch); | |
| 1832 | + if (vflag) print ("Reservation done\n" ); | |
| 1710 | 1833 | break; |
| 1711 | 1834 | |
| 1712 | 1835 | case CMexec: |