jsocket.cpp 191 KB

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