jsocket.cpp 178 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631263226332634263526362637263826392640264126422643264426452646264726482649265026512652265326542655265626572658265926602661266226632664266526662667266826692670267126722673267426752676267726782679268026812682268326842685268626872688268926902691269226932694269526962697269826992700270127022703270427052706270727082709271027112712271327142715271627172718271927202721272227232724272527262727272827292730273127322733273427352736273727382739274027412742274327442745274627472748274927502751275227532754275527562757275827592760276127622763276427652766276727682769277027712772277327742775277627772778277927802781278227832784278527862787278827892790279127922793279427952796279727982799280028012802280328042805280628072808280928102811281228132814281528162817281828192820282128222823282428252826282728282829283028312832283328342835283628372838283928402841284228432844284528462847284828492850285128522853285428552856285728582859286028612862286328642865286628672868286928702871287228732874287528762877287828792880288128822883288428852886288728882889289028912892289328942895289628972898289929002901290229032904290529062907290829092910291129122913291429152916291729182919292029212922292329242925292629272928292929302931293229332934293529362937293829392940294129422943294429452946294729482949295029512952295329542955295629572958295929602961296229632964296529662967296829692970297129722973297429752976297729782979298029812982298329842985298629872988298929902991299229932994299529962997299829993000300130023003300430053006300730083009301030113012301330143015301630173018301930203021302230233024302530263027302830293030303130323033303430353036303730383039304030413042304330443045304630473048304930503051305230533054305530563057305830593060306130623063306430653066306730683069307030713072307330743075307630773078307930803081308230833084308530863087308830893090309130923093309430953096309730983099310031013102310331043105310631073108310931103111311231133114311531163117311831193120312131223123312431253126312731283129313031313132313331343135313631373138313931403141314231433144314531463147314831493150315131523153315431553156315731583159316031613162316331643165316631673168316931703171317231733174317531763177317831793180318131823183318431853186318731883189319031913192319331943195319631973198319932003201320232033204320532063207320832093210321132123213321432153216321732183219322032213222322332243225322632273228322932303231323232333234323532363237323832393240324132423243324432453246324732483249325032513252325332543255325632573258325932603261326232633264326532663267326832693270327132723273327432753276327732783279328032813282328332843285328632873288328932903291329232933294329532963297329832993300330133023303330433053306330733083309331033113312331333143315331633173318331933203321332233233324332533263327332833293330333133323333333433353336333733383339334033413342334333443345334633473348334933503351335233533354335533563357335833593360336133623363336433653366336733683369337033713372337333743375337633773378337933803381338233833384338533863387338833893390339133923393339433953396339733983399340034013402340334043405340634073408340934103411341234133414341534163417341834193420342134223423342434253426342734283429343034313432343334343435343634373438343934403441344234433444344534463447344834493450345134523453345434553456345734583459346034613462346334643465346634673468346934703471347234733474347534763477347834793480348134823483348434853486348734883489349034913492349334943495349634973498349935003501350235033504350535063507350835093510351135123513351435153516351735183519352035213522352335243525352635273528352935303531353235333534353535363537353835393540354135423543354435453546354735483549355035513552355335543555355635573558355935603561356235633564356535663567356835693570357135723573357435753576357735783579358035813582358335843585358635873588358935903591359235933594359535963597359835993600360136023603360436053606360736083609361036113612361336143615361636173618361936203621362236233624362536263627362836293630363136323633363436353636363736383639364036413642364336443645364636473648364936503651365236533654365536563657365836593660366136623663366436653666366736683669367036713672367336743675367636773678367936803681368236833684368536863687368836893690369136923693369436953696369736983699370037013702370337043705370637073708370937103711371237133714371537163717371837193720372137223723372437253726372737283729373037313732373337343735373637373738373937403741374237433744374537463747374837493750375137523753375437553756375737583759376037613762376337643765376637673768376937703771377237733774377537763777377837793780378137823783378437853786378737883789379037913792379337943795379637973798379938003801380238033804380538063807380838093810381138123813381438153816381738183819382038213822382338243825382638273828382938303831383238333834383538363837383838393840384138423843384438453846384738483849385038513852385338543855385638573858385938603861386238633864386538663867386838693870387138723873387438753876387738783879388038813882388338843885388638873888388938903891389238933894389538963897389838993900390139023903390439053906390739083909391039113912391339143915391639173918391939203921392239233924392539263927392839293930393139323933393439353936393739383939394039413942394339443945394639473948394939503951395239533954395539563957395839593960396139623963396439653966396739683969397039713972397339743975397639773978397939803981398239833984398539863987398839893990399139923993399439953996399739983999400040014002400340044005400640074008400940104011401240134014401540164017401840194020402140224023402440254026402740284029403040314032403340344035403640374038403940404041404240434044404540464047404840494050405140524053405440554056405740584059406040614062406340644065406640674068406940704071407240734074407540764077407840794080408140824083408440854086408740884089409040914092409340944095409640974098409941004101410241034104410541064107410841094110411141124113411441154116411741184119412041214122412341244125412641274128412941304131413241334134413541364137413841394140414141424143414441454146414741484149415041514152415341544155415641574158415941604161416241634164416541664167416841694170417141724173417441754176417741784179418041814182418341844185418641874188418941904191419241934194419541964197419841994200420142024203420442054206420742084209421042114212421342144215421642174218421942204221422242234224422542264227422842294230423142324233423442354236423742384239424042414242424342444245424642474248424942504251425242534254425542564257425842594260426142624263426442654266426742684269427042714272427342744275427642774278427942804281428242834284428542864287428842894290429142924293429442954296429742984299430043014302430343044305430643074308430943104311431243134314431543164317431843194320432143224323432443254326432743284329433043314332433343344335433643374338433943404341434243434344434543464347434843494350435143524353435443554356435743584359436043614362436343644365436643674368436943704371437243734374437543764377437843794380438143824383438443854386438743884389439043914392439343944395439643974398439944004401440244034404440544064407440844094410441144124413441444154416441744184419442044214422442344244425442644274428442944304431443244334434443544364437443844394440444144424443444444454446444744484449445044514452445344544455445644574458445944604461446244634464446544664467446844694470447144724473447444754476447744784479448044814482448344844485448644874488448944904491449244934494449544964497449844994500450145024503450445054506450745084509451045114512451345144515451645174518451945204521452245234524452545264527452845294530453145324533453445354536453745384539454045414542454345444545454645474548454945504551455245534554455545564557455845594560456145624563456445654566456745684569457045714572457345744575457645774578457945804581458245834584458545864587458845894590459145924593459445954596459745984599460046014602460346044605460646074608460946104611461246134614461546164617461846194620462146224623462446254626462746284629463046314632463346344635463646374638463946404641464246434644464546464647464846494650465146524653465446554656465746584659466046614662466346644665466646674668466946704671467246734674467546764677467846794680468146824683468446854686468746884689469046914692469346944695469646974698469947004701470247034704470547064707470847094710471147124713471447154716471747184719472047214722472347244725472647274728472947304731473247334734473547364737473847394740474147424743474447454746474747484749475047514752475347544755475647574758475947604761476247634764476547664767476847694770477147724773477447754776477747784779478047814782478347844785478647874788478947904791479247934794479547964797479847994800480148024803480448054806480748084809481048114812481348144815481648174818481948204821482248234824482548264827482848294830483148324833483448354836483748384839484048414842484348444845484648474848484948504851485248534854485548564857485848594860486148624863486448654866486748684869487048714872487348744875487648774878487948804881488248834884488548864887488848894890489148924893489448954896489748984899490049014902490349044905490649074908490949104911491249134914491549164917491849194920492149224923492449254926492749284929493049314932493349344935493649374938493949404941494249434944494549464947494849494950495149524953495449554956495749584959496049614962496349644965496649674968496949704971497249734974497549764977497849794980498149824983498449854986498749884989499049914992499349944995499649974998499950005001500250035004500550065007500850095010501150125013501450155016501750185019502050215022502350245025502650275028502950305031503250335034503550365037503850395040504150425043504450455046504750485049505050515052505350545055505650575058505950605061506250635064506550665067506850695070507150725073507450755076507750785079508050815082508350845085508650875088508950905091509250935094509550965097509850995100510151025103510451055106510751085109511051115112511351145115511651175118511951205121512251235124512551265127512851295130513151325133513451355136513751385139514051415142514351445145514651475148514951505151515251535154515551565157515851595160516151625163516451655166516751685169517051715172517351745175517651775178517951805181518251835184518551865187518851895190519151925193519451955196519751985199520052015202520352045205520652075208520952105211521252135214521552165217521852195220522152225223522452255226522752285229523052315232523352345235523652375238523952405241524252435244524552465247524852495250525152525253525452555256525752585259526052615262526352645265526652675268526952705271527252735274527552765277527852795280528152825283528452855286528752885289529052915292529352945295529652975298529953005301530253035304530553065307530853095310531153125313531453155316531753185319532053215322532353245325532653275328532953305331533253335334533553365337533853395340534153425343534453455346534753485349535053515352535353545355535653575358535953605361536253635364536553665367536853695370537153725373537453755376537753785379538053815382538353845385538653875388538953905391539253935394539553965397539853995400540154025403540454055406540754085409541054115412541354145415541654175418541954205421542254235424542554265427542854295430543154325433543454355436543754385439544054415442544354445445544654475448544954505451545254535454545554565457545854595460546154625463546454655466546754685469547054715472547354745475547654775478547954805481548254835484548554865487548854895490549154925493549454955496549754985499550055015502550355045505550655075508550955105511551255135514551555165517551855195520552155225523552455255526552755285529553055315532553355345535553655375538553955405541554255435544554555465547554855495550555155525553555455555556555755585559556055615562556355645565556655675568556955705571557255735574557555765577557855795580558155825583558455855586558755885589559055915592559355945595559655975598559956005601560256035604560556065607560856095610561156125613561456155616561756185619562056215622562356245625562656275628562956305631563256335634563556365637563856395640564156425643564456455646564756485649565056515652565356545655565656575658565956605661566256635664566556665667566856695670567156725673567456755676567756785679568056815682568356845685568656875688568956905691569256935694569556965697569856995700570157025703570457055706570757085709571057115712571357145715571657175718571957205721572257235724572557265727572857295730573157325733573457355736573757385739574057415742574357445745574657475748574957505751575257535754575557565757575857595760576157625763576457655766576757685769577057715772577357745775577657775778577957805781578257835784578557865787578857895790579157925793579457955796579757985799580058015802580358045805580658075808580958105811581258135814581558165817581858195820582158225823582458255826582758285829583058315832583358345835583658375838583958405841584258435844584558465847584858495850585158525853585458555856585758585859586058615862586358645865586658675868586958705871587258735874587558765877587858795880588158825883588458855886588758885889589058915892589358945895589658975898589959005901590259035904590559065907590859095910591159125913591459155916591759185919592059215922592359245925592659275928592959305931593259335934593559365937593859395940594159425943594459455946594759485949595059515952595359545955595659575958595959605961596259635964596559665967596859695970597159725973597459755976597759785979598059815982598359845985598659875988598959905991599259935994599559965997599859996000600160026003600460056006600760086009601060116012601360146015601660176018601960206021602260236024602560266027602860296030603160326033603460356036603760386039604060416042604360446045604660476048604960506051605260536054605560566057605860596060606160626063606460656066606760686069607060716072607360746075607660776078607960806081608260836084608560866087608860896090609160926093609460956096609760986099610061016102610361046105610661076108610961106111611261136114611561166117611861196120612161226123612461256126612761286129613061316132613361346135613661376138613961406141614261436144614561466147614861496150615161526153615461556156615761586159616061616162
  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. // New IPv6 Version - IN PROGRESS
  14. /*
  15. TBD IPv6 connect
  16. multicast
  17. look at loopback
  18. */
  19. #include "platform.h"
  20. #ifdef _VER_C5
  21. #include <clwclib.h>
  22. #else
  23. #include "platform.h"
  24. #include <stdio.h>
  25. #endif
  26. #include <algorithm>
  27. #ifdef _WIN32
  28. #define WIN32_LEAN_AND_MEAN
  29. #include <windows.h>
  30. #include <winsock2.h>
  31. #include <ws2tcpip.h>
  32. #include <signal.h>
  33. #else
  34. #include <sys/types.h>
  35. #include <sys/socket.h>
  36. #include <netinet/tcp.h>
  37. #include <netinet/in.h>
  38. #include <arpa/inet.h>
  39. #include <stddef.h>
  40. #include <errno.h>
  41. #include <net/if.h>
  42. #endif
  43. #include <limits.h>
  44. #include "jmutex.hpp"
  45. #include "jsocket.hpp"
  46. #include "jexcept.hpp"
  47. #include "jio.hpp"
  48. #include "jmisc.hpp"
  49. #include "jthread.hpp"
  50. #include "jqueue.tpp"
  51. #include "jtime.hpp"
  52. #include "jprop.hpp"
  53. #include "jregexp.hpp"
  54. #include "jdebug.hpp"
  55. #include "build-config.h"
  56. // epoll only with linux
  57. #ifndef __linux__
  58. # undef _HAS_EPOLL_SUPPORT
  59. #else
  60. # define _HAS_EPOLL_SUPPORT
  61. # ifdef _HAS_EPOLL_SUPPORT
  62. # include <unistd.h>
  63. # include <sys/epoll.h>
  64. # ifndef EPOLLRDHUP
  65. // Centos 5.x bug - epoll.h does not define but its in the kernel
  66. # define EPOLLRDHUP 0x2000
  67. # endif
  68. # endif
  69. #endif
  70. // various options
  71. #define CONNECT_TIMEOUT_REFUSED_WAIT 1000 // maximum to sleep on connect_timeout
  72. #define TRACE_SLOW_BLOCK_TRANSFER
  73. #define DEFAULT_CONNECT_TIME (100*1000) // for connect_wait
  74. #ifndef _WIN32
  75. #define BLOCK_POLLED_SINGLE_CONNECTS // NB this is much slower in windows
  76. #define CENTRAL_NODE_RANDOM_DELAY
  77. #else
  78. #define USERECVSEM // to singlethread BF_SYNC_TRANSFER_PUSH
  79. #endif
  80. #ifdef _DEBUG
  81. //#define SOCKTRACE
  82. //#define EPOLLTRACE
  83. #endif
  84. #ifdef _TESTING
  85. #define _TRACE
  86. #endif
  87. #ifdef _TRACE
  88. #define THROWJSOCKEXCEPTION(exc) \
  89. { StringBuffer msg; \
  90. msg.appendf("Target: %s, Raised in: %s, line %d",tracename ,__FILE__, __LINE__); \
  91. IJSOCK_Exception *e = new SocketException(exc,msg.str());\
  92. throw e; }
  93. #define THROWJSOCKEXCEPTION2(exc) \
  94. { StringBuffer msg; \
  95. msg.appendf("Raised in: %s, line %d",__FILE__, __LINE__); \
  96. IJSOCK_Exception *e = new SocketException(exc,msg.str());\
  97. throw e; }
  98. #define LOGERR(err,ref,info) LogErr(err,ref,info,__LINE__,NULL)
  99. #define LOGERR2(err,ref,info) LogErr(err,ref,info,__LINE__,tracename)
  100. #else
  101. #define THROWJSOCKEXCEPTION(exc) \
  102. { IJSOCK_Exception *e = new SocketException(exc);\
  103. throw e; }
  104. #define THROWJSOCKEXCEPTION2(exc) THROWJSOCKEXCEPTION(exc)
  105. #define LOGERR(err,ref,info)
  106. #define LOGERR2(err,ref,info)
  107. #endif
  108. JSocketStatistics STATS;
  109. static bool IP4only=false; // slighly faster if we know no IPv6
  110. static bool IP6preferred=false; // e.g. for DNS and socket create
  111. IpSubNet PreferredSubnet(NULL,NULL); // set this if you prefer a particular subnet for debugging etc
  112. // e.g. PreferredSubnet("192.168.16.0", "255.255.255.0")
  113. static atomic_t pre_conn_unreach_cnt = ATOMIC_INIT(0); // global count of pre_connect() ENETUNREACH error
  114. #define IPV6_SERIALIZE_PREFIX (0x00ff00ff)
  115. inline void LogErr(unsigned err,unsigned ref,const char *info,unsigned lineno,const char *tracename)
  116. {
  117. if (err)
  118. PROGLOG("jsocket(%d,%d)%s%s err = %d%s%s",ref,lineno,
  119. (info&&*info)?" ":"",(info&&*info)?info:"",err,
  120. (tracename&&*tracename)?" : ":"",(tracename&&*tracename)?tracename:"");
  121. }
  122. class jlib_thrown_decl SocketException: public CInterface, public IJSOCK_Exception
  123. {
  124. public:
  125. IMPLEMENT_IINTERFACE;
  126. SocketException(int code,const char *_msg=NULL) : errcode(code)
  127. {
  128. if (_msg)
  129. msg = strdup(_msg);
  130. else
  131. msg = NULL;
  132. };
  133. ~SocketException() { free(msg); }
  134. int errorCode() const { return errcode; }
  135. static StringBuffer & geterrormessage(int err,StringBuffer &str)
  136. {
  137. switch (err) {
  138. case JSOCKERR_ok: return str.append("ok");
  139. case JSOCKERR_not_opened: return str.append("socket not opened");
  140. case JSOCKERR_bad_address: return str.append("bad address");
  141. case JSOCKERR_connection_failed: return str.append("connection failed");
  142. case JSOCKERR_broken_pipe: return str.append("connection is broken");
  143. case JSOCKERR_graceful_close: return str.append("connection closed other end");
  144. case JSOCKERR_invalid_access_mode: return str.append("invalid access mode");
  145. case JSOCKERR_timeout_expired: return str.append("timeout expired");
  146. case JSOCKERR_port_in_use: return str.append("port in use");
  147. case JSOCKERR_cancel_accept: return str.append("cancel accept");
  148. case JSOCKERR_connectionless_socket: return str.append("connectionless socket");
  149. case JSOCKERR_handle_too_large: return str.append("handle too large");
  150. case JSOCKERR_bad_netaddr: return str.append("bad net addr");
  151. case JSOCKERR_ipv6_not_implemented: return str.append("IPv6 not implemented");
  152. // OS errors
  153. #ifdef _WIN32
  154. case WSAEINTR: return str.append("WSAEINTR(10004) - Interrupted system call.");
  155. case WSAEBADF: return str.append("WSAEBADF(10009) - Bad file number.");
  156. case WSAEACCES: return str.append("WSAEACCES(10013) - Permission denied.");
  157. case WSAEFAULT: return str.append("WSAEFAULT(10014) - Bad address.");
  158. case WSAEINVAL: return str.append("WSAEINVAL(10022) - Invalid argument.");
  159. case WSAEMFILE: return str.append("WSAEMFILE(10024) - Too many open files.");
  160. case WSAEWOULDBLOCK: return str.append("WSAEWOULDBLOCK(10035) - Operation would block.");
  161. case WSAEINPROGRESS: return str.append("WSAEINPROGRESS(10036) - Operation now in progress.");
  162. case WSAEALREADY: return str.append("WSAEALREADY(10037) - Operation already in progress.");
  163. case WSAENOTSOCK: return str.append("WSAENOTSOCK(10038) - Socket operation on nonsocket.");
  164. case WSAEDESTADDRREQ: return str.append("WSAEDESTADDRREQ(10039) - Destination address required.");
  165. case WSAEMSGSIZE: return str.append("WSAEMSGSIZE(10040) - Message too long.");
  166. case WSAEPROTOTYPE: return str.append("WSAEPROTOTYPE(10041) - Protocol wrong type for socket.");
  167. case WSAENOPROTOOPT: return str.append("WSAENOPROTOOPT(10042) - Protocol not available.");
  168. case WSAEPROTONOSUPPORT: return str.append("WSAEPROTONOSUPPORT(10043) - Protocol not supported.");
  169. case WSAESOCKTNOSUPPORT: return str.append("WSAESOCKTNOSUPPORT(10044) - Socket type not supported.");
  170. case WSAEOPNOTSUPP: return str.append("WSAEOPNOTSUPP(10045) - Operation not supported on socket.");
  171. case WSAEPFNOSUPPORT: return str.append("WSAEPFNOSUPPORT(10046) - Protocol family not supported.");
  172. case WSAEAFNOSUPPORT: return str.append("WSAEAFNOSUPPORT(10047) - Address family not supported by protocol family.");
  173. case WSAEADDRINUSE: return str.append("WSAEADDRINUSE(10048) - Address already in use.");
  174. case WSAEADDRNOTAVAIL: return str.append("WSAEADDRNOTAVAIL(10049) - Cannot assign requested address.");
  175. case WSAENETDOWN: return str.append("WSAENETDOWN(10050) - Network is down.");
  176. case WSAENETUNREACH: return str.append("WSAENETUNREACH(10051) - Network is unreachable.");
  177. case WSAENETRESET: return str.append("WSAENETRESET(10052) - Network dropped connection on reset.");
  178. case WSAECONNABORTED: return str.append("WSAECONNABORTED(10053) - Software caused connection abort.");
  179. case WSAECONNRESET: return str.append("WSAECONNRESET(10054) - Connection reset by peer.");
  180. case WSAENOBUFS: return str.append("WSAENOBUFS(10055) - No buffer space available.");
  181. case WSAEISCONN: return str.append("WSAEISCONN(10056) - Socket is already connected.");
  182. case WSAENOTCONN: return str.append("WSAENOTCONN(10057) - Socket is not connected.");
  183. case WSAESHUTDOWN: return str.append("WSAESHUTDOWN(10058) - Cannot send after socket shutdown.");
  184. case WSAETOOMANYREFS: return str.append("WSAETOOMANYREFS(10059) - Too many references: cannot splice.");
  185. case WSAETIMEDOUT: return str.append("WSAETIMEDOUT(10060) - Connection timed out.");
  186. case WSAECONNREFUSED: return str.append("WSAECONNREFUSED(10061) - Connection refused.");
  187. case WSAELOOP: return str.append("WSAELOOP(10062) - Too many levels of symbolic links.");
  188. case WSAENAMETOOLONG: return str.append("WSAENAMETOOLONG(10063) - File name too long.");
  189. case WSAEHOSTDOWN: return str.append("WSAEHOSTDOWN(10064) - Host is down.");
  190. case WSAEHOSTUNREACH: return str.append("WSAEHOSTUNREACH(10065) - No route to host.");
  191. case WSASYSNOTREADY: return str.append("WSASYSNOTREADY(10091) - The network subsystem is unusable.");
  192. case WSAVERNOTSUPPORTED: return str.append("WSAVERNOTSUPPORTED(10092) - The Windows Sockets DLL cannot support this application.");
  193. case WSANOTINITIALISED: return str.append("WSANOTINITIALISED(10093) - Winsock not initialized.");
  194. case WSAEDISCON: return str.append("WSAEDISCON(10101) - Disconnect.");
  195. case WSAHOST_NOT_FOUND: return str.append("WSAHOST_NOT_FOUND(11001) - Host not found.");
  196. case WSATRY_AGAIN: return str.append("WSATRY_AGAIN(11002) - Nonauthoritative host not found.");
  197. case WSANO_RECOVERY: return str.append("WSANO_RECOVERY(11003) - Nonrecoverable error.");
  198. case WSANO_DATA: return str.append("WSANO_DATA(11004) - Valid name, no data record of requested type.");
  199. #else
  200. case ENOTSOCK: return str.append("ENOTSOCK - Socket operation on non-socket ");
  201. case EDESTADDRREQ: return str.append("EDESTADDRREQ - Destination address required ");
  202. case EMSGSIZE: return str.append("EMSGSIZE - Message too long ");
  203. case EPROTOTYPE: return str.append("EPROTOTYPE - Protocol wrong type for socket ");
  204. case ENOPROTOOPT: return str.append("ENOPROTOOPT - Protocol not available ");
  205. case EPROTONOSUPPORT: return str.append("EPROTONOSUPPORT - Protocol not supported ");
  206. case ESOCKTNOSUPPORT: return str.append("ESOCKTNOSUPPORT - Socket type not supported ");
  207. case EOPNOTSUPP: return str.append("EOPNOTSUPP - Operation not supported on socket ");
  208. case EPFNOSUPPORT: return str.append("EPFNOSUPPORT - Protocol family not supported ");
  209. case EAFNOSUPPORT: return str.append("EAFNOSUPPORT - Address family not supported by protocol family ");
  210. case EADDRINUSE: return str.append("EADDRINUSE - Address already in use ");
  211. case EADDRNOTAVAIL: return str.append("EADDRNOTAVAIL - Can't assign requested address ");
  212. case ENETDOWN: return str.append("ENETDOWN - Network is down ");
  213. case ENETUNREACH: return str.append("ENETUNREACH - Network is unreachable ");
  214. case ENETRESET: return str.append("ENETRESET - Network dropped connection because of reset ");
  215. case ECONNABORTED: return str.append("ECONNABORTED - Software caused connection abort ");
  216. case ECONNRESET: return str.append("ECONNRESET - Connection reset by peer ");
  217. case ENOBUFS: return str.append("ENOBUFS - No buffer space available ");
  218. case EISCONN: return str.append("EISCONN - Socket is already connected ");
  219. case ENOTCONN: return str.append("ENOTCONN - Socket is not connected ");
  220. case ESHUTDOWN: return str.append("ESHUTDOWN - Can't send after socket shutdown ");
  221. case ETOOMANYREFS: return str.append("ETOOMANYREFS - Too many references: can't splice ");
  222. case ETIMEDOUT: return str.append("ETIMEDOUT - Connection timed out ");
  223. case ECONNREFUSED: return str.append("ECONNREFUSED - Connection refused ");
  224. case EHOSTDOWN: return str.append("EHOSTDOWN - Host is down ");
  225. case EHOSTUNREACH: return str.append("EHOSTUNREACH - No route to host ");
  226. case EWOULDBLOCK: return str.append("EWOULDBLOCK - operation already in progress");
  227. case EINPROGRESS: return str.append("EINPROGRESS - operation now in progress ");
  228. #endif
  229. }
  230. IException *ose = makeOsException(err);
  231. ose->errorMessage(str);
  232. ose->Release();
  233. return str;
  234. }
  235. StringBuffer & errorMessage(StringBuffer &str) const
  236. {
  237. if (msg)
  238. return geterrormessage(errcode,str).append('\n').append(msg);
  239. return geterrormessage(errcode,str);
  240. }
  241. MessageAudience errorAudience() const
  242. {
  243. switch (errcode) {
  244. case JSOCKERR_port_in_use:
  245. return MSGAUD_operator;
  246. }
  247. return MSGAUD_user;
  248. }
  249. private:
  250. int errcode;
  251. char *msg;
  252. };
  253. IJSOCK_Exception *IPv6NotImplementedException(const char *filename,unsigned lineno)
  254. {
  255. StringBuffer msg;
  256. msg.appendf("%s(%d)",filename,lineno);
  257. return new SocketException(JSOCKERR_ipv6_not_implemented,msg.str());
  258. }
  259. struct MCASTREQ
  260. {
  261. struct in_addr imr_multiaddr; /* multicast group to join */
  262. struct in_addr imr_interface; /* interface to join on */
  263. MCASTREQ(const char *mcip)
  264. {
  265. imr_multiaddr.s_addr = inet_addr(mcip);
  266. imr_interface.s_addr = htonl(INADDR_ANY);
  267. }
  268. };
  269. #ifdef __APPLE__
  270. #ifndef MSG_NOSIGNAL
  271. #define MSG_NOSIGNAL 0x4000
  272. #endif
  273. #endif
  274. #if defined( _WIN32)
  275. #define T_SOCKET SOCKET
  276. #define T_FD_SET fd_set
  277. #define XFD_SETSIZE FD_SETSIZE
  278. #define ETIMEDOUT WSAETIMEDOUT
  279. #define ECONNREFUSED WSAECONNREFUSED
  280. #define XFD_ZERO(s) FD_ZERO(s)
  281. #define SEND_FLAGS 0
  282. #define BADSOCKERR(err) ((err==WSAEBADF)||(err==WSAENOTSOCK))
  283. #define CHECKSOCKRANGE(s)
  284. #elif defined(__FreeBSD__) || defined(__APPLE__)
  285. #define XFD_SETSIZE FD_SETSIZE
  286. #define T_FD_SET fd_set
  287. #define XFD_ZERO(s) FD_ZERO(s)
  288. #define T_SOCKET int
  289. #define SEND_FLAGS (MSG_NOSIGNAL)
  290. #define BADSOCKERR(err) ((err==EBADF)||(err==ENOTSOCK))
  291. #define CHECKSOCKRANGE(s)
  292. #else
  293. #define XFD_SETSIZE 32768
  294. struct xfd_set { __fd_mask fds_bits[XFD_SETSIZE / __NFDBITS]; }; // define our own
  295. // linux 64 bit
  296. #ifdef __linux__
  297. #ifdef __x86_64__
  298. #undef __FDMASK
  299. #define __FDMASK(d) (1UL << ((d) % __NFDBITS))
  300. #undef __FDELT
  301. #define __FDELT(d) ((d) / __NFDBITS)
  302. #undef __FD_SET
  303. #define __FD_SET(d, s) (__FDS_BITS (s)[__FDELT(d)] |= __FDMASK(d))
  304. #undef __FD_ISSET
  305. #define __FD_ISSET(d, s) ((__FDS_BITS (s)[__FDELT(d)] & __FDMASK(d)) != 0)
  306. #endif
  307. #define CHECKSOCKRANGE(s) { if (s>=XFD_SETSIZE) THROWJSOCKEXCEPTION2(JSOCKERR_handle_too_large); }
  308. #endif
  309. // end 64 bit
  310. #define T_FD_SET xfd_set
  311. #define XFD_ZERO(s) memset(s,0,sizeof(xfd_set))
  312. #define T_SOCKET int
  313. #define SEND_FLAGS (MSG_NOSIGNAL)
  314. #define BADSOCKERR(err) ((err==EBADF)||(err==ENOTSOCK))
  315. #endif
  316. #ifdef CENTRAL_NODE_RANDOM_DELAY
  317. static SocketEndpointArray CentralNodeArray;
  318. #endif
  319. enum SOCKETMODE { sm_tcp_server, sm_tcp, sm_udp_server, sm_udp, sm_multicast_server, sm_multicast};
  320. class CSocket: public CInterface, public ISocket
  321. {
  322. public:
  323. IMPLEMENT_IINTERFACE;
  324. static CriticalSection crit;
  325. protected:
  326. friend class CSocketConnectWait;
  327. enum { ss_open, ss_shutdown, ss_close, ss_pre_open } state;
  328. T_SOCKET sock;
  329. char* hostname; // host address
  330. unsigned short hostport; // host port
  331. SOCKETMODE sockmode;
  332. IpAddress targetip;
  333. SocketEndpoint returnep; // set by set_return_addr
  334. MCASTREQ * mcastreq;
  335. size32_t nextblocksize;
  336. unsigned blockflags;
  337. unsigned blocktimeoutms;
  338. bool owned;
  339. enum {accept_not_cancelled, accept_cancel_pending, accept_cancelled} accept_cancel_state;
  340. bool in_accept;
  341. bool nonblocking;
  342. bool nagling;
  343. static unsigned connectingcount;
  344. #ifdef USERECVSEM
  345. static Semaphore receiveblocksem;
  346. bool receiveblocksemowned; // owned by this socket
  347. #endif
  348. #ifdef _TRACE
  349. char * tracename;
  350. #endif
  351. public:
  352. void open(int listen_queue_size,bool reuseports=false);
  353. bool connect_timeout( unsigned timeout, bool noexception);
  354. void connect_wait( unsigned timems);
  355. void udpconnect();
  356. void read(void* buf, size32_t min_size, size32_t max_size, size32_t &size_read,unsigned timeoutsecs);
  357. void readtms(void* buf, size32_t min_size, size32_t max_size, size32_t &size_read, unsigned timedelaysecs);
  358. void read(void* buf, size32_t size);
  359. size32_t write(void const* buf, size32_t size);
  360. size32_t writetms(void const* buf, size32_t size, unsigned timeoutms=WAIT_FOREVER);
  361. size32_t write_multiple(unsigned num,void const**buf, size32_t *size);
  362. size32_t udp_write_to(const SocketEndpoint &ep,void const* buf, size32_t size);
  363. void close();
  364. void errclose();
  365. bool connectionless() { return (sockmode!=sm_tcp)&&(sockmode!=sm_tcp_server); }
  366. void shutdown(unsigned mode);
  367. ISocket* accept(bool allowcancel);
  368. int wait_read(unsigned timeout);
  369. int wait_write(unsigned timeout);
  370. int name(char *name,size32_t namemax);
  371. int peer_name(char *name,size32_t namemax);
  372. SocketEndpoint &getPeerEndpoint(SocketEndpoint &ep);
  373. IpAddress & getPeerAddress(IpAddress &addr);
  374. void set_return_addr(int port,const char *name); // sets returnep
  375. void cancel_accept();
  376. size32_t get_max_send_size();
  377. bool set_nonblock(bool on=true);
  378. bool set_nagle(bool on);
  379. void set_linger(int lingersecs);
  380. void set_keep_alive(bool set);
  381. virtual void set_inherit(bool inherit=false);
  382. virtual bool check_connection();
  383. // Block functions
  384. void set_block_mode(unsigned flags,size32_t recsize=0,unsigned timeoutms=0);
  385. bool send_block(const void *blk,size32_t sz);
  386. size32_t receive_block_size();
  387. size32_t receive_block(void *blk,size32_t sz);
  388. size32_t get_send_buffer_size();
  389. void set_send_buffer_size(size32_t sz);
  390. bool join_multicast_group(SocketEndpoint &ep); // for udp multicast
  391. bool leave_multicast_group(SocketEndpoint &ep); // for udp multicast
  392. size32_t get_receive_buffer_size();
  393. void set_receive_buffer_size(size32_t sz);
  394. size32_t avail_read();
  395. int pre_connect(bool block);
  396. int post_connect();
  397. CSocket(const SocketEndpoint &_ep,SOCKETMODE smode,const char *name);
  398. CSocket(T_SOCKET new_sock,SOCKETMODE smode,bool _owned);
  399. virtual ~CSocket();
  400. unsigned OShandle()
  401. {
  402. return (unsigned)sock;
  403. }
  404. private:
  405. int closesock()
  406. {
  407. if (sock!=INVALID_SOCKET) {
  408. T_SOCKET s = sock;
  409. sock = INVALID_SOCKET;
  410. STATS.activesockets--;
  411. #ifdef SOCKTRACE
  412. PROGLOG("SOCKTRACE: Closing socket %x %d (%p)", s, s, this);
  413. #endif
  414. #ifdef _WIN32
  415. return ::closesocket(s);
  416. #else
  417. return ::close(s);
  418. #endif
  419. }
  420. else
  421. return 0;
  422. }
  423. };
  424. CriticalSection CSocket::crit;
  425. unsigned CSocket::connectingcount=0;
  426. #ifdef USERECVSEM
  427. Semaphore CSocket::receiveblocksem(2);
  428. #endif
  429. #ifdef _WIN32
  430. class win_socket_library
  431. {
  432. static bool initdone; // to prevent dependancy probs very early on (e.g. jlog)
  433. public:
  434. win_socket_library() { init(); }
  435. bool init()
  436. {
  437. if (initdone)
  438. return true;
  439. WSADATA wsa;
  440. if (WSAStartup(MAKEWORD(2, 2), &wsa) != 0) {
  441. if (WSAStartup(MAKEWORD(1, 1), &wsa) != 0) {
  442. MessageBox(NULL,"Failed to initialize windows sockets","JLib Socket Error",MB_OK);
  443. return false;
  444. }
  445. }
  446. initdone = true;
  447. return true;
  448. }
  449. ~win_socket_library()
  450. {
  451. WSACleanup();
  452. }
  453. };
  454. bool win_socket_library::initdone = false;
  455. static win_socket_library ws32_lib;
  456. #define ERRNO() WSAGetLastError()
  457. #define EADDRINUSE WSAEADDRINUSE
  458. #define EINTRCALL WSAEINTR
  459. #define ECONNRESET WSAECONNRESET
  460. #define ECONNABORTED WSAECONNABORTED
  461. #define ENOTCONN WSAENOTCONN
  462. #define EWOULDBLOCK WSAEWOULDBLOCK
  463. #define EINPROGRESS WSAEINPROGRESS
  464. #define ENETUNREACH WSAENETUNREACH
  465. #define ENOTSOCK WSAENOTSOCK
  466. struct j_sockaddr_in6 {
  467. short sin6_family; /* AF_INET6 */
  468. u_short sin6_port; /* Transport level port number */
  469. u_long sin6_flowinfo; /* IPv6 flow information */
  470. struct in_addr6 sin6_addr; /* IPv6 address */
  471. u_long sin6_scope_id; /* set of interfaces for a scope */
  472. };
  473. typedef union {
  474. struct sockaddr sa;
  475. struct j_sockaddr_in6 sin6;
  476. struct sockaddr_in sin;
  477. } J_SOCKADDR;
  478. #define DEFINE_SOCKADDR(name) J_SOCKADDR name; memset(&name,0,sizeof(J_SOCKADDR))
  479. static int _inet_pton(int af, const char* src, void* dst)
  480. {
  481. DEFINE_SOCKADDR(u);
  482. int address_length;
  483. switch (af) {
  484. case AF_INET:
  485. u.sin.sin_family = AF_INET;
  486. address_length = sizeof (u.sin);
  487. break;
  488. case AF_INET6:
  489. u.sin6.sin6_family = AF_INET6;
  490. address_length = sizeof (u.sin6);
  491. break;
  492. default:
  493. #ifdef EAFNOSUPPORT
  494. errno = EAFNOSUPPORT;
  495. #else
  496. errno = 52;
  497. #endif
  498. return -1;
  499. }
  500. ws32_lib.init();
  501. int ret = WSAStringToAddress ((LPTSTR) src, af, NULL, &u.sa, &address_length);
  502. if (ret == 0) {
  503. switch (af) {
  504. case AF_INET:
  505. memcpy (dst, &u.sin.sin_addr, sizeof (struct in_addr));
  506. break;
  507. case AF_INET6:
  508. memcpy (dst, &u.sin6.sin6_addr, sizeof (u.sin6.sin6_addr));
  509. break;
  510. }
  511. return 1;
  512. }
  513. errno = WSAGetLastError();
  514. // PROGLOG("errno = %d",errno);
  515. return 0;
  516. }
  517. static const char * _inet_ntop (int af, const void *src, char *dst, socklen_t cnt)
  518. {
  519. /* struct sockaddr can't accomodate struct sockaddr_in6. */
  520. DEFINE_SOCKADDR(u);
  521. DWORD dstlen = cnt;
  522. size_t srcsize;
  523. memset(&u,0,sizeof(u));
  524. switch (af) {
  525. case AF_INET:
  526. u.sin.sin_family = AF_INET;
  527. u.sin.sin_addr = *(struct in_addr *) src;
  528. srcsize = sizeof (u.sin);
  529. break;
  530. case AF_INET6:
  531. u.sin6.sin6_family = AF_INET6;
  532. memcpy(&u.sin6.sin6_addr,src,sizeof(in_addr6));
  533. srcsize = sizeof (u.sin6);
  534. break;
  535. default:
  536. return NULL;
  537. }
  538. ws32_lib.init();
  539. if (WSAAddressToString (&u.sa, srcsize, NULL, dst, &dstlen) != 0) {
  540. errno = WSAGetLastError();
  541. return NULL;
  542. }
  543. return (const char *) dst;
  544. }
  545. int inet_aton (const char *name, struct in_addr *addr)
  546. {
  547. addr->s_addr = inet_addr (name);
  548. return (addr->s_addr == (u_long)-1)?1:0; // 255.255.255.255 has had it here
  549. }
  550. #else
  551. #define _inet_ntop inet_ntop
  552. #define _inet_pton inet_pton
  553. #define in_addr6 in6_addr
  554. typedef union {
  555. struct sockaddr sa;
  556. struct sockaddr_in6 sin6;
  557. struct sockaddr_in sin;
  558. } J_SOCKADDR;
  559. #define DEFINE_SOCKADDR(name) J_SOCKADDR name; memset(&name,0,sizeof(J_SOCKADDR))
  560. #define EINTRCALL EINTR
  561. #define ERRNO() (errno)
  562. #ifndef INADDR_NONE
  563. #define INADDR_NONE (-1)
  564. #endif
  565. #endif
  566. #ifndef INET6_ADDRSTRLEN
  567. #define INET6_ADDRSTRLEN 65
  568. #endif
  569. inline socklen_t setSockAddr(J_SOCKADDR &u, const IpAddress &ip,unsigned short port)
  570. {
  571. if (!IP6preferred) {
  572. if (ip.getNetAddress(sizeof(in_addr),&u.sin.sin_addr)==sizeof(in_addr)) {
  573. u.sin.sin_family = AF_INET;
  574. u.sin.sin_port = htons(port);
  575. return sizeof(u.sin);
  576. }
  577. }
  578. if (IP4only)
  579. IPV6_NOT_IMPLEMENTED();
  580. ip.getNetAddress(sizeof(in_addr6),&u.sin6.sin6_addr);
  581. u.sin6.sin6_family = AF_INET6;
  582. u.sin6.sin6_port = htons(port);
  583. return sizeof(u.sin6);
  584. }
  585. inline socklen_t setSockAddrAny(J_SOCKADDR &u, unsigned short port)
  586. {
  587. if (IP6preferred) {
  588. #ifdef _WIN32
  589. IN6ADDR_SETANY((PSOCKADDR_IN6)&u.sin6.sin6_addr);
  590. #else
  591. memcpy(&u.sin6.sin6_addr,&in6addr_any,sizeof(in_addr6));
  592. #endif
  593. u.sin6.sin6_family= AF_INET6;
  594. u.sin6.sin6_port = htons(port);
  595. return sizeof(u.sin6);
  596. }
  597. u.sin.sin_addr.s_addr = htonl(INADDR_ANY);
  598. u.sin.sin_family= AF_INET;
  599. u.sin.sin_port = htons(port);
  600. return sizeof(u.sin);
  601. }
  602. inline void getSockAddrEndpoint(const J_SOCKADDR &u, socklen_t ul, SocketEndpoint &ep)
  603. {
  604. if (ul==sizeof(u.sin)) {
  605. ep.setNetAddress(sizeof(in_addr),&u.sin.sin_addr);
  606. ep.port = htons(u.sin.sin_port);
  607. }
  608. else {
  609. ep.setNetAddress(sizeof(in_addr6),&u.sin6.sin6_addr);
  610. ep.port = htons(u.sin6.sin6_port);
  611. }
  612. }
  613. /* might need fcntl(F_SETFL), or ioctl(FIONBIO) */
  614. /* Posix.1g says fcntl */
  615. #if defined(O_NONBLOCK)
  616. bool CSocket::set_nonblock(bool on)
  617. {
  618. int flags = fcntl(sock, F_GETFL, 0);
  619. if (flags == -1)
  620. return nonblocking;
  621. if (on)
  622. flags |= O_NONBLOCK;
  623. else
  624. flags &= ~O_NONBLOCK;
  625. if (fcntl(sock, F_SETFL, flags)==0) {
  626. bool wasNonBlocking = nonblocking;
  627. nonblocking = on;
  628. return wasNonBlocking;
  629. }
  630. return nonblocking;
  631. }
  632. #else
  633. bool CSocket::set_nonblock(bool on)
  634. {
  635. #ifdef _WIN32
  636. u_long yes = on?1:0;
  637. if (ioctlsocket(sock, FIONBIO, &yes)==0) {
  638. #else
  639. int yes = on?1:0;
  640. if (ioctl(sock, FIONBIO, &yes)==0) {
  641. #endif
  642. bool wasNonBlocking = nonblocking;
  643. nonblocking = on;
  644. return wasNonBlocking;
  645. }
  646. return nonblocking;
  647. }
  648. #endif
  649. bool CSocket::set_nagle(bool on)
  650. {
  651. bool ret = nagling;
  652. nagling = on;
  653. int enabled = !on;
  654. if (setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, (char*)&enabled, sizeof(enabled)) != 0) {
  655. nagling = !on;
  656. }
  657. return ret;
  658. }
  659. void CSocket::set_inherit(bool inherit)
  660. {
  661. #ifndef _WIN32
  662. long flag = fcntl(sock, F_GETFD);
  663. if(inherit)
  664. flag &= ~FD_CLOEXEC;
  665. else
  666. flag |= FD_CLOEXEC;
  667. fcntl(sock, F_SETFD, flag);
  668. #endif
  669. }
  670. size32_t CSocket::avail_read()
  671. {
  672. #ifdef _WIN32
  673. u_long avail;
  674. if (ioctlsocket(sock, FIONREAD, &avail)==0)
  675. #else
  676. int avail;
  677. if (ioctl(sock, FIONREAD, &avail)==0)
  678. #endif
  679. return (size32_t)avail;
  680. int err = ERRNO();
  681. LOGERR2(err,1,"avail_read");
  682. return 0;
  683. }
  684. #define PRE_CONN_UNREACH_ELIM 100
  685. int CSocket::pre_connect (bool block)
  686. {
  687. if (NULL == hostname || '\0' == (*hostname))
  688. {
  689. StringBuffer err;
  690. err.appendf("CSocket::pre_connect - Invalid/missing host IP address raised in : %s, line %d",__FILE__, __LINE__);
  691. IJSOCK_Exception *e = new SocketException(JSOCKERR_bad_netaddr,err.str());
  692. throw e;
  693. }
  694. DEFINE_SOCKADDR(u);
  695. if (targetip.isNull()) {
  696. set_return_addr(hostport,hostname);
  697. targetip.ipset(returnep);
  698. }
  699. socklen_t ul = setSockAddr(u,targetip,hostport);
  700. sock = ::socket(u.sa.sa_family, SOCK_STREAM, targetip.isIp4()?0:PF_INET6);
  701. owned = true;
  702. state = ss_pre_open; // will be set to open by post_connect
  703. if (sock == INVALID_SOCKET) {
  704. int err = ERRNO();
  705. THROWJSOCKEXCEPTION(err);
  706. }
  707. STATS.activesockets++;
  708. int err = 0;
  709. set_nonblock(!block);
  710. int rc = ::connect(sock, &u.sa, ul);
  711. if (rc==SOCKET_ERROR) {
  712. err = ERRNO();
  713. if ((err != EINPROGRESS)&&(err != EWOULDBLOCK)&&(err != ETIMEDOUT)&&(err!=ECONNREFUSED)) { // handled by caller
  714. if (err != ENETUNREACH) {
  715. atomic_set(&pre_conn_unreach_cnt, 0);
  716. LOGERR2(err,1,"pre_connect");
  717. } else {
  718. int ecnt = atomic_read(&pre_conn_unreach_cnt);
  719. if (ecnt <= PRE_CONN_UNREACH_ELIM) {
  720. atomic_inc(&pre_conn_unreach_cnt);
  721. LOGERR2(err,1,"pre_connect network unreachable");
  722. }
  723. }
  724. } else
  725. atomic_set(&pre_conn_unreach_cnt, 0);
  726. } else
  727. atomic_set(&pre_conn_unreach_cnt, 0);
  728. #ifdef SOCKTRACE
  729. PROGLOG("SOCKTRACE: pre-connected socket%s %x %d (%p) err=%d", block?"(block)":"", sock, sock, this, err);
  730. #endif
  731. return err;
  732. }
  733. int CSocket::post_connect ()
  734. {
  735. set_nonblock(false);
  736. int err = 0;
  737. socklen_t errlen = sizeof(err);
  738. int rc = getsockopt(sock, SOL_SOCKET, SO_ERROR, (char *)&err, &errlen); // check for error
  739. if ((rc!=0)&&!err)
  740. err = ERRNO(); // some implementations of getsockopt duff
  741. if (err==0) {
  742. nagling = true;
  743. set_nagle(false);
  744. state = ss_open;
  745. }
  746. else if ((err!=ETIMEDOUT)&&(err!=ECONNREFUSED)) // handled by caller
  747. LOGERR2(err,1,"post_connect");
  748. return err;
  749. }
  750. void CSocket::open(int listen_queue_size,bool reuseports)
  751. {
  752. if (IP6preferred)
  753. sock = ::socket(AF_INET6, connectionless()?SOCK_DGRAM:SOCK_STREAM, PF_INET6);
  754. else
  755. sock = ::socket(AF_INET, connectionless()?SOCK_DGRAM:SOCK_STREAM, 0);
  756. if (sock == INVALID_SOCKET) {
  757. THROWJSOCKEXCEPTION(ERRNO());
  758. }
  759. STATS.activesockets++;
  760. #ifdef SOCKTRACE
  761. PROGLOG("SOCKTRACE: opened socket %x %d (%p)", sock,sock,this);
  762. #endif
  763. if ((hostport==0)&&(sockmode==sm_udp)) {
  764. state = ss_open;
  765. #ifdef SOCKTRACE
  766. PROGLOG("SOCKTRACE: opened socket return udp");
  767. #endif
  768. set_inherit(false);
  769. return;
  770. }
  771. #ifndef _WIN32
  772. reuseports = true; // for some reason linux requires reuse ports
  773. #endif
  774. if (reuseports) {
  775. int on = 1;
  776. setsockopt( sock, SOL_SOCKET, SO_REUSEADDR, (char *)&on, sizeof(on));
  777. }
  778. DEFINE_SOCKADDR(u);
  779. socklen_t ul;
  780. if (hostname) {
  781. if (targetip.isNull()) {
  782. set_return_addr(hostport,hostname);
  783. targetip.ipset(returnep);
  784. }
  785. ul = setSockAddr(u,targetip,hostport);
  786. }
  787. else
  788. ul = setSockAddrAny(u,hostport);
  789. int saverr;
  790. if (::bind(sock, &u.sa, ul) != 0) {
  791. saverr = ERRNO();
  792. if (saverr==EADDRINUSE) { // don't log as error (some usages probe ports)
  793. ErrPortInUse:
  794. closesock();
  795. char msg[1024];
  796. sprintf(msg,"Target: %s, port = %d, Raised in: %s, line %d",tracename,(int)hostport,__FILE__, __LINE__);
  797. IJSOCK_Exception *e = new SocketException(JSOCKERR_port_in_use,msg);
  798. throw e;
  799. }
  800. else {
  801. closesock();
  802. THROWJSOCKEXCEPTION(saverr);
  803. }
  804. }
  805. if (!connectionless()) {
  806. if (::listen(sock, listen_queue_size) != 0) {
  807. saverr = ERRNO();
  808. if (saverr==EADDRINUSE)
  809. goto ErrPortInUse;
  810. closesock();
  811. THROWJSOCKEXCEPTION(saverr);
  812. }
  813. }
  814. if (mcastreq) {
  815. if (setsockopt(sock, IPPROTO_IP, IP_ADD_MEMBERSHIP,(char*)mcastreq, sizeof(*mcastreq))!=0) {
  816. saverr = ERRNO();
  817. closesock();
  818. THROWJSOCKEXCEPTION(saverr);
  819. }
  820. }
  821. state = ss_open;
  822. #ifdef SOCKTRACE
  823. PROGLOG("SOCKTRACE: opened socket return");
  824. #endif
  825. set_inherit(false);
  826. }
  827. ISocket* CSocket::accept(bool allowcancel)
  828. {
  829. if ((accept_cancel_state!=accept_not_cancelled) && allowcancel) {
  830. accept_cancel_state=accept_cancelled;
  831. return NULL;
  832. }
  833. if (state != ss_open) {
  834. ERRLOG("invalid accept, state = %d",(int)state);
  835. THROWJSOCKEXCEPTION(JSOCKERR_not_opened);
  836. }
  837. if (connectionless()) {
  838. THROWJSOCKEXCEPTION(JSOCKERR_connectionless_socket);
  839. }
  840. T_SOCKET newsock;
  841. loop {
  842. in_accept = true;
  843. newsock = (sock!=INVALID_SOCKET)?::accept(sock, NULL, NULL):INVALID_SOCKET;
  844. in_accept = false;
  845. #ifdef SOCKTRACE
  846. PROGLOG("SOCKTRACE: accept created socket %x %d (%p)", newsock,newsock,this);
  847. #endif
  848. if (newsock!=INVALID_SOCKET) {
  849. if ((sock==INVALID_SOCKET)||(accept_cancel_state==accept_cancel_pending)) {
  850. ::close(newsock);
  851. newsock=INVALID_SOCKET;
  852. }
  853. else {
  854. accept_cancel_state = accept_not_cancelled;
  855. break;
  856. }
  857. }
  858. int saverr;
  859. saverr = ERRNO();
  860. if ((sock==INVALID_SOCKET)||(accept_cancel_state==accept_cancel_pending)) {
  861. accept_cancel_state = accept_cancelled;
  862. if (allowcancel)
  863. return NULL;
  864. THROWJSOCKEXCEPTION(JSOCKERR_cancel_accept);
  865. }
  866. if (saverr != EINTRCALL) {
  867. accept_cancel_state = accept_not_cancelled;
  868. THROWJSOCKEXCEPTION(saverr);
  869. }
  870. }
  871. if (state != ss_open) {
  872. accept_cancel_state = accept_cancelled;
  873. if (allowcancel)
  874. return NULL;
  875. THROWJSOCKEXCEPTION(JSOCKERR_cancel_accept);
  876. }
  877. CSocket *ret = new CSocket(newsock,sm_tcp,true);
  878. ret->set_inherit(false);
  879. return ret;
  880. }
  881. void CSocket::set_linger(int lingertime)
  882. {
  883. struct linger l;
  884. l.l_onoff = (lingertime>=0)?1:0;
  885. l.l_linger = (lingertime>=0)?lingertime:0;
  886. if (setsockopt(sock, SOL_SOCKET, SO_LINGER, (char*)&l, sizeof(l)) != 0) {
  887. WARNLOG("Linger not set");
  888. }
  889. }
  890. void CSocket::set_keep_alive(bool set)
  891. {
  892. int on=set?1:0;
  893. if (setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE, (char*)&on, sizeof(on)) != 0) {
  894. WARNLOG("KeepAlive not set");
  895. }
  896. }
  897. int CSocket::name(char *retname,size32_t namemax)
  898. {
  899. if (!retname)
  900. namemax = 0;
  901. if (namemax)
  902. retname[0] = 0;
  903. if (state != ss_open) {
  904. THROWJSOCKEXCEPTION(JSOCKERR_not_opened);
  905. }
  906. DEFINE_SOCKADDR(u);
  907. socklen_t ul = sizeof(u);
  908. if (::getsockname(sock,&u.sa, &ul)<0) {
  909. THROWJSOCKEXCEPTION(ERRNO());
  910. }
  911. SocketEndpoint ep;
  912. getSockAddrEndpoint(u,ul,ep);
  913. if (namemax>=1)
  914. {
  915. StringBuffer s;
  916. ep.getIpText(s);
  917. if (namemax-1<s.length())
  918. s.setLength(namemax-1);
  919. memcpy(retname,s.str(),s.length()+1);
  920. }
  921. return ep.port;
  922. }
  923. int CSocket::peer_name(char *retname,size32_t namemax)
  924. {
  925. // should not raise exceptions
  926. int ret = 0;
  927. if (!retname)
  928. namemax = 0;
  929. if (namemax)
  930. retname[0] = 0;
  931. if (state != ss_open) {
  932. return -1; // don't log as used to test socket
  933. }
  934. StringBuffer s;
  935. if (sockmode==sm_udp_server) { // udp server
  936. returnep.getIpText(s);
  937. ret = returnep.port;
  938. }
  939. else {
  940. DEFINE_SOCKADDR(u);
  941. socklen_t ul = sizeof(u);
  942. if (::getpeername(sock,&u.sa, &ul)<0)
  943. return -1; // don't log as used to test socket
  944. SocketEndpoint ep;
  945. getSockAddrEndpoint(u,ul,ep);
  946. ep.getIpText(s);
  947. ret = ep.port;
  948. }
  949. if (namemax>1) {
  950. if (namemax-1<s.length())
  951. s.setLength(namemax-1);
  952. memcpy(retname,s.str(),s.length()+1);
  953. }
  954. return ret;
  955. }
  956. SocketEndpoint &CSocket::getPeerEndpoint(SocketEndpoint &ep)
  957. {
  958. if (state != ss_open) {
  959. THROWJSOCKEXCEPTION(JSOCKERR_not_opened);
  960. }
  961. StringBuffer s;
  962. if (sockmode==sm_udp_server) { // udp server
  963. ep.set(returnep);
  964. }
  965. else {
  966. DEFINE_SOCKADDR(u);
  967. socklen_t ul = sizeof(u);
  968. if (::getpeername(sock,&u.sa, &ul)<0) {
  969. DBGLOG("getpeername failed %d",ERRNO());
  970. ep.set(NULL, 0);
  971. }
  972. else
  973. getSockAddrEndpoint(u,ul,ep);
  974. }
  975. return ep;
  976. }
  977. IpAddress & CSocket::getPeerAddress(IpAddress &addr)
  978. {
  979. SocketEndpoint ep;
  980. getPeerEndpoint(ep);
  981. addr = ep;
  982. return addr;
  983. }
  984. void CSocket::set_return_addr(int port,const char *retname)
  985. {
  986. if (!returnep.ipset(retname)) {
  987. IJSOCK_Exception *e = new SocketException(JSOCKERR_bad_address); // don't use THROWJSOCKEXCEPTION here
  988. throw e;
  989. }
  990. returnep.port = port;
  991. }
  992. void CSocket::cancel_accept()
  993. {
  994. if (connectionless()) {
  995. THROWJSOCKEXCEPTION(JSOCKERR_connectionless_socket);
  996. }
  997. #ifdef SOCKTRACE
  998. PROGLOG("SOCKTRACE: Cancel accept socket %x %d (%p)", sock, sock, this);
  999. #endif
  1000. if (!in_accept) {
  1001. accept_cancel_state = accept_cancelled;
  1002. errclose();
  1003. return;
  1004. }
  1005. accept_cancel_state = accept_cancel_pending;
  1006. errclose(); // this should cause accept to terminate (not supported on all linux though)
  1007. #ifdef _WIN32
  1008. for (unsigned i=0;i<5;i++) { // windows closes on different lower priority thread
  1009. Sleep(i);
  1010. if (accept_cancel_state==accept_cancelled)
  1011. return;
  1012. }
  1013. #else
  1014. Sleep(0);
  1015. if (accept_cancel_state==accept_cancelled)
  1016. return;
  1017. #endif
  1018. // Wakeup listener using a connection
  1019. SocketEndpoint ep(hostport,queryHostIP());
  1020. Owned<CSocket> sock = new CSocket(ep,sm_tcp,NULL);
  1021. try {
  1022. sock->connect_timeout(100,true);
  1023. }
  1024. catch (IJSOCK_Exception *e) {
  1025. EXCLOG(e,"CSocket::cancel_eccept");
  1026. e->Release();
  1027. }
  1028. }
  1029. // ================================================================================
  1030. // connect versions
  1031. ISocket* ISocket::connect( const SocketEndpoint &ep )
  1032. {
  1033. // general connect
  1034. return ISocket::connect_wait(ep,DEFAULT_CONNECT_TIME);
  1035. }
  1036. inline void refused_sleep(CTimeMon &tm, unsigned &refuseddelay)
  1037. {
  1038. unsigned remaining;
  1039. if (!tm.timedout(&remaining)) {
  1040. if (refuseddelay<remaining/4) {
  1041. Sleep(refuseddelay);
  1042. if (refuseddelay<CONNECT_TIMEOUT_REFUSED_WAIT/2)
  1043. refuseddelay *=2;
  1044. else
  1045. refuseddelay = CONNECT_TIMEOUT_REFUSED_WAIT;
  1046. }
  1047. else
  1048. Sleep(remaining/4); // towards end of timeout approach gradually
  1049. }
  1050. }
  1051. bool CSocket::connect_timeout( unsigned timeout, bool noexception)
  1052. {
  1053. // simple connect with timeout (no fancy stuff!)
  1054. unsigned startt = usTick();
  1055. CTimeMon tm(timeout);
  1056. unsigned remaining;
  1057. unsigned refuseddelay = 1;
  1058. int err;
  1059. while (!tm.timedout(&remaining)) {
  1060. err = pre_connect(false);
  1061. if ((err == EINPROGRESS)||(err == EWOULDBLOCK)) {
  1062. T_FD_SET fds;
  1063. struct timeval tv;
  1064. XFD_ZERO(&fds);
  1065. FD_SET((unsigned)sock, &fds);
  1066. T_FD_SET except;
  1067. XFD_ZERO(&except);
  1068. FD_SET((unsigned)sock, &except);
  1069. tv.tv_sec = remaining / 1000;
  1070. tv.tv_usec = (remaining % 1000)*1000;
  1071. CHECKSOCKRANGE(sock);
  1072. int rc = ::select( sock + 1, NULL, (fd_set *)&fds, (fd_set *)&except, &tv );
  1073. if (rc>0) {
  1074. // select succeeded - return error from socket (0 if connected)
  1075. socklen_t errlen = sizeof(err);
  1076. rc = getsockopt(sock, SOL_SOCKET, SO_ERROR, (char *)&err, &errlen); // check for error
  1077. if ((rc!=0)&&!err)
  1078. err = ERRNO(); // some implementations of getsockopt duff
  1079. if (err) // probably ECONNREFUSED but treat all errors same
  1080. refused_sleep(tm,refuseddelay);
  1081. }
  1082. else if (rc<0) {
  1083. err = ERRNO();
  1084. LOGERR2(err,2,"::select");
  1085. }
  1086. }
  1087. if (err==0) {
  1088. err = post_connect();
  1089. if (err==0) {
  1090. STATS.connects++;
  1091. STATS.connecttime+=usTick()-startt;
  1092. #ifdef _TRACE
  1093. char peer[256];
  1094. peer[0] = 'C';
  1095. peer[1] = '!';
  1096. strcpy(peer+2,hostname?hostname:"(NULL)");
  1097. free(tracename);
  1098. tracename = strdup(peer);
  1099. #endif
  1100. return true;
  1101. }
  1102. }
  1103. errclose();
  1104. }
  1105. #ifdef SOCKTRACE
  1106. PROGLOG("connect_timeout: failed %d",err);
  1107. #endif
  1108. STATS.failedconnects++;
  1109. STATS.failedconnecttime+=usTick()-startt;
  1110. if (!noexception)
  1111. THROWJSOCKEXCEPTION(JSOCKERR_connection_failed);
  1112. return false;
  1113. }
  1114. ISocket* ISocket::connect_timeout(const SocketEndpoint &ep,unsigned timeout)
  1115. {
  1116. if (ep.isNull()||(ep.port==0))
  1117. THROWJSOCKEXCEPTION2(JSOCKERR_bad_address);
  1118. Owned<CSocket> sock = new CSocket(ep,sm_tcp,NULL);
  1119. sock->connect_timeout(timeout,false);
  1120. return sock.getClear();
  1121. }
  1122. #define POLLTIME 50
  1123. void CSocket::connect_wait(unsigned timems)
  1124. {
  1125. // simple connect with timeout (no fancy stuff!)
  1126. unsigned startt = usTick();
  1127. CTimeMon tm(timems);
  1128. bool exit = false;
  1129. int err;
  1130. unsigned refuseddelay = 1;
  1131. while (!exit) {
  1132. #ifdef CENTRAL_NODE_RANDOM_DELAY
  1133. ForEachItemIn(cn,CentralNodeArray) {
  1134. const SocketEndpoint &ep=CentralNodeArray.item(cn);
  1135. if (ep.ipequals(targetip)) {
  1136. unsigned sleeptime = getRandom() % 1000;
  1137. StringBuffer s;
  1138. ep.getIpText(s);
  1139. PrintLog("Connection to central node %s - sleeping %d milliseconds", s.str(), sleeptime);
  1140. Sleep(sleeptime);
  1141. break;
  1142. }
  1143. }
  1144. #endif
  1145. unsigned remaining;
  1146. exit = tm.timedout(&remaining);
  1147. bool blockselect = exit; // if last time round block
  1148. {
  1149. CriticalBlock block(crit);
  1150. if (++connectingcount>4)
  1151. blockselect = true;
  1152. }
  1153. err = pre_connect(blockselect);
  1154. if (blockselect) {
  1155. if (err&&!exit)
  1156. refused_sleep(tm,refuseddelay); // probably ECONNREFUSED but treat all errors same
  1157. }
  1158. else {
  1159. unsigned timeoutms = (exit||(remaining<10000))?10000:remaining;
  1160. unsigned polltime = 1;
  1161. while (!blockselect && ((err == EINPROGRESS)||(err == EWOULDBLOCK))) {
  1162. T_FD_SET fds;
  1163. struct timeval tv;
  1164. XFD_ZERO(&fds);
  1165. FD_SET((unsigned)sock, &fds);
  1166. T_FD_SET except;
  1167. XFD_ZERO(&except);
  1168. FD_SET((unsigned)sock, &except);
  1169. #ifdef BLOCK_POLLED_SINGLE_CONNECTS
  1170. tv.tv_sec = timeoutms / 1000;
  1171. tv.tv_usec = (timeoutms % 1000)*1000;
  1172. #else
  1173. tv.tv_sec = 0;
  1174. tv.tv_usec = 0;
  1175. #endif
  1176. CHECKSOCKRANGE(sock);
  1177. int rc = ::select( sock + 1, NULL, (fd_set *)&fds, (fd_set *)&except, &tv );
  1178. if (rc>0) {
  1179. // select succeeded - return error from socket (0 if connected)
  1180. socklen_t errlen = sizeof(err);
  1181. rc = getsockopt(sock, SOL_SOCKET, SO_ERROR, (char *)&err, &errlen); // check for error
  1182. if ((rc!=0)&&!err)
  1183. err = ERRNO(); // some implementations of getsockopt duff
  1184. if (err)
  1185. refused_sleep(tm,refuseddelay); // probably ECONNREFUSED but treat all errors same
  1186. break;
  1187. }
  1188. if (rc<0) {
  1189. err = ERRNO();
  1190. LOGERR2(err,2,"::select");
  1191. break;
  1192. }
  1193. if (!timeoutms) {
  1194. #ifdef SOCKTRACE
  1195. PROGLOG("connecttimeout: timed out");
  1196. #endif
  1197. err = -1;
  1198. break;
  1199. }
  1200. #ifdef BLOCK_POLLED_SINGLE_CONNECTS
  1201. break;
  1202. #else
  1203. if (timeoutms<polltime)
  1204. polltime = timeoutms;
  1205. Sleep(polltime); // sleeps 1-50ms (to let other threads run)
  1206. timeoutms -= polltime;
  1207. if (polltime>POLLTIME/2)
  1208. polltime = POLLTIME;
  1209. else
  1210. polltime *= 2;
  1211. #endif
  1212. }
  1213. }
  1214. {
  1215. CriticalBlock block(crit);
  1216. --connectingcount;
  1217. }
  1218. if (err==0) {
  1219. err = post_connect();
  1220. if (err==0) {
  1221. STATS.connects++;
  1222. STATS.connecttime+=usTick()-startt;
  1223. #ifdef _TRACE
  1224. char peer[256];
  1225. peer[0] = 'C';
  1226. peer[1] = '!';
  1227. strcpy(peer+2,hostname?hostname:"(NULL)");
  1228. free(tracename);
  1229. tracename = strdup(peer);
  1230. #endif
  1231. return;
  1232. }
  1233. }
  1234. errclose();
  1235. }
  1236. #ifdef SOCKTRACE
  1237. PROGLOG("connect_wait: failed %d",err);
  1238. #endif
  1239. STATS.failedconnects++;
  1240. STATS.failedconnecttime+=usTick()-startt;
  1241. THROWJSOCKEXCEPTION(JSOCKERR_connection_failed);
  1242. }
  1243. ISocket* ISocket::connect_wait( const SocketEndpoint &ep, unsigned timems)
  1244. {
  1245. if (ep.isNull()||(ep.port==0))
  1246. THROWJSOCKEXCEPTION2(JSOCKERR_bad_address);
  1247. Owned<CSocket> sock = new CSocket(ep,sm_tcp,NULL);
  1248. sock->connect_wait(timems);
  1249. return sock.getClear();
  1250. }
  1251. void CSocket::udpconnect()
  1252. {
  1253. DEFINE_SOCKADDR(u);
  1254. if (targetip.isNull()) {
  1255. set_return_addr(hostport,hostname);
  1256. targetip.ipset(returnep);
  1257. }
  1258. socklen_t ul = setSockAddr(u,targetip,hostport);
  1259. sock = ::socket(u.sa.sa_family, SOCK_DGRAM, targetip.isIp4()?0:PF_INET6);
  1260. #ifdef SOCKTRACE
  1261. PROGLOG("SOCKTRACE: udp connected socket %x %d (%p)", sock, sock, this);
  1262. #endif
  1263. STATS.activesockets++;
  1264. if (sock == INVALID_SOCKET) {
  1265. THROWJSOCKEXCEPTION(ERRNO());
  1266. }
  1267. int res = ::connect(sock, &u.sa, ul);
  1268. if (res != 0) { // works for UDP
  1269. closesock();
  1270. THROWJSOCKEXCEPTION(JSOCKERR_connection_failed);
  1271. }
  1272. nagling = false; // means nothing for UDP
  1273. state = ss_open;
  1274. #ifdef _TRACE
  1275. char peer[256];
  1276. peer[0] = 'C';
  1277. peer[1] = '!';
  1278. strcpy(peer+2,hostname?hostname:"(NULL)");
  1279. free(tracename);
  1280. tracename = strdup(peer);
  1281. #endif
  1282. }
  1283. int CSocket::wait_read(unsigned timeout)
  1284. {
  1285. int ret = 0;
  1286. while (sock!=INVALID_SOCKET) {
  1287. T_FD_SET fds;
  1288. XFD_ZERO(&fds);
  1289. FD_SET((unsigned)sock, &fds);
  1290. CHECKSOCKRANGE(sock);
  1291. if (timeout==WAIT_FOREVER) {
  1292. ret = ::select( sock + 1, (fd_set *)&fds, NULL, NULL, NULL );
  1293. }
  1294. else {
  1295. struct timeval tv;
  1296. tv.tv_sec = timeout / 1000;
  1297. tv.tv_usec = (timeout % 1000)*1000;
  1298. ret = ::select( sock + 1, (fd_set *)&fds, NULL, NULL, &tv );
  1299. }
  1300. if (ret==SOCKET_ERROR) {
  1301. int err = ERRNO();
  1302. if (err!=EINTRCALL) { // else retry (should adjust time but for our usage don't think it matters that much)
  1303. LOGERR2(err,1,"wait_read");
  1304. break;
  1305. }
  1306. }
  1307. else
  1308. break;
  1309. }
  1310. return ret;
  1311. }
  1312. int CSocket::wait_write(unsigned timeout)
  1313. {
  1314. int ret = 0;
  1315. while (sock!=INVALID_SOCKET) {
  1316. T_FD_SET fds;
  1317. XFD_ZERO(&fds);
  1318. FD_SET((unsigned)sock, &fds);
  1319. CHECKSOCKRANGE(sock);
  1320. if (timeout==WAIT_FOREVER) {
  1321. ret = ::select( sock + 1, NULL, (fd_set *)&fds, NULL, NULL );
  1322. }
  1323. else {
  1324. struct timeval tv;
  1325. tv.tv_sec = timeout / 1000;
  1326. tv.tv_usec = (timeout % 1000)*1000;
  1327. ret = ::select( sock + 1, NULL, (fd_set *)&fds, NULL, &tv );
  1328. }
  1329. if (ret==SOCKET_ERROR) {
  1330. int err = ERRNO();
  1331. if (err!=EINTRCALL) { // else retry (should adjust time but for our usage don't think it matters that much)
  1332. LOGERR2(err,1,"wait_write");
  1333. break;
  1334. }
  1335. }
  1336. else
  1337. break;
  1338. }
  1339. return ret;
  1340. }
  1341. void CSocket::readtms(void* buf, size32_t min_size, size32_t max_size, size32_t &size_read,
  1342. unsigned timeoutms)
  1343. {
  1344. if (timeoutms == WAIT_FOREVER) {
  1345. read(buf,min_size, max_size, size_read,WAIT_FOREVER);
  1346. return;
  1347. }
  1348. unsigned startt=usTick();
  1349. size_read = 0;
  1350. if (state != ss_open) {
  1351. THROWJSOCKEXCEPTION(JSOCKERR_not_opened);
  1352. }
  1353. unsigned start;
  1354. unsigned timeleft;
  1355. start = msTick();
  1356. timeleft = timeoutms;
  1357. do {
  1358. int rc = wait_read(timeleft);
  1359. if (rc < 0) {
  1360. THROWJSOCKEXCEPTION(ERRNO());
  1361. }
  1362. if (rc == 0) {
  1363. THROWJSOCKEXCEPTION(JSOCKERR_timeout_expired);
  1364. }
  1365. unsigned elapsed = (msTick()-start);
  1366. if (elapsed<timeoutms)
  1367. timeleft = timeoutms-elapsed;
  1368. else
  1369. timeleft = 0;
  1370. unsigned retrycount=100;
  1371. EintrRetry:
  1372. if (sockmode==sm_udp_server) { // udp server
  1373. DEFINE_SOCKADDR(u);
  1374. socklen_t ul=sizeof(u);
  1375. rc = recvfrom(sock, (char*)buf + size_read, max_size - size_read, 0, &u.sa,&ul);
  1376. getSockAddrEndpoint(u,ul,returnep);
  1377. }
  1378. else {
  1379. rc = recv(sock, (char*)buf + size_read, max_size - size_read, 0);
  1380. }
  1381. if (rc < 0) {
  1382. int err = ERRNO();
  1383. if (BADSOCKERR(err)) {
  1384. // don't think this should happen but convert to same as shutdown while investigation
  1385. LOGERR2(err,1,"Socket closed during read");
  1386. rc = 0;
  1387. }
  1388. else if ((err==EINTRCALL)&&(retrycount--!=0)) {
  1389. LOGERR2(err,1,"EINTR retrying");
  1390. goto EintrRetry;
  1391. }
  1392. else {
  1393. VStringBuffer errMsg("readtms(timeoutms=%d)", timeoutms);
  1394. LOGERR2(err,1,errMsg.str());
  1395. if ((err==ECONNRESET)||(err==EINTRCALL)||(err==ECONNABORTED)) {
  1396. errclose();
  1397. err = JSOCKERR_broken_pipe;
  1398. }
  1399. THROWJSOCKEXCEPTION(err);
  1400. }
  1401. }
  1402. if (rc == 0) {
  1403. state = ss_shutdown;
  1404. if (min_size==0)
  1405. break; // if min_read is 0 return 0 if socket closed
  1406. THROWJSOCKEXCEPTION(JSOCKERR_graceful_close);
  1407. }
  1408. size_read += rc;
  1409. } while (size_read < min_size);
  1410. STATS.reads++;
  1411. STATS.readsize += size_read;
  1412. STATS.readtime+=usTick()-startt;
  1413. }
  1414. void CSocket::read(void* buf, size32_t min_size, size32_t max_size, size32_t &size_read,
  1415. unsigned timeoutsecs)
  1416. {
  1417. unsigned startt=usTick();
  1418. size_read = 0;
  1419. unsigned start;
  1420. unsigned timeleft = 0;
  1421. if (state != ss_open) {
  1422. THROWJSOCKEXCEPTION(JSOCKERR_not_opened);
  1423. }
  1424. if (timeoutsecs != WAIT_FOREVER) {
  1425. start = (unsigned)time(NULL);
  1426. timeleft = timeoutsecs;
  1427. }
  1428. do {
  1429. int rc;
  1430. if (timeoutsecs != WAIT_FOREVER) {
  1431. rc = wait_read(timeleft*1000);
  1432. if (rc < 0) {
  1433. THROWJSOCKEXCEPTION(ERRNO());
  1434. }
  1435. if (rc == 0) {
  1436. THROWJSOCKEXCEPTION(JSOCKERR_timeout_expired);
  1437. }
  1438. unsigned elapsed = ((unsigned)time(NULL))-start;
  1439. if (elapsed<timeoutsecs)
  1440. timeleft = timeoutsecs-elapsed;
  1441. else
  1442. timeleft = 0;
  1443. }
  1444. unsigned retrycount=100;
  1445. EintrRetry:
  1446. if (sockmode==sm_udp_server) { // udp server
  1447. DEFINE_SOCKADDR(u);
  1448. socklen_t ul=sizeof(u.sin);
  1449. rc = recvfrom(sock, (char*)buf + size_read, max_size - size_read, 0, &u.sa,&ul);
  1450. getSockAddrEndpoint(u,ul,returnep);
  1451. }
  1452. else {
  1453. rc = recv(sock, (char*)buf + size_read, max_size - size_read, 0);
  1454. }
  1455. if (rc < 0) {
  1456. int err = ERRNO();
  1457. if (BADSOCKERR(err)) {
  1458. // don't think this should happen but convert to same as shutdown while investigation
  1459. LOGERR2(err,3,"Socket closed during read");
  1460. rc = 0;
  1461. }
  1462. else if ((err==EINTRCALL)&&(retrycount--!=0)) {
  1463. if (sock==INVALID_SOCKET)
  1464. rc = 0; // convert an EINTR after closed to a graceful close
  1465. else {
  1466. LOGERR2(err,3,"EINTR retrying");
  1467. goto EintrRetry;
  1468. }
  1469. }
  1470. else {
  1471. LOGERR2(err,3,"read");
  1472. if ((err==ECONNRESET)||(err==EINTRCALL)||(err==ECONNABORTED)) {
  1473. errclose();
  1474. err = JSOCKERR_broken_pipe;
  1475. }
  1476. THROWJSOCKEXCEPTION(err);
  1477. }
  1478. }
  1479. if (rc == 0) {
  1480. state = ss_shutdown;
  1481. if (min_size==0)
  1482. break; // if min_read is 0 return 0 if socket closed
  1483. THROWJSOCKEXCEPTION(JSOCKERR_graceful_close);
  1484. }
  1485. size_read += rc;
  1486. } while (size_read < min_size);
  1487. STATS.reads++;
  1488. STATS.readsize += size_read;
  1489. STATS.readtime+=usTick()-startt;
  1490. }
  1491. void CSocket::read(void* buf, size32_t size)
  1492. {
  1493. if (!size)
  1494. return;
  1495. unsigned startt=usTick();
  1496. size32_t size_read=size;
  1497. if (state != ss_open) {
  1498. THROWJSOCKEXCEPTION(JSOCKERR_not_opened);
  1499. }
  1500. do {
  1501. unsigned retrycount=100;
  1502. EintrRetry:
  1503. int rc;
  1504. if (sockmode==sm_udp_server) { // udp server
  1505. DEFINE_SOCKADDR(u);
  1506. socklen_t ul=sizeof(u.sin);
  1507. rc = recvfrom(sock, (char*)buf, size, 0, &u.sa,&ul);
  1508. getSockAddrEndpoint(u,ul,returnep);
  1509. }
  1510. else {
  1511. rc = recv(sock, (char*)buf, size, 0);
  1512. }
  1513. if (rc < 0) {
  1514. int err = ERRNO();
  1515. if (BADSOCKERR(err)) {
  1516. // don't think this should happen but convert to same as shutdown while investigation
  1517. LOGERR2(err,5,"Socket closed during read");
  1518. rc = 0;
  1519. }
  1520. else if ((err==EINTRCALL)&&(retrycount--!=0)) {
  1521. LOGERR2(err,5,"EINTR retrying");
  1522. goto EintrRetry;
  1523. }
  1524. else {
  1525. LOGERR2(err,5,"read");
  1526. if ((err==ECONNRESET)||(err==EINTRCALL)||(err==ECONNABORTED)) {
  1527. errclose();
  1528. err = JSOCKERR_broken_pipe;
  1529. }
  1530. THROWJSOCKEXCEPTION(err);
  1531. }
  1532. }
  1533. if (rc == 0) {
  1534. state = ss_shutdown;
  1535. THROWJSOCKEXCEPTION(JSOCKERR_graceful_close);
  1536. }
  1537. buf = (char*)buf + rc;
  1538. size -= rc;
  1539. } while (size != 0);
  1540. STATS.reads++;
  1541. STATS.readsize += size_read;
  1542. STATS.readtime+=usTick()-startt;
  1543. }
  1544. size32_t CSocket::write(void const* buf, size32_t size)
  1545. {
  1546. if (size==0)
  1547. return 0;
  1548. unsigned startt=usTick();
  1549. size32_t size_writ = size;
  1550. if (state != ss_open) {
  1551. THROWJSOCKEXCEPTION(JSOCKERR_not_opened);
  1552. }
  1553. size32_t res=0;
  1554. do {
  1555. unsigned retrycount=100;
  1556. EintrRetry:
  1557. int rc;
  1558. if (sockmode==sm_udp_server) { // udp server
  1559. DEFINE_SOCKADDR(u);
  1560. socklen_t ul = setSockAddr(u,returnep,returnep.port);
  1561. rc = sendto(sock, (char*)buf, size, 0, &u.sa, ul);
  1562. }
  1563. else {
  1564. rc = send(sock, (char*)buf, size, SEND_FLAGS);
  1565. }
  1566. if (rc < 0) {
  1567. int err=ERRNO();
  1568. if (BADSOCKERR(err)) {
  1569. LOGERR2(err,7,"Socket closed during write");
  1570. rc = 0;
  1571. }
  1572. else if ((err==EINTRCALL)&&(retrycount--!=0)) {
  1573. LOGERR2(err,7,"EINTR retrying");
  1574. goto EintrRetry;
  1575. }
  1576. else {
  1577. if (((sockmode==sm_multicast)||(sockmode==sm_udp))&&(err==ECONNREFUSED))
  1578. break; // ignore
  1579. LOGERR2(err,7,"write");
  1580. if ((err==ECONNRESET)||(err==EINTRCALL)||(err==ECONNABORTED)
  1581. #ifndef _WIN32
  1582. ||(err==EPIPE)||(err==ETIMEDOUT) // linux can raise these on broken pipe
  1583. #endif
  1584. ) {
  1585. errclose();
  1586. err = JSOCKERR_broken_pipe;
  1587. }
  1588. if ((err == EWOULDBLOCK) && nonblocking)
  1589. break;
  1590. THROWJSOCKEXCEPTION(err);
  1591. }
  1592. }
  1593. res += rc;
  1594. if (rc == 0) {
  1595. state = ss_shutdown;
  1596. THROWJSOCKEXCEPTION(JSOCKERR_graceful_close);
  1597. }
  1598. if (nonblocking)
  1599. break;
  1600. buf = (char*)buf + rc;
  1601. size -= rc;
  1602. } while (size != 0);
  1603. STATS.writes++;
  1604. STATS.writesize += size_writ;
  1605. STATS.writetime+=usTick()-startt;
  1606. return res;
  1607. }
  1608. size32_t CSocket::writetms(void const* buf, size32_t size, unsigned timeoutms)
  1609. {
  1610. if (size==0)
  1611. return 0;
  1612. if (state != ss_open)
  1613. {
  1614. THROWJSOCKEXCEPTION(JSOCKERR_not_opened);
  1615. }
  1616. if (timeoutms == WAIT_FOREVER)
  1617. return write(buf, size);
  1618. const char *p = (const char *)buf;
  1619. unsigned start, elapsed;
  1620. start = msTick();
  1621. elapsed = 0;
  1622. size32_t nwritten = 0;
  1623. size32_t nleft = size;
  1624. unsigned rollover = 0;
  1625. bool prevblock = set_nonblock(true);
  1626. while ( (nwritten < size) && (elapsed <= timeoutms) )
  1627. {
  1628. size32_t amnt = write(p,nleft);
  1629. if ( ((amnt == (size32_t)-1) || (amnt == 0)) && (++rollover >= 20) )
  1630. {
  1631. rollover = 0;
  1632. Sleep(20);
  1633. }
  1634. else
  1635. {
  1636. nwritten += amnt;
  1637. nleft -= amnt;
  1638. p += amnt;
  1639. }
  1640. elapsed = msTick() - start;
  1641. }
  1642. set_nonblock(prevblock);
  1643. if (nwritten < size)
  1644. {
  1645. ERRLOG("writetms timed out; timeout: %u, nwritten: %u, size: %u", timeoutms, nwritten, size);
  1646. THROWJSOCKEXCEPTION(JSOCKERR_timeout_expired);
  1647. }
  1648. return nwritten;
  1649. }
  1650. bool CSocket::check_connection()
  1651. {
  1652. if (state != ss_open)
  1653. return false;
  1654. unsigned retrycount=100;
  1655. EintrRetry:
  1656. int rc;
  1657. if (sockmode==sm_udp_server) { // udp server
  1658. DEFINE_SOCKADDR(u);
  1659. socklen_t ul = setSockAddr(u,returnep,returnep.port);
  1660. rc = sendto(sock, NULL, 0, 0, &u.sa, ul);
  1661. }
  1662. else {
  1663. rc = send(sock, NULL, 0, SEND_FLAGS);
  1664. }
  1665. if (rc < 0) {
  1666. int err=ERRNO();
  1667. if ((err==EINTRCALL)&&(retrycount--!=0)) {
  1668. LOGERR2(err,7,"EINTR retrying");
  1669. goto EintrRetry;
  1670. }
  1671. else
  1672. return false;
  1673. }
  1674. return true;
  1675. }
  1676. size32_t CSocket::udp_write_to(const SocketEndpoint &ep, void const* buf, size32_t size)
  1677. {
  1678. if (size==0)
  1679. return 0;
  1680. unsigned startt=usTick();
  1681. size32_t size_writ = size;
  1682. if (state != ss_open) {
  1683. THROWJSOCKEXCEPTION(JSOCKERR_not_opened);
  1684. }
  1685. size32_t res=0;
  1686. DEFINE_SOCKADDR(u);
  1687. loop {
  1688. socklen_t ul = setSockAddr(u,ep,ep.port);
  1689. int rc = sendto(sock, (char*)buf, size, 0, &u.sa, ul);
  1690. if (rc < 0) {
  1691. int err=ERRNO();
  1692. if (((sockmode==sm_multicast)||(sockmode==sm_udp))&&(err==ECONNREFUSED))
  1693. break; // ignore
  1694. if (err!=EINTRCALL) {
  1695. THROWJSOCKEXCEPTION(err);
  1696. }
  1697. }
  1698. else {
  1699. res = (size32_t)rc;
  1700. break;
  1701. }
  1702. }
  1703. STATS.writes++;
  1704. STATS.writesize += res;
  1705. STATS.writetime+=usTick()-startt;
  1706. return res;
  1707. }
  1708. size32_t CSocket::write_multiple(unsigned num,const void **buf, size32_t *size)
  1709. {
  1710. assertex(sockmode!=sm_udp_server);
  1711. assertex(!nonblocking);
  1712. if (num==1)
  1713. return write(buf[0],size[0]);
  1714. size32_t total = 0;
  1715. unsigned i;
  1716. for (i=0;i<num;i++)
  1717. total += size[i];
  1718. if (total==0)
  1719. return 0;
  1720. unsigned startt=usTick();
  1721. if (state != ss_open) {
  1722. THROWJSOCKEXCEPTION(JSOCKERR_not_opened);
  1723. }
  1724. size32_t res=0;
  1725. #ifdef _WIN32
  1726. WSABUF *bufs = (WSABUF *)alloca(sizeof(WSABUF)*num);
  1727. for (i=0;i<num;i++) {
  1728. bufs[i].buf = (char *)buf[i];
  1729. bufs[i].len = size[i];
  1730. }
  1731. unsigned retrycount=100;
  1732. EintrRetry:
  1733. DWORD sent;
  1734. if (WSASendTo(sock,bufs,num,&sent,0,NULL,0,NULL,NULL)==SOCKET_ERROR) {
  1735. int err=ERRNO();
  1736. if (BADSOCKERR(err)) {
  1737. LOGERR2(err,8,"Socket closed during write");
  1738. sent = 0;
  1739. }
  1740. else if ((err==EINTRCALL)&&(retrycount--!=0)) {
  1741. LOGERR2(err,8,"EINTR retrying");
  1742. goto EintrRetry;
  1743. }
  1744. else {
  1745. LOGERR2(err,8,"write_multiple");
  1746. if ((err==ECONNRESET)||(err==EINTRCALL)||(err==ECONNABORTED)||(err==ETIMEDOUT)) {
  1747. errclose();
  1748. err = JSOCKERR_broken_pipe;
  1749. }
  1750. THROWJSOCKEXCEPTION(err);
  1751. }
  1752. }
  1753. if (sent == 0) {
  1754. state = ss_shutdown;
  1755. THROWJSOCKEXCEPTION(JSOCKERR_graceful_close);
  1756. }
  1757. res = sent;
  1758. #else
  1759. #ifdef USE_CORK
  1760. if (total>1024) {
  1761. class Copt
  1762. {
  1763. T_SOCKET sock;
  1764. bool nagling;
  1765. public:
  1766. Copt(T_SOCKET _sock,bool _nagling)
  1767. {
  1768. nagling = _nagling;
  1769. int enabled = 1;
  1770. int disabled = 0;
  1771. if (!nagling)
  1772. setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, (char*)&disabled, sizeof(disabled));
  1773. setsockopt(sock, IPPROTO_TCP, TCP_CORK, (char*)&enabled, sizeof(enabled));
  1774. }
  1775. ~Copt()
  1776. {
  1777. int enabled = 1;
  1778. int disabled = 0;
  1779. setsockopt(sock, IPPROTO_TCP, TCP_CORK, (char*)&disabled, sizeof(disabled));
  1780. if (!nagling)
  1781. setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, (char*)&enabled, sizeof(enabled));
  1782. }
  1783. } copt(sock,nagling);
  1784. for (i=0;i<num;i++)
  1785. res += write(buf[i],size[i]);
  1786. }
  1787. else {
  1788. byte b[1024];
  1789. byte *p=b;
  1790. for (i=0;i<num;i++) {
  1791. memcpy(p,buf[i],size[i]);
  1792. p += size[i];
  1793. }
  1794. res = write(b,total);
  1795. }
  1796. #else
  1797. // send in equal chunks about 64K
  1798. unsigned n = (total+0xffff)/0x10000;
  1799. size32_t outbufsize = (total+n-1)/n;
  1800. MemoryAttr ma;
  1801. byte *outbuf = (byte *)ma.allocate(outbufsize);
  1802. size32_t outwr = 0;
  1803. i = 0;
  1804. size32_t os = 0;
  1805. size32_t left = total;
  1806. byte *b = NULL;
  1807. size32_t s=0;
  1808. loop {
  1809. while (!s&&(i<num)) {
  1810. b = (byte *)buf[i];
  1811. s = size[i];
  1812. i++;
  1813. }
  1814. if ((os==0)&&(s==left)) {
  1815. write(b,s); // go for it
  1816. break;
  1817. }
  1818. else {
  1819. size32_t cpy = outbufsize-os;
  1820. if (cpy>s)
  1821. cpy = s;
  1822. memcpy(outbuf+os,b,cpy);
  1823. os += cpy;
  1824. left -= cpy;
  1825. s -= cpy;
  1826. b += cpy;
  1827. if (left==0) {
  1828. write(outbuf,os);
  1829. break;
  1830. }
  1831. else if (os==outbufsize) {
  1832. write(outbuf,os);
  1833. os = 0;
  1834. }
  1835. }
  1836. }
  1837. #endif
  1838. #endif
  1839. STATS.writes++;
  1840. STATS.writesize += res;
  1841. STATS.writetime+=usTick()-startt;
  1842. return res;
  1843. }
  1844. bool CSocket::send_block(const void *blk,size32_t sz)
  1845. {
  1846. unsigned startt=usTick();
  1847. #ifdef TRACE_SLOW_BLOCK_TRANSFER
  1848. unsigned startt2 = startt;
  1849. unsigned startt3 = startt;
  1850. #endif
  1851. if (blockflags&BF_SYNC_TRANSFER_PULL) {
  1852. size32_t rd;
  1853. bool eof = true;
  1854. readtms(&eof,sizeof(eof),sizeof(eof),rd,blocktimeoutms);
  1855. if (eof)
  1856. return false;
  1857. #ifdef TRACE_SLOW_BLOCK_TRANSFER
  1858. startt2=usTick();
  1859. #endif
  1860. }
  1861. if (!blk||!sz) {
  1862. sz = 0;
  1863. write(&sz,sizeof(sz));
  1864. try {
  1865. bool reply;
  1866. size32_t rd;
  1867. readtms(&reply,sizeof(reply),sizeof(reply),rd,blocktimeoutms);
  1868. }
  1869. catch (IJSOCK_Exception *e) {
  1870. if ((e->errorCode()!=JSOCKERR_broken_pipe)&&(e->errorCode()!=JSOCKERR_graceful_close))
  1871. EXCLOG(e,"CSocket::send_block");
  1872. e->Release();
  1873. }
  1874. return false;
  1875. }
  1876. size32_t rsz=sz;
  1877. _WINREV(rsz);
  1878. write(&rsz,sizeof(rsz));
  1879. if (blockflags&BF_SYNC_TRANSFER_PUSH) {
  1880. #ifdef TRACE_SLOW_BLOCK_TRANSFER
  1881. startt2=usTick();
  1882. #endif
  1883. size32_t rd;
  1884. bool eof = true;
  1885. readtms(&eof,sizeof(eof),sizeof(eof),rd,blocktimeoutms);
  1886. if (eof)
  1887. return false;
  1888. #ifdef TRACE_SLOW_BLOCK_TRANSFER
  1889. startt3=usTick();
  1890. #endif
  1891. }
  1892. write(blk,sz);
  1893. if (blockflags&BF_RELIABLE_TRANSFER) {
  1894. bool isok=false;
  1895. size32_t rd;
  1896. readtms(&isok,sizeof(isok),sizeof(isok),rd,blocktimeoutms);
  1897. if (!isok)
  1898. return false;
  1899. }
  1900. unsigned nowt = usTick();
  1901. unsigned elapsed = nowt-startt;
  1902. STATS.blocksendtime+=elapsed;
  1903. STATS.numblocksends++;
  1904. STATS.blocksendsize+=sz;
  1905. if (elapsed>STATS.longestblocksend) {
  1906. STATS.longestblocksend = elapsed;
  1907. STATS.longestblocksize = sz;
  1908. }
  1909. #ifdef TRACE_SLOW_BLOCK_TRANSFER
  1910. static unsigned lastreporttime=0;
  1911. static unsigned lastexceeded=0;
  1912. if (elapsed>1000000*60) { // over 1min
  1913. unsigned t = msTick();
  1914. if (1) { //((t-lastreporttime>1000*60) || // only report once per min
  1915. // (elapsed>lastexceeded*2)) {
  1916. lastexceeded = elapsed;
  1917. lastreporttime = t;
  1918. WARNLOG("send_block took %ds to %s (%d,%d,%d)",elapsed/1000000,tracename,startt2-startt,startt3-startt2,nowt-startt3);
  1919. }
  1920. }
  1921. #endif
  1922. return true;
  1923. }
  1924. #ifdef USERECVSEM
  1925. class CSemProtect
  1926. {
  1927. Semaphore *sem;
  1928. bool *owned;
  1929. public:
  1930. CSemProtect() { clear(); }
  1931. ~CSemProtect()
  1932. {
  1933. if (sem&&*owned) {
  1934. *owned = false;
  1935. sem->signal();
  1936. }
  1937. }
  1938. void set(Semaphore *_sem,bool *_owned)
  1939. {
  1940. sem = _sem;
  1941. owned = _owned;
  1942. }
  1943. bool wait(Semaphore *_sem,bool *_owned,unsigned timeout) {
  1944. if (!*_owned&&!_sem->wait(timeout))
  1945. return false;
  1946. *_owned = true;
  1947. set(_sem,_owned);
  1948. return true;
  1949. }
  1950. void clear() { sem = NULL; owned = NULL; }
  1951. };
  1952. #endif
  1953. size32_t CSocket::receive_block_size()
  1954. {
  1955. // assumed always paired with receive_block
  1956. if (nextblocksize) {
  1957. if (blockflags&BF_SYNC_TRANSFER_PULL) {
  1958. bool eof=false;
  1959. write(&eof,sizeof(eof));
  1960. }
  1961. size32_t rd;
  1962. readtms(&nextblocksize,sizeof(nextblocksize),sizeof(nextblocksize),rd,blocktimeoutms);
  1963. _WINREV(nextblocksize);
  1964. if (nextblocksize==0) { // confirm eof
  1965. try {
  1966. bool confirm=true;
  1967. write(&confirm,sizeof(confirm));
  1968. }
  1969. catch (IJSOCK_Exception *e) {
  1970. if ((e->errorCode()!=JSOCKERR_broken_pipe)&&(e->errorCode()!=JSOCKERR_graceful_close))
  1971. EXCLOG(e,"receive_block_size");
  1972. e->Release();
  1973. }
  1974. }
  1975. else if (blockflags&BF_SYNC_TRANSFER_PUSH) { // leaves receiveblocksem clear
  1976. #ifdef USERECVSEM
  1977. CSemProtect semprot; // this will catch exception in write
  1978. while (!semprot.wait(&receiveblocksem,&receiveblocksemowned,60*1000*5))
  1979. WARNLOG("Receive block stalled");
  1980. #endif
  1981. bool eof=false;
  1982. write(&eof,sizeof(eof));
  1983. #ifdef USERECVSEM
  1984. semprot.clear();
  1985. #endif
  1986. }
  1987. }
  1988. return nextblocksize;
  1989. }
  1990. size32_t CSocket::receive_block(void *blk,size32_t maxsize)
  1991. {
  1992. #ifdef USERECVSEM
  1993. CSemProtect semprot; // this will catch exceptions
  1994. #endif
  1995. size32_t sz = nextblocksize;
  1996. if (sz) {
  1997. if (sz==UINT_MAX) { // need to get size
  1998. if (!blk||!maxsize) {
  1999. if (blockflags&BF_SYNC_TRANSFER_PUSH) { // ignore block size
  2000. size32_t rd;
  2001. readtms(&nextblocksize,sizeof(nextblocksize),sizeof(nextblocksize),rd,blocktimeoutms);
  2002. }
  2003. if (blockflags&(BF_SYNC_TRANSFER_PULL|BF_SYNC_TRANSFER_PUSH)) { // signal eof
  2004. bool eof=true;
  2005. write(&eof,sizeof(eof));
  2006. nextblocksize = 0;
  2007. return 0;
  2008. }
  2009. }
  2010. sz = receive_block_size();
  2011. if (!sz)
  2012. return 0;
  2013. }
  2014. unsigned startt=usTick(); // include sem block but not initial handshake
  2015. #ifdef USERECVSEM
  2016. if (blockflags&BF_SYNC_TRANSFER_PUSH) // read_block_size sets semaphore
  2017. semprot.set(&receiveblocksem,&receiveblocksemowned); // this will reset semaphore on exit
  2018. #endif
  2019. nextblocksize = UINT_MAX;
  2020. size32_t rd;
  2021. if (sz<=maxsize) {
  2022. readtms(blk,sz,sz,rd,blocktimeoutms);
  2023. }
  2024. else { // truncate
  2025. readtms(blk,maxsize,maxsize,rd,blocktimeoutms);
  2026. sz -= maxsize;
  2027. void *tmp=malloc(sz);
  2028. readtms(tmp,sz,sz,rd,blocktimeoutms);
  2029. free(tmp);
  2030. sz = maxsize;
  2031. }
  2032. if (blockflags&BF_RELIABLE_TRANSFER) {
  2033. bool isok=true;
  2034. write(&isok,sizeof(isok));
  2035. }
  2036. unsigned elapsed = usTick()-startt;
  2037. STATS.blockrecvtime+=elapsed;
  2038. STATS.numblockrecvs++;
  2039. STATS.blockrecvsize+=sz;
  2040. }
  2041. return sz;
  2042. }
  2043. void CSocket::set_block_mode(unsigned flags, size32_t recsize, unsigned _timeoutms)
  2044. {
  2045. blockflags = flags;
  2046. nextblocksize = UINT_MAX;
  2047. blocktimeoutms = _timeoutms?_timeoutms:WAIT_FOREVER;
  2048. }
  2049. void CSocket::shutdown(unsigned mode)
  2050. {
  2051. if (state == ss_open) {
  2052. state = ss_shutdown;
  2053. #ifdef SOCKTRACE
  2054. PROGLOG("SOCKTRACE: shutdown(%d) socket %x %d (%p)", mode, sock, sock, this);
  2055. #endif
  2056. int rc = ::shutdown(sock, mode);
  2057. if (rc != 0) {
  2058. int err=ERRNO();
  2059. if (err==ENOTCONN) {
  2060. LOGERR2(err,9,"shutdown");
  2061. err = JSOCKERR_broken_pipe;
  2062. }
  2063. THROWJSOCKEXCEPTION(err);
  2064. }
  2065. }
  2066. }
  2067. void CSocket::errclose()
  2068. {
  2069. #ifdef USERECVSEM
  2070. if (receiveblocksemowned) {
  2071. receiveblocksemowned = false;
  2072. receiveblocksem.signal();
  2073. }
  2074. #endif
  2075. if (state != ss_close) {
  2076. state = ss_close;
  2077. #ifdef SOCKTRACE
  2078. PROGLOG("SOCKTRACE: errclose socket %x %d (%p)", sock, sock, this);
  2079. #endif
  2080. if (mcastreq)
  2081. setsockopt(sock, IPPROTO_IP, IP_DROP_MEMBERSHIP,(char*)mcastreq,sizeof(*mcastreq));
  2082. closesock();
  2083. }
  2084. }
  2085. void CSocket::close()
  2086. {
  2087. #ifdef USERECVSEM
  2088. if (receiveblocksemowned) {
  2089. receiveblocksemowned = false;
  2090. receiveblocksem.signal();
  2091. }
  2092. #endif
  2093. if (state != ss_close) {
  2094. #ifdef SOCKTRACE
  2095. PROGLOG("SOCKTRACE: close socket %x %d (%p)", sock, sock, this);
  2096. #endif
  2097. state = ss_close;
  2098. if (mcastreq)
  2099. setsockopt(sock, IPPROTO_IP, IP_DROP_MEMBERSHIP,(char*)mcastreq,sizeof(*mcastreq));
  2100. if (closesock() != 0) {
  2101. THROWJSOCKEXCEPTION(ERRNO());
  2102. }
  2103. }
  2104. }
  2105. size32_t CSocket::get_max_send_size()
  2106. {
  2107. size32_t maxsend=0;
  2108. socklen_t size = sizeof(maxsend);
  2109. #if _WIN32
  2110. getsockopt(sock, SOL_SOCKET, SO_MAX_MSG_SIZE, (char *) &maxsend, &size);
  2111. #else
  2112. getsockopt(sock, SOL_SOCKET, SO_SNDBUF, (char *) &maxsend, &size); // not the same but closest I can find
  2113. #endif
  2114. return maxsend;
  2115. }
  2116. size32_t CSocket::get_send_buffer_size()
  2117. {
  2118. size32_t maxsend=0;
  2119. socklen_t size = sizeof(maxsend);
  2120. getsockopt(sock, SOL_SOCKET, SO_SNDBUF, (char *) &maxsend, &size);
  2121. return maxsend;
  2122. }
  2123. void CSocket::set_send_buffer_size(size32_t maxsend)
  2124. {
  2125. if (setsockopt(sock, SOL_SOCKET, SO_SNDBUF, (char *)&maxsend, sizeof(maxsend))!=0) {
  2126. LOGERR2(ERRNO(),1,"setsockopt(SO_SNDBUF)");
  2127. }
  2128. #ifdef CHECKBUFSIZE
  2129. size32_t v;
  2130. if (getsockopt(sock, SOL_SOCKET, SO_SNDBUF, (char *)&v, sizeof(v))!=0) {
  2131. LOGERR2(ERRNO(),1,"getsockopt(SO_SNDBUF)");
  2132. }
  2133. if (v!=maxsend)
  2134. WARNLOG("set_send_buffer_size requested %d, got %d",maxsend,v);
  2135. #endif
  2136. }
  2137. size32_t CSocket::get_receive_buffer_size()
  2138. {
  2139. size32_t max=0;
  2140. socklen_t size = sizeof(max);
  2141. getsockopt(sock, SOL_SOCKET, SO_RCVBUF, (char *) &max, &size);
  2142. return max;
  2143. }
  2144. void CSocket::set_receive_buffer_size(size32_t max)
  2145. {
  2146. if (setsockopt(sock, SOL_SOCKET, SO_RCVBUF, (char *)&max, sizeof(max))!=0) {
  2147. LOGERR2(ERRNO(),1,"setsockopt(SO_RCVBUF)");
  2148. }
  2149. #ifdef CHECKBUFSIZE
  2150. size32_t v;
  2151. if (getsockopt(sock, SOL_SOCKET, SO_RCVBUF, (char *)&v, sizeof(v))!=0) {
  2152. LOGERR2(ERRNO(),1,"getsockopt(SO_RCVBUF)");
  2153. }
  2154. if (v<max)
  2155. WARNLOG("set_receive_buffer_size requested %d, got %d",max,v);
  2156. #endif
  2157. }
  2158. bool CSocket::join_multicast_group(SocketEndpoint &ep)
  2159. {
  2160. StringBuffer s;
  2161. ep.getIpText(s); // will improve later
  2162. MCASTREQ req(s.str());
  2163. if (setsockopt(sock, IPPROTO_IP, IP_ADD_MEMBERSHIP,(char*)&req, sizeof(req))!=0) {
  2164. return false;
  2165. }
  2166. return true;
  2167. }
  2168. bool CSocket::leave_multicast_group(SocketEndpoint &ep)
  2169. {
  2170. StringBuffer s;
  2171. ep.getIpText(s); // will improve later
  2172. MCASTREQ req(s.str());
  2173. if (setsockopt(sock, IPPROTO_IP, IP_DROP_MEMBERSHIP,(char*)&req, sizeof(req))!=0) {
  2174. return false;
  2175. }
  2176. return true;
  2177. }
  2178. CSocket::~CSocket()
  2179. {
  2180. if (owned)
  2181. close();
  2182. free(hostname);
  2183. hostname = NULL;
  2184. #ifdef _TRACE
  2185. free(tracename);
  2186. tracename = NULL;
  2187. #endif
  2188. delete mcastreq;
  2189. }
  2190. CSocket::CSocket(const SocketEndpoint &ep,SOCKETMODE smode,const char *name)
  2191. {
  2192. state = ss_close;
  2193. nonblocking = false;
  2194. #ifdef USERECVSEM
  2195. receiveblocksemowned = false;
  2196. #endif
  2197. nagling = true; // until turned off
  2198. hostport = ep.port;
  2199. hostname = NULL;
  2200. mcastreq = NULL;
  2201. tracename = NULL;
  2202. StringBuffer tmp;
  2203. if ((smode==sm_multicast_server)&&(name&&*name)) {
  2204. mcastreq = new MCASTREQ(name);
  2205. }
  2206. else {
  2207. if (!name&&!ep.isNull())
  2208. name = ep.getIpText(tmp).str();
  2209. hostname = name?strdup(name):NULL;
  2210. }
  2211. sock = INVALID_SOCKET;
  2212. sockmode = smode;
  2213. owned = true;
  2214. nextblocksize = 0;
  2215. in_accept = false;
  2216. accept_cancel_state = accept_not_cancelled;
  2217. #ifdef _TRACE
  2218. char peer[256];
  2219. peer[0] = name?'T':'S';
  2220. peer[1] = '>';
  2221. if (name)
  2222. strcpy(peer+2,name);
  2223. else {
  2224. SocketEndpoint self;
  2225. self.setLocalHost(0);
  2226. self.getUrlStr(peer+2,sizeof(peer)-2);
  2227. }
  2228. tracename = strdup(peer);
  2229. #endif
  2230. }
  2231. CSocket::CSocket(T_SOCKET new_sock,SOCKETMODE smode,bool _owned)
  2232. {
  2233. nonblocking = false;
  2234. #ifdef USERECVSEM
  2235. receiveblocksemowned = false;
  2236. #endif
  2237. nagling = true; // until turned off
  2238. sock = new_sock;
  2239. if (new_sock!=INVALID_SOCKET)
  2240. STATS.activesockets++;
  2241. hostname = NULL;
  2242. mcastreq = NULL;
  2243. hostport = 0;
  2244. tracename = NULL;
  2245. state = ss_open;
  2246. sockmode = smode;
  2247. owned = _owned;
  2248. nextblocksize = 0;
  2249. in_accept = false;
  2250. accept_cancel_state = accept_not_cancelled;
  2251. set_nagle(false);
  2252. //set_linger(DEFAULT_LINGER_TIME); -- experiment with removing this as closesocket should still endevour to send outstanding data
  2253. #ifdef _TRACE
  2254. char peer[256];
  2255. peer[0] = 'A';
  2256. peer[1] = '!';
  2257. peer_name(peer+2,sizeof(peer)-2);
  2258. tracename = strdup(peer);
  2259. #endif
  2260. }
  2261. ISocket* ISocket::create(unsigned short p,int listen_queue_size)
  2262. {
  2263. if (p==0)
  2264. THROWJSOCKEXCEPTION2(JSOCKERR_bad_address);
  2265. SocketEndpoint ep;
  2266. ep.port = p;
  2267. Owned<CSocket> sock = new CSocket(ep,sm_tcp_server,NULL);
  2268. sock->open(listen_queue_size);
  2269. return sock.getClear();
  2270. }
  2271. ISocket* ISocket::create_ip(unsigned short p,const char *host,int listen_queue_size)
  2272. {
  2273. if (p==0)
  2274. THROWJSOCKEXCEPTION2(JSOCKERR_bad_address);
  2275. SocketEndpoint ep(host,p);
  2276. Owned<CSocket> sock = new CSocket(ep,sm_tcp_server,host);
  2277. sock->open(listen_queue_size);
  2278. return sock.getClear();
  2279. }
  2280. ISocket* ISocket::udp_create(unsigned short p)
  2281. {
  2282. SocketEndpoint ep;
  2283. ep.port=p;
  2284. Owned<CSocket> sock = new CSocket(ep,(p==0)?sm_udp:sm_udp_server,NULL);
  2285. sock->open(0);
  2286. return sock.getClear();
  2287. }
  2288. ISocket* ISocket::multicast_create(unsigned short p, const char *mcip)
  2289. {
  2290. if (p==0)
  2291. THROWJSOCKEXCEPTION2(JSOCKERR_bad_address);
  2292. SocketEndpoint ep(mcip,p);
  2293. Owned<CSocket> sock = new CSocket(ep,sm_multicast_server,mcip);
  2294. sock->open(0,true);
  2295. return sock.getClear();
  2296. }
  2297. ISocket* ISocket::multicast_create(unsigned short p, const IpAddress &ip)
  2298. {
  2299. if (p==0)
  2300. THROWJSOCKEXCEPTION2(JSOCKERR_bad_address);
  2301. SocketEndpoint ep(p, ip);
  2302. StringBuffer tmp;
  2303. Owned<CSocket> sock = new CSocket(ep,sm_multicast_server,ip.getIpText(tmp).str());
  2304. sock->open(0,true);
  2305. return sock.getClear();
  2306. }
  2307. ISocket* ISocket::udp_connect(unsigned short p, char const* name)
  2308. {
  2309. if (!name||!*name||(p==0))
  2310. THROWJSOCKEXCEPTION2(JSOCKERR_bad_address);
  2311. SocketEndpoint ep(name, p);
  2312. Owned<CSocket> sock = new CSocket(ep,sm_udp,name);
  2313. sock->udpconnect();
  2314. return sock.getClear();
  2315. }
  2316. ISocket* ISocket::udp_connect(const SocketEndpoint &ep)
  2317. {
  2318. Owned<CSocket> sock = new CSocket(ep,sm_udp,NULL);
  2319. sock->udpconnect();
  2320. return sock.getClear();
  2321. }
  2322. ISocket* ISocket::multicast_connect(unsigned short p, char const* mcip, unsigned _ttl)
  2323. {
  2324. if (p==0)
  2325. THROWJSOCKEXCEPTION2(JSOCKERR_bad_address);
  2326. SocketEndpoint ep(mcip,p);
  2327. return multicast_connect(ep, _ttl);
  2328. }
  2329. ISocket* ISocket::multicast_connect(const SocketEndpoint &ep, unsigned _ttl)
  2330. {
  2331. Owned<CSocket> sock = new CSocket(ep,sm_multicast,NULL);
  2332. sock->udpconnect();
  2333. u_char ttl = _ttl;
  2334. setsockopt(sock->OShandle(), IPPROTO_IP, IP_MULTICAST_TTL, (char *) &ttl, sizeof(ttl));
  2335. return sock.getClear();
  2336. }
  2337. ISocket* ISocket::attach(int s, bool tcpip)
  2338. {
  2339. CSocket* sock = new CSocket((SOCKET)s, tcpip?sm_tcp:sm_udp, false);
  2340. return sock;
  2341. }
  2342. bool isInterfaceIp(const IpAddress &ip, const char *ifname)
  2343. {
  2344. #ifdef _WIN32
  2345. return false;
  2346. #else
  2347. int fd = socket(AF_INET, SOCK_DGRAM, 0); // IPV6 TBD
  2348. if (fd<0)
  2349. return false;
  2350. MemoryAttr ma;
  2351. char *buf = (char *)ma.allocate(1024);
  2352. struct ifconf ifc;
  2353. ifc.ifc_len = 1024;
  2354. ifc.ifc_buf = buf;
  2355. if(ioctl(fd, SIOCGIFCONF, &ifc) < 0) // query interfaces
  2356. {
  2357. close(fd);
  2358. return false;
  2359. }
  2360. struct ifreq *ifr = ifc.ifc_req;
  2361. unsigned n = ifc.ifc_len/sizeof(struct ifreq);
  2362. bool match = false;
  2363. for(unsigned i=0; i<n; i++)
  2364. {
  2365. struct ifreq *item = &ifr[i];
  2366. if (ifname&&*ifname)
  2367. if (!WildMatch(item->ifr_name,ifname))
  2368. continue;
  2369. IpAddress iptest((inet_ntoa(((struct sockaddr_in *)&item->ifr_addr)->sin_addr)));
  2370. if (ip.ipequals(iptest))
  2371. {
  2372. match = true;
  2373. break;
  2374. }
  2375. }
  2376. close(fd);
  2377. return match;
  2378. #endif
  2379. }
  2380. bool getInterfaceIp(IpAddress &ip,const char *ifname)
  2381. {
  2382. #if defined(_WIN32) || defined(__APPLE__)
  2383. return false;
  2384. #else
  2385. static bool recursioncheck = false;
  2386. ip.ipset(NULL);
  2387. int fd = socket(AF_INET, SOCK_DGRAM, 0); // IPV6 TBD
  2388. if (fd<0)
  2389. return false;
  2390. MemoryAttr ma;
  2391. char *buf = (char *)ma.allocate(1024);
  2392. struct ifconf ifc;
  2393. ifc.ifc_len = 1024;
  2394. ifc.ifc_buf = buf;
  2395. if(ioctl(fd, SIOCGIFCONF, &ifc) < 0) // query interfaces
  2396. {
  2397. close(fd);
  2398. return false;
  2399. }
  2400. struct ifreq *ifr = ifc.ifc_req;
  2401. unsigned n = ifc.ifc_len/sizeof(struct ifreq);
  2402. for (int loopback = 0; loopback <= 1; loopback++)
  2403. {
  2404. for (int i=0; i<n; i++)
  2405. {
  2406. bool useLoopback = (loopback==1);
  2407. struct ifreq *item = &ifr[i];
  2408. if (ifname&&*ifname)
  2409. if (!WildMatch(item->ifr_name,ifname))
  2410. continue;
  2411. IpAddress iptest((inet_ntoa(((struct sockaddr_in *)&item->ifr_addr)->sin_addr)));
  2412. if (ioctl(fd, SIOCGIFFLAGS, item) < 0)
  2413. {
  2414. if (!recursioncheck) {
  2415. recursioncheck = true;
  2416. DBGLOG("Error retrieving interface flags for interface %s", item->ifr_name);
  2417. recursioncheck = false;
  2418. }
  2419. continue;
  2420. }
  2421. bool isLoopback = iptest.isLoopBack() || ((item->ifr_flags & IFF_LOOPBACK) != 0);
  2422. bool isUp = (item->ifr_flags & IFF_UP) != 0;
  2423. if ((isLoopback==useLoopback) && isUp)
  2424. {
  2425. if (ip.isNull())
  2426. ip.ipset(iptest);
  2427. else if (!PreferredSubnet.isNull()&&!PreferredSubnet.test(ip)&&PreferredSubnet.test(iptest))
  2428. ip.ipset(iptest);
  2429. }
  2430. }
  2431. if (!ip.isNull())
  2432. break;
  2433. }
  2434. close(fd);
  2435. return !ip.isNull();
  2436. #endif
  2437. }
  2438. static StringAttr cachehostname;
  2439. static IpAddress cachehostip;
  2440. static IpAddress localhostip;
  2441. static CriticalSection hostnamesect;
  2442. static StringBuffer EnvConfPath;
  2443. const char * GetCachedHostName()
  2444. {
  2445. CriticalBlock c(hostnamesect);
  2446. if (!cachehostname.get())
  2447. {
  2448. #ifndef _WIN32
  2449. IpAddress ip;
  2450. if (EnvConfPath.length() == 0)
  2451. EnvConfPath.append(CONFIG_DIR).append(PATHSEPSTR).append("environment.conf");
  2452. Owned<IProperties> conf = createProperties(EnvConfPath.str(), true);
  2453. StringBuffer ifs;
  2454. conf->getProp("interface", ifs);
  2455. if (getInterfaceIp(ip, ifs.str()))
  2456. {
  2457. StringBuffer ips;
  2458. ip.getIpText(ips);
  2459. if (ips.length())
  2460. {
  2461. cachehostname.set(ips.str());
  2462. cachehostip.ipset(ip);
  2463. return cachehostname.get();
  2464. }
  2465. }
  2466. #endif
  2467. char temp[1024];
  2468. if (gethostname(temp, sizeof(temp))==0)
  2469. cachehostname.set(temp);
  2470. else
  2471. cachehostname.set("localhost"); // assume no NIC card
  2472. }
  2473. return cachehostname.get();
  2474. }
  2475. IpAddress & queryLocalIP()
  2476. {
  2477. CriticalBlock c(hostnamesect);
  2478. if (localhostip.isNull())
  2479. {
  2480. if (IP6preferred)
  2481. localhostip.ipset("::1"); //IPv6
  2482. else
  2483. localhostip.ipset("127.0.0.1"); //IPv4
  2484. }
  2485. return localhostip;
  2486. }
  2487. IpAddress & queryHostIP()
  2488. {
  2489. CriticalBlock c(hostnamesect);
  2490. if (cachehostip.isNull()) {
  2491. if (!cachehostip.ipset(GetCachedHostName())) {
  2492. cachehostip.ipset(queryLocalIP());
  2493. // printf("hostname %s not resolved, using localhost\n",GetCachedHostName()); // don't use jlog in case recursive
  2494. }
  2495. }
  2496. return cachehostip;
  2497. }
  2498. IpAddress &GetHostIp(IpAddress &ip)
  2499. {
  2500. ip.ipset(queryHostIP());
  2501. return ip;
  2502. }
  2503. IpAddress &localHostToNIC(IpAddress &ip)
  2504. {
  2505. if (ip.isLoopBack())
  2506. GetHostIp(ip);
  2507. return ip;
  2508. }
  2509. // IpAddress
  2510. inline bool isIp4(const unsigned *netaddr)
  2511. {
  2512. if (IP4only)
  2513. return true;
  2514. if (netaddr[2]==0xffff0000)
  2515. return (netaddr[1]==0)&&(netaddr[0]==0);
  2516. if (netaddr[2]==0)
  2517. if ((netaddr[3]==0)&&(netaddr[0]==0)&&(netaddr[1]==0))
  2518. return true; // null address
  2519. // maybe should get loopback here
  2520. return false;
  2521. }
  2522. bool IpAddress::isIp4() const
  2523. {
  2524. return ::isIp4(netaddr);
  2525. }
  2526. bool IpAddress::isNull() const
  2527. {
  2528. return (netaddr[3]==0)&&(IP4only||((netaddr[2]==0)&&(netaddr[1]==0)&&(netaddr[0]==0)));
  2529. }
  2530. bool IpAddress::isLoopBack() const
  2531. {
  2532. if (::isIp4(netaddr)&&((netaddr[3] & 0x000000ff)==0x000007f))
  2533. return true;
  2534. return (netaddr[3]==0x1000000)&&(netaddr[2]==0)&&(netaddr[1]==0)&&(netaddr[0]==0);
  2535. }
  2536. bool IpAddress::isLocal() const
  2537. {
  2538. if (isLoopBack() || isHost())
  2539. return true;
  2540. IpAddress ip(*this);
  2541. return isInterfaceIp(ip, NULL);
  2542. }
  2543. bool IpAddress::ipequals(const IpAddress & other) const
  2544. {
  2545. // reverse compare for speed
  2546. return (other.netaddr[3]==netaddr[3])&&(IP4only||((other.netaddr[2]==netaddr[2])&&(other.netaddr[1]==netaddr[1])&&(other.netaddr[0]==netaddr[0])));
  2547. }
  2548. int IpAddress::ipcompare(const IpAddress & other) const
  2549. {
  2550. return memcmp(&netaddr, &other.netaddr, sizeof(netaddr));
  2551. }
  2552. unsigned IpAddress::iphash(unsigned prev) const
  2553. {
  2554. return hashc((const byte *)&netaddr,sizeof(netaddr),prev);
  2555. }
  2556. bool IpAddress::isHost() const
  2557. {
  2558. return ipequals(queryHostIP());
  2559. }
  2560. static bool decodeNumericIP(const char *text,unsigned *netaddr)
  2561. {
  2562. if (!text)
  2563. return false;
  2564. bool isv6 = strchr(text,':')!=NULL;
  2565. StringBuffer tmp;
  2566. if ((*text=='[')&&!IP4only) {
  2567. text++;
  2568. size32_t l = strlen(text);
  2569. if ((l<=2)||(text[l-1]!=']'))
  2570. return false;
  2571. text = tmp.append(l-2,text);
  2572. }
  2573. if (!isv6&&isdigit(text[0])) {
  2574. if (_inet_pton(AF_INET, text, &netaddr[3])>0) {
  2575. netaddr[2] = netaddr[3]?0xffff0000:0; // check for NULL
  2576. netaddr[1] = 0;
  2577. netaddr[0] = 0; // special handling for loopback?
  2578. return true;
  2579. }
  2580. }
  2581. else if (isv6&&!IP4only) {
  2582. int ret = _inet_pton(AF_INET6, text, netaddr);
  2583. if (ret>=0)
  2584. return (ret>0);
  2585. int err = ERRNO();
  2586. StringBuffer tmp("_inet_pton: ");
  2587. tmp.append(text);
  2588. LOGERR(err,1,tmp.str());
  2589. }
  2590. return false;
  2591. }
  2592. static bool lookupHostAddress(const char *name,unsigned *netaddr)
  2593. {
  2594. // if IP4only or using MS V6 can only resolve IPv4 using
  2595. static bool recursioncheck = false; // needed to stop error message recursing
  2596. unsigned retry=10;
  2597. #if defined(__linux__) || defined(getaddrinfo)
  2598. if (IP4only) {
  2599. #else
  2600. {
  2601. #endif
  2602. CriticalBlock c(hostnamesect);
  2603. hostent * entry = gethostbyname(name);
  2604. while (entry==NULL) {
  2605. if (retry--==0) {
  2606. if (!recursioncheck) {
  2607. recursioncheck = true;
  2608. LogErr(h_errno,1,"gethostbyname failed",__LINE__,name);
  2609. recursioncheck = false;
  2610. }
  2611. return false;
  2612. }
  2613. {
  2614. CriticalUnblock ub(hostnamesect);
  2615. Sleep((10-retry)*100);
  2616. }
  2617. entry = gethostbyname(name);
  2618. }
  2619. if (entry->h_addr_list[0]) {
  2620. unsigned ptr = 0;
  2621. if (!PreferredSubnet.isNull()) {
  2622. loop {
  2623. ptr++;
  2624. if (entry->h_addr_list[ptr]==NULL) {
  2625. ptr = 0;
  2626. break;
  2627. }
  2628. IpAddress ip;
  2629. ip.setNetAddress(sizeof(unsigned),entry->h_addr_list[ptr]);
  2630. if (PreferredSubnet.test(ip))
  2631. break;
  2632. }
  2633. }
  2634. memcpy(&netaddr[3], entry->h_addr_list[ptr], sizeof(netaddr[3]));
  2635. netaddr[2] = 0xffff0000;
  2636. netaddr[1] = 0;
  2637. netaddr[0] = 0;
  2638. return true;
  2639. }
  2640. return false;
  2641. }
  2642. #if defined(__linux__) || defined(getaddrinfo)
  2643. struct addrinfo hints;
  2644. memset(&hints,0,sizeof(hints));
  2645. struct addrinfo *addrInfo = NULL;
  2646. loop {
  2647. memset(&hints,0,sizeof(hints));
  2648. int ret = getaddrinfo(name, NULL , &hints, &addrInfo);
  2649. if (!ret)
  2650. break;
  2651. if (retry--==0) {
  2652. if (!recursioncheck) {
  2653. recursioncheck = true;
  2654. LogErr(ret,1,"getaddrinfo failed",__LINE__,name);
  2655. #ifdef _DEBUG
  2656. PrintStackReport();
  2657. #endif
  2658. recursioncheck = false;
  2659. }
  2660. return false;
  2661. }
  2662. Sleep((10-retry)*100);
  2663. }
  2664. struct addrinfo *best = NULL;
  2665. bool snm = !PreferredSubnet.isNull();
  2666. loop {
  2667. struct addrinfo *ai;
  2668. for (ai = addrInfo; ai; ai = ai->ai_next) {
  2669. // printf("flags=%d, family=%d, socktype=%d, protocol=%d, addrlen=%d, canonname=%s\n",ai->ai_flags,ai->ai_family,ai->ai_socktype,ai->ai_protocol,ai->ai_addrlen,ai->ai_canonname?ai->ai_canonname:"NULL");
  2670. switch (ai->ai_family) {
  2671. case AF_INET: {
  2672. if (snm) {
  2673. IpAddress ip;
  2674. ip.setNetAddress(sizeof(in_addr),&(((sockaddr_in *)ai->ai_addr)->sin_addr));
  2675. if (!PreferredSubnet.test(ip))
  2676. continue;
  2677. }
  2678. if ((best==NULL)||((best->ai_family==AF_INET6)&&!IP6preferred))
  2679. best = ai;
  2680. break;
  2681. }
  2682. case AF_INET6: {
  2683. if (snm) {
  2684. IpAddress ip;
  2685. ip.setNetAddress(sizeof(in_addr6),&(((sockaddr_in6 *)ai->ai_addr)->sin6_addr));
  2686. if (!PreferredSubnet.test(ip))
  2687. continue;
  2688. }
  2689. if ((best==NULL)||((best->ai_family==AF_INET)&&IP6preferred))
  2690. best = ai;
  2691. break;
  2692. }
  2693. }
  2694. }
  2695. if (best||!snm)
  2696. break;
  2697. snm = false;
  2698. }
  2699. if (best) {
  2700. if (best->ai_family==AF_INET6)
  2701. memcpy(netaddr,&(((sockaddr_in6 *)best->ai_addr)->sin6_addr),sizeof(in6_addr));
  2702. else {
  2703. memcpy(netaddr+3,&(((sockaddr_in *)best->ai_addr)->sin_addr),sizeof(in_addr));
  2704. netaddr[2] = 0xffff0000;
  2705. netaddr[1] = 0;
  2706. netaddr[0] = 0;
  2707. }
  2708. }
  2709. freeaddrinfo(addrInfo);
  2710. return best!=NULL;
  2711. #endif
  2712. return false;
  2713. }
  2714. bool IpAddress::ipset(const char *text)
  2715. {
  2716. if (text&&*text) {
  2717. if ((text[0]=='.')&&(text[1]==0)) {
  2718. ipset(queryHostIP());
  2719. return true;
  2720. }
  2721. if (decodeNumericIP(text,netaddr))
  2722. return true;
  2723. const char *s;
  2724. for (s=text;*s;s++)
  2725. if (!isdigit(*s)&&(*s!=':')&&(*s!='.'))
  2726. break;
  2727. if (!*s)
  2728. return ipset(NULL);
  2729. if (lookupHostAddress(text,netaddr))
  2730. return true;
  2731. }
  2732. memset(&netaddr,0,sizeof(netaddr));
  2733. return false;
  2734. }
  2735. inline char * addbyte(char *s,byte b)
  2736. {
  2737. if (b>=100) {
  2738. *(s++) = b/100+'0';
  2739. b %= 100;
  2740. *(s++) = b/10+'0';
  2741. b %= 10;
  2742. }
  2743. else if (b>=10) {
  2744. *(s++) = b/10+'0';
  2745. b %= 10;
  2746. }
  2747. *(s++) = b+'0';
  2748. return s;
  2749. }
  2750. StringBuffer & IpAddress::getIpText(StringBuffer & out) const
  2751. {
  2752. if (::isIp4(netaddr)) {
  2753. const byte *ip = (const byte *)&netaddr[3];
  2754. char ips[16];
  2755. char *s = ips;
  2756. for (unsigned i=0;i<4;i++) {
  2757. if (i)
  2758. *(s++) = '.';
  2759. s = addbyte(s,ip[i]);
  2760. }
  2761. return out.append(s-ips,ips);
  2762. }
  2763. char tmp[INET6_ADDRSTRLEN];
  2764. const char *res = _inet_ntop(AF_INET6, &netaddr, tmp, sizeof(tmp));
  2765. if (!res)
  2766. throw makeOsException(errno);
  2767. return out.append(res);
  2768. }
  2769. void IpAddress::ipserialize(MemoryBuffer & out) const
  2770. {
  2771. if (((netaddr[2]==0xffff0000)||(netaddr[2]==0))&&(netaddr[1]==0)&&(netaddr[0]==0)) {
  2772. if (netaddr[3]==IPV6_SERIALIZE_PREFIX)
  2773. throw MakeStringException(-1,"Invalid network address"); // hack prevention
  2774. out.append(sizeof(netaddr[3]), &netaddr[3]);
  2775. }
  2776. else {
  2777. unsigned pfx = IPV6_SERIALIZE_PREFIX;
  2778. out.append(sizeof(pfx),&pfx).append(sizeof(netaddr),&netaddr);
  2779. }
  2780. }
  2781. void IpAddress::ipdeserialize(MemoryBuffer & in)
  2782. {
  2783. unsigned pfx;
  2784. in.read(sizeof(pfx),&pfx);
  2785. if (pfx!=IPV6_SERIALIZE_PREFIX) {
  2786. netaddr[0] = 0;
  2787. netaddr[1] = 0;
  2788. netaddr[2] = (pfx == 0 || pfx == 0x1000000) ? 0 : 0xffff0000; // catch null and loopback
  2789. netaddr[3] = pfx;
  2790. }
  2791. else
  2792. in.read(sizeof(netaddr),&netaddr);
  2793. }
  2794. unsigned IpAddress::ipdistance(const IpAddress &other,unsigned offset) const
  2795. {
  2796. if (offset>3)
  2797. offset = 3;
  2798. int i1;
  2799. _cpyrev4(&i1,&netaddr[3-offset]);
  2800. int i2;
  2801. _cpyrev4(&i2,&other.netaddr[3-offset]);
  2802. i1-=i2;
  2803. if (i1>0)
  2804. return i1;
  2805. return -i1;
  2806. }
  2807. bool IpAddress::ipincrement(unsigned count,byte minoctet,byte maxoctet,unsigned short minipv6piece,unsigned maxipv6piece)
  2808. {
  2809. unsigned base;
  2810. if (::isIp4(netaddr)) {
  2811. base = maxoctet-minoctet+1;
  2812. if (!base||(base>256))
  2813. return false;
  2814. byte * ips = (byte *)&netaddr[3];
  2815. byte * ip = ips+4;
  2816. while (count) {
  2817. if (ip==ips)
  2818. return false; // overflow
  2819. ip--;
  2820. unsigned v = (count+((*ip>minoctet)?(*ip-minoctet):0));
  2821. *ip = minoctet + v%base;
  2822. count = v/base;
  2823. }
  2824. }
  2825. else {
  2826. base = maxipv6piece-minipv6piece+1;
  2827. if (!base||(base>0x10000))
  2828. return false;
  2829. unsigned short * ps = (unsigned short *)&netaddr;
  2830. unsigned short * p = ps+8;
  2831. while (count) {
  2832. if (p==ps)
  2833. return false; // overflow (actually near impossible!)
  2834. p--;
  2835. unsigned v = (count+((*p>minipv6piece)?(*p-minipv6piece):0));
  2836. *p = minipv6piece + v%base;
  2837. count = v/base;
  2838. }
  2839. }
  2840. return true;
  2841. }
  2842. unsigned IpAddress::ipsetrange( const char *text) // e.g. 10.173.72.1-65 ('-' may be omitted)
  2843. {
  2844. unsigned e=0;
  2845. unsigned f=0;
  2846. const char *r = strchr(text,'-');
  2847. bool ok;
  2848. if (r) {
  2849. e = atoi(r+1);
  2850. StringBuffer tmp(r-text,text);
  2851. ok = ipset(tmp.str());
  2852. if (!::isIp4(netaddr))
  2853. IPV6_NOT_IMPLEMENTED(); // TBD IPv6
  2854. if (ok) {
  2855. while ((r!=text)&&(*(r-1)!='.'))
  2856. r--;
  2857. f = (r!=text)?atoi(r):0;
  2858. }
  2859. }
  2860. else
  2861. ok = ipset(text);
  2862. if ((f>e)||!ok)
  2863. return 0;
  2864. return e-f+1;
  2865. }
  2866. size32_t IpAddress::getNetAddress(size32_t maxsz,void *dst) const
  2867. {
  2868. if (maxsz==sizeof(unsigned)) {
  2869. if (::isIp4(netaddr)) {
  2870. *(unsigned *)dst = netaddr[3];
  2871. return maxsz;
  2872. }
  2873. }
  2874. else if (!IP4only&&(maxsz==sizeof(netaddr))) {
  2875. memcpy(dst,&netaddr,maxsz);
  2876. return maxsz;
  2877. }
  2878. return 0;
  2879. }
  2880. void IpAddress::setNetAddress(size32_t sz,const void *src)
  2881. {
  2882. if (sz==sizeof(unsigned)) { // IPv4
  2883. netaddr[0] = 0;
  2884. netaddr[1] = 0;
  2885. netaddr[3] = *(const unsigned *)src;
  2886. netaddr[2] = netaddr[3] ? 0xffff0000 : 0; // leave as null if Ipv4 address is null
  2887. }
  2888. else if (!IP4only&&(sz==sizeof(netaddr))) { // IPv6
  2889. memcpy(&netaddr,src,sz);
  2890. if ((netaddr[2]==0)&&(netaddr[3]!=0)&&(netaddr[3]!=0x1000000)&&(netaddr[0]==0)&&(netaddr[1]==0))
  2891. netaddr[2]=0xffff0000; // use this form only
  2892. }
  2893. else
  2894. memset(&netaddr,0,sizeof(netaddr));
  2895. }
  2896. void SocketEndpoint::deserialize(MemoryBuffer & in)
  2897. {
  2898. ipdeserialize(in);
  2899. in.read(port);
  2900. }
  2901. void SocketEndpoint::serialize(MemoryBuffer & out) const
  2902. {
  2903. ipserialize(out);
  2904. out.append(port);
  2905. }
  2906. bool SocketEndpoint::set(const char *name,unsigned short _port)
  2907. {
  2908. if (name) {
  2909. if (*name=='[') {
  2910. const char *s = name+1;
  2911. const char *t = strchr(s,']');
  2912. if (t) {
  2913. StringBuffer tmp(t-s,s);
  2914. if (t[1]==':')
  2915. _port = atoi(t+2);
  2916. return set(tmp.str(),_port);
  2917. }
  2918. }
  2919. const char * colon = strchr(name, ':');
  2920. if (colon) {
  2921. if (!IP4only&&strchr(colon+1, ':'))
  2922. colon = NULL; // hello its IpV6
  2923. }
  2924. else
  2925. colon = strchr(name, '|'); // strange hole convention
  2926. char ips[260];
  2927. if (colon) {
  2928. size32_t l = colon-name;
  2929. if (l>=sizeof(ips))
  2930. l = sizeof(ips)-1;
  2931. memcpy(ips,name,l);
  2932. ips[l] = 0;
  2933. name = ips;
  2934. _port = atoi(colon+1);
  2935. }
  2936. if (ipset(name)) {
  2937. port = _port;
  2938. return true;
  2939. }
  2940. }
  2941. ipset(NULL);
  2942. port = 0;
  2943. return false;
  2944. }
  2945. void SocketEndpoint::getUrlStr(char * str, size32_t len) const
  2946. {
  2947. if (len==0)
  2948. return;
  2949. StringBuffer _str;
  2950. getUrlStr(_str);
  2951. size32_t l = _str.length()+1;
  2952. if (l>len)
  2953. {
  2954. l = len-1;
  2955. str[l] = 0;
  2956. }
  2957. memcpy(str,_str.str(),l);
  2958. }
  2959. StringBuffer &SocketEndpoint::getUrlStr(StringBuffer &str) const
  2960. {
  2961. getIpText(str);
  2962. if (port)
  2963. str.append(':').append((unsigned)port); // TBD IPv6 put [] on
  2964. return str;
  2965. }
  2966. unsigned SocketEndpoint::hash(unsigned prev) const
  2967. {
  2968. return hashc((const byte *)&port,sizeof(port),iphash(prev));
  2969. }
  2970. //---------------------------------------------------------------------------
  2971. SocketListCreator::SocketListCreator()
  2972. {
  2973. lastPort = 0;
  2974. }
  2975. void SocketListCreator::addSocket(const SocketEndpoint &ep)
  2976. {
  2977. StringBuffer ipstr;
  2978. ep.getIpText(ipstr);
  2979. addSocket(ipstr.str(), ep.port);
  2980. }
  2981. void SocketListCreator::addSocket(const char * ip, unsigned port)
  2982. {
  2983. if (fullText.length())
  2984. fullText.append("|");
  2985. const char * prev = lastIp;
  2986. const char * startCopy = ip;
  2987. if (prev)
  2988. {
  2989. if (strcmp(ip, prev) == 0)
  2990. {
  2991. fullText.append("=");
  2992. startCopy = NULL;
  2993. }
  2994. else
  2995. {
  2996. const char * cur = ip;
  2997. loop
  2998. {
  2999. char n = *cur;
  3000. if (!n)
  3001. break;
  3002. if (n != *prev)
  3003. break;
  3004. cur++;
  3005. prev++;
  3006. if (n == '.')
  3007. startCopy = cur;
  3008. }
  3009. if (startCopy != ip)
  3010. fullText.append("*");
  3011. }
  3012. }
  3013. fullText.append(startCopy);
  3014. if (lastPort != port)
  3015. fullText.append(":").append(port);
  3016. lastIp.set(ip);
  3017. lastPort = port;
  3018. }
  3019. const char * SocketListCreator::getText()
  3020. {
  3021. return fullText.str();
  3022. }
  3023. void SocketListCreator::addSockets(SocketEndpointArray &array)
  3024. {
  3025. ForEachItemIn(i,array) {
  3026. const SocketEndpoint &sockep=array.item(i);
  3027. StringBuffer ipstr;
  3028. sockep.getIpText(ipstr);
  3029. addSocket(ipstr.str(),sockep.port);
  3030. }
  3031. }
  3032. //---------------------------------------------------------------------------
  3033. SocketListParser::SocketListParser(const char * text)
  3034. {
  3035. fullText.set(text);
  3036. cursor = NULL;
  3037. lastPort = 0;
  3038. }
  3039. void SocketListParser::first(unsigned port)
  3040. {
  3041. cursor = fullText;
  3042. lastIp.set(NULL);
  3043. lastPort = port;
  3044. }
  3045. bool SocketListParser::get(StringAttr & ip, unsigned & port, unsigned index, unsigned defport)
  3046. {
  3047. first(defport);
  3048. do
  3049. {
  3050. if (!next(ip, port))
  3051. return false;
  3052. } while (index--);
  3053. return true;
  3054. }
  3055. bool SocketListParser::next(StringAttr & ip, unsigned & port)
  3056. {
  3057. // IPV6TBD
  3058. StringBuffer ipText;
  3059. if (*cursor == 0)
  3060. return false;
  3061. if (*cursor == '=')
  3062. {
  3063. ipText.append(lastIp);
  3064. cursor++;
  3065. }
  3066. else if (*cursor == '*')
  3067. {
  3068. cursor++;
  3069. //count the number of dots in the tail
  3070. const char * cur = cursor;
  3071. unsigned count = 0;
  3072. loop
  3073. {
  3074. char c = *cur++;
  3075. switch (c)
  3076. {
  3077. case 0:
  3078. case '|':
  3079. case ',':
  3080. case ':':
  3081. goto done;
  3082. case '.':
  3083. ++count;
  3084. break;
  3085. }
  3086. }
  3087. done:
  3088. //copy up to the appropriate dot from the previous ip.
  3089. const unsigned dotCount = 3; //more what about 6 digit ip's
  3090. cur = lastIp;
  3091. loop
  3092. {
  3093. char c = *cur++;
  3094. switch (c)
  3095. {
  3096. case 0:
  3097. case '|':
  3098. case ',':
  3099. case ':':
  3100. assertex(!"Should not get here!");
  3101. goto done2;
  3102. case '.':
  3103. ipText.append(c);
  3104. if (++count == dotCount)
  3105. goto done2;
  3106. break;
  3107. default:
  3108. ipText.append(c);
  3109. break;
  3110. }
  3111. }
  3112. done2:;
  3113. }
  3114. bool inPort = false;
  3115. port = lastPort;
  3116. loop
  3117. {
  3118. char c = *cursor++;
  3119. switch (c)
  3120. {
  3121. case 0:
  3122. cursor--;
  3123. goto doneCopy;
  3124. case '|':
  3125. case ',':
  3126. goto doneCopy;
  3127. case ':':
  3128. port = atoi(cursor);
  3129. inPort = true;
  3130. break;;
  3131. default:
  3132. if (!inPort)
  3133. ipText.append(c);
  3134. break;
  3135. }
  3136. }
  3137. doneCopy:
  3138. lastIp.set(ipText.str());
  3139. ip.set(lastIp);
  3140. lastPort = port;
  3141. return true;
  3142. }
  3143. unsigned SocketListParser::getSockets(SocketEndpointArray &array,unsigned defport)
  3144. {
  3145. first(defport);
  3146. StringAttr ip;
  3147. unsigned port;
  3148. while (next(ip,port)) {
  3149. SocketEndpoint ep(ip,port);
  3150. array.append(ep);
  3151. }
  3152. return array.ordinality();
  3153. }
  3154. void getSocketStatistics(JSocketStatistics &stats)
  3155. {
  3156. // should put in simple lock
  3157. memcpy(&stats,&STATS,sizeof(stats));
  3158. }
  3159. void resetSocketStatistics()
  3160. {
  3161. unsigned activesockets=STATS.activesockets;
  3162. memset(&STATS,0,sizeof(STATS));
  3163. STATS.activesockets = activesockets;
  3164. }
  3165. static StringBuffer &appendtime(StringBuffer &s,unsigned us)
  3166. {
  3167. // attemp to get into more sensible units
  3168. if (us>10000000)
  3169. return s.append(us/1000000).append('s');
  3170. if (us>10000)
  3171. return s.append(us/1000).append("ms");
  3172. return s.append(us).append("us");
  3173. }
  3174. StringBuffer &getSocketStatisticsString(JSocketStatistics &stats,StringBuffer &str)
  3175. {
  3176. str.append("connects=").append(stats.connects).append('\n');
  3177. appendtime(str.append("connecttime="),stats.connecttime).append('\n');
  3178. str.append("failedconnects=").append(stats.failedconnects).append('\n');
  3179. appendtime(str.append("failedconnecttime="),stats.failedconnecttime).append('\n');
  3180. str.append("reads=").append(stats.reads).append('\n');
  3181. appendtime(str.append("readtime="),stats.readtime).append('\n');
  3182. str.append("readsize=").append(stats.readsize).append(" bytes\n");
  3183. str.append("writes=").append(stats.writes).append('\n');
  3184. appendtime(str.append("writetime="),stats.writetime).append('\n');
  3185. str.append("writesize=").append(stats.writesize).append(" bytes").append('\n');
  3186. str.append("activesockets=").append(stats.activesockets).append('\n');
  3187. str.append("numblockrecvs=").append(stats.numblockrecvs).append('\n');
  3188. str.append("numblocksends=").append(stats.numblocksends).append('\n');
  3189. str.append("blockrecvsize=").append(stats.blockrecvsize).append('\n');
  3190. str.append("blocksendsize=").append(stats.blocksendsize).append('\n');
  3191. str.append("blockrecvtime=").append(stats.blockrecvtime).append('\n');
  3192. str.append("blocksendtime=").append(stats.blocksendtime).append('\n');
  3193. str.append("longestblocksend=").append(stats.longestblocksend).append('\n');
  3194. str.append("longestblocksize=").append(stats.longestblocksize);
  3195. return str;
  3196. }
  3197. // ===============================================================================
  3198. // select thread for handling multiple selects
  3199. struct SelectItem
  3200. {
  3201. ISocket *sock;
  3202. T_SOCKET handle;
  3203. ISocketSelectNotify *nfy;
  3204. byte mode;
  3205. bool del;
  3206. bool add_epoll;
  3207. bool operator == (const SelectItem & other) const { return sock == other.sock; }
  3208. };
  3209. class SelectItemArray : public StructArrayOf<SelectItem> { };
  3210. #define SELECT_TIMEOUT_SECS 1 // but it does (TBD)
  3211. #ifdef _WIN32
  3212. // fd_set utility functions
  3213. inline T_FD_SET *cpyfds(T_FD_SET &dst,const T_FD_SET &src)
  3214. {
  3215. unsigned i = src.fd_count;
  3216. dst.fd_count = i;
  3217. while (i--)
  3218. dst.fd_array[i] = src.fd_array[i]; // possibly better as memcpy
  3219. return &dst;
  3220. }
  3221. inline bool findfds(T_FD_SET &s,T_SOCKET h,bool &c)
  3222. {
  3223. unsigned n = s.fd_count;
  3224. unsigned i;
  3225. for(i=0;i<n;i++) {
  3226. if (s.fd_array[i] == h) {
  3227. if (--n)
  3228. s.fd_array[i] = s.fd_array[n]; // remove item
  3229. else
  3230. c = false;
  3231. s.fd_count = n;
  3232. return true;
  3233. }
  3234. }
  3235. return false;
  3236. }
  3237. inline T_SOCKET popfds(T_FD_SET &s)
  3238. {
  3239. unsigned n = s.fd_count;
  3240. T_SOCKET ret;
  3241. if (n) {
  3242. ret = s.fd_array[--n];
  3243. s.fd_count = n;
  3244. }
  3245. else
  3246. ret = NULL;
  3247. return ret;
  3248. }
  3249. #else
  3250. #define _USE_PIPE_FOR_SELECT_TRIGGER
  3251. // not as optimized as windows but I am expecting to convert to using poll anyway
  3252. inline T_FD_SET *cpyfds(T_FD_SET &dst,const T_FD_SET &src)
  3253. {
  3254. memcpy(&dst,&src,sizeof(T_FD_SET));
  3255. return &dst;
  3256. }
  3257. inline bool findfds(T_FD_SET &s,T_SOCKET h,bool &c)
  3258. {
  3259. if ((unsigned)h>=XFD_SETSIZE)
  3260. return false;
  3261. return FD_ISSET(h,&s); // does not remove entry or set termination flag when done
  3262. }
  3263. #endif
  3264. class CSocketBaseThread: public Thread
  3265. {
  3266. protected:
  3267. bool terminating;
  3268. CriticalSection sect;
  3269. Semaphore ticksem;
  3270. atomic_t tickwait;
  3271. SelectItemArray items;
  3272. unsigned offset;
  3273. bool selectvarschange;
  3274. unsigned waitingchange;
  3275. Semaphore waitingchangesem;
  3276. int validateselecterror;
  3277. unsigned validateerrcount;
  3278. const char *selecttrace;
  3279. unsigned basesize;
  3280. #ifdef _USE_PIPE_FOR_SELECT_TRIGGER
  3281. T_SOCKET dummysock[2];
  3282. #else
  3283. T_SOCKET dummysock;
  3284. #endif
  3285. bool dummysockopen;
  3286. CSocketBaseThread(const char *trc) : Thread("CSocketBaseThread")
  3287. {
  3288. }
  3289. ~CSocketBaseThread()
  3290. {
  3291. }
  3292. public:
  3293. void triggerselect()
  3294. {
  3295. if (atomic_read(&tickwait))
  3296. ticksem.signal();
  3297. #ifdef _USE_PIPE_FOR_SELECT_TRIGGER
  3298. CriticalBlock block(sect);
  3299. char c = 0;
  3300. if(write(dummysock[1], &c, 1) != 1) {
  3301. int err = ERRNO();
  3302. LOGERR(err,1,"Socket closed during trigger select");
  3303. }
  3304. #else
  3305. closedummy();
  3306. #endif
  3307. }
  3308. void resettrigger()
  3309. {
  3310. #ifdef _USE_PIPE_FOR_SELECT_TRIGGER
  3311. CriticalBlock block(sect);
  3312. char c;
  3313. while((::read(dummysock[0], &c, sizeof(c))) == sizeof(c));
  3314. #endif
  3315. }
  3316. bool remove(ISocket *sock)
  3317. {
  3318. if (terminating)
  3319. return false;
  3320. CriticalBlock block(sect);
  3321. if (sock==NULL) { // wait until no changes outstanding
  3322. while (selectvarschange) {
  3323. waitingchange++;
  3324. CriticalUnblock unblock(sect);
  3325. waitingchangesem.wait();
  3326. }
  3327. return true;
  3328. }
  3329. ForEachItemIn(i,items) {
  3330. SelectItem &si = items.element(i);
  3331. if (!si.del&&(si.sock==sock)) {
  3332. si.del = true;
  3333. selectvarschange = true;
  3334. triggerselect();
  3335. return true;
  3336. }
  3337. }
  3338. return false;
  3339. }
  3340. void stop(bool wait)
  3341. {
  3342. terminating = true;
  3343. triggerselect();
  3344. if (wait)
  3345. join();
  3346. }
  3347. bool sockOk(T_SOCKET sock)
  3348. {
  3349. PROGLOG("CSocketBaseThread: sockOk testing %d",sock);
  3350. int err = 0;
  3351. int t=0;
  3352. socklen_t tl = sizeof(t);
  3353. if (getsockopt(sock, SOL_SOCKET, SO_TYPE, (char *)&t, &tl)!=0) {
  3354. StringBuffer sockstr;
  3355. const char *tracename = sockstr.append((unsigned)sock).str();
  3356. LOGERR2(ERRNO(),1,"CSocketBaseThread select handle");
  3357. return false;
  3358. }
  3359. T_FD_SET fds;
  3360. struct timeval tv;
  3361. XFD_ZERO(&fds);
  3362. FD_SET((unsigned)sock, &fds);
  3363. //FD_SET((unsigned)sock, &except);
  3364. tv.tv_sec = 0;
  3365. tv.tv_usec = 0;
  3366. CHECKSOCKRANGE(sock);
  3367. int rc = ::select( sock + 1, NULL, (fd_set *)&fds, NULL, &tv );
  3368. if (rc<0) {
  3369. StringBuffer sockstr;
  3370. const char *tracename = sockstr.append((unsigned)sock).str();
  3371. LOGERR2(ERRNO(),2,"CSocketBaseThread select handle");
  3372. return false;
  3373. }
  3374. else if (rc>0)
  3375. PROGLOG("CSocketBaseThread: select handle %d selected(2) %d",sock,rc);
  3376. XFD_ZERO(&fds);
  3377. FD_SET((unsigned)sock, &fds);
  3378. tv.tv_sec = 0;
  3379. tv.tv_usec = 0;
  3380. rc = ::select( sock + 1, (fd_set *)&fds, NULL, NULL, &tv );
  3381. if (rc<0) {
  3382. StringBuffer sockstr;
  3383. const char *tracename = sockstr.append((unsigned)sock).str();
  3384. LOGERR2(ERRNO(),3,"CSocketBaseThread select handle");
  3385. return false;
  3386. }
  3387. else if (rc>0)
  3388. PROGLOG("CSocketBaseThread: select handle %d selected(2) %d",sock,rc);
  3389. return true;
  3390. }
  3391. bool checkSocks()
  3392. {
  3393. bool ret = false;
  3394. ForEachItemIn(i,items) {
  3395. SelectItem &si = items.element(i);
  3396. if (si.del)
  3397. ret = true; // maybe that bad one
  3398. else if (!sockOk(si.handle)) {
  3399. si.del = true;
  3400. ret = true;
  3401. }
  3402. }
  3403. return ret;
  3404. }
  3405. virtual void closedummy() = 0;
  3406. };
  3407. class CSocketSelectThread: public CSocketBaseThread
  3408. {
  3409. void opendummy()
  3410. {
  3411. CriticalBlock block(sect);
  3412. if (!dummysockopen) {
  3413. #ifdef _USE_PIPE_FOR_SELECT_TRIGGER
  3414. if(pipe(dummysock)) {
  3415. WARNLOG("CSocketSelectThread: create pipe failed %d",ERRNO());
  3416. return;
  3417. }
  3418. for (unsigned i=0;i<2;i++) {
  3419. int flags = fcntl(dummysock[i], F_GETFL, 0);
  3420. if (flags!=-1) {
  3421. flags |= O_NONBLOCK;
  3422. fcntl(dummysock[i], F_SETFL, flags);
  3423. }
  3424. flags = fcntl(dummysock[i], F_GETFD, 0);
  3425. if (flags!=-1) {
  3426. flags |= FD_CLOEXEC;
  3427. fcntl(dummysock[i], F_SETFD, flags);
  3428. }
  3429. }
  3430. CHECKSOCKRANGE(dummysock[0]);
  3431. #else
  3432. if (IP6preferred)
  3433. dummysock = ::socket(AF_INET6, SOCK_STREAM, PF_INET6);
  3434. else
  3435. dummysock = ::socket(AF_INET, SOCK_STREAM, 0);
  3436. CHECKSOCKRANGE(dummysock);
  3437. #endif
  3438. dummysockopen = true;
  3439. }
  3440. }
  3441. void closedummy()
  3442. {
  3443. CriticalBlock block(sect);
  3444. if (dummysockopen) {
  3445. #ifdef _USE_PIPE_FOR_SELECT_TRIGGER
  3446. #ifdef SOCKTRACE
  3447. PROGLOG("SOCKTRACE: Closing dummy sockets %x %d %x %d (%p)", dummysock[0], dummysock[0], dummysock[1], dummysock[1], this);
  3448. #endif
  3449. ::close(dummysock[0]);
  3450. ::close(dummysock[1]);
  3451. #else
  3452. #ifdef _WIN32
  3453. ::closesocket(dummysock);
  3454. #else
  3455. ::close(dummysock);
  3456. #endif
  3457. #endif
  3458. dummysockopen = false;
  3459. }
  3460. }
  3461. #ifdef _WIN32
  3462. #define HASHTABSIZE 256
  3463. #define HASHNULL (HASHTABSIZE-1)
  3464. #define HASHTABMASK (HASHTABSIZE-1)
  3465. byte hashtab[HASHTABSIZE];
  3466. #define HASHSOCKET(s) ((((unsigned)s)>>2)&HASHTABMASK) // with some knowledge of windows handles
  3467. void inithash()
  3468. {
  3469. memset(&hashtab,HASHNULL,sizeof(hashtab));
  3470. assertex(FD_SETSIZE<255);
  3471. }
  3472. void reinithash()
  3473. { // done this way because index of items changes and hash table not that big
  3474. inithash();
  3475. assertex(items.ordinality()<HASHTABSIZE-1);
  3476. ForEachItemIn(i,items) {
  3477. unsigned h = HASHSOCKET(items.item(i).handle);
  3478. loop {
  3479. if (hashtab[h]==HASHNULL) {
  3480. hashtab[h] = (byte)i;
  3481. break;
  3482. }
  3483. if (++h==HASHTABSIZE)
  3484. h = 0;
  3485. }
  3486. }
  3487. }
  3488. inline SelectItem &findhash(T_SOCKET handle)
  3489. {
  3490. unsigned h = HASHSOCKET(handle);
  3491. unsigned sh = h;
  3492. loop {
  3493. SelectItem &i=items.element(hashtab[h]);
  3494. if (i.handle==handle)
  3495. return i;
  3496. if (++h==HASHTABSIZE)
  3497. h = 0;
  3498. assertex(h!=sh);
  3499. }
  3500. }
  3501. inline void processfds(T_FD_SET &s,byte mode,SelectItemArray &tonotify)
  3502. {
  3503. loop {
  3504. T_SOCKET sock = popfds(s);
  3505. if (!sock)
  3506. break;
  3507. if (sock!=dummysock) {
  3508. SelectItem si = findhash(sock); // nb copies
  3509. if (!si.del) {
  3510. si.mode = mode;
  3511. tonotify.append(si);
  3512. }
  3513. }
  3514. }
  3515. }
  3516. #endif
  3517. public:
  3518. IMPLEMENT_IINTERFACE;
  3519. CSocketSelectThread(const char *trc)
  3520. : CSocketBaseThread("CSocketSelectThread")
  3521. {
  3522. dummysockopen = false;
  3523. opendummy();
  3524. terminating = false;
  3525. atomic_set(&tickwait,0);
  3526. waitingchange = 0;
  3527. selectvarschange = false;
  3528. validateselecterror = 0;
  3529. validateerrcount = 0;
  3530. offset = 0;
  3531. selecttrace = trc;
  3532. basesize = 0;
  3533. #ifdef _WIN32
  3534. inithash();
  3535. #endif
  3536. }
  3537. ~CSocketSelectThread()
  3538. {
  3539. closedummy();
  3540. ForEachItemIn(i,items) {
  3541. try {
  3542. SelectItem &si = items.element(i);
  3543. si.sock->Release();
  3544. si.nfy->Release();
  3545. }
  3546. catch (IException *e) {
  3547. EXCLOG(e,"~CSocketSelectThread");
  3548. e->Release();
  3549. }
  3550. }
  3551. }
  3552. Owned<IException> termexcept;
  3553. void updateItems()
  3554. {
  3555. // must be in CriticalBlock block(sect);
  3556. unsigned n = items.ordinality();
  3557. bool hashupdateneeded = (n!=basesize); // additions all come at end
  3558. for (unsigned i=0;i<n;) {
  3559. SelectItem &si = items.element(i);
  3560. if (si.del) {
  3561. si.nfy->Release();
  3562. try {
  3563. #ifdef SOCKTRACE
  3564. PROGLOG("CSocketSelectThread::updateItems release %d",si.handle);
  3565. #endif
  3566. si.sock->Release();
  3567. }
  3568. catch (IException *e) {
  3569. EXCLOG(e,"CSocketSelectThread::updateItems");
  3570. e->Release();
  3571. }
  3572. n--;
  3573. if (i<n)
  3574. si = items.item(n);
  3575. items.remove(n);
  3576. hashupdateneeded = true;
  3577. }
  3578. else
  3579. i++;
  3580. }
  3581. assertex(n<=XFD_SETSIZE-1);
  3582. #ifdef _WIN32
  3583. if (hashupdateneeded)
  3584. reinithash();
  3585. #endif
  3586. basesize = n;
  3587. }
  3588. bool add(ISocket *sock,unsigned mode,ISocketSelectNotify *nfy)
  3589. {
  3590. // maybe check once to prevent 1st delay? TBD
  3591. CriticalBlock block(sect);
  3592. unsigned n=0;
  3593. ForEachItemIn(i,items) {
  3594. SelectItem &si = items.element(i);
  3595. if (!si.del) {
  3596. if (si.sock==sock) {
  3597. si.del = true;
  3598. }
  3599. else
  3600. n++;
  3601. }
  3602. }
  3603. if (n>=XFD_SETSIZE-1) // leave 1 spare
  3604. return false;
  3605. SelectItem sn;
  3606. sn.nfy = LINK(nfy);
  3607. sn.sock = LINK(sock);
  3608. sn.mode = (byte)mode;
  3609. sn.handle = (T_SOCKET)sock->OShandle();
  3610. CHECKSOCKRANGE(sn.handle);
  3611. sn.del = false;
  3612. sn.add_epoll = false;
  3613. items.append(sn);
  3614. selectvarschange = true;
  3615. triggerselect();
  3616. return true;
  3617. }
  3618. void updateSelectVars(T_FD_SET &rdfds,T_FD_SET &wrfds,T_FD_SET &exfds,bool &isrd,bool &iswr,bool &isex,unsigned &ni,T_SOCKET &max_sockid)
  3619. {
  3620. CriticalBlock block(sect);
  3621. selectvarschange = false;
  3622. if (waitingchange) {
  3623. waitingchangesem.signal(waitingchange);
  3624. waitingchange = 0;
  3625. }
  3626. if (validateselecterror) { // something went wrong so check sockets
  3627. validateerrcount++;
  3628. if (!checkSocks()) {
  3629. // bad socket not found
  3630. PROGLOG("CSocketSelectThread::updateSelectVars cannot find socket error");
  3631. if (validateerrcount>10)
  3632. throw MakeStringException(-1,"CSocketSelectThread:Socket select error %d",validateselecterror);
  3633. }
  3634. }
  3635. else
  3636. validateerrcount = 0;
  3637. updateItems();
  3638. XFD_ZERO( &rdfds );
  3639. XFD_ZERO( &wrfds );
  3640. XFD_ZERO( &exfds );
  3641. isrd=false;
  3642. iswr=false;
  3643. isex=false;
  3644. #ifdef _USE_PIPE_FOR_SELECT_TRIGGER
  3645. max_sockid=dummysockopen?dummysock[0]:0;
  3646. #else
  3647. opendummy();
  3648. max_sockid=dummysockopen?dummysock:0;
  3649. #endif
  3650. ni = items.ordinality();
  3651. #ifdef _WIN32
  3652. if (offset>=ni)
  3653. #endif
  3654. offset = 0;
  3655. unsigned j=offset;
  3656. ForEachItemIn(i,items) {
  3657. SelectItem &si = items.element(j);
  3658. j++;
  3659. if (j==ni)
  3660. j = 0;
  3661. if (si.mode & SELECTMODE_READ) {
  3662. FD_SET( si.handle, &rdfds );
  3663. isrd = true;
  3664. }
  3665. if (si.mode & SELECTMODE_WRITE) {
  3666. FD_SET( si.handle, &wrfds );
  3667. iswr = true;
  3668. }
  3669. if (si.mode & SELECTMODE_EXCEPT) {
  3670. FD_SET( si.handle, &exfds );
  3671. isex = true;
  3672. }
  3673. max_sockid=std::max(si.handle, max_sockid);
  3674. }
  3675. if (dummysockopen) {
  3676. #ifdef _USE_PIPE_FOR_SELECT_TRIGGER
  3677. FD_SET( dummysock[0], &rdfds );
  3678. isrd = true;
  3679. #else
  3680. FD_SET( dummysock, &exfds );
  3681. isex = true;
  3682. #endif
  3683. }
  3684. validateselecterror = 0;
  3685. max_sockid++;
  3686. #ifdef SOCKTRACE
  3687. PROGLOG("SOCKTRACE: selecting on %d sockets",ni);
  3688. #endif
  3689. }
  3690. int run()
  3691. {
  3692. try {
  3693. T_FD_SET rdfds;
  3694. T_FD_SET wrfds;
  3695. T_FD_SET exfds;
  3696. timeval selecttimeout;
  3697. bool isrd = false;
  3698. bool iswr = false;
  3699. bool isex = false;
  3700. T_SOCKET maxsockid = 0;
  3701. unsigned ni = 0;
  3702. selectvarschange = true;
  3703. unsigned numto = 0;
  3704. unsigned lastnumto = 0;
  3705. unsigned totnum = 0;
  3706. unsigned total = 0;
  3707. while (!terminating) {
  3708. selecttimeout.tv_sec = SELECT_TIMEOUT_SECS; // linux modifies so initialize inside loop
  3709. selecttimeout.tv_usec = 0;
  3710. if (selectvarschange) {
  3711. updateSelectVars(rdfds,wrfds,exfds,isrd,iswr,isex,ni,maxsockid);
  3712. }
  3713. if (ni==0) {
  3714. validateerrcount = 0;
  3715. atomic_inc(&tickwait);
  3716. if(!selectvarschange&&!terminating)
  3717. ticksem.wait(SELECT_TIMEOUT_SECS*1000);
  3718. atomic_dec(&tickwait);
  3719. continue;
  3720. }
  3721. T_FD_SET rs;
  3722. T_FD_SET ws;
  3723. T_FD_SET es;
  3724. T_FD_SET *rsp = isrd?cpyfds(rs,rdfds):NULL;
  3725. T_FD_SET *wsp = iswr?cpyfds(ws,wrfds):NULL;
  3726. T_FD_SET *esp = isex?cpyfds(es,exfds):NULL;
  3727. int n = ::select(maxsockid,(fd_set *)rsp,(fd_set *)wsp,(fd_set *)esp,&selecttimeout); // first parameter needed for posix
  3728. if (terminating)
  3729. break;
  3730. if (n < 0) {
  3731. CriticalBlock block(sect);
  3732. int err = ERRNO();
  3733. if (err != EINTRCALL) {
  3734. if (dummysockopen) {
  3735. LOGERR(err,12,"CSocketSelectThread select error"); // should cache error ?
  3736. validateselecterror = err;
  3737. #ifndef _USE_PIPE_FOR_SELECT_TRIGGER
  3738. closedummy(); // just in case was culprit
  3739. #endif
  3740. }
  3741. selectvarschange = true;
  3742. continue;
  3743. }
  3744. n = 0;
  3745. }
  3746. else if (n>0) {
  3747. validateerrcount = 0;
  3748. numto = 0;
  3749. lastnumto = 0;
  3750. total += n;
  3751. totnum++;
  3752. SelectItemArray tonotify;
  3753. {
  3754. CriticalBlock block(sect);
  3755. #ifdef _WIN32
  3756. if (isrd)
  3757. processfds(rs,SELECTMODE_READ,tonotify);
  3758. if (iswr)
  3759. processfds(ws,SELECTMODE_WRITE,tonotify);
  3760. if (isex)
  3761. processfds(es,SELECTMODE_EXCEPT,tonotify);
  3762. #else
  3763. unsigned i;
  3764. SelectItem *si = items.getArray(offset);
  3765. SelectItem *sie = items.getArray(ni-1)+1;
  3766. bool r = isrd;
  3767. bool w = iswr;
  3768. bool e = isex;
  3769. #ifdef _USE_PIPE_FOR_SELECT_TRIGGER
  3770. if (r&&dummysockopen&&findfds(rs,dummysock[0],r)) {
  3771. resettrigger();
  3772. --n;
  3773. }
  3774. #endif
  3775. for (i=0;(n>0)&&(i<ni);i++) {
  3776. if (r&&findfds(rs,si->handle,r)) {
  3777. if (!si->del) {
  3778. tonotify.append(*si);
  3779. tonotify.element(tonotify.length()-1).mode = SELECTMODE_READ;
  3780. }
  3781. --n;
  3782. }
  3783. if (w&&findfds(ws,si->handle,w)) {
  3784. if (!si->del) {
  3785. tonotify.append(*si);
  3786. tonotify.element(tonotify.length()-1).mode = SELECTMODE_WRITE;
  3787. }
  3788. --n;
  3789. }
  3790. if (e&&findfds(es,si->handle,e)) {
  3791. if (!si->del) {
  3792. tonotify.append(*si);
  3793. tonotify.element(tonotify.length()-1).mode = SELECTMODE_EXCEPT;
  3794. }
  3795. --n;
  3796. }
  3797. si++;
  3798. if (si==sie)
  3799. si = items.getArray();
  3800. }
  3801. #endif
  3802. }
  3803. ForEachItemIn(j,tonotify) {
  3804. const SelectItem &si = tonotify.item(j);
  3805. try {
  3806. si.nfy->notifySelected(si.sock,si.mode); // ignore return
  3807. }
  3808. catch (IException *e) { // should be acted upon by notifySelected
  3809. EXCLOG(e,"CSocketSelectThread notifySelected");
  3810. throw ;
  3811. }
  3812. }
  3813. }
  3814. else {
  3815. validateerrcount = 0;
  3816. if ((++numto>=lastnumto*2)) {
  3817. lastnumto = numto;
  3818. if (selecttrace&&(numto>4))
  3819. PROGLOG("%s: Select Idle(%d), %d,%d,%0.2f",selecttrace,numto,totnum,total,totnum?((double)total/(double)totnum):0.0);
  3820. }
  3821. /*
  3822. if (numto&&(numto%100)) {
  3823. CriticalBlock block(sect);
  3824. if (!selectvarschange)
  3825. selectvarschange = checkSocks();
  3826. }
  3827. */
  3828. }
  3829. if (++offset>=ni)
  3830. offset = 0;
  3831. }
  3832. }
  3833. catch (IException *e) {
  3834. EXCLOG(e,"CSocketSelectThread");
  3835. termexcept.setown(e);
  3836. }
  3837. CriticalBlock block(sect);
  3838. try {
  3839. updateItems();
  3840. }
  3841. catch (IException *e) {
  3842. EXCLOG(e,"CSocketSelectThread(2)");
  3843. if (!termexcept)
  3844. termexcept.setown(e);
  3845. else
  3846. e->Release();
  3847. }
  3848. return 0;
  3849. }
  3850. };
  3851. class CSocketSelectHandler: public CInterface, implements ISocketSelectHandler
  3852. {
  3853. CIArrayOf<CSocketSelectThread> threads;
  3854. CriticalSection sect;
  3855. bool started;
  3856. StringAttr selecttrace;
  3857. public:
  3858. IMPLEMENT_IINTERFACE;
  3859. CSocketSelectHandler(const char *trc)
  3860. : selecttrace(trc)
  3861. {
  3862. started = false;
  3863. }
  3864. void start()
  3865. {
  3866. CriticalBlock block(sect);
  3867. if (!started) {
  3868. started = true;
  3869. ForEachItemIn(i,threads) {
  3870. threads.item(i).start();
  3871. }
  3872. }
  3873. }
  3874. void add(ISocket *sock,unsigned mode,ISocketSelectNotify *nfy)
  3875. {
  3876. CriticalBlock block(sect);
  3877. loop {
  3878. bool added=false;
  3879. ForEachItemIn(i,threads) {
  3880. if (added)
  3881. threads.item(i).remove(sock);
  3882. else
  3883. added = threads.item(i).add(sock,mode,nfy);
  3884. }
  3885. if (added)
  3886. return;
  3887. CSocketSelectThread *thread = new CSocketSelectThread(selecttrace);
  3888. threads.append(*thread);
  3889. if (started)
  3890. thread->start();
  3891. }
  3892. }
  3893. void remove(ISocket *sock)
  3894. {
  3895. CriticalBlock block(sect);
  3896. ForEachItemIn(i,threads) {
  3897. if (threads.item(i).remove(sock)&&sock)
  3898. break;
  3899. }
  3900. }
  3901. void stop(bool wait)
  3902. {
  3903. IException *e=NULL;
  3904. CriticalBlock block(sect);
  3905. unsigned i = 0;
  3906. while (i<threads.ordinality()) {
  3907. CSocketSelectThread &t=threads.item(i);
  3908. {
  3909. CriticalUnblock unblock(sect);
  3910. t.stop(wait); // not quite as quick as could be if wait true
  3911. }
  3912. if (wait && !e && t.termexcept)
  3913. e = t.termexcept.getClear();
  3914. i++;
  3915. }
  3916. #if 0 // don't throw error as too late
  3917. if (e)
  3918. throw e;
  3919. #else
  3920. ::Release(e);
  3921. #endif
  3922. }
  3923. };
  3924. #ifdef _HAS_EPOLL_SUPPORT
  3925. class CSocketEpollThread: public CSocketBaseThread
  3926. {
  3927. int epfd;
  3928. int *epfdtbl; // table of fd<->item index for lookups
  3929. struct epoll_event *epevents;
  3930. void epoll_op(int efd, int op, int fd, unsigned int event_mask)
  3931. {
  3932. int srtn;
  3933. struct epoll_event event;
  3934. // write all bytes to eliminate uninitialized warnings
  3935. memset(&event, 0, sizeof(event));
  3936. event.events = event_mask;
  3937. event.data.fd = fd;
  3938. # ifdef EPOLLTRACE
  3939. DBGLOG("EPOLL: op(%d) fd %d to epfd %d", op, fd, efd);
  3940. # endif
  3941. srtn = ::epoll_ctl(efd, op, fd, &event);
  3942. // if another thread closed fd before here don't fail
  3943. if ( (srtn < 0) && (op != EPOLL_CTL_DEL) ){
  3944. int err = ERRNO();
  3945. LOGERR(err,1,"epoll_ctl");
  3946. THROWJSOCKEXCEPTION2(err);
  3947. }
  3948. }
  3949. void opendummy()
  3950. {
  3951. CriticalBlock block(sect);
  3952. if (!dummysockopen) {
  3953. #ifdef _USE_PIPE_FOR_SELECT_TRIGGER
  3954. if(pipe(dummysock)) {
  3955. WARNLOG("CSocketEpollThread: create pipe failed %d",ERRNO());
  3956. return;
  3957. }
  3958. for (unsigned i=0;i<2;i++) {
  3959. int flags = fcntl(dummysock[i], F_GETFL, 0);
  3960. if (flags!=-1) {
  3961. flags |= O_NONBLOCK;
  3962. fcntl(dummysock[i], F_SETFL, flags);
  3963. }
  3964. flags = fcntl(dummysock[i], F_GETFD, 0);
  3965. if (flags!=-1) {
  3966. flags |= FD_CLOEXEC;
  3967. fcntl(dummysock[i], F_SETFD, flags);
  3968. }
  3969. }
  3970. CHECKSOCKRANGE(dummysock[0]);
  3971. epoll_op(epfd, EPOLL_CTL_ADD, dummysock[0], EPOLLIN);
  3972. #else
  3973. if (IP6preferred)
  3974. dummysock = ::socket(AF_INET6, SOCK_STREAM, PF_INET6);
  3975. else
  3976. dummysock = ::socket(AF_INET, SOCK_STREAM, 0);
  3977. CHECKSOCKRANGE(dummysock);
  3978. // added EPOLLIN also because cannot find anywhere MSG_OOB is sent
  3979. // added here to match existing select() code above which sets
  3980. // the except fd_set mask.
  3981. epoll_op(epfd, EPOLL_CTL_ADD, dummysock, (EPOLLIN | EPOLLPRI));
  3982. #endif
  3983. dummysockopen = true;
  3984. }
  3985. }
  3986. void closedummy()
  3987. {
  3988. CriticalBlock block(sect);
  3989. if (dummysockopen) {
  3990. #ifdef _USE_PIPE_FOR_SELECT_TRIGGER
  3991. epoll_op(epfd, EPOLL_CTL_DEL, dummysock[0], 0);
  3992. #ifdef SOCKTRACE
  3993. PROGLOG("SOCKTRACE: Closing dummy sockets %x %d %x %d (%p)", dummysock[0], dummysock[0], dummysock[1], dummysock[1], this);
  3994. #endif
  3995. ::close(dummysock[0]);
  3996. ::close(dummysock[1]);
  3997. #else
  3998. epoll_op(epfd, EPOLL_CTL_DEL, dummysock, 0);
  3999. ::close(dummysock);
  4000. #endif
  4001. dummysockopen = false;
  4002. }
  4003. }
  4004. public:
  4005. IMPLEMENT_IINTERFACE;
  4006. CSocketEpollThread(const char *trc)
  4007. : CSocketBaseThread("CSocketEpollThread")
  4008. {
  4009. dummysockopen = false;
  4010. terminating = false;
  4011. atomic_set(&tickwait,0);
  4012. waitingchange = 0;
  4013. selectvarschange = false;
  4014. validateselecterror = 0;
  4015. validateerrcount = 0;
  4016. offset = 0;
  4017. selecttrace = trc;
  4018. epfd = ::epoll_create(XFD_SETSIZE);
  4019. if (epfd < 0) {
  4020. int err = ERRNO();
  4021. LOGERR(err,1,"epoll_create()");
  4022. THROWJSOCKEXCEPTION2(err);
  4023. }
  4024. # ifdef EPOLLTRACE
  4025. DBGLOG("CSocketEpollThread: creating epoll fd %d", epfd );
  4026. # endif
  4027. try {
  4028. epfdtbl = new int[XFD_SETSIZE];
  4029. } catch (const std::bad_alloc &e) {
  4030. int err = ERRNO();
  4031. LOGERR(err,1,"epfdtbl alloc()");
  4032. THROWJSOCKEXCEPTION2(err);
  4033. }
  4034. for (int i=0; i<XFD_SETSIZE; i++) {
  4035. epfdtbl[i] = -1;
  4036. }
  4037. try {
  4038. epevents = new struct epoll_event[XFD_SETSIZE];
  4039. } catch (const std::bad_alloc &e) {
  4040. int err = ERRNO();
  4041. LOGERR(err,1,"epevents alloc()");
  4042. THROWJSOCKEXCEPTION2(err);
  4043. }
  4044. opendummy();
  4045. }
  4046. ~CSocketEpollThread()
  4047. {
  4048. closedummy();
  4049. ForEachItemIn(i,items) {
  4050. try {
  4051. SelectItem &si = items.element(i);
  4052. epoll_op(epfd, EPOLL_CTL_DEL, si.handle, 0);
  4053. si.sock->Release();
  4054. si.nfy->Release();
  4055. }
  4056. catch (IException *e) {
  4057. EXCLOG(e,"~CSocketEpollThread");
  4058. e->Release();
  4059. }
  4060. }
  4061. if (epfd >= 0) {
  4062. # ifdef EPOLLTRACE
  4063. DBGLOG("EPOLL: closing epfd %d", epfd );
  4064. # endif
  4065. ::close(epfd);
  4066. epfd = -1;
  4067. delete [] epfdtbl;
  4068. delete [] epevents;
  4069. }
  4070. }
  4071. Owned<IException> termexcept;
  4072. void updateItems()
  4073. {
  4074. // must be in CriticalBlock block(sect);
  4075. unsigned n = items.ordinality();
  4076. bool reindex = false;
  4077. for (unsigned i=0;i<n;) {
  4078. SelectItem &si = items.element(i);
  4079. if (si.add_epoll) {
  4080. reindex = true;
  4081. }
  4082. if (si.del) {
  4083. epoll_op(epfd, EPOLL_CTL_DEL, si.handle, 0);
  4084. epfdtbl[si.handle] = -1;
  4085. reindex = true;
  4086. si.nfy->Release();
  4087. try {
  4088. #ifdef SOCKTRACE
  4089. PROGLOG("CSocketEpollThread::updateItems release %d",si.handle);
  4090. #endif
  4091. si.sock->Release();
  4092. }
  4093. catch (IException *e) {
  4094. EXCLOG(e,"CSocketEpollThread::updateItems");
  4095. e->Release();
  4096. }
  4097. n--;
  4098. if (i<n)
  4099. si = items.item(n);
  4100. items.remove(n);
  4101. }
  4102. else
  4103. i++;
  4104. }
  4105. assertex(n<=XFD_SETSIZE-1);
  4106. if (reindex) {
  4107. # ifdef EPOLLTRACE
  4108. int max_sockid = 0;
  4109. # endif
  4110. ForEachItemIn(j,items) {
  4111. SelectItem &si = items.element(j);
  4112. epfdtbl[si.handle] = j;
  4113. if (si.add_epoll) {
  4114. si.add_epoll = false;
  4115. int srtn, ep_mode;
  4116. struct epoll_event event;
  4117. if (si.mode != 0) {
  4118. ep_mode = 0;
  4119. if (si.mode & SELECTMODE_READ) {
  4120. ep_mode |= EPOLLIN;
  4121. }
  4122. if (si.mode & SELECTMODE_WRITE) {
  4123. ep_mode |= EPOLLOUT;
  4124. }
  4125. if (si.mode & SELECTMODE_EXCEPT) {
  4126. ep_mode |= EPOLLPRI;
  4127. }
  4128. if (ep_mode != 0) {
  4129. epoll_op(epfd, EPOLL_CTL_ADD, si.handle, ep_mode);
  4130. }
  4131. }
  4132. # ifdef EPOLLTRACE
  4133. max_sockid=std::max(si.handle, max_sockid);
  4134. # endif
  4135. }
  4136. # ifdef EPOLLTRACE
  4137. for(int ix=0; ix<=max_sockid; ix++) {
  4138. DBGLOG("EPOLL: epfdtbl[%d] = %d", ix, epfdtbl[ix]);
  4139. }
  4140. # endif
  4141. }
  4142. # ifdef EPOLLTRACE
  4143. DBGLOG("EPOLL: leaving updateItems(), reindex = %d", reindex);
  4144. # endif
  4145. }
  4146. }
  4147. bool add(ISocket *sock,unsigned mode,ISocketSelectNotify *nfy)
  4148. {
  4149. // maybe check once to prevent 1st delay? TBD
  4150. CriticalBlock block(sect);
  4151. unsigned n=0;
  4152. ForEachItemIn(i,items) {
  4153. SelectItem &si = items.element(i);
  4154. if (!si.del) {
  4155. if (si.sock==sock) {
  4156. si.del = true;
  4157. }
  4158. else
  4159. n++;
  4160. }
  4161. }
  4162. if (n>=XFD_SETSIZE-1) // leave 1 spare
  4163. return false;
  4164. SelectItem sn;
  4165. sn.nfy = LINK(nfy);
  4166. sn.sock = LINK(sock);
  4167. sn.mode = (byte)mode;
  4168. sn.handle = (T_SOCKET)sock->OShandle();
  4169. CHECKSOCKRANGE(sn.handle);
  4170. sn.del = false;
  4171. sn.add_epoll = true;
  4172. items.append(sn);
  4173. selectvarschange = true;
  4174. triggerselect();
  4175. return true;
  4176. }
  4177. void updateEpollVars(unsigned &ni)
  4178. {
  4179. CriticalBlock block(sect);
  4180. selectvarschange = false;
  4181. if (waitingchange) {
  4182. waitingchangesem.signal(waitingchange);
  4183. waitingchange = 0;
  4184. }
  4185. if (validateselecterror) { // something went wrong so check sockets
  4186. validateerrcount++;
  4187. if (!checkSocks()) {
  4188. // bad socket not found
  4189. PROGLOG("CSocketEpollThread::updateEpollVars cannot find socket error");
  4190. if (validateerrcount>10)
  4191. throw MakeStringException(-1,"CSocketEpollThread:Socket epoll error %d",validateselecterror);
  4192. }
  4193. }
  4194. else
  4195. validateerrcount = 0;
  4196. updateItems();
  4197. #ifndef _USE_PIPE_FOR_SELECT_TRIGGER
  4198. opendummy();
  4199. #endif
  4200. ni = items.ordinality();
  4201. validateselecterror = 0;
  4202. }
  4203. int run()
  4204. {
  4205. try {
  4206. unsigned ni = 0;
  4207. unsigned numto = 0;
  4208. unsigned lastnumto = 0;
  4209. unsigned totnum = 0;
  4210. unsigned total = 0;
  4211. selectvarschange = true;
  4212. while (!terminating) {
  4213. if (selectvarschange) {
  4214. updateEpollVars(ni);
  4215. }
  4216. if (ni==0) {
  4217. validateerrcount = 0;
  4218. atomic_inc(&tickwait);
  4219. if(!selectvarschange&&!terminating)
  4220. ticksem.wait(SELECT_TIMEOUT_SECS*1000);
  4221. atomic_dec(&tickwait);
  4222. continue;
  4223. }
  4224. int n = ::epoll_wait(epfd, epevents, XFD_SETSIZE, 1000);
  4225. # ifdef EPOLLTRACE
  4226. if(n > 0)
  4227. DBGLOG("EPOLL: after epoll_wait(), n = %d, ni = %d", n, ni);
  4228. # endif
  4229. if (terminating)
  4230. break;
  4231. if (n < 0) {
  4232. CriticalBlock block(sect);
  4233. int err = ERRNO();
  4234. if (err != EINTRCALL) {
  4235. if (dummysockopen) {
  4236. LOGERR(err,12,"CSocketEpollThread epoll error"); // should cache error ?
  4237. validateselecterror = err;
  4238. #ifndef _USE_PIPE_FOR_SELECT_TRIGGER
  4239. closedummy(); // just in case was culprit
  4240. #endif
  4241. }
  4242. selectvarschange = true;
  4243. continue;
  4244. }
  4245. n = 0;
  4246. }
  4247. else if (n>0) {
  4248. validateerrcount = 0;
  4249. numto = 0;
  4250. lastnumto = 0;
  4251. total += n;
  4252. totnum++;
  4253. SelectItemArray tonotify;
  4254. {
  4255. CriticalBlock block(sect);
  4256. for (int j=0;j<n;j++) {
  4257. # ifdef EPOLLTRACE
  4258. DBGLOG("EPOLL: epevents[%d].data.fd = %d, epfdtbl = %d, emask = %d", j, epevents[j].data.fd, epfdtbl[epevents[j].data.fd], epevents[j].events);
  4259. # endif
  4260. # ifdef _USE_PIPE_FOR_SELECT_TRIGGER
  4261. if ((dummysockopen) && (epevents[j].data.fd == dummysock[0])) {
  4262. resettrigger();
  4263. continue;
  4264. }
  4265. # endif
  4266. if (epevents[j].data.fd >= 0) {
  4267. assertex(epfdtbl[epevents[j].data.fd] >= 0);
  4268. SelectItem *epsi = items.getArray(epfdtbl[epevents[j].data.fd]);
  4269. if (!epsi->del) {
  4270. unsigned int ep_mode = 0;
  4271. if (epevents[j].events & (EPOLLIN | EPOLLHUP | EPOLLERR)) {
  4272. ep_mode |= SELECTMODE_READ;
  4273. }
  4274. if (epevents[j].events & EPOLLOUT) {
  4275. ep_mode |= SELECTMODE_WRITE;
  4276. }
  4277. if (epevents[j].events & EPOLLPRI) {
  4278. ep_mode |= SELECTMODE_EXCEPT;
  4279. }
  4280. if (ep_mode != 0) {
  4281. tonotify.append(*epsi);
  4282. tonotify.element(tonotify.length()-1).mode = ep_mode;
  4283. }
  4284. }
  4285. }
  4286. }
  4287. }
  4288. ForEachItemIn(j,tonotify) {
  4289. const SelectItem &si = tonotify.item(j);
  4290. try {
  4291. si.nfy->notifySelected(si.sock,si.mode); // ignore return
  4292. }
  4293. catch (IException *e) { // should be acted upon by notifySelected
  4294. EXCLOG(e,"CSocketEpollThread notifySelected");
  4295. throw ;
  4296. }
  4297. }
  4298. }
  4299. else {
  4300. validateerrcount = 0;
  4301. if ((++numto>=lastnumto*2)) {
  4302. lastnumto = numto;
  4303. if (selecttrace&&(numto>4))
  4304. PROGLOG("%s: Epoll Idle(%d), %d,%d,%0.2f",selecttrace,numto,totnum,total,totnum?((double)total/(double)totnum):0.0);
  4305. }
  4306. /*
  4307. if (numto&&(numto%100)) {
  4308. CriticalBlock block(sect);
  4309. if (!selectvarschange)
  4310. selectvarschange = checkSocks();
  4311. }
  4312. */
  4313. }
  4314. }
  4315. }
  4316. catch (IException *e) {
  4317. EXCLOG(e,"CSocketEpollThread");
  4318. termexcept.setown(e);
  4319. }
  4320. CriticalBlock block(sect);
  4321. try {
  4322. updateItems();
  4323. }
  4324. catch (IException *e) {
  4325. EXCLOG(e,"CSocketEpollThread(2)");
  4326. if (!termexcept)
  4327. termexcept.setown(e);
  4328. else
  4329. e->Release();
  4330. }
  4331. return 0;
  4332. }
  4333. };
  4334. class CSocketEpollHandler: public CInterface, implements ISocketSelectHandler
  4335. {
  4336. CSocketEpollThread *epollthread;
  4337. CriticalSection sect;
  4338. bool started;
  4339. StringAttr epolltrace;
  4340. public:
  4341. IMPLEMENT_IINTERFACE;
  4342. CSocketEpollHandler(const char *trc)
  4343. : epolltrace(trc)
  4344. {
  4345. epollthread = new CSocketEpollThread(epolltrace);
  4346. }
  4347. ~CSocketEpollHandler()
  4348. {
  4349. delete epollthread;
  4350. }
  4351. void start()
  4352. {
  4353. CriticalBlock block(sect);
  4354. epollthread->start();
  4355. }
  4356. void add(ISocket *sock,unsigned mode,ISocketSelectNotify *nfy)
  4357. {
  4358. CriticalBlock block(sect);
  4359. epollthread->add(sock,mode,nfy);
  4360. }
  4361. void remove(ISocket *sock)
  4362. {
  4363. CriticalBlock block(sect);
  4364. epollthread->remove(sock);
  4365. }
  4366. void stop(bool wait)
  4367. {
  4368. IException *e=NULL;
  4369. epollthread->stop(wait); // not quite as quick as could be if wait true
  4370. if (wait && !e && epollthread->termexcept)
  4371. e = epollthread->termexcept.getClear();
  4372. #if 0 // don't throw error as too late
  4373. if (e)
  4374. throw e;
  4375. #else
  4376. ::Release(e);
  4377. #endif
  4378. }
  4379. };
  4380. #endif // _HAS_EPOLL_SUPPORT
  4381. enum EpollMethod { EPOLL_INIT = 0, EPOLL_DISABLED, EPOLL_ENABLED };
  4382. static EpollMethod epoll_method = EPOLL_INIT;
  4383. static CriticalSection epollsect;
  4384. ISocketSelectHandler *createSocketSelectHandler(const char *trc)
  4385. {
  4386. #ifdef _HAS_EPOLL_SUPPORT
  4387. {
  4388. CriticalBlock block(epollsect);
  4389. // DBGLOG("createSocketSelectHandler(): epoll_method = %d",epoll_method);
  4390. if (epoll_method == EPOLL_INIT) {
  4391. Owned<IProperties> conf = createProperties(CONFIG_DIR PATHSEPSTR "environment.conf", true);
  4392. if (conf->getPropBool("use_epoll", true)) {
  4393. epoll_method = EPOLL_ENABLED;
  4394. } else {
  4395. epoll_method = EPOLL_DISABLED;
  4396. }
  4397. // DBGLOG("createSocketSelectHandler(): after reading conf file, epoll_method = %d",epoll_method);
  4398. }
  4399. }
  4400. if (epoll_method == EPOLL_ENABLED)
  4401. return new CSocketEpollHandler(trc);
  4402. else
  4403. return new CSocketSelectHandler(trc);
  4404. #else
  4405. return new CSocketSelectHandler(trc);
  4406. #endif
  4407. }
  4408. ISocketSelectHandler *createSocketEpollHandler(const char *trc)
  4409. {
  4410. #ifdef _HAS_EPOLL_SUPPORT
  4411. return new CSocketEpollHandler(trc);
  4412. #else
  4413. return new CSocketSelectHandler(trc);
  4414. #endif
  4415. }
  4416. void readBuffer(ISocket * socket, MemoryBuffer & buffer)
  4417. {
  4418. size32_t len;
  4419. socket->read(&len, sizeof(len));
  4420. _WINREV4(len);
  4421. if (len) {
  4422. void * target = buffer.reserve(len);
  4423. socket->read(target, len);
  4424. }
  4425. }
  4426. void readBuffer(ISocket * socket, MemoryBuffer & buffer, unsigned timeoutms)
  4427. {
  4428. size32_t len;
  4429. size32_t sizeGot;
  4430. socket->readtms(&len, sizeof(len), sizeof(len), sizeGot, timeoutms);
  4431. _WINREV4(len);
  4432. if (len) {
  4433. void * target = buffer.reserve(len);
  4434. socket->readtms(target, len, len, sizeGot, timeoutms);
  4435. }
  4436. }
  4437. void writeBuffer(ISocket * socket, MemoryBuffer & buffer)
  4438. {
  4439. unsigned len = buffer.length();
  4440. _WINREV4(len);
  4441. socket->write(&len, sizeof(len));
  4442. if (len)
  4443. socket->write(buffer.toByteArray(), buffer.length());
  4444. }
  4445. bool catchReadBuffer(ISocket * socket, MemoryBuffer & buffer)
  4446. {
  4447. try
  4448. {
  4449. readBuffer(socket, buffer);
  4450. return true;
  4451. }
  4452. catch (IException * e)
  4453. {
  4454. switch (e->errorCode())
  4455. {
  4456. case JSOCKERR_graceful_close:
  4457. break;
  4458. default:
  4459. EXCLOG(e,"catchReadBuffer");
  4460. break;
  4461. }
  4462. e->Release();
  4463. }
  4464. return false;
  4465. }
  4466. bool catchReadBuffer(ISocket * socket, MemoryBuffer & buffer, unsigned timeoutms)
  4467. {
  4468. try
  4469. {
  4470. readBuffer(socket, buffer, timeoutms);
  4471. return true;
  4472. }
  4473. catch (IException * e)
  4474. {
  4475. switch (e->errorCode())
  4476. {
  4477. case JSOCKERR_graceful_close:
  4478. break;
  4479. default:
  4480. EXCLOG(e,"catchReadBuffer");
  4481. break;
  4482. }
  4483. e->Release();
  4484. }
  4485. return false;
  4486. }
  4487. bool catchWriteBuffer(ISocket * socket, MemoryBuffer & buffer)
  4488. {
  4489. try
  4490. {
  4491. writeBuffer(socket, buffer);
  4492. return true;
  4493. }
  4494. catch (IException * e)
  4495. {
  4496. EXCLOG(e,"catchWriteBuffer");
  4497. e->Release();
  4498. }
  4499. return false;
  4500. }
  4501. // utility interface for simple conversations
  4502. // conversation is always between two ends,
  4503. // at any given time one end must be receiving and other sending (though these may swap during the conversation)
  4504. class CSingletonSocketConnection: public CInterface, implements IConversation
  4505. {
  4506. Owned<ISocket> sock;
  4507. Owned<ISocket> listensock;
  4508. enum { Snone, Saccept, Sconnect, Srecv, Ssend, Scancelled } state;
  4509. bool accepting;
  4510. bool cancelling;
  4511. SocketEndpoint ep;
  4512. CriticalSection crit;
  4513. public:
  4514. IMPLEMENT_IINTERFACE;
  4515. CSingletonSocketConnection(SocketEndpoint &_ep)
  4516. {
  4517. ep = _ep;
  4518. state = Snone;
  4519. cancelling = false;
  4520. }
  4521. ~CSingletonSocketConnection()
  4522. {
  4523. try {
  4524. if (sock)
  4525. sock->close();
  4526. }
  4527. catch (IException *e) {
  4528. if (e->errorCode()!=JSOCKERR_graceful_close)
  4529. EXCLOG(e,"CSingletonSocketConnection close");
  4530. e->Release();
  4531. }
  4532. }
  4533. bool connect(unsigned timeoutms)
  4534. {
  4535. CriticalBlock block(crit);
  4536. if (cancelling)
  4537. state = Scancelled;
  4538. if (state==Scancelled)
  4539. return false;
  4540. assertex(!sock);
  4541. ISocket *newsock=NULL;
  4542. state = Sconnect;
  4543. unsigned start = 0;
  4544. if (timeoutms!=(unsigned)INFINITE)
  4545. start = msTick();
  4546. while (state==Sconnect) {
  4547. try {
  4548. CriticalUnblock unblock(crit);
  4549. newsock = ISocket::connect_wait(ep,1000*60*4);
  4550. break;
  4551. }
  4552. catch (IException * e) {
  4553. if ((e->errorCode()==JSOCKERR_timeout_expired)||(e->errorCode()==JSOCKERR_connection_failed)) {
  4554. e->Release();
  4555. if ((state==Sconnect)&&(timeoutms!=(unsigned)INFINITE)&&(msTick()-start>timeoutms)) {
  4556. state = Snone;
  4557. return false;
  4558. }
  4559. }
  4560. else {
  4561. state = Scancelled;
  4562. EXCLOG(e,"CSingletonSocketConnection::connect");
  4563. e->Release();
  4564. return false;
  4565. }
  4566. }
  4567. }
  4568. if (state!=Sconnect) {
  4569. ::Release(newsock);
  4570. newsock = NULL;
  4571. }
  4572. if (!newsock) {
  4573. state = Scancelled;
  4574. return false;
  4575. }
  4576. sock.setown(newsock);
  4577. return true;
  4578. }
  4579. bool send(MemoryBuffer &mb)
  4580. {
  4581. CriticalBlock block(crit);
  4582. if (cancelling)
  4583. state = Scancelled;
  4584. if (state==Scancelled)
  4585. return false;
  4586. assertex(sock);
  4587. state = Srecv;
  4588. try {
  4589. CriticalUnblock unblock(crit);
  4590. writeBuffer(sock,mb);
  4591. }
  4592. catch (IException * e) {
  4593. state = Scancelled;
  4594. EXCLOG(e,"CSingletonSocketConnection::send");
  4595. e->Release();
  4596. return false;
  4597. }
  4598. state = Snone;
  4599. return true;
  4600. }
  4601. unsigned short setRandomPort(unsigned short base, unsigned num)
  4602. {
  4603. loop {
  4604. try {
  4605. ep.port = base+(unsigned short)(getRandom()%num);
  4606. listensock.setown(ISocket::create(ep.port));
  4607. return ep.port;
  4608. }
  4609. catch (IException *e) {
  4610. if (e->errorCode()!=JSOCKERR_port_in_use) {
  4611. state = Scancelled;
  4612. EXCLOG(e,"CSingletonSocketConnection::setRandomPort");
  4613. e->Release();
  4614. break;
  4615. }
  4616. e->Release();
  4617. }
  4618. }
  4619. return 0;
  4620. }
  4621. bool accept(unsigned timeoutms)
  4622. {
  4623. CriticalBlock block(crit);
  4624. if (cancelling)
  4625. state = Scancelled;
  4626. if (state==Scancelled)
  4627. return false;
  4628. if (!sock) {
  4629. ISocket *newsock=NULL;
  4630. state = Saccept;
  4631. loop {
  4632. try {
  4633. {
  4634. CriticalUnblock unblock(crit);
  4635. if (!listensock)
  4636. listensock.setown(ISocket::create(ep.port));
  4637. if ((timeoutms!=(unsigned)INFINITE)&&(!listensock->wait_read(timeoutms))) {
  4638. state = Snone;
  4639. return false;
  4640. }
  4641. }
  4642. if (cancelling)
  4643. state = Scancelled;
  4644. if (state==Scancelled)
  4645. return false;
  4646. {
  4647. CriticalUnblock unblock(crit);
  4648. newsock=listensock->accept(true);
  4649. break;
  4650. }
  4651. }
  4652. catch (IException *e) {
  4653. if (e->errorCode()==JSOCKERR_graceful_close)
  4654. PROGLOG("CSingletonSocketConnection: Closed socket on accept - retrying...");
  4655. else {
  4656. state = Scancelled;
  4657. EXCLOG(e,"CSingletonSocketConnection::accept");
  4658. e->Release();
  4659. break;
  4660. }
  4661. e->Release();
  4662. }
  4663. }
  4664. if (state!=Saccept) {
  4665. ::Release(newsock);
  4666. newsock = NULL;
  4667. }
  4668. if (!newsock) {
  4669. state = Scancelled;
  4670. return false;
  4671. }
  4672. sock.setown(newsock);
  4673. }
  4674. return true;
  4675. }
  4676. bool recv(MemoryBuffer &mb, unsigned timeoutms)
  4677. {
  4678. CriticalBlock block(crit);
  4679. if (cancelling)
  4680. state = Scancelled;
  4681. if (state==Scancelled)
  4682. return false;
  4683. assertex(sock);
  4684. state = Srecv;
  4685. try {
  4686. CriticalUnblock unblock(crit);
  4687. readBuffer(sock,mb,timeoutms);
  4688. }
  4689. catch (IException *e) {
  4690. if (e->errorCode()==JSOCKERR_timeout_expired)
  4691. state = Snone;
  4692. else {
  4693. state = Scancelled;
  4694. if (e->errorCode()!=JSOCKERR_graceful_close)
  4695. EXCLOG(e,"CSingletonSocketConnection::recv");
  4696. }
  4697. e->Release();
  4698. return false;
  4699. }
  4700. state = Snone;
  4701. return true;
  4702. }
  4703. virtual void cancel()
  4704. {
  4705. CriticalBlock block(crit);
  4706. while (state!=Scancelled) {
  4707. cancelling = true;
  4708. try {
  4709. switch (state) {
  4710. case Saccept:
  4711. {
  4712. if (listensock)
  4713. listensock->cancel_accept();
  4714. }
  4715. break;
  4716. case Sconnect:
  4717. // wait for timeout
  4718. break;
  4719. case Srecv:
  4720. {
  4721. if (sock)
  4722. sock->close();
  4723. }
  4724. break;
  4725. case Ssend:
  4726. // wait for finished
  4727. break;
  4728. default:
  4729. state = Scancelled;
  4730. break;
  4731. }
  4732. }
  4733. catch (IException *e) {
  4734. EXCLOG(e,"CSingletonSocketConnection::cancel");
  4735. e->Release();
  4736. }
  4737. {
  4738. CriticalUnblock unblock(crit);
  4739. Sleep(1000);
  4740. }
  4741. }
  4742. }
  4743. };
  4744. IConversation *createSingletonSocketConnection(unsigned short port,SocketEndpoint *_ep)
  4745. {
  4746. SocketEndpoint ep;
  4747. if (_ep)
  4748. ep = *_ep;
  4749. if (port)
  4750. ep.port = port;
  4751. return new CSingletonSocketConnection(ep);
  4752. }
  4753. // interface for reading from multiple sockets using the BF_SYNC_TRANSFER_PUSH protocol
  4754. class CSocketBufferReader: public CInterface, implements ISocketBufferReader
  4755. {
  4756. class SocketElem: public CInterface, implements ISocketSelectNotify
  4757. {
  4758. CSocketBufferReader *parent;
  4759. unsigned num; // top bit used for ready
  4760. MemoryAttr blk;
  4761. CriticalSection sect;
  4762. Linked<ISocket> sock;
  4763. bool active;
  4764. bool pending;
  4765. public:
  4766. IMPLEMENT_IINTERFACE;
  4767. void init(CSocketBufferReader *_parent,ISocket *_sock,unsigned _n)
  4768. {
  4769. parent = _parent;
  4770. num = _n;
  4771. sock.set(_sock);
  4772. active = true;
  4773. pending = false;
  4774. }
  4775. virtual bool notifySelected(ISocket *socket,unsigned selected)
  4776. {
  4777. assertex(sock==socket);
  4778. {
  4779. CriticalBlock block(sect);
  4780. if (pending) {
  4781. active = false;
  4782. parent->remove(sock);
  4783. return false;
  4784. }
  4785. pending = true;
  4786. unsigned t1=usTick();
  4787. size32_t sz = sock->receive_block_size();
  4788. unsigned t2=usTick();
  4789. if (sz)
  4790. sock->receive_block(blk.allocate(sz),sz);
  4791. else
  4792. parent->remove(sock);
  4793. unsigned t3=usTick();
  4794. if (t3-t1>60*1000000)
  4795. PROGLOG("CSocketBufferReader(%d): slow receive_block (%d,%d) sz=%d",num,t2-t1,t3-t2,sz);
  4796. }
  4797. parent->enqueue(this); // nb outside sect critical block
  4798. return false; // always return false
  4799. }
  4800. unsigned get(MemoryBuffer &mb)
  4801. {
  4802. CriticalBlock block(sect);
  4803. assertex(pending);
  4804. size32_t sz = blk.length();
  4805. if (sz)
  4806. mb.setBuffer(sz,blk.detach(),true);
  4807. pending = false;
  4808. if (!active) {
  4809. active = true;
  4810. parent->add(*this);
  4811. }
  4812. return num;
  4813. }
  4814. size32_t size()
  4815. {
  4816. return blk.length();
  4817. }
  4818. ISocket *getSocket() { return sock; }
  4819. } *elems;
  4820. SimpleInterThreadQueueOf<SocketElem, false> readyq;
  4821. Owned<ISocketSelectHandler> selecthandler;
  4822. size32_t buffersize;
  4823. size32_t buffermax;
  4824. unsigned bufferwaiting;
  4825. CriticalSection buffersect;
  4826. Semaphore buffersem;
  4827. bool isdone;
  4828. public:
  4829. IMPLEMENT_IINTERFACE;
  4830. CSocketBufferReader(const char *trc)
  4831. {
  4832. selecthandler.setown(createSocketSelectHandler(trc));
  4833. elems = NULL;
  4834. }
  4835. ~CSocketBufferReader()
  4836. {
  4837. delete [] elems;
  4838. }
  4839. virtual void init(unsigned num,ISocket **sockets,size32_t _buffermax)
  4840. {
  4841. elems = new SocketElem[num];
  4842. for (unsigned i=0;i<num;i++) {
  4843. ISocket *sock = sockets[i];
  4844. if (sock) { // can have gaps
  4845. elems[i].init(this,sock,i);
  4846. add(elems[i]);
  4847. }
  4848. }
  4849. buffersize = 0;
  4850. buffermax = _buffermax;
  4851. bufferwaiting = 0;
  4852. isdone = false;
  4853. selecthandler->start();
  4854. }
  4855. virtual unsigned get(MemoryBuffer &mb)
  4856. {
  4857. SocketElem &e = *readyq.dequeue();
  4858. CriticalBlock block(buffersect);
  4859. assertex(buffersize>=e.size());
  4860. buffersize-=e.size();
  4861. if (bufferwaiting) {
  4862. buffersem.signal(bufferwaiting);
  4863. bufferwaiting = 0;
  4864. }
  4865. return e.get(mb);
  4866. }
  4867. virtual void done(bool wait)
  4868. {
  4869. buffersem.signal(0x10000);
  4870. isdone = true;
  4871. selecthandler->stop(wait);
  4872. if (wait) {
  4873. delete [] elems;
  4874. elems = NULL;
  4875. }
  4876. }
  4877. void enqueue(SocketElem *elem)
  4878. {
  4879. if (elem) {
  4880. CriticalBlock block(buffersect);
  4881. size32_t sz = elem->size();
  4882. while ((buffersize>0)&&(sz>0)&&(buffersize+sz>buffermax)) {
  4883. if (isdone)
  4884. return;
  4885. bufferwaiting++;
  4886. CriticalUnblock unblock(buffersect);
  4887. buffersem.wait();
  4888. }
  4889. buffersize += sz;
  4890. }
  4891. readyq.enqueue(elem);
  4892. }
  4893. void remove(ISocket *sock)
  4894. {
  4895. selecthandler->remove(sock);
  4896. }
  4897. void add(SocketElem &elem)
  4898. {
  4899. selecthandler->add(elem.getSocket(),SELECTMODE_READ,&elem);
  4900. }
  4901. };
  4902. ISocketBufferReader *createSocketBufferReader(const char *trc)
  4903. {
  4904. return new CSocketBufferReader(trc);
  4905. }
  4906. extern jlib_decl void markNodeCentral(SocketEndpoint &ep)
  4907. {
  4908. #ifdef CENTRAL_NODE_RANDOM_DELAY
  4909. CriticalBlock block(CSocket::crit);
  4910. CentralNodeArray.append(ep);
  4911. #endif
  4912. }
  4913. static CSocket *prepareSocket(unsigned idx,const SocketEndpoint &ep, ISocketConnectNotify &inotify)
  4914. {
  4915. Owned<CSocket> sock = new CSocket(ep,sm_tcp,NULL);
  4916. int err = sock->pre_connect(false);
  4917. if ((err == EINPROGRESS)||(err == EWOULDBLOCK))
  4918. return sock.getClear();
  4919. if (err==0) {
  4920. int err = sock->post_connect();
  4921. if (err==0)
  4922. inotify.connected(idx,ep,sock);
  4923. else {
  4924. sock->errclose();
  4925. inotify.failed(idx,ep,err);
  4926. }
  4927. }
  4928. else
  4929. inotify.failed(idx,ep,err);
  4930. return NULL;
  4931. }
  4932. void multiConnect(const SocketEndpointArray &eps,ISocketConnectNotify &inotify,unsigned timeout)
  4933. {
  4934. class SocketElem: public CInterface, implements ISocketSelectNotify
  4935. {
  4936. CriticalSection *sect;
  4937. ISocketSelectHandler *handler;
  4938. unsigned *remaining;
  4939. Semaphore *notifysem;
  4940. ISocketConnectNotify *inotify;
  4941. public:
  4942. Owned<CSocket> sock;
  4943. SocketEndpoint ep;
  4944. unsigned idx;
  4945. IMPLEMENT_IINTERFACE;
  4946. void init(CSocket *_sock,unsigned _idx,const SocketEndpoint &_ep,CriticalSection *_sect,ISocketSelectHandler *_handler,ISocketConnectNotify *_inotify, unsigned *_remaining, Semaphore *_notifysem)
  4947. {
  4948. ep = _ep;
  4949. idx = _idx;
  4950. inotify = _inotify;
  4951. sock.setown(_sock),
  4952. sect = _sect;
  4953. handler = _handler;
  4954. remaining = _remaining;
  4955. notifysem = _notifysem;
  4956. }
  4957. virtual bool notifySelected(ISocket *socket,unsigned selected)
  4958. {
  4959. CriticalBlock block(*sect);
  4960. handler->remove(socket);
  4961. int err = sock->post_connect();
  4962. CSocket *newsock = NULL;
  4963. {
  4964. CriticalUnblock unblock(*sect); // up to caller to cope with multithread
  4965. if (err==0)
  4966. inotify->connected(idx,ep,sock);
  4967. else if ((err==ETIMEDOUT)||(err==ECONNREFUSED)) {
  4968. // don't give up so easily (maybe listener not yet started (i.e. racing))
  4969. newsock = prepareSocket(idx,ep,*inotify);
  4970. Sleep(100); // not very nice but without this would just loop
  4971. }
  4972. else
  4973. inotify->failed(idx,ep,err);
  4974. }
  4975. if (newsock) {
  4976. sock.setown(newsock);
  4977. handler->add(sock,SELECTMODE_WRITE|SELECTMODE_EXCEPT,this);
  4978. }
  4979. else {
  4980. sock.clear();
  4981. (*remaining)--;
  4982. notifysem->signal();
  4983. }
  4984. return false;
  4985. }
  4986. } *elems;
  4987. unsigned n = eps.ordinality();
  4988. unsigned remaining = n;
  4989. if (!n)
  4990. return;
  4991. elems = new SocketElem[n];
  4992. unsigned i;
  4993. CriticalSection sect;
  4994. Semaphore notifysem;
  4995. Owned<ISocketSelectHandler> selecthandler = createSocketSelectHandler(
  4996. #ifdef _DEBUG
  4997. "multiConnect"
  4998. #else
  4999. NULL
  5000. #endif
  5001. );
  5002. StringBuffer name;
  5003. for (i=0;i<n;i++) {
  5004. CSocket* sock = prepareSocket(i,eps.item(i),inotify);
  5005. if (sock) {
  5006. elems[i].init(sock,i,eps.item(i),&sect,selecthandler,&inotify,&remaining,&notifysem);
  5007. selecthandler->add(sock,SELECTMODE_WRITE|SELECTMODE_EXCEPT,&elems[i]);
  5008. }
  5009. else
  5010. remaining--;
  5011. }
  5012. if (remaining) {
  5013. unsigned lastremaining=remaining;
  5014. selecthandler->start();
  5015. loop {
  5016. bool to=!notifysem.wait(timeout);
  5017. {
  5018. CriticalBlock block(sect);
  5019. if (remaining==0)
  5020. break;
  5021. if (to&&(remaining==lastremaining))
  5022. break; // nothing happened recently
  5023. lastremaining = remaining;
  5024. }
  5025. }
  5026. selecthandler->stop(true);
  5027. }
  5028. selecthandler.clear();
  5029. if (remaining) {
  5030. for (unsigned j=0;j<n;j++) { // mop up timeouts
  5031. SocketElem &elem = elems[j];
  5032. if (elem.sock.get()) {
  5033. elem.sock.clear();
  5034. inotify.failed(j,elem.ep,-1);
  5035. remaining--;
  5036. if (remaining==0)
  5037. break;
  5038. }
  5039. }
  5040. }
  5041. delete [] elems;
  5042. }
  5043. void multiConnect(const SocketEndpointArray &eps, IPointerArrayOf<ISocket> &retsockets,unsigned timeout)
  5044. {
  5045. unsigned n = eps.ordinality();
  5046. if (n==0)
  5047. return;
  5048. if (n==1) { // no need for multi
  5049. ISocket *sock = NULL;
  5050. try {
  5051. sock = ISocket::connect_timeout(eps.item(0),timeout);
  5052. }
  5053. catch (IException *e) { // ignore error just append NULL
  5054. sock = NULL;
  5055. e->Release();
  5056. }
  5057. retsockets.append(sock);
  5058. return;
  5059. }
  5060. while (retsockets.ordinality()<n)
  5061. retsockets.append(NULL);
  5062. CriticalSection sect;
  5063. class cNotify: implements ISocketConnectNotify
  5064. {
  5065. CriticalSection &sect;
  5066. IPointerArrayOf<ISocket> &retsockets;
  5067. public:
  5068. cNotify(IPointerArrayOf<ISocket> &_retsockets,CriticalSection &_sect)
  5069. : retsockets(_retsockets),sect(_sect)
  5070. {
  5071. }
  5072. void connected(unsigned idx,const SocketEndpoint &ep,ISocket *sock)
  5073. {
  5074. CriticalBlock block(sect);
  5075. assertex(idx<retsockets.ordinality());
  5076. sock->Link();
  5077. retsockets.replace(sock,idx);
  5078. }
  5079. void failed(unsigned idx,const SocketEndpoint &ep,int err)
  5080. {
  5081. StringBuffer s;
  5082. PROGLOG("multiConnect failed to %s with %d",ep.getUrlStr(s).str(),err);
  5083. }
  5084. } notify(retsockets,sect);
  5085. multiConnect(eps,notify,timeout);
  5086. }
  5087. inline void flushText(StringBuffer &text,unsigned short port,unsigned &rep,unsigned &range)
  5088. {
  5089. if (rep) {
  5090. text.append('*').append(rep+1);
  5091. rep = 0;
  5092. }
  5093. else if (range) {
  5094. text.append('-').append(range);
  5095. range = 0;
  5096. }
  5097. if (port)
  5098. text.append(':').append(port);
  5099. }
  5100. StringBuffer &SocketEndpointArray::getText(StringBuffer &text)
  5101. {
  5102. unsigned count = ordinality();
  5103. if (!count)
  5104. return text;
  5105. if (count==1)
  5106. return item(0).getUrlStr(text);
  5107. byte lastip[4];
  5108. const SocketEndpoint &first = item(0);
  5109. bool lastis4 = first.getNetAddress(sizeof(lastip),&lastip)==sizeof(lastip);
  5110. unsigned short lastport = first.port;
  5111. first.getIpText(text);
  5112. unsigned rep=0;
  5113. unsigned range=0;
  5114. for (unsigned i=1;i<count;i++) {
  5115. byte ip[4];
  5116. const SocketEndpoint &ep = item(i);
  5117. bool is4 = ep.getNetAddress(sizeof(ip),&ip)==sizeof(ip);
  5118. if (!lastis4||!is4) {
  5119. flushText(text,lastport,rep,range);
  5120. text.append(',');
  5121. ep.getIpText(text);
  5122. }
  5123. else { // try and shorten
  5124. unsigned j;
  5125. for (j=0;j<4;j++)
  5126. if (ip[j]!=lastip[j])
  5127. break;
  5128. if (ep.port==lastport) {
  5129. if (j==4) {
  5130. if (range) // cant have range and rep
  5131. j--; // pretend only 3 matched
  5132. else {
  5133. rep++;
  5134. continue;
  5135. }
  5136. }
  5137. else if ((j==3)&&(lastip[3]+1==ip[3])&&(rep==0)) {
  5138. range = ip[3];
  5139. lastip[3] = (byte)range;
  5140. continue;
  5141. }
  5142. }
  5143. flushText(text,lastport,rep,range);
  5144. // output diff
  5145. text.append(',');
  5146. if (j==4)
  5147. j--;
  5148. for (unsigned k=j;k<4;k++) {
  5149. if (k>j)
  5150. text.append('.');
  5151. text.append((int)ip[k]);
  5152. }
  5153. }
  5154. memcpy(&lastip,&ip,sizeof(lastip));
  5155. lastis4 = is4;
  5156. lastport = ep.port;
  5157. }
  5158. flushText(text,lastport,rep,range);
  5159. return text;
  5160. }
  5161. inline const char *getnum(const char *s,unsigned &n)
  5162. {
  5163. n = 0;
  5164. while (isdigit(*s)) {
  5165. n = n*10+(*s-'0');
  5166. s++;
  5167. }
  5168. return s;
  5169. }
  5170. inline bool appendv4range(SocketEndpointArray *array,char *str,SocketEndpoint &ep, unsigned defport)
  5171. {
  5172. char *s = str;
  5173. unsigned dc = 0;
  5174. unsigned port = defport;
  5175. unsigned rng = 0;
  5176. unsigned rep = 1;
  5177. bool notip = false;
  5178. while (*s) {
  5179. if (*s=='.') {
  5180. dc++;
  5181. s++;
  5182. }
  5183. else if (*s==':') {
  5184. *s = 0;
  5185. s = (char *)getnum(s+1,port);
  5186. }
  5187. else if (*s=='-') {
  5188. *s = 0;
  5189. s = (char *)getnum(s+1,rng);
  5190. }
  5191. else if (*s=='*') {
  5192. *s = 0;
  5193. s = (char *)getnum(s+1,rep);
  5194. }
  5195. else {
  5196. if (!isdigit(*s))
  5197. notip = true;
  5198. s++;
  5199. }
  5200. }
  5201. ep.port = port;
  5202. if (*str) {
  5203. if (!notip&&((dc<3)&&((dc!=1)||(strlen(str)!=1)))) {
  5204. if (!ep.isIp4()) {
  5205. return false;
  5206. }
  5207. StringBuffer tmp;
  5208. ep.getIpText(tmp);
  5209. size32_t l = tmp.length();
  5210. dc++;
  5211. loop {
  5212. if (tmp.length()==0)
  5213. return false;
  5214. if (tmp.charAt(tmp.length()-1)=='.')
  5215. if (--dc==0)
  5216. break;
  5217. tmp.setLength(tmp.length()-1);
  5218. }
  5219. tmp.append(str);
  5220. if (rng) {
  5221. tmp.appendf("-%d",rng);
  5222. rep = ep.ipsetrange(tmp.str());
  5223. }
  5224. else
  5225. ep.ipset(tmp.str());
  5226. }
  5227. else if (rng) { // not nice as have to add back range (must be better way - maybe ipincrementto) TBD
  5228. StringBuffer tmp;
  5229. tmp.appendf("%s-%d",str,rng);
  5230. rep = ep.ipsetrange(tmp.str());
  5231. }
  5232. else if (*str)
  5233. ep.ipset(str);
  5234. if (ep.isNull())
  5235. ep.port = 0;
  5236. for (unsigned i=0;i<rep;i++) {
  5237. array->append(ep);
  5238. if (rng)
  5239. ep.ipincrement(1);
  5240. }
  5241. }
  5242. else {// just a port change
  5243. if (ep.isNull()) // avoid null values with ports
  5244. ep.port = 0;
  5245. array->append(ep);
  5246. }
  5247. return true;
  5248. }
  5249. void SocketEndpointArray::fromText(const char *text,unsigned defport)
  5250. {
  5251. // this is quite complicated with (mixed) IPv4 and IPv6
  5252. // only support 'full' IPv6 and no ranges
  5253. char *str = strdup(text);
  5254. char *s = str;
  5255. SocketEndpoint ep;
  5256. bool eol = false;
  5257. loop {
  5258. while (isspace(*s)||(*s==','))
  5259. s++;
  5260. if (!*s)
  5261. break;
  5262. char *e=s;
  5263. if (*e=='[') { // we have a IPv6
  5264. while (*e&&(*e!=']'))
  5265. e++;
  5266. while ((*e!=',')&&!isspace(*e)) {
  5267. if (!*s) {
  5268. eol = true;
  5269. break;
  5270. }
  5271. e++;
  5272. }
  5273. *e = 0;
  5274. ep.set(s,defport);
  5275. if (ep.isNull()) {
  5276. // Error TBD
  5277. }
  5278. append(ep);
  5279. }
  5280. else {
  5281. bool hascolon = false;
  5282. bool isv6 = false;
  5283. do {
  5284. if (*e==':') {
  5285. if (hascolon)
  5286. isv6 = true;
  5287. else
  5288. hascolon = true;
  5289. }
  5290. e++;
  5291. if (!*e) {
  5292. eol = true;
  5293. break;
  5294. }
  5295. } while (!isspace(*e)&&(*e!=','));
  5296. *e = 0;
  5297. if (isv6) {
  5298. ep.set(s,defport);
  5299. if (ep.isNull()) {
  5300. // Error TBD
  5301. }
  5302. append(ep);
  5303. }
  5304. else {
  5305. if (!appendv4range(this,s,ep,defport)) {
  5306. // Error TBD
  5307. }
  5308. }
  5309. }
  5310. if (eol)
  5311. break;
  5312. s = e+1;
  5313. }
  5314. free(str);
  5315. }
  5316. bool IpSubNet::set(const char *_net,const char *_mask)
  5317. {
  5318. if (!_net||!decodeNumericIP(_net,net)) { // _net NULL means match everything
  5319. memset(net,0,sizeof(net));
  5320. memset(mask,0,sizeof(mask));
  5321. return (_net==NULL);
  5322. }
  5323. if (!_mask||!decodeNumericIP(_mask,mask)) { // _mask NULL means match exact
  5324. memset(mask,0xff,sizeof(mask));
  5325. return (_mask==NULL);
  5326. }
  5327. if (isIp4(net)!=isIp4(mask))
  5328. return false;
  5329. for (unsigned j=0;j<4;j++)
  5330. if (net[j]&~mask[j])
  5331. return false;
  5332. return true;
  5333. }
  5334. bool IpSubNet::test(const IpAddress &ip) const
  5335. {
  5336. unsigned i;
  5337. if (ip.getNetAddress(sizeof(i),&i)==sizeof(i)) {
  5338. if (!isIp4(net))
  5339. return false;
  5340. return (i&mask[3])==(net[3]&mask[3]);
  5341. }
  5342. unsigned na[4];
  5343. if (ip.getNetAddress(sizeof(na),&na)==sizeof(na)) {
  5344. for (unsigned j=0;j<4;j++)
  5345. if ((na[j]&mask[j])!=(net[j]&mask[j]))
  5346. return false;
  5347. return true;
  5348. }
  5349. return false;
  5350. }
  5351. StringBuffer IpSubNet::getNetText(StringBuffer &text) const
  5352. {
  5353. char tmp[INET6_ADDRSTRLEN];
  5354. const char *res = ::isIp4(net) ? _inet_ntop(AF_INET, &net[3], tmp, sizeof(tmp))
  5355. : _inet_ntop(AF_INET6, &net, tmp, sizeof(tmp));
  5356. return text.append(res);
  5357. }
  5358. StringBuffer IpSubNet::getMaskText(StringBuffer &text) const
  5359. {
  5360. char tmp[INET6_ADDRSTRLEN];
  5361. // isIp4(net) is correct here
  5362. const char *res = ::isIp4(net) ? _inet_ntop(AF_INET, &mask[3], tmp, sizeof(tmp))
  5363. : _inet_ntop(AF_INET6, &mask, tmp, sizeof(tmp));
  5364. return text.append(res);
  5365. }
  5366. bool IpSubNet::isNull() const
  5367. {
  5368. for (unsigned i=0;i<4;i++)
  5369. if (net[i]||mask[i])
  5370. return false;
  5371. return true;
  5372. }
  5373. IpSubNet &queryPreferredSubnet()
  5374. {
  5375. return PreferredSubnet;
  5376. }
  5377. bool setPreferredSubnet(const char *ip,const char *mask)
  5378. {
  5379. // also resets cached host IP
  5380. if (PreferredSubnet.set(ip,mask))
  5381. {
  5382. if (!cachehostip.isNull())
  5383. {
  5384. cachehostip.ipset(NULL);
  5385. queryHostIP();
  5386. }
  5387. return true;
  5388. }
  5389. else
  5390. return false;
  5391. }
  5392. StringBuffer lookupHostName(const IpAddress &ip,StringBuffer &ret)
  5393. {
  5394. // not a common routine (no Jlib function!) only support IPv4 initially
  5395. unsigned ipa;
  5396. if (ip.getNetAddress(sizeof(ipa),&ipa)==sizeof(ipa)) {
  5397. struct hostent *phostent = gethostbyaddr( (char *) &ipa, sizeof(ipa), PF_INET);
  5398. if (phostent)
  5399. ret.append(phostent->h_name);
  5400. else
  5401. ip.getIpText(ret);
  5402. }
  5403. else
  5404. ip.getIpText(ret);
  5405. return ret;
  5406. }
  5407. struct SocketEndpointHTElem
  5408. {
  5409. IInterface *ii;
  5410. SocketEndpoint ep;
  5411. SocketEndpointHTElem(const SocketEndpoint _ep,IInterface *_ii) { ep.set(_ep); ii = _ii; }
  5412. ~SocketEndpointHTElem() { ::Release(ii); }
  5413. };
  5414. class jlib_decl CSocketEndpointHashTable : public SuperHashTableOf<SocketEndpointHTElem, SocketEndpoint>, implements ISocketEndpointHashTable
  5415. {
  5416. virtual void onAdd(void *) {}
  5417. virtual void onRemove(void *e) { delete (SocketEndpointHTElem *)e; }
  5418. unsigned getHashFromElement(const void *e) const
  5419. {
  5420. return ((const SocketEndpointHTElem *)e)->ep.hash(0);
  5421. }
  5422. unsigned getHashFromFindParam(const void *fp) const
  5423. {
  5424. return ((const SocketEndpoint *)fp)->hash(0);
  5425. }
  5426. const void * getFindParam(const void *p) const
  5427. {
  5428. return &((const SocketEndpointHTElem *)p)->ep;
  5429. }
  5430. bool matchesFindParam(const void * et, const void *fp, unsigned) const
  5431. {
  5432. return ((const SocketEndpointHTElem *)et)->ep.equals(*(SocketEndpoint *)fp);
  5433. }
  5434. IMPLEMENT_SUPERHASHTABLEOF_REF_FIND(SocketEndpointHTElem,SocketEndpoint);
  5435. public:
  5436. IMPLEMENT_IINTERFACE;
  5437. CSocketEndpointHashTable() {}
  5438. ~CSocketEndpointHashTable() { kill(); }
  5439. void add(const SocketEndpoint &ep, IInterface *i)
  5440. {
  5441. SocketEndpointHTElem *e = SuperHashTableOf<SocketEndpointHTElem,SocketEndpoint>::find(&ep);
  5442. if (e) {
  5443. ::Release(e->ii);
  5444. e->ii = i;
  5445. }
  5446. else {
  5447. e = new SocketEndpointHTElem(ep,i);
  5448. SuperHashTableOf<SocketEndpointHTElem,SocketEndpoint>::add(*e);
  5449. }
  5450. }
  5451. void remove(const SocketEndpoint &ep)
  5452. {
  5453. SuperHashTableOf<SocketEndpointHTElem,SocketEndpoint>::remove(&ep);
  5454. }
  5455. IInterface *find(const SocketEndpoint &ep)
  5456. {
  5457. SocketEndpointHTElem *e = SuperHashTableOf<SocketEndpointHTElem,SocketEndpoint>::find(&ep);
  5458. if (e)
  5459. return e->ii;
  5460. return NULL;
  5461. }
  5462. };
  5463. ISocketEndpointHashTable *createSocketEndpointHashTable()
  5464. {
  5465. CSocketEndpointHashTable *ht = new CSocketEndpointHashTable;
  5466. return ht;
  5467. }
  5468. class CSocketConnectWait: public CInterface, implements ISocketConnectWait
  5469. {
  5470. Owned<CSocket> sock;
  5471. bool done;
  5472. CTimeMon connecttm;
  5473. unsigned startt;
  5474. bool oneshot;
  5475. bool isopen;
  5476. int initerr;
  5477. void successfulConnect()
  5478. {
  5479. STATS.connects++;
  5480. STATS.connecttime+=usTick()-startt;
  5481. #ifdef _TRACE
  5482. char peer[256];
  5483. peer[0] = 'C';
  5484. peer[1] = '!';
  5485. strcpy(peer+2,sock->hostname?sock->hostname:"(NULL)");
  5486. free(sock->tracename);
  5487. sock->tracename = strdup(peer);
  5488. #endif
  5489. }
  5490. void failedConnect()
  5491. {
  5492. STATS.failedconnects++;
  5493. STATS.failedconnecttime+=usTick()-startt;
  5494. const char* tracename = sock->tracename;
  5495. THROWJSOCKEXCEPTION(JSOCKERR_connection_failed);
  5496. }
  5497. public:
  5498. IMPLEMENT_IINTERFACE;
  5499. CSocketConnectWait(SocketEndpoint &ep,unsigned connecttimeoutms)
  5500. : connecttm(connecttimeoutms)
  5501. {
  5502. oneshot = (connecttimeoutms==0); // i.e. as long as one connect takes
  5503. done = false;
  5504. startt = usTick();
  5505. sock.setown(new CSocket(ep,sm_tcp,NULL));
  5506. isopen = true;
  5507. initerr = sock->pre_connect(false);
  5508. }
  5509. ISocket *wait(unsigned timems)
  5510. {
  5511. // this is a bit spagetti due to dual timeouts etc
  5512. CTimeMon waittm(timems);
  5513. unsigned refuseddelay = 1;
  5514. bool waittimedout = false;
  5515. bool connectimedout = false;
  5516. do {
  5517. bool connectdone = false;
  5518. unsigned remaining;
  5519. connectimedout = connecttm.timedout(&remaining);
  5520. unsigned waitremaining;
  5521. waittimedout = waittm.timedout(&waitremaining);
  5522. if (oneshot||(waitremaining<remaining))
  5523. remaining = waitremaining;
  5524. int err = 0;
  5525. if (!isopen||initerr) {
  5526. isopen = true;
  5527. err = initerr?initerr:sock->pre_connect(false);
  5528. initerr = 0;
  5529. if ((err == EINPROGRESS)||(err == EWOULDBLOCK))
  5530. err = 0; // continue
  5531. else {
  5532. if (err==0)
  5533. connectdone = true; // done immediately
  5534. else if(!oneshot) // probably ECONNREFUSED but treat all errors same
  5535. refused_sleep((waitremaining==remaining)?waittm:connecttm,refuseddelay); // this stops becoming cpu bound
  5536. }
  5537. }
  5538. if (!connectdone&&(err==0)) {
  5539. SOCKET s = sock->sock;
  5540. T_FD_SET fds;
  5541. struct timeval tv;
  5542. XFD_ZERO(&fds);
  5543. FD_SET((unsigned)s, &fds);
  5544. T_FD_SET except;
  5545. XFD_ZERO(&except);
  5546. FD_SET((unsigned)s, &except);
  5547. tv.tv_sec = remaining / 1000;
  5548. tv.tv_usec = (remaining % 1000)*1000;
  5549. CHECKSOCKRANGE(s);
  5550. int rc = ::select( s + 1, NULL, (fd_set *)&fds, (fd_set *)&except, &tv );
  5551. if (rc==0)
  5552. break; // timeout
  5553. done = true;
  5554. err = 0;
  5555. if (rc>0) {
  5556. // select succeeded - return error from socket (0 if connected)
  5557. socklen_t errlen = sizeof(err);
  5558. rc = getsockopt(s, SOL_SOCKET, SO_ERROR, (char *)&err, &errlen); // check for error
  5559. if ((rc!=0)&&!err)
  5560. err = ERRNO(); // some implementations of getsockopt duff
  5561. if (err&&!oneshot) // probably ECONNREFUSED but treat all errors same
  5562. refused_sleep((waitremaining==remaining)?waittm:connecttm,refuseddelay); // this stops becoming cpu bound
  5563. }
  5564. else { // select failed
  5565. err = ERRNO();
  5566. LOGERR(err,2,"CSocketConnectWait ::select");
  5567. }
  5568. }
  5569. if (err==0) {
  5570. err = sock->post_connect();
  5571. if (err==0) {
  5572. successfulConnect();
  5573. return sock.getClear();
  5574. }
  5575. }
  5576. sock->errclose();
  5577. isopen = false;
  5578. } while (!waittimedout&&!oneshot);
  5579. if (connectimedout) {
  5580. STATS.failedconnects++;
  5581. STATS.failedconnecttime+=usTick()-startt;
  5582. const char* tracename = sock->tracename;
  5583. THROWJSOCKEXCEPTION(JSOCKERR_connection_failed);
  5584. }
  5585. return NULL;
  5586. }
  5587. };
  5588. ISocketConnectWait *nonBlockingConnect(SocketEndpoint &ep,unsigned connecttimeoutms)
  5589. {
  5590. return new CSocketConnectWait(ep,connecttimeoutms);
  5591. }
  5592. int wait_multiple(bool isRead, //IN true if wait read, false it wait write
  5593. UnsignedArray &socks, //IN sockets to be checked for readiness
  5594. unsigned timeoutMS, //IN timeout
  5595. UnsignedArray &readySocks) //OUT sockets ready
  5596. {
  5597. aindex_t numSocks = socks.length();
  5598. if (numSocks == 0)
  5599. THROWJSOCKEXCEPTION2(JSOCKERR_bad_address);
  5600. SOCKET maxSocket = 0;
  5601. T_FD_SET fds;
  5602. XFD_ZERO(&fds);
  5603. //Add each SOCKET in array to T_FD_SET
  5604. #ifdef _DEBUG
  5605. StringBuffer dbgSB("wait_multiple() on sockets :");
  5606. #endif
  5607. for (aindex_t idx = 0; idx < numSocks; idx++)
  5608. {
  5609. #ifdef _DEBUG
  5610. dbgSB.appendf(" %d",socks.item(idx));
  5611. #endif
  5612. maxSocket = socks.item(idx) > maxSocket ? socks.item(idx) : maxSocket;
  5613. FD_SET((unsigned)socks.item(idx), &fds);
  5614. }
  5615. #ifdef _DEBUG
  5616. DBGLOG("%s",dbgSB.str());
  5617. #endif
  5618. //Check socket states
  5619. int res;
  5620. if (timeoutMS == WAIT_FOREVER)
  5621. res = ::select( maxSocket + 1, isRead ? (fd_set *)&fds : NULL, isRead ? NULL : (fd_set *)&fds, NULL, NULL );
  5622. else
  5623. {
  5624. struct timeval tv;
  5625. tv.tv_sec = timeoutMS / 1000;
  5626. tv.tv_usec = (timeoutMS % 1000)*1000;
  5627. res = ::select( maxSocket + 1, isRead ? (fd_set *)&fds : NULL, isRead ? NULL : (fd_set *)&fds, NULL, &tv );
  5628. }
  5629. if (res != SOCKET_ERROR)
  5630. {
  5631. #ifdef _DEBUG
  5632. StringBuffer dbgSB("wait_multiple() ready socket(s) :");
  5633. #endif
  5634. //Build up list of socks which are ready for accept read/write without blocking
  5635. for (aindex_t idx = 0; res && idx < socks.length(); idx++)
  5636. {
  5637. if (FD_ISSET(socks.item(idx), &fds))
  5638. {
  5639. #ifdef _DEBUG
  5640. dbgSB.appendf(" %d",socks.item(idx));
  5641. #endif
  5642. readySocks.append(socks.item(idx));
  5643. if (readySocks.length() == res)
  5644. break;
  5645. }
  5646. }
  5647. #ifdef _DEBUG
  5648. if (res)
  5649. DBGLOG("%s",dbgSB.str());
  5650. #endif
  5651. }
  5652. else
  5653. {
  5654. int err = ERRNO();
  5655. if (err != EINTRCALL)
  5656. {
  5657. throw MakeStringException(-1,"wait_multiple::select error %d", err);
  5658. }
  5659. }
  5660. return res;
  5661. }
  5662. //Given a list of sockets, wait until any one or more are ready to be read (wont block)
  5663. //returns 0 if timeout, number of waiting sockets otherwise
  5664. int wait_read_multiple(UnsignedArray &socks, //IN sockets to be checked for readiness
  5665. unsigned timeoutMS, //IN timeout
  5666. UnsignedArray &readySocks) //OUT sockets ready to be read
  5667. {
  5668. return wait_multiple(true, socks, timeoutMS, readySocks);
  5669. }
  5670. int wait_write_multiple(UnsignedArray &socks, //IN sockets to be checked for readiness
  5671. unsigned timeoutMS, //IN timeout
  5672. UnsignedArray &readySocks) //OUT sockets ready to be written
  5673. {
  5674. return wait_multiple(false, socks, timeoutMS, readySocks);
  5675. }