jsocket.cpp 187 KB

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