sockfile.cpp 175 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558255925602561256225632564256525662567256825692570257125722573257425752576257725782579258025812582258325842585258625872588258925902591259225932594259525962597259825992600260126022603260426052606260726082609261026112612261326142615261626172618261926202621262226232624262526262627262826292630263126322633263426352636263726382639264026412642264326442645264626472648264926502651265226532654265526562657265826592660266126622663266426652666266726682669267026712672267326742675267626772678267926802681268226832684268526862687268826892690269126922693269426952696269726982699270027012702270327042705270627072708270927102711271227132714271527162717271827192720272127222723272427252726272727282729273027312732273327342735273627372738273927402741274227432744274527462747274827492750275127522753275427552756275727582759276027612762276327642765276627672768276927702771277227732774277527762777277827792780278127822783278427852786278727882789279027912792279327942795279627972798279928002801280228032804280528062807280828092810281128122813281428152816281728182819282028212822282328242825282628272828282928302831283228332834283528362837283828392840284128422843284428452846284728482849285028512852285328542855285628572858285928602861286228632864286528662867286828692870287128722873287428752876287728782879288028812882288328842885288628872888288928902891289228932894289528962897289828992900290129022903290429052906290729082909291029112912291329142915291629172918291929202921292229232924292529262927292829292930293129322933293429352936293729382939294029412942294329442945294629472948294929502951295229532954295529562957295829592960296129622963296429652966296729682969297029712972297329742975297629772978297929802981298229832984298529862987298829892990299129922993299429952996299729982999300030013002300330043005300630073008300930103011301230133014301530163017301830193020302130223023302430253026302730283029303030313032303330343035303630373038303930403041304230433044304530463047304830493050305130523053305430553056305730583059306030613062306330643065306630673068306930703071307230733074307530763077307830793080308130823083308430853086308730883089309030913092309330943095309630973098309931003101310231033104310531063107310831093110311131123113311431153116311731183119312031213122312331243125312631273128312931303131313231333134313531363137313831393140314131423143314431453146314731483149315031513152315331543155315631573158315931603161316231633164316531663167316831693170317131723173317431753176317731783179318031813182318331843185318631873188318931903191319231933194319531963197319831993200320132023203320432053206320732083209321032113212321332143215321632173218321932203221322232233224322532263227322832293230323132323233323432353236323732383239324032413242324332443245324632473248324932503251325232533254325532563257325832593260326132623263326432653266326732683269327032713272327332743275327632773278327932803281328232833284328532863287328832893290329132923293329432953296329732983299330033013302330333043305330633073308330933103311331233133314331533163317331833193320332133223323332433253326332733283329333033313332333333343335333633373338333933403341334233433344334533463347334833493350335133523353335433553356335733583359336033613362336333643365336633673368336933703371337233733374337533763377337833793380338133823383338433853386338733883389339033913392339333943395339633973398339934003401340234033404340534063407340834093410341134123413341434153416341734183419342034213422342334243425342634273428342934303431343234333434343534363437343834393440344134423443344434453446344734483449345034513452345334543455345634573458345934603461346234633464346534663467346834693470347134723473347434753476347734783479348034813482348334843485348634873488348934903491349234933494349534963497349834993500350135023503350435053506350735083509351035113512351335143515351635173518351935203521352235233524352535263527352835293530353135323533353435353536353735383539354035413542354335443545354635473548354935503551355235533554355535563557355835593560356135623563356435653566356735683569357035713572357335743575357635773578357935803581358235833584358535863587358835893590359135923593359435953596359735983599360036013602360336043605360636073608360936103611361236133614361536163617361836193620362136223623362436253626362736283629363036313632363336343635363636373638363936403641364236433644364536463647364836493650365136523653365436553656365736583659366036613662366336643665366636673668366936703671367236733674367536763677367836793680368136823683368436853686368736883689369036913692369336943695369636973698369937003701370237033704370537063707370837093710371137123713371437153716371737183719372037213722372337243725372637273728372937303731373237333734373537363737373837393740374137423743374437453746374737483749375037513752375337543755375637573758375937603761376237633764376537663767376837693770377137723773377437753776377737783779378037813782378337843785378637873788378937903791379237933794379537963797379837993800380138023803380438053806380738083809381038113812381338143815381638173818381938203821382238233824382538263827382838293830383138323833383438353836383738383839384038413842384338443845384638473848384938503851385238533854385538563857385838593860386138623863386438653866386738683869387038713872387338743875387638773878387938803881388238833884388538863887388838893890389138923893389438953896389738983899390039013902390339043905390639073908390939103911391239133914391539163917391839193920392139223923392439253926392739283929393039313932393339343935393639373938393939403941394239433944394539463947394839493950395139523953395439553956395739583959396039613962396339643965396639673968396939703971397239733974397539763977397839793980398139823983398439853986398739883989399039913992399339943995399639973998399940004001400240034004400540064007400840094010401140124013401440154016401740184019402040214022402340244025402640274028402940304031403240334034403540364037403840394040404140424043404440454046404740484049405040514052405340544055405640574058405940604061406240634064406540664067406840694070407140724073407440754076407740784079408040814082408340844085408640874088408940904091409240934094409540964097409840994100410141024103410441054106410741084109411041114112411341144115411641174118411941204121412241234124412541264127412841294130413141324133413441354136413741384139414041414142414341444145414641474148414941504151415241534154415541564157415841594160416141624163416441654166416741684169417041714172417341744175417641774178417941804181418241834184418541864187418841894190419141924193419441954196419741984199420042014202420342044205420642074208420942104211421242134214421542164217421842194220422142224223422442254226422742284229423042314232423342344235423642374238423942404241424242434244424542464247424842494250425142524253425442554256425742584259426042614262426342644265426642674268426942704271427242734274427542764277427842794280428142824283428442854286428742884289429042914292429342944295429642974298429943004301430243034304430543064307430843094310431143124313431443154316431743184319432043214322432343244325432643274328432943304331433243334334433543364337433843394340434143424343434443454346434743484349435043514352435343544355435643574358435943604361436243634364436543664367436843694370437143724373437443754376437743784379438043814382438343844385438643874388438943904391439243934394439543964397439843994400440144024403440444054406440744084409441044114412441344144415441644174418441944204421442244234424442544264427442844294430443144324433443444354436443744384439444044414442444344444445444644474448444944504451445244534454445544564457445844594460446144624463446444654466446744684469447044714472447344744475447644774478447944804481448244834484448544864487448844894490449144924493449444954496449744984499450045014502450345044505450645074508450945104511451245134514451545164517451845194520452145224523452445254526452745284529453045314532453345344535453645374538453945404541454245434544454545464547454845494550455145524553455445554556455745584559456045614562456345644565456645674568456945704571457245734574457545764577457845794580458145824583458445854586458745884589459045914592459345944595459645974598459946004601460246034604460546064607460846094610461146124613461446154616461746184619462046214622462346244625462646274628462946304631463246334634463546364637463846394640464146424643464446454646464746484649465046514652465346544655465646574658465946604661466246634664466546664667466846694670467146724673467446754676467746784679468046814682468346844685468646874688468946904691469246934694469546964697469846994700470147024703470447054706470747084709471047114712471347144715471647174718471947204721472247234724472547264727472847294730473147324733473447354736473747384739474047414742474347444745474647474748474947504751475247534754475547564757475847594760476147624763476447654766476747684769477047714772477347744775477647774778477947804781478247834784478547864787478847894790479147924793479447954796479747984799480048014802480348044805480648074808480948104811481248134814481548164817481848194820482148224823482448254826482748284829483048314832483348344835483648374838483948404841484248434844484548464847484848494850485148524853485448554856485748584859486048614862486348644865486648674868486948704871487248734874487548764877487848794880488148824883488448854886488748884889489048914892489348944895489648974898489949004901490249034904490549064907490849094910491149124913491449154916491749184919492049214922492349244925492649274928492949304931493249334934493549364937493849394940494149424943494449454946494749484949495049514952495349544955495649574958495949604961496249634964496549664967496849694970497149724973497449754976497749784979498049814982498349844985498649874988498949904991499249934994499549964997499849995000500150025003500450055006500750085009501050115012501350145015501650175018501950205021502250235024502550265027502850295030503150325033503450355036503750385039504050415042504350445045504650475048504950505051505250535054505550565057505850595060506150625063506450655066506750685069507050715072507350745075507650775078507950805081508250835084508550865087508850895090509150925093509450955096509750985099510051015102510351045105510651075108510951105111511251135114511551165117511851195120512151225123512451255126512751285129513051315132513351345135513651375138513951405141514251435144514551465147514851495150515151525153515451555156515751585159516051615162516351645165516651675168516951705171517251735174517551765177517851795180518151825183518451855186518751885189519051915192519351945195519651975198519952005201520252035204520552065207520852095210521152125213521452155216521752185219522052215222522352245225522652275228522952305231523252335234523552365237523852395240524152425243524452455246524752485249525052515252525352545255525652575258525952605261526252635264526552665267526852695270527152725273527452755276527752785279528052815282528352845285528652875288528952905291529252935294529552965297529852995300530153025303530453055306530753085309531053115312531353145315531653175318531953205321532253235324532553265327532853295330533153325333533453355336533753385339534053415342
  1. /*##############################################################################
  2. HPCC SYSTEMS software Copyright (C) 2012 HPCC Systems®.
  3. Licensed under the Apache License, Version 2.0 (the "License");
  4. you may not use this file except in compliance with the License.
  5. You may obtain a copy of the License at
  6. http://www.apache.org/licenses/LICENSE-2.0
  7. Unless required by applicable law or agreed to in writing, software
  8. distributed under the License is distributed on an "AS IS" BASIS,
  9. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. See the License for the specific language governing permissions and
  11. limitations under the License.
  12. ############################################################################## */
  13. // todo look at IRemoteFileServer stop
  14. #include "platform.h"
  15. #include "limits.h"
  16. #include "jlib.hpp"
  17. #include "jio.hpp"
  18. #include "jmutex.hpp"
  19. #include "jfile.hpp"
  20. #include "jmisc.hpp"
  21. #include "jthread.hpp"
  22. #include "jqueue.tpp"
  23. #include "securesocket.hpp"
  24. #include "sockfile.hpp"
  25. #include "portlist.h"
  26. #include "jsocket.hpp"
  27. #include "jencrypt.hpp"
  28. #include "jset.hpp"
  29. #include "remoteerr.hpp"
  30. #define SOCKET_CACHE_MAX 500
  31. #ifdef _DEBUG
  32. //#define SIMULATE_PACKETLOSS 1
  33. #endif
  34. #define TREECOPYTIMEOUT (60*60*1000) // 1Hr (I guess could take longer for big file but at least will stagger)
  35. #define TREECOPYPOLLTIME (60*1000*5) // for tracing that delayed
  36. #define TREECOPYPRUNETIME (24*60*60*1000) // 1 day
  37. #if SIMULATE_PACKETLOSS
  38. #define TESTING_FAILURE_RATE_LOST_SEND 10 // per 1000
  39. #define TESTING_FAILURE_RATE_LOST_RECV 10 // per 1000
  40. #define DUMMY_TIMEOUT_MAX (1000*10)
  41. static bool errorSimulationOn = true;
  42. static ISocket *timeoutreadsock = NULL; // used to trigger
  43. struct dummyReadWrite
  44. {
  45. class X
  46. {
  47. dummyReadWrite *parent;
  48. public:
  49. X(dummyReadWrite *_parent)
  50. {
  51. parent = _parent;
  52. }
  53. ~X()
  54. {
  55. delete parent;
  56. }
  57. };
  58. class TimeoutSocketException: public CInterface, public IJSOCK_Exception
  59. {
  60. public:
  61. IMPLEMENT_IINTERFACE;
  62. TimeoutSocketException()
  63. {
  64. }
  65. virtual ~TimeoutSocketException()
  66. {
  67. }
  68. int errorCode() const { return JSOCKERR_timeout_expired; }
  69. StringBuffer & errorMessage(StringBuffer &str) const
  70. {
  71. return str.append("timeout expired");
  72. }
  73. MessageAudience errorAudience() const
  74. {
  75. return MSGAUD_user;
  76. }
  77. };
  78. ISocket *sock;
  79. dummyReadWrite(ISocket *_sock)
  80. {
  81. sock = _sock;
  82. }
  83. void readtms(void* buf, size32_t min_size, size32_t max_size, size32_t &size_read, time_t timeout)
  84. {
  85. X x(this);
  86. unsigned t = msTick();
  87. unsigned r = getRandom();
  88. bool timeoutread = (timeoutreadsock==sock);
  89. timeoutreadsock=NULL;
  90. if (!timeoutread)
  91. sock->readtms(buf, min_size, max_size, size_read, timeout);
  92. if (timeoutread||(errorSimulationOn&&(TESTING_FAILURE_RATE_LOST_RECV>0)&&(r%1000<TESTING_FAILURE_RATE_LOST_RECV))) {
  93. PrintStackReport();
  94. if (timeoutread)
  95. PROGLOG("** Simulate timeout");
  96. else
  97. PROGLOG("** Simulate Packet loss (size %d,%d)",min_size,max_size);
  98. if (timeout>DUMMY_TIMEOUT_MAX)
  99. timeout = DUMMY_TIMEOUT_MAX;
  100. t = msTick()-t;
  101. if (t<timeout)
  102. Sleep(timeout-t);
  103. IJSOCK_Exception *e = new TimeoutSocketException;
  104. throw e;
  105. }
  106. }
  107. size32_t write(void const* buf, size32_t size)
  108. {
  109. X x(this);
  110. timeoutreadsock=NULL;
  111. unsigned r = getRandom();
  112. if (errorSimulationOn&&(TESTING_FAILURE_RATE_LOST_SEND>0)&&(r%1000<TESTING_FAILURE_RATE_LOST_SEND)) {
  113. PrintStackReport();
  114. PROGLOG("** Simulate Packet loss (size %d)",size);
  115. timeoutreadsock=sock;
  116. return size;
  117. }
  118. return sock->write(buf,size);
  119. }
  120. };
  121. #define SOCKWRITE(sock) (new dummyReadWrite(sock))->write
  122. #define SOCKREADTMS(sock) (new dummyReadWrite(sock))->readtms
  123. #else
  124. #define SOCKWRITE(sock) sock->write
  125. #define SOCKREADTMS(sock) sock->readtms
  126. #endif
  127. // backward compatible modes
  128. typedef enum { compatIFSHnone, compatIFSHread, compatIFSHwrite, compatIFSHexec, compatIFSHall} compatIFSHmode;
  129. static const char *VERSTRING= "DS V1.9" // dont forget FILESRV_VERSION in header
  130. #ifdef _WIN32
  131. "Windows ";
  132. #else
  133. "Linux ";
  134. #endif
  135. typedef unsigned char RemoteFileCommandType;
  136. typedef int RemoteFileIOHandle;
  137. static unsigned maxConnectTime = 0;
  138. static unsigned maxReceiveTime = 0;
  139. //Security and default port attributes
  140. static class _securitySettings
  141. {
  142. public:
  143. bool useSSL;
  144. unsigned short daFileSrvPort;
  145. const char * certificate;
  146. const char * privateKey;
  147. _securitySettings()
  148. {
  149. querySecuritySettings(&useSSL, &daFileSrvPort, &certificate, &privateKey);
  150. }
  151. } securitySettings;
  152. static CriticalSection secureContextCrit;
  153. static Owned<ISecureSocketContext> secureContext;
  154. static ISecureSocket *createSecureSocket(ISocket *sock,SecureSocketType type)
  155. {
  156. {
  157. CriticalBlock b(secureContextCrit);
  158. if (!secureContext)
  159. {
  160. if (securitySettings.certificate)
  161. secureContext.setown(createSecureSocketContextEx(securitySettings.certificate,securitySettings.privateKey, NULL, type));
  162. else
  163. secureContext.setown(createSecureSocketContext(type));
  164. }
  165. }
  166. #ifdef _DEBUG
  167. return secureContext->createSecureSocket(sock, SSLogMax);
  168. #else
  169. return secureContext->createSecureSocket(sock);
  170. #endif
  171. }
  172. void clientSetRemoteFileTimeouts(unsigned maxconnecttime,unsigned maxreadtime)
  173. {
  174. maxConnectTime = maxconnecttime;
  175. maxReceiveTime = maxreadtime;
  176. }
  177. struct sRFTM
  178. {
  179. CTimeMon *timemon;
  180. sRFTM() { timemon = maxReceiveTime?new CTimeMon(maxReceiveTime):NULL; }
  181. ~sRFTM() { delete timemon; }
  182. };
  183. const char *remoteServerVersionString() { return VERSTRING; }
  184. static bool AuthenticationEnabled = true;
  185. bool enableDafsAuthentication(bool on)
  186. {
  187. bool ret = AuthenticationEnabled;
  188. AuthenticationEnabled = on;
  189. return ret;
  190. }
  191. #define CLIENT_TIMEOUT (1000*60*60*12) // long timeout in case zombies
  192. #define CLIENT_INACTIVEWARNING_TIMEOUT (1000*60*60*12) // time between logging inactive clients
  193. #define SERVER_TIMEOUT (1000*60*5) // timeout when waiting for dafilesrv to reply after command
  194. // (increased when waiting for large block)
  195. #define DAFS_CONNECT_FAIL_RETRY_TIME (1000*60*15)
  196. #ifdef SIMULATE_PACKETLOSS
  197. #define NORMAL_RETRIES (1)
  198. #define LENGTHY_RETRIES (1)
  199. #else
  200. #define NORMAL_RETRIES (3)
  201. #define LENGTHY_RETRIES (12)
  202. #endif
  203. #ifdef _DEBUG
  204. static byte traceFlags=0x30;
  205. #else
  206. static byte traceFlags=0x20;
  207. #endif
  208. #define TF_TRACE (traceFlags&1)
  209. #define TF_TRACE_PRE_IO (traceFlags&2)
  210. #define TF_TRACE_FULL (traceFlags&4)
  211. #define TF_TRACE_CLIENT_CONN (traceFlags&8)
  212. #define TF_TRACE_TREE_COPY (traceFlags&0x10)
  213. #define TF_TRACE_CLIENT_STATS (traceFlags&0x20)
  214. enum {
  215. RFCopenIO, // 0
  216. RFCcloseIO,
  217. RFCread,
  218. RFCwrite,
  219. RFCsize,
  220. RFCexists,
  221. RFCremove,
  222. RFCrename,
  223. RFCgetver,
  224. RFCisfile,
  225. RFCisdirectory, // 10
  226. RFCisreadonly,
  227. RFCsetreadonly,
  228. RFCgettime,
  229. RFCsettime,
  230. RFCcreatedir,
  231. RFCgetdir,
  232. RFCstop,
  233. RFCexec,
  234. RFCkill,
  235. RFCredeploy, // 20
  236. RFCgetcrc,
  237. RFCmove,
  238. // 1.5 features below
  239. RFCsetsize,
  240. RFCextractblobelements,
  241. RFCcopy,
  242. RFCappend,
  243. RFCmonitordir,
  244. RFCsettrace,
  245. RFCgetinfo,
  246. RFCfirewall, // not used currently // 30
  247. RFCunlock,
  248. RFCunlockreply,
  249. RFCinvalid,
  250. RFCcopysection,
  251. // 1.7e
  252. RFCtreecopy,
  253. // 1.7e - 1
  254. RFCtreecopytmp,
  255. // 1.8
  256. RFCsetthrottle, // legacy version
  257. // 1.9
  258. RFCsetthrottle2,
  259. RFCsetfileperms,
  260. RFCmax,
  261. RFCunknown = 255 // 0 would have been more sensible, but can't break backward compatibility
  262. };
  263. #define RFCText(cmd) #cmd
  264. const char *RFCStrings[] =
  265. {
  266. RFCText(RFCopenIO),
  267. RFCText(RFCcloseIO),
  268. RFCText(RFCread),
  269. RFCText(RFCwrite),
  270. RFCText(RFCsize),
  271. RFCText(RFCexists),
  272. RFCText(RFCremove),
  273. RFCText(RFCrename),
  274. RFCText(RFCgetver),
  275. RFCText(RFCisfile),
  276. RFCText(RFCisdirectory),
  277. RFCText(RFCisreadonly),
  278. RFCText(RFCsetreadonly),
  279. RFCText(RFCgettime),
  280. RFCText(RFCsettime),
  281. RFCText(RFCcreatedir),
  282. RFCText(RFCgetdir),
  283. RFCText(RFCstop),
  284. RFCText(RFCexec),
  285. RFCText(RFCkill),
  286. RFCText(RFCredeploy),
  287. RFCText(RFCgetcrc),
  288. RFCText(RFCmove),
  289. RFCText(RFCsetsize),
  290. RFCText(RFCextractblobelements),
  291. RFCText(RFCcopy),
  292. RFCText(RFCappend),
  293. RFCText(RFCmonitordir),
  294. RFCText(RFCsettrace),
  295. RFCText(RFCgetinfo),
  296. RFCText(RFCfirewall),
  297. RFCText(RFCunlock),
  298. RFCText(RFCunlockreply),
  299. RFCText(RFCinvalid),
  300. RFCText(RFCcopysection),
  301. RFCText(RFCtreecopy),
  302. RFCText(RFCtreecopytmp),
  303. RFCText(RFCsetthrottle), // legacy version
  304. RFCText(RFCsetthrottle2),
  305. RFCText(RFCsetfileperms),
  306. RFCText(RFCunknown),
  307. };
  308. static const char *getRFCText(RemoteFileCommandType cmd)
  309. {
  310. if (cmd > RFCmax)
  311. cmd = RFCmax;
  312. return RFCStrings[cmd];
  313. }
  314. #define ThrottleText(throttleClass) #throttleClass
  315. const char *ThrottleStrings[] =
  316. {
  317. ThrottleText(ThrottleStd),
  318. ThrottleText(ThrottleSlow),
  319. };
  320. // very high upper limits that configure can't exceed
  321. #define THROTTLE_MAX_LIMIT 1000000
  322. #define THROTTLE_MAX_DELAYMS 3600000
  323. #define THROTTLE_MAX_CPUTHRESHOLD 100
  324. #define THROTTLE_MAX_QUEUELIMIT 10000000
  325. static const char *getThrottleClassText(ThrottleClass throttleClass) { return ThrottleStrings[throttleClass]; }
  326. typedef enum { ACScontinue, ACSdone, ACSerror} AsyncCommandStatus;
  327. typedef byte OnceKey[16];
  328. static void genOnce(OnceKey &key)
  329. {
  330. static __int64 inc=0;
  331. *(unsigned *)&key[0] = getRandom();
  332. *(__int64 *)&key[4] = ++inc;
  333. *(unsigned *)&key[12] = getRandom();
  334. }
  335. static void mergeOnce(OnceKey &key,size32_t sz,const void *data)
  336. {
  337. assertex(sz<=sizeof(OnceKey));
  338. const byte *p = (const byte *)data;
  339. while (sz)
  340. key[--sz] ^= *(p++);
  341. }
  342. //---------------------------------------------------------------------------
  343. class CDafsException: public CInterface, public IDAFS_Exception
  344. {
  345. StringAttr msg;
  346. int errcode;
  347. public:
  348. IMPLEMENT_IINTERFACE;
  349. CDafsException(int code,const char *_msg)
  350. : errcode(code), msg(_msg)
  351. {
  352. };
  353. int errorCode() const
  354. {
  355. return errcode;
  356. }
  357. StringBuffer & errorMessage(StringBuffer &str) const
  358. {
  359. return str.append(msg);
  360. }
  361. MessageAudience errorAudience() const
  362. {
  363. return MSGAUD_user;
  364. }
  365. };
  366. static IDAFS_Exception *createDafsException(int code,const char *msg)
  367. {
  368. return new CDafsException(code,msg);
  369. }
  370. void setDafsEndpointPort(SocketEndpoint &ep)
  371. {
  372. // odd kludge (don't do this at home)
  373. byte ipb[4];
  374. if (ep.getNetAddress(sizeof(ipb),&ipb)==sizeof(ipb)) {
  375. if ((ipb[0]==255)&&(ipb[1]==255)) {
  376. ep.port = (((unsigned)ipb[2])<<8)+ipb[3];
  377. ep.ipset(queryLocalIP());
  378. }
  379. }
  380. if (ep.port==0)
  381. {
  382. ep.port = securitySettings.daFileSrvPort;
  383. }
  384. }
  385. inline MemoryBuffer & initSendBuffer(MemoryBuffer & buff)
  386. {
  387. buff.setEndian(__BIG_ENDIAN); // transfer as big endian...
  388. buff.append((unsigned)0); // reserve space for length prefix
  389. return buff;
  390. }
  391. inline void sendBuffer(ISocket * socket, MemoryBuffer & src)
  392. {
  393. unsigned length = src.length() - sizeof(unsigned);
  394. byte * buffer = (byte *)src.toByteArray();
  395. if (TF_TRACE_FULL)
  396. PROGLOG("sendBuffer size %d, data = %d %d %d %d",length, (int)buffer[4],(int)buffer[5],(int)buffer[6],(int)buffer[7]);
  397. _WINCPYREV(buffer, &length, sizeof(unsigned));
  398. SOCKWRITE(socket)(buffer, src.length());
  399. }
  400. inline size32_t receiveBufferSize(ISocket * socket, unsigned numtries=NORMAL_RETRIES,CTimeMon *timemon=NULL)
  401. {
  402. unsigned timeout = SERVER_TIMEOUT;
  403. if (numtries==0) {
  404. numtries = 1;
  405. timeout = 10*1000; // 10s
  406. }
  407. while (numtries--) {
  408. try {
  409. if (timemon) {
  410. unsigned remaining;
  411. if (timemon->timedout(&remaining)||(remaining<10))
  412. remaining = 10;
  413. if (remaining<timeout)
  414. timeout = remaining;
  415. }
  416. size32_t szread;
  417. size32_t gotLength;
  418. SOCKREADTMS(socket)(&gotLength, sizeof(gotLength), sizeof(gotLength), szread, timeout);
  419. _WINREV(gotLength);
  420. if (TF_TRACE_FULL)
  421. PROGLOG("receiveBufferSized %d",gotLength);
  422. return gotLength;
  423. }
  424. catch (IJSOCK_Exception *e) {
  425. if ((numtries==0)||(e->errorCode()!=JSOCKERR_timeout_expired)||(timemon&&timemon->timedout())) {
  426. throw;
  427. }
  428. StringBuffer err;
  429. char peername[256];
  430. socket->peer_name(peername,sizeof(peername)-1);
  431. WARNLOG("Remote connection %s: %s",peername,e->errorMessage(err).str()); // why no peername
  432. e->Release();
  433. Sleep(500+getRandom()%1000); // ~1s
  434. }
  435. }
  436. return 0;
  437. }
  438. static void flush(ISocket *socket)
  439. {
  440. MemoryBuffer sendbuf;
  441. initSendBuffer(sendbuf);
  442. sendbuf.append((RemoteFileCommandType)RFCgetver);
  443. sendbuf.append((unsigned)RFCgetver);
  444. MemoryBuffer reply;
  445. size32_t totread=0;
  446. try {
  447. sendBuffer(socket, sendbuf);
  448. char buf[1024];
  449. loop {
  450. Sleep(1000); // breathe
  451. size32_t szread;
  452. SOCKREADTMS(socket)(buf, 1, sizeof(buf), szread, 1000*60);
  453. totread += szread;
  454. }
  455. }
  456. catch (IJSOCK_Exception *e) {
  457. if (totread)
  458. PROGLOG("%d bytes discarded",totread);
  459. if (e->errorCode()!=JSOCKERR_timeout_expired)
  460. EXCLOG(e,"flush");
  461. e->Release();
  462. }
  463. }
  464. inline void receiveBuffer(ISocket * socket, MemoryBuffer & tgt, unsigned numtries=1, size32_t maxsz=0x7fffffff)
  465. // maxsz is a guess at a resonable upper max to catch where protocol error
  466. {
  467. sRFTM tm;
  468. size32_t gotLength = receiveBufferSize(socket, numtries,tm.timemon);
  469. if (gotLength) {
  470. size32_t origlen = tgt.length();
  471. try {
  472. if (gotLength>maxsz) {
  473. StringBuffer msg;
  474. msg.appendf("receiveBuffer maximum block size exceeded %d/%d",gotLength,maxsz);
  475. PrintStackReport();
  476. throw createDafsException(DAFSERR_protocol_failure,msg.str());
  477. }
  478. unsigned timeout = SERVER_TIMEOUT*(numtries?numtries:1);
  479. if (tm.timemon) {
  480. unsigned remaining;
  481. if (tm.timemon->timedout(&remaining)||(remaining<10))
  482. remaining = 10;
  483. if (remaining<timeout)
  484. timeout = remaining;
  485. }
  486. size32_t szread;
  487. SOCKREADTMS(socket)((gotLength<4000)?tgt.reserve(gotLength):tgt.reserveTruncate(gotLength), gotLength, gotLength, szread, timeout);
  488. }
  489. catch (IJSOCK_Exception *e) {
  490. if (e->errorCode()!=JSOCKERR_timeout_expired) {
  491. EXCLOG(e,"receiveBuffer(1)");
  492. PrintStackReport();
  493. if (!tm.timemon||!tm.timemon->timedout())
  494. flush(socket);
  495. }
  496. else {
  497. EXCLOG(e,"receiveBuffer");
  498. PrintStackReport();
  499. }
  500. tgt.setLength(origlen);
  501. throw;
  502. }
  503. catch (IException *e) {
  504. EXCLOG(e,"receiveBuffer(2)");
  505. PrintStackReport();
  506. if (!tm.timemon||!tm.timemon->timedout())
  507. flush(socket);
  508. tgt.setLength(origlen);
  509. throw;
  510. }
  511. }
  512. tgt.setEndian(__BIG_ENDIAN);
  513. }
  514. struct CConnectionRec
  515. {
  516. SocketEndpoint ep;
  517. unsigned tick;
  518. IArrayOf<ISocket> socks; // relies on isShared
  519. };
  520. //---------------------------------------------------------------------------
  521. // Local mount redirect
  522. struct CLocalMountRec: public CInterface
  523. {
  524. IpAddress ip;
  525. StringAttr dir; // dir path on remote ip
  526. StringAttr local; // local dir path
  527. };
  528. static CIArrayOf<CLocalMountRec> localMounts;
  529. static CriticalSection localMountCrit;
  530. void setDafsLocalMountRedirect(const IpAddress &ip,const char *dir,const char *mountdir)
  531. {
  532. CriticalBlock block(localMountCrit);
  533. ForEachItemInRev(i,localMounts) {
  534. CLocalMountRec &mount = localMounts.item(i);
  535. if (dir==NULL) { // remove all matching mount
  536. if (!mountdir)
  537. return;
  538. if (strcmp(mount.local,mountdir)==0)
  539. localMounts.remove(i);
  540. }
  541. else if (mount.ip.ipequals(ip)&&(strcmp(mount.dir,dir)==0)) {
  542. if (mountdir) {
  543. mount.local.set(mountdir);
  544. return;
  545. }
  546. else
  547. localMounts.remove(i);
  548. }
  549. }
  550. if (dir&&mountdir) {
  551. CLocalMountRec &mount = *new CLocalMountRec;
  552. mount.ip.ipset(ip);
  553. mount.dir.set(dir);
  554. mount.local.set(mountdir);
  555. localMounts.append(mount);
  556. }
  557. }
  558. IFile *createFileLocalMount(const IpAddress &ip, const char * filename)
  559. {
  560. CriticalBlock block(localMountCrit);
  561. ForEachItemInRev(i,localMounts) {
  562. CLocalMountRec &mount = localMounts.item(i);
  563. if (mount.ip.ipequals(ip)) {
  564. size32_t bl = mount.dir.length();
  565. if (isPathSepChar(mount.dir[bl-1]))
  566. bl--;
  567. if ((memcmp((void *)filename,(void *)mount.dir.get(),bl)==0)&&(isPathSepChar(filename[bl])||!filename[bl])) { // match
  568. StringBuffer locpath(mount.local);
  569. if (filename[bl])
  570. addPathSepChar(locpath).append(filename+bl+1);
  571. locpath.replace((PATHSEPCHAR=='\\')?'/':'\\',PATHSEPCHAR);
  572. return createIFile(locpath.str());
  573. }
  574. }
  575. }
  576. return NULL;
  577. }
  578. //---------------------------------------------------------------------------
  579. static class CConnectionTable: public SuperHashTableOf<CConnectionRec,SocketEndpoint>
  580. {
  581. void onAdd(void *) {}
  582. void onRemove(void *e)
  583. {
  584. CConnectionRec *r=(CConnectionRec *)e;
  585. delete r;
  586. }
  587. unsigned getHashFromElement(const void *e) const
  588. {
  589. const CConnectionRec &elem=*(const CConnectionRec *)e;
  590. return elem.ep.hash(0);
  591. }
  592. unsigned getHashFromFindParam(const void *fp) const
  593. {
  594. return ((const SocketEndpoint *)fp)->hash(0);
  595. }
  596. const void * getFindParam(const void *p) const
  597. {
  598. const CConnectionRec &elem=*(const CConnectionRec *)p;
  599. return (void *)&elem.ep;
  600. }
  601. bool matchesFindParam(const void * et, const void *fp, unsigned) const
  602. {
  603. return ((CConnectionRec *)et)->ep.equals(*(SocketEndpoint *)fp);
  604. }
  605. IMPLEMENT_SUPERHASHTABLEOF_REF_FIND(CConnectionRec,SocketEndpoint);
  606. unsigned numsockets;
  607. public:
  608. static CriticalSection crit;
  609. CConnectionTable()
  610. {
  611. numsockets = 0;
  612. }
  613. ~CConnectionTable() {
  614. releaseAll();
  615. }
  616. ISocket *lookup(const SocketEndpoint &ep)
  617. {
  618. // always called from crit block
  619. CConnectionRec *r = SuperHashTableOf<CConnectionRec,SocketEndpoint>::find(&ep);
  620. if (r) {
  621. ForEachItemIn(i,r->socks) {
  622. ISocket *s = &r->socks.item(i);
  623. if (!QUERYINTERFACE(s, CInterface)->IsShared()) {
  624. r->tick = msTick();
  625. s->Link();
  626. return s;
  627. }
  628. }
  629. }
  630. return NULL;
  631. }
  632. void addLink(SocketEndpoint &ep,ISocket *sock)
  633. {
  634. // always called from crit block
  635. while (numsockets>=SOCKET_CACHE_MAX) {
  636. // find oldest
  637. CConnectionRec *c = NULL;
  638. unsigned oldest = 0;
  639. CConnectionRec *old = NULL;
  640. unsigned oldi;
  641. unsigned now = msTick();
  642. loop {
  643. c = (CConnectionRec *)SuperHashTableOf<CConnectionRec,SocketEndpoint>::next(c);
  644. if (!c)
  645. break;
  646. ForEachItemIn(i,c->socks) {
  647. ISocket *s = &c->socks.item(i);
  648. if (!QUERYINTERFACE(s, CInterface)->IsShared()) { // candidate to remove
  649. unsigned t = now-c->tick;
  650. if (t>oldest) {
  651. oldest = t;
  652. old = c;
  653. oldi = i;
  654. }
  655. }
  656. }
  657. }
  658. if (!old)
  659. return;
  660. old->socks.remove(oldi);
  661. numsockets--;
  662. }
  663. CConnectionRec *r = SuperHashTableOf<CConnectionRec,SocketEndpoint>::find(&ep);
  664. if (!r) {
  665. r = new CConnectionRec;
  666. r->ep = ep;
  667. SuperHashTableOf<CConnectionRec,SocketEndpoint>::add(*r);
  668. }
  669. sock->Link();
  670. r->socks.append(*sock);
  671. numsockets++;
  672. r->tick = msTick();
  673. }
  674. void remove(SocketEndpoint &ep,ISocket *sock)
  675. {
  676. // always called from crit block
  677. CConnectionRec *r = SuperHashTableOf<CConnectionRec,SocketEndpoint>::find(&ep);
  678. if (r)
  679. if (r->socks.zap(*sock)&&numsockets)
  680. numsockets--;
  681. }
  682. } *ConnectionTable = NULL;
  683. CriticalSection CConnectionTable::crit;
  684. void clientSetDaliServixSocketCaching(bool on)
  685. {
  686. CriticalBlock block(CConnectionTable::crit);
  687. if (on) {
  688. if (!ConnectionTable)
  689. ConnectionTable = new CConnectionTable;
  690. }
  691. else {
  692. delete ConnectionTable;
  693. ConnectionTable = NULL;
  694. }
  695. }
  696. //---------------------------------------------------------------------------
  697. // TreeCopy
  698. #define TREECOPY_CACHE_SIZE 50
  699. struct CTreeCopyItem: public CInterface
  700. {
  701. StringAttr net;
  702. StringAttr mask;
  703. offset_t sz; // original size
  704. CDateTime dt; // original date
  705. RemoteFilenameArray loc; // locations for file - 0 is original
  706. Owned<IBitSet> busy;
  707. unsigned lastused;
  708. CTreeCopyItem(RemoteFilename &orig, const char *_net, const char *_mask, offset_t _sz, CDateTime &_dt)
  709. : net(_net), mask(_mask)
  710. {
  711. loc.append(orig);
  712. dt.set(_dt);
  713. sz = _sz;
  714. busy.setown(createThreadSafeBitSet());
  715. lastused = msTick();
  716. }
  717. bool equals(const RemoteFilename &orig, const char *_net, const char *_mask, offset_t _sz, CDateTime &_dt)
  718. {
  719. if (!orig.equals(loc.item(0)))
  720. return false;
  721. if (strcmp(_net,net)!=0)
  722. return false;
  723. if (strcmp(_mask,mask)!=0)
  724. return false;
  725. if (sz!=_sz)
  726. return false;
  727. return (dt.equals(_dt,false));
  728. }
  729. };
  730. static CIArrayOf<CTreeCopyItem> treeCopyArray;
  731. static CriticalSection treeCopyCrit;
  732. static unsigned treeCopyWaiting=0;
  733. static Semaphore treeCopySem;
  734. #define DEBUGSAMEIP false
  735. //---------------------------------------------------------------------------
  736. class CRemoteBase: public CInterface
  737. {
  738. Owned<ISocket> socket;
  739. static SocketEndpoint lastfailep;
  740. static unsigned lastfailtime;
  741. bool useSSL;
  742. void connectSocket(SocketEndpoint &ep)
  743. {
  744. sRFTM tm;
  745. // called in CConnectionTable::crit
  746. unsigned retries = 3;
  747. if (ep.equals(lastfailep)) {
  748. if (msTick()-lastfailtime<DAFS_CONNECT_FAIL_RETRY_TIME) {
  749. StringBuffer msg("Failed to connect to dafilesrv/daliservix on ");
  750. ep.getUrlStr(msg);
  751. throw createDafsException(DAFSERR_connection_failed,msg.str());
  752. }
  753. lastfailep.set(NULL);
  754. retries = 1; // on probation
  755. }
  756. while(retries--) {
  757. CriticalUnblock unblock(CConnectionTable::crit); // allow others to connect
  758. StringBuffer eps;
  759. if (TF_TRACE_CLIENT_CONN) {
  760. ep.getUrlStr(eps);
  761. PROGLOG("Connecting %sto %s",useSSL?"SECURE ":"",eps.str());
  762. //PrintStackReport();
  763. }
  764. bool ok = true;
  765. try {
  766. if (tm.timemon) {
  767. unsigned remaining;
  768. tm.timemon->timedout(&remaining);
  769. socket.setown(ISocket::connect_timeout(ep,remaining));
  770. }
  771. else
  772. socket.setown(ISocket::connect(ep));
  773. if (useSSL)
  774. {
  775. Owned<ISecureSocket> ssock = createSecureSocket(socket.getClear(), ClientSocket);
  776. int status = ssock->secure_connect();
  777. if (status < 0)
  778. throw createDafsException(DAFSERR_connection_failed,"Failure to establish secure connection");
  779. socket.setown(ssock.getLink());
  780. }
  781. }
  782. catch (IJSOCK_Exception *e) {
  783. ok = false;
  784. if (!retries||(tm.timemon&&tm.timemon->timedout())) {
  785. if (e->errorCode()==JSOCKERR_connection_failed) {
  786. lastfailep.set(ep);
  787. lastfailtime = msTick();
  788. e->Release();
  789. StringBuffer msg("Failed to connect to dafilesrv/daliservix on ");
  790. ep.getUrlStr(msg);
  791. throw createDafsException(DAFSERR_connection_failed,msg.str());
  792. }
  793. throw;
  794. }
  795. StringBuffer err;
  796. WARNLOG("Remote file connect %s",e->errorMessage(err).str());
  797. e->Release();
  798. }
  799. if (ok) {
  800. if (TF_TRACE_CLIENT_CONN) {
  801. PROGLOG("Connected to %s",eps.str());
  802. }
  803. if (AuthenticationEnabled) {
  804. try {
  805. sendAuthentication(ep); // this will log error
  806. break;
  807. }
  808. catch (IJSOCK_Exception *e) {
  809. StringBuffer err;
  810. WARNLOG("Remote file authenticate %s for %s ",e->errorMessage(err).str(),ep.getUrlStr(eps.clear()).str());
  811. e->Release();
  812. if (!retries)
  813. break;
  814. }
  815. }
  816. else
  817. break;
  818. }
  819. unsigned sleeptime = getRandom()%3000+1000;
  820. if (tm.timemon) {
  821. unsigned remaining;
  822. tm.timemon->timedout(&remaining);
  823. if (remaining/2<sleeptime)
  824. sleeptime = remaining/2;
  825. }
  826. Sleep(sleeptime); // prevent multiple retries beating
  827. PROGLOG("Retrying %sconnect",useSSL?"SECURE ":"");
  828. }
  829. if (ConnectionTable)
  830. ConnectionTable->addLink(ep,socket);
  831. }
  832. void killSocket(SocketEndpoint &tep)
  833. {
  834. CriticalBlock block2(CConnectionTable::crit); // this is nested with crit
  835. if (socket) {
  836. try {
  837. Owned<ISocket> s = socket.getClear();
  838. if (ConnectionTable)
  839. ConnectionTable->remove(tep,s);
  840. }
  841. catch (IJSOCK_Exception *e) {
  842. e->Release(); // ignore errors closing
  843. }
  844. Sleep(getRandom()%1000*5+500); // prevent multiple beating
  845. }
  846. }
  847. protected: friend class CRemoteFileIO;
  848. StringAttr filename;
  849. CriticalSection crit;
  850. SocketEndpoint ep;
  851. void sendRemoteCommand(MemoryBuffer & src, MemoryBuffer & reply, bool retry=true, bool lengthy=false)
  852. {
  853. CriticalBlock block(crit); // serialize commands on same file
  854. SocketEndpoint tep(ep);
  855. setDafsEndpointPort(tep);
  856. unsigned nretries = retry?3:0;
  857. Owned<IJSOCK_Exception> firstexc; // when retrying return first error if fails
  858. loop {
  859. try {
  860. if (socket) {
  861. sendBuffer(socket, src);
  862. receiveBuffer(socket, reply, lengthy?LENGTHY_RETRIES:NORMAL_RETRIES);
  863. break;
  864. }
  865. }
  866. catch (IJSOCK_Exception *e) {
  867. if (!nretries--) {
  868. if (firstexc) {
  869. e->Release();
  870. e = firstexc.getClear();
  871. }
  872. killSocket(tep);
  873. throw e;
  874. }
  875. StringBuffer str;
  876. e->errorMessage(str);
  877. WARNLOG("Remote File: %s, retrying (%d)",str.str(),nretries);
  878. if (firstexc)
  879. e->Release();
  880. else
  881. firstexc.setown(e);
  882. killSocket(tep);
  883. }
  884. CriticalBlock block2(CConnectionTable::crit); // this is nested with crit
  885. if (ConnectionTable) {
  886. socket.setown(ConnectionTable->lookup(tep));
  887. if (socket) {
  888. // validate existing socket by sending an 'exists' command with short time out
  889. // (use exists for backward compatibility)
  890. bool ok = false;
  891. try {
  892. MemoryBuffer sendbuf;
  893. initSendBuffer(sendbuf);
  894. MemoryBuffer replybuf;
  895. sendbuf.append((RemoteFileCommandType)RFCexists).append(filename);
  896. sendBuffer(socket, sendbuf);
  897. receiveBuffer(socket, replybuf, 0, 1024);
  898. ok = true;
  899. }
  900. catch (IException *e) {
  901. e->Release();
  902. }
  903. if (!ok)
  904. killSocket(tep);
  905. }
  906. }
  907. if (!socket) {
  908. bool tryNonSecure = true;
  909. if (useSSL) {
  910. try {
  911. connectSocket(tep);//first try secure connect
  912. tryNonSecure = false;
  913. }
  914. catch (IException *e) {
  915. StringBuffer s;
  916. e->errorMessage(s);
  917. e->Release();
  918. WARNLOG("Secure connect failed, retrying on legacy port (%s)",s.str());
  919. useSSL = false;
  920. tep.port = DAFILESRV_PORT;//retry on nonsecure port
  921. tryNonSecure = true;
  922. }
  923. }
  924. if (tryNonSecure)
  925. connectSocket(tep);
  926. }
  927. }
  928. unsigned errCode;
  929. reply.read(errCode);
  930. if (errCode) {
  931. StringBuffer msg;
  932. if (filename.get())
  933. msg.append(filename);
  934. ep.getUrlStr(msg.append('[')).append("] ");
  935. size32_t pos = reply.getPos();
  936. if (pos<reply.length()) {
  937. size32_t len = reply.length()-pos;
  938. const byte *rest = reply.readDirect(len);
  939. if (errCode==RFSERR_InvalidCommand) {
  940. const char *s = (const char *)rest;
  941. const char *e = (const char *)rest+len;
  942. while (*s&&(s!=e))
  943. s++;
  944. msg.append(s-(const char *)rest,(const char *)rest);
  945. }
  946. else if (len&&(rest[len-1]==0))
  947. msg.append((const char *)rest);
  948. else {
  949. msg.appendf("extra data[%d]",len);
  950. for (unsigned i=0;(i<16)&&(i<len);i++)
  951. msg.appendf(" %2x",(int)rest[i]);
  952. }
  953. }
  954. else if(errCode == 8209)
  955. msg.append("Failed to open directory.");
  956. else
  957. msg.append("ERROR #").append(errCode);
  958. #ifdef _DEBUG
  959. ERRLOG("%s",msg.str());
  960. PrintStackReport();
  961. #endif
  962. throw createDafsException(errCode,msg.str());
  963. }
  964. }
  965. void sendRemoteCommand(MemoryBuffer & src, bool retry)
  966. {
  967. MemoryBuffer reply;
  968. sendRemoteCommand(src, reply, retry);
  969. }
  970. void throwUnauthenticated(const IpAddress &ip,const char *user,unsigned err=0)
  971. {
  972. if (err==0)
  973. err = RFSERR_AuthenticateFailed;
  974. StringBuffer msg;
  975. msg.appendf("Authentication for %s on ",user);
  976. ip.getIpText(msg);
  977. msg.append(" failed");
  978. throw createDafsException(err, msg.str());
  979. }
  980. void sendAuthentication(const IpAddress &serverip)
  981. {
  982. // send my sig
  983. // first send my sig which if stream unencrypted will get returned as a bad command
  984. OnceKey oncekey;
  985. genOnce(oncekey);
  986. MemoryBuffer sendbuf;
  987. initSendBuffer(sendbuf);
  988. MemoryBuffer replybuf;
  989. MemoryBuffer encbuf; // because aesEncrypt clears input
  990. sendbuf.append((RemoteFileCommandType)RFCunlock).append(sizeof(oncekey),&oncekey);
  991. try {
  992. sendBuffer(socket, sendbuf);
  993. receiveBuffer(socket, replybuf, NORMAL_RETRIES, 1024);
  994. }
  995. catch (IException *e)
  996. {
  997. EXCLOG(e,"Remote file - sendAuthentication(1)");
  998. throw;
  999. }
  1000. unsigned errCode;
  1001. replybuf.read(errCode);
  1002. if (errCode!=0) // no authentication required
  1003. return;
  1004. SocketEndpoint ep;
  1005. ep.setLocalHost(0);
  1006. byte ipdata[16];
  1007. size32_t ipds = ep.getNetAddress(sizeof(ipdata),&ipdata);
  1008. mergeOnce(oncekey,ipds,&ipdata);
  1009. StringBuffer username;
  1010. StringBuffer password;
  1011. IPasswordProvider * pp = queryPasswordProvider();
  1012. if (pp)
  1013. pp->getPassword(serverip, username, password);
  1014. if (!username.length())
  1015. username.append("sds_system"); // default account (note if exists should have restricted access!)
  1016. if (!password.length())
  1017. password.append("sds_man");
  1018. if (replybuf.remaining()<=sizeof(size32_t))
  1019. throwUnauthenticated(serverip,username.str());
  1020. size32_t bs;
  1021. replybuf.read(bs);
  1022. if (replybuf.remaining()<bs)
  1023. throwUnauthenticated(serverip,username.str());
  1024. MemoryBuffer skeybuf;
  1025. aesDecrypt(&oncekey,sizeof(oncekey),replybuf.readDirect(bs),bs,skeybuf);
  1026. if (skeybuf.remaining()<sizeof(OnceKey))
  1027. throwUnauthenticated(serverip,username.str());
  1028. OnceKey sokey;
  1029. skeybuf.read(sizeof(OnceKey),&sokey);
  1030. // now we have the key to use to send user/password
  1031. MemoryBuffer tosend;
  1032. tosend.append((byte)2).append(username).append(password);
  1033. initSendBuffer(sendbuf.clear());
  1034. sendbuf.append((RemoteFileCommandType)RFCunlockreply);
  1035. aesEncrypt(&sokey, sizeof(oncekey), tosend.toByteArray(), tosend.length(), encbuf);
  1036. sendbuf.append(encbuf.length());
  1037. sendbuf.append(encbuf);
  1038. try {
  1039. sendBuffer(socket, sendbuf);
  1040. receiveBuffer(socket, replybuf.clear(), NORMAL_RETRIES, 1024);
  1041. }
  1042. catch (IException *e)
  1043. {
  1044. EXCLOG(e,"Remote file - sendAuthentication(2)");
  1045. throw;
  1046. }
  1047. replybuf.read(errCode);
  1048. if (errCode==0) // suceeded!
  1049. return;
  1050. throwUnauthenticated(serverip,username.str(),errCode);
  1051. }
  1052. public:
  1053. SocketEndpoint &queryEp() { return ep; };
  1054. CRemoteBase(const SocketEndpoint &_ep, const char * _filename)
  1055. : filename(_filename)
  1056. {
  1057. ep = _ep;
  1058. useSSL = securitySettings.useSSL;
  1059. }
  1060. void connect()
  1061. {
  1062. CriticalBlock block(crit);
  1063. CriticalBlock block2(CConnectionTable::crit); // this shouldn't ever block
  1064. if (AuthenticationEnabled) {
  1065. SocketEndpoint tep(ep);
  1066. setDafsEndpointPort(tep);
  1067. connectSocket(tep);
  1068. }
  1069. }
  1070. void disconnect()
  1071. {
  1072. CriticalBlock block(crit);
  1073. CriticalBlock block2(CConnectionTable::crit); // this shouldn't ever block
  1074. if (socket)
  1075. {
  1076. ISocket *s = socket.getClear();
  1077. if (ConnectionTable)
  1078. {
  1079. SocketEndpoint tep(ep);
  1080. setDafsEndpointPort(tep);
  1081. ConnectionTable->remove(tep,s);
  1082. }
  1083. ::Release(s);
  1084. }
  1085. }
  1086. const char *queryLocalName()
  1087. {
  1088. return filename;
  1089. }
  1090. };
  1091. SocketEndpoint CRemoteBase::lastfailep;
  1092. unsigned CRemoteBase::lastfailtime;
  1093. //---------------------------------------------------------------------------
  1094. class CRemoteDirectoryIterator : public CInterface, implements IDirectoryDifferenceIterator
  1095. {
  1096. Owned<IFile> cur;
  1097. bool curvalid;
  1098. bool curisdir;
  1099. StringAttr curname;
  1100. CDateTime curdt;
  1101. __int64 cursize;
  1102. StringAttr dir;
  1103. SocketEndpoint ep;
  1104. byte *flags;
  1105. unsigned numflags;
  1106. unsigned curidx;
  1107. unsigned mask;
  1108. MemoryBuffer buf;
  1109. public:
  1110. static CriticalSection crit;
  1111. CRemoteDirectoryIterator(SocketEndpoint &_ep,const char *_dir)
  1112. : dir(_dir)
  1113. {
  1114. // an extended difference iterator starts with 2 (for bwd compatibility)
  1115. ep = _ep;
  1116. curisdir = false;
  1117. curvalid = false;
  1118. cursize = 0;
  1119. curidx = (unsigned)-1;
  1120. mask = 0;
  1121. numflags = 0;
  1122. flags = NULL;
  1123. }
  1124. bool appendBuf(MemoryBuffer &_buf)
  1125. {
  1126. buf.setSwapEndian(_buf.needSwapEndian());
  1127. byte hdr;
  1128. _buf.read(hdr);
  1129. if (hdr==2) {
  1130. _buf.read(numflags);
  1131. flags = (byte *)malloc(numflags);
  1132. _buf.read(numflags,flags);
  1133. }
  1134. else {
  1135. buf.append(hdr);
  1136. flags = NULL;
  1137. numflags = 0;
  1138. }
  1139. size32_t rest = _buf.length()-_buf.getPos();
  1140. const byte *rb = (const byte *)_buf.readDirect(rest);
  1141. bool ret = true;
  1142. // At the last byte of the rb (rb[rest-1]) is the stream live flag
  1143. // True if the stream has more data
  1144. // False at the end of stream
  1145. // The previous byte (rb[rest-2]) is the flag to signal there are more
  1146. // valid entries in this block
  1147. // True if there are valid directory entry follows this flag
  1148. // False if there are no more valid entry in this block aka end of block
  1149. // If there is more data in the stream, the end of block flag should be removed
  1150. if (rest&&(rb[rest-1]!=0))
  1151. {
  1152. rest--; // remove stream live flag
  1153. if(rest && (0 == rb[rest-1]))
  1154. rest--; //Remove end of block flag
  1155. ret = false; // continuation
  1156. }
  1157. buf.append(rest,rb);
  1158. return ret;
  1159. }
  1160. ~CRemoteDirectoryIterator()
  1161. {
  1162. free(flags);
  1163. }
  1164. IMPLEMENT_IINTERFACE
  1165. bool first()
  1166. {
  1167. curidx = (unsigned)-1;
  1168. buf.reset();
  1169. return next();
  1170. }
  1171. bool next()
  1172. {
  1173. loop {
  1174. curidx++;
  1175. cur.clear();
  1176. curdt.clear();
  1177. curname.clear();
  1178. cursize = 0;
  1179. curisdir = false;
  1180. if (buf.getPos()>=buf.length())
  1181. return false;
  1182. byte isValidEntry;
  1183. buf.read(isValidEntry);
  1184. curvalid = isValidEntry!=0;
  1185. if (!curvalid)
  1186. return false;
  1187. buf.read(curisdir);
  1188. buf.read(cursize);
  1189. curdt.deserialize(buf);
  1190. buf.read(curname);
  1191. // kludge for bug in old linux jlibs
  1192. if (strchr(curname,'\\')&&(getPathSepChar(dir)=='/')) {
  1193. StringBuffer temp(curname);
  1194. temp.replace('\\','/');
  1195. curname.set(temp.str());
  1196. }
  1197. if ((mask==0)||(getFlags()&mask))
  1198. break;
  1199. }
  1200. return true;
  1201. }
  1202. bool isValid()
  1203. {
  1204. return curvalid;
  1205. }
  1206. IFile & query()
  1207. {
  1208. if (!cur) {
  1209. StringBuffer full(dir);
  1210. addPathSepChar(full).append(curname);
  1211. if (ep.isNull())
  1212. cur.setown(createIFile(full.str()));
  1213. else {
  1214. RemoteFilename rfn;
  1215. rfn.setPath(ep,full.str());
  1216. cur.setown(createIFile(rfn));
  1217. }
  1218. }
  1219. return *cur;
  1220. }
  1221. StringBuffer &getName(StringBuffer &buf)
  1222. {
  1223. return buf.append(curname);
  1224. }
  1225. bool isDir()
  1226. {
  1227. return curisdir;
  1228. }
  1229. __int64 getFileSize()
  1230. {
  1231. if (curisdir)
  1232. return -1;
  1233. return cursize;
  1234. }
  1235. bool getModifiedTime(CDateTime &ret)
  1236. {
  1237. ret = curdt;
  1238. return true;
  1239. }
  1240. void setMask(unsigned _mask)
  1241. {
  1242. mask = _mask;
  1243. }
  1244. virtual unsigned getFlags()
  1245. {
  1246. if (flags&&(curidx<numflags))
  1247. return flags[curidx];
  1248. return 0;
  1249. }
  1250. static bool serialize(MemoryBuffer &mb,IDirectoryIterator *iter, size32_t bufsize, bool first)
  1251. {
  1252. bool ret = true;
  1253. byte b=1;
  1254. StringBuffer tmp;
  1255. if (first ? iter->first() : iter->next()) {
  1256. loop {
  1257. mb.append(b);
  1258. bool isdir = iter->isDir();
  1259. __int64 sz = isdir?0:iter->getFileSize();
  1260. CDateTime dt;
  1261. iter->getModifiedTime(dt);
  1262. iter->getName(tmp.clear());
  1263. mb.append(isdir).append(sz);
  1264. dt.serialize(mb);
  1265. mb.append(tmp.str());
  1266. if (bufsize&&(mb.length()>=bufsize-1)) {
  1267. ret = false;
  1268. break;
  1269. }
  1270. if (!iter->next())
  1271. break;
  1272. }
  1273. }
  1274. b = 0;
  1275. mb.append(b);
  1276. return ret;
  1277. }
  1278. static void serializeDiff(MemoryBuffer &mb,IDirectoryDifferenceIterator *iter)
  1279. {
  1280. // bit slow
  1281. MemoryBuffer flags;
  1282. ForEach(*iter)
  1283. flags.append((byte)iter->getFlags());
  1284. if (flags.length()) {
  1285. byte b = 2;
  1286. mb.append(b).append((unsigned)flags.length()).append(flags);
  1287. }
  1288. serialize(mb,iter,0,true);
  1289. }
  1290. void serialize(MemoryBuffer &mb,bool isdiff)
  1291. {
  1292. byte b;
  1293. if (isdiff&&numflags&&flags) {
  1294. b = 2;
  1295. mb.append(b).append(numflags).append(numflags,flags);
  1296. }
  1297. serialize(mb,this,0,true);
  1298. }
  1299. };
  1300. class CCritTable;
  1301. class CEndpointCS : public CriticalSection, public CInterface
  1302. {
  1303. CCritTable &table;
  1304. const SocketEndpoint ep;
  1305. public:
  1306. CEndpointCS(CCritTable &_table, const SocketEndpoint &_ep) : table(_table), ep(_ep) { }
  1307. const void *queryFindParam() const { return &ep; }
  1308. virtual void beforeDispose();
  1309. };
  1310. class CCritTable : private SimpleHashTableOf<CEndpointCS, const SocketEndpoint>
  1311. {
  1312. typedef SimpleHashTableOf<CEndpointCS, const SocketEndpoint> PARENT;
  1313. CriticalSection crit;
  1314. public:
  1315. CEndpointCS *getCrit(const SocketEndpoint &ep)
  1316. {
  1317. CriticalBlock b(crit);
  1318. Linked<CEndpointCS> clientCrit = find(ep);
  1319. if (!clientCrit || !clientCrit->isAlive()) // if !isAlive(), then it is in the process of being destroyed/removed.
  1320. {
  1321. clientCrit.setown(new CEndpointCS(*this, ep));
  1322. replace(*clientCrit); // NB table doesn't own
  1323. }
  1324. return clientCrit.getClear();
  1325. }
  1326. unsigned getHashFromElement(const void *e) const
  1327. {
  1328. const CEndpointCS &elem=*(const CEndpointCS *)e;
  1329. return getHashFromFindParam(elem.queryFindParam());
  1330. }
  1331. unsigned getHashFromFindParam(const void *fp) const
  1332. {
  1333. return ((const SocketEndpoint *)fp)->hash(0);
  1334. }
  1335. void removeExact(CEndpointCS *clientCrit)
  1336. {
  1337. CriticalBlock b(crit);
  1338. PARENT::removeExact(clientCrit); // NB may not exist, could have been replaced if detected !isAlive() in getCrit()
  1339. }
  1340. } *dirCSTable;
  1341. MODULE_INIT(INIT_PRIORITY_STANDARD)
  1342. {
  1343. dirCSTable = new CCritTable;
  1344. return true;
  1345. }
  1346. MODULE_EXIT()
  1347. {
  1348. delete dirCSTable;
  1349. }
  1350. void CEndpointCS::beforeDispose()
  1351. {
  1352. table.removeExact(this);
  1353. }
  1354. class CRemoteFile : public CRemoteBase, implements IFile
  1355. {
  1356. StringAttr remotefilename;
  1357. unsigned flags;
  1358. public:
  1359. IMPLEMENT_IINTERFACE
  1360. CRemoteFile(const SocketEndpoint &_ep, const char * _filename)
  1361. : CRemoteBase(_ep, _filename)
  1362. {
  1363. flags = ((unsigned)IFSHread)|((S_IRUSR|S_IWUSR)<<16);
  1364. }
  1365. bool exists()
  1366. {
  1367. MemoryBuffer sendBuffer;
  1368. initSendBuffer(sendBuffer);
  1369. MemoryBuffer replyBuffer;
  1370. sendBuffer.append((RemoteFileCommandType)RFCexists).append(filename);
  1371. sendRemoteCommand(sendBuffer, replyBuffer);
  1372. bool ok;
  1373. replyBuffer.read(ok);
  1374. return ok;
  1375. }
  1376. bool getTime(CDateTime * createTime, CDateTime * modifiedTime, CDateTime * accessedTime)
  1377. {
  1378. CDateTime dummyTime;
  1379. if (!createTime)
  1380. createTime = &dummyTime;
  1381. if (!modifiedTime)
  1382. modifiedTime = &dummyTime;
  1383. if (!accessedTime)
  1384. accessedTime = &dummyTime;
  1385. MemoryBuffer sendBuffer;
  1386. initSendBuffer(sendBuffer);
  1387. MemoryBuffer replyBuffer;
  1388. sendBuffer.append((RemoteFileCommandType)RFCgettime).append(filename);
  1389. sendRemoteCommand(sendBuffer, replyBuffer);
  1390. bool ok;
  1391. replyBuffer.read(ok);
  1392. if (ok) {
  1393. createTime->deserialize(replyBuffer);
  1394. modifiedTime->deserialize(replyBuffer);
  1395. accessedTime->deserialize(replyBuffer);
  1396. }
  1397. return ok;
  1398. }
  1399. bool setTime(const CDateTime * createTime, const CDateTime * modifiedTime, const CDateTime * accessedTime)
  1400. {
  1401. MemoryBuffer sendBuffer;
  1402. initSendBuffer(sendBuffer);
  1403. MemoryBuffer replyBuffer;
  1404. sendBuffer.append((RemoteFileCommandType)RFCsettime).append(filename);
  1405. if (createTime) {
  1406. sendBuffer.append((bool)true);
  1407. createTime->serialize(sendBuffer);
  1408. }
  1409. else
  1410. sendBuffer.append((bool)false);
  1411. if (modifiedTime) {
  1412. sendBuffer.append((bool)true);
  1413. modifiedTime->serialize(sendBuffer);
  1414. }
  1415. else
  1416. sendBuffer.append((bool)false);
  1417. if (accessedTime) {
  1418. sendBuffer.append((bool)true);
  1419. accessedTime->serialize(sendBuffer);
  1420. }
  1421. else
  1422. sendBuffer.append((bool)false);
  1423. sendRemoteCommand(sendBuffer, replyBuffer);
  1424. bool ok;
  1425. replyBuffer.read(ok);
  1426. return ok;
  1427. }
  1428. fileBool isDirectory()
  1429. {
  1430. MemoryBuffer sendBuffer;
  1431. initSendBuffer(sendBuffer);
  1432. MemoryBuffer replyBuffer;
  1433. sendBuffer.append((RemoteFileCommandType)RFCisdirectory).append(filename);
  1434. sendRemoteCommand(sendBuffer, replyBuffer);
  1435. unsigned ret;
  1436. replyBuffer.read(ret);
  1437. return (fileBool)ret;
  1438. }
  1439. fileBool isFile()
  1440. {
  1441. MemoryBuffer sendBuffer;
  1442. initSendBuffer(sendBuffer);
  1443. MemoryBuffer replyBuffer;
  1444. sendBuffer.append((RemoteFileCommandType)RFCisfile).append(filename);
  1445. sendRemoteCommand(sendBuffer, replyBuffer);
  1446. unsigned ret;
  1447. replyBuffer.read(ret);
  1448. return (fileBool)ret;
  1449. }
  1450. fileBool isReadOnly()
  1451. {
  1452. MemoryBuffer sendBuffer;
  1453. initSendBuffer(sendBuffer);
  1454. MemoryBuffer replyBuffer;
  1455. sendBuffer.append((RemoteFileCommandType)RFCisreadonly).append(filename);
  1456. sendRemoteCommand(sendBuffer, replyBuffer);
  1457. unsigned ret;
  1458. replyBuffer.read(ret);
  1459. return (fileBool)ret;
  1460. }
  1461. IFileIO * open(IFOmode mode,IFEflags extraFlags=IFEnone);
  1462. IFileIO * openShared(IFOmode mode,IFSHmode shmode,IFEflags extraFlags=IFEnone);
  1463. IFileAsyncIO * openAsync(IFOmode mode) { return NULL; } // not supported
  1464. const char * queryFilename()
  1465. {
  1466. if (remotefilename.isEmpty()) {
  1467. RemoteFilename rfn;
  1468. rfn.setPath(ep,filename);
  1469. StringBuffer path;
  1470. rfn.getRemotePath(path);
  1471. remotefilename.set(path);
  1472. }
  1473. return remotefilename.get();
  1474. }
  1475. void resetLocalFilename(const char *name)
  1476. {
  1477. remotefilename.clear();
  1478. filename.set(name);
  1479. }
  1480. bool remove()
  1481. {
  1482. MemoryBuffer sendBuffer;
  1483. initSendBuffer(sendBuffer);
  1484. MemoryBuffer replyBuffer;
  1485. sendBuffer.append((RemoteFileCommandType)RFCremove).append(filename);
  1486. sendRemoteCommand(sendBuffer, replyBuffer);
  1487. bool ok;
  1488. replyBuffer.read(ok);
  1489. return ok;
  1490. }
  1491. void rename(const char *newname)
  1492. {
  1493. // currently ignores directory on newname (in future versions newname will be required to be tail only and not full path)
  1494. StringBuffer path;
  1495. splitDirTail(filename,path);
  1496. StringBuffer newdir;
  1497. path.append(splitDirTail(newname,newdir));
  1498. if (newdir.length()&&(strcmp(newdir.str(),path.str())!=0))
  1499. WARNLOG("CRemoteFile::rename passed full path '%s' that may not to match original directory '%s'",newname,path.str());
  1500. MemoryBuffer sendBuffer;
  1501. initSendBuffer(sendBuffer);
  1502. MemoryBuffer replyBuffer;
  1503. sendBuffer.append((RemoteFileCommandType)RFCrename).append(filename).append(path);
  1504. sendRemoteCommand(sendBuffer, replyBuffer);
  1505. filename.set(path);
  1506. remotefilename.clear();
  1507. }
  1508. void move(const char *newname)
  1509. {
  1510. // like rename except between directories
  1511. // first create replote path
  1512. if (!newname||!*newname)
  1513. return;
  1514. RemoteFilename destrfn;
  1515. if (isPathSepChar(newname[0])&&isPathSepChar(newname[1])) {
  1516. destrfn.setRemotePath(newname);
  1517. if (!destrfn.queryEndpoint().ipequals(ep)) {
  1518. StringBuffer msg;
  1519. msg.appendf("IFile::move %s to %s, destination node must match source node", queryFilename(), newname);
  1520. throw createDafsException(RFSERR_MoveFailed,msg.str());
  1521. }
  1522. }
  1523. else
  1524. destrfn.setPath(ep,newname);
  1525. StringBuffer dest;
  1526. newname = destrfn.getLocalPath(dest).str();
  1527. MemoryBuffer sendBuffer;
  1528. initSendBuffer(sendBuffer);
  1529. MemoryBuffer replyBuffer;
  1530. StringBuffer path;
  1531. splitDirTail(filename,path);
  1532. StringBuffer newdir;
  1533. const char *newtail = splitDirTail(newname,newdir);
  1534. if (strcmp(newdir.str(),path.str())==0) {
  1535. path.append(newtail);
  1536. newname = path;
  1537. sendBuffer.append((RemoteFileCommandType)RFCrename); // use rename if we can (supported on older dafilesrv)
  1538. }
  1539. else
  1540. sendBuffer.append((RemoteFileCommandType)RFCmove);
  1541. sendBuffer.append(filename).append(newname);
  1542. sendRemoteCommand(sendBuffer, replyBuffer);
  1543. filename.set(newname);
  1544. remotefilename.clear();
  1545. }
  1546. void setReadOnly(bool set)
  1547. {
  1548. MemoryBuffer sendBuffer;
  1549. initSendBuffer(sendBuffer);
  1550. MemoryBuffer replyBuffer;
  1551. sendBuffer.append((RemoteFileCommandType)RFCsetreadonly).append(filename).append(set);
  1552. sendRemoteCommand(sendBuffer, replyBuffer);
  1553. }
  1554. void setFilePermissions(unsigned fPerms)
  1555. {
  1556. MemoryBuffer sendBuffer;
  1557. initSendBuffer(sendBuffer);
  1558. MemoryBuffer replyBuffer;
  1559. sendBuffer.append((RemoteFileCommandType)RFCsetfileperms).append(filename).append(fPerms);
  1560. sendRemoteCommand(sendBuffer, replyBuffer);
  1561. }
  1562. offset_t size()
  1563. {
  1564. #if 1 // faster method (consistant with IFile)
  1565. // do this by using dir call (could be improved with new function but this not *too* bad)
  1566. if (isSpecialPath(filename))
  1567. return 0; // queries deemed to always exist (though don't know size).
  1568. // if needed to get size I guess could use IFileIO method and cache (bit of pain though)
  1569. StringBuffer dir;
  1570. const char *tail = splitDirTail(filename,dir);
  1571. if (!dir.length())
  1572. return false;
  1573. MemoryBuffer sendBuffer;
  1574. initSendBuffer(sendBuffer);
  1575. MemoryBuffer replyBuffer;
  1576. bool includedirs = true;
  1577. bool sub=false;
  1578. {
  1579. //Could be removed with new dafilesrv change [ (stream != 0) ], since this is not streaming.
  1580. Owned<CEndpointCS> crit = dirCSTable->getCrit(ep); // NB dirCSTable doesn't own, last reference will remove from table
  1581. CriticalBlock block(*crit);
  1582. sendBuffer.append((RemoteFileCommandType)RFCgetdir).append(dir).append(tail).append(includedirs).append(sub);
  1583. sendRemoteCommand(sendBuffer, replyBuffer);
  1584. }
  1585. // now should be 0 or 1 files returned
  1586. Owned<CRemoteDirectoryIterator> iter = new CRemoteDirectoryIterator(ep, dir.str());
  1587. iter->appendBuf(replyBuffer);
  1588. if (!iter->first())
  1589. return (offset_t)-1;
  1590. return (offset_t) iter->getFileSize();
  1591. #else
  1592. IFileIO * io = open(IFOread);
  1593. offset_t length = (offset_t)-1;
  1594. if (io)
  1595. {
  1596. length = io->size();
  1597. io->Release();
  1598. }
  1599. return length;
  1600. #endif
  1601. }
  1602. bool createDirectory()
  1603. {
  1604. MemoryBuffer sendBuffer;
  1605. initSendBuffer(sendBuffer);
  1606. MemoryBuffer replyBuffer;
  1607. sendBuffer.append((RemoteFileCommandType)RFCcreatedir).append(filename);
  1608. sendRemoteCommand(sendBuffer, replyBuffer);
  1609. bool ok;
  1610. replyBuffer.read(ok);
  1611. return ok;
  1612. }
  1613. virtual IDirectoryIterator *directoryFiles(const char *mask,bool sub,bool includedirs)
  1614. {
  1615. if (mask&&!*mask)
  1616. return createDirectoryIterator("",""); // NULL iterator
  1617. CRemoteDirectoryIterator *ret = new CRemoteDirectoryIterator(ep, filename);
  1618. byte stream=1;
  1619. Owned<CEndpointCS> crit = dirCSTable->getCrit(ep); // NB dirCSTable doesn't own, last reference will remove from table
  1620. CriticalBlock block(*crit);
  1621. loop {
  1622. MemoryBuffer sendBuffer;
  1623. initSendBuffer(sendBuffer);
  1624. MemoryBuffer replyBuffer;
  1625. sendBuffer.append((RemoteFileCommandType)RFCgetdir).append(filename).append(mask?mask:"").append(includedirs).append(sub).append(stream);
  1626. sendRemoteCommand(sendBuffer, replyBuffer);
  1627. if (ret->appendBuf(replyBuffer))
  1628. break;
  1629. stream = 2;
  1630. }
  1631. return ret;
  1632. }
  1633. IDirectoryDifferenceIterator *monitorDirectory(
  1634. IDirectoryIterator *prev=NULL, // in (NULL means use current as baseline)
  1635. const char *mask=NULL,
  1636. bool sub=false,
  1637. bool includedirs=false,
  1638. unsigned checkinterval=60*1000,
  1639. unsigned timeout=(unsigned)-1,
  1640. Semaphore *abortsem=NULL) // returns NULL if timed out
  1641. {
  1642. // abortsem not yet supported
  1643. MemoryBuffer sendBuffer;
  1644. initSendBuffer(sendBuffer);
  1645. MemoryBuffer replyBuffer;
  1646. sendBuffer.append((RemoteFileCommandType)RFCmonitordir).append(filename).append(mask?mask:"").append(includedirs).append(sub);
  1647. sendBuffer.append(checkinterval).append(timeout);
  1648. __int64 cancelid=0; // not yet used
  1649. sendBuffer.append(cancelid);
  1650. byte isprev=(prev!=NULL)?1:0;
  1651. sendBuffer.append(isprev);
  1652. if (prev)
  1653. CRemoteDirectoryIterator::serialize(sendBuffer,prev,0,true);
  1654. sendRemoteCommand(sendBuffer, replyBuffer);
  1655. byte status;
  1656. replyBuffer.read(status);
  1657. if (status==1) {
  1658. CRemoteDirectoryIterator *iter = new CRemoteDirectoryIterator(ep, filename);
  1659. iter->appendBuf(replyBuffer);
  1660. return iter;
  1661. }
  1662. return NULL;
  1663. }
  1664. bool getInfo(bool &isdir,offset_t &size,CDateTime &modtime)
  1665. {
  1666. // do this by using dir call (could be improved with new function but this not *too* bad)
  1667. StringBuffer dir;
  1668. const char *tail = splitDirTail(filename,dir);
  1669. if (!dir.length())
  1670. return false;
  1671. MemoryBuffer sendBuffer;
  1672. initSendBuffer(sendBuffer);
  1673. MemoryBuffer replyBuffer;
  1674. bool includedirs = true;
  1675. bool sub=false;
  1676. {
  1677. //Could be removed with new dafilesrv change [ (stream != 0) ], since this is not streaming.
  1678. Owned<CEndpointCS> crit = dirCSTable->getCrit(ep); // NB dirCSTable doesn't own, last reference will remove from table
  1679. CriticalBlock block(*crit);
  1680. sendBuffer.append((RemoteFileCommandType)RFCgetdir).append(dir).append(tail).append(includedirs).append(sub);
  1681. sendRemoteCommand(sendBuffer, replyBuffer);
  1682. }
  1683. // now should be 0 or 1 files returned
  1684. Owned<CRemoteDirectoryIterator> iter = new CRemoteDirectoryIterator(ep, dir.str());
  1685. iter->appendBuf(replyBuffer);
  1686. if (!iter->first())
  1687. return false;
  1688. isdir = iter->isDir();
  1689. size = (offset_t) iter->getFileSize();
  1690. iter->getModifiedTime(modtime);
  1691. return true;
  1692. }
  1693. bool setCompression(bool set)
  1694. {
  1695. assertex(!"Need to implement compress()");
  1696. return false;
  1697. }
  1698. offset_t compressedSize()
  1699. {
  1700. assertex(!"Need to implement actualSize()");
  1701. return (offset_t)-1;
  1702. }
  1703. void serialize(MemoryBuffer &tgt)
  1704. {
  1705. throwUnexpected();
  1706. }
  1707. void deserialize(MemoryBuffer &src)
  1708. {
  1709. throwUnexpected();
  1710. }
  1711. unsigned getCRC()
  1712. {
  1713. MemoryBuffer sendBuffer;
  1714. initSendBuffer(sendBuffer);
  1715. MemoryBuffer replyBuffer;
  1716. sendBuffer.append((RemoteFileCommandType)RFCgetcrc).append(filename);
  1717. sendRemoteCommand(sendBuffer, replyBuffer, true, true);
  1718. unsigned crc;
  1719. replyBuffer.read(crc);
  1720. return crc;
  1721. }
  1722. void setCreateFlags(unsigned cflags)
  1723. {
  1724. flags |= (cflags<<16);
  1725. }
  1726. void setShareMode(IFSHmode shmode)
  1727. {
  1728. flags &= ~(IFSHfull|IFSHread);
  1729. flags |= (unsigned)(shmode&(IFSHfull|IFSHread));
  1730. }
  1731. void remoteExtractBlobElements(const char * prefix, ExtractedBlobArray & extracted)
  1732. {
  1733. MemoryBuffer sendBuffer;
  1734. initSendBuffer(sendBuffer);
  1735. sendBuffer.append((RemoteFileCommandType)RFCextractblobelements).append(prefix).append(filename);
  1736. MemoryBuffer replyBuffer;
  1737. sendRemoteCommand(sendBuffer, replyBuffer, true, true); // handles error code
  1738. unsigned n;
  1739. replyBuffer.read(n);
  1740. for (unsigned i=0;i<n;i++) {
  1741. ExtractedBlobInfo *item = new ExtractedBlobInfo;
  1742. item->deserialize(replyBuffer);
  1743. extracted.append(*item);
  1744. }
  1745. }
  1746. bool copySectionAsync(const char *uuid,const RemoteFilename &dest, offset_t toOfs, offset_t fromOfs, offset_t size, ICopyFileProgress *progress, unsigned timeout)
  1747. {
  1748. // now if we get here is it can be assumed the source file is local to where we send the command
  1749. StringBuffer tos;
  1750. dest.getRemotePath(tos);
  1751. MemoryBuffer sendBuffer;
  1752. initSendBuffer(sendBuffer);
  1753. MemoryBuffer replyBuffer;
  1754. sendBuffer.append((RemoteFileCommandType)RFCcopysection).append(uuid).append(queryLocalName()).append(tos).append(toOfs).append(fromOfs).append(size).append(timeout);
  1755. sendRemoteCommand(sendBuffer, replyBuffer);
  1756. unsigned status;
  1757. replyBuffer.read(status);
  1758. if (progress) {
  1759. offset_t sizeDone;
  1760. offset_t totalSize;
  1761. replyBuffer.read(sizeDone).read(totalSize);
  1762. progress->onProgress(sizeDone,totalSize);
  1763. }
  1764. return (AsyncCommandStatus)status!=ACScontinue; // should only otherwise be done as errors raised by exception
  1765. }
  1766. void copySection(const RemoteFilename &dest, offset_t toOfs, offset_t fromOfs, offset_t size, ICopyFileProgress *progress, CFflags copyFlags=CFnone)
  1767. {
  1768. StringBuffer uuid;
  1769. genUUID(uuid,true);
  1770. unsigned timeout = 60*1000; // check every minute
  1771. while(!copySectionAsync(uuid.str(),dest,toOfs,fromOfs,size,progress,timeout));
  1772. }
  1773. void copyTo(IFile *dest, size32_t buffersize, ICopyFileProgress *progress, bool usetmp, CFflags copyFlags=CFnone);
  1774. virtual IMemoryMappedFile *openMemoryMapped(offset_t ofs, memsize_t len, bool write)
  1775. {
  1776. return NULL;
  1777. }
  1778. void treeCopyTo(IFile *dest,IpSubNet &subnet,IpAddress &resfrom,bool usetmp,CFflags copyFlags=CFnone)
  1779. {
  1780. resfrom.ipset(NULL);
  1781. MemoryBuffer sendBuffer;
  1782. initSendBuffer(sendBuffer);
  1783. MemoryBuffer replyBuffer;
  1784. sendBuffer.append((RemoteFileCommandType)(usetmp?RFCtreecopytmp:RFCtreecopy));
  1785. RemoteFilename rfn;
  1786. rfn.setPath(ep,filename);
  1787. rfn.serialize(sendBuffer);
  1788. const char *d = dest->queryFilename();
  1789. if (!isAbsolutePath(d))
  1790. throw MakeStringException(-1,"treeCopyFile destination '%s' is not an absolute path", d);
  1791. rfn.setRemotePath(d);
  1792. rfn.serialize(sendBuffer);
  1793. StringBuffer tmp;
  1794. subnet.getNetText(tmp);
  1795. sendBuffer.append(tmp);
  1796. subnet.getMaskText(tmp.clear());
  1797. sendBuffer.append(tmp);
  1798. unsigned status=1;
  1799. try {
  1800. sendRemoteCommand(sendBuffer, replyBuffer);
  1801. replyBuffer.read(status);
  1802. }
  1803. catch (IDAFS_Exception *e) {
  1804. if (e->errorCode()!=RFSERR_InvalidCommand)
  1805. throw;
  1806. e->Release();
  1807. status = (unsigned)-1;
  1808. }
  1809. if (status==-1) {
  1810. resfrom.ipset(ep);
  1811. StringBuffer tmp;
  1812. WARNLOG("dafilesrv on %s does not support treeCopyTo - falling back to copyTo",resfrom.getIpText(tmp).str());
  1813. copyTo(dest,DEFAULT_COPY_BLKSIZE,NULL,usetmp,copyFlags);
  1814. status = 0;
  1815. }
  1816. else if (status==0)
  1817. resfrom.ipdeserialize(replyBuffer);
  1818. }
  1819. };
  1820. void clientCacheFileConnect(SocketEndpoint &_ep,unsigned timeout)
  1821. {
  1822. if (!timeout) {
  1823. SocketEndpoint ep(_ep);
  1824. setDafsEndpointPort(ep);
  1825. Owned<CRemoteFile> cfile = new CRemoteFile(ep, "null");
  1826. cfile->connect();
  1827. return; // frees file and adds its socket to cache
  1828. }
  1829. // timeout needed so start a thread (that may become orphaned)
  1830. class cThread: public Thread
  1831. {
  1832. SocketEndpoint ep;
  1833. public:
  1834. cThread(SocketEndpoint &_ep)
  1835. : Thread("cacheFileConnect")
  1836. {
  1837. ep = _ep;
  1838. }
  1839. int run()
  1840. {
  1841. try {
  1842. clientCacheFileConnect(ep,0);
  1843. }
  1844. catch (IException *e) {
  1845. CriticalBlock block(sect);
  1846. except.setown(e);
  1847. }
  1848. return 0;
  1849. }
  1850. Owned<IException> except;
  1851. CriticalSection sect;
  1852. } *thread;
  1853. thread = new cThread(_ep);
  1854. thread->start();
  1855. IException *e =NULL;
  1856. if (!thread->join(timeout)) {
  1857. StringBuffer msg("Timed out connecting to ");
  1858. _ep.getUrlStr(msg);
  1859. e = createDafsException(RFSERR_AuthenticateFailed,msg.str());
  1860. }
  1861. {
  1862. CriticalBlock block(thread->sect);
  1863. if (!e&&thread->except)
  1864. e = thread->except.getClear();
  1865. }
  1866. thread->Release();
  1867. if (e)
  1868. throw e;
  1869. }
  1870. void clientAddSocketToCache(SocketEndpoint &ep,ISocket *socket)
  1871. {
  1872. CriticalBlock block(CConnectionTable::crit);
  1873. if (ConnectionTable)
  1874. ConnectionTable->addLink(ep,socket);
  1875. }
  1876. IFile * createRemoteFile(SocketEndpoint &ep, const char * filename)
  1877. {
  1878. IFile *ret = createFileLocalMount(ep,filename);
  1879. if (ret)
  1880. return ret;
  1881. return new CRemoteFile(ep, filename);
  1882. }
  1883. void clientDisconnectRemoteFile(IFile *file)
  1884. {
  1885. CRemoteFile *cfile = QUERYINTERFACE(file,CRemoteFile);
  1886. if (cfile)
  1887. cfile->disconnect();
  1888. }
  1889. bool clientResetFilename(IFile *file, const char *newname) // returns false if not remote
  1890. {
  1891. CRemoteFile *cfile = QUERYINTERFACE(file,CRemoteFile);
  1892. if (!cfile)
  1893. return false;
  1894. cfile->resetLocalFilename(newname);
  1895. return true;
  1896. }
  1897. extern bool clientAsyncCopyFileSection(const char *uuid,
  1898. IFile *from, // expected to be remote
  1899. RemoteFilename &to,
  1900. offset_t toOfs, // -1 created file and copies to start
  1901. offset_t fromOfs,
  1902. offset_t size,
  1903. ICopyFileProgress *progress,
  1904. unsigned timeout) // returns true when done
  1905. {
  1906. CRemoteFile *cfile = QUERYINTERFACE(from,CRemoteFile);
  1907. if (!cfile) {
  1908. // local - do sync
  1909. from->copySection(to,toOfs,fromOfs,size,progress);
  1910. return true;
  1911. }
  1912. return cfile->copySectionAsync(uuid,to,toOfs,fromOfs, size, progress, timeout);
  1913. }
  1914. //---------------------------------------------------------------------------
  1915. class CRemoteFileIO : public CInterface, implements IFileIO
  1916. {
  1917. protected:
  1918. Linked<CRemoteFile> parent;
  1919. RemoteFileIOHandle handle;
  1920. IFOmode mode;
  1921. compatIFSHmode compatmode;
  1922. IFEflags extraFlags;
  1923. bool disconnectonexit;
  1924. public:
  1925. IMPLEMENT_IINTERFACE
  1926. CRemoteFileIO(CRemoteFile *_parent)
  1927. : parent(_parent)
  1928. {
  1929. handle = 0;
  1930. disconnectonexit = false;
  1931. }
  1932. ~CRemoteFileIO()
  1933. {
  1934. if (handle) {
  1935. try {
  1936. close();
  1937. }
  1938. catch (IException *e) {
  1939. StringBuffer s;
  1940. e->errorMessage(s);
  1941. WARNLOG("CRemoteFileIO close file: %s",s.str());
  1942. e->Release();
  1943. }
  1944. }
  1945. if (disconnectonexit)
  1946. parent->disconnect();
  1947. }
  1948. void close()
  1949. {
  1950. if (handle) {
  1951. try {
  1952. MemoryBuffer sendBuffer;
  1953. initSendBuffer(sendBuffer);
  1954. sendBuffer.append((RemoteFileCommandType)RFCcloseIO).append(handle);
  1955. parent->sendRemoteCommand(sendBuffer,false);
  1956. }
  1957. catch (IDAFS_Exception *e) {
  1958. if ((e->errorCode()!=RFSERR_InvalidFileIOHandle)&&(e->errorCode()!=RFSERR_NullFileIOHandle))
  1959. throw;
  1960. e->Release();
  1961. }
  1962. handle = 0;
  1963. }
  1964. }
  1965. bool open(IFOmode _mode,compatIFSHmode _compatmode,IFEflags _extraFlags=IFEnone)
  1966. {
  1967. MemoryBuffer sendBuffer;
  1968. initSendBuffer(sendBuffer);
  1969. MemoryBuffer replyBuffer;
  1970. const char *localname = parent->queryLocalName();
  1971. localname = skipSpecialPath(localname);
  1972. // also send _extraFlags
  1973. sendBuffer.append((RemoteFileCommandType)RFCopenIO).append(localname).append((byte)_mode).append((byte)_compatmode).append((byte)_extraFlags);
  1974. parent->sendRemoteCommand(sendBuffer, replyBuffer);
  1975. replyBuffer.read(handle);
  1976. if (!handle)
  1977. return false;
  1978. switch (_mode) {
  1979. case IFOcreate:
  1980. mode = IFOwrite;
  1981. break;
  1982. case IFOcreaterw:
  1983. mode = IFOreadwrite;
  1984. break;
  1985. default:
  1986. mode = _mode;
  1987. }
  1988. compatmode = _compatmode;
  1989. extraFlags = _extraFlags;
  1990. return true;
  1991. }
  1992. bool reopen()
  1993. {
  1994. StringBuffer s;
  1995. PROGLOG("Attempting reopen of %s on %s",parent->queryLocalName(),parent->queryEp().getUrlStr(s).str());
  1996. if (open(mode,compatmode,extraFlags)) {
  1997. return true;
  1998. }
  1999. return false;
  2000. }
  2001. offset_t size()
  2002. {
  2003. MemoryBuffer sendBuffer;
  2004. initSendBuffer(sendBuffer);
  2005. MemoryBuffer replyBuffer;
  2006. sendBuffer.append((RemoteFileCommandType)RFCsize).append(handle);
  2007. parent->sendRemoteCommand(sendBuffer, replyBuffer, false);
  2008. // Retry using reopen TBD
  2009. offset_t ret;
  2010. replyBuffer.read(ret);
  2011. return ret;
  2012. }
  2013. size32_t read(offset_t pos, size32_t len, void * data)
  2014. {
  2015. size32_t got;
  2016. MemoryBuffer replyBuffer;
  2017. const void *b = doRead(pos,len,replyBuffer,got,data);
  2018. if (b!=data)
  2019. memcpy(data,b,got);
  2020. return got;
  2021. }
  2022. virtual void flush()
  2023. {
  2024. }
  2025. const void *doRead(offset_t pos, size32_t len, MemoryBuffer &replyBuffer, size32_t &got, void *dstbuf)
  2026. {
  2027. unsigned tries=0;
  2028. loop {
  2029. try {
  2030. MemoryBuffer sendBuffer;
  2031. initSendBuffer(sendBuffer);
  2032. replyBuffer.clear();
  2033. sendBuffer.append((RemoteFileCommandType)RFCread).append(handle).append(pos).append(len);
  2034. parent->sendRemoteCommand(sendBuffer, replyBuffer,false);
  2035. // kludge dafilesrv versions <= 1.5e don't return error correctly
  2036. if (replyBuffer.length()>len+sizeof(size32_t)+sizeof(unsigned)) {
  2037. size32_t save = replyBuffer.getPos();
  2038. replyBuffer.reset(len+sizeof(size32_t)+sizeof(unsigned));
  2039. unsigned errCode;
  2040. replyBuffer.read(errCode);
  2041. if (errCode) {
  2042. StringBuffer msg;
  2043. parent->ep.getUrlStr(msg.append('[')).append("] ");
  2044. if (replyBuffer.getPos()<replyBuffer.length()) {
  2045. StringAttr s;
  2046. replyBuffer.read(s);
  2047. msg.append(s);
  2048. }
  2049. else
  2050. msg.append("ERROR #").append(errCode);
  2051. throw createDafsException(errCode, msg.str());
  2052. }
  2053. else
  2054. replyBuffer.reset(save);
  2055. }
  2056. replyBuffer.read(got);
  2057. if ((got>replyBuffer.remaining())||(got>len)) {
  2058. PROGLOG("Read beyond buffer %d,%d,%d",got,replyBuffer.remaining(),len);
  2059. throw createDafsException(RFSERR_ReadFailed, "Read beyond buffer");
  2060. }
  2061. return replyBuffer.readDirect(got);
  2062. }
  2063. catch (IJSOCK_Exception *e) {
  2064. EXCLOG(e,"CRemoteFileIO::read");
  2065. if (++tries>3)
  2066. throw;
  2067. WARNLOG("Retrying read of %s (%d)",parent->queryLocalName(),tries);
  2068. Owned<IException> exc = e;
  2069. if (!reopen())
  2070. throw exc.getClear();
  2071. }
  2072. }
  2073. got = 0;
  2074. return NULL;
  2075. }
  2076. size32_t write(offset_t pos, size32_t len, const void * data)
  2077. {
  2078. unsigned tries=0;
  2079. size32_t ret = 0;
  2080. loop {
  2081. try {
  2082. MemoryBuffer replyBuffer;
  2083. MemoryBuffer sendBuffer;
  2084. initSendBuffer(sendBuffer);
  2085. sendBuffer.append((RemoteFileCommandType)RFCwrite).append(handle).append(pos).append(len).append(len, data);
  2086. parent->sendRemoteCommand(sendBuffer, replyBuffer, false, true);
  2087. replyBuffer.read(ret);
  2088. break;
  2089. }
  2090. catch (IJSOCK_Exception *e) {
  2091. EXCLOG(e,"CRemoteFileIO::write");
  2092. if (++tries>3)
  2093. throw;
  2094. WARNLOG("Retrying write(%" I64F "d,%d) of %s (%d)",pos,len,parent->queryLocalName(),tries);
  2095. Owned<IException> exc = e;
  2096. if (!reopen())
  2097. throw exc.getClear();
  2098. }
  2099. }
  2100. if ((ret==(size32_t)-1) || (ret < len))
  2101. throw createDafsException(DISK_FULL_EXCEPTION_CODE,"write failed, disk full?");
  2102. return ret;
  2103. }
  2104. offset_t appendFile(IFile *file,offset_t pos,offset_t len)
  2105. {
  2106. MemoryBuffer sendBuffer;
  2107. initSendBuffer(sendBuffer);
  2108. MemoryBuffer replyBuffer;
  2109. const char * fname = file->queryFilename();
  2110. sendBuffer.append((RemoteFileCommandType)RFCappend).append(handle).append(fname).append(pos).append(len);
  2111. parent->sendRemoteCommand(sendBuffer, replyBuffer, false, true); // retry not safe
  2112. offset_t ret;
  2113. replyBuffer.read(ret);
  2114. if ((ret==(offset_t)-1) || (ret < len))
  2115. throw createDafsException(DISK_FULL_EXCEPTION_CODE,"append failed, disk full?"); // though could be file missing TBD
  2116. return ret;
  2117. }
  2118. void setSize(offset_t size)
  2119. {
  2120. MemoryBuffer sendBuffer;
  2121. initSendBuffer(sendBuffer);
  2122. MemoryBuffer replyBuffer;
  2123. sendBuffer.append((RemoteFileCommandType)RFCsetsize).append(handle).append(size);
  2124. parent->sendRemoteCommand(sendBuffer, replyBuffer, false, true);
  2125. // retry using reopen TBD
  2126. }
  2127. void setDisconnectOnExit(bool set) { disconnectonexit = set; }
  2128. };
  2129. void clientDisconnectRemoteIoOnExit(IFileIO *fileio,bool set)
  2130. {
  2131. CRemoteFileIO *cfileio = QUERYINTERFACE(fileio,CRemoteFileIO);
  2132. if (cfileio)
  2133. cfileio->setDisconnectOnExit(set);
  2134. }
  2135. IFileIO * CRemoteFile::openShared(IFOmode mode,IFSHmode shmode,IFEflags extraFlags)
  2136. {
  2137. assertex(((unsigned)shmode&0xffffffc7)==0);
  2138. compatIFSHmode compatmode;
  2139. unsigned fileflags = (flags>>16) & (S_IRUSR|S_IWUSR|S_IXUSR|S_IRGRP|S_IWGRP|S_IXGRP|S_IROTH|S_IWOTH|S_IXOTH);
  2140. if (fileflags&S_IXUSR) // this is bit hit and miss but backward compatible
  2141. compatmode = compatIFSHexec;
  2142. else if (fileflags&(S_IWGRP|S_IWOTH))
  2143. compatmode = compatIFSHall;
  2144. else if (shmode&IFSHfull)
  2145. compatmode = compatIFSHwrite;
  2146. else if (((shmode&(IFSHread|IFSHfull))==0) && ((fileflags&(S_IRGRP|S_IROTH))==0))
  2147. compatmode = compatIFSHnone;
  2148. else
  2149. compatmode = compatIFSHread;
  2150. Owned<CRemoteFileIO> res = new CRemoteFileIO(this);
  2151. if (res->open(mode,compatmode,extraFlags))
  2152. return res.getClear();
  2153. return NULL;
  2154. }
  2155. IFileIO * CRemoteFile::open(IFOmode mode,IFEflags extraFlags)
  2156. {
  2157. return openShared(mode,(IFSHmode)(flags&(IFSHread|IFSHfull)),extraFlags);
  2158. }
  2159. //---------------------------------------------------------------------------
  2160. void CRemoteFile::copyTo(IFile *dest, size32_t buffersize, ICopyFileProgress *progress, bool usetmp, CFflags copyFlags)
  2161. {
  2162. CRemoteFile *dstfile = QUERYINTERFACE(dest,CRemoteFile);
  2163. if (dstfile&&!dstfile->queryEp().isLocal()) {
  2164. StringBuffer tmpname;
  2165. Owned<IFile> destf;
  2166. RemoteFilename dest;
  2167. if (usetmp) {
  2168. makeTempCopyName(tmpname,dstfile->queryLocalName());
  2169. dest.setPath(dstfile->queryEp(),tmpname.str());
  2170. }
  2171. else
  2172. dest.setPath(dstfile->queryEp(),dstfile->queryLocalName());
  2173. destf.setown(createIFile(dest));
  2174. try {
  2175. // following may fail if new dafilesrv not deployed on src
  2176. copySection(dest,(offset_t)-1,0,(offset_t)-1,progress,copyFlags);
  2177. if (usetmp) {
  2178. StringAttr tail(pathTail(dstfile->queryLocalName()));
  2179. dstfile->remove();
  2180. destf->rename(tail);
  2181. }
  2182. return;
  2183. }
  2184. catch (IException *e)
  2185. {
  2186. StringBuffer s;
  2187. s.appendf("Remote File Copy (%d): ",e->errorCode());
  2188. e->errorMessage(s);
  2189. s.append(", retrying local");
  2190. WARNLOG("%s",s.str());
  2191. e->Release();
  2192. }
  2193. // delete dest
  2194. try {
  2195. destf->remove();
  2196. }
  2197. catch (IException *e)
  2198. {
  2199. EXCLOG(e,"Remote File Copy, Deleting temporary file");
  2200. e->Release();
  2201. }
  2202. }
  2203. // assumption if we get here that source remote, dest local (or equiv)
  2204. class cIntercept: implements ICopyFileIntercept
  2205. {
  2206. MemoryAttr ma;
  2207. MemoryBuffer mb;
  2208. virtual offset_t copy(IFileIO *from, IFileIO *to, offset_t ofs, size32_t sz)
  2209. {
  2210. if (ma.length()<sz)
  2211. ma.allocate(sz); // may be not used
  2212. void *buf = ma.bufferBase();
  2213. size32_t got;
  2214. CRemoteFileIO *srcio = QUERYINTERFACE(from,CRemoteFileIO);
  2215. const void *dst;
  2216. if (srcio)
  2217. dst = srcio->doRead(ofs,sz,mb.clear(),got,buf);
  2218. else {
  2219. // shouldn't ever get here if source remote
  2220. got = from->read(ofs, sz, buf);
  2221. dst = buf;
  2222. }
  2223. if (got != 0)
  2224. to->write(ofs, got, dst);
  2225. return got;
  2226. }
  2227. } intercept;
  2228. doCopyFile(dest,this,buffersize,progress,&intercept,usetmp,copyFlags);
  2229. }
  2230. unsigned getRemoteVersion(ISocket * socket, StringBuffer &ver)
  2231. {
  2232. // used to have a global critical section here
  2233. if (!socket)
  2234. return 0;
  2235. unsigned ret;
  2236. MemoryBuffer sendbuf;
  2237. initSendBuffer(sendbuf);
  2238. sendbuf.append((RemoteFileCommandType)RFCgetver);
  2239. sendbuf.append((unsigned)RFCgetver);
  2240. MemoryBuffer reply;
  2241. try {
  2242. sendBuffer(socket, sendbuf);
  2243. receiveBuffer(socket, reply, 1 ,4096);
  2244. unsigned errCode;
  2245. reply.read(errCode);
  2246. if (errCode==RFSERR_InvalidCommand) {
  2247. ver.append("DS V1.0");
  2248. return 10;
  2249. }
  2250. else if (errCode==0)
  2251. ret = 11;
  2252. else if (errCode<0x10000)
  2253. return 0;
  2254. else
  2255. ret = errCode-0x10000;
  2256. }
  2257. catch (IException *e) {
  2258. EXCLOG(e);
  2259. ::Release(e);
  2260. return 0;
  2261. }
  2262. StringAttr vers;
  2263. reply.read(vers);
  2264. ver.append(vers);
  2265. return ret;
  2266. }
  2267. extern unsigned stopRemoteServer(ISocket * socket)
  2268. {
  2269. // used to have a global critical section here
  2270. if (!socket)
  2271. return 0;
  2272. MemoryBuffer sendbuf;
  2273. initSendBuffer(sendbuf);
  2274. sendbuf.append((RemoteFileCommandType)RFCstop);
  2275. sendbuf.append((unsigned)RFCstop);
  2276. MemoryBuffer replybuf;
  2277. unsigned errCode = RFSERR_InvalidCommand;
  2278. try {
  2279. sendBuffer(socket, sendbuf);
  2280. receiveBuffer(socket, replybuf, NORMAL_RETRIES, 1024);
  2281. replybuf.read(errCode);
  2282. }
  2283. catch (IJSOCK_Exception *e) {
  2284. if ((e->errorCode()!=JSOCKERR_broken_pipe)&&(e->errorCode()!=JSOCKERR_graceful_close))
  2285. EXCLOG(e);
  2286. else
  2287. errCode = 0;
  2288. }
  2289. catch (IException *e) {
  2290. EXCLOG(e);
  2291. ::Release(e);
  2292. }
  2293. return errCode;
  2294. }
  2295. int remoteExec(ISocket * socket, const char *cmdline, const char *workdir,bool sync,
  2296. size32_t insize, void *inbuf, MemoryBuffer *outbuf)
  2297. {
  2298. if (!socket)
  2299. return -1;
  2300. bool hasoutput = (outbuf!=NULL);
  2301. if (!inbuf)
  2302. insize = 0;
  2303. MemoryBuffer sendbuf;
  2304. initSendBuffer(sendbuf);
  2305. sendbuf.append((RemoteFileCommandType)RFCexec).append(cmdline).append(workdir).append(sync).
  2306. append(hasoutput).append(insize);
  2307. if (insize)
  2308. sendbuf.append(insize, inbuf);
  2309. MemoryBuffer replybuf;
  2310. try {
  2311. sendBuffer(socket, sendbuf);
  2312. receiveBuffer(socket, replybuf, LENGTHY_RETRIES); // we don't know how long program will take really - assume <1hr
  2313. int retcode;
  2314. unsigned phandle;
  2315. size32_t outsz;
  2316. replybuf.read(retcode).read(phandle).read(outsz);
  2317. if (outsz&&outbuf)
  2318. replybuf.read(outsz,outbuf->reserve(outsz));
  2319. return retcode;
  2320. }
  2321. catch (IException *e) {
  2322. EXCLOG(e);
  2323. ::Release(e);
  2324. }
  2325. return -1;
  2326. }
  2327. int setDafsTrace(ISocket * socket,byte flags)
  2328. {
  2329. if (!socket) {
  2330. byte ret = traceFlags;
  2331. traceFlags = flags;
  2332. return ret;
  2333. }
  2334. MemoryBuffer sendbuf;
  2335. initSendBuffer(sendbuf);
  2336. sendbuf.append((RemoteFileCommandType)RFCsettrace).append(flags);
  2337. MemoryBuffer replybuf;
  2338. try {
  2339. sendBuffer(socket, sendbuf);
  2340. receiveBuffer(socket, replybuf, NORMAL_RETRIES, 1024);
  2341. int retcode;
  2342. replybuf.read(retcode);
  2343. return retcode;
  2344. }
  2345. catch (IException *e) {
  2346. EXCLOG(e);
  2347. ::Release(e);
  2348. }
  2349. return -1;
  2350. }
  2351. int setDafsThrottleLimit(ISocket * socket, ThrottleClass throttleClass, unsigned throttleLimit, unsigned throttleDelayMs, unsigned throttleCPULimit, unsigned queueLimit, StringBuffer *errMsg)
  2352. {
  2353. assertex(socket);
  2354. MemoryBuffer sendbuf;
  2355. initSendBuffer(sendbuf);
  2356. sendbuf.append((RemoteFileCommandType)RFCsetthrottle2).append((unsigned)throttleClass).append(throttleLimit);
  2357. sendbuf.append(throttleDelayMs).append(throttleCPULimit).append(queueLimit);
  2358. MemoryBuffer replybuf;
  2359. try {
  2360. sendBuffer(socket, sendbuf);
  2361. receiveBuffer(socket, replybuf, NORMAL_RETRIES, 1024);
  2362. int retcode;
  2363. replybuf.read(retcode);
  2364. if (retcode && errMsg && replybuf.remaining())
  2365. replybuf.read(*errMsg);
  2366. return retcode;
  2367. }
  2368. catch (IException *e) {
  2369. EXCLOG(e);
  2370. ::Release(e);
  2371. }
  2372. return -1;
  2373. }
  2374. int getDafsInfo(ISocket * socket, unsigned level, StringBuffer &retstr)
  2375. {
  2376. if (!socket) {
  2377. retstr.append(VERSTRING);
  2378. return 0;
  2379. }
  2380. MemoryBuffer sendbuf;
  2381. initSendBuffer(sendbuf);
  2382. sendbuf.append((RemoteFileCommandType)RFCgetinfo).append(level);
  2383. MemoryBuffer replybuf;
  2384. try {
  2385. sendBuffer(socket, sendbuf);
  2386. receiveBuffer(socket, replybuf, 1);
  2387. int retcode;
  2388. replybuf.read(retcode);
  2389. if (retcode==0) {
  2390. StringAttr s;
  2391. replybuf.read(s);
  2392. retstr.append(s);
  2393. }
  2394. return retcode;
  2395. }
  2396. catch (IException *e) {
  2397. EXCLOG(e);
  2398. ::Release(e);
  2399. }
  2400. return -1;
  2401. }
  2402. void remoteExtractBlobElements(const SocketEndpoint &ep,const char * prefix, const char * filename, ExtractedBlobArray & extracted)
  2403. {
  2404. Owned<CRemoteFile> file = new CRemoteFile (ep,filename);
  2405. file->remoteExtractBlobElements(prefix, extracted);
  2406. }
  2407. //====================================================================================================
  2408. class CAsyncCommandManager
  2409. {
  2410. class CAsyncJob: public CInterface
  2411. {
  2412. class cThread: public Thread
  2413. {
  2414. CAsyncJob *parent;
  2415. public:
  2416. cThread(CAsyncJob *_parent)
  2417. : Thread("CAsyncJob")
  2418. {
  2419. parent = _parent;
  2420. }
  2421. int run()
  2422. {
  2423. int ret = -1;
  2424. try {
  2425. ret = parent->run();
  2426. parent->setDone();
  2427. }
  2428. catch (IException *e)
  2429. {
  2430. parent->setException(e);
  2431. }
  2432. parent->signal();
  2433. return ret;
  2434. }
  2435. } *thread;
  2436. StringAttr uuid;
  2437. CAsyncCommandManager &parent;
  2438. public:
  2439. CAsyncJob(CAsyncCommandManager &_parent, const char *_uuid)
  2440. : parent(_parent), uuid(_uuid)
  2441. {
  2442. thread = new cThread(this);
  2443. hash = hashc((const byte *)uuid.get(),uuid.length(),~0U);
  2444. }
  2445. ~CAsyncJob()
  2446. {
  2447. thread->join();
  2448. thread->Release();
  2449. }
  2450. static void destroy(CAsyncJob *j)
  2451. {
  2452. j->Release();
  2453. }
  2454. void signal()
  2455. {
  2456. parent.signal();
  2457. }
  2458. void start()
  2459. {
  2460. parent.wait();
  2461. thread->start();
  2462. }
  2463. void join()
  2464. {
  2465. thread->join();
  2466. }
  2467. static unsigned getHash(const char *key)
  2468. {
  2469. return hashc((const byte *)key,strlen(key),~0U);
  2470. }
  2471. static CAsyncJob* create(const char *key) { assertex(!"CAsyncJob::create not implemented"); return NULL; }
  2472. unsigned hash;
  2473. bool eq(const char *key)
  2474. {
  2475. return stricmp(key,uuid.get())==0;
  2476. }
  2477. virtual int run()=0;
  2478. virtual void setException(IException *e)=0;
  2479. virtual void setDone()=0;
  2480. };
  2481. class CAsyncCopySection: public CAsyncJob
  2482. {
  2483. Owned<IFile> src;
  2484. RemoteFilename dst;
  2485. offset_t toOfs;
  2486. offset_t fromOfs;
  2487. offset_t size;
  2488. CFPmode mode; // not yet supported
  2489. CriticalSection sect;
  2490. offset_t done;
  2491. offset_t total;
  2492. Semaphore finished;
  2493. AsyncCommandStatus status;
  2494. Owned<IException> exc;
  2495. public:
  2496. CAsyncCopySection(CAsyncCommandManager &parent, const char *_uuid, const char *fromFile, const char *toFile, offset_t _toOfs, offset_t _fromOfs, offset_t _size)
  2497. : CAsyncJob(parent, _uuid)
  2498. {
  2499. status = ACScontinue;
  2500. src.setown(createIFile(fromFile));
  2501. dst.setRemotePath(toFile);
  2502. toOfs = _toOfs;
  2503. fromOfs = _fromOfs;
  2504. size = _size;
  2505. mode = CFPcontinue;
  2506. done = 0;
  2507. total = (offset_t)-1;
  2508. }
  2509. AsyncCommandStatus poll(offset_t &_done, offset_t &_total,unsigned timeout)
  2510. {
  2511. if (timeout&&finished.wait(timeout))
  2512. finished.signal(); // may need to call again
  2513. CriticalBlock block(sect);
  2514. if (exc)
  2515. throw exc.getClear();
  2516. _done = done;
  2517. _total = total;
  2518. return status;
  2519. }
  2520. int run()
  2521. {
  2522. class cProgress: implements ICopyFileProgress
  2523. {
  2524. CriticalSection &sect;
  2525. CFPmode &mode;
  2526. offset_t &done;
  2527. offset_t &total;
  2528. public:
  2529. cProgress(CriticalSection &_sect,offset_t &_done,offset_t &_total,CFPmode &_mode)
  2530. : sect(_sect), done(_done), total(_total), mode(_mode)
  2531. {
  2532. }
  2533. CFPmode onProgress(offset_t sizeDone, offset_t totalSize)
  2534. {
  2535. CriticalBlock block(sect);
  2536. done = sizeDone;
  2537. total = totalSize;
  2538. return mode;
  2539. }
  2540. } progress(sect,total,done,mode);
  2541. src->copySection(dst,toOfs, fromOfs, size, &progress); // exceptions will be handled by base class
  2542. return 0;
  2543. }
  2544. void setException(IException *e)
  2545. {
  2546. EXCLOG(e,"CAsyncCommandManager::CAsyncJob");
  2547. CriticalBlock block(sect);
  2548. if (exc.get())
  2549. e->Release();
  2550. else
  2551. exc.setown(e);
  2552. status = ACSerror;
  2553. }
  2554. void setDone()
  2555. {
  2556. CriticalBlock block(sect);
  2557. finished.signal();
  2558. status = ACSdone;
  2559. }
  2560. };
  2561. CMinHashTable<CAsyncJob> jobtable;
  2562. CriticalSection sect;
  2563. Semaphore threadsem;
  2564. unsigned limit;
  2565. public:
  2566. CAsyncCommandManager(unsigned _limit) : limit(_limit)
  2567. {
  2568. if (limit) // 0 == unbound
  2569. threadsem.signal(limit); // max number of async jobs
  2570. }
  2571. void join()
  2572. {
  2573. CriticalBlock block(sect);
  2574. unsigned i;
  2575. CAsyncJob *j=jobtable.first(i);
  2576. while (j) {
  2577. j->join();
  2578. j=jobtable.next(i);
  2579. }
  2580. }
  2581. void signal()
  2582. {
  2583. if (limit)
  2584. threadsem.signal();
  2585. }
  2586. void wait()
  2587. {
  2588. if (limit)
  2589. threadsem.wait();
  2590. }
  2591. AsyncCommandStatus copySection(const char *uuid, const char *fromFile, const char *toFile, offset_t toOfs, offset_t fromOfs, offset_t size, offset_t &done, offset_t &total, unsigned timeout)
  2592. {
  2593. // return 0 if continuing, 1 if done
  2594. CAsyncCopySection * job;
  2595. Linked<CAsyncJob> cjob;
  2596. {
  2597. CriticalBlock block(sect);
  2598. cjob.set(jobtable.find(uuid,false));
  2599. if (cjob) {
  2600. job = QUERYINTERFACE(cjob.get(),CAsyncCopySection);
  2601. if (!job) {
  2602. throw MakeStringException(-1,"Async job ID mismatch");
  2603. }
  2604. }
  2605. else {
  2606. job = new CAsyncCopySection(*this, uuid, fromFile, toFile, toOfs, fromOfs, size);
  2607. cjob.setown(job);
  2608. jobtable.add(cjob.getLink());
  2609. cjob->start();
  2610. }
  2611. }
  2612. AsyncCommandStatus ret = ACSerror;
  2613. Owned<IException> rete;
  2614. try {
  2615. ret = job->poll(done,total,timeout);
  2616. }
  2617. catch (IException * e) {
  2618. rete.setown(e);
  2619. }
  2620. if ((ret!=ACScontinue)||rete.get()) {
  2621. job->join();
  2622. CriticalBlock block(sect);
  2623. jobtable.remove(job);
  2624. if (rete.get())
  2625. throw rete.getClear();
  2626. }
  2627. return ret;
  2628. }
  2629. };
  2630. //====================================================================================================
  2631. inline void appendErr(MemoryBuffer &reply, RemoteFileCommandType e)
  2632. {
  2633. reply.append((unsigned)e).append(getRFCText(e));
  2634. }
  2635. inline void appendErr2(MemoryBuffer &reply, RemoteFileCommandType e, unsigned v)
  2636. {
  2637. StringBuffer msg;
  2638. msg.append(getRFCText(e)).append(':').append(v);
  2639. reply.append((unsigned)e).append(msg.str());
  2640. }
  2641. inline void appendErr3(MemoryBuffer &reply, RemoteFileCommandType e, int code, const char *errMsg)
  2642. {
  2643. StringBuffer msg;
  2644. msg.appendf("ERROR: %s(%d) '%s'", getRFCText(e), code, errMsg?errMsg:"");
  2645. // some errors are RemoteFileCommandType, some are RFSERR_*
  2646. // RFCOpenIO needs remapping to non-zero for client to know its an error
  2647. if ((RemoteFileCommandType)e == RFCopenIO)
  2648. e = (RemoteFileCommandType)RFSERR_OpenFailed;
  2649. reply.append((unsigned)e);
  2650. reply.append(msg.str());
  2651. }
  2652. #define MAPCOMMAND(c,p) case c: { ret = this->p(msg, reply) ; break; }
  2653. #define MAPCOMMANDCLIENT(c,p,client) case c: { ret = this->p(msg, reply, client); break; }
  2654. #define MAPCOMMANDCLIENTTHROTTLE(c,p,client,throttler) case c: { ret = this->p(msg, reply, client, throttler); break; }
  2655. #define MAPCOMMANDSTATS(c,p,stats) case c: { ret = this->p(msg, reply, stats); break; }
  2656. #define MAPCOMMANDCLIENTSTATS(c,p,client,stats) case c: { ret = this->p(msg, reply, client, stats); break; }
  2657. static unsigned ClientCount = 0;
  2658. static unsigned MaxClientCount = 0;
  2659. static CriticalSection ClientCountSect;
  2660. #define DEFAULT_THROTTLOG_LOG_INTERVAL_SECS 60 // log total throttled delay period
  2661. class CClientStats : public CInterface
  2662. {
  2663. SpinLock spin;
  2664. public:
  2665. CClientStats(const char *_client) : client(_client) { count = 0; bRead = 0; bWritten = 0; }
  2666. const char *queryFindString() const { return client; }
  2667. inline void addRead(unsigned len)
  2668. {
  2669. SpinBlock b(spin); // rare event, but we should change to a atomic<__int64> for >= c++11
  2670. bRead += len;
  2671. }
  2672. inline void addWrite(unsigned len)
  2673. {
  2674. SpinBlock b(spin); // rare event, but we should change to a atomic<__int64> for >= c++11
  2675. bWritten += len;
  2676. }
  2677. StringAttr client;
  2678. unsigned __int64 count;
  2679. unsigned __int64 bRead;
  2680. unsigned __int64 bWritten;
  2681. };
  2682. class CClientStatsTable : public OwningStringSuperHashTableOf<CClientStats>
  2683. {
  2684. typedef OwningStringSuperHashTableOf<CClientStats> PARENT;
  2685. CriticalSection crit;
  2686. unsigned cmdStats[RFCmax];
  2687. static int compareElement(void* const *ll, void* const *rr)
  2688. {
  2689. const CClientStats *l = (const CClientStats *) *ll;
  2690. const CClientStats *r = (const CClientStats *) *rr;
  2691. if (l->count == r->count)
  2692. return 0;
  2693. else if (l->count<r->count)
  2694. return 1;
  2695. else
  2696. return -1;
  2697. }
  2698. public:
  2699. CClientStatsTable()
  2700. {
  2701. memset(&cmdStats[0], 0, sizeof(cmdStats));
  2702. }
  2703. ~CClientStatsTable()
  2704. {
  2705. kill();
  2706. }
  2707. CClientStats *getClientReference(RemoteFileCommandType cmd, const char *client)
  2708. {
  2709. CriticalBlock b(crit);
  2710. CClientStats *stats = PARENT::find(client);
  2711. if (!stats)
  2712. {
  2713. stats = new CClientStats(client);
  2714. PARENT::replace(*stats);
  2715. }
  2716. if (cmd<RFCmax) // i.e. ignore duff command (which will be traced), but still record client connected
  2717. cmdStats[cmd]++;
  2718. ++stats->count;
  2719. return LINK(stats);
  2720. }
  2721. StringBuffer &getInfo(StringBuffer &info, unsigned level=1)
  2722. {
  2723. CriticalBlock b(crit);
  2724. unsigned __int64 totalCmds = 0;
  2725. for (unsigned c=0; c<RFCmax; c++)
  2726. totalCmds += cmdStats[c];
  2727. unsigned totalClients = PARENT::ordinality();
  2728. info.appendf("Commands processed = %" I64F "u, unique clients = %u", totalCmds, totalClients);
  2729. if (totalCmds)
  2730. {
  2731. info.append("Command stats:").newline();
  2732. for (unsigned c=0; c<RFCmax; c++)
  2733. {
  2734. unsigned __int64 count = cmdStats[c];
  2735. if (count)
  2736. info.append(getRFCText(c)).append(": ").append(count).newline();
  2737. }
  2738. }
  2739. if (totalClients)
  2740. {
  2741. SuperHashIteratorOf<CClientStats> iter(*this);
  2742. PointerArrayOf<CClientStats> elements;
  2743. ForEach(iter)
  2744. {
  2745. CClientStats &elem = iter.query();
  2746. elements.append(&elem);
  2747. }
  2748. elements.sort(&compareElement);
  2749. if (level < 10)
  2750. {
  2751. // list up to 10 clients ordered by # of commands processed
  2752. unsigned max=elements.ordinality();
  2753. if (max>10)
  2754. max = 10; // cap
  2755. info.append("Top 10 clients:").newline();
  2756. for (unsigned e=0; e<max; e++)
  2757. {
  2758. const CClientStats &element = *elements.item(e);
  2759. info.appendf("Client %s - %" I64F "d requests handled, bytes read = %" I64F "d, bytes written = % " I64F "d",
  2760. element.client.get(), element.count, element.bRead, element.bWritten).newline();
  2761. }
  2762. }
  2763. else // list all
  2764. {
  2765. info.append("All clients:").newline();
  2766. ForEachItemIn(e, elements)
  2767. {
  2768. const CClientStats &element = *elements.item(e);
  2769. info.appendf("Client %s - %" I64F "d requests handled, bytes read = %" I64F "d, bytes written = % " I64F "d",
  2770. element.client.get(), element.count, element.bRead, element.bWritten).newline();
  2771. }
  2772. }
  2773. }
  2774. return info;
  2775. }
  2776. void reset()
  2777. {
  2778. CriticalBlock b(crit);
  2779. memset(&cmdStats[0], 0, sizeof(cmdStats));
  2780. kill();
  2781. }
  2782. };
  2783. class CRemoteFileServer : public CInterface, implements IRemoteFileServer
  2784. {
  2785. class CThrottler;
  2786. class CRemoteClientHandler : public CInterface, implements ISocketSelectNotify
  2787. {
  2788. public:
  2789. CRemoteFileServer *parent;
  2790. Owned<ISocket> socket;
  2791. StringAttr peerName;
  2792. Owned<IAuthenticatedUser> user;
  2793. MemoryBuffer msg;
  2794. bool selecthandled;
  2795. size32_t left;
  2796. IArrayOf<IFileIO> openfiles; // kept in sync with handles
  2797. Owned<IDirectoryIterator> opendir;
  2798. StringAttrArray opennames; // for debug
  2799. IntArray handles;
  2800. unsigned lasttick, lastInactiveTick;
  2801. atomic_t &globallasttick;
  2802. unsigned previdx; // for debug
  2803. IMPLEMENT_IINTERFACE;
  2804. CRemoteClientHandler(CRemoteFileServer *_parent,ISocket *_socket,IAuthenticatedUser *_user,atomic_t &_globallasttick)
  2805. : socket(_socket), user(_user), globallasttick(_globallasttick)
  2806. {
  2807. previdx = (unsigned)-1;
  2808. StringBuffer peerBuf;
  2809. char name[256];
  2810. name[0] = 0;
  2811. int port = socket->peer_name(name,sizeof(name)-1);
  2812. if (port>=0)
  2813. {
  2814. peerBuf.append(name);
  2815. if (port)
  2816. peerBuf.append(':').append(port);
  2817. peerName.set(peerBuf);
  2818. }
  2819. else
  2820. {
  2821. /* There's a possibility the socket closed before got here, in which case, peer name is unavailable
  2822. * May potentially be unavailable for other reasons also.
  2823. * Must be set, as used in client stats HT.
  2824. * If socket closed, the handler will start up but notice closed and quit
  2825. */
  2826. peerName.set("UNKNOWN PEER NAME");
  2827. }
  2828. {
  2829. CriticalBlock block(ClientCountSect);
  2830. if (++ClientCount>MaxClientCount)
  2831. MaxClientCount = ClientCount;
  2832. if (TF_TRACE_CLIENT_CONN)
  2833. {
  2834. StringBuffer s;
  2835. s.appendf("Connecting(%p) [%d,%d] to ",this,ClientCount,MaxClientCount);
  2836. s.append(peerName);
  2837. PROGLOG("%s", s.str());
  2838. }
  2839. }
  2840. parent = _parent;
  2841. left = 0;
  2842. msg.setEndian(__BIG_ENDIAN);
  2843. selecthandled = false;
  2844. touch();
  2845. }
  2846. ~CRemoteClientHandler()
  2847. {
  2848. {
  2849. CriticalBlock block(ClientCountSect);
  2850. ClientCount--;
  2851. if (TF_TRACE_CLIENT_CONN) {
  2852. PROGLOG("Disconnecting(%p) [%d,%d] ",this,ClientCount,MaxClientCount);
  2853. }
  2854. }
  2855. ISocket *sock = socket.getClear();
  2856. try {
  2857. sock->Release();
  2858. }
  2859. catch (IException *e) {
  2860. EXCLOG(e,"~CRemoteClientHandler");
  2861. e->Release();
  2862. }
  2863. }
  2864. bool notifySelected(ISocket *sock,unsigned selected)
  2865. {
  2866. if (TF_TRACE_FULL)
  2867. PROGLOG("notifySelected(%p)",this);
  2868. if (sock!=socket)
  2869. WARNLOG("notifySelected - invalid socket passed");
  2870. size32_t avail = (size32_t)socket->avail_read();
  2871. if (avail)
  2872. touch();
  2873. if (left==0)
  2874. {
  2875. try
  2876. {
  2877. left = avail?receiveBufferSize(socket):0;
  2878. }
  2879. catch (IException *e)
  2880. {
  2881. EXCLOG(e,"notifySelected(1)");
  2882. e->Release();
  2883. left = 0;
  2884. }
  2885. if (left)
  2886. {
  2887. avail = (size32_t)socket->avail_read();
  2888. try
  2889. {
  2890. msg.ensureCapacity(left);
  2891. }
  2892. catch (IException *e)
  2893. {
  2894. EXCLOG(e,"notifySelected(2)");
  2895. e->Release();
  2896. left = 0;
  2897. // if too big then corrupted packet so read avail to try and consume
  2898. char fbuf[1024];
  2899. while (avail)
  2900. {
  2901. size32_t rd = avail>sizeof(fbuf)?sizeof(fbuf):avail;
  2902. try
  2903. {
  2904. socket->read(fbuf, rd); // don't need timeout here
  2905. avail -= rd;
  2906. }
  2907. catch (IException *e)
  2908. {
  2909. EXCLOG(e,"notifySelected(2) flush");
  2910. e->Release();
  2911. break;
  2912. }
  2913. }
  2914. avail = 0;
  2915. left = 0;
  2916. }
  2917. }
  2918. }
  2919. size32_t toread = left>avail?avail:left;
  2920. if (toread)
  2921. {
  2922. try
  2923. {
  2924. socket->read(msg.reserve(toread), toread); // don't need timeout here
  2925. }
  2926. catch (IException *e)
  2927. {
  2928. EXCLOG(e,"notifySelected(3)");
  2929. e->Release();
  2930. toread = left;
  2931. msg.clear();
  2932. }
  2933. }
  2934. if (TF_TRACE_FULL)
  2935. PROGLOG("notifySelected %d,%d",toread,left);
  2936. if ((left!=0)&&(avail==0))
  2937. {
  2938. WARNLOG("notifySelected: Closing mid packet, %d remaining", left);
  2939. toread = left;
  2940. msg.clear();
  2941. }
  2942. left -= toread;
  2943. if (left==0)
  2944. {
  2945. // DEBUG
  2946. parent->notify(this, msg); // consumes msg
  2947. }
  2948. return false;
  2949. }
  2950. void logPrevHandle()
  2951. {
  2952. if (previdx<opennames.ordinality())
  2953. PROGLOG("Previous handle(%d): %s",handles.item(previdx),opennames.item(previdx).text.get());
  2954. }
  2955. bool throttleCommand(MemoryBuffer &msg)
  2956. {
  2957. RemoteFileCommandType cmd = RFCunknown;
  2958. Owned<IException> e;
  2959. try
  2960. {
  2961. msg.read(cmd);
  2962. parent->throttleCommand(cmd, msg, this);
  2963. return true;
  2964. }
  2965. catch (IException *_e)
  2966. {
  2967. e.setown(_e);
  2968. }
  2969. /* processCommand() will handle most exception and replies,
  2970. * but if throttleCommand fails before it gets that far, this will handle
  2971. */
  2972. MemoryBuffer reply;
  2973. initSendBuffer(reply);
  2974. StringBuffer s;
  2975. e->errorMessage(s);
  2976. appendErr3(reply, cmd, e->errorCode(), s.str());
  2977. parent->appendError(cmd, this, cmd, reply);
  2978. sendBuffer(socket, reply);
  2979. return false;
  2980. }
  2981. void processCommand(RemoteFileCommandType cmd, MemoryBuffer &msg, CThrottler *throttler)
  2982. {
  2983. MemoryBuffer reply;
  2984. parent->processCommand(cmd, msg, initSendBuffer(reply), this, throttler);
  2985. sendBuffer(socket, reply);
  2986. }
  2987. bool immediateCommand() // returns false if socket closed or failure
  2988. {
  2989. MemoryBuffer msg;
  2990. msg.setEndian(__BIG_ENDIAN);
  2991. touch();
  2992. size32_t avail = (size32_t)socket->avail_read();
  2993. if (avail==0)
  2994. return false;
  2995. receiveBuffer(socket, msg, 5); // shouldn't timeout as data is available
  2996. touch();
  2997. if (msg.length()==0)
  2998. return false;
  2999. return throttleCommand(msg);
  3000. }
  3001. void process(MemoryBuffer &msg)
  3002. {
  3003. if (selecthandled)
  3004. throttleCommand(msg);
  3005. else
  3006. {
  3007. // msg only used/filled if process() has been triggered by notify()
  3008. while (parent->threadRunningCount()<=parent->targetActiveThreads) // if too many threads add to select handler
  3009. {
  3010. int w;
  3011. try
  3012. {
  3013. w = socket->wait_read(1000);
  3014. }
  3015. catch (IException *e)
  3016. {
  3017. EXCLOG(e, "CRemoteClientHandler::main wait_read error");
  3018. e->Release();
  3019. parent->onCloseSocket(this,1);
  3020. return;
  3021. }
  3022. if (w==0)
  3023. break;
  3024. if ((w<0)||!immediateCommand())
  3025. {
  3026. if (w<0)
  3027. WARNLOG("CRemoteClientHandler::main wait_read error");
  3028. parent->onCloseSocket(this,1);
  3029. return;
  3030. }
  3031. }
  3032. /* This is a bit confusing..
  3033. * The addClient below, adds this request to a selecthandler handled by another thread
  3034. * and passes ownership of 'this' (CRemoteClientHandler)
  3035. *
  3036. * When notified, the selecthandler will launch a new pool thread to handle the request
  3037. * If the pool thread limit is hit, the selecthandler will be blocked [ see comment in CRemoteFileServer::notify() ]
  3038. *
  3039. * Either way, a thread pool slot is occupied when processing a request.
  3040. * Blocked threads, will be blocked for up to 1 minute (as defined by createThreadPool call)
  3041. * IOW, if there are lots of incoming clients that can't be serviced by the CThrottler limit,
  3042. * a large number of pool threads will build up after a while.
  3043. *
  3044. * The CThrottler mechanism, imposes a further hard limit on how many concurrent request threads can be active.
  3045. * If the thread pool had an absolute limit (instead of just introducing a delay), then I don't see the point
  3046. * in this additional layer of throttling..
  3047. */
  3048. selecthandled = true;
  3049. parent->addClient(this); // add to select handler
  3050. // NB: this (CRemoteClientHandler) is now linked by the selecthandler and owned by the 'clients' list
  3051. }
  3052. }
  3053. bool timedOut()
  3054. {
  3055. return (msTick()-lasttick)>CLIENT_TIMEOUT;
  3056. }
  3057. bool inactiveTimedOut()
  3058. {
  3059. unsigned ms = msTick();
  3060. if ((ms-lastInactiveTick)>CLIENT_INACTIVEWARNING_TIMEOUT)
  3061. {
  3062. lastInactiveTick = ms;
  3063. return true;
  3064. }
  3065. return false;
  3066. }
  3067. void touch()
  3068. {
  3069. lastInactiveTick = lasttick = msTick();
  3070. atomic_set(&globallasttick,lasttick);
  3071. }
  3072. const char *queryPeerName()
  3073. {
  3074. return peerName;
  3075. }
  3076. bool getInfo(StringBuffer &str)
  3077. {
  3078. str.append("client(");
  3079. const char *name = queryPeerName();
  3080. bool ok;
  3081. if (name)
  3082. {
  3083. ok = true;
  3084. str.append(name);
  3085. }
  3086. else
  3087. ok = false;
  3088. unsigned ms = msTick();
  3089. str.appendf("): last touch %d ms ago (%d, %d)",ms-lasttick,lasttick,ms);
  3090. ForEachItemIn(i,handles) {
  3091. str.appendf("\n %d: ",handles.item(i));
  3092. str.append(opennames.item(i).text);
  3093. }
  3094. return ok;
  3095. }
  3096. };
  3097. class CThrottleQueueItem : public CSimpleInterface
  3098. {
  3099. public:
  3100. RemoteFileCommandType cmd;
  3101. Linked<CRemoteClientHandler> client;
  3102. MemoryBuffer msg;
  3103. CCycleTimer timer;
  3104. CThrottleQueueItem(RemoteFileCommandType _cmd, MemoryBuffer &_msg, CRemoteClientHandler *_client) : cmd(_cmd), client(_client)
  3105. {
  3106. msg.swapWith(_msg);
  3107. }
  3108. };
  3109. class CThrottler
  3110. {
  3111. CRemoteFileServer &owner;
  3112. Semaphore sem;
  3113. CriticalSection crit, configureCrit;
  3114. StringAttr title;
  3115. unsigned limit, delayMs, cpuThreshold, queueLimit;
  3116. unsigned disabledLimit;
  3117. unsigned __int64 totalThrottleDelay;
  3118. CCycleTimer totalThrottleDelayTimer;
  3119. QueueOf<CThrottleQueueItem, false> queue;
  3120. unsigned statsIntervalSecs;
  3121. public:
  3122. CThrottler(CRemoteFileServer &_owner, const char *_title) : owner(_owner), title(_title)
  3123. {
  3124. totalThrottleDelay = 0;
  3125. limit = 0;
  3126. delayMs = DEFAULT_STDCMD_THROTTLEDELAYMS;
  3127. cpuThreshold = DEFAULT_STDCMD_THROTTLECPULIMIT;
  3128. disabledLimit = 0;
  3129. queueLimit = DEFAULT_STDCMD_THROTTLEQUEUELIMIT;
  3130. statsIntervalSecs = DEFAULT_STDCMD_THROTTLECPULIMIT;
  3131. }
  3132. ~CThrottler()
  3133. {
  3134. loop
  3135. {
  3136. Owned<CThrottleQueueItem> item = queue.dequeue();
  3137. if (!item)
  3138. break;
  3139. }
  3140. }
  3141. unsigned queryLimit() const { return limit; }
  3142. unsigned queryDelayMs() const { return delayMs; };;
  3143. unsigned queryCpuThreshold() const { return cpuThreshold; }
  3144. unsigned queryQueueLimit() const { return queueLimit; }
  3145. StringBuffer &getInfoSummary(StringBuffer &info)
  3146. {
  3147. info.appendf("Throttler(%s) - limit=%u, delayMs=%u, cpuThreshold=%u, queueLimit=%u", title.get(), limit, delayMs, cpuThreshold, queueLimit).newline();
  3148. unsigned elapsedSecs = totalThrottleDelayTimer.elapsedMs()/1000;
  3149. time_t simple;
  3150. time(&simple);
  3151. simple -= elapsedSecs;
  3152. CDateTime dt;
  3153. dt.set(simple);
  3154. StringBuffer dateStr;
  3155. dt.getTimeString(dateStr, true);
  3156. info.appendf("Throttler(%s): statistics since %s", title.get(), dateStr.str()).newline();
  3157. info.appendf("Total delay of %0.2f seconds", ((double)totalThrottleDelay)/1000).newline();
  3158. info.appendf("Requests currently queued: %u", queue.ordinality());
  3159. return info;
  3160. }
  3161. void getInfo(StringBuffer &info)
  3162. {
  3163. CriticalBlock b(crit);
  3164. getInfoSummary(info).newline();
  3165. }
  3166. void configure(unsigned _limit, unsigned _delayMs, unsigned _cpuThreshold, unsigned _queueLimit)
  3167. {
  3168. if (_limit > THROTTLE_MAX_LIMIT || _delayMs > THROTTLE_MAX_DELAYMS || _cpuThreshold > THROTTLE_MAX_CPUTHRESHOLD || _queueLimit > THROTTLE_MAX_QUEUELIMIT)
  3169. throw MakeStringException(0, "Throttler(%s), rejecting configure command: limit=%u (max permitted=%u), delayMs=%u (max permitted=%u), cpuThreshold=%u (max permitted=%u), queueLimit=%u (max permitted=%u)",
  3170. title.str(), _limit, THROTTLE_MAX_LIMIT, _delayMs, THROTTLE_MAX_DELAYMS, _cpuThreshold,
  3171. THROTTLE_MAX_CPUTHRESHOLD, _queueLimit, THROTTLE_MAX_QUEUELIMIT);
  3172. CriticalBlock b(configureCrit);
  3173. int delta = 0;
  3174. if (_limit)
  3175. {
  3176. if (disabledLimit) // if transitioning from disabled to some throttling
  3177. {
  3178. assertex(0 == limit);
  3179. delta = _limit - disabledLimit; // + or -
  3180. disabledLimit = 0;
  3181. }
  3182. else
  3183. delta = _limit - limit; // + or -
  3184. }
  3185. else if (0 == disabledLimit)
  3186. {
  3187. PROGLOG("Throttler(%s): disabled, previous limit: %u", title.get(), limit);
  3188. /* disabling - set limit immediately to let all new transaction through.
  3189. * NB: the semaphore signals are not consumed in this case, because transactions could be waiting on it.
  3190. * Instead the existing 'limit' is kept in 'disabledLimit', so that if/when throttling is
  3191. * re-enabled, it is used as a basis for increasing or consuming the semaphore signal count.
  3192. */
  3193. disabledLimit = limit;
  3194. limit = 0;
  3195. }
  3196. if (delta > 0)
  3197. {
  3198. PROGLOG("Throttler(%s): Increasing limit from %u to %u", title.get(), limit, _limit);
  3199. sem.signal(delta);
  3200. limit = _limit;
  3201. // NB: If throttling was off, this doesn't effect transactions in progress, i.e. will only throttle new transactions coming in.
  3202. }
  3203. else if (delta < 0)
  3204. {
  3205. PROGLOG("Throttler(%s): Reducing limit from %u to %u", title.get(), limit, _limit);
  3206. // NB: This is not expected to take long
  3207. CCycleTimer timer;
  3208. while (delta < 0)
  3209. {
  3210. if (sem.wait(1000))
  3211. ++delta;
  3212. else
  3213. PROGLOG("Throttler(%s): Waited %0.2f seconds so far for up to a maximum of %u (previous limit) transactions to complete, %u completed", title.get(), ((double)timer.elapsedMs())/1000, limit, -delta);
  3214. }
  3215. limit = _limit;
  3216. // NB: doesn't include transactions in progress, i.e. will only throttle new transactions coming in.
  3217. }
  3218. if (_delayMs != delayMs)
  3219. {
  3220. PROGLOG("Throttler(%s): New delayMs=%u, previous: %u", title.get(), _delayMs, delayMs);
  3221. delayMs = _delayMs;
  3222. }
  3223. if (_cpuThreshold != cpuThreshold)
  3224. {
  3225. PROGLOG("Throttler(%s): New cpuThreshold=%u, previous: %u", title.get(), _cpuThreshold, cpuThreshold);
  3226. cpuThreshold = _cpuThreshold;
  3227. }
  3228. if (((unsigned)-1) != _queueLimit && _queueLimit != queueLimit)
  3229. {
  3230. PROGLOG("Throttler(%s): New queueLimit=%u%s, previous: %u", title.get(), _queueLimit, 0==_queueLimit?"(disabled)":"", queueLimit);
  3231. queueLimit = _queueLimit;
  3232. }
  3233. }
  3234. void setStatsInterval(unsigned _statsIntervalSecs)
  3235. {
  3236. if (_statsIntervalSecs != statsIntervalSecs)
  3237. {
  3238. PROGLOG("Throttler(%s): New statsIntervalSecs=%u, previous: %u", title.get(), _statsIntervalSecs, statsIntervalSecs);
  3239. statsIntervalSecs = _statsIntervalSecs;
  3240. }
  3241. }
  3242. void take(RemoteFileCommandType cmd) // cmd for info. only
  3243. {
  3244. loop
  3245. {
  3246. if (sem.wait(delayMs))
  3247. return;
  3248. PROGLOG("Throttler(%s): transaction delayed [cmd=%s]", title.get(), getRFCText(cmd));
  3249. }
  3250. }
  3251. void release()
  3252. {
  3253. sem.signal();
  3254. }
  3255. StringBuffer &getStats(StringBuffer &stats, bool reset)
  3256. {
  3257. CriticalBlock b(crit);
  3258. getInfoSummary(stats);
  3259. if (reset)
  3260. {
  3261. totalThrottleDelayTimer.reset();
  3262. totalThrottleDelay = 0;
  3263. }
  3264. return stats;
  3265. }
  3266. void addCommand(RemoteFileCommandType cmd, MemoryBuffer &msg, CRemoteClientHandler *client)
  3267. {
  3268. CCycleTimer timer;
  3269. Owned<IException> exception;
  3270. bool hadSem = true;
  3271. if (!sem.wait(delayMs))
  3272. {
  3273. CriticalBlock b(crit);
  3274. if (!sem.wait(0)) // check hasn't become available
  3275. {
  3276. unsigned cpu = getLatestCPUUsage();
  3277. if (getLatestCPUUsage()<cpuThreshold)
  3278. {
  3279. /* Allow to proceed, despite hitting throttle limit because CPU < threshold
  3280. * NB: The overall number of threads is still capped by the thread pool.
  3281. */
  3282. unsigned ms = timer.elapsedMs();
  3283. totalThrottleDelay += ms;
  3284. PROGLOG("Throttler(%s): transaction delayed [cmd=%s] for : %u milliseconds, proceeding as cpu(%u)<throttleCPULimit(%u)", title.get(), getRFCText(cmd), cpu, ms, cpuThreshold);
  3285. hadSem = false;
  3286. }
  3287. else
  3288. {
  3289. if (queueLimit && queue.ordinality()>=queueLimit)
  3290. throw MakeStringException(0, "Throttler(%s), the maxiumum number of items are queued (%u), rejecting new command[%s]", title.str(), queue.ordinality(), getRFCText(cmd));
  3291. queue.enqueue(new CThrottleQueueItem(cmd, msg, client)); // NB: takes over ownership of 'client' from running thread
  3292. PROGLOG("Throttler(%s): transaction delayed [cmd=%s], queuing (%u queueud), [client=%p, sock=%u]", title.get(), getRFCText(cmd), queue.ordinality(), client, client->socket->OShandle());
  3293. return;
  3294. }
  3295. }
  3296. }
  3297. /* Guarantee that sem is released.
  3298. * Should normally release on clean exit when queue is empty.
  3299. */
  3300. struct ReleaseSem
  3301. {
  3302. Semaphore *sem;
  3303. ReleaseSem(Semaphore *_sem) { sem = _sem; }
  3304. ~ReleaseSem() { if (sem) sem->signal(); }
  3305. } releaseSem(hadSem?&sem:NULL);
  3306. /* Whilst holding on this throttle slot (i.e. before signalling semaphore back), process
  3307. * queued items. NB: other threads that are finishing will do also.
  3308. * Queued items are processed 1st, then the current request, then anything that was queued when handling current request
  3309. * Throttle slot (semaphore) is only given back when no more to do.
  3310. */
  3311. Linked<CRemoteClientHandler> currentClient;
  3312. MemoryBuffer currentMsg;
  3313. unsigned ms;
  3314. loop
  3315. {
  3316. RemoteFileCommandType currentCmd;
  3317. {
  3318. CriticalBlock b(crit);
  3319. Owned<CThrottleQueueItem> item = queue.dequeue();
  3320. if (item)
  3321. {
  3322. currentCmd = item->cmd;
  3323. currentClient.setown(item->client.getClear());
  3324. currentMsg.swapWith(item->msg);
  3325. ms = item->timer.elapsedMs();
  3326. }
  3327. else
  3328. {
  3329. if (NULL == client) // previously handled and queue empty
  3330. {
  3331. /* Commands are only queued if semaphore is exhaused (checked inside crit)
  3332. * so only signal the semaphore inside the crit, after checking if there are no queued items
  3333. */
  3334. if (hadSem)
  3335. {
  3336. releaseSem.sem = NULL;
  3337. sem.signal();
  3338. }
  3339. break;
  3340. }
  3341. currentCmd = cmd;
  3342. currentClient.set(client); // process current request after dealing with queue
  3343. currentMsg.swapWith(msg);
  3344. ms = timer.elapsedMs();
  3345. client = NULL;
  3346. }
  3347. }
  3348. if (ms >= 1000)
  3349. {
  3350. if (ms>delayMs)
  3351. PROGLOG("Throttler(%s): transaction delayed [cmd=%s] for : %u seconds", title.get(), getRFCText(currentCmd), ms/1000);
  3352. }
  3353. {
  3354. CriticalBlock b(crit);
  3355. totalThrottleDelay += ms;
  3356. }
  3357. try
  3358. {
  3359. currentClient->processCommand(currentCmd, currentMsg, this);
  3360. }
  3361. catch (IException *e)
  3362. {
  3363. EXCLOG(e, "addCommand: processCommand failed");
  3364. e->Release();
  3365. }
  3366. }
  3367. }
  3368. };
  3369. // temporarily release a throttler slot
  3370. class CThrottleReleaseBlock
  3371. {
  3372. CThrottler &throttler;
  3373. RemoteFileCommandType cmd;
  3374. public:
  3375. CThrottleReleaseBlock(CThrottler &_throttler, RemoteFileCommandType _cmd) : throttler(_throttler), cmd(_cmd)
  3376. {
  3377. throttler.release();
  3378. }
  3379. ~CThrottleReleaseBlock()
  3380. {
  3381. throttler.take(cmd);
  3382. }
  3383. };
  3384. int lasthandle;
  3385. CriticalSection sect;
  3386. Owned<ISocket> acceptsock;
  3387. Owned<ISocket> rejectsock;//used to immediately reject nonsecure connection requests when in secure mode
  3388. Owned<ISocketSelectHandler> selecthandler;
  3389. Owned<IThreadPool> threads; // for commands
  3390. bool stopping;
  3391. unsigned clientcounttick;
  3392. unsigned closedclients;
  3393. CAsyncCommandManager asyncCommandManager;
  3394. CThrottler stdCmdThrottler, slowCmdThrottler;
  3395. CClientStatsTable clientStatsTable;
  3396. atomic_t globallasttick;
  3397. unsigned targetActiveThreads;
  3398. int getNextHandle()
  3399. {
  3400. // called in sect critical block
  3401. loop {
  3402. if (lasthandle==INT_MAX)
  3403. lasthandle = 1;
  3404. else
  3405. lasthandle++;
  3406. unsigned idx1;
  3407. unsigned idx2;
  3408. if (!findHandle(lasthandle,idx1,idx2))
  3409. return lasthandle;
  3410. }
  3411. }
  3412. bool findHandle(int handle,unsigned &clientidx,unsigned &handleidx)
  3413. {
  3414. // called in sect critical block
  3415. clientidx = (unsigned)-1;
  3416. handleidx = (unsigned)-1;
  3417. ForEachItemIn(i,clients) {
  3418. CRemoteClientHandler &client = clients.item(i);
  3419. ForEachItemIn(j,client.handles) {
  3420. if (client.handles.item(j)==handle) {
  3421. handleidx = j;
  3422. clientidx = i;
  3423. return true;
  3424. }
  3425. }
  3426. }
  3427. return false;
  3428. }
  3429. class cCommandProcessor: public CInterface, implements IPooledThread
  3430. {
  3431. Owned<CRemoteClientHandler> client;
  3432. MemoryBuffer msg;
  3433. public:
  3434. IMPLEMENT_IINTERFACE;
  3435. struct cCommandProcessorParams
  3436. {
  3437. cCommandProcessorParams() { msg.setEndian(__BIG_ENDIAN); }
  3438. CRemoteClientHandler *client;
  3439. MemoryBuffer msg;
  3440. };
  3441. void init(void *_params)
  3442. {
  3443. cCommandProcessorParams &params = *(cCommandProcessorParams *)_params;
  3444. client.setown(params.client);
  3445. msg.swapWith(params.msg);
  3446. }
  3447. void main()
  3448. {
  3449. // idea is that initially we process commands inline then pass over to select handler
  3450. try
  3451. {
  3452. client->process(msg);
  3453. }
  3454. catch (IException *e)
  3455. {
  3456. // suppress some errors
  3457. EXCLOG(e,"cCommandProcessor::main");
  3458. e->Release();
  3459. }
  3460. try
  3461. {
  3462. client.clear();
  3463. }
  3464. catch (IException *e)
  3465. {
  3466. // suppress some more errors clearing client
  3467. EXCLOG(e,"cCommandProcessor::main(2)");
  3468. e->Release();
  3469. }
  3470. }
  3471. bool stop()
  3472. {
  3473. return true;
  3474. }
  3475. bool canReuse()
  3476. {
  3477. return false; // want to free owned socket
  3478. }
  3479. };
  3480. IArrayOf<CRemoteClientHandler> clients;
  3481. class cImpersonateBlock
  3482. {
  3483. CRemoteClientHandler &client;
  3484. public:
  3485. cImpersonateBlock(CRemoteClientHandler &_client)
  3486. : client(_client)
  3487. {
  3488. if (client.user.get()) {
  3489. if (TF_TRACE)
  3490. PROGLOG("Impersonate user: %s",client.user->username());
  3491. client.user->impersonate();
  3492. }
  3493. }
  3494. ~cImpersonateBlock()
  3495. {
  3496. if (client.user.get()) {
  3497. if (TF_TRACE)
  3498. PROGLOG("Stop impersonating user: %s",client.user->username());
  3499. client.user->revert();
  3500. }
  3501. }
  3502. };
  3503. #define IMPERSONATE_USER(client) cImpersonateBlock ublock(client)
  3504. public:
  3505. IMPLEMENT_IINTERFACE
  3506. CRemoteFileServer(unsigned maxThreads, unsigned maxThreadsDelayMs, unsigned maxAsyncCopy)
  3507. : stdCmdThrottler(*this, "stdCmdThrotlter"), slowCmdThrottler(*this, "slowCmdThrotlter"), asyncCommandManager(maxAsyncCopy)
  3508. {
  3509. lasthandle = 0;
  3510. selecthandler.setown(createSocketSelectHandler(NULL));
  3511. stdCmdThrottler.configure(DEFAULT_STDCMD_PARALLELREQUESTLIMIT, DEFAULT_STDCMD_THROTTLEDELAYMS, DEFAULT_STDCMD_THROTTLECPULIMIT, DEFAULT_STDCMD_THROTTLEQUEUELIMIT);
  3512. slowCmdThrottler.configure(DEFAULT_SLOWCMD_PARALLELREQUESTLIMIT, DEFAULT_SLOWCMD_THROTTLEDELAYMS, DEFAULT_SLOWCMD_THROTTLECPULIMIT, DEFAULT_SLOWCMD_THROTTLEQUEUELIMIT);
  3513. unsigned targetMinThreads=maxThreads*20/100; // 20%
  3514. if (0 == targetMinThreads) targetMinThreads = 1;
  3515. targetActiveThreads=maxThreads*80/100; // 80%
  3516. if (0 == targetActiveThreads) targetActiveThreads = 1;
  3517. class CCommandFactory : public CSimpleInterfaceOf<IThreadFactory>
  3518. {
  3519. CRemoteFileServer &parent;
  3520. public:
  3521. CCommandFactory(CRemoteFileServer &_parent) : parent(_parent) { }
  3522. virtual IPooledThread *createNew()
  3523. {
  3524. return parent.createCommandProcessor();
  3525. }
  3526. };
  3527. Owned<IThreadFactory> factory = new CCommandFactory(*this); // NB: pool links factory, so takes ownership
  3528. threads.setown(createThreadPool("CRemoteFileServerPool", factory, NULL, maxThreads, maxThreadsDelayMs,
  3529. #ifdef __64BIT__
  3530. 0, // Unlimited stack size
  3531. #else
  3532. 0x10000,
  3533. #endif
  3534. INFINITE,targetMinThreads));
  3535. threads->setStartDelayTracing(60); // trace amount delayed every minute.
  3536. PROGLOG("CRemoteFileServer: maxThreads = %u, maxThreadsDelayMs = %u, maxAsyncCopy = %u", maxThreads, maxThreadsDelayMs, maxAsyncCopy);
  3537. stopping = false;
  3538. clientcounttick = msTick();
  3539. closedclients = 0;
  3540. atomic_set(&globallasttick,msTick());
  3541. }
  3542. ~CRemoteFileServer()
  3543. {
  3544. #ifdef _DEBUG
  3545. PROGLOG("Exiting CRemoteFileServer");
  3546. #endif
  3547. asyncCommandManager.join();
  3548. clients.kill();
  3549. #ifdef _DEBUG
  3550. PROGLOG("Exited CRemoteFileServer");
  3551. #endif
  3552. }
  3553. //MORE: The file handles should timeout after a while, and accessing an old (invalid handle)
  3554. // should throw a different exception
  3555. bool checkFileIOHandle(MemoryBuffer &reply, int handle, IFileIO *&fileio, bool del=false)
  3556. {
  3557. CriticalBlock block(sect);
  3558. fileio = NULL;
  3559. if (handle<=0) {
  3560. appendErr(reply, (RemoteFileCommandType)RFSERR_NullFileIOHandle);
  3561. return false;
  3562. }
  3563. unsigned clientidx;
  3564. unsigned handleidx;
  3565. if (findHandle(handle,clientidx,handleidx)) {
  3566. CRemoteClientHandler &client = clients.item(clientidx);
  3567. if (del) {
  3568. client.handles.remove(handleidx);
  3569. client.openfiles.remove(handleidx);
  3570. client.opennames.remove(handleidx);
  3571. client.previdx = (unsigned)-1;
  3572. }
  3573. else {
  3574. fileio = &client.openfiles.item(handleidx);
  3575. client.previdx = handleidx;
  3576. }
  3577. return true;
  3578. }
  3579. appendErr(reply, (RemoteFileCommandType)RFSERR_InvalidFileIOHandle);
  3580. return false;
  3581. }
  3582. void onCloseSocket(CRemoteClientHandler *client, int which)
  3583. {
  3584. if (!client)
  3585. return;
  3586. CriticalBlock block(sect);
  3587. #ifdef _DEBUG
  3588. StringBuffer s(client->queryPeerName());
  3589. PROGLOG("onCloseSocket(%d) %s",which,s.str());
  3590. #endif
  3591. if (client->socket)
  3592. {
  3593. try
  3594. {
  3595. /* JCSMORE - shouldn't this really be dependent on whether selecthandled=true
  3596. * It has not been added to the selecthandler
  3597. * Harmless, but wasteful if so.
  3598. */
  3599. selecthandler->remove(client->socket);
  3600. }
  3601. catch (IException *e) {
  3602. EXCLOG(e,"CRemoteFileServer::onCloseSocket.1");
  3603. e->Release();
  3604. }
  3605. }
  3606. try {
  3607. clients.zap(*client);
  3608. }
  3609. catch (IException *e) {
  3610. EXCLOG(e,"CRemoteFileServer::onCloseSocket.2");
  3611. e->Release();
  3612. }
  3613. }
  3614. bool cmdOpenFileIO(MemoryBuffer & msg, MemoryBuffer & reply, CRemoteClientHandler &client)
  3615. {
  3616. IMPERSONATE_USER(client);
  3617. Owned<StringAttrItem> name = new StringAttrItem;
  3618. byte mode;
  3619. byte share;
  3620. msg.read(name->text).read(mode).read(share);
  3621. // also try to recv extra byte
  3622. byte extra = 0;
  3623. if (msg.remaining() >= sizeof(byte))
  3624. msg.read(extra);
  3625. IFEflags extraFlags = (IFEflags)extra;
  3626. // none => nocache for remote (hint)
  3627. // can revert to previous behavior with conf file setting "allow_pgcache_flush=false"
  3628. if (extraFlags == IFEnone)
  3629. extraFlags = IFEnocache;
  3630. Owned<IFile> file = createIFile(name->text);
  3631. switch ((compatIFSHmode)share) {
  3632. case compatIFSHnone:
  3633. file->setCreateFlags(S_IRUSR|S_IWUSR);
  3634. file->setShareMode(IFSHnone);
  3635. break;
  3636. case compatIFSHread:
  3637. file->setShareMode(IFSHread);
  3638. break;
  3639. case compatIFSHwrite:
  3640. file->setShareMode(IFSHfull);
  3641. break;
  3642. case compatIFSHexec:
  3643. file->setCreateFlags(S_IRUSR|S_IWUSR|S_IXUSR|S_IRGRP|S_IXGRP|S_IROTH|S_IXOTH);
  3644. break;
  3645. case compatIFSHall:
  3646. file->setCreateFlags(S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH|S_IWOTH); // bit excessive
  3647. file->setShareMode(IFSHfull);
  3648. break;
  3649. }
  3650. if (TF_TRACE_PRE_IO)
  3651. PROGLOG("before open file '%s', (%d,%d,%d)",name->text.get(),(int)mode,(int)share,extraFlags);
  3652. IFileIO *fileio = file->open((IFOmode)mode,extraFlags);
  3653. int handle;
  3654. if (fileio) {
  3655. CriticalBlock block(sect);
  3656. handle = getNextHandle();
  3657. client.previdx = client.opennames.ordinality();
  3658. client.handles.append(handle);
  3659. client.openfiles.append(*fileio);
  3660. client.opennames.append(*name.getLink());
  3661. }
  3662. else
  3663. handle = 0;
  3664. reply.append(RFEnoerror);
  3665. reply.append(handle);
  3666. if (TF_TRACE)
  3667. PROGLOG("open file '%s', (%d,%d) handle = %d",name->text.get(),(int)mode,(int)share,handle);
  3668. return true;
  3669. }
  3670. bool cmdCloseFileIO(MemoryBuffer & msg, MemoryBuffer & reply)
  3671. {
  3672. int handle;
  3673. msg.read(handle);
  3674. IFileIO *fileio;
  3675. if (!checkFileIOHandle(reply, handle, fileio, true))
  3676. return false;
  3677. if (TF_TRACE)
  3678. PROGLOG("close file, handle = %d",handle);
  3679. reply.append(RFEnoerror);
  3680. return true;
  3681. }
  3682. bool cmdRead(MemoryBuffer & msg, MemoryBuffer & reply, CClientStats &stats)
  3683. {
  3684. int handle;
  3685. __int64 pos;
  3686. size32_t len;
  3687. msg.read(handle).read(pos).read(len);
  3688. IFileIO *fileio;
  3689. if (!checkFileIOHandle(reply, handle, fileio))
  3690. return false;
  3691. //arrange it so we read directly into the reply buffer...
  3692. unsigned posOfErr = reply.length();
  3693. reply.append((unsigned)RFEnoerror);
  3694. size32_t numRead;
  3695. unsigned posOfLength = reply.length();
  3696. if (TF_TRACE_PRE_IO)
  3697. PROGLOG("before read file, handle = %d, toread = %d",handle,len);
  3698. void * data;
  3699. {
  3700. reply.reserve(sizeof(numRead));
  3701. data = reply.reserve(len);
  3702. }
  3703. try {
  3704. numRead = fileio->read(pos,len,data);
  3705. }
  3706. catch (IException *e)
  3707. {
  3708. reply.setWritePos(posOfErr);
  3709. StringBuffer s;
  3710. e->errorMessage(s);
  3711. appendErr3(reply, (RemoteFileCommandType)RFSERR_ReadFailed,e->errorCode(),s.str());
  3712. e->Release();
  3713. return false;
  3714. }
  3715. stats.addRead(len);
  3716. if (TF_TRACE)
  3717. PROGLOG("read file, handle = %d, pos = %" I64F "d, toread = %d, read = %d",handle,pos,len,numRead);
  3718. {
  3719. reply.setLength(posOfLength + sizeof(numRead) + numRead);
  3720. reply.writeEndianDirect(posOfLength,sizeof(numRead),&numRead);
  3721. }
  3722. return true;
  3723. }
  3724. bool cmdSize(MemoryBuffer & msg, MemoryBuffer & reply)
  3725. {
  3726. int handle;
  3727. msg.read(handle);
  3728. IFileIO *fileio;
  3729. if (!checkFileIOHandle(reply, handle, fileio))
  3730. return false;
  3731. __int64 size = fileio->size();
  3732. reply.append((unsigned)RFEnoerror).append(size);
  3733. if (TF_TRACE)
  3734. PROGLOG("size file, handle = %d, size = %" I64F "d",handle,size);
  3735. return true;
  3736. }
  3737. bool cmdSetSize(MemoryBuffer & msg, MemoryBuffer & reply)
  3738. {
  3739. int handle;
  3740. offset_t size;
  3741. msg.read(handle).read(size);
  3742. IFileIO *fileio;
  3743. if (TF_TRACE)
  3744. PROGLOG("set size file, handle = %d, size = %" I64F "d",handle,size);
  3745. if (!checkFileIOHandle(reply, handle, fileio))
  3746. return false;
  3747. fileio->setSize(size);
  3748. reply.append((unsigned)RFEnoerror);
  3749. return true;
  3750. }
  3751. bool cmdWrite(MemoryBuffer & msg, MemoryBuffer & reply, CClientStats &stats)
  3752. {
  3753. int handle;
  3754. __int64 pos;
  3755. size32_t len;
  3756. msg.read(handle).read(pos).read(len);
  3757. IFileIO *fileio;
  3758. if (!checkFileIOHandle(reply, handle, fileio))
  3759. return false;
  3760. const byte *data = (const byte *)msg.readDirect(len);
  3761. if (TF_TRACE_PRE_IO)
  3762. PROGLOG("before write file, handle = %d, towrite = %d",handle,len);
  3763. size32_t numWritten = fileio->write(pos,len,data);
  3764. stats.addWrite(numWritten);
  3765. if (TF_TRACE)
  3766. PROGLOG("write file, handle = %d, towrite = %d, written = %d",handle,len,numWritten);
  3767. reply.append((unsigned)RFEnoerror).append(numWritten);
  3768. return true;
  3769. }
  3770. bool cmdExists(MemoryBuffer & msg, MemoryBuffer & reply, CRemoteClientHandler &client)
  3771. {
  3772. IMPERSONATE_USER(client);
  3773. StringAttr name;
  3774. msg.read(name);
  3775. if (TF_TRACE)
  3776. PROGLOG("exists, '%s'",name.get());
  3777. Owned<IFile> file=createIFile(name);
  3778. bool e = file->exists();
  3779. reply.append((unsigned)RFEnoerror).append(e);
  3780. return true;
  3781. }
  3782. bool cmdRemove(MemoryBuffer & msg, MemoryBuffer & reply,CRemoteClientHandler &client)
  3783. {
  3784. IMPERSONATE_USER(client);
  3785. StringAttr name;
  3786. msg.read(name);
  3787. if (TF_TRACE)
  3788. PROGLOG("remove, '%s'",name.get());
  3789. Owned<IFile> file=createIFile(name);
  3790. bool e = file->remove();
  3791. reply.append((unsigned)RFEnoerror).append(e);
  3792. return true;
  3793. }
  3794. bool cmdGetVer(MemoryBuffer & msg, MemoryBuffer & reply)
  3795. {
  3796. if (TF_TRACE)
  3797. PROGLOG("getVer");
  3798. if (msg.getPos()+sizeof(unsigned)>msg.length())
  3799. reply.append((unsigned)RFEnoerror);
  3800. else
  3801. reply.append((unsigned)FILESRV_VERSION+0x10000);
  3802. reply.append(VERSTRING);
  3803. return true;
  3804. }
  3805. bool cmdRename(MemoryBuffer & msg, MemoryBuffer & reply,CRemoteClientHandler &client)
  3806. {
  3807. IMPERSONATE_USER(client);
  3808. StringAttr fromname;
  3809. msg.read(fromname);
  3810. StringAttr toname;
  3811. msg.read(toname);
  3812. if (TF_TRACE)
  3813. PROGLOG("rename, '%s' to '%s'",fromname.get(),toname.get());
  3814. Owned<IFile> file=createIFile(fromname);
  3815. file->rename(toname);
  3816. reply.append((unsigned)RFEnoerror);
  3817. return true;
  3818. return false;
  3819. }
  3820. bool cmdMove(MemoryBuffer & msg, MemoryBuffer & reply,CRemoteClientHandler &client)
  3821. {
  3822. IMPERSONATE_USER(client);
  3823. StringAttr fromname;
  3824. msg.read(fromname);
  3825. StringAttr toname;
  3826. msg.read(toname);
  3827. if (TF_TRACE)
  3828. PROGLOG("move, '%s' to '%s'",fromname.get(),toname.get());
  3829. Owned<IFile> file=createIFile(fromname);
  3830. file->move(toname);
  3831. reply.append((unsigned)RFEnoerror);
  3832. return true;
  3833. return false;
  3834. }
  3835. bool cmdCopy(MemoryBuffer & msg, MemoryBuffer & reply, CRemoteClientHandler &client)
  3836. {
  3837. IMPERSONATE_USER(client);
  3838. StringAttr fromname;
  3839. msg.read(fromname);
  3840. StringAttr toname;
  3841. msg.read(toname);
  3842. if (TF_TRACE)
  3843. PROGLOG("copy, '%s' to '%s'",fromname.get(),toname.get());
  3844. copyFile(toname, fromname);
  3845. reply.append((unsigned)RFEnoerror);
  3846. return true;
  3847. }
  3848. bool cmdAppend(MemoryBuffer & msg, MemoryBuffer & reply, CRemoteClientHandler &client, CClientStats &stats)
  3849. {
  3850. IMPERSONATE_USER(client);
  3851. int handle;
  3852. __int64 pos;
  3853. __int64 len;
  3854. StringAttr srcname;
  3855. msg.read(handle).read(srcname).read(pos).read(len);
  3856. IFileIO *fileio;
  3857. if (!checkFileIOHandle(reply, handle, fileio))
  3858. return false;
  3859. Owned<IFile> file = createIFile(srcname.get());
  3860. __int64 written = fileio->appendFile(file,pos,len);
  3861. stats.addWrite(written);
  3862. if (TF_TRACE)
  3863. PROGLOG("append file, handle = %d, file=%s, pos = %" I64F "d len = %" I64F "d written = %" I64F "d",handle,srcname.get(),pos,len,written);
  3864. reply.append((unsigned)RFEnoerror).append(written);
  3865. return true;
  3866. }
  3867. bool cmdIsFile(MemoryBuffer &msg, MemoryBuffer &reply, CRemoteClientHandler &client)
  3868. {
  3869. IMPERSONATE_USER(client);
  3870. StringAttr name;
  3871. msg.read(name);
  3872. if (TF_TRACE)
  3873. PROGLOG("isFile, '%s'",name.get());
  3874. Owned<IFile> file=createIFile(name);
  3875. unsigned ret = (unsigned)file->isFile();
  3876. reply.append((unsigned)RFEnoerror).append(ret);
  3877. return true;
  3878. }
  3879. bool cmdIsDir(MemoryBuffer &msg, MemoryBuffer &reply, CRemoteClientHandler &client)
  3880. {
  3881. IMPERSONATE_USER(client);
  3882. StringAttr name;
  3883. msg.read(name);
  3884. if (TF_TRACE)
  3885. PROGLOG("isDir, '%s'",name.get());
  3886. Owned<IFile> file=createIFile(name);
  3887. unsigned ret = (unsigned)file->isDirectory();
  3888. reply.append((unsigned)RFEnoerror).append(ret);
  3889. return true;
  3890. }
  3891. bool cmdIsReadOnly(MemoryBuffer &msg, MemoryBuffer &reply, CRemoteClientHandler &client)
  3892. {
  3893. IMPERSONATE_USER(client);
  3894. StringAttr name;
  3895. msg.read(name);
  3896. if (TF_TRACE)
  3897. PROGLOG("isReadOnly, '%s'",name.get());
  3898. Owned<IFile> file=createIFile(name);
  3899. unsigned ret = (unsigned)file->isReadOnly();
  3900. reply.append((unsigned)RFEnoerror).append(ret);
  3901. return true;
  3902. }
  3903. bool cmdSetReadOnly(MemoryBuffer &msg, MemoryBuffer &reply, CRemoteClientHandler &client)
  3904. {
  3905. IMPERSONATE_USER(client);
  3906. StringAttr name;
  3907. bool set;
  3908. msg.read(name).read(set);
  3909. if (TF_TRACE)
  3910. PROGLOG("setReadOnly, '%s' %d",name.get(),(int)set);
  3911. Owned<IFile> file=createIFile(name);
  3912. file->setReadOnly(set);
  3913. reply.append((unsigned)RFEnoerror);
  3914. return true;
  3915. }
  3916. bool cmdSetFilePerms(MemoryBuffer &msg, MemoryBuffer &reply, CRemoteClientHandler &client)
  3917. {
  3918. IMPERSONATE_USER(client);
  3919. StringAttr name;
  3920. unsigned fPerms;
  3921. msg.read(name).read(fPerms);
  3922. if (TF_TRACE)
  3923. PROGLOG("setFilePerms, '%s' 0%o",name.get(),fPerms);
  3924. Owned<IFile> file=createIFile(name);
  3925. file->setFilePermissions(fPerms);
  3926. reply.append((unsigned)RFEnoerror);
  3927. return true;
  3928. }
  3929. bool cmdGetTime(MemoryBuffer &msg, MemoryBuffer &reply, CRemoteClientHandler &client)
  3930. {
  3931. IMPERSONATE_USER(client);
  3932. StringAttr name;
  3933. msg.read(name);
  3934. if (TF_TRACE)
  3935. PROGLOG("getTime, '%s'",name.get());
  3936. Owned<IFile> file=createIFile(name);
  3937. CDateTime createTime;
  3938. CDateTime modifiedTime;
  3939. CDateTime accessedTime;
  3940. bool ret = file->getTime(&createTime,&modifiedTime,&accessedTime);
  3941. reply.append((unsigned)RFEnoerror).append(ret);
  3942. if (ret) {
  3943. createTime.serialize(reply);
  3944. modifiedTime.serialize(reply);
  3945. accessedTime.serialize(reply);
  3946. }
  3947. return true;
  3948. }
  3949. bool cmdSetTime(MemoryBuffer &msg, MemoryBuffer &reply, CRemoteClientHandler &client)
  3950. {
  3951. IMPERSONATE_USER(client);
  3952. StringAttr name;
  3953. bool creategot;
  3954. CDateTime createTime;
  3955. bool modifiedgot;
  3956. CDateTime modifiedTime;
  3957. bool accessedgot;
  3958. CDateTime accessedTime;
  3959. msg.read(name);
  3960. msg.read(creategot);
  3961. if (creategot)
  3962. createTime.deserialize(msg);
  3963. msg.read(modifiedgot);
  3964. if (modifiedgot)
  3965. modifiedTime.deserialize(msg);
  3966. msg.read(accessedgot);
  3967. if (accessedgot)
  3968. accessedTime.deserialize(msg);
  3969. if (TF_TRACE)
  3970. PROGLOG("setTime, '%s'",name.get());
  3971. Owned<IFile> file=createIFile(name);
  3972. bool ret = file->setTime(creategot?&createTime:NULL,modifiedgot?&modifiedTime:NULL,accessedgot?&accessedTime:NULL);
  3973. reply.append((unsigned)RFEnoerror).append(ret);
  3974. return true;
  3975. }
  3976. bool cmdCreateDir(MemoryBuffer &msg, MemoryBuffer &reply, CRemoteClientHandler &client)
  3977. {
  3978. IMPERSONATE_USER(client);
  3979. StringAttr name;
  3980. msg.read(name);
  3981. if (TF_TRACE)
  3982. PROGLOG("CreateDir, '%s'",name.get());
  3983. Owned<IFile> dir=createIFile(name);
  3984. bool ret = dir->createDirectory();
  3985. reply.append((unsigned)RFEnoerror).append(ret);
  3986. return true;
  3987. }
  3988. bool cmdGetDir(MemoryBuffer &msg, MemoryBuffer &reply, CRemoteClientHandler &client)
  3989. {
  3990. IMPERSONATE_USER(client);
  3991. StringAttr name;
  3992. StringAttr mask;
  3993. bool includedir;
  3994. bool sub;
  3995. byte stream = 0;
  3996. msg.read(name).read(mask).read(includedir).read(sub);
  3997. if (msg.remaining()>=sizeof(byte)) {
  3998. msg.read(stream);
  3999. if (stream==1)
  4000. client.opendir.clear();
  4001. }
  4002. if (TF_TRACE)
  4003. PROGLOG("GetDir, '%s', '%s'",name.get(),mask.get());
  4004. Owned<IFile> dir=createIFile(name);
  4005. Owned<IDirectoryIterator> iter;
  4006. if (stream>1)
  4007. iter.set(client.opendir);
  4008. else {
  4009. iter.setown(dir->directoryFiles(mask.length()?mask.get():NULL,sub,includedir));
  4010. if (stream != 0)
  4011. client.opendir.set(iter);
  4012. }
  4013. if (!iter) {
  4014. reply.append((unsigned)RFSERR_GetDirFailed);
  4015. return false;
  4016. }
  4017. reply.append((unsigned)RFEnoerror);
  4018. if (CRemoteDirectoryIterator::serialize(reply,iter,stream?0x100000:0,stream<2)) {
  4019. if (stream != 0)
  4020. client.opendir.clear();
  4021. }
  4022. else {
  4023. bool cont=true;
  4024. reply.append(cont);
  4025. }
  4026. return true;
  4027. }
  4028. bool cmdMonitorDir(MemoryBuffer &msg, MemoryBuffer &reply, CRemoteClientHandler &client)
  4029. {
  4030. IMPERSONATE_USER(client);
  4031. StringAttr name;
  4032. StringAttr mask;
  4033. bool includedir;
  4034. bool sub;
  4035. unsigned checkinterval;
  4036. unsigned timeout;
  4037. __int64 cancelid; // not yet used
  4038. msg.read(name).read(mask).read(includedir).read(sub).read(checkinterval).read(timeout).read(cancelid);
  4039. byte isprev;
  4040. msg.read(isprev);
  4041. Owned<IDirectoryIterator> prev;
  4042. if (isprev==1) {
  4043. SocketEndpoint ep;
  4044. CRemoteDirectoryIterator *di = new CRemoteDirectoryIterator(ep,name);
  4045. di->appendBuf(msg);
  4046. prev.setown(di);
  4047. }
  4048. if (TF_TRACE)
  4049. PROGLOG("MonitorDir, '%s' '%s'",name.get(),mask.get());
  4050. Owned<IFile> dir=createIFile(name);
  4051. Owned<IDirectoryDifferenceIterator> iter=dir->monitorDirectory(prev,mask.length()?mask.get():NULL,sub,includedir,checkinterval,timeout);
  4052. reply.append((unsigned)RFEnoerror);
  4053. byte state = (iter.get()==NULL)?0:1;
  4054. reply.append(state);
  4055. if (state==1)
  4056. CRemoteDirectoryIterator::serializeDiff(reply,iter);
  4057. return true;
  4058. }
  4059. bool cmdCopySection(MemoryBuffer &msg, MemoryBuffer &reply, CRemoteClientHandler &client)
  4060. {
  4061. IMPERSONATE_USER(client);
  4062. StringAttr uuid;
  4063. StringAttr fromFile;
  4064. StringAttr toFile;
  4065. offset_t toOfs;
  4066. offset_t fromOfs;
  4067. offset_t size;
  4068. offset_t sizeDone=0;
  4069. offset_t totalSize=(offset_t)-1;
  4070. unsigned timeout;
  4071. msg.read(uuid).read(fromFile).read(toFile).read(toOfs).read(fromOfs).read(size).read(timeout);
  4072. AsyncCommandStatus status = asyncCommandManager.copySection(uuid,fromFile,toFile,toOfs,fromOfs,size,sizeDone,totalSize,timeout);
  4073. reply.append((unsigned)RFEnoerror).append((unsigned)status).append(sizeDone).append(totalSize);
  4074. return true;
  4075. }
  4076. static void treeCopyFile(RemoteFilename &srcfn, RemoteFilename &dstfn, const char *net, const char *mask, IpAddress &ip, bool usetmp, CThrottler *throttler, CFflags copyFlags=CFnone)
  4077. {
  4078. unsigned start = msTick();
  4079. Owned<IFile> dstfile = createIFile(dstfn);
  4080. // the following is really to check the dest node is up and working (otherwise not much point in continuing!)
  4081. if (dstfile->exists())
  4082. PROGLOG("TREECOPY overwriting '%s'",dstfile->queryFilename());
  4083. Owned<IFile> srcfile = createIFile(srcfn);
  4084. unsigned lastmin = 0;
  4085. if (!srcfn.queryIP().ipequals(dstfn.queryIP())) {
  4086. CriticalBlock block(treeCopyCrit);
  4087. loop {
  4088. CDateTime dt;
  4089. offset_t sz;
  4090. try {
  4091. sz = srcfile->size();
  4092. if (sz==(offset_t)-1) {
  4093. if (TF_TRACE_TREE_COPY)
  4094. PROGLOG("TREECOPY source not found '%s'",srcfile->queryFilename());
  4095. break;
  4096. }
  4097. srcfile->getTime(NULL,&dt,NULL);
  4098. }
  4099. catch (IException *e) {
  4100. EXCLOG(e,"treeCopyFile(1)");
  4101. e->Release();
  4102. break;
  4103. }
  4104. Linked<CTreeCopyItem> tc;
  4105. unsigned now = msTick();
  4106. ForEachItemInRev(i1,treeCopyArray) {
  4107. CTreeCopyItem &item = treeCopyArray.item(i1);
  4108. // prune old entries (not strictly needed buf I think better)
  4109. if (now-item.lastused>TREECOPYPRUNETIME)
  4110. treeCopyArray.remove(i1);
  4111. else if (!tc.get()&&item.equals(srcfn,net,mask,sz,dt)) {
  4112. tc.set(&item);
  4113. item.lastused = now;
  4114. }
  4115. }
  4116. if (!tc.get()) {
  4117. if (treeCopyArray.ordinality()>=TREECOPY_CACHE_SIZE)
  4118. treeCopyArray.remove(0);
  4119. tc.setown(new CTreeCopyItem(srcfn,net,mask,sz,dt));
  4120. treeCopyArray.append(*tc.getLink());
  4121. }
  4122. ForEachItemInRev(cand,tc->loc) { // rev to choose copied locations first (maybe optional?)
  4123. if (!tc->busy->testSet(cand)) {
  4124. // check file accessible and matches
  4125. if (!cand&&dstfn.equals(tc->loc.item(cand))) // hmm trying to overwrite existing, better humor
  4126. continue;
  4127. bool ok = true;
  4128. Owned<IFile> rmtfile = createIFile(tc->loc.item(cand));
  4129. if (cand) { // only need to check if remote
  4130. try {
  4131. if (rmtfile->size()!=sz)
  4132. ok = false;
  4133. else {
  4134. CDateTime fdt;
  4135. rmtfile->getTime(NULL,&fdt,NULL);
  4136. ok = fdt.equals(dt);
  4137. }
  4138. }
  4139. catch (IException *e) {
  4140. EXCLOG(e,"treeCopyFile(2)");
  4141. e->Release();
  4142. ok = false;
  4143. }
  4144. }
  4145. if (ok) { // if not ok leave 'busy'
  4146. // finally lets try and copy!
  4147. try {
  4148. if (TF_TRACE_TREE_COPY)
  4149. PROGLOG("TREECOPY(started) %s to %s",rmtfile->queryFilename(),dstfile->queryFilename());
  4150. {
  4151. CriticalUnblock unblock(treeCopyCrit); // note we have tc linked
  4152. rmtfile->copyTo(dstfile,DEFAULT_COPY_BLKSIZE,NULL,usetmp,copyFlags);
  4153. }
  4154. if (TF_TRACE_TREE_COPY)
  4155. PROGLOG("TREECOPY(done) %s to %s",rmtfile->queryFilename(),dstfile->queryFilename());
  4156. tc->busy->set(cand,false);
  4157. if (treeCopyWaiting)
  4158. treeCopySem.signal((treeCopyWaiting>1)?2:1);
  4159. // add to known locations
  4160. tc->busy->set(tc->loc.ordinality(),false); // prob already is clear
  4161. tc->loc.append(dstfn);
  4162. ip.ipset(tc->loc.item(cand).queryIP());
  4163. return;
  4164. }
  4165. catch (IException *e) {
  4166. if (cand==0) {
  4167. tc->busy->set(0,false); // don't leave busy
  4168. if (treeCopyWaiting)
  4169. treeCopySem.signal();
  4170. throw; // what more can we do!
  4171. }
  4172. EXCLOG(e,"treeCopyFile(3)");
  4173. e->Release();
  4174. }
  4175. }
  4176. }
  4177. }
  4178. // all locations busy
  4179. if (msTick()-start>TREECOPYTIMEOUT) {
  4180. WARNLOG("Treecopy %s wait timed out", srcfile->queryFilename());
  4181. break;
  4182. }
  4183. treeCopyWaiting++; // note this isn't precise - just indication
  4184. {
  4185. CriticalUnblock unblock(treeCopyCrit);
  4186. if (throttler)
  4187. {
  4188. CThrottleReleaseBlock block(*throttler, RFCtreecopy);
  4189. treeCopySem.wait(TREECOPYPOLLTIME);
  4190. }
  4191. else
  4192. treeCopySem.wait(TREECOPYPOLLTIME);
  4193. }
  4194. treeCopyWaiting--;
  4195. if ((msTick()-start)/10*1000!=lastmin) {
  4196. lastmin = (msTick()-start)/10*1000;
  4197. PROGLOG("treeCopyFile delayed: %s to %s",srcfile->queryFilename(),dstfile->queryFilename());
  4198. }
  4199. }
  4200. }
  4201. else if (TF_TRACE_TREE_COPY)
  4202. PROGLOG("TREECOPY source on same node as destination");
  4203. if (TF_TRACE_TREE_COPY)
  4204. PROGLOG("TREECOPY(started,fallback) %s to %s",srcfile->queryFilename(),dstfile->queryFilename());
  4205. try {
  4206. GetHostIp(ip);
  4207. srcfile->copyTo(dstfile,DEFAULT_COPY_BLKSIZE,NULL,usetmp,copyFlags);
  4208. }
  4209. catch (IException *e) {
  4210. EXCLOG(e,"TREECOPY(done,fallback)");
  4211. throw;
  4212. }
  4213. if (TF_TRACE_TREE_COPY)
  4214. PROGLOG("TREECOPY(done,fallback) %s to %s",srcfile->queryFilename(),dstfile->queryFilename());
  4215. }
  4216. bool cmdTreeCopy(MemoryBuffer &msg, MemoryBuffer &reply, CRemoteClientHandler &client, CThrottler *throttler, bool usetmp=false)
  4217. {
  4218. IMPERSONATE_USER(client);
  4219. RemoteFilename src;
  4220. src.deserialize(msg);
  4221. RemoteFilename dst;
  4222. dst.deserialize(msg);
  4223. StringAttr net;
  4224. StringAttr mask;
  4225. msg.read(net).read(mask);
  4226. IpAddress ip;
  4227. treeCopyFile(src,dst,net,mask,ip,usetmp,throttler);
  4228. unsigned status = 0;
  4229. reply.append((unsigned)RFEnoerror).append((unsigned)status);
  4230. ip.ipserialize(reply);
  4231. return true;
  4232. }
  4233. bool cmdTreeCopyTmp(MemoryBuffer &msg, MemoryBuffer &reply, CRemoteClientHandler &client, CThrottler *throttler)
  4234. {
  4235. return cmdTreeCopy(msg, reply, client, throttler, true);
  4236. }
  4237. bool cmdGetCRC(MemoryBuffer &msg, MemoryBuffer &reply, CRemoteClientHandler &client)
  4238. {
  4239. IMPERSONATE_USER(client);
  4240. StringAttr name;
  4241. msg.read(name);
  4242. if (TF_TRACE)
  4243. PROGLOG("getCRC, '%s'",name.get());
  4244. Owned<IFile> file=createIFile(name);
  4245. unsigned ret = file->getCRC();
  4246. reply.append((unsigned)RFEnoerror).append(ret);
  4247. return true;
  4248. }
  4249. bool cmdStop(MemoryBuffer &msg, MemoryBuffer &reply)
  4250. {
  4251. PROGLOG("Abort request received");
  4252. stopping = true;
  4253. if (acceptsock)
  4254. acceptsock->cancel_accept();
  4255. if (rejectsock)
  4256. rejectsock->cancel_accept();
  4257. reply.append((unsigned)RFEnoerror);
  4258. return false;
  4259. }
  4260. bool cmdExec(MemoryBuffer &msg, MemoryBuffer &reply, CRemoteClientHandler &client)
  4261. {
  4262. StringAttr cmdline;
  4263. StringAttr workdir;
  4264. bool sync;
  4265. bool hasoutput;
  4266. size32_t insize;
  4267. MemoryAttr inbuf;
  4268. msg.read(cmdline).read(workdir).read(sync).read(hasoutput).read(insize);
  4269. if (insize)
  4270. msg.read(insize, inbuf.allocate(insize));
  4271. Owned<IPipeProcess> pipe = createPipeProcess();
  4272. int retcode=-1;
  4273. HANDLE phandle=(HANDLE)0;
  4274. MemoryBuffer outbuf;
  4275. if (pipe->run("EXEC",cmdline,workdir,insize!=0,hasoutput)) {
  4276. if (insize) {
  4277. pipe->write(insize, inbuf.get());
  4278. pipe->closeInput();
  4279. }
  4280. if (hasoutput) {
  4281. byte buf[4096];
  4282. loop {
  4283. size32_t read = pipe->read(sizeof(buf),buf);
  4284. if (!read)
  4285. break;
  4286. outbuf.append(read,buf);
  4287. }
  4288. }
  4289. if (sync)
  4290. retcode = pipe->wait();
  4291. else {
  4292. phandle = pipe->getProcessHandle();
  4293. retcode = 0;
  4294. }
  4295. }
  4296. size32_t outsz = outbuf.length();
  4297. reply.append(retcode).append((unsigned)phandle).append(outsz);
  4298. if (outsz)
  4299. reply.append(outbuf);
  4300. return true;
  4301. }
  4302. bool cmdSetTrace(MemoryBuffer &msg, MemoryBuffer &reply)
  4303. {
  4304. byte flags;
  4305. msg.read(flags);
  4306. int retcode=-1;
  4307. if (flags!=255) { // escape
  4308. retcode = traceFlags;
  4309. traceFlags = flags;
  4310. }
  4311. reply.append(retcode);
  4312. return true;
  4313. }
  4314. bool cmdGetInfo(MemoryBuffer &msg, MemoryBuffer &reply)
  4315. {
  4316. unsigned level=1;
  4317. if (msg.remaining() >= sizeof(unsigned))
  4318. msg.read(level);
  4319. StringBuffer retstr;
  4320. getInfo(retstr, level);
  4321. reply.append(0).append(retstr.str());
  4322. return true;
  4323. }
  4324. bool cmdFirewall(MemoryBuffer &msg, MemoryBuffer &reply)
  4325. {
  4326. // TBD
  4327. StringBuffer retstr;
  4328. getInfo(retstr);
  4329. reply.append(0).append(retstr.str());
  4330. return true;
  4331. }
  4332. bool cmdExtractBlobElements(MemoryBuffer &msg, MemoryBuffer &reply, CRemoteClientHandler &client)
  4333. {
  4334. IMPERSONATE_USER(client);
  4335. StringAttr prefix;
  4336. StringAttr filename;
  4337. msg.read(prefix).read(filename);
  4338. RemoteFilename rfn;
  4339. rfn.setLocalPath(filename);
  4340. ExtractedBlobArray extracted;
  4341. extractBlobElements(prefix, rfn, extracted);
  4342. unsigned n = extracted.ordinality();
  4343. reply.append((unsigned)RFEnoerror).append(n);
  4344. for (unsigned i=0;i<n;i++)
  4345. extracted.item(i).serialize(reply);
  4346. return true;
  4347. }
  4348. bool cmdRedeploy(MemoryBuffer &msg, MemoryBuffer &reply)
  4349. {
  4350. return false; // TBD
  4351. }
  4352. bool cmdKill(MemoryBuffer & msg, MemoryBuffer & reply)
  4353. {
  4354. // TBD
  4355. appendErr2(reply, (RemoteFileCommandType)RFSERR_InvalidCommand, RFCkill);
  4356. return false;
  4357. }
  4358. bool cmdUnknown(MemoryBuffer & msg, MemoryBuffer & reply,RemoteFileCommandType cmd)
  4359. {
  4360. appendErr2(reply, (RemoteFileCommandType)RFSERR_InvalidCommand, cmd);
  4361. return false;
  4362. }
  4363. bool cmdUnlock(MemoryBuffer & msg, MemoryBuffer & reply,CRemoteClientHandler &client)
  4364. {
  4365. // this is an attempt to authenticate when we haven't got authentication turned on
  4366. if (TF_TRACE_CLIENT_STATS)
  4367. {
  4368. StringBuffer s(client.queryPeerName());
  4369. PROGLOG("Connect from %s",s.str());
  4370. }
  4371. appendErr2(reply, (RemoteFileCommandType)RFSERR_InvalidCommand, RFCunlock);
  4372. return false;
  4373. }
  4374. // legacy version
  4375. bool cmdSetThrottle(MemoryBuffer & msg, MemoryBuffer & reply)
  4376. {
  4377. unsigned limit, delayMs, cpuThreshold;
  4378. msg.read(limit);
  4379. msg.read(delayMs);
  4380. msg.read(cpuThreshold);
  4381. stdCmdThrottler.configure(limit, delayMs, cpuThreshold, (unsigned)-1);
  4382. reply.append((unsigned)RFEnoerror);
  4383. return true;
  4384. }
  4385. bool cmdSetThrottle2(MemoryBuffer & msg, MemoryBuffer & reply)
  4386. {
  4387. unsigned throttleClass, limit, delayMs, cpuThreshold, queueLimit;
  4388. msg.read(throttleClass);
  4389. msg.read(limit);
  4390. msg.read(delayMs);
  4391. msg.read(cpuThreshold);
  4392. msg.read(queueLimit);
  4393. setThrottle((ThrottleClass)throttleClass, limit, delayMs, cpuThreshold, queueLimit);
  4394. reply.append((unsigned)RFEnoerror);
  4395. return true;
  4396. }
  4397. void appendError(RemoteFileCommandType cmd, CRemoteClientHandler *client, unsigned ret, MemoryBuffer &reply)
  4398. {
  4399. if (reply.length()>=sizeof(unsigned)*2)
  4400. {
  4401. reply.reset();
  4402. unsigned z;
  4403. unsigned e;
  4404. reply.read(z).read(e);
  4405. StringBuffer err("ERR(");
  4406. err.append(e).append(") ");
  4407. if (client)
  4408. {
  4409. const char *peer = client->queryPeerName();
  4410. if (peer)
  4411. err.append(peer);
  4412. }
  4413. if (e&&(reply.getPos()<reply.length()))
  4414. {
  4415. StringAttr es;
  4416. reply.read(es);
  4417. err.append(" : ").append(es);
  4418. }
  4419. reply.reset();
  4420. if (cmd!=RFCunlock)
  4421. PROGLOG("%s",err.str()); // supress authentication logging
  4422. if (client)
  4423. client->logPrevHandle();
  4424. }
  4425. }
  4426. void throttleCommand(RemoteFileCommandType cmd, MemoryBuffer &msg, CRemoteClientHandler *client)
  4427. {
  4428. switch(cmd)
  4429. {
  4430. case RFCexec:
  4431. case RFCgetcrc:
  4432. case RFCcopy:
  4433. case RFCappend:
  4434. case RFCtreecopy:
  4435. case RFCtreecopytmp:
  4436. slowCmdThrottler.addCommand(cmd, msg, client);
  4437. return;
  4438. case RFCcloseIO:
  4439. case RFCopenIO:
  4440. case RFCread:
  4441. case RFCsize:
  4442. case RFCwrite:
  4443. case RFCexists:
  4444. case RFCremove:
  4445. case RFCrename:
  4446. case RFCgetver:
  4447. case RFCisfile:
  4448. case RFCisdirectory:
  4449. case RFCisreadonly:
  4450. case RFCsetreadonly:
  4451. case RFCsetfileperms:
  4452. case RFCgettime:
  4453. case RFCsettime:
  4454. case RFCcreatedir:
  4455. case RFCgetdir:
  4456. case RFCmonitordir:
  4457. case RFCstop:
  4458. case RFCextractblobelements:
  4459. case RFCkill:
  4460. case RFCredeploy:
  4461. case RFCmove:
  4462. case RFCsetsize:
  4463. case RFCsettrace:
  4464. case RFCgetinfo:
  4465. case RFCfirewall:
  4466. case RFCunlock:
  4467. stdCmdThrottler.addCommand(cmd, msg, client);
  4468. return;
  4469. // NB: The following commands are still bound by the the thread pool
  4470. case RFCsetthrottle: // legacy version
  4471. case RFCsetthrottle2:
  4472. case RFCcopysection: // slightly odd, but has it's own limit
  4473. default:
  4474. {
  4475. client->processCommand(cmd, msg, NULL);
  4476. break;
  4477. }
  4478. }
  4479. }
  4480. bool processCommand(RemoteFileCommandType cmd, MemoryBuffer & msg, MemoryBuffer & reply, CRemoteClientHandler *client, CThrottler *throttler)
  4481. {
  4482. Owned<CClientStats> stats = clientStatsTable.getClientReference(cmd, client->queryPeerName());
  4483. bool ret = true;
  4484. try
  4485. {
  4486. switch(cmd)
  4487. {
  4488. MAPCOMMANDSTATS(RFCread, cmdRead, *stats);
  4489. MAPCOMMANDSTATS(RFCwrite, cmdWrite, *stats);
  4490. MAPCOMMANDCLIENTSTATS(RFCappend, cmdAppend, *client, *stats);
  4491. MAPCOMMAND(RFCcloseIO, cmdCloseFileIO);
  4492. MAPCOMMANDCLIENT(RFCopenIO, cmdOpenFileIO, *client);
  4493. MAPCOMMAND(RFCsize, cmdSize);
  4494. MAPCOMMANDCLIENT(RFCexists, cmdExists, *client);
  4495. MAPCOMMANDCLIENT(RFCremove, cmdRemove, *client);
  4496. MAPCOMMANDCLIENT(RFCrename, cmdRename, *client);
  4497. MAPCOMMAND(RFCgetver, cmdGetVer);
  4498. MAPCOMMANDCLIENT(RFCisfile, cmdIsFile, *client);
  4499. MAPCOMMANDCLIENT(RFCisdirectory, cmdIsDir, *client);
  4500. MAPCOMMANDCLIENT(RFCisreadonly, cmdIsReadOnly, *client);
  4501. MAPCOMMANDCLIENT(RFCsetreadonly, cmdSetReadOnly, *client);
  4502. MAPCOMMANDCLIENT(RFCsetfileperms, cmdSetFilePerms, *client);
  4503. MAPCOMMANDCLIENT(RFCgettime, cmdGetTime, *client);
  4504. MAPCOMMANDCLIENT(RFCsettime, cmdSetTime, *client);
  4505. MAPCOMMANDCLIENT(RFCcreatedir, cmdCreateDir, *client);
  4506. MAPCOMMANDCLIENT(RFCgetdir, cmdGetDir, *client);
  4507. MAPCOMMANDCLIENT(RFCmonitordir, cmdMonitorDir, *client);
  4508. MAPCOMMAND(RFCstop, cmdStop);
  4509. MAPCOMMANDCLIENT(RFCexec, cmdExec, *client);
  4510. MAPCOMMANDCLIENT(RFCextractblobelements, cmdExtractBlobElements, *client);
  4511. MAPCOMMAND(RFCkill, cmdKill);
  4512. MAPCOMMAND(RFCredeploy, cmdRedeploy); // only Windows
  4513. MAPCOMMANDCLIENT(RFCgetcrc, cmdGetCRC, *client);
  4514. MAPCOMMANDCLIENT(RFCmove, cmdMove, *client);
  4515. MAPCOMMANDCLIENT(RFCcopy, cmdCopy, *client);
  4516. MAPCOMMAND(RFCsetsize, cmdSetSize);
  4517. MAPCOMMAND(RFCsettrace, cmdSetTrace);
  4518. MAPCOMMAND(RFCgetinfo, cmdGetInfo);
  4519. MAPCOMMAND(RFCfirewall, cmdFirewall);
  4520. MAPCOMMANDCLIENT(RFCunlock, cmdUnlock, *client);
  4521. MAPCOMMANDCLIENT(RFCcopysection, cmdCopySection, *client);
  4522. MAPCOMMANDCLIENTTHROTTLE(RFCtreecopy, cmdTreeCopy, *client, &slowCmdThrottler);
  4523. MAPCOMMANDCLIENTTHROTTLE(RFCtreecopytmp, cmdTreeCopyTmp, *client, &slowCmdThrottler);
  4524. MAPCOMMAND(RFCsetthrottle, cmdSetThrottle); // legacy version
  4525. MAPCOMMAND(RFCsetthrottle2, cmdSetThrottle2);
  4526. default:
  4527. ret = cmdUnknown(msg,reply,cmd);
  4528. break;
  4529. }
  4530. }
  4531. catch (IException *e)
  4532. {
  4533. ret = false;
  4534. StringBuffer s;
  4535. e->errorMessage(s);
  4536. appendErr3(reply, cmd, e->errorCode(), s.str());
  4537. e->Release();
  4538. }
  4539. if (!ret) // append error string
  4540. appendError(cmd, client, cmd, reply);
  4541. return ret;
  4542. }
  4543. IPooledThread *createCommandProcessor()
  4544. {
  4545. return new cCommandProcessor();
  4546. }
  4547. void run(SocketEndpoint &listenep, bool useSSL)
  4548. {
  4549. if (listenep.isNull())
  4550. acceptsock.setown(ISocket::create(listenep.port));
  4551. else {
  4552. StringBuffer ips;
  4553. listenep.getIpText(ips);
  4554. acceptsock.setown(ISocket::create_ip(listenep.port,ips.str()));
  4555. }
  4556. if (useSSL) {
  4557. if (!securitySettings.certificate)
  4558. throw createDafsException(DAFSERR_connection_failed,"SSL Certificate information not found in environment.conf");
  4559. if (listenep.port <= 0)
  4560. {
  4561. assertex(FALSE);//should never get here
  4562. listenep.port = securitySettings.daFileSrvPort;
  4563. }
  4564. //Create unsecure socket to reject non-ssl client requests
  4565. if (listenep.isNull())
  4566. rejectsock.setown(ISocket::create(DAFILESRV_PORT));
  4567. else
  4568. {
  4569. StringBuffer ips;
  4570. listenep.getIpText(ips);
  4571. rejectsock.setown(ISocket::create_ip(DAFILESRV_PORT,ips.str()));
  4572. }
  4573. }
  4574. #ifdef _DEBUG
  4575. StringBuffer sb;
  4576. listenep.getUrlStr(sb);
  4577. DBGLOG("Server accepting %sfrom %s", useSSL?"SECURE ":"", sb.str());
  4578. #endif
  4579. selecthandler->start();
  4580. UnsignedArray readSocks;
  4581. if (useSSL)
  4582. {
  4583. readSocks.append(acceptsock->OShandle());
  4584. readSocks.append(rejectsock->OShandle());
  4585. }
  4586. loop {
  4587. Owned<ISocket> sock;
  4588. bool sockavail = false;
  4589. try {
  4590. if (!useSSL)
  4591. sockavail = acceptsock->wait_read(1000*60*1)!=0;
  4592. else
  4593. {
  4594. UnsignedArray waitingSocks;
  4595. //SSL Enabled. Listen for non SSL connection on DAFILESRV_PORT and reject them
  4596. int numReady = wait_read_multiple(readSocks, 1000*60*1, waitingSocks);
  4597. if (numReady)
  4598. {
  4599. for (int idx = 0; idx < numReady; idx++)
  4600. {
  4601. if (waitingSocks.item(idx) == rejectsock->OShandle())
  4602. {
  4603. //Unsecure connection attemped, reject !
  4604. Owned<ISocket> s;
  4605. s.setown(rejectsock->accept(true));
  4606. IpAddress ip;
  4607. StringBuffer sb;
  4608. s->getPeerAddress(ip);
  4609. ip.getIpText(sb);
  4610. DBGLOG("Rejecting nonsecure connect from %s",sb.str());
  4611. s->close();
  4612. }
  4613. else
  4614. {
  4615. sockavail = true;
  4616. }
  4617. }
  4618. }
  4619. }
  4620. #if 0
  4621. if (!sockavail) {
  4622. JSocketStatistics stats;
  4623. getSocketStatistics(stats);
  4624. StringBuffer s;
  4625. getSocketStatisticsString(stats,s);
  4626. PROGLOG( "Socket statistics : \n%s\n",s.str());
  4627. }
  4628. #endif
  4629. }
  4630. catch (IException *e) {
  4631. EXCLOG(e,"CRemoteFileServer(1)");
  4632. e->Release();
  4633. // not sure what to do so just accept
  4634. sockavail = true;
  4635. }
  4636. if (stopping)
  4637. break;
  4638. if (sockavail) {
  4639. try {
  4640. sock.setown(acceptsock->accept(true));
  4641. if (useSSL)
  4642. {
  4643. Owned<ISecureSocket> ssock = createSecureSocket(sock.getClear(), ServerSocket);
  4644. int status = ssock->secure_accept();
  4645. if (status < 0)
  4646. throw createDafsException(DAFSERR_connection_failed,"Failure to establish secure connection");
  4647. sock.setown(ssock.getLink());
  4648. }
  4649. if (!sock||stopping)
  4650. break;
  4651. runClient(sock.getClear());
  4652. }
  4653. catch (IException *e) {
  4654. EXCLOG(e,"CRemoteFileServer");
  4655. e->Release();
  4656. sock.clear();
  4657. if (!QUERYINTERFACE(e, IJSOCK_Exception))
  4658. break;
  4659. }
  4660. }
  4661. else
  4662. checkTimeout();
  4663. }
  4664. if (TF_TRACE_CLIENT_STATS)
  4665. PROGLOG("CRemoteFileServer:run exiting");
  4666. selecthandler->stop(true);
  4667. }
  4668. void processUnauthenticatedCommand(RemoteFileCommandType cmd, ISocket *socket, MemoryBuffer &msg)
  4669. {
  4670. // these are unauthenticated commands
  4671. if (cmd != RFCgetver)
  4672. cmd = RFCinvalid;
  4673. MemoryBuffer reply;
  4674. processCommand(cmd, msg, initSendBuffer(reply), NULL, NULL);
  4675. sendBuffer(socket, reply);
  4676. }
  4677. bool checkAuthentication(ISocket *socket, IAuthenticatedUser *&ret)
  4678. {
  4679. ret = NULL;
  4680. if (!AuthenticationEnabled)
  4681. return true;
  4682. MemoryBuffer reqbuf;
  4683. MemoryBuffer reply;
  4684. MemoryBuffer encbuf; // because aesEncrypt clears input
  4685. initSendBuffer(reply);
  4686. receiveBuffer(socket, reqbuf, 1);
  4687. RemoteFileCommandType typ=0;
  4688. if (reqbuf.remaining()<sizeof(RemoteFileCommandType))
  4689. return false;
  4690. reqbuf.read(typ);
  4691. if (typ!=RFCunlock) {
  4692. processUnauthenticatedCommand(typ,socket,reqbuf);
  4693. return false;
  4694. }
  4695. if (reqbuf.remaining()<sizeof(OnceKey))
  4696. return false;
  4697. OnceKey oncekey;
  4698. reqbuf.read(sizeof(oncekey),&oncekey);
  4699. IpAddress ip;
  4700. socket->getPeerAddress(ip);
  4701. byte ipdata[16];
  4702. size32_t ipds = ip.getNetAddress(sizeof(ipdata),&ipdata);
  4703. mergeOnce(oncekey,sizeof(ipdata),&ipdata); // this is clients key
  4704. OnceKey mykey;
  4705. genOnce(mykey);
  4706. reply.append((unsigned)0); // errcode
  4707. aesEncrypt(&oncekey,sizeof(oncekey),&mykey,sizeof(oncekey),encbuf);
  4708. reply.append(encbuf.length()).append(encbuf);
  4709. sendBuffer(socket, reply); // send my oncekey
  4710. reqbuf.clear();
  4711. receiveBuffer(socket, reqbuf, 1);
  4712. if (reqbuf.remaining()>sizeof(RemoteFileCommandType)+sizeof(size32_t)) {
  4713. reqbuf.read(typ);
  4714. if (typ==RFCunlockreply) {
  4715. size32_t bs;
  4716. reqbuf.read(bs);
  4717. if (bs<=reqbuf.remaining()) {
  4718. MemoryBuffer userbuf;
  4719. aesDecrypt(&mykey,sizeof(mykey),reqbuf.readDirect(bs),bs,userbuf);
  4720. byte n;
  4721. userbuf.read(n);
  4722. if (n>=2) {
  4723. StringAttr user;
  4724. StringAttr password;
  4725. userbuf.read(user).read(password);
  4726. Owned<IAuthenticatedUser> iau = createAuthenticatedUser();
  4727. if (iau->login(user,password)) {
  4728. initSendBuffer(reply.clear());
  4729. reply.append((unsigned)0);
  4730. sendBuffer(socket, reply); // send OK
  4731. ret = iau;
  4732. return true;
  4733. }
  4734. }
  4735. }
  4736. }
  4737. }
  4738. reply.clear();
  4739. appendErr(reply, (RemoteFileCommandType)RFSERR_AuthenticateFailed);
  4740. sendBuffer(socket, reply); // send OK
  4741. return false;
  4742. }
  4743. void runClient(ISocket *sock)
  4744. {
  4745. cCommandProcessor::cCommandProcessorParams params;
  4746. IAuthenticatedUser *user=NULL;
  4747. bool authenticated = false;
  4748. try {
  4749. if (checkAuthentication(sock,user))
  4750. authenticated = true;
  4751. }
  4752. catch (IException *e) {
  4753. e->Release();
  4754. }
  4755. if (!authenticated) {
  4756. try {
  4757. sock->Release();
  4758. }
  4759. catch (IException *e) {
  4760. e->Release();
  4761. }
  4762. return;
  4763. }
  4764. params.client = new CRemoteClientHandler(this,sock,user,globallasttick);
  4765. {
  4766. CriticalBlock block(sect);
  4767. clients.append(*LINK(params.client));
  4768. }
  4769. // NB: This could be blocked, by thread pool limit
  4770. threads->start(&params);
  4771. }
  4772. void stop()
  4773. {
  4774. // stop accept loop
  4775. if (TF_TRACE_CLIENT_STATS)
  4776. PROGLOG("CRemoteFileServer::stop");
  4777. if (acceptsock)
  4778. acceptsock->cancel_accept();
  4779. if (rejectsock)
  4780. rejectsock->cancel_accept();
  4781. threads->stopAll();
  4782. threads->joinAll(true,60*1000);
  4783. }
  4784. bool notify(CRemoteClientHandler *_client, MemoryBuffer &msg)
  4785. {
  4786. Linked<CRemoteClientHandler> client;
  4787. client.set(_client);
  4788. if (TF_TRACE_FULL)
  4789. PROGLOG("notify %d", msg.length());
  4790. if (msg.length())
  4791. {
  4792. if (TF_TRACE_FULL)
  4793. PROGLOG("notify CRemoteClientHandler(%p), msg length=%u", _client, msg.length());
  4794. cCommandProcessor::cCommandProcessorParams params;
  4795. params.client = client.getClear();
  4796. params.msg.swapWith(msg);
  4797. /* This can block because the thread pool is full and therefore block the selecthandler
  4798. * This is akin to the main server blocking post accept() for the same reason.
  4799. */
  4800. threads->start(&params);
  4801. }
  4802. else
  4803. onCloseSocket(client,3); // removes owned handles
  4804. return false;
  4805. }
  4806. void addClient(CRemoteClientHandler *client)
  4807. {
  4808. if (client&&client->socket)
  4809. selecthandler->add(client->socket,SELECTMODE_READ,client);
  4810. }
  4811. void checkTimeout()
  4812. {
  4813. if (msTick()-clientcounttick>1000*60*60) {
  4814. CriticalBlock block(ClientCountSect);
  4815. if (TF_TRACE_CLIENT_STATS && (ClientCount || MaxClientCount))
  4816. PROGLOG("Client count = %d, max = %d", ClientCount, MaxClientCount);
  4817. clientcounttick = msTick();
  4818. MaxClientCount = ClientCount;
  4819. if (closedclients) {
  4820. if (TF_TRACE_CLIENT_STATS)
  4821. PROGLOG("Closed client count = %d",closedclients);
  4822. closedclients = 0;
  4823. }
  4824. }
  4825. CriticalBlock block(sect);
  4826. ForEachItemInRev(i,clients) {
  4827. CRemoteClientHandler &client = clients.item(i);
  4828. if (client.timedOut()) {
  4829. StringBuffer s;
  4830. bool ok = client.getInfo(s); // will spot duff sockets
  4831. if (ok&&(client.handles.ordinality()!=0)) {
  4832. if (TF_TRACE_CLIENT_CONN && client.inactiveTimedOut())
  4833. WARNLOG("Inactive %s",s.str());
  4834. }
  4835. else {
  4836. #ifndef _DEBUG
  4837. if (TF_TRACE_CLIENT_CONN)
  4838. #endif
  4839. PROGLOG("Timing out %s",s.str());
  4840. closedclients++;
  4841. onCloseSocket(&client,4); // removes owned handles
  4842. }
  4843. }
  4844. }
  4845. }
  4846. void getInfo(StringBuffer &info, unsigned level=1)
  4847. {
  4848. {
  4849. CriticalBlock block(ClientCountSect);
  4850. info.append(VERSTRING).append('\n');
  4851. info.appendf("Client count = %d\n",ClientCount);
  4852. info.appendf("Max client count = %d",MaxClientCount);
  4853. }
  4854. CriticalBlock block(sect);
  4855. ForEachItemIn(i,clients)
  4856. {
  4857. info.newline().append(i).append(": ");
  4858. clients.item(i).getInfo(info);
  4859. }
  4860. info.newline().appendf("Running threads: %u", threadRunningCount());
  4861. info.newline();
  4862. stdCmdThrottler.getInfo(info);
  4863. info.newline();
  4864. slowCmdThrottler.getInfo(info);
  4865. clientStatsTable.getInfo(info, level);
  4866. }
  4867. unsigned threadRunningCount()
  4868. {
  4869. if (!threads)
  4870. return 0;
  4871. return threads->runningCount();
  4872. }
  4873. unsigned idleTime()
  4874. {
  4875. unsigned t = (unsigned)atomic_read(&globallasttick);
  4876. return msTick()-t;
  4877. }
  4878. void setThrottle(ThrottleClass throttleClass, unsigned limit, unsigned delayMs, unsigned cpuThreshold, unsigned queueLimit)
  4879. {
  4880. switch (throttleClass)
  4881. {
  4882. case ThrottleStd:
  4883. stdCmdThrottler.configure(limit, delayMs, cpuThreshold, queueLimit);
  4884. break;
  4885. case ThrottleSlow:
  4886. slowCmdThrottler.configure(limit, delayMs, cpuThreshold, queueLimit);
  4887. break;
  4888. default:
  4889. {
  4890. StringBuffer availableClasses("{ ");
  4891. for (unsigned c=0; c<ThrottleClassMax; c++)
  4892. {
  4893. availableClasses.append(c).append(" = ").append(getThrottleClassText((ThrottleClass)c));
  4894. if (c+1<ThrottleClassMax)
  4895. availableClasses.append(", ");
  4896. }
  4897. availableClasses.append(" }");
  4898. throw MakeStringException(0, "Unknown throttle class: %u, available classes are: %s", (unsigned)throttleClass, availableClasses.str());
  4899. }
  4900. }
  4901. }
  4902. StringBuffer &getStats(StringBuffer &stats, bool reset)
  4903. {
  4904. CriticalBlock block(sect);
  4905. stdCmdThrottler.getStats(stats, reset).newline();
  4906. slowCmdThrottler.getStats(stats, reset);
  4907. if (reset)
  4908. clientStatsTable.reset();
  4909. return stats;
  4910. }
  4911. };
  4912. IRemoteFileServer * createRemoteFileServer(unsigned maxThreads, unsigned maxThreadsDelayMs, unsigned maxAsyncCopy)
  4913. {
  4914. #if SIMULATE_PACKETLOSS
  4915. errorSimulationOn = false;
  4916. #endif
  4917. return new CRemoteFileServer(maxThreads, maxThreadsDelayMs, maxAsyncCopy);
  4918. }